AI Agent开发实战：从基础对话到企业级多步骤任务规划

在 AI 应用开发领域，大模型本身的能力已经足够强大，但真正让这些能力落地到具体业务场景、解决复杂问题的，往往是 Agent 技术。一个设计良好的 Agent 能够理解用户意图、规划执行步骤、调用工具、处理异常，并最终交付可靠结果。对于希望从基础功能调用进阶到复杂系统开发的工程师来说，掌握 Agent 开发是企业级 AI 应用的核心竞争力。

本文将以企业级实战为目标，通过 12 个由浅入深的案例，带你系统掌握 Agent 的开发、调试、部署和优化。每个案例都围绕一个具体业务场景，包含完整的环境准备、代码实现、参数调优和问题排查路径。学完后，你将能独立设计并实现满足生产要求的 AI Agent 应用。

1. 理解 AI Agent 的核心机制与典型架构

AI Agent 不是简单的大模型 API 调用封装，而是一个具备感知、决策、执行和反思能力的自治系统。在企业环境中，Agent 需要处理不确定的外部输入、管理多步任务状态、安全地调用外部工具，并在出错时具备恢复或降级能力。

1.1 Agent 的基本工作流程

一个典型的 Agent 执行流程包括任务解析、规划、执行和总结四个阶段。以“帮用户查询北京明天天气并推荐穿衣”为例：

1. 任务解析：模型理解用户请求，识别关键信息（地点：北京，时间：明天，动作：查询天气并推荐穿衣）。
2. 规划：模型决定需要调用哪些工具（天气查询 API）、按什么顺序执行（先查天气，再根据温度推荐穿衣）。
3. 执行：按规划调用工具，获取实时数据。
4. 总结：将工具执行结果整合成自然语言回复。

这个流程中，模型需要维护对话历史、工具调用结果和当前执行状态，这对架构设计提出了明确要求。

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_name="gpt-3.5-turbo"): self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) self.model_name = model_name def chat_completion(self, messages, temperature=0.7): try: response = self.client.chat.completions.create( model=self.model_name, messages=messages, temperature=temperature ) return response.choices[0].message.content except Exception as e: print(f"模型调用失败: {e}") return None

2.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_response

3.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(f"Agent: {response}") # 测试复杂计算 response = agent.process_message("那再计算 95 除以 5 呢？") print(f"用户: 那再计算 95 除以 5 呢？") print(f"Agent: {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"], key=lambda 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[f"step_{step['step_id']}"] = { "result": step_result["result"] } return { "task_id": f"task_{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, temperature=0.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_items=20): 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, count=5) -> 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_results=3) -> 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', encoding='utf-8') as f: json.dump(memory_data, f, ensure_ascii=False, indent=2) def load_from_file(self, filepath: str): """从文件加载记忆""" try: with open(filepath, 'r', encoding='utf-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_context=conversation_context, relevant_memories=memories_text, tools_descriptions=tools_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_file=None): 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(f"Agent: {response1}") # 第二轮对话，测试记忆 print("\n=== 第二轮对话 ===") response2 = agent.process_message("你还记得我的名字吗？") print(f"用户: 你还记得我的名字吗？") print(f"