跳到主要内容

数据截至 (上游 commit 4d1c35a6dd5e)

编排层:通信流与 send_message 工具

这章讲 Agency Swarm 最有特色的那层:你写的「谁能给谁发消息」是怎么变成 agent 手里一个真工具的。读完你会明白 agent1 > agent2 背后发生了什么,以及 send_message(求助)和 Handoff(转交)的区别。

3.1 通信流:从运算符到边列表​

它要解决的小问题。 你想用最自然的语法声明委派关系。Agency Swarm 让你写 manager > analyst,意思是「manager 可以给 analyst 发消息」。这是一条有向边——反过来 analyst 不能主动找 manager。

思路。 Python 允许重载 > / <。Agent.__gt__ 不返回布尔,而返回一个 AgentFlow 对象,把这条边记下来;链式的 a > b > c 会累积成多条边。

# 示意,非源码:运算符如何攒边
flow = manager > analyst        # AgentFlow,内部 _all_flows = [(manager, analyst)]
flow = manager > analyst > writer  # _all_flows = [(manager,analyst),(analyst,writer)]

真实实现。 Agent.__gt__ 直接构造 AgentFlow([self, other]),见 src/agency_swarm/agent/core.py:862-870。AgentFlow 在构造时把相邻对拆成边存进 _all_flows,见 src/agency_swarm/agent/agent_flow.py:25-33、链式追加见 agent_flow.py:35-50。

一个坑(也是巧妙之处)。 Python 的比较链 a > b > c 实际等价于 (a > b) and (b > c),中间结果会被 and 吃掉。为接住链式里被 and 求值的中间 AgentFlow,AgentFlow.__bool__ 被重载成「把自己的边塞进一个类级全局列表 _chain_flows」再返回 True,见 agent_flow.py:73-90。Agency 解析时用 AgentFlow.get_and_clear_chain_flows() 把这些「比较链副产物」捞回来。作者在源码注释里直白承认这是 a hack to work around Python's comparison chaining(agent_flow.py:76)。

解析入口。 parse_agent_flows(src/agency_swarm/agency/setup.py:82-182)统一处理所有支持的 flow 写法,产出三样东西:

产物含义
basic_flows(sender, receiver) 边列表
tool_class_mapping(sender_name, receiver_name) → [自定义 SendMessage 子类...]
default_tool_pairs用默认 SendMessage 工具的边集合

支持的 flow 条目形态(由 parse_agent_flows 分支处理):

  • (Agent, Agent) —— 基本边,见 setup.py:113-119
  • AgentFlow 或 (AgentFlow, tool_class) —— 链展开,见 setup.py:121-147
  • (Agent, Agent, tool_class) —— 给这条边指定自定义工具类,见 setup.py:155-177

3.2 send_message 工具:按边「长」出来​

它要解决的小问题。 每个 sender agent 需要一个工具,能列出「我能找谁」并把消息发过去。这个工具的可选 recipient 列表必须随通信流而变。

思路。 Agency 构造末尾调 configure_agents,对每条边在 sender 身上调 register_subagent:第一次创建一个 SendMessage 工具实例,之后每多一个 recipient 就往同一个工具的 enum 里 add_recipient。

下图是「构造期把边变成工具」的流向:

 communication_flows=[manager>analyst, manager>writer]
        │ parse_agent_flows()
        ▼
  basic_flows = [(manager,analyst), (manager,writer)]
        │ configure_agents()  (agency/setup.py)
        ▼
  对 sender=manager 的每个 recipient:
        register_subagent(manager, analyst, runtime_state)
        register_subagent(manager, writer,  runtime_state)
        │
        ▼
  manager 的 runtime_state.send_message_tools["SendMessage"]
     = 一个 SendMessage 实例,recipients = {analyst, writer}
     enum: recipient_agent ∈ ["RiskAnalyst", "ReportGenerator"]

真实实现。

  • configure_agents 先建 sender → [recipients] 映射,再逐对调 register_subagent,见 src/agency_swarm/agency/setup.py:237-322。注意它还会拦截:如果一个 agent 被设成 supports_outbound_communication=False(只收不发)却出现在某条边的 sender 位,直接抛错(setup.py:260-265)。
  • register_subagent(src/agency_swarm/agent/subagents.py:34-110)在 runtime_state.send_message_tools 里按工具类名做 key 复用同一个工具实例;新 recipient 走 add_recipient 增量更新 enum,见 subagents.py:85-109。
  • 工具实例自己在 SendMessage.__init__ 里据 recipients 构造 JSON schema:recipient_agent 是个 enum,外加 message、additional_instructions 两个参数,见 src/agency_swarm/tools/send_message.py:87-119。工具描述里还会把每个 recipient 的 description 列出来,让模型知道「该找谁」,见 send_message.py:166-173。

