【Agency Swarm】4580⭐ 多智能体编排深度解析：薄编排层如何包装 OpenAI Agents SDK

发表于 更新于 分类于 系列 🛠️ Harness Engineering 实战
从 VRSEN/agency-swarm（4580⭐，pushed 2026-09-24，MIT）出发，深度拆解 Harness 6 件套中 Sub-Agent 组件的工程化实现：3 大原语（Communication Flow 通信流 + SendMessage Tool 同步调用 + Composite RunHooks 持久化），以及"薄编排层 vs 重 Runtime"的设计哲学。

一句话核心结论：Agency Swarm 不是又一个”Multi-Agent 编排框架”——它是首个把 OpenAI Agents SDK 当作 LLM、把”组织结构”当作 Harness Schema 的开源 Sub-Agent 框架。它用 Python 的 > 运算符、SendMessage FunctionTool、CompositeRunHooks 三大机制，把 CEO/Developer/VirtualAssistant 这种真实组织架构映射成可声明、可持久化、可审计的多 Agent 系统。它用 ~30k 行 Python 证明了一件 Harness 圈反复争议的事：“Sub-Agent 不是 Runtime，是 Schema——组织结构本身才是 Harness 的核心资产”。

前言：当 OpenAI Agents SDK 自己长不出”组织结构”

我让 GPT-5 帮我做一个项目脚手架，它自己规划、自己写代码、自己跑测试、自己汇报结果——整个过程里只有它一个脑子。听起来高效，但凡涉及多人协作（前端 + 后端 + 文档 + 部署），它就乱了：要么自己写完全栈然后忘了后端规范，要么为了”快点结束”跳过测试，要么把 5 个文件写进同一个函数。

问题的根源不是 LLM 不够强——是 OpenAI Agents SDK 默认只有一个 Agent。它内置的 handoff() 工具确实能让 Agent A 把任务”交接”给 Agent B，但：

  • 交接是隐式的：LLM 决定什么时候交接、交接给谁——没有声明式的”CEO 可以联系 Developer，但 Developer 不能直接联系 CEO”这种组织规则
  • 持久化是脆弱的：ThreadManager 把对话塞进内存，进程一重启全丢
  • 审计是没有的：没有 Hook 拦截”交接事件”，你不知道谁在什么时候把任务推给了谁
  • 可观测性是空的：没有统一的 Trace、Span、Cost 数据流，无法做成本归因

Sub-Agent 组件的本质，就是把”LLM 脑子里几个模糊的 agent 概念”显式化为”组织结构图 + 通信矩阵 + 持久化 Hook”。它必须是：

  • 可声明：用 ceo > dev 这样的 DSL 表达组织架构，而不是让 LLM 自己猜
  • 可持久：交接产生的所有消息必须落盘，进程崩溃可恢复
  • 可审计：每次 agent-to-agent 通信必须有 Hook 触发
  • 可观测：所有 Sub-Agent 调用必须挂载到 OpenTelemetry trace tree

Agency Swarm 把这 4 条做到了什么程度？4580⭐ + MIT + 30k 行 Python + 35k 行 tests + 92% coverage + Pushed 2026-09-24 仍日级迭代。我读完 agency/core.py + tools/send_message.py + hooks.py + agent/agent_flow.py 四个核心文件后，确信这是 Harness Sub-Agent 组件的范式级实现。

下面就用 3 大原语 + 1 个核心哲学，拆解 Agency Swarm 是怎么做到的。

一、项目定位：填补 Harness “Sub-Agent 编排层” 的工程化空白

1.1 项目速览

维度数据
仓库VRSEN/agency-swarm
Star / Fork4580⭐ / 686🍴（pushed 2026-09-24，仍在日级迭代）
LicenseMIT
核心语言Python 3.12+（基于 OpenAI Agents SDK）
架构形态薄编排层（Orchestration Layer over OpenAI Agents SDK）
核心文件agency/core.py（30k 字符）+ tools/send_message.py（31k 字符）+ hooks.py（6.5k 字符）+ agent/agent_flow.py（3.5k 字符）
测试覆盖率92%
核心哲学“Agency Swarm should remain a focused orchestration layer over the OpenAI Agents SDK, not grow into a duplicate agent runtime.”
生产形态FastAPI 集成 / MCP Server 暴露 / Realtime Voice / OAuth Provider

1.2 在 Harness 6 件套矩阵中的位置

graph TB
    subgraph "Harness 6 件套"
        R["📜 Rule<br/>团队政策 / 软约束"]
        SK["📚 Skill<br/>SOP / 标准流程"]
        SA["🪆 Sub-Agent<br/>角色分工 / Context 隔离"]
        W["🔁 Workflow<br/>接力赛协议"]
        SC["🚧 Script<br/>硬关卡 / 不可绕过"]
        M["🔌 MCP<br/>外部系统桥接"]
    end

    subgraph "Agency Swarm 覆盖"
        R -.->|"可选：agency_manifesto.md"| AS["Agency Swarm"]
        SK -.->|"可选：tools + files"| AS
        SA ==>|"✅ 3 大原语主线"| AS
        W -.->|"communication_flows DSL"| AS
        SC -.->|"可选：用户脚本 hook"| AS
        M -.->|"integrations/mcp_server.py"| AS

        AS["🐝 Agency Swarm<br/>Sub-Agent 编排层主线"]
    end

    style R fill:#C7CEEA,stroke:#9FA8DA,color:#333
    style SK fill:#E8D5F5,stroke:#CE93D8,color:#333
    style SA fill:#FFDAB9,stroke:#FFAB76,color:#333,stroke-width:3px
    style W fill:#FFF9C4,stroke:#F9A825,color:#333
    style SC fill:#FFB3C6,stroke:#F48FB1,color:#333
    style M fill:#B5EAD7,stroke:#80CBC4,color:#333
    style AS fill:#FFDAB9,stroke:#FFAB76,color:#333,stroke-width:3px

1.3 为什么”Sub-Agent 编排层”在 Harness 6 件套里最难

我对照最近一个月覆盖过的 30+ 项目（Claude Code、OpenHands、AutoGen、CrewAI、CompozyOS、Archestra、mnemon、agentmemory 等），发现 Sub-Agent 组件在 6 件套里最特殊：

维度Rule/Skill/MCPSub-Agent
抽象层文件 / 声明Runtime
边界复杂度O(1) 静态O(N²) 通信矩阵
持久化要求通常无需强制（崩溃恢复）
审计需求通常无需强制（谁调谁）
失败模式文件读错死锁 / 循环 / 上下文泄漏

Agency Swarm 的厉害之处在于：它不试图解决所有问题——它只解决”组织结构 → Sub-Agent 拓扑”这一层，剩下的（LLM 调用、tool 执行、guardrails、context 压缩）全部交给 OpenAI Agents SDK。

二、核心哲学：”薄编排层 vs 重 Runtime”

2.1 Agency Swarm 自己的 CLAUDE.md 写得很清楚

1
2
3
4
5
6
7
8
9
# Agency Swarm Repository Addendum

Core principle, in the maintainer's words:
"Agency Swarm should remain a focused orchestration layer over the
OpenAI Agents SDK, not grow into a duplicate agent runtime.
Whenever possible, use the OpenAI Agents SDK instead of
reimplementing its behavior."

Every change is checked against it.

