企业官网建设流程全解析

1. 项目概述：Root Agent 不是“超级大脑”，而是多 Agent 系统的调度中枢

你有没有遇到过这样的场景：一个 Agent 能写周报，另一个能查数据库，第三个能调用天气 API，但三者之间像三座孤岛——你得手动把周报草稿复制给数据库 Agent 去查数据，再把结果粘贴给天气 Agent 补充实时信息，最后自己拼成终稿？这不是智能，这是智能时代的“人肉管道工”。而Root Agent的本质，恰恰就是来终结这种低效协作的。它不是要取代其他 Agent，而是像交响乐团的指挥家：不拉小提琴，也不吹长笛，但清楚每件乐器何时入场、奏什么音高、强弱如何变化——它负责理解用户原始意图、拆解任务逻辑、动态选择并协调下游 Agent、统一收口输出。这个角色在多 Agent 架构中不可替代，尤其当系统规模从“单点工具”迈向“可扩展服务”时，Root Agent 就是那个让整个系统不散架的骨架。

我第一次在真实项目里落地 Root Agent，是在一个企业级知识助手重构中。原有单 Agent 模型在处理“对比分析近三年销售数据与行业政策变动对华东区毛利率的影响”这类复合查询时，准确率跌到 42%。原因很直接：它试图在一个推理链里完成语义解析、SQL 生成、政策文档检索、归因建模四件事，中间任一环节出错，全盘崩塌。引入 Root Agent 后，我们把它拆成四个原子 Agent：Query Decomposer（问题拆解）、SQL Generator（结构化查询）、Policy Retriever（政策文档召回）、Margin Analyst（归因计算）。Root Agent 只做三件事：① 接收用户原始 query；② 判定是否需多步协同（比如含“对比”“影响”“原因”等关键词）；③ 按预设 workflow 编排子 Agent 执行顺序，并传递上下文。实测下来，复杂 query 准确率升至 89%，响应耗时反而下降 31%，因为每个子 Agent 都在自己最擅长的领域专注执行，没有“跨域思考”的认知损耗。

这里必须划清一个关键认知边界：Root Agent ≠ 更强的 LLM，≠ 更大的模型，≠ 更多的 prompt 工程。它的核心价值在于架构设计权——把“让一个模型硬扛所有事”的蛮力模式，转向“让多个专业模型各司其职”的工程模式。而落地难点，从来不在模型能力，而在如何让 Agent 之间说同一种语言、遵守同一套契约、共享同一份上下文。这正是claude.md文件存在的底层逻辑：它不是一份文档，而是一份Agent 间的运行时契约（Runtime Contract）。就像两个程序员合作开发，光有接口文档不够，还得约定错误码含义、重试策略、超时阈值、日志格式——claude.md就是为 Agent 世界写的这份“工程协议”。它定义了 Agent 的能力边界（能做什么）、输入规范（要什么参数）、输出结构（返回什么格式）、失败反馈（错在哪）、调用约束（能否并发、最大重试次数）。没有它，再多的 Agent 也只是散装零件；有了它，才能组装成可维护、可测试、可演进的系统。所以本指南不讲“怎么选大模型”，只聚焦一件事：如何用最小必要配置，让 Root Agent 真正跑起来，并让下游 Agent 愿意听它指挥。

2. 架构设计与方案选型：为什么 Root Agent 必须“轻量”且“契约驱动”

2.1 Root Agent 的三种常见误入歧途

很多团队在落地 Root Agent 时，第一反应是堆算力、上大模型、写超长 system prompt。我见过三个典型失败案例，都是踩过坑后才明白“轻量”和“契约”的分量：

• 陷阱一：“Root = 最强模型”幻觉
  某金融客户坚持用 72B 参数模型做 Root Agent，认为“它最聪明，应该最会调度”。结果上线后发现，90% 的调度决策其实只需要判断“当前 query 是否含时间范围”“是否涉及多个数据源”“是否需要人工复核”三类简单规则。用大模型做这些判断，就像用航空母舰去送快递——延迟高、成本高、还容易因微小扰动（如 prompt 微调）导致调度逻辑漂移。我们后来换成 7B 模型 + 规则引擎混合调度，稳定性提升 3 倍，推理成本降为原来的 1/12。

