在 AI 应用开发领域大模型本身的能力已经足够强大但真正让这些能力落地到具体业务场景、解决复杂问题的往往是 Agent 技术。一个设计良好的 Agent 能够理解用户意图、规划执行步骤、调用工具、处理异常并最终交付可靠结果。对于希望从基础功能调用进阶到复杂系统开发的工程师来说掌握 Agent 开发是企业级 AI 应用的核心竞争力。本文将以企业级实战为目标通过 12 个由浅入深的案例带你系统掌握 Agent 的开发、调试、部署和优化。每个案例都围绕一个具体业务场景包含完整的环境准备、代码实现、参数调优和问题排查路径。学完后你将能独立设计并实现满足生产要求的 AI Agent 应用。1. 理解 AI Agent 的核心机制与典型架构AI Agent 不是简单的大模型 API 调用封装而是一个具备感知、决策、执行和反思能力的自治系统。在企业环境中Agent 需要处理不确定的外部输入、管理多步任务状态、安全地调用外部工具并在出错时具备恢复或降级能力。1.1 Agent 的基本工作流程一个典型的 Agent 执行流程包括任务解析、规划、执行和总结四个阶段。以“帮用户查询北京明天天气并推荐穿衣”为例任务解析模型理解用户请求识别关键信息地点北京时间明天动作查询天气并推荐穿衣。规划模型决定需要调用哪些工具天气查询 API、按什么顺序执行先查天气再根据温度推荐穿衣。执行按规划调用工具获取实时数据。总结将工具执行结果整合成自然语言回复。这个流程中模型需要维护对话历史、工具调用结果和当前执行状态这对架构设计提出了明确要求。1.2 企业级 Agent 的架构组件生产环境中的 Agent 系统通常包含以下核心组件推理引擎负责理解用户输入、规划执行路径的核心模型。可以是 GPT-4、Claude 3 等通用大模型也可以是针对特定领域微调的专用模型。工具集Agent 可以调用的外部功能如数据库查询、API 调用、文件操作、计算器等。工具需要明确定义输入输出格式和错误处理方式。记忆模块存储对话历史、工具调用结果、用户偏好等上下文信息。短期记忆保存在会话中长期记忆可持久化到数据库。控制逻辑管理执行流程包括重试机制、超时控制、并行执行、异常处理等。安全与监控权限验证、输入输出过滤、操作审计、性能指标收集等企业级功能。下面是一个简化版 Agent 系统的 Python 类结构示意class EnterpriseAgent: def __init__(self, model, tools, memory): self.model model # 推理引擎 self.tools tools # 工具字典 {name: tool_function} self.memory memory # 记忆存储 self.max_retries 3 # 最大重试次数 def parse_intent(self, user_input): 解析用户意图识别需要调用的工具 pass def plan_execution(self, intent): 根据意图规划执行步骤 pass def execute_tool(self, tool_name, parameters): 安全执行单个工具 pass def run(self, user_input): 主执行流程 pass这种架构确保了各组件职责清晰便于测试、扩展和维护。2. 搭建基础的 Agent 开发环境企业级 Agent 开发需要稳定的模型服务、版本化的工具管理和可重现的测试环境。下面以 OpenAI API 和本地工具开发为例搭建最小可行环境。2.1 模型服务配置对于大多数企业场景建议从云服务商的大模型 API 开始降低部署复杂度。以 OpenAI 为例# 安装必要的 Python 包 pip install openai python-dotenv # 创建环境变量文件 .env echo OPENAI_API_KEY你的API密钥 .env创建基础的模型调用封装# model_client.py import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() class ModelClient: def __init__(self, model_namegpt-3.5-turbo): self.client OpenAI(api_keyos.getenv(OPENAI_API_KEY)) self.model_name model_name def chat_completion(self, messages, temperature0.7): try: response self.client.chat.completions.create( modelself.model_name, messagesmessages, temperaturetemperature ) return response.choices[0].message.content except Exception as e: print(f模型调用失败: {e}) return None2.2 工具管理系统工具是 Agent 能力的扩展需要统一的管理接口。每个工具应该包含名称、描述、参数定义和执行函数# tools/calculator.py def calculator(a: float, b: float, operator: str) - float: 基础计算器工具 Args: a: 第一个数字 b: 第二个数字 operator: 运算符支持 、-、*、/ Returns: 计算结果 if operator : return a b elif operator -: return a - b elif operator *: return a * b elif operator /: if b 0: raise ValueError(除数不能为零) return a / b else: raise ValueError(f不支持的运算符: {operator}) # 工具注册表 TOOL_REGISTRY { calculator: { function: calculator, description: 执行基础数学运算, parameters: { a: {type: number, description: 第一个操作数}, b: {type: number, description: 第二个操作数}, operator: {type: string, description: 运算符} } } }2.3 验证环境完整性创建测试脚本验证环境配置是否正确# test_environment.py from model_client import ModelClient from tools.calculator import calculator def test_model_connection(): 测试模型连接 client ModelClient() response client.chat_completion([{role: user, content: 回复OK}]) assert response OK, f模型连接测试失败: {response} print(✓ 模型连接正常) def test_tool_function(): 测试工具功能 result calculator(10, 5, ) assert result 15, f工具测试失败: {result} print(✓ 工具功能正常) if __name__ __main__: test_model_connection() test_tool_function() print(环境验证通过)这个基础环境为后续的 Agent 实战提供了稳定的开发基础。3. 实现第一个可工作的对话型 Agent有了环境基础现在实现一个能够理解用户意图并调用相应工具的简单 Agent。这个 Agent 将具备基础的对话能力和工具调用能力。3.1 设计 Agent 的提示词模板提示词是 Agent 的大脑需要明确定义角色、能力和行为规范# prompts/system_prompt.py SYSTEM_PROMPT 你是一个有帮助的AI助手可以调用工具来解决用户问题。 你可以使用的工具 {tools_descriptions} 工作流程 1. 理解用户请求判断是否需要调用工具 2. 如果需要调用工具明确指定工具名称和参数 3. 等待工具执行结果 4. 根据结果生成最终回复 工具调用格式 tool {{ name: 工具名称, parameters: {{ 参数名: 参数值 }} }}如果不需要调用工具直接回复用户问题。 保持回复专业、简洁、有帮助。 def build_system_prompt(tools_info): 构建系统提示词 tools_descriptions [] for name, info in tools_info.items(): params_desc , .join([f{pname}({pinfo[type]}) for pname, pinfo in info[parameters].items()]) tools_descriptions.append(f- {name}: {info[description]} 参数: {params_desc})return SYSTEM_PROMPT.format(tools_descriptions\n.join(tools_descriptions))### 3.2 实现工具调用逻辑 工具调用需要处理参数验证、执行异常和结果格式化 python # agent/tool_executor.py import json import inspect from typing import Dict, Any class ToolExecutor: def __init__(self, tool_registry): self.tool_registry tool_registry def parse_tool_call(self, model_response: str) - Dict[str, Any]: 从模型响应中解析工具调用指令 if tool not in model_response: return None try: # 提取工具调用代码块 start_idx model_response.find(tool) 7 end_idx model_response.find(, start_idx) tool_json model_response[start_idx:end_idx].strip() tool_data json.loads(tool_json) return tool_data except (json.JSONDecodeError, KeyError) as e: print(f工具调用解析失败: {e}) return None def validate_parameters(self, tool_name: str, parameters: Dict) - bool: 验证工具参数是否合法 if tool_name not in self.tool_registry: return False tool_info self.tool_registry[tool_name] expected_params set(tool_info[parameters].keys()) provided_params set(parameters.keys()) return expected_params provided_params def execute_tool(self, tool_name: str, parameters: Dict) - Any: 执行工具并返回结果 if tool_name not in self.tool_registry: return f错误: 未知工具 {tool_name} if not self.validate_parameters(tool_name, parameters): return f错误: 工具 {tool_name} 参数验证失败 try: tool_func self.tool_registry[tool_name][function] result tool_func(**parameters) return result except Exception as e: return f工具执行错误: {str(e)}3.3 构建完整的 Agent 类整合模型调用、工具执行和对话管理# agent/basic_agent.py from model_client import ModelClient from prompts.system_prompt import build_system_prompt from tool_executor import ToolExecutor class BasicAgent: def __init__(self, model_client, tool_registry): self.model_client model_client self.tool_executor ToolExecutor(tool_registry) self.conversation_history [] # 初始化系统提示词 system_prompt build_system_prompt(tool_registry) self.conversation_history.append({role: system, content: system_prompt}) def process_message(self, user_input: str) - str: 处理用户输入并返回回复 # 添加用户消息到历史 self.conversation_history.append({role: user, content: user_input}) # 获取模型响应 model_response self.model_client.chat_completion(self.conversation_history) if not model_response: return 抱歉暂时无法处理您的请求 # 检查是否需要工具调用 tool_call self.tool_executor.parse_tool_call(model_response) if tool_call: # 执行工具调用 tool_result self.tool_executor.execute_tool( tool_call[name], tool_call[parameters] ) # 添加工具结果到对话历史 self.conversation_history.append({ role: user, content: f工具执行结果: {tool_result} }) # 获取最终回复 final_response self.model_client.chat_completion(self.conversation_history) self.conversation_history.append({role: assistant, content: final_response}) return final_response else: # 直接使用模型回复 self.conversation_history.append({role: assistant, content: model_response}) return model_response3.4 测试第一个 Agent创建测试脚本验证 Agent 功能# test_basic_agent.py from model_client import ModelClient from tools.calculator import TOOL_REGISTRY from agent.basic_agent import BasicAgent def test_calculation_agent(): 测试计算器Agent model_client ModelClient() agent BasicAgent(model_client, TOOL_REGISTRY) # 测试简单计算 response agent.process_message(请计算 25 乘以 38 等于多少) print(f用户: 请计算 25 乘以 38 等于多少) print(fAgent: {response}) # 测试复杂计算 response agent.process_message(那再计算 95 除以 5 呢) print(f用户: 那再计算 95 除以 5 呢) print(fAgent: {response}) if __name__ __main__: test_calculation_agent()这个基础 Agent 已经具备了理解用户意图、调用工具和维持对话的能力为后续更复杂的 Agent 开发奠定了基础。4. 开发多步骤任务规划 Agent实际业务中的问题往往需要多个步骤才能解决。多步骤任务规划 Agent 能够将复杂问题分解为可执行的子任务并按顺序或并行执行。4.1 设计任务规划提示词多步骤规划需要模型具备任务分解和依赖分析能力# prompts/planning_prompt.py PLANNING_SYSTEM_PROMPT 你是一个任务规划专家能够将复杂问题分解为可执行的步骤。 你的工作流程 1. 分析用户请求识别需要完成的主要任务 2. 将任务分解为逻辑上连贯的步骤序列 3. 为每个步骤指定要使用的工具和输入参数 4. 考虑步骤之间的依赖关系和数据传递 输出格式 plan { task: 原始任务描述, steps: [ { step_id: 1, description: 步骤描述, tool: 工具名称, parameters: {参数: 值}, depends_on: [] // 依赖的步骤ID } ] }确保每个步骤都是具体、可执行的。 def create_planning_messages(user_input, available_tools): 创建规划专用的消息列表 tools_list \n.join([f- {name}: {info[description]} for name, info in available_tools.items()])return [ {role: system, content: PLANNING_SYSTEM_PROMPT}, {role: user, content: f可用工具:\n{tools_list}\n\n用户请求: {user_input}} ]### 4.2 实现任务执行引擎 任务执行引擎需要管理步骤状态、处理依赖关系和收集执行结果 python # agent/task_engine.py import json import asyncio from typing import List, Dict, Any from datetime import datetime class TaskExecutionEngine: def __init__(self, tool_executor): self.tool_executor tool_executor self.task_state {} # 存储任务执行状态 def parse_plan(self, plan_response: str) - Dict[str, Any]: 解析规划结果 try: start_idx plan_response.find(plan) 7 end_idx plan_response.find(, start_idx) plan_json plan_response[start_idx:end_idx].strip() return json.loads(plan_json) except (json.JSONDecodeError, ValueError) as e: raise ValueError(f规划解析失败: {e}) def validate_plan(self, plan: Dict) - bool: 验证规划合理性 if steps not in plan or not isinstance(plan[steps], list): return False step_ids set() for step in plan[steps]: # 检查必要字段 if not all(key in step for key in [step_id, tool, parameters]): return False # 检查步骤ID唯一性 if step[step_id] in step_ids: return False step_ids.add(step[step_id]) # 检查依赖关系 if depends_on in step: for dep_id in step[depends_on]: if dep_id not in step_ids: return False return True def execute_step(self, step: Dict, context: Dict) - Dict: 执行单个步骤 try: # 解析参数中的变量引用如 ${step_1.result} resolved_params self.resolve_parameters(step[parameters], context) # 执行工具 result self.tool_executor.execute_tool(step[tool], resolved_params) return { step_id: step[step_id], status: completed, result: result, timestamp: datetime.now().isoformat() } except Exception as e: return { step_id: step[step_id], status: failed, error: str(e), timestamp: datetime.now().isoformat() } def resolve_parameters(self, parameters: Dict, context: Dict) - Dict: 解析参数中的变量引用 resolved {} for key, value in parameters.items(): if isinstance(value, str) and value.startswith(${) and value.endswith(}): # 变量引用如 ${step_1.result} var_path value[2:-1].split(.) var_value context try: for part in var_path: var_value var_value[part] resolved[key] var_value except KeyError: resolved[key] value # 保持原值 else: resolved[key] value return resolved def execute_plan(self, plan: Dict) - Dict[str, Any]: 执行整个规划 if not self.validate_plan(plan): raise ValueError(无效的任务规划) execution_context {} results [] # 按步骤ID排序执行简单实现实际应该考虑依赖关系 sorted_steps sorted(plan[steps], keylambda x: x[step_id]) for step in sorted_steps: # 检查依赖是否就绪 if depends_on in step: dependencies_ready all( dep_id in [r[step_id] for r in results if r[status] completed] for dep_id in step[depends_on] ) if not dependencies_ready: results.append({ step_id: step[step_id], status: skipped, error: 依赖未就绪, timestamp: datetime.now().isoformat() }) continue # 执行步骤 step_result self.execute_step(step, execution_context) results.append(step_result) # 更新执行上下文 if step_result[status] completed: execution_context[fstep_{step[step_id]}] { result: step_result[result] } return { task_id: ftask_{datetime.now().strftime(%Y%m%d_%H%M%S)}, original_task: plan[task], status: completed if all(r[status] in [completed, skipped] for r in results) else failed, results: results, execution_time: datetime.now().isoformat() }4.3 创建规划型 Agent整合规划能力和执行引擎# agent/planning_agent.py from model_client import ModelClient from prompts.planning_prompt import create_planning_messages from task_engine import TaskExecutionEngine class PlanningAgent: def __init__(self, model_client, tool_registry): self.model_client model_client self.task_engine TaskExecutionEngine(ToolExecutor(tool_registry)) self.available_tools tool_registry def create_plan(self, user_input: str) - Dict: 创建任务规划 messages create_planning_messages(user_input, self.available_tools) plan_response self.model_client.chat_completion(messages, temperature0.3) if not plan_response: raise ValueError(规划创建失败) return self.task_engine.parse_plan(plan_response) def execute_task(self, user_input: str) - Dict: 执行完整任务 # 创建规划 plan self.create_plan(user_input) print(f任务规划: {plan[task]}) # 执行规划 execution_result self.task_engine.execute_plan(plan) return execution_result def format_execution_result(self, result: Dict) - str: 格式化执行结果为用户友好的格式 if result[status] completed: successful_steps [r for r in result[results] if r[status] completed] return f任务执行完成共执行 {len(successful_steps)} 个步骤。 else: failed_steps [r for r in result[results] if r[status] failed] return f任务执行失败{len(failed_steps)} 个步骤出错。4.4 测试多步骤任务 Agent创建包含多个工具的场景进行测试# tools/advanced_tools.py import requests from datetime import datetime, timedelta def get_weather(city: str) - Dict: 获取天气信息模拟实现 # 实际项目中应该调用真实天气API weather_data { 北京: {temperature: 25, condition: 晴, humidity: 40}, 上海: {temperature: 28, condition: 多云, humidity: 65}, 深圳: {temperature: 30, condition: 雨, humidity: 80} } return weather_data.get(city, {error: 城市不存在}) def suggest_clothing(temperature: int, condition: str) - str: 根据天气推荐穿衣 if temperature 28: return 建议穿短袖、短裤等夏装 elif temperature 20: return 建议穿长袖T恤、薄外套 else: return 建议穿毛衣、厚外套 def calculate_travel_time(distance_km: float, traffic: str normal) - float: 计算旅行时间 base_speed 60 # km/h if traffic heavy: base_speed * 0.6 elif traffic light: base_speed * 1.2 return distance_km / base_speed ADVANCED_TOOL_REGISTRY { get_weather: { function: get_weather, description: 获取城市天气信息, parameters: { city: {type: string, description: 城市名称} } }, suggest_clothing: { function: suggest_clothing, description: 根据天气条件推荐穿衣, parameters: { temperature: {type: number, description: 温度}, condition: {type: string, description: 天气状况} } }, calculate_travel_time: { function: calculate_travel_time, description: 计算旅行时间, parameters: { distance_km: {type: number, description: 距离公里}, traffic: {type: string, description: 交通状况} } } } # test_planning_agent.py from model_client import ModelClient from tools.advanced_tools import ADVANCED_TOOL_REGISTRY from agent.planning_agent import PlanningAgent def test_travel_planning(): 测试旅行规划任务 model_client ModelClient() agent PlanningAgent(model_client, ADVANCED_TOOL_REGISTRY) # 测试复杂任务 user_request 我明天要去北京出差请帮我查询天气并推荐合适的衣服再计算从机场到市中心的30公里路程需要多长时间 result agent.execute_task(user_request) print(f任务结果: {agent.format_execution_result(result)}) # 打印详细执行结果 for step_result in result[results]: if step_result[status] completed: print(f步骤 {step_result[step_id]}: 成功 - 结果: {step_result[result]}) else: print(f步骤 {step_result[step_id]}: 失败 - 错误: {step_result[error]}) if __name__ __main__: test_travel_planning()这个多步骤任务规划 Agent 展示了如何处理复杂业务逻辑将用户需求分解为可执行的工作流为构建企业级自动化系统奠定了基础。5. 实现记忆管理与上下文保持在实际对话场景中Agent 需要记住之前的交互内容避免用户重复提供信息。记忆管理包括短期会话记忆和长期知识记忆。5.1 设计记忆存储系统记忆系统需要支持多种类型的记忆存储和检索# memory/memory_manager.py from typing import Dict, List, Any from datetime import datetime import json class MemoryManager: def __init__(self, max_short_term_items20): self.short_term_memory [] # 短期会话记忆 self.long_term_memory {} # 长期知识记忆 self.max_short_term_items max_short_term_items def add_conversation(self, user_input: str, agent_response: str, metadata: Dict None): 添加对话记录到短期记忆 memory_item { timestamp: datetime.now().isoformat(), user_input: user_input, agent_response: agent_response, metadata: metadata or {} } self.short_term_memory.append(memory_item) # 保持短期记忆大小 if len(self.short_term_memory) self.max_short_term_items: self.short_term_memory self.short_term_memory[-self.max_short_term_items:] def get_recent_conversations(self, count5) - List[Dict]: 获取最近的对话记录 return self.short_term_memory[-count:] if self.short_term_memory else [] def add_fact(self, key: str, value: Any, category: str general): 添加事实到长期记忆 if category not in self.long_term_memory: self.long_term_memory[category] {} self.long_term_memory[category][key] { value: value, timestamp: datetime.now().isoformat(), access_count: 0 } def get_fact(self, key: str, category: str None) - Any: 从长期记忆获取事实 if category: if category in self.long_term_memory and key in self.long_term_memory[category]: self.long_term_memory[category][key][access_count] 1 return self.long_term_memory[category][key][value] else: # 在所有类别中搜索 for cat in self.long_term_memory.values(): if key in cat: cat[key][access_count] 1 return cat[key][value] return None def search_memories(self, query: str, max_results3) - List[Dict]: 基于关键词搜索记忆 results [] # 搜索短期记忆 for memory in self.short_term_memory: if query.lower() in memory[user_input].lower() or query.lower() in memory[agent_response].lower(): results.append({ type: conversation, content: memory, relevance: high # 简单实现实际应该计算相关性分数 }) # 搜索长期记忆 for category, facts in self.long_term_memory.items(): for key, fact in facts.items(): if query.lower() in key.lower() or query.lower() in str(fact[value]).lower(): results.append({ type: fact, category: category, key: key, value: fact[value], relevance: medium }) return results[:max_results] def save_to_file(self, filepath: str): 保存记忆到文件 memory_data { short_term: self.short_term_memory, long_term: self.long_term_memory } with open(filepath, w, encodingutf-8) as f: json.dump(memory_data, f, ensure_asciiFalse, indent2) def load_from_file(self, filepath: str): 从文件加载记忆 try: with open(filepath, r, encodingutf-8) as f: memory_data json.load(f) self.short_term_memory memory_data.get(short_term, []) self.long_term_memory memory_data.get(long_term, {}) except FileNotFoundError: print(记忆文件不存在使用空记忆)5.2 实现上下文感知的提示词构建基于记忆内容动态构建提示词让 Agent 具备上下文感知能力# prompts/context_aware_prompt.py CONTEXT_AWARE_SYSTEM_PROMPT 你是一个有帮助的AI助手能够记住之前的对话并根据上下文提供更好的服务。 当前对话上下文 {conversation_context} 相关记忆信息 {relevant_memories} 工作流程 1. 首先回顾对话历史和相关信息 2. 理解当前用户请求与上下文的关联 3. 如果需要调用工具按格式执行 4. 生成自然、连贯的回复 如果用户引用之前的对话内容请基于记忆信息准确回应。 保持对话的连贯性和一致性。 def build_context_aware_prompt(memory_manager, current_input, tools_descriptions): 构建包含上下文的系统提示词 # 获取最近对话 recent_conversations memory_manager.get_recent_conversations(3) conversation_context \n.join([ f用户: {conv[user_input]}\n助手: {conv[agent_response]} for conv in recent_conversations ]) if recent_conversations else 暂无历史对话 # 搜索相关记忆 relevant_memories memory_manager.search_memories(current_input) memories_text \n.join([ f- {mem[type]}: {mem.get(key, )} {mem.get(value, )} for mem in relevant_memories ]) if relevant_memories else 暂无相关记忆 return CONTEXT_AWARE_SYSTEM_PROMPT.format( conversation_contextconversation_context, relevant_memoriesmemories_text, tools_descriptionstools_descriptions )5.3 创建具备记忆能力的 Agent整合记忆管理到基础 Agent# agent/memory_agent.py from model_client import ModelClient from prompts.context_aware_prompt import build_context_aware_prompt from memory.memory_manager import MemoryManager from tool_executor import ToolExecutor class MemoryAgent: def __init__(self, model_client, tool_registry, memory_fileNone): self.model_client model_client self.tool_executor ToolExecutor(tool_registry) self.memory_manager MemoryManager() self.tool_registry tool_registry # 加载现有记忆 if memory_file: self.memory_manager.load_from_file(memory_file) def process_message(self, user_input: str) - str: 处理用户输入考虑上下文记忆 # 构建上下文感知的提示词 tools_descriptions \n.join([ f- {name}: {info[description]} for name, info in self.tool_registry.items() ]) system_prompt build_context_aware_prompt( self.memory_manager, user_input, tools_descriptions ) # 准备对话消息 messages [ {role: system, content: system_prompt}, {role: user, content: user_input} ] # 获取模型响应 model_response self.model_client.chat_completion(messages) if not model_response: return 抱歉暂时无法处理您的请求 # 处理工具调用简化版 tool_call self.tool_executor.parse_tool_call(model_response) if tool_call: tool_result self.tool_executor.execute_tool( tool_call[name], tool_call[parameters] ) # 获取最终回复 messages.append({role: user, content: f工具结果: {tool_result}}) final_response self.model_client.chat_completion(messages) else: final_response model_response # 保存到记忆 self.memory_manager.add_conversation(user_input, final_response) # 尝试从对话中提取重要信息保存为长期记忆 self._extract_and_save_facts(user_input, final_response) return final_response def _extract_and_save_facts(self, user_input: str, agent_response: str): 从对话中提取重要信息保存为长期记忆 # 简单的关键词匹配提取实际应该用更智能的方法 important_keywords [我叫, 我的名字是, 我喜欢, 我讨厌, 我住在] for keyword in important_keywords: if keyword in user_input: # 提取关键词后的内容 start_idx user_input.find(keyword) len(keyword) fact_value user_input[start_idx:].split(。)[0].split()[0].strip() if fact_value: fact_key keyword.replace(我, 用户).strip() self.memory_manager.add_fact(fact_key, fact_value, user_preferences) break def save_memory(self, filepath: str): 保存记忆到文件 self.memory_manager.save_to_file(filepath)5.4 测试记忆功能验证 Agent 的记忆能力# test_memory_agent.py from model_client import ModelClient from tools.calculator import TOOL_REGISTRY from agent.memory_agent import MemoryAgent def test_memory_functionality(): 测试记忆功能 model_client ModelClient() agent MemoryAgent(model_client, TOOL_REGISTRY, test_memory.json) # 第一轮对话 print( 第一轮对话 ) response1 agent.process_message(你好我叫张三) print(f用户: 你好我叫张三) print(fAgent: {response1}) # 第二轮对话测试记忆 print(\n 第二轮对话 ) response2 agent.process_message(你还记得我的名字吗) print(f用户: 你还记得我的名字吗) print(f