Swarm Blackboard 模式：无监督 P2P（研究用）
解决什么问题
传统 Manager-Worker 有中心节点瓶颈。如果我们让一群 Agent 自主协作呢？ Swarm Blackboard 是一种去中心化模式：没有 Manager，所有 Agent 通过一块共享黑板 (Blackboard) 读写信息、自主认领任务。
这是一种研究性质的模式——适合探索性任务（如旅行灵感收集），但生产可控性差。
| 场景 | 为什么选它 |
|---|---|
| 旅行灵感收集 | 多个 Agent 各自搜索，把发现写到黑板 |
| 学术调研 | 无中心调度，各自探索再汇总 |
| 创意头脑风暴 | 不需要严格分工，自由碰撞 |
复杂度
⭐⭐⭐⭐⭐ 高。去中心化 = 难调试 + 难控制收敛。
和其他模式的关系
| 模式 | 谁决定下一步 | 什么时候用 |
|---|---|---|
| Swarm Blackboard | 每个 Agent 自主决定（读黑板→写黑板） | 发散性探索任务，不需要严格控制 |
| Manager-Worker | Manager 集中分配 | 确定性任务、需要可控产出 |
| Group Chat | 主持人控制轮次 | 多角色需要有序讨论 |
Swarm 比 Manager-Worker 更自由（无中心调度），比 Group Chat 更松散（无主持人、无轮次控制）。代价是可控性差。
角色关系
┌───────────────────────────────┐
│ Blackboard (共享黑板) │
│ "东京樱花季3月下旬" │
│ "京都岚山竹林必去" │
│ "大阪道顿堀美食街" │
└──┬──────┬──────┬──────┬───────┘
│ │ │ │ 读/写
▼ ▼ ▼ ▼
┌───┐ ┌───┐ ┌───┐ ┌───┐
│ A1 │ │ A2 │ │ A3 │ │ A4 │ ← P2P，无 Manager
└───┘ └───┘ └───┘ └───┘
- 所有 Agent 地位对等
- 通过 Blackboard 间接通信
- 每个 Agent 自主决定"读什么、写什么、何时停止"
完整 Python 实现
"""
Swarm Blackboard: 日本旅行灵感收集
4个 Agent 自主搜索，把灵感写到共享黑板，互相参考补充。
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from typing import Any
# ── MockLLM ──────────────────────────────────────────────
class MockLLM:
def __init__(self):
self.call_log: list[dict] = []
def chat(self, role: str, prompt: str) -> str:
self.call_log.append({"role": role, "prompt": prompt[:80]})
if role == "culture_scout":
if "黑板" in prompt and "樱花" in prompt:
return json.dumps({"action": "write", "content": "补充: 上野公园也是赏樱名所", "done": True})
return json.dumps({"action": "write", "content": "东京樱花季3月下旬至4月初，推荐目黑川、千鸟渊", "done": False})
if role == "nature_scout":
if "黑板" in prompt and "岚山" in prompt:
return json.dumps({"action": "write", "content": "补充: 岚山小火车需提前1周订票", "done": True})
return json.dumps({"action": "write", "content": "京都岚山竹林必去，嵯峨野观光小火车绝美", "done": False})
if role == "food_scout":
return json.dumps({"action": "write", "content": "大阪道顿堀: 章鱼烧、拉面、蟹道乐", "done": True})
if role == "budget_scout":
if "黑板" in prompt:
return json.dumps({"action": "write", "content": "预算参考: 日本7日人均¥8000-12000，JR Pass可省30%交通费", "done": True})
return json.dumps({"action": "write", "content": "交通建议: 买JR Pass全国版¥1500，覆盖新干线", "done": False})
return json.dumps({"action": "noop", "content": "", "done": True})
# ── Blackboard ───────────────────────────────────────────
@dataclass
class Blackboard:
"""共享黑板：所有 Agent 可读可写。"""
entries: list[dict] = field(default_factory=list)
def write(self, author: str, content: str):
entry = {"author": author, "content": content, "id": len(self.entries)}
self.entries.append(entry)
print(f" [黑板 ← {author}] {content[:50]}")
def read_recent(self, n: int = 5) -> list[dict]:
return self.entries[-n:]
def read_all(self) -> list[dict]:
return list(self.entries)
def summary(self) -> str:
return "\n".join(f" [{e['author']}] {e['content']}" for e in self.entries)
# ── Swarm Agent ──────────────────────────────────────────
@dataclass
class SwarmAgent:
name: str
role: str
llm: MockLLM = field(repr=False)
is_done: bool = False
def act(self, blackboard: Blackboard) -> bool:
"""读黑板 → 思考 → 写黑板。返回 True 表示自己认为完成了。"""
if self.is_done:
return True
# 读取黑板内容
recent = blackboard.read_recent(5)
bb_text = "\n".join(f"[{e['author']}] {e['content']}" for e in recent)
prompt = f"你是{self.name}。黑板内容:\n{bb_text}\n\n请决定下一步行动(write/noop)，用JSON回复:"
raw = self.llm.chat(self.role, prompt)
try:
action = json.loads(raw)
except json.JSONDecodeError:
action = {"action": "noop", "content": "", "done": True}
if action.get("action") == "write" and action.get("content"):
blackboard.write(self.name, action["content"])
if action.get("done"):
self.is_done = True
print(f" [{self.name}] 宣布完成")
return self.is_done
# ── Swarm Runner ─────────────────────────────────────────
@dataclass
class SwarmRunner:
agents: list[SwarmAgent]
blackboard: Blackboard = field(default_factory=Blackboard)
max_ticks: int = 5
def run(self, topic: str) -> Blackboard:
print(f"[Swarm] 话题: {topic}")
self.blackboard.write("系统", f"探索话题: {topic}")
for tick in range(self.max_ticks):
print(f"\n--- Tick {tick+1} ---")
all_done = True
for agent in self.agents:
if not agent.is_done:
done = agent.act(self.blackboard)
if not done:
all_done = False
if all_done:
print(f"\n[Swarm] 所有 Agent 完成，共 {tick+1} 轮")
break
print(f"\n[Swarm] 黑板最终内容:")
print(self.blackboard.summary())
return self.blackboard
# ── 运行入口 ─────────────────────────────────────────────
def main():
llm = MockLLM()
agents = [
SwarmAgent("文化探子", "culture_scout", llm),
SwarmAgent("自然探子", "nature_scout", llm),
SwarmAgent("美食探子", "food_scout", llm),
SwarmAgent("预算探子", "budget_scout", llm),
]
runner = SwarmRunner(agents=agents, max_ticks=5)
bb = runner.run("日本旅行灵感收集")
# 验证
assert len(bb.entries) >= 5 # 至少系统+4个 Agent 各写一条
contents = " ".join(e["content"] for e in bb.entries)
assert "樱花" in contents
assert "道顿堀" in contents or "章鱼烧" in contents
assert "JR Pass" in contents or "交通" in contents
print(f"\n✓ 共 {len(llm.call_log)} 次 LLM 调用，黑板收集 {len(bb.entries)} 条灵感")
if __name__ == "__main__":
main()
运行记录
[Swarm] 话题: 日本旅行灵感收集
[黑板 ← 系统] 探索话题: 日本旅行灵感收集
--- Tick 1 ---
[黑板 ← 文化探子] 东京樱花季3月下旬至4月初，推荐目黑川、千鸟渊
[黑板 ← 自然探子] 京都岚山竹林必去，嵯峨野观光小火车绝美
[黑板 ← 美食探子] 大阪道顿堀: 章鱼烧、拉面、蟹道乐
[美食探子] 宣布完成
[黑板 ← 预算探子] 交通建议: 买JR Pass全国版¥1500，覆盖新干线
--- Tick 2 ---
[黑板 ← 文化探子] 补充: 上野公园也是赏樱名所
[文化探子] 宣布完成
[黑板 ← 自然探子] 补充: 岚山小火车需提前1周订票
[自然探子] 宣布完成
[黑板 ← 预算探子] 预算参考: 日本7日人均¥8000-12000，JR Pass可省30%交通费
[预算探子] 宣布完成
[Swarm] 所有 Agent 完成，共 2 轮
[Swarm] 黑板最终内容:
[系统] 探索话题: 日本旅行灵感收集
[文化探子] 东京樱花季3月下旬至4月初，推荐目黑川、千鸟渊
[自然探子] 京都岚山竹林必去，嵯峨野观光小火车绝美
[美食探子] 大阪道顿堀: 章鱼烧、拉面、蟹道乐
[预算探子] 交通建议: 买JR Pass全国版¥1500，覆盖新干线
[文化探子] 补充: 上野公园也是赏樱名所
[自然探子] 补充: 岚山小火车需提前1周订票
[预算探子] 预算参考: 日本7日人均¥8000-12000，JR Pass可省30%交通费
✓ 共 8 次 LLM 调用，黑板收集 8 条灵感
踩坑记录
| 坑 | 现象 | trace 信号 | 修法 |
|---|---|---|---|
| 黑板爆炸 | 几百条垃圾信息 | blackboard entries 数量远超 Agent 数 × 预期每人贡献数 | 限制每个 Agent 最多写 N 条 |
| 永不收敛 | Agent 一直觉得没写完 | tick 数触达 max_ticks 且仍有 Agent 未 done | 设 max_ticks + "黑板超过 M 条就停" |
| 重复内容 | 3个 Agent 写了一样的 | 黑板中多条 entry 的文本相似度 > 0.9 | 写入前查重，或 prompt 里加"不要重复黑板已有内容" |
| 搭便车 | 有的 Agent 只读不写 | 某 Agent 的 write 事件数为 0 但 read 事件数正常 | 要求每个 Agent 至少贡献 1 条才能宣布完成 |
工程备忘
- 仅限研究/探索场景：生产环境不推荐纯 Swarm，可控性太差。
- 黑板 = 共享内存：生产中可以用 Redis 或数据库实现。
- 读写控制：可以加读写锁防止并发冲突，但 Swarm 本身就是松散的。
- 收敛策略：(a) 所有 Agent 宣布 done (b) 达到 max_ticks (c) 黑板条数达上限。
- 质量控制：可以加一个"评审 Agent"定期清理黑板上的低质量条目。
- 与 Manager-Worker 的对比：Manager-Worker 有中心调度，适合确定性任务；Swarm 无中心，适合发散性任务。
- 扩展方向：加入投票机制（Agent 给黑板条目点赞），高票内容优先保留。
读完以后
如果需要中心化调度、确定性产出，回到 Manager-Worker。 如果 Agent 之间需要有序讨论而非自由写黑板，看 Group Chat。 如果任务有明确的依赖关系和分支逻辑，看 Graph Orchestration。
参考资料
- Blackboard Systems (Wikipedia)
- Swarm Intelligence: From Natural to Artificial Systems
- OpenAI Swarm — 名字相似但实际是 Handoff 模式，对比阅读有助于理解区别