数据截至 (上游 commit 375d8081c48c)
Swarms — 高层协作拓扑
本章讲清楚:同一套底层引擎,怎么变出“总监分工、专家混合、群聊、投票”等不同协作风格,以及
SwarmRouter如何把它们统一。
1. 拓扑总览
swarms/structs/ 下有约 60 个协作结构。它们按“协作风格”可归几类:
| 类别 | 代表 | 一句话 |
|---|---|---|
| 流水/并行 | SequentialWorkflow、ConcurrentWorkflow、AgentRearrange、GraphWorkflow | 按顺序/并行/DAG 跑(见第 2 章) |
| 层级分工 | HierarchicalSwarm、PlannerWorkerSwarm | 一个头儿拆任务、派给工人 |
| 集体智慧 | MixtureOfAgents、MajorityVoting、CouncilAsAJudge、LLMCouncil | 多个独立回答再综合/表决 |
| 对话/博弈 | GroupChat、DebateWithJudge、multi_agent_debates | 多智能体对话、辩论、评审 |
| 深度分析 | HeavySwarm | 多阶段、多轮的重型研究流程 |
| 统一入口 | SwarmRouter | 一个参数在上述之间切换 |
下面挑三个最能代表不同范式的细讲。
2. 总监制:HierarchicalSwarm
要解决的小问题: 一个大任务,想让“经理”拆解并派活给“专才”,而不是所有人都做同一件事。
思路: 一个 director(总监)先产出一份结构化计划——一段 plan + 一串 orders(每条 order = 派给某个 agent 的具体指令),然后逐条执行,最后可选地由总监给反馈并再来一轮。
task
│
▼
┌─────────────┐ 产出 SwarmSpec { plan, orders[] }
│ Director │───────────────┐
└─────────────┘ │ 每条 order 指名 agent + 具体任务
▲ ▼
│ ┌──────┬──────┬──────┐
反馈(可选,再一轮) ▼ ▼ ▼ ▼
│ WorkerA WorkerB WorkerC ... ← 按 order 执行,结果写回黑板
└──────────────┴──────┴──────┘
真实实现:
- 计划的数据结构是 Pydantic 模型
SwarmSpec(含plan字段和orders: List[HierarchicalOrder],structs/hiearchical_swarm.py:85-111);每条HierarchicalOrder指定agent_name和该 agent 要做的任务(structs/hiearchical_swarm.py:41-64)。 - 一步
step()(structs/hiearchical_swarm.py:601):run_director让总监产出 →parse_orders抽出 plan 和 orders →execute_orders逐条派给对应 worker(call_single_agent,structs/hiearchical_swarm.py:931)。 - 多轮:
run()(structs/hiearchical_swarm.py:684)按max_loops循环,第 2 轮起把“上一轮结果 + 原任务”作为新的 loop 任务喂回去(structs/hiearchical_swarm.py:744-752);若director_feedback_on=True还会有一轮总监反馈(feedback_director,structs/hiearchical_swarm.py:838)。
用总监制强制 LLM 输出结构化 orders,是这类框架的常见手法:让“派单”这一步机器可读、可校验,而不是从自然语言里猜谁该干什么。
3. 专家混合:MixtureOfAgents
要解决的小问题: 想要多个(可以是不同厂商的)模型各自独立作答,再由一个“综合器”融合成一份更稳的答案。
思路: 分“层(layers)”跑——每层让所有 worker 并行作答、把答案都堆进共享黑板;跑完所有层,最后由 aggregator 读整块黑板做综合。
task ──► [ 第1层 ] worker1 ∥ worker2 ∥ worker3 → 各自答案写入黑板
│ (重复 layers 层)
▼
[ 第N层 ] ...
│
▼
Aggregator 读整块黑板 → 综合出最终答案
真实实现(MixtureOfAgents._run,structs/mixture_of_agents.py:177-215):for i in range(self.layers)(mixture_of_agents.py:196)每层跑一批 worker、结果 self.conversation.add(...) 进黑板;循环后 aggregator_output = self.aggregator_agent.run(task=self.conversation.get_str()) 综合(mixture_of_agents.py:209)。适合高风险任务想要“多个独立视角 + 一次融合”,尤其配不同模型时效果更好。
4. 自选发言群聊:GroupChat
要解决的小问题: 让若干 Agent 像开会一样自由发言,但又不能人人同时抢话。
思路(竞价发言): 没有固定轮次或“选人函数”。每一轮,所有 Agent 私下竞价——被强制调用一个 respond(score, message) 工具,给出“我有多想发言”的分数(0..1)和拟发内容;分数最高且超过 threshold 的那一个拿到发言权,只有它的话被广播。没人超过阈值就停(冷场即结束)。
真实实现:
RESPOND_TOOL(structs/groupchat.py:74-98)是一个强制函数调用的 schema:注释直言“让发言决策机器可读,避免解析自然语言”(structs/groupchat.py:75-76)。- 机制概述写在模块顶部注释(
structs/groupchat.py:17-23):每轮所有 agent 私下respond(score, message)竞价 → 单个最高竞价者(经新近度调整)超过阈值者拿到发言权 → 若一轮无人超过阈值则停止。 - 解析竞价时分数被夹到
0..1,缺失/非法视为“沉默”(0.0, "")(structs/groupchat.py:140-144)。
注意:是“最高者独占发言”,不是“所有超过阈值者都发言”。这是从源码读到的真实语义。调高
threshold→ 更挑剔、更安静的会议室;调低 → 更热闹。
5. 统一入口:SwarmRouter
要解决的小问题: 上面这么多拓扑,能不能用一个统一 API、只改一个参数就切换?
思路: SwarmRouter 内部维护一张工厂表——swarm_type 字符串 → 构造对应 Swarm 的函数。run 时按类型查表、构造(带缓存)、再把任务转交给那个 Swarm 的 run。
SwarmRouter(agents=[...], swarm_type="HierarchicalSwarm").run(task)
│
▼
_swarm_factory["HierarchicalSwarm"] → 构造 HierarchicalSwarm(带缓存)
│
▼
swarm.run(task) ← 真正干活的还是对应拓扑
真实实现:
- 支持的类型是一个
Literal,共 17 种(SwarmType,structs/swarm_router.py:154-172):SequentialWorkflow、ConcurrentWorkflow、AgentRearrange、MixtureOfAgents、HierarchicalSwarm、GroupChat、MajorityVoting、CouncilAsAJudge、HeavySwarm、MultiAgentRouter、DebateWithJudge、RoundRobin、PlannerWorkerSwarm、BatchedGridWorkflow、LLMCouncil、AutoSwarmBuilder、auto。 - 工厂表
_initialize_swarm_factory(structs/swarm_router.py:568-591)把每种类型映射到_create_xxx方法。 _create_swarm(structs/swarm_router.py:813)先查缓存_swarm_cache,未命中再调工厂并缓存(structs/swarm_router.py:834-857)——同配置只构造一次。_run(structs/swarm_router.py:902)构造后把task/tasks/img转交给swarm.run(structs/swarm_router.py:925-932)。
swarm_type="auto" 则让路由自己(借模型)选一种拓扑——适合“还不确定该用哪种结构”的原型阶段。
继续看设计取舍、边界与对比 → 04-deep-dive.md。