1. 从概念到落地：为什么你的AI Agent总是“不好用”？
最近和几个团队交流，发现大家在做AI Agent时，普遍陷入一个怪圈：Demo跑得飞快，功能演示天花乱坠，一旦想把它放到真实业务流里，立马就“水土不服”。要么是工具调用时灵时不灵，要么是对话聊着聊着就忘了上下文，要么是处理复杂任务时逻辑混乱。这背后的核心，往往不是大模型本身的能力问题，而是我们忽略了构建一个生产级AI Agent所必需的工程化架构——特别是工具调用与记忆架构这两个支柱。
一个能上生产线的Agent，和我们在Notebook里随手写的脚本，完全是两码事。生产级意味着稳定、可靠、可观测、可维护。它需要像一个经验丰富的员工，不仅知道怎么用工具（工具调用），还能记住之前的对话、任务状态和自己的“经验教训”（记忆架构），并且在复杂环境中做出连贯、合理的决策。很多人直接把OpenAI的Function Calling或者LangChain的AgentExecutor拿来就用，结果就是面对稍微复杂点的场景，Agent的表现就变得不可预测，调试起来更是噩梦。
这篇文章，我想结合自己从零搭建并部署多个业务Agent的实战经验，抛开那些华而不实的框架包装，直接深入到工具调用与记忆系统的设计原理和工程细节里。我们会聊清楚：一个健壮的工具调用链路应该如何设计错误处理和状态管理？记忆系统到底该存什么、怎么存、存多久？如何让Agent拥有“工作记忆”和“长期经验”？我会用具体的代码示例和架构图（文字描述）来拆解，目标是让你看完后，能直接着手改造或构建一个真正能扛住生产流量、解决实际问题的AI Agent系统。
2. 超越简单封装：构建鲁棒的工具调用引擎
工具调用（Tool Calling）是AI Agent与外部世界交互的手和脚。但很多实现仅仅是把大模型的函数调用结果解析出来，然后去执行对应的函数，这离“生产级”还差得很远。一个生产级的工具调用引擎，必须考虑编排、验证、容错、状态管理与可观测性。
2.1 工具的定义与描述：给模型清晰的“说明书”
首先，工具的定义不能马虎。模型的调用效果，很大程度上取决于你如何描述这个工具。一个常见的错误是描述过于简略或模糊。
# 不佳的示例：描述模糊，缺乏约束 tools = [ { "type": "function", "function": { "name": "get_weather", "description": "获取天气信息", "parameters": { "type": "object", "properties": { "location": {"type": "string"} } } } } ] # 改进后的示例：描述清晰，参数有详细说明和约束 tools = [ { "type": "function", "function": { "name": "get_current_weather", "description": "获取指定城市当前的具体天气状况，包括温度、体感温度、天气现象（晴、雨、雪等）、湿度和风速风向。请确保城市名称是明确且存在的。", "parameters": { "type": "object", "properties": { "location": { "type": "string", "description": "城市名称，必须是完整的、公认的城市名，例如‘北京市’、‘New York’。不要使用缩写、别名或模糊指代。" }, "unit": { "type": "string", "enum": ["celsius", "fahrenheit"], "description": "温度单位，默认为‘celsius’（摄氏度）。", "default": "celsius" } }, "required": ["location"], # 新增：严格模式，禁止模型传入未定义的参数 "additionalProperties": False }, # 非OpenAI标准，但可作为元信息传递给执行层 "strict": True } } ]关键点：
- 描述要具体：说明工具做什么、输入是什么、输出是什么。避免“获取信息”这种泛泛之谈。
- 参数描述是重点：告诉模型需要什么样的输入。比如
location，要说明是“完整的城市名”，这能显著减少模型传递“北京天气咋样”这种口语化字符串的概率。 - 使用
enum和default：对于有限选项的参数，用enum明确列出；提供合理的默认值，简化模型的决策。 - 设置
additionalProperties: False：这能防止模型“幻觉”出你工具函数不支持的参数，减少调用错误。
2.2 调用执行与错误处理：设计容错链路
模型返回的调用请求，需要被安全地执行。这里绝不能简单地try-except然后抛出一个错误消息给模型。
import logging from typing import Any, Dict, Callable from pydantic import BaseModel, ValidationError logger = logging.getLogger(__name__) class ToolExecutionResult(BaseModel): """工具执行结果的标准化封装""" success: bool data: Any = None error_message: str = "" tool_name: str = "" raw_arguments: Dict = None class RobustToolExecutor: def __init__(self, tools_registry: Dict[str, Dict]): self.tools = tools_registry # 工具注册表，包含函数和schema def execute(self, tool_call: Dict) -> ToolExecutionResult: """ 执行单个工具调用，包含完整的验证和错误处理。 tool_call 结构: {"name": "get_weather", "arguments": {"location": "Beijing"}} """ tool_name = tool_call.get("name") if not tool_name or tool_name not in self.tools: return ToolExecutionResult( success=False, error_message=f"工具 '{tool_name}' 未注册或不可用。", tool_name=tool_name or "unknown" ) tool_info = self.tools[tool_name] func = tool_info["function"] schema = tool_info["schema"] # 对应的JSON Schema # 1. 参数验证与类型转换 raw_args = tool_call.get("arguments", {}) try: # 使用Pydantic模型进行严格验证和类型转换 validated_args = self._validate_and_convert_args(schema, raw_args) except ValidationError as e: logger.warning(f"工具 {tool_name} 参数验证失败: {e}") # 给模型反馈具体的错误字段，帮助其修正 error_detail = "; ".join([f"{err['loc'][0]}: {err['msg']}" for err in e.errors()]) return ToolExecutionResult( success=False, error_message=f"参数错误: {error_detail}。请检查参数格式和类型。", tool_name=tool_name, raw_arguments=raw_args ) # 2. 安全执行 try: result_data = func(**validated_args) except Exception as e: logger.exception(f"工具 {tool_name} 执行时发生异常") # 区分业务逻辑错误和系统错误，反馈给模型的信息详略程度可以不同 # 例如，网络超时和“城市不存在”是两种不同的错误 user_friendly_msg = self._make_error_user_friendly(e) return ToolExecutionResult( success=False, error_message=f"执行失败: {user_friendly_msg}", tool_name=tool_name, raw_arguments=validated_args ) # 3. 结果标准化 return ToolExecutionResult( success=True, data=result_data, tool_name=tool_name, raw_arguments=validated_args ) def _validate_and_convert_args(self, schema: Dict, raw_args: Dict) -> Dict: """使用Pydantic根据schema验证和转换参数""" # 此处简化，实际可根据schema动态生成Pydantic模型 # 核心是进行类型转换，比如把字符串"123"转成整数123 validated = {} for key, prop_schema in schema["properties"].items(): if key in raw_args: value = raw_args[key] expected_type = prop_schema.get("type") # 执行简单的类型转换逻辑 if expected_type == "integer" and isinstance(value, str): if value.isdigit(): validated[key] = int(value) else: raise ValidationError(f"字段 '{key}' 需要整数，但收到 '{value}'") elif expected_type == "boolean" and isinstance(value, str): if value.lower() in ["true", "1"]: validated[key] = True elif value.lower() in ["false", "0"]: validated[key] = False else: raise ValidationError(f"字段 '{key}' 需要布尔值，但收到 '{value}'") else: validated[key] = value elif key in schema.get("required", []): raise ValidationError(f"缺少必需参数: '{key}'") return validated def _make_error_user_friendly(self, exception: Exception) -> str: """将系统异常转换为对模型友好的错误信息""" # 这里可以根据不同的异常类型返回不同的提示 if isinstance(exception, TimeoutError): return "请求外部服务超时，请稍后重试。" elif isinstance(exception, ValueError): return f"输入数据有误: {str(exception)}" else: # 生产环境中，避免将内部错误堆栈直接暴露给模型 return "工具执行过程中发生意外错误。"设计要点：
- 标准化结果封装：使用
ToolExecutionResult这样的类统一返回格式，包含成功状态、数据、错误信息和原始参数。这为后续的日志、监控和记忆存储提供了便利。 - 分层错误处理：
- 工具不存在：直接返回错误，避免调用未定义函数。
- 参数验证失败：使用如Pydantic的库进行严格验证，并给出字段级的错误反馈（例如“location: 必须是字符串类型”），这比单纯的“参数错误”更能帮助模型在下一次调用中修正。
- 执行时异常：捕获所有异常，并区分是业务逻辑错误（如“城市不存在”）还是系统错误（如“网络超时”）。反馈给模型的信息应有助于其进行后续决策（例如，网络超时可以建议重试，城市不存在则需要用户澄清）。
- 参数类型转换：大模型输出的参数永远是字符串。你的工具函数可能期望整数、布尔值或日期对象。必须在执行前进行安全的类型转换，这是避免运行时类型错误的关键。
- 日志与监控：在每个关键节点（调用开始、验证失败、执行异常、执行成功）记录结构化的日志。这对于生产环境调试和性能监控至关重要。
2.3 流式、并行与编排：应对复杂任务
当Agent需要连续调用多个工具，或者一个工具调用依赖前一个工具的结果时，简单的串行循环就不够了。
class ToolOrchestrator: def __init__(self, executor: RobustToolExecutor): self.executor = executor self.call_history = [] # 用于记忆和复盘 def run_agent_loop(self, initial_query: str, max_turns: int = 10): """运行一个多轮工具调用的Agent循环""" messages = [{"role": "user", "content": initial_query}] for turn in range(max_turns): # 1. 调用大模型，获取包含工具调用的响应 llm_response = call_llm(messages, tools=self.executor.tools) messages.append(llm_response) # 检查是否需要调用工具 tool_calls = llm_response.get("tool_calls") if not tool_calls: # 没有工具调用，直接返回最终答案 final_answer = llm_response["content"] self.call_history.append({"turn": turn, "action": "final_answer", "content": final_answer}) return final_answer # 2. 并行执行多个工具调用（如果模型同时返回了多个） parallel_results = [] for tc in tool_calls: result = self.executor.execute(tc) self.call_history.append({ "turn": turn, "tool": tc["name"], "args": tc.get("arguments"), "result": result.dict() if hasattr(result, 'dict') else result }) parallel_results.append(result) # 3. 将工具执行结果整理成消息，反馈给模型 tool_messages = [] for result in parallel_results: if result.success: # 成功：将结果以结构化方式返回 # 注意：直接返回原始数据可能过于冗长，可以考虑总结或提取关键信息 tool_messages.append({ "role": "tool", "content": f"工具 {result.tool_name} 执行成功。结果: {result.data}", "tool_call_id": tc.get("id") # 关联具体的tool call }) else: # 失败：提供清晰的错误信息，指导模型下一步操作 tool_messages.append({ "role": "tool", "content": f"工具 {result.tool_name} 执行失败。原因: {result.error_message}。请检查输入或尝试其他方法。", "tool_call_id": tc.get("id") }) messages.extend(tool_messages) # 4. 检查终止条件（例如，所有必要工具已成功，或出现不可恢复错误） if self._should_stop(parallel_results): logger.info(f"Agent在 {turn+1} 轮后停止。") break # 循环结束（可能达到最大轮数） final_llm_response = call_llm(messages) # 最后一次，不提供工具，让模型总结 return final_llm_response["content"] def _should_stop(self, results: List[ToolExecutionResult]) -> bool: """基于本轮工具执行结果判断是否应停止Agent循环""" # 策略1：所有计划中的工具都成功执行完毕 all_success = all(r.success for r in results) # 策略2：出现了关键路径上的失败，且无法通过重试或替代方案解决 critical_failure = any(r.error_message and "无法找到" in r.error_message for r in results) return all_success or critical_failure编排逻辑的核心：
- 消息历史管理：严格遵循OpenAI的消息格式（
user,assistant,tool），确保模型能正确理解上下文。每次工具执行结果都必须以tool角色消息追加。 - 并行执行优化：如果模型一次性建议了多个不依赖的工具调用（例如，同时查询天气和股票），应该并行执行它们以降低延迟。
- 循环终止策略：必须设计明确的停止条件，防止Agent陷入无限循环或无效尝试。常见策略包括：成功完成所有必要步骤、达到最大轮数限制、遇到无法修复的关键错误、模型自己决定输出最终答案。
- 结果处理与摘要：直接将庞大的JSON结果扔回给模型会浪费token且可能干扰判断。考虑对工具返回的数据进行摘要或提取关键字段。例如，数据库查询结果可能返回10行，你只需要把最相关的3行摘要给模型。
注意：工具调用链路的稳定性，一半靠清晰的工具定义，另一半靠严谨的执行层容错。千万不要假设模型返回的调用请求总是正确和完整的。
3. 记忆架构设计：让Agent拥有“上下文”与“经验”
记忆是AI Agent的“大脑皮层”，决定了其连贯性和智能水平。生产级的记忆系统不能只是一个简单的对话历史列表。它需要分层、结构化、可持久化，并且能高效地被检索和利用。
3.1 记忆的层次：工作记忆、会话记忆与长期记忆
我将记忆分为三个层次，这对应了人类不同的记忆机制：
工作记忆 (Working Memory)：相当于Agent的“桌面”。它存储当前任务执行过程中产生的临时、高活跃度的信息。例如，在多轮工具调用中，上一步的查询结果、当前步骤的中间状态、用户的即时反馈。它的特点是容量小、存取快、生命周期短（通常与一个任务或会话绑定）。在代码中，这通常体现为程序运行时的变量或一个短期的缓存对象。
会话记忆 (Session Memory / Conversation Memory)：存储单次对话会话的完整历史。这是最常见的记忆形式，即整个
messages数组。它保证了对话的连贯性。但随着对话拉长，直接使用全部历史会导致token消耗剧增和模型注意力分散。因此，需要对会话记忆进行摘要或选择性保留。长期记忆 (Long-term Memory)：Agent的“知识库”或“经验库”。它存储跨越不同会话的重要事实、用户偏好、学到的知识、历史决策与结果。例如，用户说过“我对花生过敏”，这个信息应该被存入长期记忆，并在未来相关的会话中被检索出来。长期记忆需要持久化存储（数据库、向量库），并配备高效的检索机制。
3.2 会话记忆的优化：摘要与关键信息提取
直接存储所有原始消息是不可持续的。以下是一个结合了ConversationSummaryBufferMemory思想和关键信息提取的混合策略实现。
from typing import List, Dict, Any from datetime import datetime import hashlib class HybridConversationMemory: """ 混合会话记忆：结合完整最近消息 + 历史摘要 + 关键实体存储。 """ def __init__(self, llm_client, max_recent_tokens=1000, summary_interval=10): self.llm = llm_client self.max_recent_tokens = max_recent_tokens # 保留的最近消息的token上限 self.summary_interval = summary_interval # 每N轮对话生成一次摘要 self.recent_messages: List[Dict] = [] # 最近的原始消息 self.summaries: List[str] = [] # 历史摘要列表 self.key_entities: Dict[str, Any] = {} # 提取出的关键实体（如用户偏好、决策点） self.message_count = 0 def add_message(self, message: Dict): """添加一条新消息到记忆""" self.recent_messages.append(message) self.message_count += 1 # 检查是否需要进行摘要 if self.message_count % self.summary_interval == 0: self._create_summary() # 定期清理recent_messages，防止token超限 self._prune_recent_messages() # 尝试从消息中提取关键实体（例如，用户声明了偏好） if message['role'] == 'user': self._extract_key_entities(message['content']) def get_context_for_llm(self) -> List[Dict]: """组装用于提交给LLM的上下文消息""" context_messages = [] # 1. 添加历史摘要（如果有） if self.summaries: summary_text = "\n\n".join([f"先前对话摘要({i+1}): {s}" for i, s in enumerate(self.summaries)]) context_messages.append({ "role": "system", "content": f"以下是本次对话之前的历史摘要，供你参考：\n{summary_text}" }) # 2. 添加关键实体提示（如果有） if self.key_entities: entities_text = ", ".join([f"{k}: {v}" for k, v in self.key_entities.items()]) context_messages.append({ "role": "system", "content": f"请注意以下在本对话中已确认的关键信息：{entities_text}" }) # 3. 添加最近的原始消息（保证最新交互的完整性） context_messages.extend(self.recent_messages[-self._estimate_recent_message_count():]) # 控制数量 return context_messages def _create_summary(self): """调用LLM对最近的对话内容生成摘要""" if len(self.recent_messages) < 3: # 消息太少时不摘要 return # 准备要摘要的文本 text_to_summarize = "\n".join([f"{m['role']}: {m['content']}" for m in self.recent_messages]) prompt = f""" 请将以下对话内容浓缩成一个简洁的摘要，保留关于事实、用户需求、已做出的决策和待办事项的关键信息。 摘要语言请使用中文。 对话内容： {text_to_summarize} 摘要： """ try: response = self.llm.chat_completion([{"role": "user", "content": prompt}]) new_summary = response.choices[0].message.content.strip() self.summaries.append(new_summary) # 生成摘要后，可以清空或部分清空recent_messages，因为其信息已浓缩 # self.recent_messages = self.recent_messages[-5:] # 可选：只保留最后几条 logger.info(f"已生成新的对话摘要: {new_summary[:100]}...") except Exception as e: logger.error(f"生成对话摘要失败: {e}") def _extract_key_entities(self, user_input: str): """从用户输入中提取关键实体信息（简化示例）""" # 这里可以使用规则、NER模型或调用一个小型LLM来提取 # 例如，简单规则匹配 import re allergy_pattern = r"(?:我对|我)对(.+?)过敏" match = re.search(allergy_pattern, user_input) if match: allergen = match.group(1).strip() self.key_entities['user_allergy'] = allergen logger.info(f"提取到关键实体[用户过敏物]: {allergen}") def _prune_recent_messages(self): """估算token数并修剪最近消息列表""" # 简化实现：按条数修剪。实际应使用tiktoken等库估算token。 max_recent_messages = 20 # 假设一个保守的条数上限 if len(self.recent_messages) > max_recent_messages: # 移除最旧的消息，但至少保留最后5条以保证连贯性 remove_count = len(self.recent_messages) - max_recent_messages keep_from = max(5, remove_count) # 确保不移除过于新的消息 self.recent_messages = self.recent_messages[keep_from:] def _estimate_recent_message_count(self) -> int: """估算返回多少条最近消息合适""" # 更复杂的实现可以动态计算token return min(10, len(self.recent_messages)) # 默认返回最多10条最新消息这个混合策略的优势：
- 控制Token消耗：用摘要代表遥远的过去，只保留最近的原始消息，有效控制了上下文长度。
- 保留关键信息：独立存储
key_entities，确保重要事实（如过敏信息）不会被摘要过程稀释或遗忘，并能被精准提示给模型。 - 维持连贯性：最近的原始消息保证了最新几轮对话的完整细节，使模型能理解细微的指代和即时意图。
3.3 长期记忆的实现：向量检索与结构化存储
长期记忆的核心是“存得进去，找得出来”。对于非结构化文本知识（如产品文档、历史对话精华），向量检索是首选。对于结构化信息（如用户档案、系统配置），则用传统数据库。
向量检索长期记忆示例：
import chromadb # 或其他向量数据库客户端 from chromadb.config import Settings from sentence_transformers import SentenceTransformer import uuid class VectorLongTermMemory: def __init__(self, persist_dir: str = "./chroma_db"): self.client = chromadb.PersistentClient(path=persist_dir, settings=Settings(allow_reset=True)) # 使用一个轻量且高效的嵌入模型，例如 all-MiniLM-L6-v2 self.embedder = SentenceTransformer('all-MiniLM-L6-v2') self.collection = self.client.get_or_create_collection(name="agent_long_term_memory") def store(self, text: str, metadata: Dict[str, Any]): """存储一段文本到长期记忆""" embedding = self.embedder.encode(text).tolist() doc_id = str(uuid.uuid4()) self.collection.add( documents=[text], embeddings=[embedding], metadatas=[metadata], # 可以存储来源、时间、类型、重要性分数等 ids=[doc_id] ) return doc_id def search(self, query: str, top_k: int = 3, filter_conditions: Dict = None) -> List[Dict]: """从长期记忆中检索与查询最相关的片段""" query_embedding = self.embedder.encode(query).tolist() results = self.collection.query( query_embeddings=[query_embedding], n_results=top_k, where=filter_conditions # 例如，只检索某种类型的记忆 ) retrieved = [] if results['documents']: for i in range(len(results['documents'][0])): retrieved.append({ 'text': results['documents'][0][i], 'metadata': results['metadatas'][0][i], 'distance': results['distances'][0][i] }) return retrieved def retrieve_for_context(self, current_query: str, user_id: str = None) -> str: """为当前查询检索相关长期记忆，并格式化为上下文字符串""" filter_ = {"user_id": user_id} if user_id else None memories = self.search(current_query, top_k=2, filter_conditions=filter_) if not memories: return "" context_parts = ["以下是从长期记忆中检索到的相关信息："] for mem in memories: # 可以基于距离（相关性）进行过滤或加权 if mem['distance'] < 0.4: # 设定一个相关性阈值 context_parts.append(f"- {mem['text']} (来源: {mem['metadata'].get('source', 'N/A')})") return "\n".join(context_parts)使用时机：
- 存储：当会话结束时，将本次对话的最终摘要或学到的关键知识存入向量记忆。
- 检索：在新会话开始时，或会话中用户提到相关话题时，用当前查询去向量记忆中搜索，将结果作为系统提示的一部分注入上下文。
结构化长期记忆示例： 对于用户偏好、设置等，直接用SQL/NoSQL数据库。
# 伪代码示例 class StructuredUserMemory: def get_user_preference(self, user_id: str) -> Dict: # 从数据库读取用户偏好 return db.query("SELECT * FROM user_preferences WHERE user_id = ?", user_id) def update_user_fact(self, user_id: str, fact_key: str, fact_value: str): # 更新或插入用户相关事实 db.upsert("user_facts", {"user_id": user_id, "key": fact_key, "value": fact_value})记忆的融合：在实际的Agent循环中，get_context_for_llm方法需要整合工作记忆、会话记忆和长期记忆。
def build_full_context(hybrid_memory: HybridConversationMemory, vector_memory: VectorLongTermMemory, user_id: str, current_query: str) -> List[Dict]: """构建完整的上下文消息列表""" messages = [] # 1. 系统提示词（包含角色、核心指令） system_prompt = """你是一个专业的助手。请根据对话历史和以下提供的相关信息来回答问题或执行任务。""" messages.append({"role": "system", "content": system_prompt}) # 2. 注入长期记忆（向量检索结果） long_term_context = vector_memory.retrieve_for_context(current_query, user_id) if long_term_context: messages.append({"role": "system", "content": long_term_context}) # 3. 注入会话记忆（摘要+最近消息） messages.extend(hybrid_memory.get_context_for_llm()) # 4. 加入当前用户查询 messages.append({"role": "user", "content": current_query}) return messages注意：记忆不是越多越好。无关信息的注入会形成“噪声”，干扰模型判断。必须设计精密的检索和过滤策略，确保提供给模型的记忆是高相关、高价值的。
4. 实战集成：构建一个完整的订单查询Agent
让我们把工具调用和记忆架构组合起来，设计一个简单的“订单查询助手”Agent。这个Agent能调用工具查询订单状态，并能记住用户偏好的查询方式（例如，总是优先显示物流信息）。
步骤1：定义工具
order_tools = [ { "type": "function", "function": { "name": "get_order_status", "description": "根据订单号查询订单的当前状态、支付情况、物流信息。", "parameters": { "type": "object", "properties": { "order_id": {"type": "string", "description": "完整的订单编号"}, "include_logistics": {"type": "boolean", "description": "是否包含详细的物流跟踪信息", "default": True} }, "required": ["order_id"], "additionalProperties": False } } } ] # 模拟的工具实现 def mock_get_order_status(order_id: str, include_logistics: bool = True) -> Dict: # 模拟数据库查询 return { "order_id": order_id, "status": "已发货", "payment_status": "已支付", "logistics": { "company": "某快递", "tracking_number": "SF1234567890", "latest_update": "已到达上海转运中心" } if include_logistics else None }步骤2：初始化记忆与执行器
# 初始化记忆系统 conversation_memory = HybridConversationMemory(llm_client=llm_client) long_term_memory = VectorLongTermMemory() # 初始化工具执行器 tool_registry = { "get_order_status": { "function": mock_get_order_status, "schema": order_tools[0]["function"]["parameters"] } } executor = RobustToolExecutor(tool_registry) orchestrator = ToolOrchestrator(executor)步骤3：模拟对话流程
user_id = "user_123" # 第一轮对话 query_1 = "帮我查一下订单 ABC-20240520-001 到哪里了。" # 构建上下文（假设这是新会话，长期记忆暂无相关内容） context_1 = build_full_context(conversation_memory, long_term_memory, user_id, query_1) # Agent决策并调用工具 # ... (调用LLM，得到工具调用请求) tool_call = {"name": "get_order_status", "arguments": {"order_id": "ABC-20240520-001"}} result = executor.execute(tool_call) # 将结果和原始消息存入会话记忆 conversation_memory.add_message({"role": "user", "content": query_1}) conversation_memory.add_message({"role": "tool", "content": str(result.data)}) # LLM生成最终回复给用户: “您的订单已发货，最新物流信息：已到达上海转运中心，快递单号SF1234567890。” # 用户表达偏好 query_2 = “以后查订单，直接告诉我到哪了就行，不用再说支付状态。” conversation_memory.add_message({"role": "user", "content": query_2}) # 从这句话中提取关键实体（用户偏好） # 通过 `_extract_key_entities` 会提取到 key_entities['preferred_order_detail'] = 'logistics' # 并且，在会话结束时，可以将“用户偏好物流信息”这个知识存入长期记忆 memory_text = “用户 user_123 在查询订单状态时，明确表示更关注物流信息，而非支付状态。” long_term_memory.store(memory_text, metadata={"user_id": user_id, "type": "preference", "key": "order_detail_priority"}) # 几天后，新一轮会话 new_query = “订单 DEF-20240525-002 什么情况了？” # 构建上下文时，长期记忆检索会匹配到“物流信息”偏好 # 检索到的记忆会作为系统提示注入，影响Agent行为 # 例如，系统提示中会包含：“注意：该用户偏好优先查看订单物流信息。” # Agent在调用 `get_order_status` 工具时，可能会更倾向于设置 `include_logistics=True`，并在组织回复时重点强调物流部分。步骤4：设计智能体主循环
def agent_loop(user_input: str, user_id: str, memory_systems: dict, max_turns=6): """简化的智能体主循环""" conv_memory = memory_systems['conversation'] long_memory = memory_systems['long_term'] # 1. 检索长期记忆，构建完整上下文 full_context = build_full_context(conv_memory, long_memory, user_id, user_input) # 2. 调用LLM，获取可能包含工具调用的响应 llm_response = call_llm_with_tools(full_context, tools=order_tools) # 3. 处理工具调用（使用之前的Orchestrator） final_answer = orchestrator.run_agent_loop_with_memory( initial_messages=full_context, llm_initial_response=llm_response, memory=conv_memory ) # 4. 更新记忆 conv_memory.add_message({"role": "user", "content": user_input}) conv_memory.add_message({"role": "assistant", "content": final_answer}) # 5. （可选）在会话合适节点，将重要信息沉淀到长期记忆 if should_save_to_long_term(user_input, final_answer): summary_for_storage = create_memory_snippet(conv_memory) long_memory.store(summary_for_storage, metadata={"user_id": user_id, "session_id": session_id}) return final_answer通过这个例子，你可以看到工具调用和记忆系统是如何协同工作的：记忆系统为Agent提供了个性化的上下文（用户偏好），影响了Agent的决策（调用工具时的参数倾向和回复组织方式）；而工具调用的结果又反过来丰富了记忆（订单查询的结果可以作为对话历史的一部分被摘要或存储）。