关键细节:工具是「运行期」可见的。 send_message 工具不是塞进 agent.tools 静态列表,而是存在 runtime_state.send_message_tools 里。Agent.get_all_tools 在每次运行时把它们和静态工具合并暴露给 SDK,见 src/agency_swarm/agent/core.py:382-400。这是「Agent 无状态、通信能力按 agency 注入」设计的直接体现(详见第 2 章)。

3.3 调用 send_message 会发生什么​

核心一句话:调用 send_message = 同步地把 recipient agent 跑一整轮,拿它的 final_output 当工具返回值。 sender 的模型拿到这个字符串后继续自己的推理。

SendMessage.on_invoke_tool 的主干(src/agency_swarm/tools/send_message.py:314-539):

  1. 解析参数、校验 recipient 在 enum 里(大小写不敏感查找),见 send_message.py:334-372。
  2. 去重锁:同一个 thread 内,若对同一个 recipient 已有一条消息在处理中,直接返回错误「别同时发两条」,见 send_message.py:379-393。这对应工具 docstring 里那句 Do not send more than 1 message to the same recipient agent at the same time。
  3. 给 recipient 造一份临时 AgencyContext(_create_recipient_agency_context,send_message.py:268-312),然后:
    • 流式模式:调 recipient.get_response_stream,把子 agent 的事件转发到父流,同时收集 final_output,见 send_message.py:408-498。
    • 非流式:直接 await recipient.get_response(...),见 send_message.py:499-539。
  4. 把子 agent 的 raw_responses 连同其模型名塞进 wrapper.context._sub_agent_raw_responses,供后续逐响应成本计算,见 send_message.py:481-492、513-524。
  5. finally 里无论成败都把 recipient 从待处理集合里移除(线程安全),见 send_message.py:560-567。
# 示意,非源码:send_message 调用的本质
async def on_invoke_tool(wrapper, args_json):
    recipient = self.recipients[args["recipient_agent"].lower()]
    # 关键:这就是又跑了一整轮一个 agent
    response = await recipient.get_response(
        message=args["message"],
        sender_name=self.sender_agent.name,        # 标记「这条来自某 agent」
        additional_instructions=args.get("additional_instructions") or None,
        agency_context=recipient_ctx,              # 临时上下文,共享同一个 thread_manager
        parent_run_id=wrapper.tool_call_id,        # 把工具调用 id 当父 run id,串起调用树
    )
    return response.final_output                   # 字符串塞回 sender 的模型

自定义 send_message。 你可以子类化 SendMessage 给委派加额外参数(比如 priority),框架支持三种声明方式:嵌套 ExtraParams(BaseModel)、类属性 extra_params_model,或直接在子类上写带 Field() 的注解。自动发现逻辑见 _discover_inline_fields(send_message.py:221-266),它会把这些字段合进工具 schema 并在调用时做 Pydantic 校验(send_message.py:344-355)。

3.4 Handoff:另一种通信(转交,不是求助)​

区别一句话:send_message 是「我派活给你,你干完把结果还我,我接着干」;Handoff 是「我把控制权整个交给你,后面由你接着跟用户对话」。 前者是子程序调用,后者是控制权转移。

Handoff(src/agency_swarm/tools/send_message.py:570-642)基于 SDK 的 handoff() 构造一个 transfer_to_<Agent> 工具,并:

  • 把输入 schema 收紧成只含 recipient_agent 的 Literal,以统一 send_message 和 handoff 的调用形态,见 send_message.py:585-589。
  • 默认 add_reminder=True:转交时往历史里追加一条 system 提醒「Transfer completed. You are . Please continue the task.」,让接手的 agent 知道自己现在是谁,见 send_message.py:594-641。提醒文案可由 Agent(..., handoff_reminder=...) 覆盖(send_message.py:605-611)。

Handoff 在 configure_agents 里走单独分支:它生成的不是 send_message 工具,而是被塞进 runtime_state.handoffs,见 src/agency_swarm/agency/setup.py:282-301。SendMessageHandoff 是 Handoff 的已弃用别名(send_message.py:645-647)。

何时用哪个:

维度send_messageHandoff
语义同步求助、拿结果回来转交控制权,自己退出
之后谁跟用户对话还是原 agent接手的 agent
底层自定义 FunctionTool,递归调 get_responseSDK 原生 handoff() + 历史提醒
配置入口通信流默认就是它在通信流里给边指定 Handoff 类

接下来​

通信工具调用时反复出现一个词:AgencyContext / runtime_state / parent_run_id。这些「运行期注入的状态」是理解 Agency Swarm 的关键,下一章 02-execution-and-context.md 端到端拆开它。