Agent Teams
协作Teammates + Mailboxes
Claude Code 可以同时跑几个独立的 agent,每个有自己的人设、自己的对话历史。它们之间通过一个共享邮箱互相发消息,能互相看见、互相回应。模拟评审小组、跨学科咨询这类需要多视角讨论的场景就靠这套机制。
组一支 4 人审稿队读 07_论文写作/01_主表/PC_ESG_主表汇总.docx 与 docs/01_实证研究设计.md,针对"A2 主口径 OLS β=0.0004 不显著 + IV1 0.0052** / IV3 0.0021** / PSM 0.0032*** / Placebo p_perm=0.000 全部显著"这种 OLS 与 IV 系数相差 10 倍的事实做四类视角的评审:方法学审稿人盯识别策略、计量审稿人盯标准误聚类、写作审稿人盯叙事一致性、新意审稿人盯相对唐亮 2025 与李思飞 2025 的边际贡献。
s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > [ s09 ] s10 > s11 > s12
"任务太大一个人干不完, 要能分给队友" -- 持久化队友 + JSONL 邮箱。
Harness 层: 团队邮箱 -- 多个模型, 通过文件协调。
问题
Subagent (s04) 是一次性的: 生成、干活、返回摘要、消亡。没有身份, 没有跨调用的记忆。Background Tasks (s08) 能跑 shell 命令, 但做不了 LLM 引导的决策。
真正的团队协作需要三样东西: (1) 能跨多轮对话存活的持久 Agent, (2) 身份和生命周期管理, (3) Agent 之间的通信通道。
解决方案
Teammate lifecycle:
spawn -> WORKING -> IDLE -> WORKING -> ... -> SHUTDOWN
Communication:
.team/
config.json <- team roster + statuses
inbox/
alice.jsonl <- append-only, drain-on-read
bob.jsonl
lead.jsonl
+--------+ send("alice","bob","...") +--------+
| alice | -----------------------------> | bob |
| loop | bob.jsonl << {json_line} | loop |
+--------+ +--------+
^ |
| BUS.read_inbox("alice") |
+---- alice.jsonl -> read + drain ---------+
工作原理
- TeammateManager 通过 config.json 维护团队名册。
class TeammateManager:
def __init__(self, team_dir: Path):
self.dir = team_dir
self.dir.mkdir(exist_ok=True)
self.config_path = self.dir / "config.json"
self.config = self._load_config()
self.threads = {}
spawn()创建队友并在线程中启动 agent loop。
def spawn(self, name: str, role: str, prompt: str) -> str:
member = {"name": name, "role": role, "status": "working"}
self.config["members"].append(member)
self._save_config()
thread = threading.Thread(
target=self._teammate_loop,
args=(name, role, prompt), daemon=True)
thread.start()
return f"Spawned teammate '{name}' (role: {role})"
- MessageBus: append-only 的 JSONL 收件箱。
send()追加一行;read_inbox()读取全部并清空。
class MessageBus:
def send(self, sender, to, content, msg_type="message", extra=None):
msg = {"type": msg_type, "from": sender,
"content": content, "timestamp": time.time()}
if extra:
msg.update(extra)
with open(self.dir / f"{to}.jsonl", "a") as f:
f.write(json.dumps(msg) + "\n")
def read_inbox(self, name):
path = self.dir / f"{name}.jsonl"
if not path.exists(): return "[]"
msgs = [json.loads(l) for l in path.read_text().strip().splitlines() if l]
path.write_text("") # drain
return json.dumps(msgs, indent=2)
- 每个队友在每次 LLM 调用前检查收件箱, 将消息注入上下文。
def _teammate_loop(self, name, role, prompt):
messages = [{"role": "user", "content": prompt}]
for _ in range(50):
inbox = BUS.read_inbox(name)
if inbox != "[]":
messages.append({"role": "user",
"content": f"<inbox>{inbox}</inbox>"})
response = client.messages.create(...)
if response.stop_reason != "tool_use":
break
# execute tools, append results...
self._find_member(name)["status"] = "idle"
相对 s08 的变更
| 组件 | 之前 (s08) | 之后 (s09) |
|---|---|---|
| Tools | 6 | 9 (+spawn/send/read_inbox) |
| Agent 数量 | 单一 | 领导 + N 个队友 |
| 持久化 | 无 | config.json + JSONL 收件箱 |
| 线程 | 后台命令 | 每线程完整 agent loop |
| 生命周期 | 一次性 | idle -> working -> idle |
| 通信 | 无 | message + broadcast |
试一试
cd claude-code-for-researchers
python agents/s09_agent_teams.py
试试这些 prompt (英文 prompt 对 LLM 效果更好, 也可以用中文):
Spawn method_reviewer and stat_reviewer. method_reviewer sends stat_reviewer a note about IV3 (PC lag).Broadcast "main.tex v2 ready for mock review — IV/PSM done, Myopia mechanism added" to all reviewersCheck the author inbox for review summaries- 输入
/team查看团队名册和状态 - 输入
/inbox手动检查领导的收件箱