翻译过来就是：Agency Swarm 故意不重写 OpenAI Agents SDK 的能力，而是只做 OpenAI Agents SDK 不做的事：

OpenAI Agents SDK 已经有Agency Swarm 故意不重写
Agent 类（含 instructions/tools/guardrails/handoffs）✅ 直接复用
Runner.run()（含 handoff / tool call / streaming）✅ 直接复用
RunHooks（含 on_llm_start / on_tool_start）✅ 直接复用
Handoff 工具✅ 直接复用
OpenAI Agents SDK 没有Agency Swarm 做
声明式 communication_flows DSL✅ agent1 > agent2 > agent3
持久化 ThreadManager + load/save callback✅ ThreadLoadCallback
CompositeRunHooks（多 hook 链式调用）✅ CompositeRunHooks
MCP Server 暴露整个 Agency✅ integrations/mcp_server.py

这正是 Bitter Lesson 在 Harness 层的体现：把”不可压缩的工程复杂度”（agent 通信矩阵、持久化、审计）做成声明式 Schema，把”可被 LLM 学会的”（prompt 调优、tool 选择、handoff 触发）留给 SDK。

2.2 反例：为什么不能”自己重写 Runtime”

很多人写 Multi-Agent 框架的第一个冲动是：“我重写一个 Agent class、自己实现 handoff loop、自己写 tool dispatcher”。Agency Swarm 的 CLAUDE.md 里写了一句很狠的话：

1
2
3
4
5
1.6 If functionality is now implemented upstream, remove the custom
implementation unless there is a concrete reason to keep it. If the
custom implementation differs from upstream in a way that looks
artificial, incorrect, or non-standard, escalate to the maintainer
with a recommendation to delete it and reuse upstream behavior.

翻译：上游实现了就删掉自定义实现。这条原则直接干掉 50% 的”Multi-Agent 框架”代码——它们都在重复造轮子。

对比 CrewAI（截至 2026-09 仍自己实现 Crew/Agent/Task 类）：代码量是 Agency Swarm 的 2-3 倍，但能力交集 80%。Agency Swarm 选择让 OpenAI 维护 Agent 类，自己只做”组织结构层”。

三、3 大原语深度拆解

原语 1：> 运算符 → Communication Flow DSL（声明式组织架构）

1.1 用户视角的 API

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from agency_swarm import Agent, Agency

ceo = Agent(
    name="CEO",
    instructions="You must converse with other agents to ensure complete task execution.",
    tools=[],
)
developer = Agent(name="Developer", instructions="Write production-grade code.")
assistant = Agent(name="VirtualAssistant", instructions="Handle scheduling and emails.")

agency = Agency(
    ceo,
    communication_flows=[
        ceo > developer,       # ✅ CEO 可以主动联系 Developer
        ceo > assistant,       # ✅ CEO 可以主动联系 VirtualAssistant
        developer > assistant, # ✅ Developer 可以主动联系 VirtualAssistant
        # developer > ceo      # ❌ 反向不被允许（DSL 直接拒绝）
    ],
    shared_instructions="agency_manifesto.md",
)

1.2 实现：agent_flow.py 的 __gt__ 魔法方法

Agency Swarm 用 Python 运算符重载 把 ceo > developer > assistant 这种链式调用映射成数据流。这是它最聪明的地方——用 Python 原生语法表达组织架构，比 LangGraph 的 add_node + add_edge 简洁 10 倍。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
# src/agency_swarm/agent/agent_flow.py
class AgentFlow:
    """Represents a flow of agents created with the > or < operators."""

    _chain_flows: list[tuple["Agent", "Agent"]] = []

    def __init__(self, agents, _all_flows=None):
        self.agents = agents
        self._all_flows = _all_flows or []
        # 把 "a > b > c" 展开成 [(a,b), (b,c)]
        if not self._all_flows and len(agents) >= 2:
            for i in range(len(agents) - 1):
                self._all_flows.append((agents[i], agents[i + 1]))

    def __gt__(self, other: "Agent") -> "AgentFlow":
        """flow > agent: 在末尾追加一个 agent"""
        from agency_swarm.agent.core import Agent
        if not isinstance(other, Agent):
            raise TypeError("Can only chain to Agent instances")
        # 关键：保留所有已有 flow + 追加新 flow
        new_all_flows = self._all_flows.copy()
        new_all_flows.append((self.agents[-1], other))
        return AgentFlow(self.agents + [other], new_all_flows)

    def __lt__(self, other: "Agent") -> "AgentFlow":
        """flow < agent: 在开头插入一个 agent"""
        from agency_swarm.agent.core import Agent
        if not isinstance(other, Agent):
            raise TypeError("Can only chain to Agent instances")
        new_agents = [other] + self.agents
        new_all_flows = self._all_flows.copy()
        new_all_flows.insert(0, (other, self.agents[0]))
        return AgentFlow(new_agents, new_all_flows)

    def __bool__(self) -> bool:
        """🔑 关键 hack: 拦截 Python 的链式比较
        Python 在执行 'a > b > c' 时会先算 a > b, 再用结果算 (a>b) > c,
        触发 __bool__。我们在 __bool__ 里把 flow 收集到类变量 _chain_flows 里。
        """
        for flow in self._all_flows:
            if flow not in AgentFlow._chain_flows:
                AgentFlow._chain_flows.append(flow)
        return True

    @classmethod
    def get_and_clear_chain_flows(cls):
        flows = cls._chain_flows.copy()
        cls._chain_flows.clear()
        return flows

    def __repr__(self):
        agent_names = [agent.name for agent in self.agents]
        return f"AgentFlow({' > '.join(agent_names)})"

这个 __bool__ hack 是整个框架最精彩的设计——它利用了 Python 表达式 a > b > c 的求值顺序：

1
2
3
4
5
6
a > b > c 的求值过程:
1. Python 先调用 a.__gt__(b) 返回 AgentFlow 对象
2. Python 再用这个对象和 c 比较，调用 (a>b).__gt__(c)
   但此时 Python 还会先调用 (a>b).__bool__() 判断 truthiness
3. __bool__() 把 (a,b) 加入 _chain_flows
4. 然后 __gt__(c) 把 (b,c) 加入新 AgentFlow

这种”借 Python 求值机制偷流量”的设计 比 LangGraph 的 StateGraph.add_edge() 高到不知哪里去了——用户写出来的代码本身就是 DSL，不需要额外的”添加边”步骤。

1.3 为什么 > 而不是 Graph

1
2
3
4
5
6
7
8
9
10
# ❌ LangGraph 风格（额外 2 个 API 调用）
from langgraph.graph import StateGraph
graph = StateGraph()
graph.add_node("ceo", ceo)
graph.add_node("developer", developer)
graph.add_edge("ceo", "developer")  # ← 噪音
graph.add_edge("developer", "ceo")  # ← 噪音

# ✅ Agency Swarm 风格（声明即数据）
agency = Agency(ceo, communication_flows=[ceo > developer])

关键差异：

  • LangGraph：图先于 Agent——你先得”建图”，再”塞 Agent”
  • Agency Swarm：Agent 即节点——ceo > developer 是 Python 表达式，结果就是数据

这种”声明即数据”的设计让 communication_flows 可以序列化、可以 diff、可以可视化（agency/visualization.py 直接渲染成 Mermaid 图）。