• 陷阱二：“万能 Router”执念
  有人试图让 Root Agent 动态学习“哪个子 Agent 适合处理哪类 query”，训练一个专用分类器。这看似智能，实则埋下巨大隐患：当新增一个子 Agent（比如加了个财报解读 Agent），整个分类器就得重新训练、验证、上线，迭代周期长达 2 周。而业务需求往往要求“今天提需求，明天就上线”。我们最终采用显式路由表（Explicit Routing Table）：在config.yaml中明确定义query_pattern: agent_name映射，例如r".*毛利率.*影响.*": "margin_analyst"。新增 Agent 只需在 config 中加一行，5 分钟内生效，运维心智负担极低。

• 陷阱三：“自由协作”浪漫主义
  初期设计曾允许子 Agent 在执行中“自主决定是否调用其他 Agent”，形成网状调用。结果一次线上故障排查花了 3 天：某个子 Agent 因网络抖动失败，触发了它内部的“备用调用逻辑”，又去调用了另一个未授权的 Agent，后者又触发了第三层调用……最终形成调用风暴，压垮了整个服务。我们强制改为单向流水线（Linear Pipeline）：Root Agent 是唯一调度者，子 Agent 只能向 Root 返回结果或错误，不能越级调用。所有分支逻辑（如“查不到数据则转人工”）都在 Root 层统一决策。虽然牺牲了一点理论上的灵活性，但换来的是可预测性、可观测性和可调试性——这对生产环境而言，是远比“理论上更智能”更重要的东西。

2.2 为什么claude.md是契约核心，而非config.yaml

很多人混淆config.yaml和claude.md的分工。简单说：config.yaml是静态部署配置，claude.md是动态运行契约。这个区分决定了整个系统的可维护性。

• config.yaml解决的是“谁在哪”：
  它记录 Agent 的部署地址（http://sql-gen:8000）、健康检查端点（/health）、基础超时（timeout: 30s）、重试次数（retries: 2）。这些是运维视角的配置，修改后通常需要重启服务或热加载。

• claude.md解决的是“谁能干啥、怎么干、干不好怎么办”：
  它以 Markdown 格式声明 Agent 的能力契约。例如sql_generator的claude.md文件内容可能如下：

# SQL Generator Agent ## Capabilities - Generate valid SQL queries for PostgreSQL databases - Support JOIN, WHERE, GROUP BY, ORDER BY clauses - **Does NOT support** DDL (CREATE/DROP), subqueries in SELECT clause, or non-deterministic functions (e.g., NOW()) ## Input Schema ```json { "user_query": "string, required, e.g. 'Show top 5 customers by revenue in Q3 2023'", "schema_context": "string, optional, description of relevant DB tables and columns", "allowed_tables": ["customers", "orders", "products"] }

Output Schema

{ "sql": "string, the generated SQL query", "explanation": "string, plain-language explanation of what the query does", "confidence": "number, 0.0-1.0, confidence in correctness" }

Error Handling

• Ifuser_queryis ambiguous: return error codeERR_AMBIGUOUS_QUERY, withsuggestionfield proposing clarifying questions
• Ifallowed_tablesrestriction violated: return error codeERR_TABLE_NOT_ALLOWED
• Max execution time: 8 seconds (hard limit)

