1. 多智能体系统架构选型的真实困境
1.1 从一个踩坑项目说起
去年下半年,我接手了一个内部工具链的智能化改造项目。需求说起来不复杂:让几个不同职责的智能体协作完成一份技术调研报告的生成——一个负责检索资料,一个负责归纳整理,一个负责撰写初稿,还有一个负责审校和格式化输出。四个角色,串行加部分并行,听起来用任何一个多智能体框架都能快速搭起来。
一开始我们选了一条看起来最省事的路:直接上通用编排框架。社区活跃、文档齐全、示例丰富,两天就跑通了Demo。但真正往生产环境推的时候,问题一个接一个冒出来。最典型的是状态管理——框架自带的状态机模型和我们业务里的“报告版本迭代”语义对不上,每次审校环节回退都要手动打补丁。再就是可观测性,框架封装的调用链在排查“为什么第三个智能体拿到了错误的上下文”时几乎帮不上忙,日志打出来是一团黑盒。
后来团队里有人提议:干脆全部自研,反正逻辑也不复杂。我们花了大概一周时间用最朴素的方式重写了一遍——自己定义消息协议、自己写调度循环、自己管上下文窗口。结果确实可控了,但新的问题又来了:并发控制、失败重试、超时熔断这些基础设施全部要自己造,而且造得远不如成熟框架稳。
这个来回折腾的过程,最终把我们逼到了一个中间地带:混合架构。核心编排逻辑自研,保证业务语义的精确表达;底层的基础设施能力复用通用框架,不重复造轮子。这篇文章就把这次架构抉择的完整思考过程、技术细节和踩过的坑摊开来讲,适合正在做多智能体系统选型、或者已经在自研和框架之间纠结的同行参考。
1.2 为什么这个问题值得单独拿出来讲
多智能体系统的架构选型,表面上看是一个“用不用框架”的技术决策,实际上它牵扯到三个层面的权衡:业务语义的表达能力、工程基础设施的成熟度、以及团队长期维护的成本。这三个维度在不同项目阶段的权重是完全不同的。
我见过不少团队在项目初期为了快速验证,直接套用通用框架,结果业务逻辑稍微复杂一点就被框架的抽象层卡住;也见过一些团队一上来就追求“完全自主可控”,把大量时间花在重造调度器、重试机制这些通用能力上,真正核心的业务逻辑反而没时间打磨。这两种极端都有各自的合理性,但也都埋着隐患。
混合架构的价值就在于,它承认了一个现实:没有任何一种方案能同时在这三个维度上做到最优。与其在两端之间反复横跳,不如划一条清晰的分界线,让自研的部分专注解决业务特有的问题,让框架的部分承担那些已经被验证过的通用能力。
2. 自研编排与通用框架的核心差异拆解
2.1 通用框架到底帮我们解决了什么
先把话说清楚,通用多智能体框架不是没有价值。恰恰相反,它在几个方面的积累是自研很难在短期内追上的。
第一是消息传递的可靠性。成熟的框架通常内置了消息队列、确认机制、死信处理等一整套基础设施。智能体之间的通信不是简单的函数调用,而是异步的、可能失败的消息交换。自己实现这套东西,光是处理“消息发出去了但对方没收到”这种边界情况就够喝一壶的。
第二是上下文管理的标准化。多智能体协作最头疼的问题之一就是上下文窗口的分配和裁剪。框架一般会提供统一的上下文对象模型,支持按角色、按轮次、按优先级来组织信息。自研的话,每个智能体都要自己处理“我该看到哪些历史消息”这个问题,很容易出现信息不一致。
第三是可观测性的开箱即用。调用链追踪、耗时统计、Token消耗监控,这些在框架里往往是标配。自研系统要做到同等水平,需要额外接入日志、指标、追踪三件套,工作量不小。
但框架的代价也很明显:抽象层带来的语义损耗。框架设计者预设的协作模式(比如常见的“管理者-工作者”模式或“流水线”模式)未必贴合你的业务。当业务逻辑需要更灵活的控制流时,你只能在框架的约束下打补丁,补丁越打越多,代码就越难维护。
2.2 自研编排的诱惑与陷阱
自研最大的吸引力在于完全的控制力。你可以精确地定义每个智能体在什么条件下被激活、以什么顺序执行、失败后如何回退、上下文如何传递。这种控制力在处理复杂业务语义时是决定性的。
举个例子,在我们的报告生成场景里,审校智能体发现事实性错误时,需要把问题定位到具体的段落,然后只让撰写智能体重写那一段,而不是整篇重来。这种“局部回退”的语义,在通用框架的状态机里表达起来非常别扭,但自研的调度循环里就是几行判断逻辑的事。
但自研的陷阱在于,你很容易低估基础设施的复杂度。我最初以为调度循环就是“按顺序调用几个函数”,实际写起来才发现要考虑:并发调用时的资源竞争、单个智能体超时后的降级策略、整个流程的幂等性保证、以及最要命的——当智能体数量从4个扩展到10个时,调度逻辑的复杂度是指数级上升的。
还有一个容易被忽视的点是测试成本。自研系统意味着所有的边界情况都要自己写测试覆盖。框架至少帮你把通用路径测过了,自研的话,光是模拟“消息丢失”“智能体返回格式错误”“上下文超长”这些异常场景,就要花掉大量时间。
2.3 一张表看清两种路线的取舍
| 维度 | 通用框架 | 纯自研 | 混合架构 |
|---|---|---|---|
| 业务语义表达 | 受框架抽象限制 | 完全自由 | 核心逻辑自由,基础设施标准化 |
| 开发速度 | 初期快 | 初期慢 | 中等 |
| 基础设施成熟度 | 高 | 依赖自建 | 复用框架能力 |
| 可观测性 | 开箱即用 | 需自行搭建 | 框架提供基础,自研补充业务指标 |
| 长期维护成本 | 受框架升级影响 | 完全自主但负担重 | 边界清晰,维护可控 |
| 团队学习曲线 | 需学习框架概念 | 需理解全部实现 | 只需理解分界线两侧的约定 |
这张表不是要得出“混合架构最好”的结论,而是想说:选型的本质是匹配项目阶段和团队能力。如果项目还在验证期,通用框架的快速启动优势是压倒性的;如果业务逻辑极其特殊且团队基础设施能力强,纯自研也未尝不可。混合架构适合的是那种“业务逻辑已经相对清晰、但基础设施不想重复造”的中间状态。
3. 混合架构的落地设计与关键决策
3.1 分界线的划定原则
混合架构最关键的一步,是决定“什么自研、什么复用”。这条线划在哪里,直接决定了后续的维护成本和扩展性。
我们最终确定的原则是:与业务语义强相关的编排逻辑自研,与业务无关的通用能力复用框架。具体来说:
自研的部分包括:智能体的角色定义与激活条件、上下文在角色间的传递规则、局部回退与重试的业务判断、以及整个流程的状态快照与恢复。这些逻辑和我们的报告生成业务紧密绑定,换一个业务场景就要重写,放在框架里反而会被框架的抽象束缚。
复用框架的部分包括:消息的底层传输与确认、智能体调用的超时与熔断、Token消耗的统计与上报、以及调用链的追踪与可视化。这些能力和业务无关,任何多智能体系统都需要,而且成熟框架的实现比我们自研的更稳定。
这条分界线的判断标准其实很简单:问自己“这个逻辑换一个业务场景还能用吗”。如果能用,就属于通用能力,优先考虑复用;如果不能用,就属于业务逻辑,应该自研。
3.2 自研编排层的核心数据结构
自研编排层最核心的设计是一个任务图结构。每个节点代表一个智能体的一次执行,边代表数据依赖和触发条件。这个图不是静态的,而是根据运行时的反馈动态调整的。
class TaskNode: def __init__(self, agent_role, input_context, retry_policy): self.agent_role = agent_role # 智能体角色标识 self.input_context = input_context # 输入上下文快照 self.retry_policy = retry_policy # 重试策略 self.status = "pending" # pending/running/done/failed self.output = None # 执行结果 self.dependencies = [] # 依赖的上游节点ID class TaskGraph: def __init__(self): self.nodes = {} # node_id -> TaskNode self.edges = {} # node_id -> [下游节点ID] def add_node(self, node_id, node): self.nodes[node_id] = node def add_edge(self, from_id, to_id): self.edges.setdefault(from_id, []).append(to_id) self.nodes[to_id].dependencies.append(from_id) def get_ready_nodes(self): """返回所有依赖已满足且状态为pending的节点""" ready = [] for nid, node in self.nodes.items(): if node.status != "pending": continue deps_done = all( self.nodes[d].status == "done" for d in node.dependencies ) if deps_done: ready.append(nid) return ready这个结构的好处是,局部回退变得非常自然。当审校节点发现某段内容有问题时,只需要把对应的撰写节点重置为pending,并清除其下游节点的输出,调度循环下一轮就会自动重新执行这些节点。不需要在框架的状态机里绕来绕去。
3.3 与通用框架的对接方式
自研编排层和通用框架的对接,我们选择了一个相对松耦合的方式:把框架当作一个“智能体执行器”来用。
具体来说,框架负责的是单个智能体的调用——包括向模型发起请求、处理流式响应、管理Token预算、以及失败重试。自研的编排层则负责决定“什么时候调用哪个智能体”以及“调用结果如何影响后续流程”。
对接的接口设计得很薄:
class AgentExecutor: """封装通用框架的智能体调用能力""" def __init__(self, framework_client): self.client = framework_client def execute(self, agent_role, context, timeout=30): """ 调用指定角色的智能体,返回结构化结果。 框架负责底层的重试、熔断、Token统计。 """ try: response = self.client.invoke( agent=agent_role, messages=context.to_messages(), timeout=timeout ) return AgentResult( success=True, content=response.content, token_usage=response.usage, trace_id=response.trace_id ) except FrameworkTimeoutError: return AgentResult(success=False, error="timeout") except FrameworkRateLimitError: return AgentResult(success=False, error="rate_limit")这样设计的好处是,框架的升级或替换不会影响编排层的业务逻辑。如果将来换一个框架,只需要重新实现AgentExecutor这一层,上层的任务图和调度循环完全不用动。这也是混合架构在长期维护上的一个隐性优势。
注意:对接层的异常处理要区分“框架层异常”和“业务层异常”。框架层异常(超时、限流)应该触发重试或降级;业务层异常(智能体返回了不符合预期的内容)应该触发局部回退。两者混在一起处理,会导致重试策略失效。
4. 实操过程中的关键环节与避坑记录
4.1 上下文传递的裁剪策略
多智能体系统里,上下文窗口是最稀缺的资源。四个智能体如果每个都把完整历史带上,Token消耗会迅速失控。我们的做法是按角色定制上下文视图。
检索智能体只需要看到原始需求;归纳智能体需要看到检索结果;撰写智能体需要看到归纳后的要点和原始需求;审校智能体需要看到撰写初稿和原始需求。每个角色的上下文都是独立裁剪的,而不是共享一个全局历史。
具体实现上,我们给每个角色定义了一个context_builder函数:
def build_context_for(role, global_state): if role == "retriever": return Context(messages=[global_state.original_request]) elif role == "summarizer": return Context(messages=[ global_state.original_request, global_state.retrieval_results ]) elif role == "writer": return Context(messages=[ global_state.original_request, global_state.summary, global_state.previous_draft # 局部回退时带上上一版 ]) elif role == "reviewer": return Context(messages=[ global_state.original_request, global_state.current_draft ])这个策略实测下来,Token消耗比“全量历史共享”降低了大约60%,而且因为每个角色只看到自己需要的信息,输出质量反而更稳定。
4.2 局部回退的触发条件设计
局部回退是混合架构里最能体现自研价值的地方,但触发条件的设计需要非常小心。触发太敏感会导致频繁回退、流程震荡;触发太迟钝又会让错误累积到无法修复。
我们的做法是分级触发:
- 一级回退:审校智能体明确标记某段落为“事实错误”或“逻辑矛盾”,只回退对应的撰写节点。
- 二级回退:审校智能体连续两次给出“需要大幅修改”的评价,回退到归纳节点,重新生成要点。
- 三级回退:整个流程超过预设的迭代次数上限,终止并输出当前最佳版本,同时记录失败原因供人工介入。
每一级回退都有明确的计数器和上限,防止无限循环。这个设计是在踩过一次“审校和撰写互相踢皮球导致流程跑了两个小时”的坑之后才定下来的。
4.3 状态快照与恢复的实现
多智能体流程可能跑很长时间,中间任何一步失败都从头再来是不可接受的。我们实现了一个轻量的状态快照机制:每次任务图发生变更(节点完成、回退触发)时,把整个TaskGraph序列化到持久化存储。
import json import hashlib def snapshot(graph, storage): """将任务图状态持久化""" state = { "nodes": { nid: { "role": node.agent_role, "status": node.status, "output_hash": hashlib.md5( str(node.output).encode() ).hexdigest() if node.output else None, "dependencies": node.dependencies } for nid, node in graph.nodes.items() }, "edges": graph.edges } storage.save(json.dumps(state)) def restore(storage, graph): """从快照恢复任务图""" state = json.loads(storage.load()) for nid, node_state in state["nodes"].items(): node = graph.nodes[nid] node.status = node_state["status"] # 输出内容需要从单独的存储按hash取回 if node_state["output_hash"]: node.output = storage.load_output(node_state["output_hash"])这里有个细节:输出内容单独存储,快照里只存hash。因为智能体的输出可能很大,全部塞进快照会让序列化变得很慢。按内容寻址的方式存储,还能顺便去重——如果两个节点的输出相同,只存一份。
4.4 可观测性的补充建设
框架提供的可观测性主要覆盖单次智能体调用,但多智能体流程的全局视图需要自研补充。我们在编排层埋了几个关键指标:
- 流程级指标:总耗时、总Token消耗、回退次数、最终状态。
- 节点级指标:每个节点的执行耗时、重试次数、输出长度。
- 回退级指标:回退触发原因分布、回退后修复成功率。
这些指标通过一个简单的埋点接口上报,和框架的调用链追踪通过trace_id关联。排查问题时,先看流程级指标定位到异常节点,再通过trace_id下钻到框架层的详细调用记录。
实操心得:埋点不要贪多,先上最核心的五个指标,跑一段时间后再根据实际排查需求补充。一开始就追求大而全的监控体系,往往导致埋点代码比业务代码还多,维护成本极高。
5. 常见问题与排查技巧实录
5.1 智能体输出格式不稳定怎么办
这是多智能体系统里最高频的问题。同一个智能体,同样的输入,有时候返回结构化JSON,有时候返回一段自然语言。我们的应对策略是三层防护:
第一层是提示词约束,在系统提示里明确要求输出格式,并给出示例。这一层能解决大部分问题,但不能指望100%可靠。
第二层是解析容错,写一个宽松的解析器,能处理常见的格式偏差(比如JSON被包在Markdown代码块里、字段名大小写不一致)。解析失败时不要直接抛异常,而是尝试提取关键信息。
第三层是格式修复智能体,当解析彻底失败时,把原始输出交给一个专门的修复智能体,让它转换成目标格式。这个智能体的提示词非常简单,就是“把下面的内容转换成JSON格式”。
三层防护下来,格式问题的自动修复率能到95%以上,剩下的5%才需要人工介入。
5.2 流程卡死或无限循环的排查
流程卡死通常有两个原因:依赖环或者回退震荡。
依赖环的排查相对简单,在构建任务图时做一次拓扑排序检测即可。如果存在环,说明节点依赖关系设计有问题,需要重新梳理。
回退震荡更隐蔽。表现是流程一直在跑,但状态在“撰写-审校-回退-撰写”之间反复横跳。排查方法是给每个节点加一个执行计数器,当某个节点的执行次数超过阈值时,强制终止并输出诊断信息。
我们后来还加了一个回退冷却机制:同一个节点在短时间内被连续回退时,回退间隔逐渐拉长,给智能体“换个思路”的机会。这个机制听起来有点玄学,但实测确实减少了一部分无效震荡。
5.3 常见问题速查表
| 问题现象 | 可能原因 | 排查方向 | 解决手段 |
|---|---|---|---|
| 流程卡死不动 | 依赖环或节点一直pending | 检查任务图拓扑结构 | 拓扑排序检测,修复依赖关系 |
| 反复回退不收敛 | 回退触发条件过于敏感 | 查看回退原因分布 | 调整触发阈值,加冷却机制 |
| Token消耗异常高 | 上下文未裁剪或重复携带 | 检查各角色context_builder | 按角色定制上下文视图 |
| 智能体输出格式错乱 | 提示词约束不足 | 查看原始输出 | 三层防护:约束+容错+修复 |
| 恢复后状态不一致 | 快照与输出存储不同步 | 对比快照hash与实际输出 | 快照和输出存储放在同一事务里 |
| 并发调用互相干扰 | 共享状态未隔离 | 检查全局变量使用 | 每个任务图实例独立状态 |
5.4 几个反直觉的实操经验
第一个经验是:不要追求智能体之间的“充分沟通”。一开始我们设计了一个“所有智能体共享一个消息板”的机制,结果信息过载导致每个智能体都被无关信息干扰。后来改成定向传递,每个智能体只接收自己需要的上游输出,效果反而更好。
第二个经验是:审校智能体的能力上限决定了整个系统的质量上限。如果审校智能体只能发现表面错误,那整个流程就是在低水平上循环。我们在审校智能体上投入的提示词优化时间,比其他三个加起来还多。
第三个经验是:混合架构的分界线不是一成不变的。项目初期我们把重试逻辑放在自研层,后来发现框架的重试机制更完善,就迁移了过去。分界线应该随着对框架和业务的理解加深而动态调整,不要一开始就定死。
6. 混合架构的扩展与长期维护
6.1 从四个智能体扩展到更多角色的路径
当业务需要增加新的智能体角色时,混合架构的优势就体现出来了。因为编排层是自研的,增加一个角色只需要:定义新的TaskNode类型、在context_builder里增加对应的上下文视图、在任务图构建逻辑里加入新的节点和边。框架层完全不用动。
我们后来增加了一个“数据可视化”智能体,负责把报告里的关键数据转成图表描述。整个接入过程花了不到半天,主要时间花在提示词调优上,架构层面几乎没有改动。
6.2 框架升级时的隔离策略
通用框架的版本升级是不可避免的。混合架构的隔离设计让升级风险可控:因为所有框架调用都收敛在AgentExecutor这一层,升级时只需要确保这一层的接口兼容,上层的编排逻辑不受影响。
我们实际经历过一次框架的大版本升级,消息协议有破坏性变更。因为隔离层做得好,只改了AgentExecutor里的适配代码,业务逻辑一行没动。这也是当初坚持划清分界线的回报。
6.3 团队协作上的隐性收益
混合架构还有一个容易被忽视的好处:它让团队分工更清晰。对框架熟悉的同学可以专注优化AgentExecutor和基础设施对接;对业务熟悉的同学可以专注打磨编排逻辑和提示词。两拨人的工作界面就是那个薄薄的接口层,沟通成本很低。
相比之下,纯自研需要每个人都理解全部实现,纯框架则需要每个人都理解框架的全部抽象。混合架构把学习曲线拆成了两段,每段都更平缓。
我个人在实际操作中的体会是,混合架构不是一种“妥协”,而是一种有意识的边界设计。它的核心不是“用一半自研一半框架”,而是“想清楚哪些能力是业务的核心竞争力、必须自己掌控,哪些能力是通用基础设施、可以站在别人肩膀上”。这条线划对了,后面的扩展和维护都会顺很多。如果非要给一个建议,我会说:先花时间把这条线想清楚,再动手写代码,比急着跑通Demo重要得多。