企业官网建设流程全解析

1. 多 Agent 协作的幻觉：单测全绿，联调崩溃

做过分布式系统的人都有一个共识：单元测试全绿，不代表系统能跑。这个道理放在多 Agent 协作系统里，同样成立，而且更加隐蔽。单个 Agent 的 Prompt 调好了，工具调用没问题，输出格式也稳定，你信心满满地把它们串起来，结果系统一跑就崩。更让人抓狂的是，崩的方式千奇百怪：有时候是信息在传递过程中丢失了关键字段，有时候是两个 Agent 互相等待对方先行动，有时候是第三个 Agent 拿到了前两个 Agent 的混合输出，直接开始胡言乱语。

这不是某一个框架的 Bug，也不是模型能力不够，而是多 Agent 协作本身引入的系统性复杂度。我前后搭过几套多 Agent 系统，从简单的双 Agent 对话到五六个 Agent 分工协作的流程，踩过的坑基本可以归成一句话：单 Agent 的可靠性，无法通过简单叠加传递到系统层面。每个 Agent 单独看都是 90 分，串起来可能只有 40 分。

这篇文章想聊的就是这件事。我会从协作架构的设计思路讲起，拆解信息传递、状态管理、冲突消解这几个核心环节里最容易出问题的地方，然后给出可落地的排查方法和实操方案。如果你正在搭多 Agent 系统，或者已经被“单测全绿、联调崩溃”折磨过，这篇内容应该能帮你省下不少调试时间。

2. 为什么单 Agent 测试通过，系统仍然崩溃

2.1 单 Agent 测试的盲区在哪里

先说说我们通常怎么测一个 Agent。拿一个负责信息提取的 Agent 举例，你会准备一批输入样本，检查它输出的 JSON 结构对不对、字段值准不准、有没有幻觉。跑个几十条，准确率 95% 以上，你就觉得这个 Agent 没问题了。

但这种测试有一个根本性的盲区：它假设输入是干净且确定的。真实的多 Agent 系统里，一个 Agent 的输入往往来自另一个 Agent 的输出，而这个输出可能包含格式偏差、语义模糊、甚至完全错误的内容。你的信息提取 Agent 在测试集上表现很好，是因为测试集的输入是你精心准备的；但在系统里，它拿到的可能是上一个 Agent 用自然语言描述的一段话，里面夹杂着不确定的措辞和冗余信息。

我踩过最典型的一个坑：一个负责分类的 Agent，单测时输入都是标准的用户问题，分类准确率很高。但接入系统后，它的输入变成了上游 Agent 的总结文本，里面包含了“根据分析，这个问题可能属于 A 类，但也存在 B 类的可能性”这样的表述。分类 Agent 直接懵了，因为它从来没在训练和测试中见过这种“带不确定性的输入”。

2.2 协作引入的三类系统性风险

多 Agent 协作带来的问题，大致可以分成三类。

第一类是信息衰减与失真。Agent 之间传递信息时，每一层都可能丢失细节或引入噪声。A Agent 输出一段分析，B Agent 基于这段分析做决策，C Agent 再基于 B 的决策执行操作。每经过一层，原始信息就衰减一次。如果中间某个 Agent 做了“总结”或“压缩”，关键细节可能直接消失。

第二类是状态不一致。多个 Agent 并行工作时，它们各自维护自己的上下文和状态。如果 Agent A 更新了某个数据，Agent B 还在用旧版本，就会产生冲突。更麻烦的是，这种不一致往往不会立刻暴露，而是在某个特定条件下才触发，排查起来非常困难。

第三类是交互死锁与循环。两个 Agent 互相等待对方先输出，或者三个 Agent 陷入“A 等 B、B 等 C、C 等 A”的循环，系统直接卡死。这类问题在单 Agent 测试中根本不可能发现，因为单 Agent 没有交互对象。

2.3 一个真实的崩溃案例拆解

我搭过一个内容审核流程，包含三个 Agent：一个负责提取关键信息，一个负责判断风险等级，一个负责生成审核意见。单测时每个 Agent 都表现良好，但联调时出现了这样的问题：

