多智能体原语模型

多智能体原语模型

本节摘要:每六个月就冒出一个新多智能体框架——AutoGen、CrewAI、LangGraph、OpenAI Swarm、Google ADK、Microsoft Agent Framework——每个都自称「正确的抽象」。别被烧穿。它们底下只有四个原语:Agent、交接(Handoff)、共享状态(Shared State)、编排者(Orchestrator)。这四个原语张成一个四维设计空间,2026 年所有主流框架都是其中的一个点。本节用标准库 Python 从零实现这四个原语,把同一条三 Agent 流水线分别跑过三种编排者,然后把所有主流框架映射到同一组坐标轴上。读完本节,你能在一段话内读懂任何新框架发布——因为它只是在四个旋钮上挑了一组默认值。

学习目标

阅读完本节,你应当能够:

1. 说清多智能体的四个原语(Agent、交接、共享状态、编排者)及其各自的无状态性。
2. 用标准库 Python 从零实现这四个原语与三种编排者(静态、LLM 选择、交接驱动)。
3. 把 AutoGen、LangGraph、CrewAI、OpenAI Swarm/Agents SDK、Microsoft Agent Framework、Google ADK 映射到同一组坐标轴上。
4. 用三个核心问题(路由信任谁、共享状态是全量还是投影、Agent 能否改彼此提示)判断一个框架是否适合给定问题。

一、问题与直觉

每六个月一个新框架。AutoGen(2023)、CrewAI(2024)、LangGraph 与 OpenAI Swarm(2024)、Google ADK(2025/04)、Microsoft Agent Framework RC(2026/02)。每个发布都自称「正确的抽象」。

逐一学你会烧穿。API 长得不同,文档对「什么是 Agent」莫衷一是;一个把共享记忆叫「黑板」,一个叫「消息池」,第三个叫「StateGraph」。你开始怀疑这领域就是原地打转。

不是的。营销话术底下,四个原语是稳定的。学一次,以后每个新框架一段话就读懂。

关键洞察:无状态性

除了共享状态,其他三个原语都是无状态的。Agent 是 (prompt, tools) 的函数;交接是一次函数调用;编排者是个调度器。系统里唯一有状态的东西是共享状态——所有有趣的 bug 都住在这里:记忆投毒(第 15 节)、消息乱序、版本化、写冲突。

隐藏共享状态的框架(Swarm)把问题甩给调用者;集中化的框架(LangGraph checkpoint、AutoGen pool)让它可被检查,却把协调成本挪到了共享状态的实现上。

⚠️ 这就是「协调难点在状态而非数量」的根源:Agent 数再多,只要共享状态干净,系统就稳;Agent 不多但共享状态被污染,系统照样崩。

单个原语剖析

Agent = (system_prompt, tools, model, 可选 name)。无记忆无状态。系统提示和工具相同的两个 Agent 可互换。看起来像「每个 Agent 的私有状态」的东西,实际都在共享状态或交接协议里。

交接 = (from, to, reason, payload)。三种主流实现:

• 函数返回:工具返回下一个 Agent(OpenAI Swarm 模式,路由编进工具 schema)。
• 图边:LangGraph,边是声明式的;LLM 产出一个值,条件选出下一个节点。
• 发言者选择:AutoGen GroupChat,一个选择函数(有时本身是 LLM 调用)读消息池选下一个发言者。

共享状态 = { messages: [], artifacts: {}, context: {} }。至少是消息列表,常含结构化产物、类型化上下文、外部记忆。两种拓扑:全量池(每个 Agent 看到所有消息,简单但难扩展)与投影(Agent 看到角色作用域视图,可扩展但需预先设计 schema)。

编排者 = ({state, last_speaker}) -> next_agent。四种风味:静态(图构建时固定,LangGraph/CrewAI Sequential)、LLM 选择(AutoGen/CrewAI Hierarchical)、交接驱动(Swarm)、队列驱动(worker 从共享队列拉取,无显式下一位)。

一旦看懂原语,框架对比成了短清单

• 编排者信任 LLM 路由(Swarm)还是把路由钉死在代码里(LangGraph)?
• 共享状态是全量历史(GroupChat)还是投影(StateGraph reducer)?
• Agent 能改彼此提示(CrewAI manager)还是只能交接(Swarm)?

三个问题回答 80% 的「哪个框架适合这个问题」。你不再逛「最佳多智能体框架」,而是为你在意的轴做设计。

二、从零实现

code/main.py 用约 150 行标准库 Python 实现四个原语。没有真实 LLM——每个 Agent 是脚本化策略,焦点留在协调结构上。

Agent 与交接

from dataclasses import dataclass, field
from typing import Callable, Optional

@dataclass
class Agent:
    name: str
    system_prompt: str
    tools: list = field(default_factory=list)
    policy: Callable[[str, "SharedState"], tuple[str, Optional["Agent"]] | None] = None