这个文件的关键在于： ① **能力声明是机器可读的**：Root Agent 可以解析 `Capabilities` 部分，自动过滤掉不支持 `JOIN` 的 Agent； ② **输入输出是结构化的**：Root Agent 能自动生成符合 `Input Schema` 的 JSON 请求体，无需硬编码字段名； ③ **错误是标准化的**：`ERR_AMBIGUOUS_QUERY` 这种错误码，让 Root Agent 能统一触发“追问用户”流程，而不是每个子 Agent 自己写一套错误处理逻辑。 我们曾做过对比实验：当 `claude.md` 缺失时，Root Agent 对子 Agent 的调用成功率仅 63%，因为大量时间花在适配不同 Agent 的“方言”上（有的返回 `{"error": "xxx"}`，有的返回 `{"code": 500, "msg": "xxx"}`，有的甚至直接抛 HTTP 500）。引入 `claude.md` 后，调用成功率稳定在 99.2%，且新增一个 Agent 的集成时间从平均 8 小时压缩到 45 分钟以内——因为 Root Agent 的调度逻辑不再需要为每个新成员重写，只需读取它的 `claude.md` 即可自动适配。 ### 2.3 为什么 React 架构不是必须，但“工具调用”范式是刚需 网络热词里常问：“调用 MCP 工具需要 Agent 以 React 的架构调用吗？”这个问题背后，是对 Agent 交互范式的误解。React（ReAct）是一种**推理时的 prompt 技巧**，用于引导 LLM 在生成答案前先“思考步骤、调用工具、观察结果、再综合”，它解决的是“单个 LLM 如何使用工具”的问题。而 Root Agent 架构解决的是“多个独立 Agent 如何协同”的问题——两者维度不同，不互斥，但也不等价。 在 Root Agent 场景下，“工具调用”范式之所以成为刚需，是因为它定义了**跨 Agent 通信的最小公约数**。无论下游 Agent 是用 LangChain、LlamaIndex 还是自研框架实现，只要它遵循“接收标准输入 → 执行 → 返回标准输出/错误”的范式，Root Agent 就能无差别调度。我们强制所有子 Agent 实现以下统一接口： ```bash # 标准调用方式（HTTP POST） curl -X POST http://<agent-host>/invoke \ -H "Content-Type: application/json" \ -d '{ "input": { ... }, # 严格按 claude.md 的 Input Schema "context": { # Root Agent 注入的全局上下文 "session_id": "sess_abc123", "user_id": "usr_xyz789", "trace_id": "trc_def456" } }'

这个接口的设计哲学是：剥离模型细节，聚焦契约履行。子 Agent 内部用什么模型、什么框架、什么推理策略，Root Agent 完全不关心；它只关心“你能不能按claude.md说的那样，正确解析input字段，并在output或error字段里返回符合约定的内容”。这就让技术栈异构成为可能：SQL Generator 用微调的 CodeLlama，Policy Retriever 用 RAG+Embedding 检索，Margin Analyst 用 Python 数值计算库——它们在 Root Agent 看来，只是三个遵守同一份claude.md契约的服务端点。

提示：不要试图在 Root Agent 层实现 React 式的“思考-行动-观察”循环。那会让 Root Agent 变得臃肿且难以调试。真正的 React 应该下沉到每个子 Agent 内部——比如 SQL Generator 自己用 ReAct 框架生成 SQL，Policy Retriever 自己用 ReAct 框架做多跳检索。Root Agent 只需相信它们能按契约交付结果，就像你不会关心快递员是骑电动车还是开货车，你只关心他能不能在承诺时间内把包裹送到。

3. 核心配置与实操步骤：从零搭建可运行的 Root Agent 系统

3.1 环境准备与依赖安装：轻量起步，拒绝过度工程

Root Agent 的核心逻辑其实非常轻量，不需要复杂的分布式框架。我们推荐用 Python + FastAPI + Pydantic 实现，原因很实在：

• FastAPI 自动生成 OpenAPI 文档，方便调试和监控；
• Pydantic 提供强大的 Schema 验证能力，天然适配claude.md的输入输出结构；
• 整个服务打包后镜像体积 < 150MB，启动时间 < 2 秒，运维友好。

以下是精简版依赖清单（requirements.txt），剔除了所有非必要组件：

fastapi==0.115.0 uvicorn==0.32.0 pydantic==2.9.2 httpx==0.27.2 jinja2==3.1.4 python-dotenv==1.0.1

注意：不安装任何 LLM 相关包（如 transformers、llama-cpp-python）。Root Agent 本身不执行推理，只做调度，因此不需要 GPU 或大内存。所有模型推理都由下游子 Agent 承担。这大幅降低了 Root Agent 的资源消耗和故障面——我们线上集群的 Root Agent 实例，全部运行在 2C4G 的通用云主机上，CPU 平均占用率 < 15%。

安装命令（建议在干净虚拟环境中执行）：

python -m venv root_agent_env source root_agent_env/bin/activate # Linux/Mac # root_agent_env\Scripts\activate # Windows pip install --upgrade pip pip install -r requirements.txt

注意：不要用pip install fastapi[all]。[all]会安装大量调试和文档依赖（如 uvicorn[standard]、jinja2），在生产环境纯属冗余。我们线上只保留uvicorn作为 ASGI 服务器，其他全删。

