把 multi-agent 系统建模成一个"内部 Slack 频道"——每个 Agent 是一个 channel member,所有协作通过"消息"完成。你不需要写状态机或显式控制流,而是定义谁能加入、谁发起、谁有权回话,让对话自然演进出结果。后端类比:把传统 RPC 编排换成事件驱动消息总线。
单个 Agent 上下文有限、技能受限——面对"读代码 + 写测试 + 跑测试 + 修 bug"这类多技能任务,你可以把它塞给一个 Agent,但代价是 prompt 越来越臃肿,模型注意力被稀释。AutoGen(微软 2023 年开源,2024 推出 v0.4 重写)的解法:把任务拆给多个专门化 Agent,让它们通过对话协作。它的核心抽象只有两个:
ConversableAgent——能收、能发、能调工具的"对话节点",底层就是 LLM + tools + memory;GroupChat + GroupChatManager——决定下一个发言者的"群主",常见策略:round-robin、由 LLM 选下一个、按角色规则选。整套系统类似 actor model——每个 Agent 是一个 actor,消息是唯一耦合方式。AutoGen v0.4 把这点做得更彻底,引入了真正的异步消息总线和分布式 runtime。
from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.teams import RoundRobinGroupChat from autogen_ext.models.openai import OpenAIChatCompletionClient model = OpenAIChatCompletionClient(model="gpt-4o") # 每个 Agent 只负责一件事——system prompt 决定它的"人格" coder = AssistantAgent("coder", model_client=model, system_message="你是 Python 工程师。只输出可运行代码,不写解释。") reviewer = AssistantAgent("reviewer", model_client=model, system_message="你是 code reviewer。指出 bug 和改进点;满意时回复 'APPROVED'。") # 群聊:轮流发言,直到 reviewer 说 APPROVED 就终止 team = RoundRobinGroupChat([coder, reviewer], termination_condition=lambda msgs: "APPROVED" in msgs[-1].content) async def main(): async for msg in team.run_stream(task="写一个二分查找,要处理空数组"): print(f"[{msg.source}] {msg.content}")
max_turns 兜底。如果 AutoGen 像"Slack 频道",CrewAI 就像"带 PM 的项目团队"——每个 Agent 有明确职责(Role)、目标(Goal)、背景(Backstory);任务(Task)被显式分配给某个 Agent;可以走顺序流(Sequential)也可以走层级流(Hierarchical,有 Manager Agent)。后端类比:从"消息驱动"变回"显式工作流引擎",更像 Airflow + Slack 的混合体。
AutoGen 的对话式协作灵活但难预测——同一个任务跑两次可能走完全不同的对话路径,对生产环境不友好。CrewAI(2024 年开源,迅速成为最流行的 multi-agent 框架之一)的设计哲学相反:让流程显式、让角色稳定。你预先定义好"3 个 Agent + 5 个 Task 的有向图",运行时按图执行。两种执行模式:
"Backstory" 不是装饰——它显著影响模型的措辞和决策风格。给 Agent 写"你是有 15 年经验的资深财报分析师,以严谨保守著称"和写"你是分析师",输出质量差距明显(已被多个 benchmark 复现)。
from crewai import Agent, Task, Crew, Process researcher = Agent( role="市场研究员", goal="找出 {topic} 的 3 个最新趋势,附数据来源", backstory="你是有十年经验的行业分析师,擅长一手数据挖掘", tools=[search_tool]) writer = Agent( role="内容编辑", goal="把研究报告改写成 800 字精炼简报", backstory="你曾是《经济学人》编辑,追求事实密度和可读性") # Task 显式分配给 Agent;context 字段声明依赖关系 research = Task(description="研究 AI Agent 2026 趋势", agent=researcher, expected_output="3 条趋势 + 引用链接") write = Task(description="改写为简报", agent=writer, context=[research], # 等 research 完成后才跑 expected_output="800 字 markdown") crew = Crew(agents=[researcher, writer], tasks=[research, write], process=Process.sequential) result = crew.kickoff(inputs={"topic": "AI Agent"})
把一个 Agent 拆成多个角色,对应后端世界的"单体 → 微服务"——不是因为"分工"是美德,而是因为小上下文 + 专门 prompt + 受限工具集三件事能显著降低单 Agent 的认知负载,让每一步推理都更准。代价也和微服务一样:通信、调试、整体一致性都变难。
给一个 Agent 同时挂上 20 个工具 + 5 段长 system prompt + 完整任务上下文,模型很容易"意图漂移":写代码写到一半开始解释架构,做研究做到一半开始问用户。Anthropic 2024 年的"Constitutional AI"和 OpenAI 的"Specialized Agents" 实验都验证了同一件事:缩窄角色范围 → 任务完成率显著提升。背后是三个机制:
判断"该不该拆"的简单启发:当某个 Agent 的失败案例大多集中在"做错了类型"(该写代码却给建议、该总结却扩写)而不是"做得不够好",就该拆。
# 反面:一个全能 Agent 既写又审又跑 generalist = AssistantAgent("engineer", system_message="你是工程师,会写代码、审代码、跑测试、修 bug、写文档...", tools=[write_code, run_tests, lint, format, git, search_docs]) # 正面:三个专门角色,每个 prompt 简短、工具收敛 coder = AssistantAgent("coder", model_client=opus, system_message="只写实现。不写测试不写文档。", tools=[write_code, search_docs]) tester = AssistantAgent("tester", model_client=haiku, # 用更便宜的模型 system_message="只写 pytest 用例并执行。报告 pass/fail。", tools=[run_tests]) reviewer = AssistantAgent("reviewer", model_client=opus, system_message="只审代码。指出最严重的 1-3 个问题,不重写。", tools=[lint])
多 Agent 之间"怎么交换信息和达成一致",是经典分布式系统协调问题的 LLM 版本——Paxos、leader election、gossip 协议、blackboard pattern 全都有对应版本。今天的 multi-agent 框架本质都是在这几种范式里组合。
当你有 N 个 Agent,必须回答两个问题:(1) 谁在什么时候发言/行动?(控制流)(2) 共享状态怎么管?(数据流)。主流四种协议:
选哪种取决于任务:任务可拆且结果可合并用 Hierarchical;有明确依赖顺序用 Sequential;需要多视角辩论用 Debate;任意子集需要访问任意中间结果用 Blackboard。生产中常混合——外层 Sequential,某一步内部 Debate。
2024-2025 年还出现了新协议层:Google 的 A2A (Agent-to-Agent) 协议把跨厂商 Agent 互操作标准化(类似 Day 6 MCP 之于工具);Anthropic Claude 的 Computer Use + sub-agents 模式让 Agent 能 spawn 子 Agent。本质都还是这四种范式的工程化。
# LangGraph:用图 + 共享状态实现混合协议(实战最常用) from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator class State(TypedDict): question: str research: Annotated[list, operator.add] # blackboard 共享区 answer: str def researcher(s): return {"research": [search_web(s["question"])]} def critic(s): return {"research": [critique(s["research"])]} def writer(s): return {"answer": synthesize(s["research"])} g = StateGraph(State) g.add_node("research", researcher); g.add_node("critic", critic) g.add_node("writer", writer) g.add_edge("research", "critic") # sequential g.add_conditional_edges("critic", # 条件路由 lambda s: "writer" if good_enough(s) else "research") g.add_edge("writer", END) g.set_entry_point("research") app = g.compile() result = app.invoke({"question": "2026 年 AI Agent 趋势?", "research": []})
总 token = Σ (每个 Agent 的 system_prompt + 累计对话历史 + 工具描述) × 该 Agent 被调用次数。最容易低估的隐藏开销:(a) 对话历史膨胀——AutoGen GroupChat 默认所有 Agent 共享完整历史,5 个 Agent × 10 轮 = 50 次 LLM 调用,每次都带全量历史,token 是 O(N²) 增长;(b) 工具描述重复——同一个工具被 3 个 Agent 用,schema 在每个 Agent 的 prompt 里都要塞一遍;(c) 失败重试——Agent 输出格式错被 reviewer 打回重做,token 翻倍;(d) 不必要的"礼貌寒暄"——Agent 之间互相确认"好的,我开始做"占了 5-10% token。降本招数:使用 prompt caching(Day 6 提到过)、定期 summarize 对话历史压缩成 memory、关闭礼貌 prompt("直接交付结果,不要确认")、用 Hierarchical 替代 Debate(前者通信量是 O(N)、后者是 O(N²))。一个典型 5-Agent 任务从 50K token 优化到 8K token 是常见的,单次任务成本能降 80%。