提取 Agent 输出的 JSON 里，有一个字段叫risk_indicators，是一个数组。判断 Agent 期望这个字段始终存在且为数组类型。但在某些输入下，提取 Agent 认为没有风险指标，直接省略了这个字段。判断 Agent 拿到没有该字段的输入后，没有做空值处理，直接抛异常。异常被上层捕获后，系统重试了整个流程，但重试时输入没变，于是再次失败，最终超时崩溃。

这个问题的根源不在于任何一个 Agent 的逻辑错误，而在于接口契约没有被严格定义和验证。提取 Agent 认为“没有风险指标就不输出该字段”是合理的，判断 Agent 认为“该字段必须存在”也是合理的，但两者之间的契约没有对齐。

3. 协作架构设计的核心考量

3.1 集中式调度 vs 去中心化协商

多 Agent 协作的架构，大体上分两种思路。

集中式调度是有一个 Orchestrator（编排器）角色，由它来决定哪个 Agent 在什么时候做什么。所有 Agent 之间的通信都经过 Orchestrator 中转。这种架构的好处是控制流清晰，容易调试，出问题时能快速定位是哪个环节卡住了。坏处是 Orchestrator 容易成为瓶颈，而且它的逻辑会越来越复杂，最终变成一个难以维护的“上帝对象”。

去中心化协商是 Agent 之间直接通信，没有统一的调度中心。每个 Agent 根据自己的状态和收到的消息决定下一步行动。这种架构更灵活，扩展性更好，但调试难度直线上升。你很难回答“为什么系统做了这个决策”这样的问题，因为决策是多个 Agent 局部交互涌现出来的。

我的建议是：除非你有非常明确的理由选择去中心化，否则优先用集中式调度。多 Agent 系统本身的复杂度已经够高了，再引入去中心化的不确定性，调试成本会高到难以承受。集中式调度虽然看起来“不够优雅”，但它的可观测性和可控性是去中心化架构很难比拟的。

3.2 通信协议：结构化消息是底线

Agent 之间怎么传递信息，这是协作架构里最关键的设计决策之一。我见过不少项目直接用自然语言做 Agent 间的通信，理由是“模型能理解自然语言，这样更灵活”。这种做法在 Demo 阶段看起来很酷，但到了生产环境基本都会出问题。

自然语言通信的问题在于歧义性和不可验证性。A Agent 说“这个内容有点风险”，B Agent 理解成“需要进一步审查”，C Agent 理解成“直接拒绝”。三个 Agent 对同一句话的理解不一致，系统行为就不可预测了。

我的做法是：Agent 间的通信必须用结构化格式，通常是 JSON。每个消息包含明确的字段：发送者、接收者、消息类型、负载数据、时间戳、以及可选的元信息。负载数据的 Schema 要严格定义，并且要有验证机制。如果某个 Agent 输出的消息不符合 Schema，系统应该立即报错，而不是让错误的消息继续往下传。

{ "from": "extractor_agent", "to": "risk_judge_agent", "type": "extraction_result", "payload": { "content_id": "abc123", "risk_indicators": [], "confidence": 0.87 }, "timestamp": "2025-01-15T10:30:00Z", "trace_id": "trace_xyz789" }

上面这个例子里，risk_indicators即使为空也必须存在，这就是契约的一部分。trace_id用于全链路追踪，后面排查问题时会非常有用。

3.3 状态管理：共享状态还是消息传递

多 Agent 系统里的状态管理，有两种主流模式。

一种是共享状态，所有 Agent 读写同一个状态存储（比如一个共享的字典或数据库）。这种模式的好处是状态一致性强，Agent 之间不需要显式传递状态。坏处是并发读写需要加锁，而且 Agent 之间的耦合度变高，一个 Agent 改了状态可能影响其他 Agent 的行为。

另一种是消息传递，每个 Agent 维护自己的状态，通过消息来同步信息。这种模式解耦更彻底，每个 Agent 可以独立测试和替换。坏处是状态同步有延迟，而且需要处理消息丢失和重复的问题。

我倾向于混合模式：核心的、需要强一致的状态放在共享存储里，比如任务 ID、当前阶段、全局配置；而 Agent 各自的中间结果和上下文用消息传递。这样既保证了关键状态的一致性，又保留了 Agent 的独立性。

