数据截至 (上游 commit 375d8081c48c)
Swarms — 编排引擎
本章讲清楚:多个 Agent 之间到底靠什么“串起来、传上下文、并行跑”。看懂这一层,上层几十种 Swarm 就都是变体。
1. 三个底层部件
多智能体编排听起来花样繁多,但底座就三件套:
| 部件 | 职责 | 文件 |
|---|---|---|
流程引擎 AgentRearrange | 解析 a -> b, c 这种流程串,决定谁先谁后、谁和谁并发 | structs/agent_rearrange.py |
对话黑板 Conversation | 所有 Agent 共读共写的一份对话历史,充当传话筒 | structs/conversation.py |
并发原语 run_agents_concurrently | 用线程池同时跑一批 Agent | structs/multi_agent_exec.py |
核心心智模型:黑板 + 调度器。 有一块共享黑板(Conversation),引擎按流程串的规则,轮流(或并行)让 Agent 上台:每个 Agent 读黑板上已有的一切、说一段话、把话写回黑板。下一个 Agent 因此能“看到”前面所有人的产出。上下文就是这么传递的——不是显式的返回值链,而是共享的对话历史。
2. 流程 DSL:a -> b, c 怎么解析
它要解决的小问题: 怎么用一行字符串表达“先 A,然后 B 和 C 并行,最后 D”。
思路(受 einsum 启发): 用两个符号就够了——-> 表示先后(顺序),, 表示同时(并发)。
流程串: "researcher -> writer, editor -> publisher"
解析:先按 -> 切成“步”,每步再按 , 切成“并发组”
步1 步2 步3
researcher → writer, editor → publisher
(单个) (两个并发) (单个)
执行顺序(→ 表示先后,│ 表示同一步内并行):
researcher
│
▼
writer editor ← 这两个并行跑,读同一份黑板
│ │
└───┬──┘
▼
publisher ← 读到 writer + editor 都写回后 的黑板
真实实现。 核心就在 AgentRearrange._run(structs/agent_rearrange.py:669):
- 先把任务写进黑板:
self.conversation.add("User", task)(structs/agent_rearrange.py:716)。 - 按
->切步:tasks = self.flow.split("->")(structs/agent_rearrange.py:716)。 - 每步再按
,切:agent_names = [name.strip() for name in task.split(",")](structs/agent_rearrange.py:741-743);长度 >1 走并发_run_concurrent_workflow,否则走_run_sequential_workflow(structs/agent_rearrange.py:739-755)。
顺序执行怎么读黑板。 _run_sequential_workflow(structs/agent_rearrange.py:584)里,每个 Agent 拿到的“任务”其实是整块黑板:agent.run(task=self.conversation.get_str(), ...),跑完 self.conversation.add(agent.agent_name, current_task) 写回(structs/agent_rearrange.py:664-674)。并发分支 _run_concurrent_workflow 同理,只是用并发原语一次跑一组(structs/agent_rearrange.py:562-566)。
一个巧妙处:顺序感知不污染黑板。 当开启团队感知时,引擎会告诉每个 Agent “你在流程里的位置、上一个是谁、下一个是谁”。这段“感知信息”不是写进共享黑板,而是临时拼到该 Agent 的 system_prompt 后面,跑完再还原(structs/agent_rearrange.py:630-668 的 try/finally)。源码注释给了两个理由:① 共享记录保持干净,下游看不到别人的私有感知;② system_prompt 内容默认不回显进对话,不会泄漏(structs/agent_rearrange.py:627-636)。
3. 并发原语:线程池怎么并行跑
这节讲什么: , 并发和 ConcurrentWorkflow 底下真正干活的那个函数。
run_agents_concurrently(structs/multi_agent_exec.py:96)非常直白——一个线程池,把每个 agent.run 提交进去:
# 简化示意,非源码
max_workers = int(os.cpu_count() * 0.95) # 默认吃满 ~95% 核
with ThreadPoolExecutor(max_workers=max_workers) as ex:
futures = [ex.submit(agent.run, task=task) for agent in agents]
results = []
for fut in as_completed(futures): # 按“完成先后”收集
try:
results.append(fut.result())
except Exception as e:
results.append(e) # 失败不炸,装进结果
对应真实源码:线程数 int(num_cores * 0.95)(structs/multi_agent_exec.py:142-144);默认用 as_completed 收集(structs/multi_agent_exec.py:180),所以结果顺序是完成顺序、不是输入顺序;单个 Agent 抛异常会被 catch 进结果列表而非中断整体(structs/multi_agent_exec.py:187)。想要“按输入顺序的字典”得传 return_agent_output_dict=True(structs/multi_agent_exec.py:162-177)。
为什么用线程池而不是多进程?因为 Agent 的瓶颈是等模型 API 返回(I/O 密集),线程足够并行;GIL 在等网络时会释放。仓库也提供了
run_agents_concurrently_async(asyncio 版,structs/multi_agent_exec.py:66)和 multiprocess 版(structs/multi_agent_exec.py:197)。
4. Sequential/Concurrent 只是薄封装
关 键发现:SequentialWorkflow 自己不做编排,它把活转包给了 AgentRearrange。
看 SequentialWorkflow.__init__:它把 agents 列表拼成一条 a -> b -> c 的流程串,构造一个内部的 AgentRearrange(structs/sequential_workflow.py:152);run() 直接 result = self.agent_rearrange.run(**run_kwargs)(structs/sequential_workflow.py:294)。
SequentialWorkflow(agents=[a,b,c]).run(task)
│ 内部生成流程串 "a -> b -> c"
▼
AgentRearrange(flow="a -> b -> c").run(task) ← 真正的引擎
所以这几个“工作流类”的关系是:
| 类 | 本质 | 说明 |
|---|---|---|
AgentRearrange | 引擎本体 | 全功能:顺序 + 并发混合 |
SequentialWorkflow | 引擎 + 固定 a -> b -> c 流程串 | 只顺序 |
ConcurrentWorkflow | 直接用并发原语 | 所有 Agent 同任务并行 |
这解释了为什么 SwarmRouter 要为 SequentialWorkflow 特判去取 self.swarm.agent_rearrange.conversation——因为它的黑板藏在内部那个 AgentRearrange 里(structs/swarm_router.py:885-888)。
额外能力:漂移检测。 SequentialWorkflow 可配一个 drift_agent:流水线跑完后,由它给“最终输出是否还对齐原始任务”打分,低于阈值就重跑,直到达标或判据无法解析才罢手(structs/sequential_workflow.py:231 的 _run_drift_detection,run 里在 structs/sequential_workflow.py:297-299 触发)。
下一章:在这套引擎之上,Swarms 怎么玩出总监制、专家混合、群聊、投票等花样 → 03-swarm-patterns.md。