1. 为什么“七要素”不是设计清单，而是故障排查地图

我第一次在团队里讲 AI Agent 架构时，画了张漂亮的七要素图：Memory、Planning、Action、Observation、Tool Use、Reasoning、Self-Correction——每个框都配了图标，还标了箭头循环。结果上线第三天，用户反馈“智能体卡在‘思考中’不动了”，运维查日志只看到一串重复的LLM call timeout；第五天，另一个任务跑着跑着开始反复调用同一个天气 API，直到被限频熔断；第七天，客户问：“你们说的‘自主容错’在哪？怎么连输入里多打了个句号都直接崩？”

那一刻我才意识到：市面上几乎所有讲“Agent 七要素”的文章，都在教你怎么画一张正确但无用的架构图，而不是告诉你——当真实请求涌进来、模型开始胡说、工具突然失联、内存溢出、token 耗尽、网络抖动、用户中途改指令……这七个模块里，哪个环节最先扛不住？它为什么会扛不住？你该先看哪行日志？该加什么 guardrail 而不是加个 retry？

“七要素”真正的价值，根本不是教学大纲，而是一张工程级故障定位地图。它对应的是七个必须被显式设计、独立压测、单独监控、可开关降级的决策点。比如：

• “Planning” 不是“让模型写个 step-by-step plan”，而是决策点1：是否启动规划？
  → 输入长度超 8k？跳过规划，直奔 Action；
  → 用户指令含“马上”“立刻”“别想太多”？强制 bypass planning loop；
  → 上一轮规划失败超2次？切换到 fallback planner（比如规则引擎）。

• “Memory” 不是“存点聊天记录”，而是决策点2：本次推理该读哪段记忆？读多少？以什么格式注入？
  → 不是“把全部历史喂给 LLM”，而是：
  ▪️ 近3轮对话摘要（用轻量 summarizer 提取 action intent）；
  ▪️ 本次任务相关的知识片段（从向量库 top-3 检索，带 score threshold）；
  ▪️ 上一轮失败的 error trace（只保留 error code + failed tool name + input hash）；
  ▪️ 全部不塞进 prompt，而是分层注入：摘要放 system prompt，知识片段放 user message 前置块，error trace 放最后 special instruction。

这才是“工程实现”的起点：把抽象概念翻译成可配置、可拦截、可熔断、可灰度的代码决策点。后面所有环节——从 token 预估、tool schema 校验、observation 解析容错，到 self-correction 的触发阈值——全建立在这个基础上。没把这个想透，后面堆再多 LangChain、LangGraph、LlamaIndex，也只是在沙上建塔。

提示：别急着选框架。先拿纸笔，对着你手头一个真实业务场景（比如“帮用户订会议室+同步日历+发提醒邮件”），逐个问自己：

• 如果 Planning 模块返回空数组，系统该沉默失败？还是 fallback 到单步执行？
• 如果 Memory 检索返回 5 条相似度 0.62 的结果，该全用？还是只取最高那条并加 confidence flag？
• 如果 Observation 解析出“{'status': 'success', 'data': null}”，算成功还是失败？要不要重试？重试几次？
  把这七个问题的答案写下来，比读十篇架构图都有用。

2. 决策点3：Tool Calling 不是函数调用，是协议协商与契约管理

绝大多数 Agent 教程把 Tool Calling 讲成“LLM 输出 JSON，程序 parse 后执行函数”。这在 demo 里能跑通，在生产环境里是定时炸弹。我见过三个最典型的崩塌现场：

• Schema 漂移：LLM 调用search_weather(city: str)，但实际传入{"city": "上海,中国"}—— 后端服务要求 city 必须是 ISO 国家码+城市名（如"CN-Shanghai"），结果返回 400，Agent 却当成“天气查询成功”，继续往下走；
• 语义歧义：LLM 调用book_meeting(start_time: str, duration: int)，传入{"start_time": "3pm", "duration": 30}—— 后端解析为“今天下午3点”，但用户本意是“明天下午3点”，没人校验时间上下文；
• 副作用失控：LLM 调用send_email(to: str, subject: str, body: str)，body 里含未过滤的用户原始输入，结果一封带 XSS payload 的邮件发给了全组。

根本原因在于：Tool 不是函数，是微服务接口；LLM 不是程序员，是不可靠的协作者。工程上必须建立三层契约：

2.1 协议层：Tool Description 必须带机器可读的约束声明

不能只写"Get current weather for a city"，而要像 OpenAPI 一样声明：