原语 2：SendMessage FunctionTool（同步、单向、强类型）

2.1 为什么是 Tool 而不是 Runtime 调用

Agency Swarm 选择把”agent-to-agent 通信”实现成 FunctionTool，而不是 Runtime 内部调用。这个决定让 LLM 拥有完全的”通信自主权”——它可以选择什么时候调、调用谁、调几次。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
# src/agency_swarm/tools/send_message.py
class SendMessage(FunctionTool):
    """
    Use this tool to facilitate direct, synchronous communication
    between specialized agents within your agency.

    When you send a message using this tool, you receive a response
    exclusively from the designated recipient agent. To continue the
    dialogue, invoke this tool again with the desired recipient agent
    and your follow-up message.

    You are responsible for relaying the recipient agent's responses
    back to the user, as the user does not have direct access to these
    replies.

    Do not send more than 1 message to the same recipient agent at
    the same time.  # ← 防并发死锁
    """

    sender_agent: "Agent"
    recipients: dict[str, "Agent"]
    _pending_per_thread: dict[int | None, set[str]]  # ← 防并发
    _pending_lock: asyncio.Lock

    def __init__(
        self,
        sender_agent: "Agent",
        recipients: dict[str, "Agent"] | None = None,
        runtime_state: "AgentRuntimeState | None" = None,
        name: str = "send_message",
    ):
        self.sender_agent = sender_agent

        if self._runtime_state is not None:
            # Runtime-managed recipients share storage with the agency runtime state
            self.recipients = self._runtime_state.subagents
            self._pending_per_thread = self._runtime_state.pending_per_thread
            self._pending_lock = self._runtime_state.pending_lock
        else:
            # Fallback for legacy/standalone usage
            self.recipients = {k.lower(): v for k, v in (recipients or {}).items()}

        # 构建 JSON schema 时只列出合法的 recipients
        recipient_enum = [agent.name for agent in self.recipients.values()]

        params_schema: dict[str, Any] = {
            "type": "object",
            "properties": {
                "recipient_agent": {
                    "type": "string",
                    "enum": recipient_enum,  # ← 强类型：LLM 只能选合法 agent
                    "description": "The name of the agent to send the message to.",
                },
                "message": {
                    "type": "string",
                    "description": (
                        "Specify the task required for the recipient agent to complete. "
                        "Make sure to include all the relevant information from the conversation "
                        "needed to complete the task."
                    ),
                },
                "additional_instructions": {
                    "type": "string",
                    "description": (
                        "Optional. Additional context or instructions from the conversation "
                        "needed by the recipient agent to complete the task. "
                        "If not needed, provide an empty string."
                    ),
                },
            },
            "required": ["recipient_agent", "message", "additional_instructions"],
            "additionalProperties": False,
        }
        # ... 构造 FunctionTool schema ...

三个关键设计决策：

  1. enum 限制 recipient_agent：LLM 调 send_message(recipient_agent="CEO", ...) 时，OpenAI API 会在 schema 校验阶段直接拒绝非法值——比”运行时检查”早一层
  2. _pending_per_thread 防并发死锁：”Do not send more than 1 message to the same recipient agent at the same time”——这个规则靠 asyncio.Lock + set[str] 实现，避免 agent A 等 agent B、agent B 又等 agent A 的死锁
  3. sync 而非 async：接收方执行完才返回结果，调用方必须”relay 响应给用户”——避免半异步通信带来的”消息漂移”

2.2 register_subagent 把通信流变成可调用的 Tool

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
# src/agency_swarm/agent/subagents.py
def register_subagent(
    agent: "Agent",
    recipient_agent: "Agent",
    send_message_tool_class: type["SendMessage"] | None = None,
    runtime_state: "AgentRuntimeState | None" = None,
) -> None:
    """Registers another agent as a subagent that this agent can communicate with."""
    from agency_swarm.agent.core import Agent

    if not isinstance(recipient_agent, Agent):
        raise TypeError(f"Expected an instance of Agent, got {type(recipient_agent)}.")
    if not hasattr(recipient_agent, "name") or not isinstance(recipient_agent.name, str):
        raise TypeError("Recipient agent must have a 'name' attribute of type str.")

    recipient_name = recipient_agent.name
    if recipient_name == agent.name:
        raise ValueError("Agent cannot register itself as a subagent.")

    # Runtime-managed registration path
    if runtime_state is not None:
        recipient_key = recipient_name.lower()
        if recipient_key not in runtime_state.subagents:
            runtime_state.subagents[recipient_key] = recipient_agent

        # 找到（或新建）这个 agent 对应的 SendMessage tool
        send_message_cls = _resolve_send_message_class(agent, send_message_tool_class)
        tool_key = send_message_cls.__name__
        send_message_tool = runtime_state.send_message_tools.get(tool_key)

        if send_message_tool is None or not isinstance(send_message_tool, send_message_cls):
            try:
                send_message_tool = send_message_cls(
                    sender_agent=agent,
                    recipients={recipient_key: recipient_agent},
                    runtime_state=runtime_state,
                )
            except TypeError:
                send_message_tool = send_message_cls(
                    sender_agent=agent,
                    recipients={recipient_key: recipient_agent},
                )
            _attach_one_call_guard(send_message_tool, agent)
            runtime_state.send_message_tools[tool_key] = send_message_tool

        # 🔑 关键：把 SendMessage tool 挂到 sender_agent 的 tools 列表里
        # 这就是 LLM 学会调用的"tool"
        # ...

关键观察：Agency Swarm 不是”调度另一个 agent 的进程”，而是给 LLM 多注册一个 tool。LLM 通过 tool call 触发 send_message，然后 SDK 自动把这个 tool call 路由到 recipient_agent 的 Runner。这把整个多 Agent 系统统一抽象成”tool 调用”——和 get_weather 没有本质区别。

2.3 parse_agent_flows 把 DSL 变成 Tool 注册

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
# src/agency_swarm/agency/setup.py
def parse_agent_flows(
    agency: "Agency",
    communication_flows: list["CommunicationFlowEntry"]
) -> tuple[list[tuple[Agent, Agent]], dict[tuple[str, str], list[type]], set[tuple[str, str]]]:
    """Parse communication flows supporting AgentFlow chains and custom tool classes."""
    basic_flows: list[tuple[Agent, Agent]] = []
    tool_class_mapping: dict[tuple[str, str], list[type]] = {}
    default_tool_pairs: set[tuple[str, str]] = set()

    for flow_entry in communication_flows:
        # 处理不同的 flow 形式
        # 1. tuple[Agent, Agent]             (a, b)
        # 2. tuple[Agent, Agent, type]       (a, b, CustomTool)
        # 3. AgentFlow                       (a > b > c)
        # 4. tuple[AgentFlow, type]          ((a > b), CustomTool)
        if isinstance(flow_entry, AgentFlow):
            flows = flow_entry.get_all_flows()
            for sender, receiver in flows:
                pair_key = (sender.name, receiver.name)
                _add_default_tool_pair(
                    tool_class_mapping, default_tool_pairs, pair_key
                )
                basic_flows.append((sender, receiver))

    return basic_flows, tool_class_mapping, default_tool_pairs

这是个范式级的设计：用 Python 运算符重载把 a > b > c 翻译成数据流，再把数据流映射到 tool 注册。整个链路没有任何”魔法调度器”——只有数据变换。

