构建生产级AI Agent：工具调用与记忆架构的工程化实践

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( successFalse, error_messagef工具 {tool_name} 未注册或不可用。, tool_nametool_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( successFalse, error_messagef参数错误: {error_detail}。请检查参数格式和类型。, tool_nametool_name, raw_argumentsraw_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( successFalse, error_messagef执行失败: {user_friendly_msg}, tool_nametool_name, raw_argumentsvalidated_args ) # 3. 结果标准化 return ToolExecutionResult( successTrue, dataresult_data, tool_nametool_name, raw_argumentsvalidated_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, toolsself.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(fAgent在 {turn1} 轮后停止。) 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_tokens1000, summary_interval10): 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先前对话摘要({i1}): {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(pathpersist_dir, settingsSettings(allow_resetTrue)) # 使用一个轻量且高效的嵌入模型例如 all-MiniLM-L6-v2 self.embedder SentenceTransformer(all-MiniLM-L6-v2) self.collection self.client.get_or_create_collection(nameagent_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_resultstop_k, wherefilter_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_k2, filter_conditionsfilter_) 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_clientllm_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_logisticsTrue并在组织回复时重点强调物流部分。步骤4设计智能体主循环def agent_loop(user_input: str, user_id: str, memory_systems: dict, max_turns6): 简化的智能体主循环 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, toolsorder_tools) # 3. 处理工具调用使用之前的Orchestrator final_answer orchestrator.run_agent_loop_with_memory( initial_messagesfull_context, llm_initial_responsellm_response, memoryconv_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的决策调用工具时的参数倾向和回复组织方式而工具调用的结果又反过来丰富了记忆订单查询的结果可以作为对话历史的一部分被摘要或存储。