3.2config.yaml详解：定义系统拓扑与基础策略

config.yaml是 Root Agent 的“地图”，它告诉系统：有哪些 Agent、它们在哪、基本行为规则是什么。以下是我们生产环境使用的精简模板，已去除所有注释和空行，确保可被程序直接加载：

version: "1.0" agents: query_decomposer: endpoint: "http://decomposer:8000/invoke" health_check: "/health" timeout: 15 retries: 1 circuit_breaker: failure_threshold: 3 reset_timeout: 60 sql_generator: endpoint: "http://sql-gen:8000/invoke" health_check: "/health" timeout: 30 retries: 2 circuit_breaker: failure_threshold: 5 reset_timeout: 120 policy_retriever: endpoint: "http://policy-retriever:8000/invoke" health_check: "/health" timeout: 25 retries: 1 circuit_breaker: failure_threshold: 2 reset_timeout: 300 routing: default: "query_decomposer" patterns: - regex: ".*毛利率.*影响.*|.*利润率.*波动.*" agent: "margin_analyst" - regex: ".*销售.*Q[1-4].*|.*季度.*营收.*" agent: "sql_generator" - regex: ".*政策.*更新.*|.*法规.*修订.*" agent: "policy_retriever" - regex: ".*总结.*|.*概览.*|.*汇总.*" agent: "query_decomposer"

关键配置项说明：

• timeout：单次调用的最大等待时间（秒）。设置原则是：必须小于下游 Agent 自身的超时设置。例如 SQL Generator 的claude.md声明“Max execution time: 8 seconds”，那么 Root Agent 的timeout必须设为 ≤ 8（我们设 30 是为了留出网络传输和序列化开销，但实际会通过httpx.AsyncClient的timeout参数精确控制）。
• retries：重试次数。永远不要设为 0。网络抖动是常态，1~2 次重试能解决 80% 的瞬时失败。但也不要设太高（如 5），否则会放大故障影响面。
• circuit_breaker：熔断器配置。当某 Agent 连续失败failure_threshold次，Root Agent 会将其标记为“熔断”，在reset_timeout秒内直接返回错误，不再尝试调用。这是防止雪崩的关键。我们的阈值设定基于 SLO：SQL Generator 要求 99.9% 的请求在 30 秒内成功，因此熔断阈值设为 5（对应约 0.1% 的失败率）。
• routing.patterns：正则路由表。所有正则必须用.*包裹，避免匹配失败。例如写r"毛利率"会漏掉“毛利率影响”，而r".*毛利率.*"则能覆盖所有变体。我们用regex字段而非keyword，是因为正则能表达更复杂的语义关系（如Q[1-4]匹配任意季度）。

3.3claude.md文件编写规范：让 Agent 说“普通话”

claude.md不是随意写的文档，它是一份可被程序解析的契约文件。我们制定了严格的编写规范，确保 Root Agent 能自动提取关键信息。以下是sql_generator的完整claude.md示例（已脱敏），并附上 Root Agent 解析它的 Python 代码逻辑：

# SQL Generator Agent ## Capabilities - Generate valid SQL queries for PostgreSQL databases - Support JOIN, WHERE, GROUP BY, ORDER BY clauses - **Does NOT support** DDL (CREATE/DROP), subqueries in SELECT clause, or non-deterministic functions (e.g., NOW()) ## Input Schema ```json { "user_query": "string, required, e.g. 'Show top 5 customers by revenue in Q3 2023'", "schema_context": "string, optional, description of relevant DB tables and columns", "allowed_tables": ["customers", "orders", "products"] }

Output Schema

{ "sql": "string, the generated SQL query", "explanation": "string, plain-language explanation of what the query does", "confidence": "number, 0.0-1.0, confidence in correctness" }

Error Handling

• Ifuser_queryis ambiguous: return error codeERR_AMBIGUOUS_QUERY, withsuggestionfield proposing clarifying questions
• Ifallowed_tablesrestriction violated: return error codeERR_TABLE_NOT_ALLOWED
• Max execution time: 8 seconds (hard limit)