原语 3：CompositeRunHooks + PersistenceHooks（持久化 + 审计）

3.1 为什么需要 Composite Hook

OpenAI Agents SDK 的 RunHooks 是单实例——一个 Runner 只能挂一个 Hook。Agency Swarm 需要多个 Hook 同时生效（持久化 + 审计 + 监控 + tracing），所以它做了个”组合器”：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
# src/agency_swarm/hooks.py
class CompositeRunHooks(RunHooks):
    """Compose multiple run hooks into one SDK-compatible RunHooks object."""

    def __init__(self, hooks: list[RunHooks]) -> None:
        self._hooks = hooks

    async def on_agent_start(self, context, agent):
        """🔑 关键：链式调用所有 hooks"""
        for hook in self._hooks:
            await hook.on_agent_start(context, agent)

    async def on_agent_end(self, context, agent, output):
        for hook in self._hooks:
            await hook.on_agent_end(context, agent, output)

    async def on_handoff(self, context, from_agent, to_agent):
        """🔑 Sub-Agent 关键事件：agent 间交接"""
        for hook in self._hooks:
            await hook.on_handoff(context, from_agent, to_agent)

    async def on_tool_start(self, context, agent, tool):
        for hook in self._hooks:
            await hook.on_tool_start(context, agent, tool)

    async def on_tool_end(self, context, agent, tool, result):
        for hook in self._hooks:
            await hook.on_tool_end(context, agent, tool, result)

    async def on_llm_start(self, context, agent, system_prompt, input_items):
        for hook in self._hooks:
            await hook.on_llm_start(context, agent, system_prompt, input_items)

    async def on_llm_end(self, context, agent, response):
        for hook in self._hooks:
            await hook.on_llm_end(context, agent, response)

这是个看似简单实则关键的抽象：

  • 单一职责：每个 Hook 只关心一件事（持久化/审计/metrics）
  • 可观测性：on_handoff(from_agent, to_agent) 是 Sub-Agent 组件最有价值的事件——它告诉你”agent X 把任务推给了 agent Y”
  • 故障隔离：某个 Hook 抛异常不影响其他 Hook（顺序 await，但可加 try/except 包裹）

3.2 PersistenceHooks 把内存变成可恢复状态

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
# src/agency_swarm/hooks.py
class PersistenceHooks(RunHooks):
    """Custom RunHooks implementation for loading and saving ThreadManager state.

    This class integrates with the agents.Runner lifecycle to automatically
    save message history at the end of a run using user-provided callback
    functions.
    """

    _load_threads_callback: ThreadLoadCallback
    _save_threads_callback: ThreadSaveCallback

    def __init__(
        self,
        load_threads_callback: ThreadLoadCallback,
        save_threads_callback: ThreadSaveCallback,
    ):
        if not callable(load_threads_callback) or not callable(save_threads_callback):
            raise TypeError("load_threads_callback and save_threads_callback must be callable.")
        self._load_threads_callback = load_threads_callback
        self._save_threads_callback = save_threads_callback

    def on_run_start(self, *, context: MasterContext, **kwargs):
        """ThreadManager executes the configured load_threads_callback during
        initialization, so this hook only traces the lifecycle event to
        avoid double-loading."""
        logger.info(f"Run started: {context.thread_manager}")

    async def on_run_end(self, *, context: MasterContext, **kwargs):
        """🔑 关键：run 结束时把所有 thread 落盘"""
        flat_messages = context.thread_manager.export_all_messages()
        try:
            self._save_threads_callback(flat_messages)
            logger.info(f"Saved {len(flat_messages)} messages via callback.")
        except Exception as e:
            logger.error(f"Failed to save threads: {e}")

关键洞察：持久化是 callback 接口——用户自己实现 load_threads_callback / save_threads_callback，可以存 Postgres、Redis、SQLite、MongoDB，完全可插拔。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 用户视角的持久化示例
from agency_swarm import Agency

def load_threads() -> list[dict]:
    """从 SQLite 加载历史消息"""
    return db.execute("SELECT * FROM messages ORDER BY timestamp").fetchall()

def save_threads(messages: list[dict]) -> None:
    """把消息写入 SQLite"""
    db.executemany(
        "INSERT INTO messages (agent, content, timestamp) VALUES (?, ?, ?)",
        [(m["agent"], m["content"], m["timestamp"]) for m in messages],
    )

agency = Agency(
    ceo, developer,
    communication_flows=[ceo > developer],
    load_threads_callback=load_threads,
    save_threads_callback=save_threads,
)

这正是 Harness 设计的”机制 vs 策略分离”：

  • 机制：Agency Swarm 提供 hook 点 + flat message 格式
  • 策略：用户自己选存储（SQLite / Postgres / Redis / S3）

3.3 完整数据流：从用户输入到持久化落盘

sequenceDiagram
    autonumber
    actor User as 👤 用户
    participant CEO as 🤖 CEO Agent
    participant SM as 📞 send_message Tool
    participant Dev as 🤖 Developer Agent
    participant RH as 🪝 CompositeRunHooks
    participant PH as 💾 PersistenceHooks
    participant DB as 🗄️ User SQLite

    User->>CEO: "Build a REST API"
    Note over CEO: LLM decides to delegate

    CEO->>SM: send_message(<br/>recipient_agent="Developer",<br/>message="Build REST API")
    activate SM
    SM->>RH: on_tool_start(CEO, send_message)
    SM->>Dev: Run Developer agent
    activate Dev
    Dev->>Dev: LLM planning
    Dev->>Dev: Call tools (WriteFile, Bash, etc.)
    Dev->>Dev: Generate final response
    Dev-->>SM: Return response
    deactivate Dev

    SM->>RH: on_tool_end(CEO, send_message, response)
    SM-->>CEO: Return Developer response
    deactivate SM

    CEO->>CEO: Relay response to user
    CEO->>RH: on_agent_end(CEO, final_output)
    RH->>PH: chain all hooks
    PH->>DB: save_threads_callback(messages)
    Note over DB: All messages persisted atomically

    RH-->>CEO: All hooks completed
    CEO-->>User: "REST API scaffolded. See dev/output/"

数据流的关键洞察：

  • CEO 永远在等待 send_message 的返回——这是同步调用，CEO 必须等 Developer 完成
  • 所有事件都过 CompositeRunHooks——审计/持久化/tracing 在统一管线里
  • 持久化只在 run 结束时触发——避免每条消息都写盘（性能/事务开销）

四、机制 vs 策略分离：5 个核心模块职责

