引言AI Agent 的工程化挑战在当今的 AI 应用浪潮中能够自主调用工具、与环境交互并完成复杂任务的 AI Agent 正成为技术焦点。然而从简单的提示词工程到构建一个稳定、可扩展、具备长期记忆与多工具协同能力的生产级 Agent 系统其间横亘着巨大的工程鸿沟。本文将深入探讨一个工程化的 AI Agent 系统架构设计聚焦于多工具的动态调度与协同以及记忆机制的模块化实现旨在为开发者提供一套可落地的架构蓝图与核心代码实践。1. 核心架构概览一个完整的 AI Agent 系统通常由以下几个核心模块组成是否“用户输入/环境事件”“推理与规划引擎”“需要工具调用”“工具执行器”“工具A”“工具B”“工具N”“结果整合与格式化”“记忆系统”“短期记忆”“长期记忆”“生成最终响应”“输出给用户”架构分层解析控制层Orchestration Layer包含推理与规划引擎是 Agent 的“大脑”负责解析目标、制定计划、决定下一步行动思考、调用工具或直接回答。执行层Execution Layer即工具执行器与具体的工具集合是 Agent 的“手脚”负责与外部世界API、数据库、文件系统等交互。记忆层Memory Layer负责存储和检索交互历史、知识、用户偏好等是 Agent 的“经验”分为短期会话上下文和长期向量数据库、图数据库等记忆。接口层Interface Layer处理与用户或其它系统的输入输出。2. 多工具协同动态调度与编排工具调用是 Agent 扩展能力边界的关键。简单的“if-else”式工具选择无法应对复杂场景我们需要一个动态、可扩展的调度系统。2.1 工具的统一抽象与注册首先我们需要一个统一的工具描述格式通常包含名称、描述、参数模式JSON Schema和执行函数。fromtypingimportAny,Dict,CallablefrompydanticimportBaseModel,FieldimportinspectclassTool(BaseModel):工具的统一抽象name:strdescription:strargs_schema:type[BaseModel]func:CallableclassConfig:arbitrary_types_allowedTrueclassToolRegistry:工具注册中心def__init__(self):self._tools:Dict[str,Tool]{}defregister(self,tool:Tool):self._tools[tool.name]tooldefget_tool(self,name:str)-Tool:returnself._tools.get(name)deflist_tools(self)-list[Tool]:returnlist(self._tools.values())# 示例定义一个搜索工具classSearchInput(BaseModel):query:strField(description搜索查询词)defsearch_web(query:str)-str:# 模拟搜索returnf关于{query}的搜索结果...search_toolTool(nameweb_search,description在互联网上搜索信息,args_schemaSearchInput,funcsearch_web)registryToolRegistry()registry.register(search_tool)2.2 基于语义的工具发现与选择当 Agent 需要决定使用哪个工具时简单的关键词匹配不够鲁棒。我们可以利用嵌入Embedding模型进行语义匹配。importnumpyasnpfromsentence_transformersimportSentenceTransformerclassSemanticToolSelector:基于语义的工具选择器def__init__(self,registry:ToolRegistry,model_nameall-MiniLM-L6-v2):self.registryregistry self.modelSentenceTransformer(model_name)self._build_tool_index()def_build_tool_index(self):为所有工具的描述构建向量索引self.tool_listself.registry.list_tools()# 将工具名称和描述组合成文本tool_texts[f{tool.name}:{tool.description}fortoolinself.tool_list]self.tool_embeddingsself.model.encode(tool_texts,normalize_embeddingsTrue)defselect_tools(self,query:str,top_k:int3)-list[Tool]:根据查询语义选择最相关的工具query_embeddingself.model.encode([query],normalize_embeddingsTrue)[0]# 计算余弦相似度similaritiesnp.dot(self.tool_embeddings,query_embedding)top_indicesnp.argsort(similarities)[-top_k:][::-1]return[self.tool_list[i]foriintop_indices]# 使用示例selectorSemanticToolSelector(registry)user_request我想知道今天的天气和新闻头条relevant_toolsselector.select_tools(user_request,top_k2)print([tool.namefortoolinrelevant_tools])# 可能输出[web_search, get_weather]2.3 工具链Workflow与条件执行复杂任务往往需要按顺序或条件执行多个工具构成一个工作流。fromtypingimportList,OptionalclassWorkflowStep(BaseModel):tool_name:strinput_expression:str# 例如{{user_query}} 或 {{previous_step.output}}classSequentialWorkflow:顺序执行的工作流def__init__(self,steps:List[WorkflowStep],registry:ToolRegistry):self.stepssteps self.registryregistry self.context{}defexecute(self,initial_context:dict)-dict:self.context.update(initial_context)forstepinself.steps:# 1. 解析输入表达式tool_inputself._evaluate_expression(step.input_expression)# 2. 获取工具并执行toolself.registry.get_tool(step.tool_name)ifnottool:raiseValueError(f工具未找到:{step.tool_name})# 此处应进行参数验证与构建resulttool.func(**tool_input)# 3. 将结果存入上下文供后续步骤使用self.context[f{step.tool_name}_output]resultreturnself.contextdef_evaluate_expression(self,expr:str):# 简单的模板渲染生产环境需更安全的方式try:returneval(expr,{__builtins__:{}},self.context)except:returnexpr3. 记忆机制从短期会话到长期知识记忆是 Agent 实现连贯性和个性化的核心。我们需要一个分层的记忆系统。3.1 短期记忆对话上下文管理短期记忆通常由大模型的上下文窗口来维护但我们需要更精细的管理策略如摘要、关键信息提取等。fromdatetimeimportdatetimefrompydanticimportBaseModelclassConversationTurn(BaseModel):role:str# user, assistant, system, toolcontent:strtimestamp:datetimeField(default_factorydatetime.now)classShortTermMemory:短期记忆管理最近对话轮次def__init__(self,max_turns:int20):self.max_turnsmax_turns self.conversation:List[ConversationTurn][]defadd_turn(self,turn:ConversationTurn):self.conversation.append(turn)# 超出限制时可触发摘要或淘汰最旧记录iflen(self.conversation)self.max_turns:self._compress_memory()def_compress_memory(self):内存压缩策略例如将最早的若干轮次总结为一句话# 简化实现移除最旧的一些记录remove_countlen(self.conversation)-self.max_turnsifremove_count0:self.conversationself.conversation[remove_count:]defget_recent_context(self,last_n:int10)-List[ConversationTurn]:returnself.conversation[-last_n:]defto_chat_messages(self)-List[dict]:转换为LLM所需的聊天消息格式return[{role:turn.role,content:turn.content}forturninself.conversation]3.2 长期记忆向量化存储与检索长期记忆用于存储超越上下文窗口的重要信息通常借助向量数据库实现。# 假设使用 ChromaDBimportchromadbfromchromadb.configimportSettingsfromsentence_transformersimportSentenceTransformerclassLongTermMemory:基于向量数据库的长期记忆def__init__(self,persist_directory:str./chroma_db):self.clientchromadb.PersistentClient(pathpersist_directory)self.collectionself.client.get_or_create_collection(nameagent_memory)self.embedderSentenceTransformer(all-MiniLM-L6-v2)defstore(self,text:str,metadata:dictNone):存储一段文本到长期记忆embeddingself.embedder.encode(text).tolist()doc_idfdoc_{datetime.now().timestamp()}self.collection.add(embeddings[embedding],documents[text],metadatas[metadata]ifmetadataelse[{}],ids[doc_id])defretrieve(self,query:str,top_k:int5)-List[str]:根据查询检索最相关的记忆片段query_embeddingself.embedder.encode(query).tolist()resultsself.collection.query(query_embeddings[query_embedding],n_resultstop_k)returnresults[documents][0]ifresults[documents]else[]# 使用示例long_memoryLongTermMemory()# 存储一次重要的用户偏好long_memory.store(用户小明喜欢用Python进行数据分析常用pandas和matplotlib。,metadata{user:xiaoming,type:preference})# 当用户再次提到数据分析时可以检索出来relevant_memorieslong_memory.retrieve(用户想分析数据,top_k3)3.3 记忆的融合与优先级在实际决策时Agent 需要综合短期记忆和长期记忆。classMemoryFusion:记忆融合器决定在给定上下文中使用哪些记忆def__init__(self,short_memory:ShortTermMemory,long_memory:LongTermMemory):self.short_memoryshort_memory self.long_memorylong_memorydefget_context_for_llm(self,user_query:str)-str:为LLM构建增强的上下文提示# 1. 获取短期记忆最近对话recent_chatself.short_memory.get_recent_context(last_n5)short_term_context\n.join([f{turn.role}:{turn.content}forturninrecent_chat])# 2. 从长期记忆中检索相关片段long_term_memoriesself.long_memory.retrieve(user_query,top_k3)long_term_context\n.join([f[记忆]{mem}formeminlong_term_memories])iflong_term_memorieselse# 3. 融合并格式化enhanced_contextf 近期对话{short_term_context}相关长期记忆{long_term_context}当前用户请求{user_query}returnenhanced_context4. 工程化实践构建一个可复用的 Agent 内核将上述模块组合起来我们可以构建一个核心的 Agent 执行引擎。classAgentCore:AI Agent 核心执行引擎def__init__(self,llm_client,tool_registry:ToolRegistry,memory_fusion:MemoryFusion):self.llmllm_client self.toolstool_registry self.memorymemory_fusion self.short_memorymemory_fusion.short_memorydefrun(self,user_input:str)-str:# 1. 更新短期记忆self.short_memory.add_turn(ConversationTurn(roleuser,contentuser_input))# 2. 获取融合了记忆的增强上下文contextself.memory.get_context_for_llm(user_input)# 3. 规划与决策简化版让LLM决定是否调用工具# 此处应使用更复杂的提示词或规划算法promptf{context}你是一个AI助手可以调用工具。请分析是否需要调用工具来回答用户。 可用工具{[t.namefortinself.tools.list_tools()]}如果需要请以JSON格式回复包含 action: tool_call, tool_name, arguments。 如果不需要请直接回复答案。 llm_responseself.llm.chat(prompt)# 4. 解析并执行工具调用此处为简化逻辑ifself._needs_tool_call(llm_response):tool_name,argsself._parse_tool_call(llm_response)toolself.tools.get_tool(tool_name)iftool:resulttool.func(**args)# 将工具调用和结果也存入记忆self.short_memory.add_turn(ConversationTurn(roletool,contentf调用{tool_name}结果{result}))# 基于工具结果生成最终回答final_answerself.llm.chat(f工具调用结果{result}。请基于此生成对用户的回答。)else:final_answerf抱歉无法调用工具{tool_name}。else:final_answerllm_response# 5. 将助手回复存入记忆并返回self.short_memory.add_turn(ConversationTurn(roleassistant,contentfinal_answer))returnfinal_answerdef_needs_tool_call(self,response:str)-bool:# 简化实现实际应解析LLM的特定输出格式returntool_callinresponsedef_parse_tool_call(self,response:str):# 简化实现实际应解析JSON# 这里假设response是JSON字符串importjson datajson.loads(response)returndata.get(tool_name),data.get(arguments,{})5. 总结与展望本文探讨了工程化 AI Agent 系统的两个核心支柱多工具协同与记忆机制。通过工具注册中心、语义选择器和工作流编排我们实现了灵活可扩展的工具调用能力。通过分层记忆系统短期对话管理、长期向量存储和记忆融合器我们赋予了 Agent 连贯的上下文意识和个性化能力。进一步的工程化考虑可观测性Observability为 Agent 的每一步决策、工具调用和记忆检索添加日志与追踪便于调试和优化。弹性与容错工具调用可能失败需要重试、降级或备用方案。安全性严格校验工具输入防止注入攻击对工具的执行权限进行管控。评估与持续学习建立评估体系利用用户反馈或自动评估来优化 Agent 的决策策略和记忆存储策略。构建强大的 AI Agent 并非一蹴而就而是一个需要精心设计架构并持续迭代的工程过程。希望本文提供的设计模式与代码片段能为你搭建自己的智能体系统提供坚实的起点。