name: search_weather description: Get current weather for a city. City must be in format "COUNTRY-CODE-CITY_NAME" (e.g., "US-NewYork"). parameters: city: type: string pattern: ^[A-Z]{2}-[A-Za-z]+$ required: true description: ISO country code + city name, dash-separated.

→ 实际落地时，我们用 Pydantic Model 自动生成此描述，并在 LLM 调用前做 runtime schema check（不是 parse 失败才报错，而是提前 reject 不合规的 candidate）。

2.2 执行层：Call Wrapper 必须封装重试、降级、熔断逻辑

我们不用tool(**args)直接调用，而是统一走ToolExecutor.execute(tool_name, args, context)：

• 第一次失败（network timeout）→ 自动重试 1 次，间隔 200ms；
• 第二次失败（4xx）→ 触发validate_input钩子，检查 args 是否符合 pattern，若不符则返回 structured error 给 LLM 修正；
• 第三次失败（5xx 或连续 timeout）→ 熔断 60 秒，返回 fallback response（如"天气服务暂不可用，已为您记录需求，稍后重试"）；
• 若 fallback 也失败 → 触发全局降级：跳过此 tool，进入 human-in-the-loop 流程。

2.3 结果层：Observation Parsing 必须定义 success/failure 的明确边界

不是“有 response 就算成功”。我们定义：

注意：failure 的返回值不是字符串"call failed"，而是结构化对象：

{"tool": "search_weather", "error_code": "INVALID_INPUT", "suggestion": "请确认城市格式为 'CN-Shanghai'"}

这样 LLM 才能真正理解问题，而不是瞎猜。我们实测发现，加了这个结构化 error 后，LLM 自我修正成功率从 37% 提升到 89%。

3. 决策点4：Observation 解析不是文本提取，是语义归一化与可信度标注

LLM 的输出是“自然语言”，Tool 的返回是“结构化数据”，但中间的 Observation 环节，常被当成透明管道——把 JSON 字符串原样塞回 prompt。这是高危操作。真实场景中，Observation 有三大陷阱：

• 格式污染：API 返回{"data": {"temp": 25.3, "unit": "°C"}}，LLM 却在 prompt 里看到"data": {"temp": 25.3, "unit": "°C"}—— 引号、冒号、换行全在，token 消耗暴增，且 LLM 容易把°C当成乱码忽略；
• 语义失真：天气 API 返回"condition": "Partly cloudy"，LLM 却理解成“多云”，而实际业务要求区分“晴/多云/阴/雨”，必须映射为标准枚举；
• 可信度缺失：搜索 API 返回 10 条结果，但其中 3 条来自低权重源，2 条是广告，LLM 却一视同仁地引用。

我们的解法是：Observation 层必须做三件事——清洗、归一、打标。

3.1 清洗：从 raw response 到 minimal semantic payload

不传整个 response body，只提取关键字段并标准化格式：

# 原始 response { "code": 0, "message": "success", "data": { "temperature": 25.3, "weather": "Partly cloudy", "humidity": "65%", "wind_speed": "12 km/h" } } # 清洗后注入 prompt 的 observation { "weather": "partly_cloudy", # 归一为 snake_case 枚举 "temperature_celsius": 25.3, # 单位显式标注，数值转 float "humidity_percent": 65.0, # 去掉 % 符号，转数字 "wind_speed_kmh": 12.0 # 同上 }

→ 这步由ObservationNormalizer统一处理，每个 tool 对应一个 normalizer class，确保 LLM 看到的永远是干净、一致、无歧义的字段。

3.2 归一：将 domain-specific value 映射为通用语义

比如会议系统返回{"status": "confirmed"}，CRM 系统返回{"state": "booked"}，邮件系统返回{"result": "sent"}—— 在 Observation 层全部归一为{"booking_status": "confirmed"}。这样 LLM 无需学习不同系统的术语，只需理解统一语义。

3.3 打标：为每条 observation 附加可信度分数（confidence score）

不是简单标记“可信/不可信”，而是量化：

→ LLM 在后续推理中，会优先采信 confidence > 0.7 的 observation；若所有 observation confidence < 0.5，则自动触发 human escalation。

实操心得：别让 LLM 自己判断“这条信息可靠吗”。人类定义规则，机器执行规则。我们曾让 LLM 给 observation 打分，结果它给一条明显错误的天气数据打了 0.95 分——因为 response 里有“权威”“实时”“官方”三个词。信任必须来自可验证的来源，而非文字修饰。

4. 决策点5：Self-Correction 不是重试，是状态机驱动的主动干预