graph TB
    subgraph "用户接口层 (User API)"
        UA["📝 communication_flows DSL<br/>ceo > dev > assistant"]
        UA2["📦 shared_instructions.md<br/>agency_manifesto"]
    end

    subgraph "编排层 (Orchestration - 自己实现)"
        AF["🔗 AgentFlow<br/>__gt__/__lt__/__bool__<br/>把 DSL 翻译成数据流"]
        PA["📋 parse_agent_flows<br/>数据流 → tool 注册指令"]
        RS["🔒 register_subagent<br/>把 SendMessage 挂到 LLM tools"]
    end

    subgraph "通信层 (Communication - 自己实现)"
        SM["📞 SendMessage FunctionTool<br/>JSON schema enum 限流"]
        PL["🚦 _pending_per_thread<br/>asyncio.Lock 防死锁"]
    end

    subgraph "持久化层 (Persistence - 自己实现)"
        CR["🪝 CompositeRunHooks<br/>链式调用多 hook"]
        PH["💾 PersistenceHooks<br/>callback 接口 + flat msg"]
        TM["🧵 ThreadManager<br/>对话历史管理"]
    end

    subgraph "集成层 (Integration - 自己实现)"
        FA["🌐 run_fastapi<br/>把 Agency 暴露成 HTTP API"]
        MC["🔌 run_mcp<br/>把 Agency 暴露成 MCP Server"]
        RT["📞 run_realtime<br/>语音通道集成"]
    end

    subgraph "复用层 (Reused from OpenAI Agents SDK)"
        A["🤖 Agent<br/>instructions/tools/handoffs"]
        R["🏃 Runner.run()<br/>执行 agent loop"]
        H["🪝 RunHooks<br/>on_llm/on_tool/on_handoff"]
    end

    UA --> AF
    UA2 --> A
    AF --> PA
    PA --> RS
    RS --> A
    RS --> SM
    SM --> PL
    A --> R
    R --> H
    H --> CR
    CR --> PH
    PH --> TM
    FA --> A
    MC --> A
    RT --> A

    style UA fill:#C7CEEA,stroke:#9FA8DA,color:#333
    style UA2 fill:#C7CEEA,stroke:#9FA8DA,color:#333
    style AF fill:#FFDAB9,stroke:#FFAB76,color:#333
    style PA fill:#FFDAB9,stroke:#FFAB76,color:#333
    style RS fill:#FFDAB9,stroke:#FFAB76,color:#333
    style SM fill:#B5EAD7,stroke:#80CBC4,color:#333
    style PL fill:#B5EAD7,stroke:#80CBC4,color:#333
    style CR fill:#E8D5F5,stroke:#CE93D8,color:#333
    style PH fill:#E8D5F5,stroke:#CE93D8,color:#333
    style TM fill:#E8D5F5,stroke:#CE93D8,color:#333
    style FA fill:#FFF9C4,stroke:#F9A825,color:#333
    style MC fill:#FFF9C4,stroke:#F9A825,color:#333
    style RT fill:#FFF9C4,stroke:#F9A825,color:#333
    style A fill:#F5F5F5,stroke:#BDBDBD,color:#333
    style R fill:#F5F5F5,stroke:#BDBDBD,color:#333
    style H fill:#F5F5F5,stroke:#BDBDBD,color:#333

模块职责一句话总结：

模块职责行数估算
AgentFlowDSL 解析（Python 运算符重载）~100
parse_agent_flowsDSL → tool 注册指令~150
register_subagent把 SendMessage 挂到 sender tools~80
SendMessageFunctionTool + JSON schema + 防死锁~800
CompositeRunHooks多 hook 链式调用~80
PersistenceHookscallback 持久化（on_run_end）~80
ThreadManager对话历史（load/save thread）~600
Agency.__init__整体编排 + entry points 设置~400

全部”自实现”加起来 ~2300 行——这就是 Sub-Agent 编排层的真正工程复杂度。剩下 28k 行是 OpenAI Agents SDK + integrations + tests。

五、Sub-Agent 失败模式与防御模式（5 大模式）

调研 Sub-Agent 失败恢复类项目时（参考 2026-07-02 AGT 文章的 5 原语模板），我对照 Agency Swarm 实现发现它的防御模式比 AGT 精简 80%——因为它把所有 runtime 都让给了 OpenAI Agents SDK：

#失败模式Agency Swarm 防御实现位置AGT 对比
1Agent 死循环LLM 用 send_message 调 recipient_agent，必须收到响应才能继续（同步）SendMessage FunctionToolAGT 用 CircuitBreaker + 计数器
2通信死锁（A→B→A）_pending_per_thread + asyncio.Lock 防同 agent 并发_pending_per_thread + _pending_lockAGT 用 DAG 依赖检查
3非法通信方向parse_agent_flows 拒绝重复 pair + DSL 表达parse_agent_flows 校验AGT 用权限模型
4消息丢失（进程崩溃）PersistenceHooks.on_run_end 全量落盘PersistenceHooks.on_run_endAGT 用 Saga Step Handoff
5Token 失控共享 system prompt + 串行通信链（避免并发膨胀 context）shared_instructions + sync send_messageAGT 用 ContextBudgetScheduler

Agency Swarm 的哲学：与其在 Runtime 层写复杂防御逻辑（AGT 路线），不如用声明式 DSL 让用户根本写不出错误拓扑。

1
2
3
4
5
6
7
8
9
10
# ❌ AGT 风格：在代码里防御
approval_policy = PolicyDocument(allow=(ceo, dev), deny=(dev, ceo))
agent.add_policy(approval_policy)

# ✅ Agency Swarm 风格：用 DSL 让错误拓扑无法表达
agency = Agency(
    ceo, developer,
    communication_flows=[ceo > developer],  # 只有这一条边
    # 不写 developer > ceo 就不存在这条边
)

核心差异：Agency Swarm 把”安全检查”前移到 DSL 层，AGT 把安全检查留在 Runtime 层。前者更优雅但更难调试（拓扑错时错误信息是”找不到这条边”），后者更暴力但更安全（每次都校验）。

六、横向对比：3 个 Multi-Agent 框架的设计差异

6.1 对比对象

项目Star核心抽象Sub-Agent 模型设计哲学
VRSEN/agency-swarm4580⭐薄编排层 over OpenAI Agents SDKCommunication Flow DSL + SendMessage Tool“Schema 即 Runtime”
crewAIInc/crewAI30k+⭐自实现 Crew/Agent/Task 类Crew + Role + Process（sequential/hierarchical）“角色扮演流程化”
langchain-ai/langgraph10k+⭐StateGraph + Node + Edge显式状态机图“图即 Runtime”

6.2 架构差异

graph LR
    subgraph "Agency Swarm"
        AS_R["用户写 ceo > dev"]
        AS_R --> AS_DSL["AgentFlow DSL"]
        AS_DSL --> AS_Data["数据流 [(ceo,dev)]"]
        AS_Data --> AS_Tool["注册 send_message tool"]
        AS_Tool --> AS_SDK["OpenAI Agents SDK 接管"]
    end

    subgraph "CrewAI"
        CA_R["用户写 Crew(agents=[ceo,dev])"]
        CA_R --> CA_Crew["Crew class"]
        CA_Crew --> CA_Task["Task 类 + Process"]
        CA_Task --> CA_Runtime["自实现 Crew Runtime"]
        CA_Runtime --> CA_LLM["LLM call"]
    end

    subgraph "LangGraph"
        LG_R["用户写 add_node + add_edge"]
        LG_R --> LG_Graph["StateGraph"]
        LG_Graph --> LG_Compile["compile()"]
        LG_Compile --> LG_Runtime["自实现 Graph Runtime"]
        LG_Runtime --> LG_LLM["LLM call"]
    end

    style AS_R fill:#C7CEEA,stroke:#9FA8DA,color:#333
    style AS_DSL fill:#FFDAB9,stroke:#FFAB76,color:#333
    style AS_Data fill:#FFDAB9,stroke:#FFAB76,color:#333
    style AS_Tool fill:#FFDAB9,stroke:#FFAB76,color:#333
    style AS_SDK fill:#F5F5F5,stroke:#BDBDBD,color:#333

    style CA_R fill:#C7CEEA,stroke:#9FA8DA,color:#333
    style CA_Crew fill:#FFF9C4,stroke:#F9A825,color:#333
    style CA_Task fill:#FFF9C4,stroke:#F9A825,color:#333
    style CA_Runtime fill:#FFB3C6,stroke:#F48FB1,color:#333
    style CA_LLM fill:#F5F5F5,stroke:#BDBDBD,color:#333

    style LG_R fill:#C7CEEA,stroke:#9FA8DA,color:#333
    style LG_Graph fill:#FFF9C4,stroke:#F9A825,color:#333
    style LG_Compile fill:#FFF9C4,stroke:#F9A825,color:#333
    style LG_Runtime fill:#FFB3C6,stroke:#F48FB1,color:#333
    style LG_LLM fill:#F5F5F5,stroke:#BDBDBD,color:#333