@dataclass
class Handoff:
    from_agent: Agent
    to_agent: Agent
    reason: str
    payload: str
​

共享状态(线程安全消息池)

import threading
class SharedState:
    def __init__(self):
        self._lock = threading.Lock()
        self.messages: list[dict] = []
        self.artifacts: dict = {}
    def append(self, sender, content):
        with self._lock:
            self.messages.append({"from": sender, "content": content})
    def view(self, role=None):
        # role 为 None 返回全量池;否则返回投影视图
        with self._lock:
            return list(self.messages)
​

三种编排者

class StaticOrchestrator:
    """路由钉死在代码里,固定序列。"""
    def __init__(self, agents: list[Agent]):
        self.agents = agents; self.idx = 0
    def next(self, state, last_speaker):
        if self.idx < len(self.agents) - 1:
            self.idx += 1; return self.agents[self.idx]
        return None

class HandoffOrchestrator:
    """当前 Agent 通过 policy 的返回值决定下一个。Swarm 模式。"""
    def next(self, state, last_speaker):
        result = last_speaker.policy(last_speaker.name, state)
        if result is None: return None
        content, nxt = result
        state.append(last_speaker.name, content)
        return nxt

class LLMSelectorOrchestrator:
    """选择函数(此处脚本化,真实里是 LLM)读池选下一个。"""
    def __init__(self, agents, selector):
        self.agents = agents; self.selector = selector
    def next(self, state, last_speaker):
        return self.selector(self.agents, state.view())
​

设计要点:三种编排者跑同一条研究→写作→评审流水线,Agent 与共享状态完全相同,唯一差别是谁来选下一个。交接驱动版若研究员提前判定「够了」,会跳过后两个 Agent——这正是 LLM 路由权衡的缩影:灵活但不确定。

一段话跑通

state = SharedState()
agents = [researcher, writer, reviewer]
orch = StaticOrchestrator(agents)
current = agents[0]
while current is not None:
    content = current.policy(current.name, state)
    state.append(current.name, content)
    current = orch.next(state, current)
print(state.view())
​

三、框架对比:映射到同一组坐标轴

表面差异看起来巨大。底下:同样的四个旋钮。各框架剩余的设计决策都是搭在原语之上的:记忆策略(临时 vs 持久 checkpoint)、安全边界(谁批准交接,人在环)、成本核算(每 Agent token 预算)、可观测性(追踪交接、持久化以便重放)。这些都可在原语上实现,没有一个是新原语。

💡 心法:读新框架先画原语映射图,再读 API 文档。 给新成员发映射图比发 API 文档更有效;框架版本变了 diff 映射图,而非 changelog。

四、可复用产物

outputs/skill-primitive-mapper.md 是一个技能:读任何多智能体代码库或框架文档,返回四原语映射。在新框架发布时跑一遍,得到一段话的理解,再决定要不要深入读文档。

采用新框架前:先为它写下原语映射。写不出来说明文档不全,或它在发明第五个原语(罕见——多半是你没见过的一种共享状态变体)。

五、练习

1. 跑三种编排者:运行 code/main.py 三次,改 Agent 策略,观察编排者选择如何改变哪些 Agent 实际运行。
2. 第四种编排者:实现队列驱动版——Agent 轮询共享状态取活。会发生什么死锁?如何检测?
3. LangGraph 映射:把 LangGraph quickstart 改写成四原语。LangGraph 哪些抽象与原语 1:1,哪些是便利包装?
4. Swarm 人体工学:读 OpenAI Swarm cookbook,识别它把哪个原语做得最顺手,把哪个甩给了调用者。
5. 隐藏共享状态的代价:在表中找一个完全隐藏共享状态的框架。当 Agent 需要在交接间协调而不重读历史时,会坏在哪?

本节要点回顾

1. 只有四个原语:Agent、交接、共享状态、编排者——所有主流框架都是这四个旋钮上的一组默认值。
2. 唯一有状态的是共享状态:其他三个原语都是无状态的函数/调度器;所有有趣的 bug(记忆投毒、乱序、写冲突)都在共享状态里。
3. 交接有三种实现:函数返回(Swarm)、图边(LangGraph)、发言者选择(AutoGen)。
4. 编排者有四种风味:静态、LLM 选择、交接驱动、队列驱动——决定谁下一个发言。
5. 共享状态两种拓扑:全量池(简单难扩展)与投影(可扩展需 schema 设计)。
6. 框架对比三条问:路由信任谁、共享状态全量还是投影、Agent 能否改彼此提示——回答 80% 的适配问题。
7. 剩余决策都是上层:记忆策略、安全边界、成本核算、可观测性都搭在原语之上,非新原语。
8. 读新框架先画映射图:写不出来说明文档不全;团队交接发映射图比发 API 文档更有效。

下一节,我们聚焦最常用的编排模式——监督者/编排者-工人模式,看一个「聪明的中心」如何规划、派发、综合,以及它何时会变成瓶颈。