多智能体架构全景指南：从单兵作战到群体智能
多智能体架构全景指南：从单兵作战到群体智能
当一个 Agent 解决不了问题，一群 Agent 该如何协作？这是 2025 年之后所有严肃的 AI 工程团队都在回答的问题。
引言：为什么我们需要多智能体
如果你在 2024 年写过 Agent，大概率是这样的：一个 Prompt，一组工具，一个 While 循环，模型自己决定什么时候调工具、什么时候停下来。这就是 ReAct 范式 —— 单智能体（Single-Agent System, SAS）。
SAS 在简单任务上表现很好：查天气、订机票、写一段代码。但当你试图让一个 Agent 完成"分析这份 50 页的财报，对比同行三年数据，生成投研报告"这种复合任务时，问题就来了：
- Prompt 膨胀：你塞进 system prompt 的工具越多，模型越容易选错工具。Anthropic 的研究表明，工具数量超过 15 个后，调用准确率会断崖式下降。
- 上下文污染：一个 Agent 既要读财报、又要算财务指标、又要写报告，这些子任务的中间结果全堆在同一个上下文窗口里，互相干扰。检索增强（RAG）也救不了 —— 你检索回来的是"信息"，不是"分工"。
- 无法并行：财报分析、同行对比、宏观环境这三个子任务本来可以并行，但 SAS 只能串行执行，延迟翻三倍。
- 无法专业化：一个 Agent 的 system prompt 里既写"你是财务专家"又写"你是行业分析师"，模型在两种身份间来回切换，深度都不够。
这些问题的本质是：单 Agent 是一个"通才"，而复杂任务需要"专家分工"。多智能体系统（Multi-Agent System, MAS）就是为解决这个矛盾而生的。
但"多 Agent"本身只是一个状态，不是一种架构。三个 Agent 放在一起，可以是"各干各的最后投票"，也可以是"一个主管指挥两个下属"，还可以是"互相质询直到达成共识"。架构决定了协作方式，协作方式决定了系统能力和边界。
本文系统梳理 10 种多智能体架构：你已了解的 5 种（SAS、独立并行、中心化、去中心化辩论、混合式），以及 4 种业界在用但你尚未提及的（层级型、竞争型、市场型、黑板型）。每种架构都包含概念解释、适用场景、Mermaid 图、Python 骨架代码（讲原理）和 LangGraph / Spring AI Alibaba 工程实现对照（讲落地）。
一、架构全景：一张图看懂分类逻辑
在深入每种架构之前，先建立分类框架。我倾向于用 三个维度 来划分所有多智能体架构：
- 控制流维度：谁是决策者？—— 单点控制 / 分布式控制 / 混合控制
- 通信流维度：Agent 之间怎么说话？—— 无通信 / 通过中心 / 点对点 / 共享状态
- 协作粒度维度：Agent 之间是什么关系？—— 独立 / 层级 / 平等 / 对抗 / 交易
这三个维度的组合，就形成了下面这张全景图：
▼mermaid复制代码graph TD ROOT[多智能体架构全景] ROOT --> SAS[单智能体 SAS<br/>基线参照] ROOT --> MAS[多智能体 MAS] MAS --> INDEP[独立并行型<br/>无通信 各干各的] MAS --> COORD[协调型<br/>有中心控制] COORD --> CENT[中心化 Orchestrator<br/>主管分派任务] COORD --> HIER[层级型 Hierarchical<br/>树状分派] COORD --> BB[黑板型 Blackboard<br/>共享状态空间] MAS --> PEER[对等型<br/>点对点通信] PEER --> DEBATE[去中心化辩论<br/>多轮质询共识] PEER --> COMP[竞争型 Adversarial<br/>红蓝对抗] PEER --> HYBRID[混合型 Hybrid<br/>层级+Peer] MAS --> ECON[经济型<br/>市场机制] ECON --> MARKET[市场/拍卖型<br/>招标竞标] style SAS fill:#e1f5ff style CENT fill:#fff3e0 style DEBATE fill:#f3e5f5 style HYBRID fill:#e8f5e9 style MARKET fill:#fce4ec
读图说明：这张图不是严格的"包含关系"，而是"控制流倾向"。比如混合型既属于协调型（有层级），也属于对等型（允许 Peer 通信），所以放在对等型分支下但用了绿色标注其"混合"属性。后续每种架构我都会单独展开。
下面进入正题。为了让你建立直觉，我会先讲最简单的，最后讲最复杂的，每种架构都用同一个例子贯穿："生成一篇高质量的投研报告"，这样你能直观对比不同架构的协作方式差异。
二、SAS 单智能体系统：基线与边界
📖 概念
SAS（Single-Agent System）严格来说不是"多智能体"，但它是所有 MAS 的参照基线，必须先讲清楚。SAS 的核心是：一个 LLM 实例 + 一组工具 + 一个控制循环（ReAct / Plan-Execute 等），模型在边界内自主决定何时调用工具、何时返回最终答案。
SAS 的"智能"体现在控制循环上。最经典的是 ReAct（Reasoning + Acting）：模型先思考（Thought），再决定行动（Action），观察结果（Observation），再思考……如此循环直到任务完成。LangChain 的 AgentExecutor、Spring AI 的 ChatClient + Tool Calling 本质都是这个范式。
▼mermaid复制代码flowchart LR USER[用户输入] --> AGENT[单一 Agent<br/>LLM + Prompt + Tools] AGENT -->|Thought| THINK{需要工具吗?} THINK -->|是| TOOL[调用 Tool] TOOL -->|Observation| AGENT THINK -->|否| FINAL[返回最终答案] FINAL --> USER style AGENT fill:#e1f5ff style TOOL fill:#fff3e0
![[blue whale girl overloaded with work 4_3 Chinese text.png]]
🎯 适用场景
- 任务边界清晰：单一目标、可分解为线性步骤，如"查询北京天气并建议穿搭"。
- 工具数量 ≤ 10 个：模型能稳定选择工具，不会出现工具混淆。
- 延迟敏感场景：SAS 无 Agent 间通信开销，端到端延迟最低。
- 原型验证阶段：先用 SAS 验证核心逻辑，再决定是否引入多 Agent。
⚠️ 局限性
- Prompt 膨胀导致工具选择准确率下降
- 子任务中间结果互相污染上下文
- 无法并行执行独立子任务
- 单一 Agent 难以兼顾多个专业领域
💻 Python 骨架（讲原理）
▼python复制代码from typing import Annotated from langchain_core.tools import tool from langchain_core.messages import HumanMessage from langgraph.prebuilt import create_react_agent from langchain_openai import ChatOpenAI # 1. 定义工具 —— SAS 的"手脚" @tool def search_financial_report(company: str, year: int) -> str: """查询某公司某年的财务报告""" return f"{company} {year}年营收: 100亿, 净利润: 15亿" @tool def calculate_pe_ratio(price: float, eps: float) -> float: """计算市盈率""" return price / eps # 2. 创建 ReAct Agent —— SAS 的"大脑+循环" model = ChatOpenAI(model="gpt-4o") tools = [search_financial_report, calculate_pe_ratio] agent = create_react_agent(model, tools) # 3. 执行 —— 一个 while 循环搞定 result = agent.invoke({ "messages": [HumanMessage(content="分析贵州茅台2024年财报，计算市盈率（股价1500，EPS 50），给出投资建议")] }) print(result["messages"][-1].content)
🔧 Spring AI Alibaba 工程实现
▼java复制代码@SpringBootApplication public class SasDemo { public static void main(String[] args) { SpringApplication.run(SasDemo.class, args); } @Bean public ChatClient chatClient(ChatClient.Builder builder) { // Spring AI 1.0 的 ChatClient + Tool Calling = SAS return builder .defaultSystem("你是专业的金融分析师") .defaultTools(new FinancialTools()) .build(); } static class FinancialTools { @Tool(description = "查询某公司某年的财务报告") String searchFinancialReport(String company, int year) { return company + year + "年营收: 100亿, 净利润: 15亿"; } @Tool(description = "计算市盈率") double calculatePeRatio(double price, double eps) { return price / eps; } } } // 调用 @RestController class ReportController { private final ChatClient chatClient; ReportController(ChatClient chatClient) { this.chatClient = chatClient; } @GetMapping("/analyze") String analyze() { return chatClient.prompt("分析贵州茅台2024年财报，计算市盈率（股价1500，EPS 50），给出投资建议") .call() .content(); } }
关键差异：Spring AI 1.0 用 @Tool 注解声明工具，ChatClient 内部自动处理 ReAct 循环；LangGraph 的 create_react_agent 是显式构造。两者都是 SAS 范式，只是抽象层级不同。
三、MAS 独立并行型：多 Agent 各干各的
📖 概念
这是最简单的 MAS：多个结构相同或相似的 Agent 并行执行同一种任务，最后聚合结果。Agent 之间没有任何通信，互不知道对方的存在。它的价值在于"用数量换质量"——单个 Agent 可能出错或偏差，但多个 Agent 独立完成后取平均/多数表决/择优，能显著降低方差。
这种架构在学术上叫 Self-Consistency（自洽性）或 Ensemble（集成），本质是机器学习中"集成学习"思想在 LLM 上的应用。OpenAI o1 系列在数学推理上的高分，部分就来自这种并行采样 + 多数表决的推理时计算（inference-time compute）策略。
一个容易被忽视的要点：并行型不是"分工"，而是"冗余"。三个 Agent 都在做"生成投研报告"，而不是"一个写宏观、一个写行业、一个写公司"。前者是独立并行，后者是中心化分工。区分这两个架构的关键就在这里。
▼mermaid复制代码flowchart TD USER[用户输入: 生成投研报告] --> SPLIT[任务分发器] SPLIT --> A1[Agent 1<br/>Prompt A] SPLIT --> A2[Agent 2<br/>Prompt B] SPLIT --> A3[Agent 3<br/>Prompt C] A1 --> AGG[结果聚合器<br/>择优/投票/合并] A2 --> AGG A3 --> AGG AGG --> FINAL[最终报告] style SPLIT fill:#fff3e0 style AGG fill:#e8f5e9
🎯 适用场景
- 创意生成：生成广告文案、slogan，多个 Agent 各出方案，人工或模型择优。
- 高风险决策：投资建议、医疗诊断，多个独立 Agent 各自推理后投票，降低单点错误。
- 数学/代码推理：复杂推理任务，多次采样取多数答案（Self-Consistency 经典用法）。
- 数据标注/评估：多个 Agent 对同一文本打标签或评分，取一致结果提升标注质量。
- A/B 测试 Prompt：同一任务用不同 Prompt 并行跑，对比哪个 Prompt 效果好。
⚠️ 局限性
- 成本是 SAS 的 N 倍（N = Agent 数量）
- 不适合需要分工的复合任务
- 聚合策略本身是个难题（多数表决？加权？LLM 裁判？）
💻 Python 骨架（讲原理）
▼python复制代码import asyncio from langchain_openai import ChatOpenAI async def run_agent(prompt: str, temperature: float) -> str: """单个 Agent，用不同 temperature 增加多样性""" model = ChatOpenAI(model="gpt-4o", temperature=temperature) resp = await model.ainvoke(prompt) return resp.content async def parallel_agents(task: str, n: int = 3) -> list[str]: """并行启动 N 个 Agent，互不通信""" # 用不同 temperature 让结果有多样性 prompts = [ f"你是一位严谨的价值投资分析师，请完成：{task}", f"你是一位成长股投资分析师，请完成：{task}", f"你是一位量化分析师，请完成：{task}", ] temps = [0.3, 0.7, 0.9] # asyncio.gather 实现真并行 results = await asyncio.gather(*[ run_agent(p, t) for p, t in zip(prompts, temps) ]) return results async def aggregate(results: list[str]) -> str: """用 LLM 做裁判，择优合并""" judge = ChatOpenAI(model="gpt-4o", temperature=0) joined = "\n\n---\n\n".join([f"方案 {i+1}:\n{r}" for i, r in enumerate(results)]) return (await judge.ainvoke( f"以下是3份投研报告，请综合三份的优点，生成一份最终报告：\n{joined}" )).content # 主流程 async def main(): results = await parallel_agents("分析贵州茅台2024年投资价值", n=3) final = await aggregate(results) print(final) asyncio.run(main())
🔧 LangGraph 工程实现
▼python复制代码from langgraph.graph import StateGraph, START, END from typing import TypedDict import asyncio class State(TypedDict): task: str results: list[str] final: str def fan_out(state: State) -> State: """并行分发 —— 这里返回多个节点名实现扇出""" return state # 实际由 conditional_edges 控制 def agent_node(state: State, prompt_style: str) -> dict: # 省略 LLM 调用，返回 {"results": [report]} return {"results": [f"{prompt_style} 的报告"]} def aggregate_node(state: State) -> dict: # LLM 裁判聚合 return {"final": "综合报告"} # 构建图：fan-out → 3个并行 agent → fan-in → aggregate graph = StateGraph(State) graph.add_node("agent_1", lambda s: agent_node(s, "价值投资")) graph.add_node("agent_2", lambda s: agent_node(s, "成长股")) graph.add_node("agent_3", lambda s: agent_node(s, "量化")) graph.add_node("aggregate", aggregate_node) graph.add_edge(START, "agent_1") graph.add_edge(START, "agent_2") graph.add_edge(START, "agent_3") # 同一起点引出3条边 = 并行 graph.add_edge("agent_1", "aggregate") graph.add_edge("agent_2", "aggregate") graph.add_edge("agent_3", "aggregate") graph.add_edge("aggregate", END) app = graph.compile()
LangGraph 的并行秘密：从同一个节点（如 START）引出多条边，LangGraph 会自动并行执行后续节点；多个节点指向同一个节点（如 aggregate），会自动等待所有上游完成再执行（barrier）。这就是"Map-Reduce"模式。
🔧 Spring AI Alibaba 对照
▼java复制代码@Service public class ParallelReportService { private final ChatClient chatClient; public ParallelReportService(ChatClient.Builder builder) { this.chatClient = builder.build(); } // 用 CompletableFuture 实现并行 public String generateParallel(String task) { List<String> personas = List.of("价值投资分析师", "成长股分析师", "量化分析师"); List<CompletableFuture<String>> futures = personas.stream() .map(persona -> CompletableFuture.supplyAsync(() -> chatClient.prompt() .system("你是专业的" + persona) .user(task) .call() .content() )) .toList(); // 等待全部完成 CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); List<String> results = futures.stream() .map(CompletableFuture::join) .toList(); // LLM 裁判聚合 return chatClient.prompt() .system("综合以下多份报告的优点，生成最终报告") .user(String.join("\n---\n", results)) .call() .content(); } }
Spring 侧要点：用 CompletableFuture.supplyAsync 实现并行，本质和 Python 的 asyncio.gather 一样。Spring AI Alibaba 暂未提供原生的多 Agent 编排图（截至 1.0），需要自己用 CompletableFuture 或 Reactor 组合，这也是它的抽象层级比 LangGraph 低的地方。
四、MAS 中心化（Centralized / Orchestrator）：主管分派
📖 概念
中心化架构是工业界最常见的 MAS 形态，也是大多数人对"多 Agent 协作"的第一直觉。它的核心是 一个协调者（Orchestrator / Supervisor / Planner）+ 多个专业 Agent：协调者负责理解用户意图、分解任务、把子任务分派给合适的 Agent、收集结果、裁决冲突、最终汇总。子 Agent 之间不直接通信，所有协作都通过协调者中转。
这种架构的本质是 "主管-下属"模式 映射到 AI 系统。协调者像一个项目经理：他不需要自己写代码、做设计、写文档，但他知道每个下属擅长什么，能把一个复杂项目拆成可执行的子任务，并保证最终交付质量。下属之间不串门聊天，只对主管汇报。
中心化架构有两种子模式，区分它们很重要：
- 静态分派（Plan-then-Execute）：协调者先一次性规划好所有子任务和分派方案，再按计划执行。适合任务结构清晰、子任务之间有依赖关系的场景。LangGraph 的
StateGraph显式定义边、Spring AI Alibaba 的Graph编排都属于这类。 - 动态分派（Reactive Supervisor）：协调者在每一步根据当前状态决定下一步该谁干活，子任务序列在运行时动态生成。适合任务不可预测、需要根据中间结果调整的场景。LangGraph 的
Command+Send、AutoGen 的GroupChat都支持这种模式。
▼mermaid复制代码flowchart TD USER[用户: 生成茅台投研报告] --> ORCH[Orchestrator 协调者<br/>分解任务+分派+汇总] ORCH -->|子任务1: 宏观分析| A1[宏观分析师 Agent] ORCH -->|子任务2: 行业分析| A2[行业分析师 Agent] ORCH -->|子任务3: 公司分析| A3[公司分析师 Agent] ORCH -->|子任务4: 估值计算| A4[量化 Agent] A1 -->|结果| ORCH A2 -->|结果| ORCH A3 -->|结果| ORCH A4 -->|结果| ORCH ORCH -->|裁决+整合| FINAL[最终投研报告] ORCH -.->|不满意,要求重做| A3 style ORCH fill:#fff3e0,stroke:#e65100,stroke-width:3px
注意图中的虚线：协调者对结果不满意时，可以让某个 Agent 重做。这是中心化架构的一大优势——它具有全局视角，可以做质量控制和迭代。
🎯 适用场景
- 复合任务需要专业分工：投研报告（宏观+行业+公司+估值）、技术方案设计（架构+安全+成本）、内容生产（选题+写作+配图+审校）。
- 子任务有依赖关系：必须先做 A 再做 B，协调者负责管理执行顺序。
- 需要质量控制：协调者可以检查每个子结果，不满意要求重做或转交其他 Agent。
- 企业级 Workflow：客服中心（路由 Agent → 售前/售后/技术 Agent）、RPA 流程自动化。
- 工具数量超过单 Agent 上限：把工具按领域分组，每个专业 Agent 只装载自己领域的工具，规避 Prompt 膨胀。
⚠️ 局限性
- 单点瓶颈：协调者是性能瓶颈和单点故障，它挂了整个系统就停了。
- 协调者能力决定上限：如果协调者的任务分解能力差，整个系统输出质量就差。这是"主管能力决定团队产出"的 AI 版本。
- 通信开销：所有子结果都要经过协调者中转，延迟比直接通信高。
- 不适合开放性探索：任务不可预测、需要 Agent 之间灵活协商的场景，中心化的固定分派反而僵化。
💻 Python 骨架（讲原理）
▼python复制代码from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI class State(TypedDict): task: str macro_report: str # 宏观分析结果 industry_report: str # 行业分析结果 company_report: str # 公司分析结果 valuation: str # 估值结果 final_report: str next: str # 协调者决定下一步 model = ChatOpenAI(model="gpt-4o") # 1. 协调者：分解任务 + 决定下一步 def orchestrator(state: State) -> dict: """主管：看当前状态，决定下一步该谁干活""" if not state.get("macro_report"): return {"next": "macro_agent"} elif not state.get("industry_report"): return {"next": "industry_agent"} elif not state.get("company_report"): return {"next": "company_agent"} elif not state.get("valuation"): return {"next": "valuation_agent"} else: return {"next": "summarize"} # 2. 专业 Agent（每个只管自己的一块） def macro_agent(state: State) -> dict: resp = model.invoke(f"你是宏观经济分析师，分析当前经济环境对白酒行业的影响。任务：{state['task']}") return {"macro_report": resp.content} def industry_agent(state: State) -> dict: resp = model.invoke(f"你是白酒行业分析师。已知宏观背景：{state.get('macro_report','')}。分析白酒行业趋势") return {"industry_report": resp.content} def company_agent(state: State) -> dict: resp = model.invoke(f"你是公司分析师。行业背景：{state.get('industry_report','')}。分析贵州茅台公司基本面") return {"company_report": resp.content} def valuation_agent(state: State) -> dict: resp = model.invoke(f"你是量化分析师。公司基本面：{state.get('company_report','')}。给出估值") return {"valuation": resp.content} def summarize(state: State) -> dict: resp = model.invoke(f"整合以下内容生成投研报告：\n宏观：{state['macro_report']}\n行业：{state['industry_report']}\n公司：{state['company_report']}\n估值：{state['valuation']}") return {"final_report": resp.content} # 3. 条件路由：协调者决定下一步 def route(state: State) -> str: return state["next"] # 4. 构建图 graph = StateGraph(State) graph.add_node("orchestrator", orchestrator) graph.add_node("macro_agent", macro_agent) graph.add_node("industry_agent", industry_agent) graph.add_node("company_agent", company_agent) graph.add_node("valuation_agent", valuation_agent) graph.add_node("summarize", summarize) graph.add_edge(START, "orchestrator") graph.add_conditional_edges("orchestrator", route, { "macro_agent": "macro_agent", "industry_agent": "industry_agent", "company_agent": "company_agent", "valuation_agent": "valuation_agent", "summarize": "summarize", }) # 每个 agent 干完活回到 orchestrator，形成"汇报-分派"循环 for agent in ["macro_agent", "industry_agent", "company_agent", "valuation_agent"]: graph.add_edge(agent, "orchestrator") graph.add_edge("summarize", END) app = graph.compile()
这段代码的精髓：conditional_edges + "每个 agent 回到 orchestrator" 的边，构成了一个 "协调者循环"。协调者每次被唤醒都重新评估状态、决定下一步，这就是"动态分派"。如果改成显式地 START → macro → industry → company → valuation → summarize，就是"静态分派"。同一个框架，两种模式，区别只在边的定义。
🔧 LangGraph 的 Command + Send 模式（更高级）
LangGraph 0.2+ 引入了 Command 和 Send，让协调者可以动态扇出子任务，比上面的固定循环更灵活：
▼python复制代码from langgraph.types import Command, Send def orchestrator(state: State) -> Command[Literal["agent"]]: """协调者动态决定派发哪些子任务""" subtasks = ["宏观分析", "行业分析", "公司分析", "估值计算"] # Send 让协调者把不同 payload 发给同一个节点，自动并行 return Command(goto=[Send("agent", {"task": t}) for t in subtasks]) def agent(state: dict) -> dict: # 单一 agent 节点，处理任意子任务 return {"results": [...]}
Send 的价值：你不需要预先定义 N 个 agent 节点，一个节点就能接收 N 个不同任务并行执行。这在子任务数量不固定时特别有用。
🔧 Spring AI Alibaba Graph 对照
Spring AI Alibaba 1.x 提供了 spring-ai-alibaba-graph 模块，灵感来自 LangGraph：
▼java复制代码// Spring AI Alibaba Graph 定义（伪代码，展示 API 风格） @Graph public class InvestmentReportGraph { @Node public State orchestrator(State state) { // 协调者：决定下一步 if (state.get("macro_report") == null) state.set("next", "macro_agent"); else if (state.get("industry_report") == null) state.set("next", "industry_agent"); else state.set("next", "summarize"); return state; } @Node public State macroAgent(State state) { String result = chatClient.prompt() .system("你是宏观经济分析师") .user(state.get("task")) .call().content(); state.put("macro_report", result); return state; } @Edge(source = "orchestrator") public String route(State state) { return state.get("next"); } @Edge(source = "macro_agent", target = "orchestrator") public void macroBack() {} @Edge(source = "START", target = "orchestrator") public void start() {} @Edge(source = "summarize", target = "END") public void end() {} }
对照要点：Spring AI Alibaba Graph 的 @Node + @Edge 注解风格与 LangGraph 的 add_node + add_edge 一一对应，但 Java 侧用注解声明，Python 侧用 DSL 构造。两者都支持条件路由（@Edge 返回字符串 / add_conditional_edges）。Spring 侧的优势是和 Spring 生态（事务、配置、Actuator 监控）无缝集成。
五、MAS 去中心化辩论（Decentralized Debate）：点对点质询
📖 概念
去中心化辩论架构放弃了"主管"，让 Agent 之间 点对点交流、互相质询、多轮改进，最终通过多数表决或共识达成结论。这种架构的灵感来自学术界的同行评审（Peer Review）和辩论赛：真理越辩越明，多个有不同立场的 Agent 通过对抗性讨论，能发现单一 Agent 发现不了的漏洞。
这种架构最著名的实现是 Multi-Agent Debate（MAD） 论文（Du et al., 2023），核心发现是：让两个 LLM 对同一问题各自生成答案，然后互相看对方的答案并改进，多轮迭代后准确率显著提升，甚至超过简单扩大模型参数的效果。这就是 "Society of Minds" 思想——智能不只来自单个模型，更来自模型间的交互。
辩论架构的关键设计要素有三个：
- 角色分配：Agent 要有不同的立场或专长，否则辩论会变成"附和"。常见模式有：正方/反方/裁判、多专家+批评者、生成者+审核者。
- 辩论轮数控制：不能无限辩下去，要有最大轮数限制或收敛检测。否则成本失控。
- 收敛机制：如何从分歧到达共识？多数表决、裁判裁决、或者最后一轮让 Agent 自己合并。
▼mermaid复制代码sequenceDiagram participant U as 用户 participant A1 as Agent 1<br/>(多头视角) participant A2 as Agent 2<br/>(空头视角) participant A3 as Agent 3<br/>(裁判) U->>A1: 生成初版分析 U->>A2: 生成初版分析 A1-->>A2: 我看多,理由是... A2-->>A1: 我看空,你的论据有漏洞... Note over A1,A2: 第1轮辩论 A1->>A1: 改进观点(吸收A2的批评) A2->>A2: 改进观点(吸收A1的批评) A1-->>A2: 第2轮质询... A2-->>A1: 第2轮反驳... Note over A1,A2: 第N轮辩论 A1->>A3: 提交最终观点 A2->>A3: 提交最终观点 A3->>U: 裁判综合,输出结论
注意：A3（裁判）是可选的。如果没有裁判，就是纯粹的多数表决（两个 Agent 各自输出，取一致部分）；如果有裁判，裁判本身也是一个 Agent，负责综合各方观点。裁判的存在让架构略微"中心化"了，所以纯粹的辩论架构通常不用裁判，而是用"最后一轮合并"的方式收敛。
🎯 适用场景
- 高风险判断需要多角度验证：投资决策（多空辩论）、医疗诊断（不同专科医生讨论）、法律案件分析。
- 开放性问题没有标准答案：产品策略、技术路线选型，多个 Agent 代表不同利益方辩论，能暴露盲点。
- 对抗 Prompt 攻击：一个 Agent 生成内容，另一个 Agent 专门找漏洞（这其实更接近"竞争型"，见后文）。
- 推理质量提升：数学证明、逻辑推理，多个 Agent 互相挑错能提升准确率。
- 减少 LLM 幻觉：辩论过程中 Agent 会指出对方的幻觉，最终结果更可靠。
⚠️ 局限性
- 成本和延迟高：N 个 Agent 辩论 M 轮，调用次数是 N×M，比 SAS 高一个数量级。
- 可能陷入"附和"：如果 Agent 角色设计不好，容易变成"你说得对，我补充一点"，失去对抗性。
- 收敛困难：分歧很大时，多轮辩论可能无法收敛，需要人工干预或强制裁决。
- 不适合事实性任务：查天气、订机票这种有明确答案的任务，辩论纯属浪费。
💻 Python 骨架（讲原理）
▼python复制代码from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI class State(TypedDict): question: str round: int answer_1: str answer_2: str final_answer: str MAX_ROUNDS = 3 model = ChatOpenAI(model="gpt-4o") def agent_1(state: State) -> dict: """多头分析师：看多茅台""" round_num = state.get("round", 0) if round_num == 0: # 第一轮：独立生成 resp = model.invoke(f"你是看多茅台的分析师。问题：{state['question']}。给出你的分析。") else: # 后续轮：看对方的观点并反驳/改进 resp = model.invoke( f"你是看多茅台的分析师。对手的观点是：\n{state['answer_2']}\n\n" f"请指出对方论据的漏洞，并改进你之前的观点。你上一轮的观点：\n{state['answer_1']}" ) return {"answer_1": resp.content} def agent_2(state: State) -> dict: """空头分析师：看空茅台""" round_num = state.get("round", 0) if round_num == 0: resp = model.invoke(f"你是看空茅台的分析师。问题：{state['question']}。给出你的分析。") else: resp = model.invoke( f"你是看空茅台的分析师。对手的观点是：\n{state['answer_1']}\n\n" f"请指出对方论据的漏洞，并改进你之前的观点。你上一轮的观点：\n{state['answer_2']}" ) return {"answer_2": resp.content} def should_continue(state: State) -> Literal["debate", "conclude"]: """判断是否继续辩论""" if state["round"] >= MAX_ROUNDS: return "conclude" return "debate" def conclude(state: State) -> dict: """最后一轮：让一个中立的 Agent 综合两方观点""" resp = model.invoke( f"你是中立的裁判。以下是多空双方经过{state['round']}轮辩论的最终观点：\n" f"多方：{state['answer_1']}\n\n空方：{state['answer_2']}\n\n" f"请综合双方合理观点，给出平衡的结论。" ) return {"final_answer": resp.content} # 构建辩论图 graph = StateGraph(State) graph.add_node("agent_1", agent_1) graph.add_node("agent_2", agent_2) graph.add_node("conclude", conclude) # 辩论循环：agent_1 → agent_2 → 判断是否继续 graph.add_edge(START, "agent_1") graph.add_edge("agent_1", "agent_2") graph.add_conditional_edges("agent_2", should_continue, { "debate": "agent_1", # 继续辩论 "conclude": "conclude", # 进入结论 }) graph.add_edge("conclude", END) app = graph.compile()
辩论架构的图模式：核心是一个 "agent_1 → agent_2 → 判断 → 回到 agent_1"的环，环的退出条件是轮数达到上限。这种"环 + 条件退出"是所有迭代型 MAS 的通用模式。
🔧 Spring AI Alibaba 对照
▼java复制代码@Service public class DebateService { private final ChatClient chatClient; private static final int MAX_ROUNDS = 3; public DebateService(ChatClient.Builder builder) { this.chatClient = builder.build(); } public String debate(String question) { String answer1 = ""; String answer2 = ""; for (int round = 0; round <= MAX_ROUNDS; round++) { final int r = round; final String prev1 = answer1; final String prev2 = answer2; // 两个 Agent 并行生成各自的回应 CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> r == 0 ? chatClient.prompt().system("你是看多分析师").user(question).call().content() : chatClient.prompt() .system("你是看多分析师。对手观点：" + prev2 + "。请反驳并改进你上轮观点：" + prev1) .user(question).call().content() ); CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> r == 0 ? chatClient.prompt().system("你是看空分析师").user(question).call().content() : chatClient.prompt() .system("你是看空分析师。对手观点：" + prev1 + "。请反驳并改进你上轮观点：" + prev2) .user(question).call().content() ); CompletableFuture.allOf(f1, f2).join(); answer1 = f1.join(); answer2 = f2.join(); } // 裁判综合 return chatClient.prompt() .system("你是中立裁判，综合多空双方观点给出平衡结论") .user("多方：" + answer1 + "\n空方：" + answer2) .call().content(); } }
对照要点：Spring 侧因为暂无原生的"图循环"抽象，辩论架构用 for 循环 + CompletableFuture 实现，比 LangGraph 的图模式更直观但更"过程式"。如果你用 Spring AI Alibaba Graph，可以用 @Edge 的条件路由实现类似的循环，但当前版本对循环的支持还在完善中。这是 LangGraph 在 MAS 编排上的核心优势——原生支持循环和条件退出。
六、MAS 混合式（Hybrid）：层级 + Peer
📖 概念
混合式架构是中心化和去中心化的"折中产物"：保留一个协调者做全局管控，但允许子 Agent 之间按需直接通信。这是对纯中心化架构僵化问题的改进——有些子任务之间的协作如果都要经过协调者中转，延迟高且信息损耗大，不如让它们直接对话。
一个典型场景：投研报告生成中，"行业分析师"和"公司分析师"需要反复对齐——行业分析师发现行业增速下滑，要通知公司分析师下调预期；公司分析师发现公司有新产能，要通知行业分析师修正行业供给预测。如果走纯中心化，每次都要"公司分析师 → 协调者 → 行业分析师 → 协调者 → 公司分析师"，来回四次。混合式允许他们直接对话：发现新信息后直接 Peer-to-Peer 交流，对齐后再各自向协调者汇报。
混合式的关键设计挑战是 "什么时候该走层级，什么时候该走 Peer"。常见的策略有：
- 白名单 Peer：协调者预先指定哪些 Agent 对可以直接通信（如行业和公司分析师可以互连，但都不能直接找估值 Agent）。
- 请求式 Peer：Agent 想和另一个 Agent 通信时，先向协调者申请，协调者批准后建立临时通道。
- 阶段式切换：某些阶段走层级（如任务分解、最终汇总），某些阶段走 Peer（如中间对齐）。
▼mermaid复制代码flowchart TD USER[用户] --> ORCH[Orchestrator 协调者] ORCH -->|分派| A1[宏观分析师] ORCH -->|分派| A2[行业分析师] ORCH -->|分派| A3[公司分析师] ORCH -->|分派| A4[量化分析师] A2 <-.Peer 通信.-> A3 A1 <-.Peer 通信.-> A2 A1 -->|汇报| ORCH A2 -->|汇报| ORCH A3 -->|汇报| ORCH A4 -->|汇报| ORCH ORCH --> FINAL[最终报告] style ORCH fill:#fff3e0,stroke:#e65100,stroke-width:3px style A2 fill:#e8f5e9 style A3 fill:#e8f5e9
图中的实线是层级汇报，虚线是 Peer 通信。注意 Peer 通信是双向的，且不经过协调者。
🎯 适用场景
- 子任务之间有强耦合：如行业分析和公司分析需要反复对齐，纯中心化的中转效率太低。
- 大型复杂项目：层级保证整体秩序，Peer 保证局部高效，适合几十个 Agent 的大型系统。
- 需要局部协商：多个 Agent 共同决策某个子问题时，让它们直接讨论比都汇报给主管更高效。
- MetaGPT 式软件开发：产品经理、架构师、工程师、测试之间有层级（PM 指挥工程师），但架构师和工程师之间需要频繁直接沟通技术细节。
⚠️ 局限性
- 复杂度最高：既有层级又有 Peer，系统设计和调试难度都很大。
- 通信管理复杂：需要明确哪些 Agent 对可以 Peer，哪些必须走层级，否则容易混乱。
- 死锁风险：Peer 通信如果设计不当，可能出现 A 等 B、B 等 A 的死锁。
- 可观测性差：Agent 之间的 Peer 通信发生在协调者视线之外，调试和审计更难。
💻 Python 骨架（LangGraph 实现）
LangGraph 实现混合式的关键是 共享 State：所有 Agent 读写同一个状态对象，Peer 通信本质上就是"一个 Agent 写入 State，另一个 Agent 读取"，不需要显式的消息传递。
▼python复制代码from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI class State(TypedDict): task: str macro_report: str industry_report: str company_report: str final_report: str # Peer 通信通道：行业和公司分析师通过这两个字段对齐 industry_to_company_note: str # 行业分析师给公司分析师的备注 company_to_industry_note: str # 公司分析师给行业分析师的备注 model = ChatOpenAI(model="gpt-4o") def orchestrator(state: State) -> dict: """协调者：分派任务，这里简化为静态分派""" # 实际中会根据状态动态决定 pass def macro_agent(state: State) -> dict: resp = model.invoke(f"宏观分析。任务：{state['task']}") return {"macro_report": resp.content, "industry_to_company_note": "宏观环境偏紧，请行业分析师关注"} # Peer 通道 def industry_agent(state: State) -> dict: # 读取公司分析师发来的 Peer 备注（如果有） peer_note = state.get("company_to_industry_note", "") prompt = f"行业分析。宏观背景：{state.get('macro_report','')}" if peer_note: prompt += f"\n公司分析师的反馈：{peer_note}" resp = model.invoke(prompt) return {"industry_report": resp.content} def company_agent(state: State) -> dict: # 读取行业分析师发来的 Peer 备注 peer_note = state.get("industry_to_company_note", "") prompt = f"公司分析。行业背景：{state.get('industry_report','')}" if peer_note: prompt += f"\n行业分析师的提醒：{peer_note}" resp = model.invoke(prompt) # 公司分析师也通过 Peer 通道反馈 return {"company_report": resp.content, "company_to_industry_note": "公司有新产能投产，请行业分析师修正供给预测"} def summarize(state: State) -> dict: resp = model.invoke(f"整合生成报告：宏观{state['macro_report']} 行业{state['industry_report']} 公司{state['company_report']}") return {"final_report": resp.content} # 构建图：行业和公司之间通过共享 State 实现 Peer graph = StateGraph(State) graph.add_node("macro_agent", macro_agent) graph.add_node("industry_agent", industry_agent) graph.add_node("company_agent", company_agent) graph.add_node("summarize", summarize) # 层级：顺序执行 graph.add_edge(START, "macro_agent") graph.add_edge("macro_agent", "industry_agent") graph.add_edge("industry_agent", "company_agent") # 行业 → 公司，公司能读到行业的 Peer note graph.add_edge("company_agent", "summarize") graph.add_edge("summarize", END) app = graph.compile()
混合式在 LangGraph 中的实现精髓：不需要额外的"Peer 通信协议"，共享 State 本身就是通信介质。任何 Agent 写入 State 的字段，其他 Agent 都能读到。如果你需要双向对齐（多轮 Peer 通信），就在行业和公司之间加一个循环边，让它们互相读对方的 note 并迭代。
🔧 Spring AI Alibaba 对照
Spring 侧实现 Peer 通信最自然的方式是用 共享的上下文对象（类似 LangGraph 的 State）：
▼java复制代码public class SharedContext { private Map<String, String> data = new ConcurrentHashMap<>(); public void put(String key, String value) { data.put(key, value); } public String get(String key) { return data.get(key); } } @Service public class HybridReportService { private final ChatClient chatClient; public String generate(String task) { SharedContext ctx = new SharedContext(); ctx.put("task", task); // 1. 宏观分析（层级执行） String macro = chatClient.prompt() .system("你是宏观分析师") .user(task).call().content(); ctx.put("macro_report", macro); ctx.put("industry_to_company_note", "宏观偏紧，请关注"); // Peer 通道 // 2. 行业分析（读取公司可能发来的 note） String peerNote = ctx.get("company_to_industry_note"); String industry = chatClient.prompt() .system("你是行业分析师。宏观：" + macro + (peerNote != null ? " 公司反馈：" + peerNote : "")) .user(task).call().content(); ctx.put("industry_report", industry); // 3. 公司分析（读取行业的 note，并通过 ctx 反馈） String peerFromIndustry = ctx.get("industry_to_company_note"); String company = chatClient.prompt() .system("你是公司分析师。行业：" + industry + (peerFromIndustry != null ? " 行业提醒：" + peerFromIndustry : "")) .user(task).call().content(); ctx.put("company_report", company); ctx.put("company_to_industry_note", "公司有新产能，请修正供给预测"); // Peer 反馈 // 实际项目中这里可以加一个循环让行业和公司多轮对齐 return chatClient.prompt() .system("整合生成报告") .user("宏观:" + macro + " 行业:" + industry + " 公司:" + company) .call().content(); } }
对照要点：Spring 侧的 SharedContext（本质是个 ConcurrentHashMap）承担了 LangGraph 中 State 的角色。这是混合式架构在两种语言里的一致实现思路——用共享状态代替显式消息传递。如果你需要更复杂的 Peer 通信（如异步通知），Spring 侧可以引入 ApplicationEventPublisher 做事件驱动，LangGraph 侧可以用 Command 做动态路由。
七、层级型（Hierarchical）：树状分派
📖 概念
层级型是中心化的"放大版"。当系统规模大到单一协调者管不过来时，就引入多级协调者，形成树状结构：顶层主管 → 中层经理 → 一线执行 Agent。每一级只管它的直接下级，不越级指挥。这是对人类组织架构的直接映射——公司 CEO 不会直接指挥一线程序员，而是通过 CTO → 技术总监 → 开发组长层层下达。
层级型和中心化的关键区别在于 "管理幅度"。中心化是一个主管直接管 N 个 Agent，当 N 超过 7±2（著名的 Miller 数字）时，主管的决策质量会下降——它要同时理解 N 个 Agent 的状态、分配任务、裁决冲突，认知负荷过载。层级型的解决方案是把 N 个 Agent 分组，每组 5-7 个，由一个中层经理管，主管只和几个经理打交道。
一个具体例子：生成一份覆盖 50 家公司的行业研究报告。中心化架构下，一个协调者要同时管 50 个公司分析 Agent，状态管理极其复杂。层级型做法：顶层协调者分派给 5 个行业经理（每个管一个子行业），每个行业经理再分派给 10 个公司分析 Agent。顶层协调者只需要和 5 个经理交互，每个经理只需要和 10 个 Agent 交互，复杂度从 O(50) 降到 O(5)+O(10)。
▼mermaid复制代码graph TD TOP[顶层 Orchestrator<br/>统筹全局] TOP -->|分派行业| M1[消费行业经理] TOP -->|分派行业| M2[科技行业经理] TOP -->|分派行业| M3[金融行业经理] M1 --> A1[公司分析 Agent 1] M1 --> A2[公司分析 Agent 2] M1 --> A3[公司分析 Agent 3] M2 --> A4[公司分析 Agent 4] M2 --> A5[公司分析 Agent 5] M3 --> A6[公司分析 Agent 6] M3 --> A7[公司分析 Agent 7] A1 -->|汇报| M1 A2 -->|汇报| M1 M1 -->|汇总| TOP TOP --> FINAL[最终行业报告] style TOP fill:#fff3e0,stroke:#e65100,stroke-width:3px style M1 fill:#ffe0b2 style M2 fill:#ffe0b2 style M3 fill:#ffe0b2
🎯 适用场景
- 大规模 Agent 系统（>10 个 Agent）：单一协调者管不过来，必须分层。
- 天然层级任务：如组织级报告生成（总部 → 分公司 → 部门）、多层级 RAG（粗检索 → 细检索 → 精排）。
- MetaGPT 式软件开发：PM → 架构师 → 工程师 → 测试，天然是层级结构。
- 多地区/多业务线：全球化公司，顶层管地区经理，地区经理管本地 Agent。
⚠️ 局限性
- 信息失真：信息层层上报会损耗，类似人类组织的"层层过滤"。
- 延迟累积：每层都有决策延迟，层数越多端到端延迟越高。
- 上层故障影响面大：顶层协调者挂了，整个系统停摆（比中心化更脆弱）。
- 不适合需要跨层级协作的任务：如果一线 Agent 需要直接和顶层沟通，层级反而成了障碍。
💻 Python 骨架（递归式 LangGraph）
层级型的实现精髓是 "递归"：每个经理节点内部又是一个完整的中心化子图。
▼python复制代码from typing import TypedDict from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI class State(TypedDict): task: str sub_industries: list[str] sub_reports: dict[str, str] # 行业 → 报告 final_report: str model = ChatOpenAI(model="gpt-4o") def top_orchestrator(state: State) -> dict: """顶层：把任务分解为子行业""" resp = model.invoke(f"把以下任务分解为3个子行业方向：{state['task']}") # 简化：实际用结构化输出解析 return {"sub_industries": ["白酒", "食品", "日化"]} def make_industry_manager(industry: str): """工厂函数：为每个子行业创建一个经理节点""" def manager(state: State) -> dict: # 经理内部再分解为公司级子任务（这里简化为直接生成） resp = model.invoke(f"你是{industry}行业研究经理，分析该行业3家代表公司，汇总成子报告") sub_reports = state.get("sub_reports", {}) sub_reports[industry] = resp.content return {"sub_reports": sub_reports} return manager def final_summary(state: State) -> dict: resp = model.invoke(f"综合各子行业报告，生成最终报告：{state['sub_reports']}") return {"final_report": resp.content} graph = StateGraph(State) graph.add_node("top_orchestrator", top_orchestrator) graph.add_node("final_summary", final_summary) # 为每个子行业创建经理节点 industries = ["白酒", "食品", "日化"] for ind in industries: graph.add_node(f"manager_{ind}", make_industry_manager(ind)) graph.add_edge("top_orchestrator", f"manager_{ind}") # 并行分派 graph.add_edge(f"manager_{ind}", "final_summary") # 汇聚 graph.add_edge(START, "top_orchestrator") graph.add_edge("final_summary", END) app = graph.compile()
层级型 = 嵌套的中心化。make_industry_manager 是个工厂函数，每个经理节点内部还可以再嵌套一个完整的中心化子图（经理 → 公司 Agent → 汇总）。这就是"递归"——所有层级用同一套模式。
🔧 Spring AI Alibaba 对照
▼java复制代码@Service public class HierarchicalReportService { private final ChatClient chatClient; public String generate(String task) { // 1. 顶层：分解为子行业 List<String> subIndustries = List.of("白酒", "食品", "日化"); // 2. 中层：每个子行业一个经理，并行执行 List<CompletableFuture<String>> managerFutures = subIndustries.stream() .map(industry -> CompletableFuture.supplyAsync(() -> // 经理内部可以再嵌套（这里简化为直接调用） chatClient.prompt() .system("你是" + industry + "行业研究经理，分析3家代表公司并汇总") .user(task).call().content() )) .toList(); CompletableFuture.allOf(managerFutures.toArray(new CompletableFuture[0])).join(); List<String> subReports = managerFutures.stream() .map(CompletableFuture::join).toList(); // 3. 顶层：汇总 return chatClient.prompt() .system("综合各子行业报告生成最终报告") .user(String.join("\n", subReports)) .call().content(); } }
对照要点：Java 侧用 CompletableFuture 嵌套实现层级，本质和 LangGraph 的图嵌套一样——每一层都是"fan-out + fan-in"模式。如果层级很深，Spring 侧建议把每层封装成独立的 @Service，通过依赖注入组合，避免方法嵌套太深。
八、竞争型（Competition / Adversarial）：红蓝对抗
📖 概念
竞争型架构让 Agent 之间形成 对抗关系 而非协作关系：一个 Agent 负责生成（攻击），另一个 Agent 负责挑错（防御），通过多轮对抗提升最终输出质量。这种架构直接借鉴了 GAN（生成对抗网络）的思想，也类似网络安全中的"红蓝对抗"演练。
和"辩论型"的区别在于：辩论是"平等的多方各持观点互相质询"，目标是达成共识；竞争是"明确的攻防角色，一方生成一方挑错"，目标是让生成方在压力下产出更鲁棒的结果。辩论追求"共识"，竞争追求"进化"。
最典型的应用是 代码生成 + 安全审计：Generator Agent 生成代码，Validator Agent 专门找安全漏洞，Generator 根据漏洞报告修复，Validator 再找新漏洞……多轮迭代后，代码的安全性显著提升。这就是 OpenAI 的 "Critique and Revise" 范式，也是 Constitutional AI 的核心思想——用一组"宪法规则"作为 Validator 的判断标准，让模型自我修正。
另一个重要变体是 Red Teaming（红队测试）：一个 Agent 扮演攻击者，试图让另一个 Agent 输出有害内容，用于发现和修补 LLM 的安全漏洞。这是大模型安全评估的标准做法。
▼mermaid复制代码flowchart LR TASK[任务: 生成安全代码] --> GEN[Generator<br/>生成代码] GEN --> VAL[Validator<br/>挑错/安全审计] VAL -->|发现漏洞: SQL注入| GEN GEN -->|修复后| VAL VAL -->|还有漏洞: XSS| GEN GEN -->|再次修复| VAL VAL -->|无漏洞 通过| FINAL[最终代码] VAL -.->|记录漏洞报告| REPORT[漏洞修复日志] style GEN fill:#e8f5e9 style VAL fill:#ffebee style FINAL fill:#e1f5ff
🎯 适用场景
- 代码生成 + 审查：Generator 写代码，Validator 做 Code Review，多轮迭代。
- 安全场景：代码安全审计、Prompt 注入防御测试、越狱攻击演练。
- 内容质量提升：写作 Agent 生成，编辑 Agent 挑错，多轮打磨。
- 合规检查：生成 Agent 输出营销文案，合规 Agent 检查是否违规。
- 事实核查：生成 Agent 写新闻摘要，Fact-checker Agent 核实每个事实。
⚠️ 局限性
- 可能过度优化：Validator 太严格会导致 Generator 过度修改，失去原始意图。
- 收敛标准难定：什么时候算"没有漏洞了"？需要明确的通过标准。
- 成本高：每轮都是两次 LLM 调用，N 轮就是 2N 次。
- Validator 能力是瓶颈：如果 Validator 本身能力不足，发现不了真问题，对抗就失去意义。
💻 Python 骨架（LangGraph 循环 + 退出条件）
▼python复制代码from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI class State(TypedDict): task: str code: str issues: list[str] round: int final_code: str MAX_ROUNDS = 5 model = ChatOpenAI(model="gpt-4o") def generator(state: State) -> dict: """生成代码，如果有上一轮的问题则修复""" issues = state.get("issues", []) if not issues: # 首次生成 resp = model.invoke(f"你是高级开发工程师，完成任务：{state['task']}，生成代码") else: # 根据问题修复 resp = model.invoke( f"你之前的代码有以下问题：{issues}。请修复。原代码：\n{state['code']}" ) return {"code": resp.content, "round": state.get("round", 0) + 1} def validator(state: State) -> dict: """挑错：审查代码的安全性和质量问题""" resp = model.invoke( f"你是严格的安全审计员。审查以下代码的安全漏洞和质量问题：\n{state['code']}\n" f"如果没有问题，返回 'PASS'；否则列出问题。" ) content = resp.content if "PASS" in content: return {"issues": []} # 无问题 return {"issues": [content]} def should_continue(state: State) -> Literal["generator", "end"]: """退出条件：无问题 或 达到最大轮数""" if not state.get("issues"): # Validator 通过 return "end" if state["round"] >= MAX_ROUNDS: # 达到上限 return "end" return "generator" # 继续修复 def finalize(state: State) -> dict: return {"final_code": state["code"]} graph = StateGraph(State) graph.add_node("generator", generator) graph.add_node("validator", validator) graph.add_node("finalize", finalize) graph.add_edge(START, "generator") graph.add_edge("generator", "validator") graph.add_conditional_edges("validator", should_continue, { "generator": "generator", # 有问题，继续修复 "end": "finalize", # 通过或达到上限 }) graph.add_edge("finalize", END) app = graph.compile()
竞争型的图模式：generator → validator → 判断 → 回到 generator 或 结束。和辩论型的环结构几乎一样，区别在于：辩论是平等的多方，竞争是明确的攻防角色；辩论的退出是"轮数到"，竞争的退出是"Validator 通过"。
🔧 Spring AI Alibaba 对照
▼java复制代码@Service public class AdversarialCodeService { private final ChatClient chatClient; private static final int MAX_ROUNDS = 5; public String generate(String task) { String code = ""; List<String> issues = List.of(); int round = 0; while (round < MAX_ROUNDS) { // 1. Generator final List<String> finalIssues = issues; final String prevCode = code; code = chatClient.prompt() .system(issues.isEmpty() ? "你是高级开发工程师" : "你是高级开发工程师。修复以下问题：" + finalIssues + "。原代码：" + prevCode) .user(task).call().content(); round++; // 2. Validator String review = chatClient.prompt() .system("你是安全审计员，审查代码。无问题返回 PASS，否则列出问题") .user(code).call().content(); if (review.contains("PASS")) break; // 通过 issues = List.of(review); } return code; } }
对照要点：Spring 侧用 while 循环 + break 实现攻防迭代，比 LangGraph 的图模式更直白。如果你用 Spring AI Alibaba 的 Advisor 机制，可以把 Validator 实现成一个 Advisor，在每次 ChatClient 调用后自动拦截审查——这是 Spring 特有的优雅实现方式。
九、市场/拍卖型（Market / Auction）：招标竞标
📖 概念
市场型架构把经济学中的 市场机制 引入多 Agent 系统：任务作为"标的物"，Agent 作为"竞标者"，通过拍卖/竞价机制决定谁来做。每个 Agent 根据自己的能力、当前负载、预期收益给出一个"标书"（包括能力匹配度、预计时间、质量承诺），拍卖者（Auctioneer）根据标书选择最优 Agent 执行。
这种架构解决的是 "任务该分派给谁"的决策问题，本质是一种去中心化的资源分配机制。和中心化协调者的区别在于：协调者是"指定"分派（主管说谁干就谁干），市场型是"竞争"分派（谁觉得自己合适就投标，系统择优）。
市场型的优势在于 自适应性：当某个 Agent 负载高或能力不足时，它会主动降低竞标意愿（标书质量下降或弃标），任务自然流向更合适的 Agent。这比协调者硬编码的"路由规则"灵活得多。劣势在于 每次任务都要走一遍竞标流程，延迟较高，且 Agent 的自我评估能力直接决定系统效果。
常见的拍卖机制有四种：
- 英式拍卖（加价）：Agent 轮流出价，最高者得。适合"成本最低"目标。
- 荷兰式拍卖（降价）：拍卖者从高价往下降，第一个接受的得。适合紧急任务。
- 密封投标（Sealed-bid）：所有 Agent 同时提交标书，最优者得。最常用。
- Vickrey 拍卖（二价）：密封投标，但中标者付第二高价。鼓励诚实报价。
▼mermaid复制代码flowchart TD TASK[新任务到达] --> AUC[Auctioneer 拍卖者] AUC -->|广播任务| A1[Agent 1<br/>能力: 金融 0.9<br/>负载: 高] AUC -->|广播任务| A2[Agent 2<br/>能力: 金融 0.7<br/>负载: 低] AUC -->|广播任务| A3[Agent 3<br/>能力: 金融 0.3<br/>负载: 低] A1 -->|标书: 质量0.85 时间慢| AUC A2 -->|标书: 质量0.75 时间快| AUC A3 -->|弃标: 能力不足| AUC AUC -->|择优: Agent 2| EXEC[Agent 2 执行] EXEC --> RESULT[结果] RESULT --> AUC style AUC fill:#fce4ec,stroke:#c2185b,stroke-width:3px style EXEC fill:#e8f5e9
🎯 适用场景
- Agent 能力异构且动态变化：不同 Agent 擅长不同领域，且负载实时变化，硬编码路由不可行。
- 资源调度：云计算任务调度、客服路由（哪个客服 Agent 现在空闲且擅长这个问题）。
- 众包式任务分发：多个 Agent 都是"自由职业者"，通过竞标获取任务。
- 成本敏感场景：不同 Agent 调用成本不同（GPT-4o vs GPT-4o-mini），通过竞标让性价比最高的 Agent 接手。
⚠️ 局限性
- 延迟高：每次任务都要走竞标流程，比直接分派慢。
- Agent 自评能力是关键：如果 Agent 高估或低估自己，分派效果差。
- 可能"寡头垄断"：强 Agent 总是中标，弱 Agent 永远得不到锻炼机会。
- 实现复杂：需要设计标书格式、评标算法、结算机制。
💻 Python 骨架
▼python复制代码from typing import TypedDict from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI class Bid(TypedDict): agent_id: str capability_score: float # 能力匹配度 0-1 estimated_time: float # 预计耗时 available: bool # 是否竞标 class State(TypedDict): task: str bids: list[Bid] winner: str result: str model = ChatOpenAI(model="gpt-4o") # 三个异构 Agent，各有不同的能力画像 AGENT_PROFILES = { "agent_1": {"finance": 0.9, "tech": 0.3, "load": 0.8}, # 金融强但负载高 "agent_2": {"finance": 0.7, "tech": 0.6, "load": 0.2}, # 均衡且空闲 "agent_3": {"finance": 0.3, "tech": 0.9, "load": 0.3}, # 技术强 } def auctioneer(state: State) -> dict: """拍卖者：广播任务，收集标书，择优""" # 简化：用规则模拟竞标（实际中每个 Agent 用 LLM 评估） bids = [] task_domain = "finance" # 简化：假设任务领域是金融 for agent_id, profile in AGENT_PROFILES.items(): capability = profile[task_domain] load = profile["load"] # 能力 × (1-负载) 作为综合评分 score = capability * (1 - load) bids.append({ "agent_id": agent_id, "capability_score": capability, "estimated_time": load * 10, # 负载越高越慢 "available": score > 0.2, }) # 择优：选综合评分最高的 valid_bids = [b for b in bids if b["available"]] winner = max(valid_bids, key=lambda b: b["capability_score"] * (1 - b["estimated_time"]/20)) return {"bids": bids, "winner": winner["agent_id"]} def execute_task(state: State) -> dict: """中标 Agent 执行任务""" winner = state["winner"] profile = AGENT_PROFILES[winner] resp = model.invoke(f"你是能力评分{profile}的分析师，完成任务：{state['task']}") return {"result": resp.content} graph = StateGraph(State) graph.add_node("auctioneer", auctioneer) graph.add_node("execute", execute_task) graph.add_edge(START, "auctioneer") graph.add_edge("auctioneer", "execute") graph.add_edge("execute", END) app = graph.compile()
🔧 Spring AI Alibaba 对照
▼java复制代码@Service public class AuctionTaskService { private final ChatClient chatClient; private final Map<String, Map<String, Double>> agentProfiles = Map.of( "agent_1", Map.of("finance", 0.9, "tech", 0.3, "load", 0.8), "agent_2", Map.of("finance", 0.7, "tech", 0.6, "load", 0.2), "agent_3", Map.of("finance", 0.3, "tech", 0.9, "load", 0.3) ); public String dispatch(String task, String domain) { // 1. 竞标：计算每个 Agent 的综合评分 Map<String, Double> scores = new HashMap<>(); agentProfiles.forEach((id, profile) -> { double capability = profile.getOrDefault(domain, 0.0); double load = profile.get("load"); scores.put(id, capability * (1 - load)); }); // 2. 择优 String winner = scores.entrySet().stream() .filter(e -> e.getValue() > 0.2) .max(Map.Entry.comparingByValue()) .map(Map.Entry::getKey) .orElseThrow(); // 3. 中标者执行 return chatClient.prompt() .system("你是 Agent " + winner + "，能力画像：" + agentProfiles.get(winner)) .user(task).call().content(); } }
对照要点：市场型架构在两种语言里都没有现成的框架支持，需要自己实现竞标逻辑。Spring 侧的优势是可以把 Agent 能力画像存到数据库，通过 @Cacheable 缓存，配合 @Async 实现异步竞标，更适合生产环境。
十、黑板型（Blackboard）：共享状态空间
📖 概念
黑板架构是最古老的 AI 架构之一，源自 1980 年代的 HEARSAY-II 语音识别系统。它的核心是 一个共享的"黑板"（共享状态空间）+ 多个独立的"知识源"（Knowledge Source, KS）：所有 Agent 都盯着黑板，当黑板上的内容触发自己的专长时，就主动修改黑板。没有中心协调者，Agent 之间也不直接通信，所有协作都通过读写黑板完成。
黑板架构和混合式的共享 State 看起来相似，但有关键区别：混合式仍然有协调者管控流程，黑板型 完全没有流程控制，靠的是"事件触发"——每个 Agent 注册自己关心的事件条件，条件满足就执行。这类似发布-订阅（Pub-Sub）模式，黑板是消息总线，Agent 是订阅者。
黑板型的典型工作流：初始问题写入黑板 → 专家 A 发现自己能贡献，写入部分解 → 专家 B 看到 A 的部分解，触发自己的推理，写入新信息 → 专家 C 综合 A 和 B 的结果，得出最终答案。整个过程是 数据驱动 的，而非流程驱动的。
在现代 LLM 系统中，黑板型常用于 知识密集型推理：多个专业 Agent（医学、法律、财务）各自从不同角度分析同一个案例，把发现写到黑板上，最终由一个"综合 Agent"或人工整合所有发现。这也类似人类专家会诊——每个医生看同一份病历，各自写诊断意见，最后主任医师综合。
▼mermaid复制代码flowchart TD INIT[初始问题写入黑板] --> BB((黑板<br/>共享状态空间)) KS1[知识源1: 医学专家] -->|观察触发| BB KS2[知识源2: 法律专家] -->|观察触发| BB KS3[知识源3: 财务专家] -->|观察触发| BB KS4[知识源4: 综合专家] -->|观察触发| BB BB -->|写入诊断意见| KS1 BB -->|写入法律分析| KS2 BB -->|写入财务影响| KS3 BB -->|读取所有意见<br/>写入最终结论| KS4 BB --> FINAL[最终结论] style BB fill:#fff9c4,stroke:#f57f17,stroke-width:3px style KS1 fill:#e1f5ff style KS2 fill:#e1f5ff style KS3 fill:#e1f5ff style KS4 fill:#e8f5e9
🎯 适用场景
- 知识密集型推理：医疗会诊、法律案件分析、故障诊断，多个领域专家各自贡献。
- 渐进式问题求解：问题无法一次性分解，需要随着信息积累逐步推进。
- 异步协作：Agent 可以异步地读写黑板，不需要同步等待。
- 事件驱动系统：如舆情监控，多个 Agent 监控不同信号源，发现异常写入黑板触发响应。
- 传统 AI 遗留系统：HEARSAY 式的专家系统升级，保留黑板架构但用 LLM 替换规则引擎。
⚠️ 局限性
- 无流程控制，难以预测：什么时候哪个 Agent 会被触发？执行顺序不确定，调试困难。
- 冲突处理：多个 Agent 同时写黑板可能冲突，需要锁机制或冲突解决策略。
- 终止条件难定：什么时候算"问题解决了"？需要专门的"终止检测 Agent"或人工判断。
- 不适合线性任务：有明确步骤的任务，黑板型反而比中心化低效。
💻 Python 骨架
▼python复制代码from typing import TypedDict from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI class Blackboard(TypedDict): case: str # 初始案例 medical_opinion: str # 医学意见 legal_opinion: str # 法律意见 financial_opinion: str # 财务意见 final_conclusion: str model = ChatOpenAI(model="gpt-4o") def medical_expert(state: Blackboard) -> dict: """医学专家：读案例，写医学意见""" resp = model.invoke(f"你是医学专家，分析案例：{state['case']}") return {"medical_opinion": resp.content} def legal_expert(state: Blackboard) -> dict: """法律专家：读案例和医学意见，写法律分析""" context = f"案例：{state['case']}\n医学意见：{state.get('medical_opinion','')}" resp = model.invoke(f"你是法律专家，基于以下信息分析法律责任：\n{context}") return {"legal_opinion": resp.content} def financial_expert(state: Blackboard) -> dict: """财务专家：读案例，评估财务影响""" resp = model.invoke(f"你是财务专家，评估案例的财务影响：{state['case']}") return {"financial_opinion": resp.content} def synthesizer(state: Blackboard) -> dict: """综合专家：读所有意见，写最终结论""" resp = model.invoke( f"综合以下专家意见，得出最终结论：\n" f"医学：{state['medical_opinion']}\n" f"法律：{state['legal_opinion']}\n" f"财务：{state['financial_opinion']}" ) return {"final_conclusion": resp.content} # 黑板型：所有专家读写同一个 Blackboard graph = StateGraph(Blackboard) graph.add_node("medical", medical_expert) graph.add_node("legal", legal_expert) graph.add_node("financial", financial_expert) graph.add_node("synthesizer", synthesizer) # 三个专家可以并行（都只读 case），synthesizer 等所有专家完成 graph.add_edge(START, "medical") graph.add_edge(START, "legal") graph.add_edge(START, "financial") graph.add_edge("medical", "synthesizer") graph.add_edge("legal", "synthesizer") graph.add_edge("financial", "synthesizer") graph.add_edge("synthesizer", END) app = graph.compile()
黑板型 vs 独立并行型的区别：看起来都是"fan-out + fan-in"，但黑板型的专家可以读取其他专家的输出（法律专家能看医学意见），独立并行型的 Agent 互不可见。在 LangGraph 中，这个区别体现在节点对 State 字段的读取上——如果法律节点读了 medical_opinion 字段，就是黑板型；如果只读 case，就是独立并行型。
🔧 Spring AI Alibaba 对照
▼java复制代码// 黑板 = 共享上下文对象 public class Blackboard { private final Map<String, String> data = new ConcurrentHashMap<>(); public void put(String key, String value) { data.put(key, value); } public String get(String key) { return data.get(key); } } @Service public class BlackboardConsultService { private final ChatClient chatClient; public String consult(String caseText) { Blackboard bb = new Blackboard(); bb.put("case", caseText); // 三个专家并行读写黑板 CompletableFuture<Void> medical = CompletableFuture.runAsync(() -> bb.put("medical_opinion", chatClient.prompt() .system("你是医学专家").user(caseText).call().content()) ); CompletableFuture<Void> financial = CompletableFuture.runAsync(() -> bb.put("financial_opinion", chatClient.prompt() .system("你是财务专家").user(caseText).call().content()) ); // 法律专家等医学专家完成（体现"读其他专家意见"的黑板特性） CompletableFuture<Void> legal = medical.thenRun(() -> bb.put("legal_opinion", chatClient.prompt() .system("你是法律专家。医学意见：" + bb.get("medical_opinion")) .user(caseText).call().content()) ); CompletableFuture.allOf(financial, legal).join(); // 综合专家 return chatClient.prompt() .system("综合医学、法律、财务意见得出结论") .user("医学:" + bb.get("medical_opinion") + " 法律:" + bb.get("legal_opinion") + " 财务:" + bb.get("financial_opinion")) .call().content(); } }
对照要点：Spring 侧用 CompletableFuture.thenRun 实现"法律专家等医学专家完成"的依赖关系，这正是黑板型"Agent 根据黑板状态决定是否行动"的特性。如果你想做更纯粹的"事件驱动"黑板，可以用 Spring 的 ApplicationEventPublisher：医学专家完成后发布 MedicalOpinionReadyEvent，法律专家作为监听器响应。
十一、架构选型决策指南
讲完 10 种架构，你可能会问：那我该选哪个？ 这一节给你两个工具——一张对比矩阵和一棵决策树。
📊 对比矩阵
| 架构 | 协作复杂度 | 实现难度 | 成本倍数 | 容错性 | 适用任务类型 | 典型框架支持 |
|---|---|---|---|---|---|---|
| SAS 单智能体 | 无 | ⭐ | 1x | 低（单点） | 简单线性任务 | LangChain/Spring AI 原生 |
| 独立并行型 | 无 | ⭐⭐ | Nx | 中（冗余） | 创意生成/高风险验证 | LangGraph Map-Reduce |
| 中心化 Orchestrator | 中 | ⭐⭐⭐ | 2-5x | 中（看协调者） | 复合任务分工 | LangGraph/LangChain Supervisor |
| 去中心化辩论 | 高 | ⭐⭐⭐⭐ | NxM | 高（多视角） | 开放性问题/推理 | LangGraph 循环图 |
| 混合式 Hybrid | 极高 | ⭐⭐⭐⭐⭐ | 3-6x | 中高 | 子任务强耦合 | LangGraph 共享 State |
| 层级型 Hierarchical | 高 | ⭐⭐⭐⭐ | 3-5x | 中（顶层脆弱） | 大规模系统 | 嵌套 LangGraph |
| 竞争型 Adversarial | 中 | ⭐⭐⭐ | 2Nx | 高（攻防迭代） | 代码/内容质量提升 | LangGraph 循环+退出 |
| 市场/拍卖型 | 中 | ⭐⭐⭐⭐ | 2-3x | 高（自适应） | 异构Agent调度 | 需自研 |
| 黑板型 Blackboard | 高 | ⭐⭐⭐ | 3-5x | 高（解耦） | 知识密集型推理 | LangGraph 共享 State |
读表说明：
- 成本倍数：相对 SAS 的 LLM 调用次数倍数（N=Agent数，M=辩论轮数）。
- 容错性：单个 Agent 出错时，系统整体是否还能产出可用结果。
- 典型框架支持：指该架构在主流框架中是否有现成模式，"需自研"意味着框架只提供基础能力，架构逻辑要自己实现。
🌳 选型决策树
▼mermaid复制代码flowchart TD START[开始选型] --> Q1{任务能否由单个<br/>Agent 线性完成?} Q1 -->|能| SAS[✅ SAS 单智能体] Q1 -->|不能| Q2{需要多个 Agent<br/>做同一件事再聚合?} Q2 -->|是| PARALLEL[✅ 独立并行型] Q2 -->|否| Q3{Agent 数量 > 10?} Q3 -->|是| Q4{任务可分层分解?} Q4 -->|是| HIER[✅ 层级型] Q4 -->|否| MARKET[✅ 市场/拍卖型] Q3 -->|否| Q5{Agent 之间是协作<br/>还是对抗关系?} Q5 -->|对抗| Q6{一方生成一方挑错?} Q6 -->|是| COMP[✅ 竞争型] Q6 -->|否 辩论| DEBATE[✅ 去中心化辩论] Q5 -->|协作| Q7{需要中心协调者吗?} Q7 -->|不需要| Q8{靠共享状态协作<br/>还是点对点通信?} Q8 -->|共享状态| BB[✅ 黑板型] Q8 -->|点对点| DEBATE2[✅ 去中心化辩论] Q7 -->|需要| Q9{子任务之间需要<br/>直接通信对齐?} Q9 -->|是| HYBRID[✅ 混合式] Q9 -->|否| CENT[✅ 中心化 Orchestrator] style SAS fill:#e1f5ff style PARALLEL fill:#e8f5e9 style CENT fill:#fff3e0 style DEBATE fill:#f3e5f5 style HYBRID fill:#e8f5e9 style HIER fill:#ffe0b2 style COMP fill:#ffebee style MARKET fill:#fce4ec style BB fill:#fff9c4
决策树使用方法：从顶部开始，按箭头回答每个菱形判断框的问题，走到对应的矩形就是推荐架构。几个补充建议：
- 从最简单的开始：如果你不确定，先试 SAS，跑不通再升级。不要一上来就用混合式，过度工程的代价很高。
- 成本敏感时优先选低复杂度：辩论型和混合式成本高，如果预算有限，优先考虑独立并行型或中心化。
- 生产环境优先选有框架支持的：LangGraph 和 Spring AI 对中心化、独立并行、层级型支持最好；辩论型和混合式需要更多自研。
- 混合现实：实际项目常常是多种架构的组合。比如"中心化协调者 + 内部用辩论型子流程"很常见——协调者分派任务，某个子任务内部用两个 Agent 辩论提升质量。架构不是非此即彼。
🎯 场景速查表
| 业务场景 | 推荐架构 | 理由 |
|---|---|---|
| 客服机器人（单轮问答） | SAS | 任务简单，延迟敏感 |
| 投研报告生成 | 中心化 / 层级 | 复合任务需分工，子任务有依赖 |
| 广告文案生成（多方案择优） | 独立并行 | 需要多样性，最后择优 |
| 代码生成 + 安全审查 | 竞争型 | 生成-审计-修复的迭代 |
| 医疗会诊 | 黑板型 / 辩论型 | 多领域专家各自贡献 |
| 投资决策（多空辩论） | 去中心化辩论 | 需要多角度对抗验证 |
| 大型软件开发（MetaGPT 式） | 层级型 | PM→架构师→工程师天然分层 |
| 多模型路由（GPT-4o/Claude/Gemini） | 市场/拍卖型 | 根据任务特征择优分配 |
| 内容生产流水线（选题→写→配图→审） | 中心化 | 线性流程，协调者管控 |
| 复杂故障诊断 | 黑板型 | 多专家渐进式推理 |
十二、工程落地的关键问题
选对架构只是开始，落地时还有四个绕不开的工程问题。这些问题的解决方案往往比架构选型本身更影响系统效果。
1. Agent 间通信协议设计
多 Agent 系统的核心是"通信"，但通信不是简单地"把字符串传过去"。你需要回答：
- 消息格式：是自然语言（"请帮我分析行业趋势"）还是结构化（
{"action": "analyze", "domain": "industry"}）？自然语言灵活但易误解，结构化精确但需要预定义 schema。LangGraph 的State倾向于结构化，AutoGen 的GroupChat倾向于自然语言。 - 寻址方式：Agent 怎么知道消息发给谁？中心化架构里协调者直接指定，去中心化架构里需要"广播 + 过滤"或"注册表查找"。
- 协议标准化：业界正在推进 A2A（Agent-to-Agent）协议 和 MCP（Model Context Protocol）。A2A 解决 Agent 之间怎么对话，MCP 解决 Agent 和工具之间怎么连接。你的系统设计应该尽量向这两个标准靠拢，避免被锁定在特定框架。
2. 状态管理与容错
多 Agent 系统的状态管理比单 Agent 复杂得多。关键问题：
- 状态持久化：LangGraph 的
State默认在内存里，进程挂了就丢了。生产环境必须接Checkpointer（如 SQLite/PostgreSQL 后端）做持久化，支持崩溃恢复和断点续跑。 - 幂等性：网络抖动导致重试时，Agent 的执行必须幂等——同样的输入执行两次结果一样。否则会产生重复副作用（如重复下单）。
- 超时与降级：某个 Agent 卡住了怎么办？需要设置超时，超时后走降级策略（用简单模型兜底、或跳过该步骤）。Spring 侧用
@Retryable+@Fallback，LangGraph 侧用timeout参数 + 条件边。 - 事务性：涉及外部副作用的操作（如发邮件、写数据库），多个 Agent 的操作需要事务性保证。这在分布式 Agent 系统里是个硬问题，目前业界还在探索。
3. 成本与延迟控制
多 Agent 系统的 LLM 调用次数是 SAS 的几倍甚至几十倍，成本和延迟很容易失控：
- 模型分级：不是每个 Agent 都需要 GPT-4o。协调者用强模型（分解任务需要推理能力），专业 Agent 用中等模型，简单校验用轻量模型。LangGraph 支持每个节点指定不同 model，Spring AI 支持多个
ChatModelBean。 - 缓存：相同输入的 Agent 调用结果可以缓存。LangChain 的
langchain.cache、Spring 的@Cacheable都能用。 - 早停：辩论型和竞争型要设置最大轮数，避免无限循环。更重要的是设置"质量阈值"——达到阈值就提前停止。
- 并行化：独立子任务尽量并行，能用
asyncio.gather/CompletableFuture就别串行。这是免费的延迟优化。
4. 终止条件设计
"什么时候算任务完成了"这个问题看似简单，在多 Agent 系统里极其棘手：
- 固定步骤：最简单，但可能过早结束（质量不够）或浪费（已经够了还在跑）。
- 质量阈值：用另一个 LLM 或规则评估输出质量，达到阈值就停。问题是"质量"难以客观定义。
- 收敛检测：辩论型中，如果连续两轮 Agent 的观点变化很小，说明已收敛，可以停止。
- 人工介入：高风险场景（医疗、法律）设置人工 checkpoint，由人决定是否继续。LangGraph 的
interrupt机制支持这种 human-in-the-loop。
十三、未来趋势：从框架到协议到 OS
最后聊聊多智能体架构的发展方向，帮你判断该把学习精力投在哪里。
1. 从"框架锁定"到"协议标准化"
当前的多 Agent 生态是"框架林立"：LangGraph、AutoGen、CrewAI、MetaGPT、Spring AI Alibaba……每个框架有自己的 Agent 抽象、通信方式、编排语法，互不兼容。这类似早期微服务框架混战，最终被 gRPC + REST 标准化。
A2A 协议（Agent-to-Agent）正在做这件事：定义 Agent 之间的标准通信协议，让不同框架的 Agent 能互相对话。MCP（Model Context Protocol）则标准化了 Agent 和工具的连接方式。未来你可能会看到"LangGraph 的 Agent 通过 A2A 和 Spring AI 的 Agent 协作"——这才是真正的"多智能体生态"。
学习建议：不要过度投入某一个框架的细节，要理解架构本质。框架会迭代、会消亡，但"中心化/去中心化/混合"这些架构模式不会变。你现在学 LangGraph 的 StateGraph，本质是在学"图式编排"这种范式，换到任何框架都通用。
2. 从"静态编排"到"动态涌现"
当前的 MAS 大多是"人预定义编排"——你画好图、定义好边，Agent 按图执行。未来的方向是 "涌现式协作"：只给 Agent 一组目标和通信协议，让它们自己协商分工、动态组建团队。
这类似人类组织从"科层制"向"自组织团队"的演进。早期的探索有 AutoGen 的 GroupChat（Agent 自己决定谁发言）、CAMEL 的角色扮演协商。但这条路还很远——当前 LLM 的"自主协商"能力不足以支撑复杂任务，容易陷入无意义对话。
3. 从"应用"到"基础设施"：Agent OS
更激进的观点认为，多 Agent 系统会成为一种 "操作系统"——Agent OS。就像传统 OS 管理进程、内存、文件系统，Agent OS 管理 Agent 的生命周期、资源调度、权限控制、通信总线。
你已经能看到雏形：LangGraph 的 State 像共享内存，Checkpointer 像虚拟内存换页，interrupt 像系统中断。Spring AI 的 Advisor 链像内核拦截器。未来的"Agent OS"可能提供：Agent 注册中心（进程表）、消息总线（IPC）、权限沙箱（容器隔离）、计费与限额（cgroups）。
学习建议：关注 Agent 系统的"操作系统化"趋势。你现在设计的多 Agent 架构，未来可能跑在某种"Agent OS"上，就像现在 Web 应用跑在 Linux 上。理解 OS 的概念（调度、IPC、隔离）能帮你设计更好的 Agent 系统。
4. 多 Agent 与 RAG 的融合
RAG 解决"Agent 知道什么"，MAS 解决"Agent 怎么协作"。两者融合是必然趋势：
- 多 Agent RAG：不同 Agent 各自维护自己的知识库（领域专家），通过协作整合多源知识。比单一 RAG 的覆盖面更广。
- Agent 化的检索器：检索本身变成一个 Agent 任务——检索 Agent 根据查询动态决定检索策略（用 BM25 还是向量、是否 rerank、是否跨库）。
- 记忆共享：多 Agent 共享一个长期记忆库，一个 Agent 学到的经验能被其他 Agent 复用。这是 LangGraph 的
Store和 Spring AI 的VectorStore正在探索的方向。
结语：架构是手段，不是目的
讲了 10 种架构、4 个工程问题、4 个未来趋势，最后回到一个根本问题：为什么要用多智能体？
答案不是"多 Agent 比 Single Agent 更先进"，而是"任务本身的复杂度超过了单 Agent 的能力边界"。如果你用一个 Agent 就能稳定完成任务，千万别用多 Agent——那是给自己找麻烦。多智能体是"用工程复杂度换任务能力"的权衡：你付出了更高的成本、更复杂的调试、更多的故障点，换来的是并行、专业、容错、对抗验证。
架构是手段，完成任务才是目的。 选型时永远从任务出发，而不是从架构出发。当你发现单 Agent 在某个任务上反复出问题（工具混淆、上下文污染、无法并行），那才是引入多 Agent 的正确时机。引入时，从最简单的架构（独立并行或中心化）开始，跑通了再迭代到更复杂的。
多智能体架构还在快速演进，今天的"最佳实践"明天可能就被新协议、新范式颠覆。但有些东西不会变：分工、协作、对抗、共识——这些是人类社会几千年验证过的组织智慧，只是现在被映射到了 AI 系统上。理解了这些底层逻辑，无论框架怎么变，你都能快速适应。
下一步学习建议：
- 用 LangGraph 实现本文的 5 种核心架构（SAS/并行/中心化/辩论/混合），用同一个"投研报告"任务跑通对比。
- 阅读 LangGraph 官方的 Multi-Agent Tutorial 和 AutoGen 的 GroupChat 文档。
- 关注 A2A 协议和 MCP 规范的进展，尝试用 MCP 连接一个外部工具到你的 Agent 系统。
- 在 Spring AI Alibaba 中用
Graph模块复刻本文的中心化架构，对比和 LangGraph 的开发体验差异。
