第 16 章 / 共 21 章
团队协议：请求-响应的显式契约
上一章留下的问题：现在队友之间能发消息了，但两件事仍然做不好。
第一件是优雅关闭。你想让 alice 停下来，直接杀线程——它可能正写文件写到一半，留下一个半截文件。你需要的是”我请你停，你收尾完告诉我一声，我再让你走”。
第二件是方案审批。队友要执行一个高风险操作（删表、改生产配置），应该先把方案交给 Lead 看，批准了再做。
这两件事表面无关，底层是同一个东西：**一次请求，一次对应的回应，中间有状态。**这就是协议。
16.1 协议的三个零件
| 零件 | 作用 | 没有它会怎样 |
|---|---|---|
request_id | 把请求和响应关联起来 | 收到一条”同意”，不知道同意的是哪件事 |
type | 标明协议种类 | 关机的回复被当成审批的回复，误批一个危险操作 |
status | 记录生命周期（pending → approved / rejected） | 不知道请求有没有被回应，也没法超时处理 |
from dataclasses import dataclass, field
@dataclass
class ProtocolState:
request_id: str
type: str # "shutdown" | "plan_approval"
status: str = "pending" # pending | approved | rejected
sender: str = ""
target: str = ""
payload: dict = field(default_factory=dict)
pending_requests: dict[str, ProtocolState] = {}
_req_counter = 0
def new_request_id() -> str:
global _req_counter
_req_counter += 1
return f"req_{_req_counter:06d}"
request_id 在四步里的流转——它从创建时生成，随请求发出，被响应原样带回，最后用来定位状态。没有这根线，异步消息就只是一堆互不相干的字条。16.2 四个步骤
第一步，发起方创建状态并发出请求：
def request_shutdown(target: str) -> str:
req_id = new_request_id()
pending_requests[req_id] = ProtocolState(
request_id=req_id, type="shutdown", sender="lead", target=target)
BUS.send("lead", target, "shutdown_request", metadata={"request_id": req_id})
return f"已向 {target} 发出关机请求（{req_id}），等待其收尾确认。"第二步，接收方按类型路由：
def dispatch_message(agent_name: str, msg: dict) -> bool:
"""返回 True 表示这是一条协议消息，已被处理，不需要注入模型上下文。"""
mtype = msg.get("type")
req_id = msg.get("metadata", {}).get("request_id")
if mtype == "shutdown_request":
cleanup(agent_name) # 保存进度、关文件
BUS.send(agent_name, "lead", "shutdown_response",
metadata={"request_id": req_id, "approve": True})
return True
if mtype == "plan_approval_request":
# Lead 侧：把方案交给人或模型判断
approved = review_plan(msg.get("content", ""))
BUS.send("lead", msg["from"], "plan_approval_response",
metadata={"request_id": req_id, "approve": approved})
return True
return False第三步，响应带着同一个 request_id 回去（上面代码里已经体现）。
第四步，发起方匹配并更新状态：
def match_response(response_type: str, request_id: str, approve: bool) -> None:
state = pending_requests.get(request_id)
if state is None:
return # 未知或已处理
expected = {"shutdown": "shutdown_response",
"plan_approval": "plan_approval_response"}[state.type]
if response_type != expected:
return # 类型不匹配，丢弃
state.status = "approved" if approve else "rejected"那个类型校验不是多余的谨慎。想象一下：Lead 同时发出了一个关机请求和一个方案审批请求，如果代码只按 request_id 匹配而不校验类型，一次编码错误就可能让”同意关机”被当成”同意删表”。协议出错的代价，往往比业务代码出错高一个量级。
16.3 统一的收件箱消费
有一个顺序问题必须处理好：协议消息要先被路由，剩下的才注入模型上下文。
def consume_inbox(agent_name: str) -> list[dict]:
remaining = []
for msg in BUS.read_inbox(agent_name):
if msg.get("type", "").endswith("_response"):
meta = msg.get("metadata", {})
match_response(msg["type"], meta.get("request_id", ""), meta.get("approve", False))
continue
if dispatch_message(agent_name, msg):
continue
remaining.append(msg) # 普通消息，交给模型
return remaining如果反过来——先把消息注入模型上下文，再想办法处理协议——协议状态就会漏掉那些消息，状态机永远停在 pending。这也是为什么要有一个统一的消费入口，而不是在几个地方各读一次收件箱。
16.4 走一遍：一次优雅关闭
| 步骤 | Lead | alice | 状态 |
|---|---|---|---|
| 1 | request_shutdown("alice")，写入 req_000001 | — | pending |
| 2 | — | 收到 shutdown_request | pending |
| 3 | — | 保存当前进度、提交未完成的编辑、写一份交接说明 | pending |
| 4 | — | 回 shutdown_response（approve=True，带 req_000001） | pending |
| 5 | match_response 校验类型后置为 approved | 退出循环 | approved |
| 6 | 确认 alice 已安全退出，继续安排 | — | approved |
对比直接 thread.kill()：第 3 步的收尾工作不会发生，你会得到一个写了一半的文件和一个不知道自己被杀了的任务。
这里还有一个第15章遗留的坑被顺手填上了：队友必须在空闲时也持续读收件箱，否则它已经跑完 10 轮退出了，关机请求发过去根本没人接。第17章的空闲循环会正式解决这件事。
16.5 走一遍：一次方案审批
carol（队友）：我准备执行 `DROP TABLE legacy_sessions;` 来清理旧表。
→ send_message("carol", "lead", "plan_approval_request",
content="执行 DROP TABLE legacy_sessions，理由：迁移已完成，该表无引用",
metadata={"request_id": "req_000002"})
→ 进入等待，不执行
lead：收到，路由到 review_plan
→ 检查：迁移真的完成了吗？有没有备份？
→ 回 plan_approval_response(approve=False, reason="请先确认备份存在")
carol：收到拒绝，改为先做备份，再重新申请注意 carol 在等待期间没有执行那条 SQL。这是协议和”打个招呼”的根本区别：协议的请求方会真的阻塞在那里等结果。
16.6 工程现实
- **教学版不做执行拦截。**这里 carol 是”自觉”等待的。真实系统会在执行层拦截：未获批准的高危操作根本走不到执行函数——不要把安全押在被审批者的自觉上。
- **需要超时。**请求发出去石沉大海怎么办？每个
ProtocolState应该带一个截止时间，超时后按预设策略处理（关机默认强杀，审批默认拒绝）。 - **协议要能扩展。**关机和审批只是开头，后面还会有任务转交、资源申请、冲突仲裁。所以要有一张”类型 → 处理器”的注册表，而不是一串
if/elif——这和第2章的工具调度表是同一个模式。 - **消息要幂等。**重发、重复消费在分布式系统里是常态。
match_response里那句if state is None: return就是在做这件事。