队友能干活之后，下一个暴露出来的问题往往不是能力，而是协调。
前面一个版本里，Lead 可以 spawn 队友、给他们发消息、接收结果。但消息是松散的——Lead 说“去创建这个文件”，队友去做，做完发个结果回来。这套流程在顺利路径下没问题，但一旦涉及需要确认的操作，就开始出裂缝。
两个具体场景触发了这个问题。
场景一：关机。 Lead 想让 Alice 停下来。最简单的办法是直接杀线程，但 Alice 正在写文件，内容写到一半，进程死了，磁盘上留着一个残缺的文件。正确的做法是握手：Lead 发请求，Alice 收到之后先把手头的事情收尾，然后回复“我准备好了，可以关”，Lead 确认后再关。
场景二：计划审批。 Bob 要重构认证模块。这是高风险操作——改错了影响面很大。合理的流程是 Bob 先把计划报上来，Lead 看了之后决定批还是驳，Bob 拿到批准再动手。
这两个场景看起来不一样，但抽象出来是同一个结构：一方发请求，另一方给回复，请求和回复通过同一个 ID 关联，有状态机追踪整个过程。
这就是当前版本要解决的问题。
︱为什么“发消息等回复”不够用
直觉上，用现有的 MessageBus 就能实现：Lead 发一条“请关机”，Alice 回一条“好的”，不就完了？
问题在于没有关联。
当 Lead 收到 Alice 回复的“好的”时，它怎么知道这是在回应哪个请求？如果 Lead 同时发了关机请求给 Alice，又发了一个任务分配消息，Alice 的“好的”对应的是哪一条？
更糟糕的是，如果 Lead 收到回复之前又发了一个新的关机请求（比如第一个请求超时了），两条请求的响应混在 inbox 里，根本分不清。
这是分布式系统里的经典问题：没有关联 ID，就没有办法把响应和请求对齐。
前面一个版本的消息总线只管把消息送达，不管语义层面的关联。当前版本要在上面加一层协议，核心就是三样东西：ProtocolState（请求状态追踪）、dispatch_message（按消息类型路由）、match_response（通过 request_id 关联响应）。
︱ProtocolState：给每个请求建一条档案
每发出一个协议请求，就在内存里创建一条状态记录。
@dataclass
class ProtocolState:
request_id: str # 唯一 ID，如 "req_004281"
type: str # "shutdown" | "plan_approval"
sender: str # 发起方，如 "lead"
target: str # 接收方，如 "alice"
status: str # pending | approved | rejected
payload: str # 计划文本或关机原因
created_at: float # 创建时间戳
pending_requests: dict[str, ProtocolState] = {}
pending_requests 是一个全局字典，key 是 request_id，value 是对应的 ProtocolState。
请求发出时，往这个字典里写一条记录，status 初始为 pending。收到响应时，根据响应里带的 request_id 找到这条记录，把 status 改成 approved 或 rejected。
request_id 的生成很简单：
def new_request_id() -> str:
return f"req_{random.randint(0, 999999):06d}"
格式是 req_ 加六位零填充数字，比如 req_004281。不需要 UUID 的强度，能唯一标识一次会话内的请求就够了。
︱四步协议流程
以关机握手为例，走一遍完整链路。
第一步：Lead 创建请求并发消息
def run_request_shutdown(teammate: str) -> str:
req_id = new_request_id()
pending_requests[req_id] = ProtocolState(
request_id=req_id, type="shutdown",
sender="lead", target=teammate,
status="pending", payload="")
BUS.send("lead", teammate, "Please shut down gracefully.",
"shutdown_request",
{"request_id": req_id})
return f"Shutdown request sent to {teammate} (req: {req_id})"
注意两点：先在 pending_requests 里注册，再发消息。顺序不能反——如果先发消息，队友可能在注册完成之前就回复了，回复到达时找不到对应的请求记录。
消息类型是 shutdown_request，元数据里携带 request_id。MessageBus 的 send 方法把这些打包进消息体写到队友的 inbox 文件。
第二步：队友收到消息，dispatch 路由到处理器
队友的主循环在每轮开始时检查 inbox：
inbox = BUS.read_inbox(name)
for msg in inbox:
if msg.get("type") in ("shutdown_request", "plan_approval_response"):
should_stop = handle_inbox_message(name, msg, messages)
handle_inbox_message 按消息类型分发：
def handle_inbox_message(name: str, msg: dict, messages: list) -> bool:
msg_type = msg.get("type", "message")
meta = msg.get("metadata", {})
req_id = meta.get("request_id", "")
if msg_type == "shutdown_request":
BUS.send(name, "lead", "Shutting down gracefully.",
"shutdown_response",
{"request_id": req_id, "approve": True})
return True # 停止循环
if msg_type == "plan_approval_response":
approve = meta.get("approve", False)
if approve:
messages.append({"role": "user",
"content": "[Plan approved] Proceed with the task."})
else:
messages.append({"role": "user",
"content": f"[Plan rejected] Feedback: {msg['content']}"})
return False # 继续循环
返回 True 表示队友应该停止，False 表示继续工作。这个设计让 dispatch 函数同时承担“处理消息”和“控制生命周期”两个职责，调用方根据返回值决定是否退出循环。
第三步：队友发回响应，携带同一个 request_id
BUS.send(name, "lead", "Shutting down gracefully.",
"shutdown_response",
{"request_id": req_id, "approve": True})
关键是 request_id 原样带回去。这是整个协议的关联键——Lead 收到这条消息时，靠这个 ID 找到最初发出的请求。
第四步：Lead 收到响应，match_response 关联并更新状态
def match_response(response_type: str, request_id: str, approve: bool):
state = pending_requests.get(request_id)
if not state:
return # 找不到请求记录，忽略
# 验证响应类型是否匹配请求类型
if state.type == "shutdown" and response_type != "shutdown_response":
return
if state.type == "plan_approval" and response_type != "plan_approval_response":
return
if state.status != "pending":
return # 已经处理过了，忽略重复响应
state.status = "approved" if approve else "rejected"
三层过滤：
request_id 必须存在于 pending_requests- 响应类型必须和请求类型匹配（
shutdown_response 不能 approve 一个 plan_approval 请求）
- 状态必须是
pending，已经 resolved 的请求不再处理
第二层过滤值得多说一句。如果只按 request_id 查表，不做类型校验，那么理论上一条 shutdown_response 可以意外地 resolve 一个 plan_approval 请求——只要 request_id 恰好相同。当前版本 request_id 生成用的是六位随机数，碰撞概率不高，但类型校验是正确性的保证，不能省。
︱dispatch_message 的设计：可扩展的类型路由
handle_inbox_message 的核心是一组 if 分支，每种消息类型对应一段处理逻辑。
这个结构看起来朴素，但它的扩展方式很清晰：加新的协议类型，就加一个新的 if 分支。不需要改任何现有代码，不需要注册表，不需要反射。
if msg_type == "shutdown_request":
# 处理关机请求
...
return True
if msg_type == "plan_approval_response":
# 处理计划审批响应
...
return False
# 普通消息：注入到 messages 让 LLM 处理
普通消息（不带协议类型的）直接注入到 LLM 的消息历史，让模型决定怎么响应。协议消息在代码层面处理，不进 LLM 上下文——这个分层很重要。协议逻辑是确定性的，不应该依赖 LLM 的判断。
︱统一的 inbox 消费入口
前面一个版本有一个隐患：check_inbox 工具（供 LLM 调用）和主循环末尾都会读 Lead 的 inbox。如果两处读取顺序不对，协议响应可能被其中一处读走，但没有触发 match_response，导致 pending_requests 里的记录永远卡在 pending。
当前版本的解法是把两处统一成一个函数：
def consume_lead_inbox(route_protocol: bool = True) -> list[dict]:
msgs = BUS.read_inbox("lead")
if not msgs:
return []
if route_protocol:
for msg in msgs:
meta = msg.get("metadata", {})
req_id = meta.get("request_id", "")
msg_type = msg.get("type", "")
if req_id and msg_type.endswith("_response"):
approve = meta.get("approve", False)
match_response(msg_type, req_id, approve)
return msgs
无论是 run_check_inbox()（工具调用）还是主循环末尾，都调用这个函数。协议路由在返回消息列表之前完成，不存在“读走但没路由”的问题。
msg_type.endswith("_response") 这个判断是个小技巧：所有协议响应消息的类型都以 _response 结尾，用这个规则统一过滤，不需要枚举所有响应类型。
︱队友的 idle loop
前面一个版本的队友跑完任务就退出——最多执行 10 轮 LLM 回合，结束。但关机握手要求队友在完成任务后继续存在，等待 Lead 发来 shutdown_request，收到之后才真正退出。
当前版本给队友加了一个 idle 循环：
if not response.choices[0].message.tool_calls:
# LLM 回合没有工具调用，说明任务完成，进入 idle
while not shutdown_requested:
time.sleep(1)
inbox = BUS.read_inbox(name)
if not inbox:
continue
for msg in inbox:
if msg.get("type") in ("shutdown_request", "plan_approval_response"):
should_stop = handle_inbox_message(name, msg, messages)
if should_stop:
shutdown_requested = True
break
else:
non_protocol.append(msg)
if shutdown_requested:
break
if non_protocol:
# 有新的普通消息，注入并回到 LLM 回合
inbox_json = json.dumps(non_protocol)
messages.append({"role": "user",
"content": "<inbox>" + inbox_json + "</inbox>"})
break
逻辑是：LLM 回合结束且没有工具调用，进入每秒轮询 inbox 的 idle 状态。
idle 里有两条出路：
- 收到
shutdown_request：回复 shutdown_response，设置 shutdown_requested = True，退出
- 收到普通消息：注入到
messages，break 出 idle 循环，回到 LLM 回合继续处理
这个设计让队友变成了真正的长期存在的 Agent，而不是一次性的任务执行器。
︱计划审批协议：从队友方向发起的请求
关机握手是 Lead → 队友方向。计划审批反过来，是队友 → Lead 方向。
队友用 submit_plan 工具提交计划：
def _teammate_submit_plan(from_name: str, plan: str) -> str:
req_id = new_request_id()
pending_requests[req_id] = ProtocolState(
request_id=req_id, type="plan_approval",
sender=from_name, target="lead",
status="pending", payload=plan)
BUS.send(from_name, "lead", plan,
"plan_approval_request",
{"request_id": req_id})
return f"Plan submitted ({req_id}). Waiting for approval..."
Lead 收到 plan_approval_request 后，用 review_plan 工具审批：
def run_review_plan(request_id: str, approve: bool, feedback: str = "") -> str:
state = pending_requests.get(request_id)
if not state:
return f"Request {request_id} not found"
if state.status != "pending":
return f"Request {request_id} already {state.status}"
state.status = "approved" if approve else "rejected"
BUS.send("lead", state.sender, feedback or ("Approved" if approve else "Rejected"),
"plan_approval_response",
{"request_id": request_id, "approve": approve})
return f"Plan {'approved' if approve else 'rejected'} ({request_id})"
审批结果通过 plan_approval_response 发回队友，队友的 handle_inbox_message 收到后，根据 approve 字段注入对应的消息到 LLM 上下文：
if msg_type == "plan_approval_response":
approve = meta.get("approve", False)
if approve:
messages.append({"role": "user",
"content": "[Plan approved] Proceed with the task."})
else:
messages.append({"role": "user",
"content": f"[Plan rejected] Feedback: {msg['content']}"})
注意这里有一个局限：当前版本的执行门控是靠 LLM 自觉的——模型收到 "waiting for approval" 之后应该等待，不去执行高风险操作，但代码层面没有拦截。队友在等待期间理论上还是能调 bash 或 write_file。真正的执行门控需要在工具分发层面做判断：如果当前有未批准的 plan_approval 请求且操作风险等级超过阈值，就拦截工具调用，返回 "waiting for plan approval"。这是当前版本的已知简化。
︱完整链路：一次关机握手的全程追踪
把所有机制串起来，走一遍：
1. 用户输入："spawn alice，让她创建 config.py，然后关机"
2. Lead: run_spawn_teammate("alice", "backend", "创建 config.py")
→ alice 线程启动
3. alice: LLM 决定调用 write_file("config.py", "...")
→ 文件写完
→ LLM 回合无 tool_call，进入 idle
4. Lead: run_request_shutdown("alice")
→ req_id = "req_004281"
→ pending_requests["req_004281"] = ProtocolState(type="shutdown", status="pending")
→ BUS.send("lead", "alice", "shutdown_request", {request_id: "req_004281"})
5. alice idle 轮询（每秒）
→ BUS.read_inbox("alice") 读到消息
→ msg_type == "shutdown_request"
→ handle_inbox_message 处理
→ BUS.send("alice", "lead", "shutdown_response", {request_id: "req_004281", approve: True})
→ 返回 True，设 shutdown_requested = True，退出循环
→ active_teammates.pop("alice")
6. 主循环末尾: consume_lead_inbox(route_protocol=True)
→ 读到 alice 的 shutdown_response
→ match_response("shutdown_response", "req_004281", approve=True)
→ pending_requests["req_004281"].status = "approved"
→ 消息注入 history
7. LLM 看到 "[Inbox] From alice: Shutting down gracefully."
→ 向用户报告关机完成
每一步有 request_id 可以追踪。如果中间某一步出了问题——比如 alice 没回复、回复类型不对、重复回复——match_response 的三层过滤会处理掉，不会污染状态。
︱相对前一个版本的核心变化
| 组件 |
前面一个版本 |
当前版本 |
| 协调方式 |
松散文本消息 |
结构化请求-响应协议 |
| 请求追踪 |
无 |
ProtocolState + pending_requests |
| 消息路由 |
全部当文本处理 |
dispatch 按类型分发 |
| 关机方式 |
自然退出或杀线程 |
request_id 握手 |
| 队友生命周期 |
最多 10 轮退出 |
idle loop，等待 inbox 消息 |
| Lead inbox |
多处独立读取 |
统一 consume_lead_inbox |
| 新消息类型 |
message, result |
+ shutdown_request/response, plan_approval_request/response |
| 新工具 |
— |
request_shutdown, request_plan, review_plan（Lead）；submit_plan（队友） |
︱协议的本质：把隐式约定变成可追踪的状态
回头看这套机制，它解决的不是一个技术问题，而是一个契约问题。
前面一个版本里，“请关机”和“好的”之间的关联是隐式的——靠时序、靠上下文、靠模型的理解去猜。这种隐式关联在正常路径下能工作，一旦出现并发、延迟、重试，就会乱。
request_id 把这个隐式关联变成了显式的：每个请求有 ID，每个响应携带 ID，状态机追踪 pending → approved / rejected。无论消息以什么顺序到达，无论中间有多少延迟，总能通过 ID 把请求和响应对齐。
这也是为什么两个看起来完全不同的场景——关机握手和计划审批——能用同一套机制处理。方向不同（Lead→队友 vs 队友→Lead），内容不同（关机 vs 计划文本），但底层结构一样：一方发出带 ID 的请求，另一方携带同一个 ID 回复，FSM 追踪状态。
Agent 团队越复杂，这层协议基础设施的价值就越大。有了它，Lead 可以同时管理多个队友的多个飞行中的请求，每一条都有状态、有历史、可追溯。没有它，规模一上来就会乱成一锅粥。
下一个要解决的问题是任务分配方式。当前版本里，Lead 必须手动告诉每个队友做什么——“Alice 做这个，Bob 做那个”。任务看板上有 10 个未认领的任务，Lead 得逐个 assign。这本身就是一个瓶颈。
下一步是让队友自己看任务看板、自己认领、自己完成——Lead 只需要创建任务，不需要分配。这就是自组织 Agent 团队的方向。
在云栈社区，我们持续追踪 AI 与后端架构 的前沿实践，欢迎一同交流探讨。
发表于 1 小时前
|
查看: 4|
回复: 0