注意：无论选哪种模式，都要给状态加上版本号或时间戳。当两个 Agent 同时更新同一个状态时，版本号能帮你检测冲突，而不是默默地覆盖。

4. 核心细节解析与实操要点

4.1 接口契约的严格定义与验证

前面反复提到接口契约，这里展开说说怎么落地。

每个 Agent 的输入和输出都要有明确的 Schema 定义。我通常用 JSON Schema 或者 Pydantic 模型来定义，然后在 Agent 的入口和出口做验证。验证不通过的消息，直接拒绝并记录错误，不要尝试“容错处理”。容错处理听起来很美好，但实际上会让问题隐藏得更深，等到爆发时更难排查。

from pydantic import BaseModel, Field from typing import List, Optional class ExtractionResult(BaseModel): content_id: str risk_indicators: List[str] = Field(default_factory=list) confidence: float = Field(ge=0.0, le=1.0) raw_text: Optional[str] = None class Config: extra = "forbid" # 禁止未定义的字段

上面这个 Pydantic 模型里，risk_indicators有默认值，所以即使上游没传也不会报错。extra = "forbid"确保不会有多余的字段混进来。confidence有范围约束，超出范围会直接报错。

验证的时机也很重要。我建议在消息发送前和接收后都做验证。发送前验证确保 Agent 不会发出不合规的消息，接收后验证确保 Agent 不会处理不合规的消息。双重验证看起来冗余，但能帮你快速定位是哪个环节出了问题。

4.2 超时、重试与熔断机制

多 Agent 系统里，一个 Agent 卡住可能导致整个系统挂起。所以超时机制是必须的。

每个 Agent 调用都要设置超时时间。超时时间设多长，取决于这个 Agent 的预期处理时间。我的经验是：取 P99 处理时间的 2 到 3 倍。比如一个 Agent 通常 2 秒内完成，P99 是 5 秒，那超时时间可以设 10 到 15 秒。设太短会导致正常请求被误杀，设太长会让系统在异常时卡太久。

重试策略要谨慎。不是所有失败都适合重试。如果是网络抖动导致的失败，重试是合理的；如果是输入数据格式错误导致的失败，重试多少次都一样。我通常只对超时和明确的临时性错误做重试，重试次数不超过 2 次，并且要加退避策略。

熔断机制用于防止故障扩散。如果某个 Agent 连续失败多次，暂时停止调用它，直接返回降级结果。这样可以避免整个系统被一个故障 Agent 拖垮。

import time from functools import wraps def with_retry(max_retries=2, backoff_base=1.0): def decorator(func): @wraps(func) def wrapper(*args, **kwargs): last_exception = None for attempt in range(max_retries + 1): try: return func(*args, **kwargs) except (TimeoutError, ConnectionError) as e: last_exception = e if attempt < max_retries: sleep_time = backoff_base * (2 ** attempt) time.sleep(sleep_time) raise last_exception return wrapper return decorator

4.3 日志与链路追踪的设计

多 Agent 系统的调试，没有链路追踪基本没法做。你需要能够回答：一个请求进来后，经过了哪些 Agent，每个 Agent 的输入输出是什么，在哪个环节出了问题。

我的做法是给每个请求分配一个唯一的trace_id，这个 ID 在所有 Agent 之间传递，每条日志都带上这个 ID。这样你就能通过一个 ID 把整个链路的日志串起来。

日志的内容也有讲究。不要只记录“Agent A 处理完成”这种模糊信息，要记录输入摘要、输出摘要、处理耗时、是否命中缓存、是否有重试。输入输出摘要不要记录全文，记录关键字段和哈希值即可，避免日志量过大。

import logging import hashlib import json logger = logging.getLogger(__name__) def log_agent_io(trace_id, agent_name, input_data, output_data, duration_ms): input_hash = hashlib.md5(json.dumps(input_data, sort_keys=True).encode()).hexdigest()[:8] output_hash = hashlib.md5(json.dumps(output_data, sort_keys=True).encode()).hexdigest()[:8] logger.info( f"trace_id={trace_id} agent={agent_name} " f"input_hash={input_hash} output_hash={output_hash} " f"duration_ms={duration_ms}" )

提示：trace_id的生成要保证全局唯一，通常用 UUID 或者“时间戳+随机数”的组合。不要用自增 ID，因为在分布式环境下容易冲突。