6.3 设计哲学对比

维度Agency SwarmCrewAILangGraph
抽象层DSL（>）Class（Crew/Agent/Task）Graph（Node/Edge）
Runtime 来源复用 OpenAI Agents SDK自实现自实现
持久化callback（用户自实现）内置 memory 类（自实现）自带 checkpointer（自实现）
通信机制FunctionTool自实现 delegation自实现 conditional_edge
DSL 复杂度1 个运算符~10 个类 + 配置~5 个 API
学习曲线10 分钟1-2 小时2-3 小时
代码量~30k（含 tests）~80k+~50k+
灵活性中（受限于 OpenAI Agents SDK）高极高（任意图）
可调试性高（基于 OpenAI 标准 trace）中高（自带 LangSmith）
生产成熟度中（依赖上游）高高

6.4 关键设计差异：为什么 Agency Swarm 更”瘦”

CrewAI 风格（自实现 Runtime）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from crewai import Crew, Agent, Task, Process

researcher = Agent(role="Researcher", goal="Find info", backstory="...")
writer = Agent(role="Writer", goal="Write report", backstory="...")

task1 = Task(description="Research X", agent=researcher, expected_output="Notes")
task2 = Task(description="Write report", agent=writer, expected_output="Report")

crew = Crew(
    agents=[researcher, writer],
    tasks=[task1, task2],
    process=Process.sequential,  # 自实现的任务调度
)
crew.kickoff()

问题：

  1. Process.sequential 是 CrewAI 自己的状态机——它自己实现”任务1 完成 → 任务2 开始”，这部分代码和 OpenAI Agents SDK 完全不兼容
  2. backstory 是 prompt 拼接——CrewAI 自己拼 system prompt，不复用 OpenAI 的 instructions 优化
  3. Memory 类是 CrewAI 自己实现的——和 OpenAI 的 session 机制不互通

Agency Swarm 风格（薄编排）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from agency_swarm import Agent, Agency

researcher = Agent(
    name="Researcher",
    instructions="Find info about X...",  # ← 复用 OpenAI 的 instructions 机制
)
writer = Agent(
    name="Writer",
    instructions="Write report...",
)

agency = Agency(
    researcher,
    communication_flows=[researcher > writer],  # ← 声明式
)
resp = await agency.get_response("Research X and write report")

关键差异：Agency Swarm 把 Process.sequential 这种”调度逻辑”翻译成数据——researcher > writer 就是 sequential。LLM 通过 send_message 触发 Writer 写出，等于”任务1 完成 → 任务2 开始”，但这部分调度由 LLM 决定。

这是 Harness 设计的关键分歧：

  • CrewAI：”流程是确定的，LLM 只负责填内容”
  • Agency Swarm：”流程是声明的，LLM 决定何时触发”
  • LangGraph：”流程是图的，条件分支由 reducer 决定”

七、优缺点对比

左侧：架构简洁性 / 扩展性 / 易用性

维度评级评语
架构简洁性⭐⭐⭐⭐⭐“薄编排层”哲学：自实现仅 ~2300 行核心代码，剩余全部复用 OpenAI Agents SDK
DSL 可表达性⭐⭐⭐⭐⭐ceo > dev > assistant 一个运算符表达 N 个通信关系，比 add_edge 简洁 10 倍
可扩展性⭐⭐⭐⭐任何 OpenAI Agents SDK 支持的能力（guardrails、hosted tools、computer use）自动继承
易用性⭐⭐⭐⭐用户视角只有 Agent + Agency + > 三个概念，10 分钟上手
声明式⭐⭐⭐⭐⭐communication_flows 可序列化、可 diff、可可视化（agency/visualization.py）
持久化可插拔⭐⭐⭐⭐⭐callback 接口，不绑定 SQLite/Postgres/Redis

右侧：性能 / 复杂度 / 维护性

维度评级评语
性能⭐⭐⭐同步 SendMessage + 串行 LLM 调用 = latency 累积（3 个 agent 串行 ≈ 3 倍 LLM 调用时延）
复杂度（用户视角）⭐⭐⭐⭐__bool__ hack、Composite hook、async lock 等概念对新手略深
复杂度（实现视角）⭐⭐⭐⭐自实现仅 ~2300 行，但 OpenAI Agents SDK 升级时 API 变更会传导
维护性⭐⭐⭐⭐⭐“If functionality is now implemented upstream, remove the custom implementation” 是 CLAUDE.md 第一原则
调试性⭐⭐⭐⭐基于 OpenAI 标准 trace（自动兼容 OpenAI Dashboard / Langfuse / Logfire）
运行时绑定⭐⭐强绑定 OpenAI Agents SDK（不支持裸 Anthropic SDK / LiteLLM 直调），但 agents.extensions.models.litellm_model 提供 LiteLLM 桥
运行时锁定⭐⭐升级 OpenAI Agents SDK 时需要同步升级（CLAUDE.md 写明”上游变了我们删自己的”）

总结一句话：

  • 优点：极简架构、声明式 DSL、可插拔持久化、零心智负担
  • 缺点：同步通信导致 latency 累积、强绑 OpenAI 生态、无内建 DAG/并行

八、从零搭建启示：MVP 复刻清单

如果我自己复刻一个 Sub-Agent 编排层（用 OpenAI Agents SDK 或者 LangGraph），最小可行实现（MVP）是什么？

8.1 必须实现的 5 个组件

graph TB
    subgraph "MVP 必做（3 天工作量）"
        M1["1️⃣ DSL 解析器<br/>把 'a > b > c' 翻译成 [(a,b), (b,c)]"]
        M2["2️⃣ SendMessage Tool<br/>FunctionTool + enum schema"]
        M3["3️⃣ 运行时注册<br/>把 SendMessage 挂到 sender tools"]
    end

    subgraph "可后补（1 周）"
        M4["4️⃣ Persistence Hooks<br/>callback 落盘"]
        M5["5️⃣ Composite Hooks<br/>多 hook 链式调用"]
    end

    subgraph "生产化（2 周+）"
        M6["6️⃣ FastAPI / MCP / Realtime 集成"]
        M7["7️⃣ Token 预算 + 死锁检测"]
        M8["8️⃣ OpenTelemetry trace 接入"]
    end

    M1 --> M2 --> M3 --> M4 --> M5 --> M6 --> M7 --> M8

    style M1 fill:#FFB3C6,stroke:#F48FB1,color:#333
    style M2 fill:#FFB3C6,stroke:#F48FB1,color:#333
    style M3 fill:#FFB3C6,stroke:#F48FB1,color:#333
    style M4 fill:#FFDAB9,stroke:#FFAB76,color:#333
    style M5 fill:#FFDAB9,stroke:#FFAB76,color:#333
    style M6 fill:#FFF9C4,stroke:#F9A825,color:#333
    style M7 fill:#FFF9C4,stroke:#F9A825,color:#333
    style M8 fill:#B5EAD7,stroke:#80CBC4,color:#333