“Self-Correction” 是最被神化的概念。很多教程说：“让 Agent 发现错误就重试”。但真实世界里，重试是最廉价的错误处理，也是最危险的默认行为。我们线上一个订餐 Agent 曾因重试逻辑失控，30 秒内向餐厅 API 发了 17 次下单请求，导致用户被扣 17 笔钱。

真正的 Self-Correction，是基于当前 execution state 的主动决策，它必须回答三个问题：

1. 这次失败是偶发（network glitch）还是必然（逻辑缺陷）？
2. 重试能否解决？如果不能，该降级到什么方案？
3. 用户是否需要知情？以什么方式告知？

我们用状态机实现，而非 if-else 堆砌：

graph TD A[Start] --> B{Error Type?} B -->|Timeout/5xx| C[Retry once with jitter] B -->|4xx/Schema Error| D[Validate & Correct Input] B -->|LLM Output Invalid| E[Re-prompt with stricter constraints] B -->|Tool Result Inconsistent| F[Cross-check with fallback source] C --> G{Success?} G -->|Yes| H[Continue] G -->|No| I[Trigger fallback] D --> J{Input fixable?} J -->|Yes| K[Auto-correct & retry] J -->|No| L[Ask user clarify]

关键细节：

• Error Type 分类必须细粒度：不是“HTTP error”，而是TOOL_TIMEOUT,TOOL_SCHEMA_VIOLATION,LLM_OUTPUT_MALFORMED,OBSERVATION_INCONSISTENT—— 每种对应不同 handler；
• Retry 有严格条件：仅对TOOL_TIMEOUT且retry_count < 1允许；每次 retry 加 jitter（100ms~500ms 随机），避免雪崩；
• Auto-correct 有安全边界：比如 LLM 传{"city": "shanghai"}，normalizer 可 auto-correct 为"CN-Shanghai"；但若传{"city": "火星"}，则拒绝修正，直接报错；
• Fallback 不是兜底，是预案：每个 tool 必须配 fallback：
  ▪️search_weatherfallback → 本地缓存（last 1h 数据）；
  ▪️book_meetingfallback → 日历空闲时段 API（不依赖具体会议室系统）；
  ▪️send_emailfallback → 企业微信消息（降级通道）。

踩坑实录：我们最初用 LLM 自我反思做 correction，prompt 是 “请检查上一步是否有错误，如有请修正”。结果 LLM 在 82% 的 case 里都说“没有错误”，哪怕 observation 明显是{"error": "not_found"}。后来改成 rule-based detection：当 observation 包含"error"key，或 LLM output 中出现"I don't know"、"unable to"等 trigger phrase，立即进入 correction flow。准确率从 18% 跃升至 99.2%。

5. 决策点6：Token 管控不是预算分配，是动态流控与成本感知路由

“AI Agent 怎么扛并发”——热搜第一的问题，本质是 token 管控失效。很多人以为加个 Redis 计数器、设个 rate limit 就行。但真实瓶颈不在 QPS，而在token throughput。一个复杂 Agent 流程（Plan → Tool1 → Obs1 → Tool2 → Obs2 → …）可能消耗 3000+ tokens，而一个简单问答只用 200 tokens。如果按请求计费，高 token 请求会饿死低 token 请求。

我们的方案是：Token-aware Request Router + Per-Step Budgeting。

5.1 Token 预估：不是 guess，是 model-driven estimation

不用len(prompt)粗略算，而是训练轻量 regressor：

• 输入：prompt template + variable length（如 history turns, tool result size）；
• 输出：预测 token count（MAE < 15 tokens）；
• 模型：XGBoost（训练快、解释性强），特征包括：
  ▪️ history turn count；
  ▪️ tool result JSON depth；
  ▪️ 最长字符串字段长度；
  ▪️ 是否启用 memory retrieval（是/否）；
  ▪️ 当前 LLM temperature（影响输出长度）。

→ 每个请求进来，先 run estimator，得到predicted_tokens。

5.2 动态流控：按 token 而非请求数限流

用令牌桶（token bucket），但桶容量单位是token-seconds（token × time）：

• 桶容量：10000 token-seconds / minute；
• 每个请求消耗：predicted_tokens × response_time_seconds；
• 若请求 predicted_tokens=2000，预估响应时间 2s，则消耗 4000 token-seconds；
• 若桶剩余 < 4000，拒绝请求，返回"系统繁忙，请稍后重试（预计等待 12s）"。

→ 这样，一个 200-token 的快速请求不会被 3000-token 的慢请求堵死。

