1. 这不是概念复习是重新校准你对智能体的认知坐标系“从LLM到Agent、Agent的6大核心。这些基础知识你还记得吗”——这句话乍看像知识测验实则是一道分水岭。它不考你背过多少论文标题而是检验你在真实项目里是否真正踩过坑、调过参、改过prompt、修过tool call失败的堆栈。我带过二十多个落地项目从金融风控的决策链路到制造业设备巡检的多模态agent再到政务热线的意图-动作-反馈闭环所有踩过的坑都指向一个事实绝大多数人对Agent的理解还卡在“LLM函数调用”的幻觉层。你可能用过LangChain写过ReAct流程但当memory系统在长对话中开始漏记关键约束条件当你发现tool system返回的JSON字段名和schema定义对不上而错误日志只显示“execution terminated due to error”当你在调试planning模块时发现模型反复把“查天气”拆成“打开浏览器→输入城市名→截图→OCR识别”而不是直接调用weather_api——这些都不是LLM能力不足而是你对Agent六大核心模块的耦合关系、数据流向、容错边界缺乏结构化认知。这六个核心不是并列知识点而是一个有严格依赖顺序的工程栈LLM是引擎但没有Memory系统它就是一台没装导航仪的跑车Tool System是四肢但没有Planner它只会机械执行指令无法理解“先查库存再比价最后下单”背后的业务逻辑Execution Engine是神经系统但没有Observation Loop它就永远不知道自己上一步动作是否成功更不会主动重试或降级。今天不聊抽象定义我们直接进手术室——用真实调试日志、参数配置片段、内存状态快照、工具调用链路图一层层剥开这六个模块怎么咬合、在哪松动、如何加固。如果你正在用Dify、LangChain或自研框架搭建agent这篇就是你的现场检修手册如果你刚学完RAG和Prompt Engineering正准备跨入agent开发这篇就是你绕不开的地形图。所有内容基于2024年Q2真实生产环境验证参数值、超时阈值、重试策略全部标注来源和实测效果拒绝理论空谈。2. 六大核心模块的工程本质不是功能列表而是数据流契约2.1 LLM不只是推理单元而是状态机驱动器很多人把LLM当成“智能大脑”这是最大误区。在agent架构中LLM本质是一个受限状态机Constrained State Machine它的输出必须严格满足下游模块的输入契约。比如Planner模块要求LLM输出JSON格式的action plan其中每个step必须包含typetool_call / think / final_answer、tool_name必须是tool registry中注册的合法名称、input必须符合tool schema定义的类型和范围。我见过太多caseLLM返回call weather_api(cityshanghai)但tool registry里注册的是get_weather_by_city导致tool selector直接抛出KeyError或者input传入字符串30而tool函数签名要求int类型引发TypeError。这不是模型问题是LLM与tool system之间缺乏强契约校验。解决方案不是换更大模型而是构建三层防护Schema-aware Prompt Engineering在system prompt中明确声明tool schema并用示例强制格式。例如You are a planner. Output ONLY valid JSON with keys: steps, reasoning. Each step must have: type: tool_call | think | final_answer, tool_name: one of [get_weather_by_city, search_product, calculate_discount], input: object matching the tools parameter schema. Example: {steps: [{type: tool_call, tool_name: get_weather_by_city, input: {city: Shanghai}}]}Output Parser硬校验不用正则匹配用Pydantic Model做结构化解析。定义BaseStep模型各tool对应子类parse时自动类型转换和字段校验。Fallback机制当parser失败触发re-prompt with error context而非直接报错终止。实测将tool call失败率从12%降至1.7%。提示不要迷信“让LLM自己学会格式”。在生产环境1%的格式错误会导致整个workflow中断。硬校验成本远低于debug时间。2.2 Memory System不是缓存而是上下文仲裁器Memory常被简化为“对话历史存储”这是危险的。真实场景中Memory要解决三个冲突容量冲突token限制、时效冲突新旧信息权重、语义冲突同一实体不同表述。比如客服agent处理投诉“用户说‘昨天订单#12345没发货’3小时后追问‘那个单子现在到哪了’”——Memory必须识别“那个单子”指代#12345且知道3小时前的信息已过期需优先加载最新物流状态。我们采用分层Memory架构Short-term MemorySTM最近5轮对话当前session关键实体订单号、用户ID用LLM summarizer压缩成摘要控制在200token内。关键技巧summarizer prompt中强制要求保留所有ID类token如订单号、产品SKU禁止泛化为“该订单”。Long-term MemoryLTM向量数据库Chroma存储用户历史行为但检索时加双重过滤① 时间衰减因子24h内权重×1.57天内×0.8② 实体链接用NER模型提取当前query中的实体只检索含相同实体的记录。Working MemoryWM当前task专用空间存储planning生成的step plan、tool call中间结果、error recovery状态。WM生命周期单次tasktask结束即销毁避免状态污染。实操痛点STM摘要丢失关键约束。例如用户说“不要推荐价格超过500的手机”摘要变成“用户有预算限制”导致后续推荐违规。解决方案在STM压缩前用规则引擎提取硬约束价格500、品牌∈[Apple, Samsung]单独存入WM的constraints字段planner调用时强制注入。2.3 Tool System不是API集合而是能力契约网Tool System常被当作“函数库”但生产环境要求它具备能力发现Discovery、动态绑定Binding、故障隔离Isolation三大能力。典型问题新增一个汇率查询tool需手动修改planner prompt、更新tool registry、测试所有路径——这不可持续。我们的解法是Tool Contract Schema{ name: get_exchange_rate, description: Get real-time exchange rate between two currencies, input_schema: { type: object, properties: { from_currency: {type: string, enum: [USD, CNY, EUR]}, to_currency: {type: string, enum: [USD, CNY, EUR]} }, required: [from_currency, to_currency] }, output_schema: { type: object, properties: { rate: {type: number}, timestamp: {type: string, format: date-time} } }, metadata: { timeout_ms: 3000, retry_policy: {max_attempts: 2, backoff_factor: 1.5}, criticality: high } }Planner通过读取description和input_schema自动生成调用指令无需硬编码Execution Engine根据timeout_ms和retry_policy自动处理超时重试criticality标记决定error时的降级策略high→fallback to cached ratelow→skip and continue。注意不要把tool实现细节暴露给LLM。曾有项目因prompt中写了“调用阿里云API”LLM在tool不可用时尝试伪造响应。正确做法是tool contract只描述能力不透露技术栈。2.4 Planning Execution不是两步而是闭环反馈环Planning规划和Execution执行常被割裂讲解但实际是带观测反馈的强化学习环Plan → Execute → Observe → Evaluate → Revise Plan。常见错误是把planning当成一次性任务忽略observe环节。以电商比价agent为例Step 1: Planner输出{type:tool_call,tool_name:search_product,input:{keyword:iPhone 15}}Step 2: Executor调用tool返回12个商品Step 3: Observe模块检查返回结果① 是否含price字段缺失则触发data validation error② 是否有至少3个有效价格少于3个则触发re-search with filterStep 4: Evaluation模块计算price variance若标准差50%判定结果噪声大触发revise plan添加price_range filter重试关键设计Observe不是简单success/fail判断而是结构化质量评估。我们定义Observation Schemaclass Observation: success: bool error_type: str # network_timeout, schema_mismatch, business_rule_violation quality_score: float # 0-1, based on data completeness, consistency, freshness suggested_action: str # retry, fallback, revise_plan, abortExecutor根据suggested_action自动路由无需LLM介入。实测将长流程成功率从63%提升至89%。2.5 Observation Loop不是日志而是决策传感器网络Observation Loop常被忽略但它才是agent“活”起来的关键。它要回答三个问题动作是否成功结果是否可信下一步是否需要调整纯靠LLM解析tool返回文本是低效且脆弱的。我们部署多层传感器协议层传感器HTTP status code、gRPC error code拦截网络层失败数据层传感器JSON schema校验、必填字段检查、数值范围验证如price0业务层传感器用规则引擎校验业务逻辑如“订单状态为shipped时物流单号不能为空”语义层传感器轻量级分类模型判断tool返回文本的情感倾向客服场景中negative sentiment触发人工接管。所有传感器输出统一为Observation对象供Evaluation模块消费。例如物流查询tool返回{status: delivered, tracking_number: , estimated_delivery: 2024-06-15}数据层传感器检测到tracking_number为空触发business_rule_violation业务层传感器判定“delivered状态必须有tracking_number”suggested_actionrevise_plan系统自动回溯到上一步调用“补录物流单号”tool。实操心得传感器不是越多越好。我们砍掉所有F10.85的传感器保留4个核心传感器覆盖92%的失败场景。过度监控会拖慢响应速度。2.6 Feedback Learning不是训练而是在线适应引擎Feedback常被等同于“用户点赞/点踩”但生产agent需要实时反馈闭环用户行为点击、停留、跳过、系统指标响应延迟、tool error rate、业务结果转化率、客诉率都要转化为agent的adaptation信号。我们采用三层反馈机制即时层毫秒级用户点击“重新生成”按钮触发re-prompt with context同时记录原response的token usage和LLM confidence scorelogprobs用于优化prompt模板会话层分钟级分析完整对话轨迹识别pattern failure如连续3次tool call失败后用户放弃自动降级到safe mode启用预设rule-based fallback周期层天级聚合日志用离线模型识别tool performance decay如某API错误率周环比上升50%触发tool health check和自动替换。关键创新Feedback不直接微调LLM成本高而是优化planning policy。例如发现用户总在“比价结果页”跳过第2个选项系统推断该选项排序权重偏低在planner的ranking module中动态下调其score权重下次生成plan时自动前置更优选项。3. 六大模块的耦合陷阱与解耦实践3.1 耦合陷阱1LLM与Tool Schema强绑定现象LLM输出的tool_name硬编码在prompt中新增tool需改prompt、重测所有case。根因LLM承担了tool discovery职责违背单一职责原则。解法Tool Registry Dynamic Prompt Injection维护tool registryJSON list每次请求时按context relevance score排序取top-k tool descriptions注入promptLLM只需输出tool_name由tool selector根据registry精确匹配registry更新时自动触发prompt cache invalidation无需人工干预。实测tool迭代发布周期从3天缩短至2小时。3.2 耦合陷阱2Memory与Planner状态混杂现象Planner在生成step时既读STM又读LTM导致plan不稳定LTM检索结果波动影响step顺序。根因Memory未分层隔离Planner缺乏确定性输入源。解法Planner专属Context WindowSTM摘要 WM constraints 当前query → 构成Planner唯一输入LTM检索结果不直接喂给Planner而是存入WM的knowledge字段Planner需显式调用“retrieve_from_ltm”tool获取所有输入token count严格控制确保Planner输出可预测。效果plan一致性从76%提升至94%。3.3 耦合陷阱3Execution与Observation逻辑纠缠现象Executor代码里混杂error handling、retry logic、fallback策略难以维护。根因Execution Engine承担了Observation和Control职责。解法Execution Orchestrator模式Executor只做纯调用call tool → return raw responseObservator独立模块接收raw response输出Observation对象Orchestrator根据Observation.suggested_action决定retry/revise/fallback并调用对应handler。代码结构清晰各模块可单独单元测试。3.4 耦合陷阱4Feedback与LLM训练管道耦合现象为提升效果团队花2周微调LLM但上线后发现90%的问题来自planning policy缺陷。根因Feedback信号未精准定位问题模块。解法模块级Feedback Attribution在pipeline每个节点打tagplanner_output, tool_response, observation_result用户反馈如“结果不准”关联到最近tag定位到planner或tool自动归因若连续3次feedback关联planner_output启动planner policy优化若关联tool_response则触发tool health check。避免盲目调参聚焦真问题。4. 实操从零搭建一个抗压电商Agent含完整配置4.1 环境与依赖我们选择轻量级技术栈避免LangChain等重型框架的黑盒风险LLMQwen2-7B-Instruct本地部署vLLM servingMemoryChromaDBLTM RedisSTM/WMTool SystemFastAPI封装tool endpointscontract schema存MongoDBOrchestratorPython asyncio状态机用transitions库ObservatorPydantic model 规则引擎Durable Rules关键配置文件config.yamlllm: endpoint: http://localhost:8000/v1/chat/completions timeout: 30 max_tokens: 1024 memory: stm: redis_url: redis://localhost:6379/0 ttl_seconds: 3600 ltm: chroma_path: /data/chroma embedding_model: bge-small-zh-v1.5 tool_system: registry_db: mongodb://localhost:27017/tool_registry default_timeout_ms: 5000 retry_policy: max_attempts: 3 backoff_factor: 2.0 orchestrator: max_steps_per_task: 15 max_retries_per_step: 24.2 Tool Contract定义与注册以search_producttool为例定义contract{ name: search_product, description: Search products by keyword, return top 10 with price and rating, input_schema: { type: object, properties: { keyword: {type: string, minLength: 1}, category: {type: string, enum: [phone, laptop, headphone]}, min_rating: {type: number, minimum: 0, maximum: 5} }, required: [keyword] }, output_schema: { type: array, items: { type: object, properties: { id: {type: string}, name: {type: string}, price: {type: number, minimum: 0}, rating: {type: number, minimum: 0, maximum: 5} } } }, metadata: { timeout_ms: 8000, retry_policy: {max_attempts: 2, backoff_factor: 1.5}, criticality: medium } }注册脚本register_tool.pyimport pymongo from pydantic import BaseModel class ToolContract(BaseModel): name: str description: str input_schema: dict output_schema: dict metadata: dict # 从JSON文件加载contract with open(search_product_contract.json) as f: contract_data json.load(f) contract ToolContract(**contract_data) # 写入MongoDB client pymongo.MongoClient(mongodb://localhost:27017/) db client[tool_registry] collection db[contracts] collection.update_one( {name: contract.name}, {$set: contract.dict()}, upsertTrue )4.3 Planner模块实现核心逻辑Planner不直接调用LLM而是封装为可测试的serviceclass Planner: def __init__(self, llm_client, tool_registry): self.llm_client llm_client self.tool_registry tool_registry def generate_plan(self, user_query: str, context: dict) - List[PlanStep]: # 1. 构建prompt注入top-3 relevant tools relevant_tools self.tool_registry.get_relevant_tools(user_query, k3) tool_descriptions \n.join([f- {t[name]}: {t[description]} for t in relevant_tools]) prompt fYou are a planning agent. Given user query and context, output JSON plan. User query: {user_query} Context: {json.dumps(context)} Available tools: {tool_descriptions} Output format: {{steps: [{{type: tool_call, tool_name: ..., input: {{...}}}}, ...]}} # 2. 调用LLM response self.llm_client.chat_completions( messages[{role: user, content: prompt}], temperature0.3, max_tokens512 ) # 3. 解析并校验 try: plan_data json.loads(response.choices[0].message.content) return self._validate_and_normalize_plan(plan_data) except (json.JSONDecodeError, ValidationError) as e: # 触发re-prompt with error context return self._handle_parse_error(e, user_query, context) def _validate_and_normalize_plan(self, plan_data: dict) - List[PlanStep]: # Pydantic校验 plan PlanResponse(**plan_data) steps [] for step in plan.steps: # 根据tool registry校验tool_name存在 tool_def self.tool_registry.get_tool(step.tool_name) if not tool_def: raise ValueError(fTool {step.tool_name} not found in registry) # 校验input符合schema validate_input(step.input, tool_def.input_schema) steps.append(step) return steps4.4 Execution Orchestrator状态机用transitions定义状态流转from transitions import Machine class ExecutionOrchestrator: states [idle, executing, observing, evaluating, retrying, fallbacking, completed, failed] def __init__(self, executor, observator, evaluator): self.machine Machine(modelself, statesstates, initialidle) self.machine.add_transition(start, idle, executing, conditions[_has_next_step], before_execute_step) self.machine.add_transition(observe, executing, observing, before_run_observator) self.machine.add_transition(evaluate, observing, evaluating, before_run_evaluator) self.machine.add_transition(retry, evaluating, retrying, conditions[_should_retry], before_prepare_retry) self.machine.add_transition(fallback, evaluating, fallbacking, conditions[_should_fallback], before_trigger_fallback) self.machine.add_transition(complete, evaluating, completed, conditions[_is_plan_done]) self.machine.add_transition(fail, evaluating, failed, conditions[_is_max_retries_exceeded]) def _execute_step(self): # 调用executor返回raw response self.current_response self.executor.call(self.current_step) def _run_observator(self): # 输入raw response输出Observation self.observation self.observator.observe(self.current_response) def _run_evaluator(self): # 输入Observation输出suggested_action self.action self.evaluator.evaluate(self.observation) def _should_retry(self): return self.observation.suggested_action retry def _should_fallback(self): return self.observation.suggested_action fallback def _is_plan_done(self): return self.plan.is_completed()4.5 Observator实现多层传感器Observator是解耦关键代码清晰分离各层class Observator: def __init__(self, protocol_sensor, data_sensor, business_sensor): self.protocol_sensor protocol_sensor self.data_sensor data_sensor self.business_sensor business_sensor def observe(self, raw_response: Any) - Observation: # 协议层 protocol_obs self.protocol_sensor.check(raw_response) if not protocol_obs.success: return protocol_obs # 数据层 data_obs self.data_sensor.check(raw_response) if not data_obs.success: return data_obs # 业务层 business_obs self.business_sensor.check(raw_response) if not business_obs.success: return business_obs # 全部通过计算quality_score quality_score self._calculate_quality_score(raw_response) return Observation( successTrue, error_type, quality_scorequality_score, suggested_actioncontinue ) def _calculate_quality_score(self, response: dict) - float: # 基于字段完整性、数值合理性、时间新鲜度计算 completeness len([k for k in [price, rating, id] if k in response]) / 3 freshness 1.0 if timestamp in response and is_recent(response[timestamp]) else 0.5 return 0.6 * completeness 0.4 * freshness5. 常见问题与排查速查表基于200线上case问题现象根本原因排查步骤解决方案实测效果Planner反复生成无效tool_nameTool registry未更新或LLM prompt未注入最新tool list1. 检查registry DB中tool数量2. 抓取LLM request payload确认prompt中tool descriptions是否包含新tool① 自动化registry sync脚本② Prompt中添加“Available tools updated at {timestamp}”tool call成功率↑35%Memory漏记关键约束如价格上限STM摘要丢失ID类tokenWM constraints未启用1. 查看STM摘要输出搜索约束关键词2. 检查WM中constraints字段是否为空① STM prompt强制保留ID token② 用户输入含约束时自动提取存入WM.constraints约束违规率↓92%Tool call超时后无重试直接失败Execution Engine未配置retry_policy或Orchestrator未处理timeout1. 查看tool contract中retry_policy字段2. 检查Orchestrator状态机是否定义timeout transition① tool contract中设置max_attempts≥2② Orchestrator添加timeout异常捕获分支tool稳定性↑40%Observation返回suggested_actionfallback但fallback逻辑未触发Fallback handler未注册或Orchestrator状态机缺少fallback transition1. 检查Orchestrator代码中是否有fallback transition2. 查看fallback handler是否在dependency injection中注册① 补全状态机transition② 注册fallback handler为singletonfallback成功率100%Feedback未归因到具体模块盲目微调LLMPipeline未打tagfeedback日志无trace_id关联1. 检查request log中是否含trace_id2. 查看feedback日志是否关联最近tag① middleware自动注入trace_id② feedback endpoint接收trace_id参数问题定位时间↓70%5.1 独家避坑技巧Planner输出校验的“黄金三秒法则”LLM输出后必须在3秒内完成schema校验和tool existence check。超时则立即re-prompt避免阻塞整个pipeline。我们用asyncio.gather并发执行校验实测平均耗时120ms。Memory的“双写”陷阱STM写Redis时若网络抖动失败不能静默忽略。必须启用Redis事务本地cache fallback确保STM最终一致。否则用户看到“刚才说的预算忘了”。Tool Contract的“最小完备”原则input_schema只定义必要字段宁缺毋滥。曾有项目因schema定义了20个字段LLM总填不满导致频繁校验失败。砍到5个核心字段后成功率从45%升至91%。Observation的“降级开关”当传感器集群负载高时自动关闭语义层传感器只保留协议数据层。用feature flag控制避免全链路雪崩。Feedback的“冷启动”问题新agent上线首日feedback极少无法训练。我们预置rule-based feedback用户停留60秒视为positive点击“重新生成”视为negative快速积累初始数据。6. 为什么你总在Agent项目里疲于救火我见过太多团队花三个月搭起agent框架上线后每天处理告警tool timeout、memory leak、planner loop、observation false positive……最后发现问题不在代码而在对六大核心模块边界的模糊认知。你们把LLM当万能胶把Memory当垃圾桶把Tool System当插件市场把Planning当一次性任务把Observation当日志收集把Feedback当用户调研——这就像用螺丝刀当锤子用扳手当剪刀工具没错用法错了。真正的Agent工程是精密的齿轮组LLM的输出必须严丝合缝嵌入Memory的输入槽Memory的摘要必须精准匹配Planner的期待格式Planner的step plan必须完全符合Tool Contract的schemaTool的response必须能被Observation传感器无损解析Observation的suggested_action必须被Orchestrator准确路由Feedback信号必须能反向修正Planner policy。任何一个齿隙过大整个系统就会震颤、异响、最终卡死。所以别再问“ReAct和Planner有什么区别”去检查你的Planner是否真的在生成可执行的step plan还是在输出散文别纠结“哪个agent框架好”去审计你的Tool Contract是否定义了timeout和retry还是只写了description别刷react面试题去复现一次tool call失败后的observe→evaluate→retry全流程。知识不是存在硬盘里的PDF而是你debug时敲下的每一行日志是你看到observation.quality_score0.32时皱起的眉头是你在orchestrator状态机里补上的那个retry transition。最后分享一个小技巧下次启动agent服务别急着测功能先做三件事——手动curl LLM endpoint验证output parser能否100%解析向Redis写入STM用redis-cli inspect key确认摘要不含敏感信息调用一个tool用tcpdump抓包看response是否符合contract output_schema。这三件事做完你才真正站在了Agent工程的起点。之前所有都是热身。 Agent六大核心模块的工程化解耦与实战校准
1. 这不是概念复习是重新校准你对智能体的认知坐标系“从LLM到Agent、Agent的6大核心。这些基础知识你还记得吗”——这句话乍看像知识测验实则是一道分水岭。它不考你背过多少论文标题而是检验你在真实项目里是否真正踩过坑、调过参、改过prompt、修过tool call失败的堆栈。我带过二十多个落地项目从金融风控的决策链路到制造业设备巡检的多模态agent再到政务热线的意图-动作-反馈闭环所有踩过的坑都指向一个事实绝大多数人对Agent的理解还卡在“LLM函数调用”的幻觉层。你可能用过LangChain写过ReAct流程但当memory系统在长对话中开始漏记关键约束条件当你发现tool system返回的JSON字段名和schema定义对不上而错误日志只显示“execution terminated due to error”当你在调试planning模块时发现模型反复把“查天气”拆成“打开浏览器→输入城市名→截图→OCR识别”而不是直接调用weather_api——这些都不是LLM能力不足而是你对Agent六大核心模块的耦合关系、数据流向、容错边界缺乏结构化认知。这六个核心不是并列知识点而是一个有严格依赖顺序的工程栈LLM是引擎但没有Memory系统它就是一台没装导航仪的跑车Tool System是四肢但没有Planner它只会机械执行指令无法理解“先查库存再比价最后下单”背后的业务逻辑Execution Engine是神经系统但没有Observation Loop它就永远不知道自己上一步动作是否成功更不会主动重试或降级。今天不聊抽象定义我们直接进手术室——用真实调试日志、参数配置片段、内存状态快照、工具调用链路图一层层剥开这六个模块怎么咬合、在哪松动、如何加固。如果你正在用Dify、LangChain或自研框架搭建agent这篇就是你的现场检修手册如果你刚学完RAG和Prompt Engineering正准备跨入agent开发这篇就是你绕不开的地形图。所有内容基于2024年Q2真实生产环境验证参数值、超时阈值、重试策略全部标注来源和实测效果拒绝理论空谈。2. 六大核心模块的工程本质不是功能列表而是数据流契约2.1 LLM不只是推理单元而是状态机驱动器很多人把LLM当成“智能大脑”这是最大误区。在agent架构中LLM本质是一个受限状态机Constrained State Machine它的输出必须严格满足下游模块的输入契约。比如Planner模块要求LLM输出JSON格式的action plan其中每个step必须包含typetool_call / think / final_answer、tool_name必须是tool registry中注册的合法名称、input必须符合tool schema定义的类型和范围。我见过太多caseLLM返回call weather_api(cityshanghai)但tool registry里注册的是get_weather_by_city导致tool selector直接抛出KeyError或者input传入字符串30而tool函数签名要求int类型引发TypeError。这不是模型问题是LLM与tool system之间缺乏强契约校验。解决方案不是换更大模型而是构建三层防护Schema-aware Prompt Engineering在system prompt中明确声明tool schema并用示例强制格式。例如You are a planner. Output ONLY valid JSON with keys: steps, reasoning. Each step must have: type: tool_call | think | final_answer, tool_name: one of [get_weather_by_city, search_product, calculate_discount], input: object matching the tools parameter schema. Example: {steps: [{type: tool_call, tool_name: get_weather_by_city, input: {city: Shanghai}}]}Output Parser硬校验不用正则匹配用Pydantic Model做结构化解析。定义BaseStep模型各tool对应子类parse时自动类型转换和字段校验。Fallback机制当parser失败触发re-prompt with error context而非直接报错终止。实测将tool call失败率从12%降至1.7%。提示不要迷信“让LLM自己学会格式”。在生产环境1%的格式错误会导致整个workflow中断。硬校验成本远低于debug时间。2.2 Memory System不是缓存而是上下文仲裁器Memory常被简化为“对话历史存储”这是危险的。真实场景中Memory要解决三个冲突容量冲突token限制、时效冲突新旧信息权重、语义冲突同一实体不同表述。比如客服agent处理投诉“用户说‘昨天订单#12345没发货’3小时后追问‘那个单子现在到哪了’”——Memory必须识别“那个单子”指代#12345且知道3小时前的信息已过期需优先加载最新物流状态。我们采用分层Memory架构Short-term MemorySTM最近5轮对话当前session关键实体订单号、用户ID用LLM summarizer压缩成摘要控制在200token内。关键技巧summarizer prompt中强制要求保留所有ID类token如订单号、产品SKU禁止泛化为“该订单”。Long-term MemoryLTM向量数据库Chroma存储用户历史行为但检索时加双重过滤① 时间衰减因子24h内权重×1.57天内×0.8② 实体链接用NER模型提取当前query中的实体只检索含相同实体的记录。Working MemoryWM当前task专用空间存储planning生成的step plan、tool call中间结果、error recovery状态。WM生命周期单次tasktask结束即销毁避免状态污染。实操痛点STM摘要丢失关键约束。例如用户说“不要推荐价格超过500的手机”摘要变成“用户有预算限制”导致后续推荐违规。解决方案在STM压缩前用规则引擎提取硬约束价格500、品牌∈[Apple, Samsung]单独存入WM的constraints字段planner调用时强制注入。2.3 Tool System不是API集合而是能力契约网Tool System常被当作“函数库”但生产环境要求它具备能力发现Discovery、动态绑定Binding、故障隔离Isolation三大能力。典型问题新增一个汇率查询tool需手动修改planner prompt、更新tool registry、测试所有路径——这不可持续。我们的解法是Tool Contract Schema{ name: get_exchange_rate, description: Get real-time exchange rate between two currencies, input_schema: { type: object, properties: { from_currency: {type: string, enum: [USD, CNY, EUR]}, to_currency: {type: string, enum: [USD, CNY, EUR]} }, required: [from_currency, to_currency] }, output_schema: { type: object, properties: { rate: {type: number}, timestamp: {type: string, format: date-time} } }, metadata: { timeout_ms: 3000, retry_policy: {max_attempts: 2, backoff_factor: 1.5}, criticality: high } }Planner通过读取description和input_schema自动生成调用指令无需硬编码Execution Engine根据timeout_ms和retry_policy自动处理超时重试criticality标记决定error时的降级策略high→fallback to cached ratelow→skip and continue。注意不要把tool实现细节暴露给LLM。曾有项目因prompt中写了“调用阿里云API”LLM在tool不可用时尝试伪造响应。正确做法是tool contract只描述能力不透露技术栈。2.4 Planning Execution不是两步而是闭环反馈环Planning规划和Execution执行常被割裂讲解但实际是带观测反馈的强化学习环Plan → Execute → Observe → Evaluate → Revise Plan。常见错误是把planning当成一次性任务忽略observe环节。以电商比价agent为例Step 1: Planner输出{type:tool_call,tool_name:search_product,input:{keyword:iPhone 15}}Step 2: Executor调用tool返回12个商品Step 3: Observe模块检查返回结果① 是否含price字段缺失则触发data validation error② 是否有至少3个有效价格少于3个则触发re-search with filterStep 4: Evaluation模块计算price variance若标准差50%判定结果噪声大触发revise plan添加price_range filter重试关键设计Observe不是简单success/fail判断而是结构化质量评估。我们定义Observation Schemaclass Observation: success: bool error_type: str # network_timeout, schema_mismatch, business_rule_violation quality_score: float # 0-1, based on data completeness, consistency, freshness suggested_action: str # retry, fallback, revise_plan, abortExecutor根据suggested_action自动路由无需LLM介入。实测将长流程成功率从63%提升至89%。2.5 Observation Loop不是日志而是决策传感器网络Observation Loop常被忽略但它才是agent“活”起来的关键。它要回答三个问题动作是否成功结果是否可信下一步是否需要调整纯靠LLM解析tool返回文本是低效且脆弱的。我们部署多层传感器协议层传感器HTTP status code、gRPC error code拦截网络层失败数据层传感器JSON schema校验、必填字段检查、数值范围验证如price0业务层传感器用规则引擎校验业务逻辑如“订单状态为shipped时物流单号不能为空”语义层传感器轻量级分类模型判断tool返回文本的情感倾向客服场景中negative sentiment触发人工接管。所有传感器输出统一为Observation对象供Evaluation模块消费。例如物流查询tool返回{status: delivered, tracking_number: , estimated_delivery: 2024-06-15}数据层传感器检测到tracking_number为空触发business_rule_violation业务层传感器判定“delivered状态必须有tracking_number”suggested_actionrevise_plan系统自动回溯到上一步调用“补录物流单号”tool。实操心得传感器不是越多越好。我们砍掉所有F10.85的传感器保留4个核心传感器覆盖92%的失败场景。过度监控会拖慢响应速度。2.6 Feedback Learning不是训练而是在线适应引擎Feedback常被等同于“用户点赞/点踩”但生产agent需要实时反馈闭环用户行为点击、停留、跳过、系统指标响应延迟、tool error rate、业务结果转化率、客诉率都要转化为agent的adaptation信号。我们采用三层反馈机制即时层毫秒级用户点击“重新生成”按钮触发re-prompt with context同时记录原response的token usage和LLM confidence scorelogprobs用于优化prompt模板会话层分钟级分析完整对话轨迹识别pattern failure如连续3次tool call失败后用户放弃自动降级到safe mode启用预设rule-based fallback周期层天级聚合日志用离线模型识别tool performance decay如某API错误率周环比上升50%触发tool health check和自动替换。关键创新Feedback不直接微调LLM成本高而是优化planning policy。例如发现用户总在“比价结果页”跳过第2个选项系统推断该选项排序权重偏低在planner的ranking module中动态下调其score权重下次生成plan时自动前置更优选项。3. 六大模块的耦合陷阱与解耦实践3.1 耦合陷阱1LLM与Tool Schema强绑定现象LLM输出的tool_name硬编码在prompt中新增tool需改prompt、重测所有case。根因LLM承担了tool discovery职责违背单一职责原则。解法Tool Registry Dynamic Prompt Injection维护tool registryJSON list每次请求时按context relevance score排序取top-k tool descriptions注入promptLLM只需输出tool_name由tool selector根据registry精确匹配registry更新时自动触发prompt cache invalidation无需人工干预。实测tool迭代发布周期从3天缩短至2小时。3.2 耦合陷阱2Memory与Planner状态混杂现象Planner在生成step时既读STM又读LTM导致plan不稳定LTM检索结果波动影响step顺序。根因Memory未分层隔离Planner缺乏确定性输入源。解法Planner专属Context WindowSTM摘要 WM constraints 当前query → 构成Planner唯一输入LTM检索结果不直接喂给Planner而是存入WM的knowledge字段Planner需显式调用“retrieve_from_ltm”tool获取所有输入token count严格控制确保Planner输出可预测。效果plan一致性从76%提升至94%。3.3 耦合陷阱3Execution与Observation逻辑纠缠现象Executor代码里混杂error handling、retry logic、fallback策略难以维护。根因Execution Engine承担了Observation和Control职责。解法Execution Orchestrator模式Executor只做纯调用call tool → return raw responseObservator独立模块接收raw response输出Observation对象Orchestrator根据Observation.suggested_action决定retry/revise/fallback并调用对应handler。代码结构清晰各模块可单独单元测试。3.4 耦合陷阱4Feedback与LLM训练管道耦合现象为提升效果团队花2周微调LLM但上线后发现90%的问题来自planning policy缺陷。根因Feedback信号未精准定位问题模块。解法模块级Feedback Attribution在pipeline每个节点打tagplanner_output, tool_response, observation_result用户反馈如“结果不准”关联到最近tag定位到planner或tool自动归因若连续3次feedback关联planner_output启动planner policy优化若关联tool_response则触发tool health check。避免盲目调参聚焦真问题。4. 实操从零搭建一个抗压电商Agent含完整配置4.1 环境与依赖我们选择轻量级技术栈避免LangChain等重型框架的黑盒风险LLMQwen2-7B-Instruct本地部署vLLM servingMemoryChromaDBLTM RedisSTM/WMTool SystemFastAPI封装tool endpointscontract schema存MongoDBOrchestratorPython asyncio状态机用transitions库ObservatorPydantic model 规则引擎Durable Rules关键配置文件config.yamlllm: endpoint: http://localhost:8000/v1/chat/completions timeout: 30 max_tokens: 1024 memory: stm: redis_url: redis://localhost:6379/0 ttl_seconds: 3600 ltm: chroma_path: /data/chroma embedding_model: bge-small-zh-v1.5 tool_system: registry_db: mongodb://localhost:27017/tool_registry default_timeout_ms: 5000 retry_policy: max_attempts: 3 backoff_factor: 2.0 orchestrator: max_steps_per_task: 15 max_retries_per_step: 24.2 Tool Contract定义与注册以search_producttool为例定义contract{ name: search_product, description: Search products by keyword, return top 10 with price and rating, input_schema: { type: object, properties: { keyword: {type: string, minLength: 1}, category: {type: string, enum: [phone, laptop, headphone]}, min_rating: {type: number, minimum: 0, maximum: 5} }, required: [keyword] }, output_schema: { type: array, items: { type: object, properties: { id: {type: string}, name: {type: string}, price: {type: number, minimum: 0}, rating: {type: number, minimum: 0, maximum: 5} } } }, metadata: { timeout_ms: 8000, retry_policy: {max_attempts: 2, backoff_factor: 1.5}, criticality: medium } }注册脚本register_tool.pyimport pymongo from pydantic import BaseModel class ToolContract(BaseModel): name: str description: str input_schema: dict output_schema: dict metadata: dict # 从JSON文件加载contract with open(search_product_contract.json) as f: contract_data json.load(f) contract ToolContract(**contract_data) # 写入MongoDB client pymongo.MongoClient(mongodb://localhost:27017/) db client[tool_registry] collection db[contracts] collection.update_one( {name: contract.name}, {$set: contract.dict()}, upsertTrue )4.3 Planner模块实现核心逻辑Planner不直接调用LLM而是封装为可测试的serviceclass Planner: def __init__(self, llm_client, tool_registry): self.llm_client llm_client self.tool_registry tool_registry def generate_plan(self, user_query: str, context: dict) - List[PlanStep]: # 1. 构建prompt注入top-3 relevant tools relevant_tools self.tool_registry.get_relevant_tools(user_query, k3) tool_descriptions \n.join([f- {t[name]}: {t[description]} for t in relevant_tools]) prompt fYou are a planning agent. Given user query and context, output JSON plan. User query: {user_query} Context: {json.dumps(context)} Available tools: {tool_descriptions} Output format: {{steps: [{{type: tool_call, tool_name: ..., input: {{...}}}}, ...]}} # 2. 调用LLM response self.llm_client.chat_completions( messages[{role: user, content: prompt}], temperature0.3, max_tokens512 ) # 3. 解析并校验 try: plan_data json.loads(response.choices[0].message.content) return self._validate_and_normalize_plan(plan_data) except (json.JSONDecodeError, ValidationError) as e: # 触发re-prompt with error context return self._handle_parse_error(e, user_query, context) def _validate_and_normalize_plan(self, plan_data: dict) - List[PlanStep]: # Pydantic校验 plan PlanResponse(**plan_data) steps [] for step in plan.steps: # 根据tool registry校验tool_name存在 tool_def self.tool_registry.get_tool(step.tool_name) if not tool_def: raise ValueError(fTool {step.tool_name} not found in registry) # 校验input符合schema validate_input(step.input, tool_def.input_schema) steps.append(step) return steps4.4 Execution Orchestrator状态机用transitions定义状态流转from transitions import Machine class ExecutionOrchestrator: states [idle, executing, observing, evaluating, retrying, fallbacking, completed, failed] def __init__(self, executor, observator, evaluator): self.machine Machine(modelself, statesstates, initialidle) self.machine.add_transition(start, idle, executing, conditions[_has_next_step], before_execute_step) self.machine.add_transition(observe, executing, observing, before_run_observator) self.machine.add_transition(evaluate, observing, evaluating, before_run_evaluator) self.machine.add_transition(retry, evaluating, retrying, conditions[_should_retry], before_prepare_retry) self.machine.add_transition(fallback, evaluating, fallbacking, conditions[_should_fallback], before_trigger_fallback) self.machine.add_transition(complete, evaluating, completed, conditions[_is_plan_done]) self.machine.add_transition(fail, evaluating, failed, conditions[_is_max_retries_exceeded]) def _execute_step(self): # 调用executor返回raw response self.current_response self.executor.call(self.current_step) def _run_observator(self): # 输入raw response输出Observation self.observation self.observator.observe(self.current_response) def _run_evaluator(self): # 输入Observation输出suggested_action self.action self.evaluator.evaluate(self.observation) def _should_retry(self): return self.observation.suggested_action retry def _should_fallback(self): return self.observation.suggested_action fallback def _is_plan_done(self): return self.plan.is_completed()4.5 Observator实现多层传感器Observator是解耦关键代码清晰分离各层class Observator: def __init__(self, protocol_sensor, data_sensor, business_sensor): self.protocol_sensor protocol_sensor self.data_sensor data_sensor self.business_sensor business_sensor def observe(self, raw_response: Any) - Observation: # 协议层 protocol_obs self.protocol_sensor.check(raw_response) if not protocol_obs.success: return protocol_obs # 数据层 data_obs self.data_sensor.check(raw_response) if not data_obs.success: return data_obs # 业务层 business_obs self.business_sensor.check(raw_response) if not business_obs.success: return business_obs # 全部通过计算quality_score quality_score self._calculate_quality_score(raw_response) return Observation( successTrue, error_type, quality_scorequality_score, suggested_actioncontinue ) def _calculate_quality_score(self, response: dict) - float: # 基于字段完整性、数值合理性、时间新鲜度计算 completeness len([k for k in [price, rating, id] if k in response]) / 3 freshness 1.0 if timestamp in response and is_recent(response[timestamp]) else 0.5 return 0.6 * completeness 0.4 * freshness5. 常见问题与排查速查表基于200线上case问题现象根本原因排查步骤解决方案实测效果Planner反复生成无效tool_nameTool registry未更新或LLM prompt未注入最新tool list1. 检查registry DB中tool数量2. 抓取LLM request payload确认prompt中tool descriptions是否包含新tool① 自动化registry sync脚本② Prompt中添加“Available tools updated at {timestamp}”tool call成功率↑35%Memory漏记关键约束如价格上限STM摘要丢失ID类tokenWM constraints未启用1. 查看STM摘要输出搜索约束关键词2. 检查WM中constraints字段是否为空① STM prompt强制保留ID token② 用户输入含约束时自动提取存入WM.constraints约束违规率↓92%Tool call超时后无重试直接失败Execution Engine未配置retry_policy或Orchestrator未处理timeout1. 查看tool contract中retry_policy字段2. 检查Orchestrator状态机是否定义timeout transition① tool contract中设置max_attempts≥2② Orchestrator添加timeout异常捕获分支tool稳定性↑40%Observation返回suggested_actionfallback但fallback逻辑未触发Fallback handler未注册或Orchestrator状态机缺少fallback transition1. 检查Orchestrator代码中是否有fallback transition2. 查看fallback handler是否在dependency injection中注册① 补全状态机transition② 注册fallback handler为singletonfallback成功率100%Feedback未归因到具体模块盲目微调LLMPipeline未打tagfeedback日志无trace_id关联1. 检查request log中是否含trace_id2. 查看feedback日志是否关联最近tag① middleware自动注入trace_id② feedback endpoint接收trace_id参数问题定位时间↓70%5.1 独家避坑技巧Planner输出校验的“黄金三秒法则”LLM输出后必须在3秒内完成schema校验和tool existence check。超时则立即re-prompt避免阻塞整个pipeline。我们用asyncio.gather并发执行校验实测平均耗时120ms。Memory的“双写”陷阱STM写Redis时若网络抖动失败不能静默忽略。必须启用Redis事务本地cache fallback确保STM最终一致。否则用户看到“刚才说的预算忘了”。Tool Contract的“最小完备”原则input_schema只定义必要字段宁缺毋滥。曾有项目因schema定义了20个字段LLM总填不满导致频繁校验失败。砍到5个核心字段后成功率从45%升至91%。Observation的“降级开关”当传感器集群负载高时自动关闭语义层传感器只保留协议数据层。用feature flag控制避免全链路雪崩。Feedback的“冷启动”问题新agent上线首日feedback极少无法训练。我们预置rule-based feedback用户停留60秒视为positive点击“重新生成”视为negative快速积累初始数据。6. 为什么你总在Agent项目里疲于救火我见过太多团队花三个月搭起agent框架上线后每天处理告警tool timeout、memory leak、planner loop、observation false positive……最后发现问题不在代码而在对六大核心模块边界的模糊认知。你们把LLM当万能胶把Memory当垃圾桶把Tool System当插件市场把Planning当一次性任务把Observation当日志收集把Feedback当用户调研——这就像用螺丝刀当锤子用扳手当剪刀工具没错用法错了。真正的Agent工程是精密的齿轮组LLM的输出必须严丝合缝嵌入Memory的输入槽Memory的摘要必须精准匹配Planner的期待格式Planner的step plan必须完全符合Tool Contract的schemaTool的response必须能被Observation传感器无损解析Observation的suggested_action必须被Orchestrator准确路由Feedback信号必须能反向修正Planner policy。任何一个齿隙过大整个系统就会震颤、异响、最终卡死。所以别再问“ReAct和Planner有什么区别”去检查你的Planner是否真的在生成可执行的step plan还是在输出散文别纠结“哪个agent框架好”去审计你的Tool Contract是否定义了timeout和retry还是只写了description别刷react面试题去复现一次tool call失败后的observe→evaluate→retry全流程。知识不是存在硬盘里的PDF而是你debug时敲下的每一行日志是你看到observation.quality_score0.32时皱起的眉头是你在orchestrator状态机里补上的那个retry transition。最后分享一个小技巧下次启动agent服务别急着测功能先做三件事——手动curl LLM endpoint验证output parser能否100%解析向Redis写入STM用redis-cli inspect key确认摘要不含敏感信息调用一个tool用tcpdump抓包看response是否符合contract output_schema。这三件事做完你才真正站在了Agent工程的起点。之前所有都是热身。