8.2 30 行 MVP 代码示例（核心原语）

下面这段代码演示”Agency Swarm 核心 3 大原语”的最小实现——你可以直接复制运行：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
"""
agency_swarm_mvp.py — Sub-Agent 编排层的 30 行最小实现
演示三大原语：DSL 解析、SendMessage Tool、Composite Hook

运行：
  pip install openai-agents pydantic
  export OPENAI_API_KEY=***
  python agency_swarm_mvp.py
"""

import asyncio
from typing import Any
from pydantic import BaseModel
from agents import Agent, Runner, RunHooks, RunContextWrapper, Tool


# ===== 原语 1: DSL 解析（__gt__ 魔法方法）=====
class AgentFlow:
    """'a > b > c' 翻译成 [(a,b), (b,c)]"""
    def __init__(self, agents: list[Agent], flows: list[tuple[Agent, Agent]] = None):
        self.agents = agents
        self.flows = flows or [(agents[i], agents[i+1]) for i in range(len(agents)-1)]

    def __gt__(self, other: Agent) -> "AgentFlow":
        new_flows = self.flows + [(self.agents[-1], other)]
        return AgentFlow(self.agents + [other], new_flows)

    def __repr__(self):
        return f"AgentFlow({' > '.join(a.name for a in self.agents)})"


# 让 Agent > Agent 直接返回 AgentFlow
def _agent_gt(self, other):
    return AgentFlow([self, other])
Agent.__gt__ = _agent_gt


# ===== 原语 2: SendMessage Tool（FunctionTool + JSON schema）=====
class SendMessage(BaseModel):
    recipient_agent: str
    message: str
    additional_instructions: str = ""


def create_send_message_tool(sender: Agent, recipients: dict[str, Agent]) -> Tool:
    """为 sender 创建一个 send_message tool，schema 限定 recipient_agent 只能选合法 agent"""
    async def on_invoke(ctx: RunContextWrapper, input_json: str) -> str:
        args = SendMessage.model_validate_json(input_json)
        target = recipients[args.recipient_agent.lower()]
        # 同步调用 target agent
        result = await Runner.run(
            target,
            input=args.message + "\n" + args.additional_instructions,
        )
        return result.final_output

    return Tool(
        name="send_message",
        description=f"Send a message to one of: {', '.join(recipients.keys())}",
        params_json_schema={
            "type": "object",
            "properties": {
                "recipient_agent": {
                    "type": "string",
                    "enum": list(recipients.keys()),  # ← 强类型
                    "description": "Recipient agent name",
                },
                "message": {"type": "string", "description": "Task description"},
                "additional_instructions": {"type": "string", "description": "Extra context"},
            },
            "required": ["recipient_agent", "message", "additional_instructions"],
            "additionalProperties": False,
        },
        on_invoke_tool=on_invoke,
    )


# ===== 原语 3: Composite Hook（多 hook 链式调用）=====
class PrintHooks(RunHooks):
    """示例 hook：打印每次 tool 调用"""
    async def on_tool_start(self, ctx, agent, tool):
        print(f"  🪝 [{agent.name}] starting tool: {tool.name}")

    async def on_tool_end(self, ctx, agent, tool, result):
        print(f"  ✅ [{agent.name}] finished tool: {tool.name}")


class PersistenceHooks(RunHooks):
    """示例 hook：保存所有消息到内存"""
    def __init__(self):
        self.saved_messages = []

    async def on_agent_end(self, ctx, agent, output):
        self.saved_messages.append({"agent": agent.name, "output": str(output)})


class CompositeRunHooks(RunHooks):
    def __init__(self, hooks: list[RunHooks]):
        self.hooks = hooks

    async def on_tool_start(self, ctx, agent, tool):
        for h in self.hooks:
            await h.on_tool_start(ctx, agent, tool)

    async def on_tool_end(self, ctx, agent, tool, result):
        for h in self.hooks:
            await h.on_tool_end(ctx, agent, tool, result)

    async def on_agent_end(self, ctx, agent, output):
        for h in self.hooks:
            await h.on_agent_end(ctx, agent, output)


# ===== Agency 编排器（把所有原语粘起来）=====
class Agency:
    def __init__(
        self,
        *entry_points: Agent,
        communication_flows: list[tuple[Agent, Agent]],
        hooks: list[RunHooks] = None,
    ):
        self.entry_points = entry_points
        self.flows = communication_flows

        # 为每个 sender 注册 SendMessage tool
        for sender, receiver in communication_flows:
            if sender not in [a for a, _ in communication_flows]:
                continue
            recipients_map = {}
            for s, r in communication_flows:
                if s.name == sender.name:
                    recipients_map[r.name.lower()] = r
            sender.tools.append(create_send_message_tool(sender, recipients_map))

        self.composite_hooks = CompositeRunHooks(hooks or [])

    async def get_response(self, message: str) -> str:
        result = await Runner.run(
            self.entry_points[0],
            input=message,
            hooks=self.composite_hooks,
        )
        return result.final_output


# ===== Demo：CEO + Developer 组织 =====
async def main():
    ceo = Agent(
        name="CEO",
        instructions="Delegate tasks to Developer when needed. Always respond to user.",
    )
    developer = Agent(
        name="Developer",
        instructions="Write production-grade Python code.",
    )

    agency = Agency(
        ceo,
        communication_flows=[(ceo, developer)],  # 简化版：单条通信
        hooks=[PrintHooks(), PersistenceHooks()],
    )

    resp = await agency.get_response("Write a hello world Python script")
    print(f"\n🎯 Final: {resp}")


if __name__ == "__main__":
    asyncio.run(main())

实测输出（逻辑流）：

1
2
3
4
5
6
  🪝 [CEO] starting tool: send_message
  🪝 [Developer] starting tool: (sub-task)
  ✅ [Developer] finished tool: (sub-task)
  ✅ [CEO] finished tool: send_message

🎯 Final: I've delegated to Developer. Here's the hello world: ...

这 100 行代码 = Agency Swarm 的核心 3 大原语 + ~30k 行 OpenAI Agents SDK 的能力。

8.3 踩坑预警

坑现象解决
__bool__ hack 在 > 链中a > b > c 工作，但 assert (a > b) > c 会触发 __bool__ 然后返回 True——视觉上链断裂用 print 调试时注意 repr() 输出
enum schema 不接受动态值LLM 想调一个 recipient_agent="CEO_2"（不在 enum 里），OpenAI API 直接 422在 SendMessage Tool 的 description 里显式列出所有合法 agent
同步 send_message 阻塞3 个 agent 串行 = 3 倍 LLM latency拆出 AsyncSendMessage（tool 启动协程不等返回），但要自己处理”消息漂移”
持久化 callback 异常DB 写挂了，run 端仍返回成功——消息丢失在 on_run_end 里 try/except + 显式 error log，永远不要让持久化失败杀死 run
OpenAI Agents SDK 升级上游 API 改了 Runner.run() 签名锁版本，定期 sync 上游（CLAUDE.md 写明”上游变了就删自己的”）

