在业务迭代中引入AI Agent时我们常常面临一个核心矛盾Agent的“聪明”与“可靠”难以兼得。它或许能理解复杂指令但在调用工具、处理长上下文、拆解多步骤任务以及与现有系统稳定交互时却频频“掉链子”导致落地困难。近期美团技术团队发布了一份基于外卖、酒店、打车等核心业务场景的Agent实践手册其价值正在于它没有空谈理论而是直面这些一线工程挑战提供了一套经过大规模流量验证的解决方案。本文将深入解读这份手册的精髓并结合行业通用实践拆解如何构建一个不仅能“听懂话”更能“办成事”的实用型AI Agent。内容涵盖从核心架构设计、上下文处理策略、任务自动拆解到与后端系统安全集成的全流程。无论你是正在探索Agent落地的架构师还是希望提升智能体可靠性的开发者都能从中获得可直接复用的代码模式与避坑指南。1. Agent的核心挑战与美团实践启示在深入技术细节之前我们首先要明确在像美团这样高并发、高可用的业务系统中一个合格的Agent需要跨越哪些鸿沟。这不仅仅是调用大模型API那么简单。1.1 从“玩具”到“工具”生产级Agent的四大门槛可靠的工具调用Agent需要准确理解何时调用工具、调用哪个工具、传递什么参数。这涉及到工具描述的规范化、参数校验的严格性以及调用失败的重试与降级策略。网络搜索中提到的codex工具调用审批失败、agent terminated due to error等问题正是工具调用不可靠的典型表现。有效的上下文管理对话可能很长但大模型的上下文窗口有限且昂贵。如何从冗长的历史中精准提取与当前任务相关的信息避免无关信息干扰噪声和关键信息丢失遗忘是保证Agent持续有效对话的关键。复杂的任务拆解用户的一个指令如“帮我订一张明天从北京到上海、下午出发、价格低于1000元的机票并预约一辆下午4点到浦东机场的接机车”涉及多个子任务查询航班、过滤、预订、查询接送服务、预订车辆。Agent必须具备将模糊的、复合型的用户目标拆解为清晰、可顺序或并行执行的动作序列的能力。安全的系统集成Agent最终需要操作真实系统如订单系统、支付系统、库存系统。这涉及到权限控制、操作幂等性、数据一致性以及防止恶意提示注入等安全问题。美团mtgsig、sig签名算法等关键词正反映了在开放平台环境中对API调用安全性的高度重视。1.2 美团实践手册的聚焦点美团的实践手册正是围绕上述门槛在其海量业务场景中锤炼出的经验总结。其核心思想是将Agent视为一个需要精密设计的系统组件而非一个黑盒魔法。手册强调了规划-执行-观察Plan-Act-Observe的闭环并特别关注于场景化工具设计针对外卖、酒店、打车等不同业务的独特逻辑封装高内聚、低耦合的工具。分层级的记忆与上下文管理区分会话记忆、工具执行历史、实体知识等采用摘要、提取等策略压缩信息。结构化任务规划利用大模型进行任务分解并将子任务以标准化的数据结构如DAG有向无环图进行描述和执行跟踪。系统化安全与治理包括工具调用的审批流程呼应codex工具调用审批失败、输入输出过滤、操作审计等。接下来我们将脱离具体厂商框架以一套可落地的技术架构为例逐一拆解这些挑战的解决方案。2. 构建生产级Agent的系统架构一个健壮的Agent系统通常采用分层架构以下是一个参考模型用户请求 | v [ 接口网关 安全过滤 ] | v [ 智能体核心引擎 ] |-----------------------| v v [ 任务规划器 ] [ 上下文管理器 ] | | v v [ 工具执行器 ] --- [ 记忆存储器 ] | | v v [ 后端业务系统 ] [ 向量库/数据库 ]2.1 核心组件职责智能体核心引擎请求的总入口和调度中心协调其他组件工作。任务规划器分析用户意图将复杂任务拆解为子任务图Plan。上下文管理器负责维护、压缩、检索与当前对话相关的历史信息。工具执行器负责加载工具、校验参数、安全调用并返回标准化结果。记忆存储器持久化存储对话历史、工具执行结果、用户偏好等。3. 实战实现一个可拆解任务与调用工具的Agent我们将使用Python和流行的LangChain框架来构建一个简化版但功能完整的Agent模拟一个“旅行助手”场景它需要处理复合请求、调用工具查询天气、查询航班、计算差价。3.1 环境准备与依赖首先确保你的Python环境建议3.9并安装必要依赖。我们使用OpenAI的GPT模型作为核心LLM你也可以替换为其他兼容API的模型。pip install langchain langchain-openai langchain-community python-dotenv创建.env文件存储你的API密钥OPENAI_API_KEYyour_openai_api_key_here3.2 定义标准化工具工具是Agent的手和脚。每个工具必须有清晰的名称、描述和参数模式。# tools/custom_tools.py from langchain.tools import BaseTool from pydantic import BaseModel, Field from typing import Optional, Type import requests import datetime class WeatherQueryInput(BaseModel): 查询天气的输入参数。 city: str Field(description需要查询天气的城市名称例如北京) class WeatherTool(BaseTool): name get_weather description 获取指定城市当前的天气情况。 args_schema: Type[BaseModel] WeatherQueryInput def _run(self, city: str) - str: # 模拟一个天气API调用实际项目中替换为真实API # 这里返回模拟数据 weather_data { 北京: 晴15~25°C微风, 上海: 多云18~28°C东南风3级, 广州: 阵雨23~32°C南风4级, } return weather_data.get(city, f未找到{city}的天气信息。) class FlightQueryInput(BaseModel): 查询航班的输入参数。 departure: str Field(description出发城市) arrival: str Field(description到达城市) date: str Field(description出发日期格式YYYY-MM-DD) class FlightTool(BaseTool): name search_flights description 查询指定日期、出发地和目的地的航班信息。 args_schema: Type[BaseModel] FlightQueryInput def _run(self, departure: str, arrival: str, date: str) - str: # 模拟航班查询 # 实际应调用航班搜索API这里返回模拟结果 flights [ fCA1234 {departure}-{arrival} 08:00-10:30 经济舱 ¥1200, fMU5678 {departure}-{arrival} 14:00-16:45 经济舱 ¥1100, ] return \n.join(flights) class PriceDiffInput(BaseModel): 计算价格差值的输入参数。 price_a: float Field(description第一个价格) price_b: float Field(description第二个价格) class PriceDiffTool(BaseTool): name calculate_price_difference description 计算两个价格之间的绝对差值。 args_schema: Type[BaseModel] PriceDiffInput def _run(self, price_a: float, price_b: float) - str: diff abs(price_a - price_b) return f两个价格之间的差值为¥{diff:.2f}3.3 构建智能体与任务拆解策略我们使用LangChain的create_react_agent它实现了ReActReasoning Acting范式能让Agent在思考推理下一步和行动调用工具间循环。# agent/core_agent.py import os from dotenv import load_dotenv from langchain import hub from langchain.agents import create_react_agent, AgentExecutor from langchain_openai import ChatOpenAI from tools.custom_tools import WeatherTool, FlightTool, PriceDiffTool load_dotenv() # 1. 初始化大模型 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, api_keyos.getenv(OPENAI_API_KEY)) # 2. 加载工具 tools [WeatherTool(), FlightTool(), PriceDiffTool()] # 3. 获取ReAct提示词模板LangChain Hub上的优质模板 prompt hub.pull(hwchase17/react) # 4. 创建ReAct Agent agent create_react_agent(llm, tools, prompt) # 5. 创建执行器并设置verboseTrue以观察Agent的思考过程 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 6. 测试一个复杂任务 if __name__ __main__: complex_request 我打算明天从北京飞上海想知道上海的天气怎么样并且帮我查一下下午的航班最后告诉我最便宜的那个航班和今天查到的价格假设是1000元差多少钱。 print(f用户请求{complex_request}\n) try: result agent_executor.invoke({input: complex_request}) print(f\n最终结果{result[output]}) except Exception as e: print(f执行过程中出现错误{e})运行上述代码你将看到类似以下的输出verbose模式用户请求我打算明天从北京飞上海想知道上海的天气怎么样并且帮我查一下下午的航班最后告诉我最便宜的那个航班和今天查到的价格假设是1000元差多少钱。 Entering new AgentExecutor chain... 我需要先理解用户的请求。用户想明天从北京飞上海需要做三件事1. 查询上海的天气2. 查询明天从北京到上海的下午航班3. 从查到的航班中找到最便宜的一个并计算其价格与1000元的差价。 首先我应该查询上海的天气。 Action: get_weather Action Input: {city: 上海} Observation: 多云18~28°C东南风3级 Thought: 天气查询完成。接下来需要查询明天从北京到上海的下午航班。但用户说明天我需要知道具体日期。假设今天是2023-10-27那么明天就是2023-10-28。 Action: search_flights Action Input: {departure: 北京, arrival: 上海, date: 2023-10-28} Observation: CA1234 北京-上海 08:00-10:30 经济舱 ¥1200 MU5678 北京-上海 14:00-16:45 经济舱 ¥1100 Thought: 我查询到了两个航班。用户要求“下午的航班”MU5678是14:00起飞符合下午的条件。它是两个航班中更便宜的价格是1100元。现在需要计算这个价格与1000元的差价。 Action: calculate_price_difference Action Input: {price_a: 1100, price_b: 1000} Observation: 两个价格之间的差值为¥100.00 Thought: 我已经完成了所有步骤查询了上海天气找到了明天下午从北京到上海的航班MU5678¥1100并计算了它与1000元的差价¥100。现在可以给出最终答案了。 Final Answer: 上海的天气是多云18~28°C东南风3级。明天下午从北京飞上海的航班有MU567814:00-16:45经济舱价格¥1100。这个最便宜的下午航班价格与您提到的1000元相差¥100。 Finished chain. 最终结果上海的天气是多云18~28°C东南风3级。明天下午从北京飞上海的航班有MU567814:00-16:45经济舱价格¥1100。这个最便宜的下午航班价格与您提到的1000元相差¥100。关键点分析任务自动拆解Agent通过ReAct提示词的引导自动将复杂请求分解为“查询天气 - 查询航班 - 计算差价”三个顺序子任务。工具调用每个Action步骤都精准地调用了我们定义的工具并传递了正确的参数JSON格式。上下文连贯Agent在思考Thought中能够引用上一步的Observation如航班价格1100并将其作为下一步的输入实现了上下文传递。4. 进阶实现显式的任务规划与上下文管理上述ReAct Agent实现了隐式的任务拆解。对于更复杂、步骤固定的业务流程我们可以引入显式的任务规划器先规划再执行提升可控性。4.1 实现一个简单的规划器# planner/task_planner.py from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from pydantic import BaseModel from typing import List import json class SubTask(BaseModel): 子任务定义 id: int description: str tool_name: str # 需要调用的工具名 tool_input: dict # 工具输入参数 depends_on: List[int] [] # 依赖的前置任务ID class TaskPlan(BaseModel): 任务计划 tasks: List[SubTask] class TaskPlanner: def __init__(self, llm): self.llm llm self.prompt ChatPromptTemplate.from_messages([ (system, 你是一个任务规划专家。请将用户的复杂请求拆解成一个有序的子任务列表。 每个子任务必须对应一个可用的工具。可用的工具列表如下 {tools_list} 请以JSON格式输出严格遵循以下结构 json {{ tasks: [ {{id: 1, description: ..., tool_name: ..., tool_input: {{...}}, depends_on: []}}, {{id: 2, description: ..., tool_name: ..., tool_input: {{...}}, depends_on: [1]}} ] }} ), (human, 用户请求{request}) ]) def plan(self, user_request: str, available_tools: list) - TaskPlan: tools_list_str \n.join([f- {tool.name}: {tool.description} for tool in available_tools]) chain self.prompt | self.llm response chain.invoke({request: user_request, tools_list: tools_list_str}) # 从LLM响应中提取JSON部分 content response.content json_start content.find(json) json_end content.find(, json_start 7) if json_start ! -1 and json_end ! -1: json_str content[json_start 7:json_end].strip() else: # 如果没有代码块尝试直接解析整个内容 json_str content.strip() try: plan_dict json.loads(json_str) return TaskPlan(**plan_dict) except json.JSONDecodeError as e: print(fJSON解析失败: {e}\n原始内容:\n{content}) # 返回一个空的计划或抛出异常 return TaskPlan(tasks[])4.2 构建基于规划器的执行引擎# agent/planned_agent_executor.py from typing import Dict, Any from planner.task_planner import TaskPlanner, TaskPlan from tools.custom_tools import WeatherTool, FlightTool, PriceDiffTool class PlannedAgentExecutor: def __init__(self, llm, tools): self.planner TaskPlanner(llm) self.tools {tool.name: tool for tool in tools} def execute(self, user_request: str) - str: print( 阶段1任务规划 ) plan: TaskPlan self.planner.plan(user_request, list(self.tools.values())) print(f生成任务计划{plan.model_dump_json(indent2)}) print(\n 阶段2顺序执行 ) executed_results: Dict[int, Any] {} final_output [] # 简单按ID顺序执行实际应根据depends_on构建DAG并拓扑排序 for task in sorted(plan.tasks, keylambda x: x.id): print(f\n执行任务{task.id}: {task.description}) print(f 调用工具: {task.tool_name}, 参数: {task.tool_input}) tool self.tools.get(task.tool_name) if not tool: result f错误未找到工具 {task.tool_name} else: try: # 这里可以添加参数预处理例如将依赖任务的结果注入到当前任务的输入中 resolved_input self._resolve_input(task.tool_input, executed_results) result tool.invoke(resolved_input) except Exception as e: result f工具执行失败: {e} executed_results[task.id] result final_output.append(f任务{task.id}结果: {result}) print(f 结果: {result}) print(\n 执行完成 ) return \n.join(final_output) def _resolve_input(self, tool_input: dict, results: dict) - dict: 一个简单的输入解析器未来可扩展为支持模板变量如 {{task_1_result}} # 当前直接返回原输入进阶版可以解析依赖并替换值 return tool_input # 使用示例 if __name__ __main__: from langchain_openai import ChatOpenAI import os from dotenv import load_dotenv load_dotenv() llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) tools [WeatherTool(), FlightTool(), PriceDiffTool()] executor PlannedAgentExecutor(llm, tools) result executor.execute(查询北京和上海的天气然后告诉我哪里更暖和。) print(f\n最终整合回答\n{result})这种“先规划后执行”的模式将任务拆解的逻辑显式化、结构化带来了额外好处可预测性在执行前就能看到完整的任务流程图。可调试性哪个子任务失败一目了然。可优化可以分析任务依赖关系对可并行任务进行并发执行提升效率。5. 上下文处理与记忆管理实战Agent在长对话中会面临上下文窗口限制。我们需要主动管理历史而非简单地将所有历史对话都塞给模型。5.1 策略摘要式记忆与向量检索记忆# memory/advanced_memory.py from langchain.memory import ConversationSummaryBufferMemory from langchain_openai import ChatOpenAI from langchain_community.embeddings import OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document import os class HybridMemoryManager: 混合记忆管理器摘要记忆 向量检索记忆 def __init__(self, llm, embedding_model, persist_directory./chroma_db): # 1. 摘要记忆用于维持对话连贯性但只保留近期原始对话和长期摘要 self.summary_memory ConversationSummaryBufferMemory( llmllm, max_token_limit1000, # 控制保留的原始对话token数 return_messagesTrue, memory_keychat_history ) # 2. 向量记忆用于长期、详细的记忆检索 self.embeddings embedding_model self.vector_store Chroma( collection_nameconversation_memory, embedding_functionself.embeddings, persist_directorypersist_directory ) self.text_splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) def save_interaction(self, human_input: str, ai_output: str): 保存一次交互到两种记忆系统 # 保存到摘要记忆 self.summary_memory.save_context({input: human_input}, {output: ai_output}) # 保存到向量记忆 interaction_text fHuman: {human_input}\nAI: {ai_output} docs [Document(page_contentinteraction_text, metadata{type: interaction})] split_docs self.text_splitter.split_documents(docs) self.vector_store.add_documents(split_docs) self.vector_store.persist() def get_relevant_memories(self, query: str, k3) - str: 根据当前查询从向量记忆中检索最相关的历史片段 if self.vector_store._collection.count() 0: return retrieved_docs self.vector_store.similarity_search(query, kk) contexts [doc.page_content for doc in retrieved_docs] return \n\n--- 相关历史 ---\n \n---\n.join(contexts) def get_chat_history_for_prompt(self) - str: 获取格式化后的聊天历史供提示词使用 # 从摘要记忆中获取近期历史 history_messages self.summary_memory.load_memory_variables({})[chat_history] history_str \n.join([f{msg.type}: {msg.content} for msg in history_messages]) return history_str # 在Agent中使用 if __name__ __main__: load_dotenv() llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) embeddings OpenAIEmbeddings() memory_manager HybridMemoryManager(llm, embeddings) # 模拟多轮对话 dialogues [ (我喜欢吃辣推荐个上海的本帮菜馆吧。, 上海的老吉士酒家本帮菜做得很地道不过口味偏甜可能不够辣。), (那有没有辣一点的菜系推荐, 您可以尝试上海的湘菜馆比如‘望湘园’辣味十足。), (我明天要去北京出差了。, 北京天气干燥注意保湿。出差顺利) ] for human, ai in dialogues: memory_manager.save_interaction(human, ai) # 在新一轮对话中利用记忆 new_query 我上次问的关于辣味餐厅你还记得推荐了什么吗 relevant_mem memory_manager.get_relevant_memories(new_query) chat_history memory_manager.get_chat_history_for_prompt() print(当前对话历史摘要) print(chat_history) print(\n检索到的相关记忆) print(relevant_mem) # 构建包含记忆的最终提示词 enhanced_prompt f 以下是当前的对话历史 {chat_history} {relevant_mem} 请根据以上信息回答用户的最新问题 用户{new_query} AI print(\n 增强后的提示词 ) print(enhanced_prompt[:500]) # 打印部分通过这种混合记忆策略Agent既能保持对话的短期连贯性通过摘要又能从漫长的历史中精准回忆起与当前问题最相关的细节通过向量检索有效突破了固定上下文窗口的限制。6. 生产环境常见问题与排查思路在实际部署Agent时你会遇到各种工程化挑战。下表总结了常见问题及其解决方案问题现象可能原因排查思路与解决方案Agent无限循环或重复调用工具1. ReAct提示词引导不佳。2. 工具返回结果未能让Agent识别任务完成。3. 最大迭代次数设置过高。1. 优化提示词明确给出终止条件如“Final Answer”。2. 确保工具返回清晰、结构化的结果。3. 在AgentExecutor中设置max_iterations如10和early_stopping_method。工具调用参数错误1. 工具描述不清晰LLM无法理解。2. 参数schema定义不准确或太复杂。3. LLM幻觉生成不存在的参数。1. 精炼工具name和description使用明确动词和示例。2. 使用Pydantic严格定义参数类型和约束。3. 在调用工具前增加一层参数验证和清洗逻辑。处理长文档或复杂上下文时性能下降1. 将全部上下文送入模型导致token消耗大、速度慢、成本高。2. 无关信息干扰模型判断。1. 采用“摘要检索”策略如第5节所述。2. 对输入文档进行预处理提取关键实体和关系。3. 考虑使用具有更长上下文窗口的模型如GPT-4 Turbo。与内部系统集成时的安全问题1. Agent被用户恶意提示操纵执行危险操作。2. 工具权限过大。3. 敏感信息泄露。1.输入过滤对用户输入进行严格的指令检测和过滤。2.权限最小化每个工具只授予完成其功能所需的最小权限。3.操作确认对于关键操作如支付、删除设计用户确认或审批流程呼应codex工具调用审批失败的启示。4.输出过滤对Agent返回的内容进行脱敏处理。Agent execution terminated due to error1. 工具执行抛出未处理异常。2. 网络超时或依赖服务不可用。3. 模型输出无法被解析为有效动作。1. 在工具函数内部进行完善的异常捕获和友好错误返回。2. 为工具调用设置超时和重试机制。3. 在AgentExecutor中设置handle_parsing_errorsTrue并提供一个兜底的错误处理提示。7. 最佳实践与工程化建议结合美团等大厂的一线经验要将Agent从Demo推向生产必须关注以下几点工具设计标准化单一职责每个工具只做一件事并做好。避免“万能工具”。强类型校验使用Pydantic等库严格定义输入输出从源头减少错误。幂等与重试工具实现应尽可能幂等并考虑网络波动内置重试逻辑。完备的文档工具的description属性至关重要应清晰说明功能、输入输出示例和边界情况。提示词工程系统化模板化管理将提示词抽离为可配置的模板文件便于A/B测试和迭代。少样本示例在提示词中包含1-3个高质量的任务拆解和工具调用示例能极大提升Agent表现。明确边界在系统提示词中明确告知Agent“能做什么”和“不能做什么”减少幻觉和越权行为。可观测性与监控全链路日志记录每一次用户输入、模型思考、工具调用、模型输出的完整链这是排查问题的黄金数据。关键指标监控平均对话轮次、工具调用成功率、用户任务完成率、Token消耗成本等。评估体系建立自动化测试用例定期评估Agent在核心场景上的表现是否达标。安全与合规底线沙箱环境Agent在操作生产系统前应在沙箱环境充分测试。人工审核链路对于高风险操作必须保留人工审核或二次确认的通道。数据隐私确保用户对话数据、通过Agent查询的业务数据符合隐私保护规定。架构解耦将Agent大脑LLM、规划模块、工具集、记忆模块解耦允许独立升级和扩展。例如可以轻松更换不同的LLM提供商或为特定场景添加专用工具包。这份从美团等一线实践中提炼出的Agent构建指南旨在为你提供一个从理论到实践的完整路线图。记住一个成功的生产级Agent不是一个炫技的AI模型而是一个将大语言模型的认知能力与现有软件系统的可靠性和业务逻辑深度融合的、精心设计的系统工程。
在业务迭代中引入AI Agent时我们常常面临一个核心矛盾Agent的“聪明”与“可靠”难以兼得。它或许能理解复杂指令但在调用工具、处理长上下文、拆解多步骤任务以及与现有系统稳定交互时却频频“掉链子”导致落地困难。近期美团技术团队发布了一份基于外卖、酒店、打车等核心业务场景的Agent实践手册其价值正在于它没有空谈理论而是直面这些一线工程挑战提供了一套经过大规模流量验证的解决方案。本文将深入解读这份手册的精髓并结合行业通用实践拆解如何构建一个不仅能“听懂话”更能“办成事”的实用型AI Agent。内容涵盖从核心架构设计、上下文处理策略、任务自动拆解到与后端系统安全集成的全流程。无论你是正在探索Agent落地的架构师还是希望提升智能体可靠性的开发者都能从中获得可直接复用的代码模式与避坑指南。1. Agent的核心挑战与美团实践启示在深入技术细节之前我们首先要明确在像美团这样高并发、高可用的业务系统中一个合格的Agent需要跨越哪些鸿沟。这不仅仅是调用大模型API那么简单。1.1 从“玩具”到“工具”生产级Agent的四大门槛可靠的工具调用Agent需要准确理解何时调用工具、调用哪个工具、传递什么参数。这涉及到工具描述的规范化、参数校验的严格性以及调用失败的重试与降级策略。网络搜索中提到的codex工具调用审批失败、agent terminated due to error等问题正是工具调用不可靠的典型表现。有效的上下文管理对话可能很长但大模型的上下文窗口有限且昂贵。如何从冗长的历史中精准提取与当前任务相关的信息避免无关信息干扰噪声和关键信息丢失遗忘是保证Agent持续有效对话的关键。复杂的任务拆解用户的一个指令如“帮我订一张明天从北京到上海、下午出发、价格低于1000元的机票并预约一辆下午4点到浦东机场的接机车”涉及多个子任务查询航班、过滤、预订、查询接送服务、预订车辆。Agent必须具备将模糊的、复合型的用户目标拆解为清晰、可顺序或并行执行的动作序列的能力。安全的系统集成Agent最终需要操作真实系统如订单系统、支付系统、库存系统。这涉及到权限控制、操作幂等性、数据一致性以及防止恶意提示注入等安全问题。美团mtgsig、sig签名算法等关键词正反映了在开放平台环境中对API调用安全性的高度重视。1.2 美团实践手册的聚焦点美团的实践手册正是围绕上述门槛在其海量业务场景中锤炼出的经验总结。其核心思想是将Agent视为一个需要精密设计的系统组件而非一个黑盒魔法。手册强调了规划-执行-观察Plan-Act-Observe的闭环并特别关注于场景化工具设计针对外卖、酒店、打车等不同业务的独特逻辑封装高内聚、低耦合的工具。分层级的记忆与上下文管理区分会话记忆、工具执行历史、实体知识等采用摘要、提取等策略压缩信息。结构化任务规划利用大模型进行任务分解并将子任务以标准化的数据结构如DAG有向无环图进行描述和执行跟踪。系统化安全与治理包括工具调用的审批流程呼应codex工具调用审批失败、输入输出过滤、操作审计等。接下来我们将脱离具体厂商框架以一套可落地的技术架构为例逐一拆解这些挑战的解决方案。2. 构建生产级Agent的系统架构一个健壮的Agent系统通常采用分层架构以下是一个参考模型用户请求 | v [ 接口网关 安全过滤 ] | v [ 智能体核心引擎 ] |-----------------------| v v [ 任务规划器 ] [ 上下文管理器 ] | | v v [ 工具执行器 ] --- [ 记忆存储器 ] | | v v [ 后端业务系统 ] [ 向量库/数据库 ]2.1 核心组件职责智能体核心引擎请求的总入口和调度中心协调其他组件工作。任务规划器分析用户意图将复杂任务拆解为子任务图Plan。上下文管理器负责维护、压缩、检索与当前对话相关的历史信息。工具执行器负责加载工具、校验参数、安全调用并返回标准化结果。记忆存储器持久化存储对话历史、工具执行结果、用户偏好等。3. 实战实现一个可拆解任务与调用工具的Agent我们将使用Python和流行的LangChain框架来构建一个简化版但功能完整的Agent模拟一个“旅行助手”场景它需要处理复合请求、调用工具查询天气、查询航班、计算差价。3.1 环境准备与依赖首先确保你的Python环境建议3.9并安装必要依赖。我们使用OpenAI的GPT模型作为核心LLM你也可以替换为其他兼容API的模型。pip install langchain langchain-openai langchain-community python-dotenv创建.env文件存储你的API密钥OPENAI_API_KEYyour_openai_api_key_here3.2 定义标准化工具工具是Agent的手和脚。每个工具必须有清晰的名称、描述和参数模式。# tools/custom_tools.py from langchain.tools import BaseTool from pydantic import BaseModel, Field from typing import Optional, Type import requests import datetime class WeatherQueryInput(BaseModel): 查询天气的输入参数。 city: str Field(description需要查询天气的城市名称例如北京) class WeatherTool(BaseTool): name get_weather description 获取指定城市当前的天气情况。 args_schema: Type[BaseModel] WeatherQueryInput def _run(self, city: str) - str: # 模拟一个天气API调用实际项目中替换为真实API # 这里返回模拟数据 weather_data { 北京: 晴15~25°C微风, 上海: 多云18~28°C东南风3级, 广州: 阵雨23~32°C南风4级, } return weather_data.get(city, f未找到{city}的天气信息。) class FlightQueryInput(BaseModel): 查询航班的输入参数。 departure: str Field(description出发城市) arrival: str Field(description到达城市) date: str Field(description出发日期格式YYYY-MM-DD) class FlightTool(BaseTool): name search_flights description 查询指定日期、出发地和目的地的航班信息。 args_schema: Type[BaseModel] FlightQueryInput def _run(self, departure: str, arrival: str, date: str) - str: # 模拟航班查询 # 实际应调用航班搜索API这里返回模拟结果 flights [ fCA1234 {departure}-{arrival} 08:00-10:30 经济舱 ¥1200, fMU5678 {departure}-{arrival} 14:00-16:45 经济舱 ¥1100, ] return \n.join(flights) class PriceDiffInput(BaseModel): 计算价格差值的输入参数。 price_a: float Field(description第一个价格) price_b: float Field(description第二个价格) class PriceDiffTool(BaseTool): name calculate_price_difference description 计算两个价格之间的绝对差值。 args_schema: Type[BaseModel] PriceDiffInput def _run(self, price_a: float, price_b: float) - str: diff abs(price_a - price_b) return f两个价格之间的差值为¥{diff:.2f}3.3 构建智能体与任务拆解策略我们使用LangChain的create_react_agent它实现了ReActReasoning Acting范式能让Agent在思考推理下一步和行动调用工具间循环。# agent/core_agent.py import os from dotenv import load_dotenv from langchain import hub from langchain.agents import create_react_agent, AgentExecutor from langchain_openai import ChatOpenAI from tools.custom_tools import WeatherTool, FlightTool, PriceDiffTool load_dotenv() # 1. 初始化大模型 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, api_keyos.getenv(OPENAI_API_KEY)) # 2. 加载工具 tools [WeatherTool(), FlightTool(), PriceDiffTool()] # 3. 获取ReAct提示词模板LangChain Hub上的优质模板 prompt hub.pull(hwchase17/react) # 4. 创建ReAct Agent agent create_react_agent(llm, tools, prompt) # 5. 创建执行器并设置verboseTrue以观察Agent的思考过程 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 6. 测试一个复杂任务 if __name__ __main__: complex_request 我打算明天从北京飞上海想知道上海的天气怎么样并且帮我查一下下午的航班最后告诉我最便宜的那个航班和今天查到的价格假设是1000元差多少钱。 print(f用户请求{complex_request}\n) try: result agent_executor.invoke({input: complex_request}) print(f\n最终结果{result[output]}) except Exception as e: print(f执行过程中出现错误{e})运行上述代码你将看到类似以下的输出verbose模式用户请求我打算明天从北京飞上海想知道上海的天气怎么样并且帮我查一下下午的航班最后告诉我最便宜的那个航班和今天查到的价格假设是1000元差多少钱。 Entering new AgentExecutor chain... 我需要先理解用户的请求。用户想明天从北京飞上海需要做三件事1. 查询上海的天气2. 查询明天从北京到上海的下午航班3. 从查到的航班中找到最便宜的一个并计算其价格与1000元的差价。 首先我应该查询上海的天气。 Action: get_weather Action Input: {city: 上海} Observation: 多云18~28°C东南风3级 Thought: 天气查询完成。接下来需要查询明天从北京到上海的下午航班。但用户说明天我需要知道具体日期。假设今天是2023-10-27那么明天就是2023-10-28。 Action: search_flights Action Input: {departure: 北京, arrival: 上海, date: 2023-10-28} Observation: CA1234 北京-上海 08:00-10:30 经济舱 ¥1200 MU5678 北京-上海 14:00-16:45 经济舱 ¥1100 Thought: 我查询到了两个航班。用户要求“下午的航班”MU5678是14:00起飞符合下午的条件。它是两个航班中更便宜的价格是1100元。现在需要计算这个价格与1000元的差价。 Action: calculate_price_difference Action Input: {price_a: 1100, price_b: 1000} Observation: 两个价格之间的差值为¥100.00 Thought: 我已经完成了所有步骤查询了上海天气找到了明天下午从北京到上海的航班MU5678¥1100并计算了它与1000元的差价¥100。现在可以给出最终答案了。 Final Answer: 上海的天气是多云18~28°C东南风3级。明天下午从北京飞上海的航班有MU567814:00-16:45经济舱价格¥1100。这个最便宜的下午航班价格与您提到的1000元相差¥100。 Finished chain. 最终结果上海的天气是多云18~28°C东南风3级。明天下午从北京飞上海的航班有MU567814:00-16:45经济舱价格¥1100。这个最便宜的下午航班价格与您提到的1000元相差¥100。关键点分析任务自动拆解Agent通过ReAct提示词的引导自动将复杂请求分解为“查询天气 - 查询航班 - 计算差价”三个顺序子任务。工具调用每个Action步骤都精准地调用了我们定义的工具并传递了正确的参数JSON格式。上下文连贯Agent在思考Thought中能够引用上一步的Observation如航班价格1100并将其作为下一步的输入实现了上下文传递。4. 进阶实现显式的任务规划与上下文管理上述ReAct Agent实现了隐式的任务拆解。对于更复杂、步骤固定的业务流程我们可以引入显式的任务规划器先规划再执行提升可控性。4.1 实现一个简单的规划器# planner/task_planner.py from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from pydantic import BaseModel from typing import List import json class SubTask(BaseModel): 子任务定义 id: int description: str tool_name: str # 需要调用的工具名 tool_input: dict # 工具输入参数 depends_on: List[int] [] # 依赖的前置任务ID class TaskPlan(BaseModel): 任务计划 tasks: List[SubTask] class TaskPlanner: def __init__(self, llm): self.llm llm self.prompt ChatPromptTemplate.from_messages([ (system, 你是一个任务规划专家。请将用户的复杂请求拆解成一个有序的子任务列表。 每个子任务必须对应一个可用的工具。可用的工具列表如下 {tools_list} 请以JSON格式输出严格遵循以下结构 json {{ tasks: [ {{id: 1, description: ..., tool_name: ..., tool_input: {{...}}, depends_on: []}}, {{id: 2, description: ..., tool_name: ..., tool_input: {{...}}, depends_on: [1]}} ] }} ), (human, 用户请求{request}) ]) def plan(self, user_request: str, available_tools: list) - TaskPlan: tools_list_str \n.join([f- {tool.name}: {tool.description} for tool in available_tools]) chain self.prompt | self.llm response chain.invoke({request: user_request, tools_list: tools_list_str}) # 从LLM响应中提取JSON部分 content response.content json_start content.find(json) json_end content.find(, json_start 7) if json_start ! -1 and json_end ! -1: json_str content[json_start 7:json_end].strip() else: # 如果没有代码块尝试直接解析整个内容 json_str content.strip() try: plan_dict json.loads(json_str) return TaskPlan(**plan_dict) except json.JSONDecodeError as e: print(fJSON解析失败: {e}\n原始内容:\n{content}) # 返回一个空的计划或抛出异常 return TaskPlan(tasks[])4.2 构建基于规划器的执行引擎# agent/planned_agent_executor.py from typing import Dict, Any from planner.task_planner import TaskPlanner, TaskPlan from tools.custom_tools import WeatherTool, FlightTool, PriceDiffTool class PlannedAgentExecutor: def __init__(self, llm, tools): self.planner TaskPlanner(llm) self.tools {tool.name: tool for tool in tools} def execute(self, user_request: str) - str: print( 阶段1任务规划 ) plan: TaskPlan self.planner.plan(user_request, list(self.tools.values())) print(f生成任务计划{plan.model_dump_json(indent2)}) print(\n 阶段2顺序执行 ) executed_results: Dict[int, Any] {} final_output [] # 简单按ID顺序执行实际应根据depends_on构建DAG并拓扑排序 for task in sorted(plan.tasks, keylambda x: x.id): print(f\n执行任务{task.id}: {task.description}) print(f 调用工具: {task.tool_name}, 参数: {task.tool_input}) tool self.tools.get(task.tool_name) if not tool: result f错误未找到工具 {task.tool_name} else: try: # 这里可以添加参数预处理例如将依赖任务的结果注入到当前任务的输入中 resolved_input self._resolve_input(task.tool_input, executed_results) result tool.invoke(resolved_input) except Exception as e: result f工具执行失败: {e} executed_results[task.id] result final_output.append(f任务{task.id}结果: {result}) print(f 结果: {result}) print(\n 执行完成 ) return \n.join(final_output) def _resolve_input(self, tool_input: dict, results: dict) - dict: 一个简单的输入解析器未来可扩展为支持模板变量如 {{task_1_result}} # 当前直接返回原输入进阶版可以解析依赖并替换值 return tool_input # 使用示例 if __name__ __main__: from langchain_openai import ChatOpenAI import os from dotenv import load_dotenv load_dotenv() llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) tools [WeatherTool(), FlightTool(), PriceDiffTool()] executor PlannedAgentExecutor(llm, tools) result executor.execute(查询北京和上海的天气然后告诉我哪里更暖和。) print(f\n最终整合回答\n{result})这种“先规划后执行”的模式将任务拆解的逻辑显式化、结构化带来了额外好处可预测性在执行前就能看到完整的任务流程图。可调试性哪个子任务失败一目了然。可优化可以分析任务依赖关系对可并行任务进行并发执行提升效率。5. 上下文处理与记忆管理实战Agent在长对话中会面临上下文窗口限制。我们需要主动管理历史而非简单地将所有历史对话都塞给模型。5.1 策略摘要式记忆与向量检索记忆# memory/advanced_memory.py from langchain.memory import ConversationSummaryBufferMemory from langchain_openai import ChatOpenAI from langchain_community.embeddings import OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document import os class HybridMemoryManager: 混合记忆管理器摘要记忆 向量检索记忆 def __init__(self, llm, embedding_model, persist_directory./chroma_db): # 1. 摘要记忆用于维持对话连贯性但只保留近期原始对话和长期摘要 self.summary_memory ConversationSummaryBufferMemory( llmllm, max_token_limit1000, # 控制保留的原始对话token数 return_messagesTrue, memory_keychat_history ) # 2. 向量记忆用于长期、详细的记忆检索 self.embeddings embedding_model self.vector_store Chroma( collection_nameconversation_memory, embedding_functionself.embeddings, persist_directorypersist_directory ) self.text_splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) def save_interaction(self, human_input: str, ai_output: str): 保存一次交互到两种记忆系统 # 保存到摘要记忆 self.summary_memory.save_context({input: human_input}, {output: ai_output}) # 保存到向量记忆 interaction_text fHuman: {human_input}\nAI: {ai_output} docs [Document(page_contentinteraction_text, metadata{type: interaction})] split_docs self.text_splitter.split_documents(docs) self.vector_store.add_documents(split_docs) self.vector_store.persist() def get_relevant_memories(self, query: str, k3) - str: 根据当前查询从向量记忆中检索最相关的历史片段 if self.vector_store._collection.count() 0: return retrieved_docs self.vector_store.similarity_search(query, kk) contexts [doc.page_content for doc in retrieved_docs] return \n\n--- 相关历史 ---\n \n---\n.join(contexts) def get_chat_history_for_prompt(self) - str: 获取格式化后的聊天历史供提示词使用 # 从摘要记忆中获取近期历史 history_messages self.summary_memory.load_memory_variables({})[chat_history] history_str \n.join([f{msg.type}: {msg.content} for msg in history_messages]) return history_str # 在Agent中使用 if __name__ __main__: load_dotenv() llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) embeddings OpenAIEmbeddings() memory_manager HybridMemoryManager(llm, embeddings) # 模拟多轮对话 dialogues [ (我喜欢吃辣推荐个上海的本帮菜馆吧。, 上海的老吉士酒家本帮菜做得很地道不过口味偏甜可能不够辣。), (那有没有辣一点的菜系推荐, 您可以尝试上海的湘菜馆比如‘望湘园’辣味十足。), (我明天要去北京出差了。, 北京天气干燥注意保湿。出差顺利) ] for human, ai in dialogues: memory_manager.save_interaction(human, ai) # 在新一轮对话中利用记忆 new_query 我上次问的关于辣味餐厅你还记得推荐了什么吗 relevant_mem memory_manager.get_relevant_memories(new_query) chat_history memory_manager.get_chat_history_for_prompt() print(当前对话历史摘要) print(chat_history) print(\n检索到的相关记忆) print(relevant_mem) # 构建包含记忆的最终提示词 enhanced_prompt f 以下是当前的对话历史 {chat_history} {relevant_mem} 请根据以上信息回答用户的最新问题 用户{new_query} AI print(\n 增强后的提示词 ) print(enhanced_prompt[:500]) # 打印部分通过这种混合记忆策略Agent既能保持对话的短期连贯性通过摘要又能从漫长的历史中精准回忆起与当前问题最相关的细节通过向量检索有效突破了固定上下文窗口的限制。6. 生产环境常见问题与排查思路在实际部署Agent时你会遇到各种工程化挑战。下表总结了常见问题及其解决方案问题现象可能原因排查思路与解决方案Agent无限循环或重复调用工具1. ReAct提示词引导不佳。2. 工具返回结果未能让Agent识别任务完成。3. 最大迭代次数设置过高。1. 优化提示词明确给出终止条件如“Final Answer”。2. 确保工具返回清晰、结构化的结果。3. 在AgentExecutor中设置max_iterations如10和early_stopping_method。工具调用参数错误1. 工具描述不清晰LLM无法理解。2. 参数schema定义不准确或太复杂。3. LLM幻觉生成不存在的参数。1. 精炼工具name和description使用明确动词和示例。2. 使用Pydantic严格定义参数类型和约束。3. 在调用工具前增加一层参数验证和清洗逻辑。处理长文档或复杂上下文时性能下降1. 将全部上下文送入模型导致token消耗大、速度慢、成本高。2. 无关信息干扰模型判断。1. 采用“摘要检索”策略如第5节所述。2. 对输入文档进行预处理提取关键实体和关系。3. 考虑使用具有更长上下文窗口的模型如GPT-4 Turbo。与内部系统集成时的安全问题1. Agent被用户恶意提示操纵执行危险操作。2. 工具权限过大。3. 敏感信息泄露。1.输入过滤对用户输入进行严格的指令检测和过滤。2.权限最小化每个工具只授予完成其功能所需的最小权限。3.操作确认对于关键操作如支付、删除设计用户确认或审批流程呼应codex工具调用审批失败的启示。4.输出过滤对Agent返回的内容进行脱敏处理。Agent execution terminated due to error1. 工具执行抛出未处理异常。2. 网络超时或依赖服务不可用。3. 模型输出无法被解析为有效动作。1. 在工具函数内部进行完善的异常捕获和友好错误返回。2. 为工具调用设置超时和重试机制。3. 在AgentExecutor中设置handle_parsing_errorsTrue并提供一个兜底的错误处理提示。7. 最佳实践与工程化建议结合美团等大厂的一线经验要将Agent从Demo推向生产必须关注以下几点工具设计标准化单一职责每个工具只做一件事并做好。避免“万能工具”。强类型校验使用Pydantic等库严格定义输入输出从源头减少错误。幂等与重试工具实现应尽可能幂等并考虑网络波动内置重试逻辑。完备的文档工具的description属性至关重要应清晰说明功能、输入输出示例和边界情况。提示词工程系统化模板化管理将提示词抽离为可配置的模板文件便于A/B测试和迭代。少样本示例在提示词中包含1-3个高质量的任务拆解和工具调用示例能极大提升Agent表现。明确边界在系统提示词中明确告知Agent“能做什么”和“不能做什么”减少幻觉和越权行为。可观测性与监控全链路日志记录每一次用户输入、模型思考、工具调用、模型输出的完整链这是排查问题的黄金数据。关键指标监控平均对话轮次、工具调用成功率、用户任务完成率、Token消耗成本等。评估体系建立自动化测试用例定期评估Agent在核心场景上的表现是否达标。安全与合规底线沙箱环境Agent在操作生产系统前应在沙箱环境充分测试。人工审核链路对于高风险操作必须保留人工审核或二次确认的通道。数据隐私确保用户对话数据、通过Agent查询的业务数据符合隐私保护规定。架构解耦将Agent大脑LLM、规划模块、工具集、记忆模块解耦允许独立升级和扩展。例如可以轻松更换不同的LLM提供商或为特定场景添加专用工具包。这份从美团等一线实践中提炼出的Agent构建指南旨在为你提供一个从理论到实践的完整路线图。记住一个成功的生产级Agent不是一个炫技的AI模型而是一个将大语言模型的认知能力与现有软件系统的可靠性和业务逻辑深度融合的、精心设计的系统工程。
![洛谷 P7912：[CSP-J 2021 T4] 小熊的果篮 ← 双向链表](http://pic.xiahunao.cn/yaotu/洛谷 P7912：[CSP-J 2021 T4] 小熊的果篮 ← 双向链表)
![[光学原理与应用-521]：对光的错误理解与纠偏](http://pic.xiahunao.cn/yaotu/[光学原理与应用-521]：对光的错误理解与纠偏)