5.3 成本感知路由：把请求导向性价比最高的 LLM

不是所有任务都需 GPT-4。我们维护 LLM profile 表：

→ Router 根据predicted_tokens和任务类型（is_planning_task,is_tool_call,is_summarize）选择 model：

• predicted_tokens < 500且is_tool_call→claude-3-haiku；
• predicted_tokens > 2000且is_planning_task→gpt-4-turbo；
• 其他 →llama3-70b（成本最低）。

关键经验：token 预估不准，比没预估更危险。我们上线初期用固定系数（prompt length × 1.3），结果高估 40%，大量请求被误拒；低估 30%，GPU 显存爆满 OOM。最终靠 real-time feedback loop 修正：每次请求 actual_tokens 与 predicted_tokens 的差值，实时更新 regressor weights。现在误差稳定在 ±8 tokens。

6. 决策点7：Memory 管理不是存储，是生命周期编排与上下文蒸馏

把 Memory 当数据库用，是 Agent 工程最大误区。我们曾有个客服 Agent，内存存了 200 轮对话，每次推理都把全部历史塞进 prompt，结果 token 溢出、响应变慢、LLM 开始胡编。后来发现，90% 的对话中，LLM 真正用到的只有最近 3 轮 + 1 条关键知识。

Memory 的工程核心，是Context Distillation：从海量信息中，实时蒸馏出本次推理所需的最小有效上下文。

6.1 生命周期分层：不是“存”和“删”，是“热/温/冷”三级流转

• Hot Memory（< 5min）：当前 session 的 active context，存于 Redis，带 TTL；
• Warm Memory（5min ~ 24h）：用户 profile、偏好、近期任务状态，存于 PostgreSQL，可关联查询；
• Cold Memory（> 24h）：归档日志、审计记录，存于 S3，只读。

→ 每次推理，只从 Hot + Warm 中提取，Cold 不参与实时决策。

6.2 蒸馏策略：三阶段筛选，不是关键词匹配

1. Turn-level filtering：

   • 丢弃所有systemrole messages；
   • 丢弃usermessages 中无 action intent 的（如“你好”“谢谢”）；
   • 保留assistantmessages 中含 tool call 或 final answer 的。

2. Semantic summarization：

   • 对保留的 turns，用轻量 summarizer（T5-small）生成摘要：
     "User asked to book meeting on Friday, assistant checked calendar and found slot 2-3pm, confirmed with user."
   • 摘要长度严格限制 ≤ 128 tokens。

3. Knowledge grounding：

   • 基于当前 query，从 Warm Memory 检索相关知识：
     ▪️ 用户历史 booking 偏好（会议室大小、是否需投影仪）；
     ▪️ 企业 policy（会议超 1h 需 manager approval）；
     ▪️ 设备状态（投影仪维修中，本周不可用）；
   • 检索结果带 relevance score，只取 score > 0.6 的 top-2 条。

→ 最终注入 prompt 的 memory，不超过 300 tokens，且 100% 与当前任务强相关。

6.3 安全隔离：Memory 不是共享池，是租户级沙箱

• 每个用户 session 有独立 memory namespace；
• 不同业务线（客服/销售/HR）memory 物理隔离；
• 敏感字段（手机号、身份证）自动 redact，替换为<PHONE>，且 redaction 规则可配置。

实测对比：未蒸馏时，平均 prompt length 2850 tokens，P95 响应 4.2s；蒸馏后，平均 210 tokens，P95 响应 0.8s，LLM 准确率提升 22%（因噪声减少）。更重要的是，内存泄漏风险归零——Hot Memory TTL 到期自动清理，无需人工干预。

7. 工程落地 checklist：七个决策点，每个都必须有 concrete implementation

讲完理论，给一份我们团队用的Agent 工程落地 checklist。不是“建议”，是上线前必须填满的项。少一项，就等于埋一颗雷。

最后分享一个血泪教训：我们曾认为 “Observation Normalizer” 是个 trivial task，让 junior engineer 用正则写了三天。上线后发现，天气 API 的{"temp": "25.3°C"}被正则错切成{"temp": "25.3"}，丢了单位，LLM 把 25.3 当成华氏度，结论是“极寒天气”。后来重写为 Pydantic Model + custom validator，加了 unit check，问题根除。

所以记住：Agent 工程没有 trivial part。每个决策点，都是生产环境的守门人。画七要素图只要十分钟，让七个决策点在高并发、低延迟、强一致的环境下稳稳跑起来，需要三个月的迭代、压测、调优。别省这个时间。