告别脆弱的单体应用，用多智能体网络构建稳定的生产力工具

来源DeepHub IMBA 本文约9000字建议阅读10分钟多智能体系统的发展轨迹指向越来越自主、自组织的智能体生态。多智能体系统代表了 AI 应用设计上的一次根本性转向在过去几年主流的一直是单智能体模型一个 LLM、一条提示链prompt chain、一个系统包办所有事。这种范式在简单任务上能跑到了复杂任务上就有一些力不从心了。这里主要有三点第一值得解决的问题在本质上就是多智能体问题。供应链优化牵涉供应商、制造商、分销商、零售商每一方都有自己的目标与约束。金融交易牵涉跨多个时间维度运行的多种策略。城市交通管理牵涉成千上万个交叉路口需要在本地协调的同时服务于全局通行目标。这些问题硬塞进单智能体架构就会得到脆弱的系统。第二工具链已经成熟。两年前要构建一套多智能体系统得从零写自定义编排代码今天CrewAI、Microsoft AutoGen、CAMEL 这类框架直接提供生产级的智能体通信、任务委派、冲突解决抽象门槛下降了一个数量级。第三多智能体系统在失败时的退化方式比单体平缓得多。多智能体系统中某一个智能体挂掉其它智能体可以补位、改道或者让整体优雅降级。单体系统一旦失效就是整体失效。那家迪拜客户用三天的人工订单处理换来了这条教训。你可以靠读这篇文章来换。协作型与竞争型智能体协作型智能体Cooperative agents共享一个共同目标工作时自由交换信息为整体结果做优化。比如说物流系统中路由智能体和分配智能体之间没有任何理由互相隐瞒路由智能体发现某条高速拥堵的瞬间就把信息分享出来让分配智能体相应调整。当所有智能体都在你掌控之下、服务于同一个利益相关方时协作型系统是合适的选择。竞争型智能体Competitive agents目标相互冲突。每个智能体为自己的收益做优化一方所得可能就是另一方所失。金融交易是经典案例一套多智能体交易系统里可能有一个动量momentum智能体想在价格上行时买入有一个均值回归mean-reversion智能体想在价格高位卖出还有一个做市market-making智能体不管方向、只想从买卖价差里吃肉。它们之间不是协作而是为共享资本池中的资金分配相互竞争——系统的整体表现就源自这些策略之间的张力。现实世界里的多数系统介于两者之间混合动机mixed-motive系统智能体在某些维度上协作在另一些维度上竞争。比如说供应链里每个节点优化自身库存同时在需求信号共享上协作就属于这一类自动驾驶车队里每辆车优化自身行程时间同时在交通流上配合也属于这一类。所以先要确定选定多智能体的类型因为这会决定通信协议、协调机制和失败处理策略。一旦选错要么给一个竞争型问题构建出协作型系统让智能体天真地分享本该被门控的信息要么给一个协作型问题构建出竞争型系统让智能体把精力浪费在没有意义的战略性行为上。通信协议智能体之间如何对话无法通信的智能体也无法协调通信协议是多智能体系统的神经系统把它设计正确是整个工程里最难的设计决策之一。基本的通信模式有三种。直接消息Direct messaging是最简单的一种。智能体 A 给智能体 B 发一条消息。它适合智能体数量少、彼此关系明确的场景路由智能体把某条高速拥堵的消息告诉分配智能体需求预测器把更新过的预测发给库存监控器。这种方式高效但扩展性不行——在 N 个智能体的系统里潜在通信通道有 N 的平方那么多管理起来会变成组合爆炸式的噩梦。广播通信Broadcast communication是另一个极端一个智能体把消息同时丢给其它所有智能体。它对系统级状态变更有用比如编排器宣布系统进入降级模式。坏处是噪声很大所有人都收到所有事情会把周期浪费在跟自己无关的消息上。发布-订阅Publish-subscribepub/sub是大多数生产系统的实用折中。智能体把消息发布到命名主题topic其它智能体订阅自己关心的主题。路由智能体发布到route-updates主题分配智能体和配送排程智能体订阅需求预测器不订阅因为它不需要实时路由信息。pub/sub 把生产者和消费者解耦扩展性好新接入一个智能体不需要重连既有通道。在传输模式之上还得有消息 schema。智能体之间究竟说什么在多智能体系统的学术研究里标准参考是 FIPA-ACLFoundation for Intelligent Physical Agents — Agent Communication Language它定义了一组施为语performative包括 inform、request、propose、accept、reject。在生产中大多数系统用更简单的 schema——带 type、sender、timestamp 与 payload 字段的 JSON。要点是每条消息必须是自描述的接收方在没有任何带外out-of-band上下文的情况下能判断它是哪种消息、谁发的、期待什么样的回应。下面是我在生产系统里用过的一份最小通信协议from dataclasses import dataclass, field from enum import Enum from typing import Any import time import uuid class MessageType(Enum): INFORM inform REQUEST request PROPOSE propose ACCEPT accept REJECT reject DELEGATE delegate dataclass class AgentMessage: sender: str receiver: str msg_type: MessageType content: dict[str, Any] correlation_id: str field(default_factorylambda: str(uuid.uuid4())) timestamp: float field(default_factorytime.time) reply_to: str | None None def create_reply(self, msg_type: MessageType, content: dict) - AgentMessage: return AgentMessage( senderself.receiver, receiverself.sender, msg_typemsg_type, contentcontent, reply_toself.correlation_id, ) class MessageBus: def __init__(self): self._subscriptions: dict[str, list[callable]] {} self._message_log: list[AgentMessage] [] def subscribe(self, topic: str, handler: callable) - None: if topic not in self._subscriptions: self._subscriptions[topic] [] self._subscriptions[topic].append(handler) def publish(self, topic: str, message: AgentMessage) - None: self._message_log.append(message) for handler in self._subscriptions.get(topic, []): handler(message) def get_history(self, correlation_id: str) - list[AgentMessage]: return [m for m in self._message_log if m.correlation_id correlation_id]这套实现是带类型的消息、用于请求-响应配对的关联追踪、一条 pub/sub 总线以及一份用于调试的消息日志。简单到容易读懂和扩展对最多几十个智能体规模的系统已经够稳健。规模再上去可以把消息总线背后换成 Redis 或一个正式的消息代理broker抽象层本身不变。任务分解与委派把问题切成智能体大小的块我怎么决定每个智能体做什么答案很简单任务分解把一个复杂问题切成能够自然映射到单个智能体上的子问题。任务分解应遵循三条原则。原则一沿自然的领域边界进行分解。如果你的问题有不同的子领域各有不同的数据需求、不同的优化目标、不同的时间动态那么这些边界就划出了智能体。在迪拜那套系统里路由、分配、预测、库存监控就是自然的领域各自有自己的数据源GPS 信号、仓库传感器、历史销售有自己的优化目标最小化通行时间、最大化仓库利用率、最小化预测误差、防止缺货有自己的更新频率路由实时、预测按小时、库存每几分钟。原则二尽量缩小智能体之间的依赖。智能体之间越多依赖彼此的输出通信开销越大系统越脆弱。目标是让智能体能半自治地跑——在有用的时候分享信息但不在彼此的回应上发生阻塞。智能体 A 必须先拿到智能体 B 的响应才能做任何事要么是依赖关系需要松绑要么是这两个功能本就该合并到一个智能体里。原则三让智能体的粒度匹配决策的粒度。别给每一个微任务都开一个智能体。要在做出有意义决策的层级上开智能体。一个决定走哪条高速的路由智能体——粒度合适一个决定走哪条车道的路由智能体——太细一个统管所有物流的智能体——太粗那就是你想摆脱的那种单体。委派是让工作在智能体之间流动的机制。最简单的委派模式是层级式的一个编排器智能体接到一个高层级任务把它拆开再把子任务委派给专职智能体。多数框架默认就用这种模式权责结构清晰时它工作得很好。迪拜那套系统里编排器接到一笔入站订单会创建用于需求核验、路线规划、仓库分配的子任务再把每一项交给对应的专家。更复杂的委派会用上合同网contract-net协议编排器把任务发布出来智能体们根据自身当前的容量和专长来投标。当多个智能体都能处理同一任务、希望系统动态做负载均衡时这种方式很合适。柏林一位客户把这种模式用在文档处理流水线pipeline上多个专职智能体法律分析、财务抽取、合规审查各自能处理某些类型的文档合同网协议保证工作是按当前队列深度分发而非靠静态分配。群体智能无中心控制的涌现式协调并不是所有多智能体系统都需要编排器。一些非常有力的多智能体行为是从大量智能体遵循简单本地规则中涌现出来的根本没有中心化协调。这就是群体智能swarm intelligence是这个领域里最反直觉的想法之一。经典的生物学例子是蚁群和蜂群。没有任何一只蚂蚁知道从蚁穴到食物源的最优路径但蚁群通过共识协作stigmergy找到了它——蚂蚁行进时留下信息素痕迹其它蚂蚁优先沿更强的痕迹前进最短路径累积出最强的信息素浓度因为蚂蚁更快完成往返。蚁群层面的行为最优路径搜索从个体规则沿更强信息素走、行进时释放信息素里涌现出来没有任何一只蚂蚁理解或主导这个过程。在软件里群体智能在优化、分布式感知、机器人协调上有实际应用。粒子群优化PSOParticle Swarm Optimization用一群候选解在搜索空间里探索每个候选解都受自身已知最佳位置与群体已知最佳位置的影响。蚁群优化ACOAnt Colony Optimization用类似的信息素机制求解像旅行商问题这样的组合优化问题。群体智能的关键在于它用最优性换鲁棒泛化性。中心化优化器能找到全局最优解但要求完整信息中央节点宕掉就是灾难性失败。群体则会在信息不完整的情况下找到一个接近最优的解并在个体智能体失败时平缓退化。对许多现实问题这个权衡是绝对划算的。多智能体系统中的博弈论智能体目标相互冲突时互动就带上了策略性。每个智能体的最优行动取决于它对其它智能体行动的预期。这是博弈论的论点它为竞争型与混合动机型多智能体系统的设计提供了数学基础。核心概念是纳什均衡Nash equilibrium在这种状态下给定其它所有智能体的策略没有任何一方能通过单方面改变自身策略来提升收益。在一套多智能体交易系统里纳什均衡可能呈现为这样的局面——动量智能体、均值回归智能体、做市智能体各自的资本配置使得在其它人配置不变的前提下谁都没法靠重新配置来改善自身的期望收益。解析地求解纳什均衡对小型博弈是可行的对生产级多智能体系统里那种复杂的、连续动作博弈则计算上不可处理。在实践里智能体通过反复互动来学到自己的策略。多智能体强化学习MARLMulti-Agent Reinforcement Learning是处理这件事的主要框架。每个智能体学一个把观测映射到动作的策略policy策略在成千上万乃至上百万次互动 episode 中协同演化。MARL 的麻烦在于非平稳性non-stationarity。从任何一个智能体的视角看环境在变化因为其它智能体在同步学习并调整自身行为。单智能体强化学习中存在的收敛性保证在多智能体场景下通常不再成立。训练可能震荡、可能发散也可能收敛到次优均衡。在多智能体设计里反复出现的另一个博弈论概念是委托代理问题principal-agent problem。编排器把一个任务委派给专职智能体时怎么核实专职智能体真的把任务做对了尤其是当专职智能体掌握编排器不具备的私有信息时在基于 LLM 的多智能体系统里它表现为验证问题——编排器怎么知道一个研究智能体的摘要是准确的或者一个写代码的智能体的输出真的能编译。实际答案是冗余加交叉验证让多个智能体独立执行同一任务再比较输出设一个专门的验证者智能体来检查其它智能体的工作把系统设计成智能体在给出输出时必须附上证据引用、测试结果、中间推理。这些方案没有一个是完美的但它们把负担从盲目信任搬到了结构化验证之上。构建多智能体系统的框架多智能体系统的工具链已经快速成熟。下面推荐一些用于生产工作的框架附上对其优劣的诚实评估。CrewAI 是我对多数团队的默认推荐。它围绕智能体、任务task以及 crew共同协作的智能体集合给出了简洁的抽象智能体以角色、目标和背景故事backstory来定义任务以描述与期望输出来定义crew 负责编排执行。CrewAI 与 LangChain 工具集成支持顺序与层级化的任务执行是我用过的所有多智能体框架里开发者体验最好的一个。Microsoft AutoGen 更灵活也更复杂。它支持对话式多智能体模式——智能体之间靠类似聊天的消息交换互动。AutoGen 的强项是对人类参与human-in-the-loop模式的支持你可以在对话中任意时刻插入一个人类智能体这对高风险应用至关重要弱点则在于当智能体数量众多、依赖关系复杂时简单的对话式会产生问题。CAMELCommunicative Agents for Mind Exploration of Large Language Model Society是另一条路径靠智能体之间的角色扮演来解决任务。一个智能体扮演AI 用户另一个扮演AI 助手二者通过结构化对话协作。CAMEL 在研究和探索方面好用对需要确定性行为的生产系统则不太合适。LangGraph本身并不是一个纯粹的多智能体框架而是一个面向 LLM 应用的图graph式编排框架——但它提供了构建多智能体系统所需的原语可以对智能体互动、状态管理、条件路由做细粒度控制。如果需要比 CrewAI 更多的控制并且愿意多写一些代码LangGraph 是合适的选择。用 CrewAI 构建多智能体系统下面给一个具体实现构建一个多智能体研究与分析系统三个智能体协作完成对某一主题的研究、对发现的分析以及一份结构化报告的产出。它对应了我为柏林和多伦多客户部署过的内容情报系统模式。from crewai import Agent, Task, Crew, Process from crewai.tools import tool tool(search_knowledge_base) def search_knowledge_base(query: str) - str: 搜索内部知识库中的相关文档。 # 生产环境中这里会连接到你的向量库 return fFound 12 relevant documents for: {query} tool(analyze_data) def analyze_data(data_description: str) - str: 对提供的数据执行定量分析。 return fAnalysis complete for: {data_description} # 定义专职智能体 researcher Agent( roleSenior Research Analyst, goalConduct thorough research on the given topic, identifying key trends, data points, and expert perspectives, backstoryYou are an experienced research analyst with 15 years in market intelligence. You excel at finding non-obvious connections between data points and synthesizing complex information into actionable insights., tools[search_knowledge_base], verboseTrue, allow_delegationTrue, ) analyst Agent( roleQuantitative Analyst, goalAnalyze research findings with rigorous quantitative methods, identifying statistical significance and causal relationships, backstoryYou are a quantitative analyst with a PhD in applied statistics. You never accept correlations at face value and always look for confounding variables and alternative explanations., tools[analyze_data], verboseTrue, allow_delegationFalse, ) writer Agent( roleReport Synthesizer, goalTransform research and analysis into a clear, structured report that non-technical stakeholders can understand and act on, backstoryYou are a technical writer who has worked with C-suite executives at Fortune 500 companies. You translate complex analysis into business language without losing precision., tools[], verboseTrue, allow_delegationFalse, ) # 定义带有清晰依赖关系的任务 research_task Task( descriptionResearch the current state of {topic}. Identify the top 5 trends, 3 key challenges, and 3 emerging opportunities. Include specific data points and source references., expected_outputA structured research brief with sections for trends, challenges, opportunities, and supporting data., agentresearcher, ) analysis_task Task( descriptionAnalyze the research findings. Validate the claimed trends with quantitative evidence. Identify which findings are statistically robust and which are speculative. Rank opportunities by potential impact and feasibility., expected_outputA quantitative analysis report with confidence levels for each finding and a ranked opportunity matrix., agentanalyst, context[research_task], # 依赖研究产出 ) report_task Task( descriptionSynthesize the research and analysis into a final executive report. Lead with the highest-confidence, highest-impact findings. Include an executive summary, detailed sections, and specific recommendations with timelines., expected_outputA polished executive report of 1500-2000 words with executive summary, findings, analysis, and recommendations., agentwriter, context[research_task, analysis_task], ) # 组装并运行 crew crew Crew( agents[researcher, analyst, writer], tasks[research_task, analysis_task, report_task], processProcess.sequential, verboseTrue, ) result crew.kickoff(inputs{topic: multi-agent AI systems in logistics}) print(result)这段代码里有几处设计可以注意。智能体之间的角色不重叠研究员找信息分析师做验证写作者负责合成哪个智能体处理哪一面没有歧义。任务之间的依赖是显式声明的——分析任务把研究任务作为上下文研究未完成前分析不能启动报告任务把研究和分析都作为上下文。这张依赖图保证信息按正确方向流动。委派是有选择性的研究员可以委派研究过程中可能请分析师核某个具体数据点而分析师和写作者不行——避免了循环委派的死循环让系统行为可预测。编排模式在框架层级的抽象之上还有更高层级的架构模式用来组织多智能体系统。模式的选择塑造了系统的可扩展性、容错能力与行为特征。顺序流水线Sequential pipeline。智能体一个接一个执行每个把输出传给下一个。这是最简单的模式也是上面 CrewAI 示例所用的模式。问题本身具备线性流程时——先研究、再分析、再写作——它工作得很好。弱点是慢总延迟等于所有智能体延迟之和也没有并行性。层级委派Hierarchical delegation。一个编排器智能体居于顶部负责拆解问题、把子任务交给专职智能体、收齐结果并合成最终输出。这就是迪拜物流系统里用的模式。编排器是唯一的权责中心让系统易于推理但也制造了一个单点故障。协作式轮转Collaborative round-robin。智能体轮流给一个共享产物添砖加瓦。智能体 A 起草一个章节B 评审并修订C 补充定量支撑A 再依据 C 的补充做下一轮修订依此类推。这种模式靠迭代细化产出高质量内容需要谨慎设计终止条件避免陷入死循环。黑板架构Blackboard architecture。智能体共享一个共同的数据结构——黑板——各自在掌握相关知识时独立地向其中提交内容。一个 agenda 机制根据黑板当前状态决定下一个动作的智能体。这是上世纪 80 年代的一种经典 AI 架构模式在基于 LLM 的系统里重获新生。问题求解过程是机会主义而非计划式的——任何智能体在任何时刻都可能给出有用的输入——这时黑板模式最合适。基于市场的协调Market-based coordination。智能体作为一个内部市场的参与者对任务竞标、交易资源。一种内部货币追踪价值创造。竞争型与混合动机系统中希望按已证实的价值而非中央规划来分配资源时这种模式合适。多伦多那套交易系统用的是简化版本每个交易智能体按自身近期的 Sharpe 比率为资本配置竞标风险管理智能体扮演做市商。下面的代码演示了黑板模式。我发现它对那些事先并不知道哪个智能体会贡献最多的开放式分析任务尤其有效from dataclasses import dataclass, field from typing import Any dataclass class BlackboardEntry: agent_id: str entry_type: str content: Any confidence: float timestamp: float class Blackboard: def __init__(self): self.entries: list[BlackboardEntry] [] self.status: str active def post(self, entry: BlackboardEntry) - None: self.entries.append(entry) def read(self, entry_type: str None) - list[BlackboardEntry]: if entry_type: return [e for e in self.entries if e.entry_type entry_type] return self.entries.copy() def get_latest(self, entry_type: str) - BlackboardEntry | None: matching self.read(entry_type) return matching[-1] if matching else None class BlackboardAgent: def __init__(self, agent_id: str, specialties: list[str]): self.agent_id agent_id self.specialties specialties def can_contribute(self, blackboard: Blackboard) - bool: 检查该智能体是否对当前状态具有相关专长。 current_types {e.entry_type for e in blackboard.entries} # 智能体的专长尚未出现或它能细化已有条目则可以贡献 return any(s not in current_types for s in self.specialties) def contribute(self, blackboard: Blackboard) - None: 向黑板提交一个贡献。 for specialty in self.specialties: if not blackboard.get_latest(specialty): entry BlackboardEntry( agent_idself.agent_id, entry_typespecialty, contentfAnalysis from {self.agent_id} on {specialty}, confidence0.85, timestamp__import__(time).time(), ) blackboard.post(entry) class BlackboardOrchestrator: def __init__(self, agents: list[BlackboardAgent], max_rounds: int 10): self.agents agents self.max_rounds max_rounds self.blackboard Blackboard() def run(self) - Blackboard: for round_num in range(self.max_rounds): contributors [a for a in self.agents if a.can_contribute(self.blackboard)] if not contributors: break for agent in contributors: agent.contribute(self.blackboard) self.blackboard.status complete return self.blackboard扩展多智能体系统——从原型到生产把一套多智能体系统从三个智能体的原型扩展到数十个智能体的生产系统会带出在小规模上并不明显的挑战。状态管理每个智能体都维护着自身的内部状态——当前任务、近期观测、对各类结论的置信度。原型阶段这些状态在内存里就行到了生产阶段需要持久化、能在崩溃后恢复、并在多个副本之间保持一致。我建议像对待微服务状态那样对待智能体状态——把它外化到数据库或状态存储把智能体设计成从外部源读写状态的无状态进程。可观测性observability。靠读取单个智能体的日志是没法调试多智能体系统的。需要的是分布式追踪从编排器开始追一个任务经过的每一个智能体呈现它们之间交换的消息、做出的决策以及每个阶段花掉的时间。带自定义 span 的 OpenTelemetry用来追踪智能体互动是最低限度可行的可观测性方案。成本管理。在基于 LLM 的多智能体系统里每一次智能体调用都要花钱。五个智能体的系统每个智能体每个任务做 3 次 LLM 调用每小时处理 100 个任务就是每小时 1500 次 LLM 调用。按当前顶级模型的定价这数字累加得很快。要剖析智能体的调用模式缓存重复查询对不需要前沿模型能力的简单智能体使用更小的模型。测试。单独测试单个智能体不算难测试智能体之间的互动才难。我推荐一种三层测试策略——单元测试覆盖单个智能体逻辑集成测试覆盖智能体两两之间的互动研究员的输出能不能成功解析为分析师的输入端到端测试覆盖完整工作流。在单元与集成测试中要把 LLM 层 mock 掉让测试保持快速和确定性。多智能体 AI 的未来多智能体系统的发展轨迹指向越来越自主、自组织的智能体生态。有几个趋势值得关注。通信协议。当前系统使用的是人工设计的通信协议。涌现式通信emergent communication方向的研究——智能体通过互动发展出自己的语言——表明智能体最终可能学会比任何手工设计协议都更高效的方案。DeepMind 与 Meta AI 在多智能体强化学习中的涌现语言相关工作已经显示智能体能发展出压缩的、任务特定的通信系统其表现优于人工设计版本。智能体市场。多智能体系统合乎逻辑的延伸是一个让来自不同开发者和组织的智能体能够互相发现、协商交互条款、共同处理任务的市场。它需要标准化的智能体接口、信任机制以及经济协议。这个方向仍在早期——Fetch.ai 和 SingularityNET 等项目正在做基于区块链的智能体市场距离主流落地仍有相当距离。递归式自我改进。一套多智能体系统如果其中一部分智能体负责监控和改进其它智能体的表现就是一种温和形式的递归式自我改进。编排器注意到研究智能体的输出质量下滑便生成新的 prompt 或微调fine-tuning数据来改进它。这件事已经在落地——我构建过一些系统其中元智能体评估其它智能体的输出并基于质量指标调整它们的系统 prompt。多模态智能体团队。当前的多智能体系统主要在文本上工作。下一代会包含处理图像、视频、音频、传感器数据的智能体跨模态协作。一个监控系统里视觉智能体检测异常、音频智能体关联声音、推理智能体合二者以评估威胁就是一个临近案例。by Gulshan Yadav编辑文婧关于我们数据派THU作为数据科学类公众号背靠清华大学大数据研究中心分享前沿数据科学与大数据技术创新研究动态、持续传播数据科学知识努力建设数据人才聚集平台、打造中国大数据最强集团军。新浪微博数据派THU微信视频号数据派THU今日头条数据派THU