Root Agent 的解析逻辑（`contract_parser.py`）： ```python import re import json from typing import Dict, List, Optional def parse_claude_md(content: str) -> Dict: """Parse claude.md content into structured contract""" contract = { "name": "", "capabilities": [], "input_schema": {}, "output_schema": {}, "error_codes": [] } # Extract name from first heading name_match = re.search(r"^#\s+(.+)$", content, re.MULTILINE) if name_match: contract["name"] = name_match.group(1).strip() # Extract capabilities (lines under "## Capabilities") cap_start = content.find("## Capabilities") if cap_start != -1: cap_end = content.find("\n## ", cap_start + 1) cap_section = content[cap_start:cap_end] if cap_end != -1 else content[cap_start:] # Get lines starting with "- " or "**Does NOT support**" cap_lines = [line.strip()[2:] for line in cap_section.split('\n') if line.strip().startswith("- ") or line.strip().startswith("**Does NOT support**")] contract["capabilities"] = cap_lines # Extract input schema (JSON block after "## Input Schema") input_match = re.search(r"## Input Schema\s+```json\s+([\s\S]+?)\s+```", content) if input_match: try: contract["input_schema"] = json.loads(input_match.group(1)) except json.JSONDecodeError: pass # Skip invalid JSON, keep empty dict # Extract output schema (JSON block after "## Output Schema") output_match = re.search(r"## Output Schema\s+```json\s+([\s\S]+?)\s+```", content) if output_match: try: contract["output_schema"] = json.loads(output_match.group(1)) except json.JSONDecodeError: pass # Extract error codes (lines containing "return error code") error_lines = re.findall(r"return error code `(.*?)`", content) contract["error_codes"] = list(set(error_lines)) # deduplicate return contract # Usage with open("agents/sql_generator/claude.md", "r") as f: contract = parse_claude_md(f.read()) print(json.dumps(contract, indent=2, ensure_ascii=False))

这个解析器的关键设计点：

• 不依赖外部 Markdown 解析库：用正则直接提取关键区块，轻量、快速、无依赖；
• 容忍格式瑕疵：如果Input Schema的 JSON 有语法错误，解析器会跳过并继续，不会崩溃；
• 错误码自动提取：return error codexxx`` 这种写法，让 Root Agent 能自动构建错误码映射表，用于后续统一错误处理。

实操心得：claude.md文件必须放在每个 Agent 服务的根目录下，并通过 HTTP 接口暴露（如GET /contract）。Root Agent 启动时会主动抓取所有已注册 Agent 的claude.md，缓存到内存。这样即使 Agent 服务重启，Root Agent 也能用缓存的契约继续调度，直到下次健康检查刷新。我们线上将claude.md的缓存 TTL 设为 5 分钟，平衡了新鲜度和稳定性。

3.4 Root Agent 核心调度逻辑：三层决策流与上下文注入

Root Agent 的调度逻辑不是简单的“if-else”，而是三层递进式决策流。以下是我们生产代码的核心骨架（main.py），已简化为可读性优先的伪代码逻辑：

from fastapi import FastAPI, HTTPException from pydantic import BaseModel import httpx import asyncio import re app = FastAPI() class RootInput(BaseModel): user_query: str session_id: str user_id: str class RootOutput(BaseModel): final_result: str intermediate_steps: list status: str # "success", "error", "partial" @app.post("/invoke", response_model=RootOutput) async def invoke_root(input_data: RootInput): # Step 1: Route Selection (Based on config.yaml routing.patterns) selected_agent = select_agent_by_regex(input_data.user_query) # Step 2: Context Enrichment (Inject global context) enriched_input = { "input": { "user_query": input_data.user_query, # Add fields required by selected_agent's claude.md input_schema # e.g., if sql_generator needs "schema_context", fetch from DB }, "context": { "session_id": input_data.session_id, "user_id": input_data.user_id, "trace_id": generate_trace_id(), "timestamp": int(time.time()) } } # Step 3: Agent Invocation & Error Handling try: async with httpx.AsyncClient() as client: response = await client.post( url=f"http://{selected_agent}/invoke", json=enriched_input, timeout=selected_agent_config["timeout"] ) if response.status_code == 200: result = response.json() # Validate against claude.md output_schema if validate_output_schema(result, selected_agent_contract["output_schema"]): return RootOutput( final_result=result.get("final_answer", str(result)), intermediate_steps=[{"agent": selected_agent, "output": result}], status="success" ) else: raise ValueError("Output schema validation failed") elif response.status_code == 400: # Handle standard error codes from claude.md error_data = response.json() error_code = error_data.get("error_code") if error_code in selected_agent_contract["error_codes"]: # Map to unified error handling if error_code == "ERR_AMBIGUOUS_QUERY": return handle_ambiguous_query(error_data) elif error_code == "ERR_TABLE_NOT_ALLOWED": return handle_table_restriction(error_data) raise HTTPException(status_code=400, detail=f"Agent error: {error_data}") else: raise HTTPException(status_code=response.status_code, detail="Agent internal error") except httpx.TimeoutException: # Circuit breaker logic here trigger_circuit_breaker(selected_agent) raise HTTPException(status_code=503, detail=f"Agent {selected_agent} timeout") except Exception as e: raise HTTPException(status_code=500, detail=str(e)) def select_agent_by_regex(query: str) -> str: """Select agent based on regex patterns in config.yaml""" for pattern in config["routing"]["patterns"]: if re.search(pattern["regex"], query, re.IGNORECASE): return pattern["agent"] return config["routing"]["default"] def handle_ambiguous_query(error_data: dict) -> RootOutput: """Unified handling for ambiguous query errors""" suggestion = error_data.get("suggestion", "Could you please clarify your request?") return RootOutput( final_result=suggestion, intermediate_steps=[{"agent": "root", "action": "ask_for_clarification"}], status="error" )

这个调度逻辑的精妙之处在于三层解耦：

1. 路由层（Route Selection）：纯正则匹配，毫秒级响应，不涉及任何模型推理；
2. 上下文层（Context Enrichment）：根据claude.md的Input Schema，动态注入所需字段。例如，若sql_generator的 Schema 要求schema_context，Root Agent 会从本地缓存或配置中心获取对应数据库的表结构描述，自动填充进去；
3. 执行层（Invocation & Error Handling）：用httpx.AsyncClient异步调用，配合超时、重试、熔断。最关键的是，它会用claude.md的Output Schema自动校验返回结果——如果子 Agent 返回了{"sql": "...", "confidence": 0.95}，但 Schema 要求confidence是0.0-1.0的 number，校验通过；如果返回了{"confidence": "high"}（字符串），校验失败，Root Agent 立即报错，而不是把错误数据传给用户。

这种设计让 Root Agent 成为系统的“质量守门员”，确保所有进入和流出的数据都符合契约，极大提升了系统鲁棒性。

4. 实操过程与核心环节实现：从本地调试到生产部署

4.1 本地开发调试：用 Mock Agent 快速验证调度逻辑

在真实子 Agent 都未开发完成前，如何验证 Root Agent 的调度逻辑？我们用Mock Agent解决。这不是简单的 HTTP mock，而是模拟claude.md契约的完整行为。以下是一个mock_sql_generator的 FastAPI 实现（mock_sql_gen.py）：

from fastapi import FastAPI from pydantic import BaseModel import json app = FastAPI() class MockInput(BaseModel): input: dict context: dict @app.post("/invoke") def mock_invoke(input_data: MockInput): # Simulate real SQL Generator behavior user_query = input_data.input.get("user_query", "") # Hardcoded responses based on query keywords if "top 5 customers" in user_query.lower(): return { "sql": "SELECT c.name, SUM(o.amount) as revenue FROM customers c JOIN orders o ON c.id = o.customer_id GROUP BY c.name ORDER BY revenue DESC LIMIT 5;", "explanation": "This query joins customers and orders tables, sums up order amounts per customer, and returns the top 5 by revenue.", "confidence": 0.98 } elif "average order value" in user_query.lower(): return { "sql": "SELECT AVG(amount) as avg_order_value FROM orders;", "explanation": "This query calculates the average amount across all orders.", "confidence": 0.95 } else: # Return standardized error per claude.md return { "error_code": "ERR_AMBIGUOUS_QUERY", "error_message": "The query is ambiguous. Please specify which metrics and dimensions you want.", "suggestion": "Could you clarify: 'What is the average order value for customers in Shanghai?'" } @app.get("/health") def health_check(): return {"status": "ok"}

启动命令：

uvicorn mock_sql_gen:app --host 0.0.0.0 --port 8000 --reload

然后在config.yaml中将sql_generator的endpoint指向http://localhost:8000/invoke，即可用真实 Root Agent 调度这个 Mock 服务。好处是：

• 开发 Root Agent 时，不依赖下游进度；
• 可以精准控制 Mock 的返回（成功/失败/慢响应），测试各种边界场景；
• 所有claude.md契约都能被真实验证——因为 Mock 服务也实现了相同的输入输出结构。

实操心得：我们为每个子 Agent 都准备了对应的 Mock 服务，并写成 Docker Compose 文件。开发时只需docker-compose up -d，就能一键启动 Root Agent + 全套 Mock Agent，5 分钟内完成端到端调试。这比写单元测试快得多，且更贴近真实调用链。

4.2 生产部署：容器化与健康检查闭环

生产环境部署 Root Agent，我们坚持“最小可行镜像”原则。Dockerfile 如下：

FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . # Remove dev dependencies and docs RUN rm -rf tests/ docs/ README.md # Create non-root user RUN addgroup -g 1001 -f appgroup && adduser -S appuser -u 1001 USER appuser EXPOSE 8000 CMD ["uvicorn", "main:app", "--host", "0.0.0.0:8000", "--port", "8000", "--workers", "4"]

关键点：

• 基础镜像用python:3.11-slim，而非python:3.11，体积减少 60%；
• --workers 4：根据 CPU 核数设置，我们 2C 实例设为 4，充分利用 I/O 并发；
• 绝不以 root 用户运行：创建appuser，降低安全风险。

健康检查是生产稳定的生命线。我们在config.yaml中为每个 Agent 配置了health_check，Root Agent 启动后会定期（默认 30 秒）调用所有 Agent 的健康端点。健康检查逻辑（health_checker.py）：

import asyncio import httpx from typing import Dict, List async def check_agent_health(agent_config: Dict) -> bool: """Check if agent is healthy""" try: async with httpx.AsyncClient(timeout=5.0) as client: response = await client.get( f"{agent_config['endpoint'].rsplit('/', 1)[0]}{agent_config['health_check']}" ) return response.status_code == 200 and response.json().get("status") == "ok" except Exception: return False async def run_health_checks(agent_configs: List[Dict]): """Run concurrent health checks""" tasks = [check_agent_health(config) for config in agent_configs] results = await asyncio.gather(*tasks) # Update internal agent status map for i, config in enumerate(agent_configs): agent_name = list(config.keys())[0] agent_status_map[agent_name] = results[i] # Run every 30 seconds async def health_loop(): while True: await run_health_checks(config["agents"].values()) await asyncio.sleep(30)

这个健康检查闭环的意义在于：

• 当某个子 Agent 宕机，Root Agent 会在 30 秒内感知，并将其状态标记为unhealthy；
• 下次路由时，如果正则匹配到该 Agent，Root Agent 会自动 fallback 到defaultAgent 或返回友好的降级提示；
• 运维人员可通过/health端点（Root Agent 自身的健康检查）看到所有下游 Agent 的实时状态，无需登录每台机器。

我们线上将健康检查失败的 Agent 自动加入“熔断池”，并在 Grafana 看板中高亮显示，平均故障发现时间（MTTD）从小时级缩短到秒级。

4.3 上下文管理与 trace_id 注入：让每一次调用可追溯

多 Agent 协作最大的调试噩梦是：用户反馈“结果错了”，你却不知道是哪个 Agent、哪一步出了问题。解决方案是全链路 trace_id 注入。Root Agent 在收到用户请求时，生成唯一trace_id，并将其注入到所有下游调用的context字段中。每个子 Agent 在处理时，必须将此trace_id记录到自己的日志中。

Root Agent 日志格式（logging_config.py）：

import logging import json class TraceIdFilter(logging.Filter): def filter(self, record): # Inject trace_id from current request context record.trace_id = getattr(record, 'trace_id', 'N/A') return True LOGGING_CONFIG = { "version": 1, "filters": { "trace_id": { "()": TraceIdFilter } }, "handlers": { "console": { "class": "logging.StreamHandler", "filters": ["trace_id"], "formatter": "detailed" } }, "formatters": { "