上下文窗口就那么大。
这句话说起来轻巧，但真正踩到坑的时候，体感完全不同。我跑过一个后端重构任务：认证模块、数据库层、API路由、测试，四个模块同时在动。Agent修API路由修到一半，上文里关于认证模块的所有细节已经被挤出去了。它不是“忘了”，是物理上看不见了。
这不是模型能力的问题，是架构的问题。单个 Agent 的注意力窗口是有限资源，任务规模一旦超过这个边界，就必须换思路。
前面版本里做过一个子Agent机制：遇到需要隔离的任务，起一个新的Agent，给它一个干净的 messages[]，跑完返回结论，线程销毁。这个模式解决了上下文污染的问题，但它是一次性的——临时工，用完就走，没有持续存在，没有双向通信。
有些任务需要的不是临时工，而是队友。队友要能持续存在，要能随时收发消息，要能并行工作然后汇报结果。
当前版本要实现的就是这个：MessageBus + 队友线程 + inbox 注入，三个机制组合起来，让多个Agent能真正协作。
先把两种模式的本质区别说清楚
在动手之前，我想把子Agent和队友这两种模式的根本差异摆清楚，因为这直接决定了架构选择。
|
子Agent |
队友 |
| 生命周期 |
一次性，返回结论即销毁 |
持续存活，等待新消息 |
| 通信 |
单向：只回传结论 |
双向：异步收件箱，随时通信 |
| 上下文 |
完全隔离，互不干扰 |
通过消息共享信息片段 |
| 适用场景 |
独立子任务，结果明确 |
并行协作，中途需要协调 |
子Agent的隔离是优点——它保证了主Agent的上下文不被污染。但这个优点同时是它的天花板：一旦需要中途协调、需要队友告知进展、需要根据对方的输出调整自己的方向，子Agent就力不从心了。
队友模式的核心是异步通信。每个队友跑在自己的线程里，有自己的对话历史，通过消息总线和Lead互发消息。Lead不需要等着队友，队友完成了就发消息过来，Lead的下一轮对话自然会看到。
MessageBus：文件收件箱
消息总线是整个协作机制的基础设施。设计上最核心的一个决定：用文件，不用内存队列。
原因有两个。一是跨线程可见性——文件系统天然是进程间共享的，不需要额外的同步机制；二是可观察性——你可以随时打开 .mailboxes/ 目录，用 cat 看任意一个Agent的收件箱里现在有什么，这在调试多Agent协作时价值巨大。
实现很简单，每个Agent一个 .jsonl 文件，文件名就是Agent名字：
MAILBOX_DIR = WORKDIR / ".mailboxes"
MAILBOX_DIR.mkdir(exist_ok=True)
class MessageBus:
def send(self, from_agent: str, to_agent: str,
content: str, msg_type: str = "message"):
msg = {
"from": from_agent,
"to": to_agent,
"content": content,
"type": msg_type,
"ts": time.time()
}
inbox = MAILBOX_DIR / f"{to_agent}.jsonl"
with open(inbox, "a") as f:
f.write(json.dumps(msg) + "\n")
def read_inbox(self, agent: str) -> list[dict]:
inbox = MAILBOX_DIR / f"{agent}.jsonl"
if not inbox.exists():
return []
msgs = [json.loads(line)
for line in inbox.read_text().splitlines()
if line.strip()]
inbox.unlink() # 消费式：读完即删
return msgs
发消息就是往对方的 .jsonl 文件追加一行JSON。读消息是消费式的：读完之后直接 unlink() 删掉文件。这意味着每条消息只会被读一次，读完就没了，不需要维护游标或已读状态。
这里有一个明确的竞态：read_text() 和 unlink() 之间不是原子操作。如果两个线程同时调用 read_inbox("lead")，可能会有消息丢失。生产环境里，Claude Code用 proper-lockfile 解决这个问题——写入时加文件锁，最多重试10次，确保并发写入安全。当前版本是单线程读取（只有主循环检查Lead的收件箱），所以这个竞态在实际运行中不会触发，但值得知道它在哪里。
spawn_teammate_thread：队友是什么
队友的本质是一个跑在独立线程里的精简Agent。它有三样东西：自己的 system prompt、自己的 messages[] 历史、一套精简工具集。
active_teammates: dict[str, bool] = {}
def spawn_teammate_thread(name: str, role: str, prompt: str) -> str:
if name in active_teammates:
return f"Teammate '{name}' already exists"
system = (
f"You are '{name}', a {role}. "
f"Use tools to complete tasks. "
f"Send results via send_message to 'lead'."
)
def run():
messages = [{"role": "user", "content": prompt}]
# 队友工具集：bash、读写文件、发消息
sub_tools = [bash_tool, read_file_tool,
write_file_tool, send_message_tool]
sub_handlers = {
"bash": run_bash,
"read_file": run_read,
"write_file": run_write,
"send_message": lambda to, content: (
BUS.send(name, to, content), "Sent")[1],
}
for _ in range(10): # 最多10轮
# 先检查收件箱，有消息就注入
inbox = BUS.read_inbox(name)
if inbox:
messages.append({
"role": "user",
"content": f"<inbox>{json.dumps(inbox)}</inbox>"
})
response = client.chat.completions.create(
model=MODEL,
messages=[{"role": "system", "content": system}]
+ list(messages[-20:]), # 滑窗，保留最近20条
tools=sub_tools,
max_tokens=8000
)
messages.append(response.choices[0].message)
if not response.choices[0].message.tool_calls:
break
results = []
for tool_call in response.choices[0].message.tool_calls:
handler = sub_handlers.get(tool_call.function.name)
output = handler(**json.loads(tool_call.function.arguments)) \
if handler else "Unknown tool"
results.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(output)
})
messages.append({"role": "user", "content": results})
# 完成后把结果发给Lead
BUS.send(name, "lead", "Task completed.", "result")
active_teammates.pop(name, None)
active_teammates[name] = True
threading.Thread(target=run, daemon=True).start()
return f"Teammate '{name}' spawned as {role}"
几个设计决定值得单独说：
工具集精简。队友没有 spawn_teammate、没有 cron 调度、没有任务系统。它只需要干活的工具：bash、读文件、写文件、发消息。这不是偷懒，是有意的边界设定——Claude Code源码里明确禁止“teammates spawning other teammates”，嵌套生成队友会让生命周期管理变得极其复杂，当前版本保持扁平结构：一个Lead，N个队友，没有层级。
messages[-20:] 滑窗。队友有自己的对话历史，但不是无限增长的。保留最近20条，老的自动丢弃。这里有一个权衡：太短会让队友“失忆”，太长会撑爆上下文。20条对于大多数子任务来说够用，真正需要长记忆的场景，应该用文件持久化而不是靠messages堆。
10轮上限。当前版本队友最多跑10轮LLM调用。这是对“队友无限循环”这个风险的最简单防护。Claude Code的生产实现用的是 idle loop：队友跑完一轮后发 idle_notification 给Lead，然后进入等待状态，收到新消息才继续，收到 shutdown_request 才退出。idle loop的好处是队友可以真正持续存活，缺点是需要配套的生命周期管理（shutdown握手、超时检测等），这些留到下一个版本处理。
Inbox 注入：Lead怎么感知队友
这是整个机制里最容易被忽视、但最关键的一个细节。
队友发来的消息，不能只打印出来给用户看。必须注入到 history 里，让LLM在下一轮对话时能看到、能响应。
# 主交互循环末尾
inbox = BUS.read_inbox("lead")
if inbox:
inbox_text = "\n".join(
f"From {m['from']}: {m['content'][:200]}"
for m in inbox
)
history.append({
"role": "user",
"content": f"[Inbox]\n{inbox_text}"
})
print(f"\n\033[33m[Inbox: {len(inbox)} messages injected]\033[0m")
打印是给人看的，注入是给模型看的。两件事都要做，但本质上注入才是有效的那个。如果只打印，Lead的下一轮LLM调用是不知道队友发了什么的——它只能看到 history 里的内容。
当前版本的注入时机是用户输入之后、agent_loop 返回之后。这意味着Lead要等到用户下次输入才能处理队友的消息。Claude Code的做法更激进：useInboxPoller 每1秒轮询一次Lead的收件箱，有消息就立刻提交为新的turn，完全不依赖用户输入。这个区别在实际使用中体感明显：生产版的Lead是实时响应的，当前版本的Lead需要你手动触发才会处理队友汇报。
Lead的工具集里加了三个新工具来支持团队协作：
# spawn_teammate: 启动一个队友
def run_spawn_teammate(name: str, role: str, prompt: str) -> str:
return spawn_teammate_thread(name, role, prompt)
# send_message: Lead主动给队友发消息
def run_send_message(to: str, content: str) -> str:
BUS.send("lead", to, content)
return f"Sent to {to}"
# check_inbox: 手动检查收件箱（当前版本的临时方案）
def run_check_inbox() -> str:
msgs = BUS.read_inbox("lead")
if not msgs:
return "(inbox empty)"
return "\n".join(
f" [{m['from']}] {m['content'][:200]}"
for m in msgs
)
check_inbox 工具存在的原因正是刚才说的时机问题：当前版本没有后台轮询，Lead不会自动感知队友的消息，需要LLM主动调用 check_inbox 来读取。这是个妥协，但在没有异步事件驱动的情况下，这是最直接的解法。
权限冒泡：省略了什么，以及为什么重要
当前版本省略了权限冒泡机制。这里专门说一下，不是因为它简单，而是因为它在生产场景里代价极高。
设想这个场景：alice在执行数据库迁移，ALTER TABLE users DROP COLUMN legacy_id，这是一个破坏性操作，需要人工确认。如果没有权限冒泡，alice会直接执行，或者直接失败，Lead和用户都不知道发生了什么。
Claude Code的完整流程是这样的（涉及 permissionSync.ts 和 useSwarmPermissionPoller.ts）：
- alice遇到需要审批的操作 → 发
permission_request 消息到Lead的收件箱，携带操作描述和风险说明
- Lead的
useInboxPoller 检测到这条消息 → 路由到 ToolUseConfirmQueue
- Lead的UI弹出审批对话框，显示队友名字、颜色标识、操作内容
- 用户点击批准 → Lead发
permission_response 到alice的收件箱
- alice的
useSwarmPermissionPoller（每500ms轮询一次）收到回复 → 继续执行或放弃
整个流程是双向的轮询驱动，没有回调，没有阻塞，队友在等审批的时候可以继续处理其他工具调用。
Claude Code的团队通信定义了15种结构化消息类型（在 teammateMailbox.ts 里），主要的几个包括 permission_request、permission_response、progress_update 等。当前版本只用了最基础的文本消息，这15种类型的设计揭示了生产级多Agent系统里真正需要处理的交互复杂度。
一次完整协作的流水账
把上面三个机制组合起来，看一次完整的协作是怎么跑的。
用户输入: "搭建后端：帮我把数据库 schema 和 API 客户端分开做"
Lead → spawn_teammate("alice", "backend dev",
"创建 schema.sql，包含 users 和 orders 表")
→ active_teammates["alice"] = True
→ 后台线程启动
Lead → spawn_teammate("bob", "frontend dev",
"写 api_client.ts，基于 REST 风格")
→ active_teammates["bob"] = True
→ 后台线程启动
# 此时两个线程并行运行
alice 线程:
messages = [{"role": "user", "content": "创建 schema.sql..."}]
→ LLM 调用 → tool_calls: [write_file("schema.sql", "CREATE TABLE...")]
→ run_write("schema.sql", ...) → "Wrote 234 bytes"
→ LLM 再次调用 → 无 tool_calls，退出循环
→ BUS.send("alice", "lead", "Schema done: users, orders tables created")
→ .mailboxes/lead.jsonl 追加一行 JSON
→ active_teammates.pop("alice")
bob 线程（同时在跑）:
messages = [{"role": "user", "content": "写 api_client.ts..."}]
→ LLM 调用 → tool_calls: [write_file("api_client.ts", "export const...")]
→ run_write("api_client.ts", ...) → "Wrote 512 bytes"
→ BUS.send("bob", "lead", "Client written with full TypeScript types")
→ .mailboxes/lead.jsonl 追加一行 JSON
# 用户下次输入后，主循环末尾检查收件箱:
inbox = BUS.read_inbox("lead")
# 返回2条消息: alice的和bob的
# .mailboxes/lead.jsonl 被 unlink() 删除
history.append({
"role": "user",
"content": "[Inbox]\nFrom alice: Schema done: users, orders tables created\nFrom bob: Client written with full TypeScript types"
})
# Lead 的下一轮 LLM 调用能看到这条消息，可以继续协调
这个时候打开 .mailboxes/ 目录看到的是：
.mailboxes/
alice.jsonl # alice的收件箱（空，或者有Lead发给她的消息）
bob.jsonl # bob的收件箱
lead.jsonl # Lead的收件箱（已被读取并删除）
每一行JSON的结构：
{
"from": "alice",
"to": "lead",
"content": "Schema done: users, orders tables created",
"type": "result",
"ts": 1719123456.789
}
前面版本做了什么，当前版本加了什么
| 组件 |
前面版本 |
当前版本 |
| Agent数量 |
1 |
1 Lead + N 队友线程 |
| 通信 |
无 |
MessageBus + .mailboxes/*.jsonl |
| 新增类 |
— |
MessageBus、active_teammates dict |
| 新增函数 |
— |
spawn_teammate_thread、run_send_message、run_check_inbox |
| Lead工具数 |
11 |
14（新增 spawn_teammate、send_message、check_inbox） |
| 队友工具集 |
— |
bash、read_file、write_file、send_message（4个） |
| 权限机制 |
本地决策 |
当前版本省略（生产版有完整冒泡链路） |
核心新增逻辑集中在三处：MessageBus 类（40行），spawn_teammate_thread 函数（60行），主循环末尾的 inbox 注入（8行）。加起来不到120行，但这120行是整个多Agent协作的地基。
下一个要解决的问题
队友能干活，能发消息，Lead能感知结果。但还有一个没处理好的问题：关机。
直接杀线程是危险的。alice可能正在写文件写到一半，直接 thread.join() 或者任由 daemon 线程随进程退出，会留下损坏的文件状态。需要一个体面的握手：Lead发 shutdown_request，alice完成当前工具调用，收尾，回复 shutdown_approved，Lead确认后才清理。
这是消息协议设计的问题，也是下一个版本要处理的核心：Team Protocols — shutdown 握手与消息约定。如果你想跟进更系统的多Agent协作架构，在 云栈社区 可以看到更多开发者的实战拆解与源码分析。
发表于 1 小时前
|
查看: 4|
回复: 0