5. 实操过程与核心环节实现

5.1 从零搭建一个可调试的多 Agent 协作流程

假设我们要搭一个三 Agent 的内容处理流程：提取 Agent 负责从原始文本中提取结构化信息，判断 Agent 负责根据提取结果做分类判断，生成 Agent 负责根据判断结果生成最终输出。

第一步是定义消息格式和接口契约。三个 Agent 的输入输出都用 Pydantic 模型定义，确保类型和字段明确。

class RawContent(BaseModel): content_id: str text: str class ExtractionOutput(BaseModel): content_id: str entities: List[str] categories: List[str] confidence: float class JudgmentOutput(BaseModel): content_id: str decision: str # "approve", "reject", "review" reason: str confidence: float class GenerationOutput(BaseModel): content_id: str final_text: str metadata: dict

第二步是实现 Orchestrator。Orchestrator 负责接收原始请求，依次调用三个 Agent，处理超时和重试，记录链路日志。

class Orchestrator: def __init__(self, extractor, judge, generator): self.extractor = extractor self.judge = judge self.generator = generator def process(self, raw_content: RawContent) -> GenerationOutput: trace_id = generate_trace_id() extraction = self._call_with_timeout( self.extractor, raw_content, trace_id, timeout=10 ) judgment = self._call_with_timeout( self.judge, extraction, trace_id, timeout=10 ) result = self._call_with_timeout( self.generator, judgment, trace_id, timeout=15 ) return result def _call_with_timeout(self, agent, input_data, trace_id, timeout): start = time.time() try: output = agent.process(input_data) duration = (time.time() - start) * 1000 log_agent_io(trace_id, agent.name, input_data.dict(), output.dict(), duration) return output except Exception as e: logger.error(f"trace_id={trace_id} agent={agent.name} error={str(e)}") raise

第三步是给每个 Agent 加上输入输出验证。Agent 的process方法入口先验证输入，出口再验证输出。

class ExtractorAgent: name = "extractor" def process(self, input_data: RawContent) -> ExtractionOutput: # 输入验证 if not isinstance(input_data, RawContent): raise ValueError(f"Invalid input type: {type(input_data)}") # 实际处理逻辑 result = self._do_extraction(input_data) # 输出验证 output = ExtractionOutput(**result) return output

5.2 关键参数的计算与选择

超时时间的设定，我通常用历史数据的 P99 乘以 2.5。假设提取 Agent 的 P50 是 800ms，P95 是 2s，P99 是 4s，那超时时间设为 10s。这个值既能覆盖绝大多数正常请求，又不会让异常请求卡太久。

重试次数的选择，要考虑重试的成功率和重试的成本。如果第一次失败后重试的成功率只有 10%，那重试的价值就不大。我一般设 2 次重试，第一次重试间隔 1 秒，第二次间隔 2 秒。如果两次都失败，基本可以认为不是临时性问题。

并发数的控制也很重要。如果多个请求同时进来，每个请求都要调用多个 Agent，系统的并发压力会很大。我通常用信号量或者队列来控制并发，避免把下游 Agent 打挂。

import asyncio class ConcurrencyLimiter: def __init__(self, max_concurrent): self.semaphore = asyncio.Semaphore(max_concurrent) async def run(self, coro): async with self.semaphore: return await coro

5.3 联调阶段的测试策略

单 Agent 测试通过后，不要直接上生产环境。先做集成测试，用模拟的上游输出作为输入，验证 Agent 之间的接口是否对齐。

集成测试的用例要覆盖边界情况：空输入、超长输入、格式异常的输入、包含特殊字符的输入。这些情况在单 Agent 测试中往往被忽略，但在系统里很容易触发。

然后做端到端测试，用真实的请求走完整流程。端到端测试不要只测正常路径，要故意制造异常：让某个 Agent 超时、返回错误、返回格式不对的数据，观察系统是否能正确处理。

我通常会写一个“故障注入”的测试脚本，随机让某个 Agent 失败或延迟，然后跑一批请求，看系统的整体成功率和错误分布。这个测试能暴露很多正常测试发现不了的问题。

6. 常见问题与排查技巧实录