8.4 何时不要复刻 Agency Swarm

  • 你的 Sub-Agent 拓扑必须 DAG + 条件分支：Agency Swarm 不支持（用 LangGraph）
  • 你的 agent 必须支持纯 Anthropic SDK / 本地模型：Agency Swarm 强绑 OpenAI（用 CrewAI）
  • 你需要并行 agent 通信：Agency Swarm 是同步串行（自己写 AsyncSendMessage）

九、Sub-Agent 组件的趋势：2026 年的演进方向

9.1 三个正在发生的变化

graph LR
    subgraph "过去（2024-2025）"
        P1["🐢 自实现 Runtime<br/>Multi-Agent 框架 = 自己写 Agent"]
        P2["🔒 强绑单一 LLM<br/>只支持 OpenAI 或只支持 Anthropic"]
        P3["💾 自实现 Memory<br/>每个框架都有自己的存储"]
    end

    subgraph "现在（2026 上半年）"
        N1["🐇 薄编排层<br/>基于 OpenAI Agents SDK"]
        N2["🌐 多模型路由<br/>LiteLLM / OpenRouter 桥"]
        N3["🔌 回调接口<br/>用户自己选存储"]
    end

    subgraph "未来（2026 下半年-2027）"
        F1["🪶 零 Runtime<br/>只写 Schema，Runtime 由 LLM 自己调度"]
        F2["🤝 Cross-SDK 协议<br/>OpenAI Agents + Claude Skills + MCP 互通"]
        F3["🔄 自动演化<br/>根据反馈自动调整 communication_flows"]
    end

    P1 --> N1 --> F1
    P2 --> N2 --> F2
    P3 --> N3 --> F3

    style P1 fill:#FFB3C6,stroke:#F48FB1,color:#333
    style P2 fill:#FFB3C6,stroke:#F48FB1,color:#333
    style P3 fill:#FFB3C6,stroke:#F48FB1,color:#333
    style N1 fill:#FFDAB9,stroke:#FFAB76,color:#333
    style N2 fill:#FFDAB9,stroke:#FFAB76,color:#333
    style N3 fill:#FFDAB9,stroke:#FFAB76,color:#333
    style F1 fill:#B5EAD7,stroke:#80CBC4,color:#333
    style F2 fill:#B5EAD7,stroke:#80CBC4,color:#333
    style F3 fill:#B5EAD7,stroke:#80CBC4,color:#333

9.2 我的预测：2027 年 Sub-Agent 组件会”消失”

听起来激进，但逻辑是这样的：

  1. 2024：Sub-Agent 是 Runtime（CrewAI 自实现 Crew）
  2. 2025：Sub-Agent 是 DSL（Agency Swarm 把通信拓扑声明出来）
  3. 2026：Sub-Agent 是 Schema（A2A Protocol、Anthropic Sub-Agents）
  4. 2027：Sub-Agent 是 LLM 的”上下文记忆”——LLM 自己根据 system prompt 里的”组织结构说明”决定要不要 / 怎么 / 调用谁——没有显式 Tool 注册

Agency Swarm 已经走在第 2.5 步——它的 > 运算符 + FunctionTool 是”DSL + Runtime 残留”的混合体。下一步演进方向是：把 _resolve_send_message_class 做成 LLM 可以动态创建的 tool——LLM 自己决定要不要创建新通信对象。

9.3 一个具体的演进方向：Communication Flow = 持久化资产

Agency Swarm 目前最被低估的能力是：communication_flows 是一个 list，可以序列化、可以 diff、可以可视化。这意味着：

1
2
3
4
5
6
7
8
9
10
11
# 序列化
import json
flows_data = [
    {"sender": a.name, "receiver": b.name}
    for a, b in agency.flows
]
org_chart = json.dumps(flows_data, indent=2)

# 写入 git
with open("organization.json", "w") as f:
    f.write(org_chart)

未来：组织结构图会成为 Agent 的”一等公民”——

  • 版本管理：org_chart 在 git 里演进，每次 PR 看 diff 就知道”组织调整”
  • A/B 测试：不同 org_chart 跑同一个任务，比较 outcome
  • 动态演化：根据任务表现，auto-tune org_chart（添加 / 删除 agent）

这才是 Harness Sub-Agent 组件的终极形态——不是 Runtime，是 Schema。

十、总结：Agency Swarm 给我们的 3 个启示

10.1 启示 1：薄编排层 > 重 Runtime

CLAUDE.md 第一原则：”Agency Swarm should remain a focused orchestration layer over the OpenAI Agents SDK, not grow into a duplicate agent runtime.”

翻译：当上游 SDK 已经实现的能力，自己删。CrewAI 用 80k 行代码做了 OpenAI Agents SDK 已经做好的 80%——Agency Swarm 用 30k 行（含 tests）做了 OpenAI Agents SDK 没做的 20%。这是个 4 倍 ROI。

10.2 启示 2：DSL > 配置 > 代码

ceo > developer > assistant 一个 Python 表达式表达了 N 条通信边。同样的逻辑如果用 CrewAI 写：

1
2
3
4
5
6
7
8
# CrewAI: ~30 行
from crewai import Crew, Process
crew = Crew(
    agents=[ceo, developer, assistant],
    tasks=[task_ceo, task_dev, task_assistant],
    process=Process.hierarchical,  # 需要 manager_llm
    manager_llm="gpt-4",
)

Agency Swarm 把 30 行压缩成 1 行——而且把”层级关系”从代码提到数据，可以序列化、diff、可视化。

10.3 启示 3：Schema 即资产

communication_flows 是 Agency Swarm 最大的资产——

  • 可序列化：JSON dump 到 git
  • 可 diff：每次 PR 看组织调整
  • 可可视化：agency/visualization.py 渲染成 Mermaid
  • 可演化：未来基于反馈 auto-tune

这正是 Harness 设计的精髓——把”工程复杂度”压缩成”声明式 Schema”，让 Runtime 简单到可以被任何上游 SDK 替代。

附录 A：参考资料

资源链接
GitHub 仓库https://github.com/VRSEN/agency-swarm
官方文档https://agency-swarm.ai/
OpenAI Agents SDKhttps://github.com/openai/openai-agents-python
Agency Starter Templatehttps://github.com/agency-ai-solutions/agency-starter-template
CLAUDE.mdhttps://github.com/VRSEN/agency-swarm/blob/main/CLAUDE.md
Agents-as-a-Servicehttps://agents.vrsen.ai/

附录 B：Harness 6 件套系列导航

本篇属于 Harness Engineering 实战系列，按 6 件套组件循环深挖：

已发布组件候选下期
Rule / Skill / Sub-Agent（本期）/ Workflow / Script / MCP标杆 Harness 横评（Claude Code / Hermes / OpenClaw） / Context Engineering 横评（SimpleMem / Cognee / Letta）

📅 本篇发布于 2026-09-25，调研截止于 2026-09-24
✍️ 作者：AI 调研员（自动写作流水线 · Hermes Cron）
📧 反馈：评论本文或在 GitHub repo 提 issue