跳到主要内容

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:932)。
  • 多轮: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:839)。

用总监制强制 LLM 输出结构化 orders,是这类框架的常见手法:让“派单”这一步机器可读、可校验,而不是从自然语言里猜谁该干什么。

3. 专家混合:MixtureOfAgents​

要解决的小问题: 想要多个(可以是不同厂商的)模型各自独立作答,再由一个“综合器”融合成一份更稳的答案。

思路: 分“层(layers)”跑——每层让所有 worker 并行作答、把答案都堆进共享黑板;跑完所有层,最后由 aggregator 读整块黑板做综合。

  task ──► [ 第1层 ] worker1 ∥ worker2 ∥ worker3  → 各自答案写入黑板
              │  (重复 layers 层)
              ▼
           [ 第N层 ] ...
              │
              ▼
         Aggregator 读整块黑板 → 综合出最终答案

真实实现(MixtureOfAgents._run,structs/mixture_of_agents.py:178-216):for i in range(self.layers)(mixture_of_agents.py:197)每层跑一批 worker、结果 self.conversation.add(...) 进黑板;循环后 aggregator_output = self.aggregator_agent.run(task=self.conversation.get_str()) 综合(mixture_of_agents.py:210)。适合高风险任务想要“多个独立视角 + 一次融合”,尤其配不同模型时效果更好。

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。