6.1 信息在传递过程中丢失或变形

这是最常见的问题。表现是：上游 Agent 明明输出了某个字段，下游 Agent 却说没收到，或者收到的值不对。

排查思路：先看日志，确认上游 Agent 的输出里有没有这个字段。如果有，再看下游 Agent 的输入日志，确认它收到的是什么。如果上游输出有、下游输入没有，那问题出在传输层；如果下游输入有但处理结果不对，那问题出在下游 Agent 的逻辑。

常见原因有几个：一是序列化和反序列化不一致，比如上游用json.dumps输出，下游用eval解析，遇到特殊字符就出问题；二是字段名大小写不一致，上游用contentId，下游期望content_id；三是嵌套结构被扁平化，上游输出的是嵌套 JSON，中间某个环节把它转成了字符串。

避坑技巧：在 Agent 之间的传输层加一个“消息校验”环节，对每条消息做 Schema 验证。验证不通过的消息直接拒绝并告警，不要让它继续往下传。

6.2 Agent 之间互相等待导致死锁

死锁的表现是系统卡住，没有任何 Agent 在输出，日志也停在某个点不动。

排查思路：看日志的最后几条，确认是哪个 Agent 在等待。然后看它等待的条件是什么，是等另一个 Agent 的消息，还是等某个资源释放。

常见原因：两个 Agent 互相等待对方先发消息；或者一个 Agent 在等另一个 Agent 的结果，但那个 Agent 已经崩溃了，消息永远不会来。

解决方法：给每个等待操作加超时。等待超时后，要么重试，要么走降级逻辑，不要无限等待。另外，尽量避免设计成“A 等 B、B 等 A”的交互模式，改成“A 发消息给 B，B 处理完后回调 A”或者“由 Orchestrator 统一调度”。

6.3 输出格式不一致导致解析失败

这个问题在接入新 Agent 时特别常见。新 Agent 的输出格式和现有系统不一致，导致解析失败。

排查思路：对比新 Agent 的输出和系统期望的格式，找出差异点。常见差异包括：日期格式不同（2025-01-15vs01/15/2025）、数字类型不同（字符串 vs 数字）、空值表示不同（nullvs""vs 字段缺失）。

解决方法：在 Agent 的输出层加一个“格式适配器”，把 Agent 的原始输出转换成系统统一格式。适配器要做严格的验证，确保转换后的格式符合预期。

6.4 常见问题速查表

6.5 独家避坑经验

第一个经验：不要相信任何 Agent 的输出。即使这个 Agent 在单测中表现完美，在系统里也要对它保持怀疑。每个 Agent 的输出都要验证，验证不通过就拒绝。

第二个经验：日志要记录足够多的上下文。我见过很多项目日志只记录“处理失败”，但不记录输入是什么、失败原因是什么。这种日志在排查时基本没用。日志要记录 trace_id、Agent 名称、输入摘要、输出摘要、耗时、错误信息。

第三个经验：先做同步版本，再做异步版本。同步版本虽然性能差，但调试简单，逻辑清晰。等同步版本跑通了，再考虑改成异步。很多项目一上来就搞异步，结果调试成本高到离谱。

第四个经验：给每个 Agent 加一个“健康检查”接口。这个接口不处理业务逻辑，只返回 Agent 的基本状态：是否就绪、当前负载、最近错误率。Orchestrator 在调用 Agent 前可以先检查健康状态，避免调用一个已经挂掉的 Agent。

第五个经验：版本管理要严格。Agent 的接口契约变更时，要同步更新所有依赖它的 Agent。我通常用版本号来管理契约，比如v1、v2，不同版本之间保持兼容。如果必须做不兼容变更，就新开一个接口，旧接口保留一段时间。

7. 协作模式的扩展与演进

7.1 从固定流程到动态编排

最开始搭多 Agent 系统时，我用的都是固定流程：A 完了调 B，B 完了调 C。这种模式简单直接，但灵活性差。如果某个请求不需要经过 B，或者需要多调一次 C，固定流程就处理不了。

后来我开始尝试动态编排：Orchestrator 根据当前状态和请求特征，动态决定下一步调用哪个 Agent。这种模式灵活很多，但复杂度也上去了。Orchestrator 的逻辑会变得越来越复杂，最终可能变成一个难以维护的决策树。

我的折中方案是：用配置来定义流程，而不是硬编码。把流程定义成一个有向图，每个节点是一个 Agent，边是转移条件。Orchestrator 根据配置和当前状态来决定下一步。这样既保留了灵活性，又避免了硬编码的混乱。

workflow = { "start": "extractor", "nodes": { "extractor": { "next": "judge", "condition": lambda state: state.get("has_content", True) }, "judge": { "next": "generator", "condition": lambda state: state["decision"] != "reject" }, "generator": { "next": None } } }

7.2 引入人工审核节点

有些场景下，Agent 的判断不够可靠，需要人工介入。比如内容审核里，Agent 判断为“需要复核”的内容，要转给人工审核。

人工审核节点的引入，会带来新的问题：人工审核的延迟不确定，可能几秒，也可能几小时。系统不能一直等着，需要有超时和降级机制。另外，人工审核的结果要能反馈回系统，用于后续的 Agent 优化。

我的做法是：人工审核节点作为一个特殊的 Agent，它的处理是异步的。Orchestrator 把请求发给人工审核节点后，不等待结果，而是把请求状态标记为“待审核”，然后继续处理其他请求。人工审核完成后，通过回调通知 Orchestrator，Orchestrator 再继续后续流程。

7.3 多 Agent 系统的监控与告警

多 Agent 系统上线后，监控是必不可少的。需要监控的指标包括：每个 Agent 的调用量、成功率、平均耗时、P99 耗时、错误分布；整个系统的端到端成功率、平均耗时；以及队列长度、并发数等资源指标。

告警的阈值要根据历史数据来定。比如某个 Agent 的错误率突然从 1% 涨到 10%，那肯定有问题，要告警。但如果某个 Agent 的错误率一直是 5%，那告警阈值就要设得更高，避免误报。

我通常用 Prometheus 加 Grafana 来做监控和可视化。每个 Agent 暴露一个 metrics 接口，Prometheus 定期抓取，Grafana 做展示。告警用 Alertmanager，配置好规则后自动发通知。

from prometheus_client import Counter, Histogram agent_calls = Counter('agent_calls_total', 'Total agent calls', ['agent_name', 'status']) agent_duration = Histogram('agent_duration_ms', 'Agent duration in ms', ['agent_name']) def track_agent_call(agent_name, duration_ms, success): status = 'success' if success else 'error' agent_calls.labels(agent_name=agent_name, status=status).inc() agent_duration.labels(agent_name=agent_name).observe(duration_ms)

这套监控搭起来后，系统出问题时能快速定位是哪个 Agent 的问题，而不是盲目地翻日志。我实际用下来，排查效率至少提升了一倍。

8. 一些个人体会

多 Agent 协作系统的复杂度，不是线性增长的，而是指数增长的。两个 Agent 协作，要考虑的交互路径有 2 条；三个 Agent 协作，交互路径可能变成 6 条甚至更多。每增加一个 Agent，系统的状态空间就扩大一圈，潜在的故障模式也成倍增加。

所以我的核心建议是：能少用一个 Agent 就少用一个。不要为了“架构优雅”而拆分 Agent。如果一个 Agent 能搞定的事，就不要拆成两个。Agent 拆分的唯一理由是：单个 Agent 的职责太重，Prompt 太长，或者需要不同的模型来处理不同类型的任务。如果只是因为“看起来更清晰”而拆分，那是在给自己找麻烦。

另外，测试策略要跟着架构走。单 Agent 测试只能验证 Agent 本身的逻辑，不能验证协作。协作的验证要靠集成测试和端到端测试。而且这些测试要覆盖异常路径，不能只测正常流程。我现在的习惯是：每加一个 Agent，就写一批故障注入的测试用例，确保系统在异常情况下也能优雅降级，而不是直接崩溃。

最后分享一个我最近在用的调试技巧：给每个 Agent 加一个“影子模式”。影子模式下，Agent 正常处理请求，但不把结果返回给下游，而是把结果记录到日志里。这样你可以在不影响系统行为的前提下，观察每个 Agent 的实际输出，对比它们和预期的差异。这个技巧在排查“信息在传递过程中变形”这类问题时特别有用。