AgentTeam：多Agent协作的工程实践

Agent Harness 工程：Agent Team 与多 Agent 协作

搜索功能已经被拆成三项互不依赖的工作：

Backend：实现查询接口Frontend：完成搜索界面QA：准备验收用例

Task System 清楚地知道它们都已就绪，Scheduler 也能不断产生新任务，可当前 Harness 里只有一个 Agent。

于是 Backend 做完，Frontend 才开始；Frontend 结束，QA 才接手。任务图画出了并行机会，执行端却仍然排成一条长队。

最直觉的改法是同时调用几个 Subagent。但并发调用不等于团队。一个真正能长期协作的 Agent Team，还要回答：

• 每个成员是谁，拥有什么能力和工具；

• 它们从哪里领取工作；

• 一条消息是否真的改变了任务状态；

• 成员崩溃后，谁能接手；

• 不同上下文中的结果如何交给下一位；

• 大家同时修改代码时，怎样避免互相覆盖。

这一章不追求“让更多模型一起聊天”，而是让多个执行者围绕同一组可靠事实工作。

Subagent 是一次委派，Team Member 是稳定角色

第 7 章的 Subagent 很适合局部问题。

主 Agent 把任务和必要上下文打包，子 Agent 在干净环境里完成工作，返回结果，然后结束。它像一次有边界的函数调用：

task package -> subagent -> compressed result

Team Member 的生命周期更长：

稳定身份-> 等待任务-> 认领 Task-> 完成或交接-> 休眠-> 再次被唤醒

“长期”并不意味着永远保留同一条messages。长期存在的是成员身份、角色、邮箱和恢复能力；每个 Task 仍然使用一段新的上下文。

这道区别非常重要：

Subagent 隔离一次局部思考，Agent Team 协调一群可以反复工作的执行者。

团队共享事实，不共享一团对话

一个最小团队需要三类共享记录：

1. Task Store

   ：目标、依赖、所有权、checkpoint 和最终结果；

2. Message Bus

   ：请求、进度、阻塞和唤醒通知；

3. Artifact Store

   ：代码提交、接口文档、截图、报告和日志。

成员拥有独立上下文和权限，只通过 Task、Message 与 Artifact 交换必要信息。

三者不能互相替代。

Backend 发来“接口完成了”，不会自动把 Task 变成completed；Task 完成也不需要把 Backend 的全部对话复制给 QA；Artifact 只是产物引用，还需要 Task Result 说明它解决了什么、怎样验证。

所以团队必须遵守一条底线：

Task Store 是工作状态的事实来源，Message Bus 只负责协调和唤醒。

消息可以迟到，甚至重复；只要 Task 状态已经持久化，下游工作就不会永远等在一封没有送达的信上。

成员身份由 Harness 注册

不要让模型临时发明角色、能力和工具权限。Harness 先维护一份可审计的 Team Registry：

from dataclasses import dataclassfrom typing import LiteralWorkspaceMode = Literal["read_only","isolated_write","integration",]@dataclass(frozen=True)class AgentProfile:id: strrole: strcapabilities: frozenset[str]tool_names: frozenset[str]workspace_mode: WorkspaceModerole_instructions: strmax_active_tasks: int = 1TEAM_PROFILES = {"lead": AgentProfile(id="lead",role="Team Lead",capabilities=frozenset({"planning", "review", "integration"}),tool_names=frozenset({"task_create","task_claim","task_complete","team_send_message","bash",}),workspace_mode="integration",role_instructions=("拆解目标、协调依赖、审查结果，""不要包办所有实现。"),),"backend": AgentProfile(id="backend",role="Backend Engineer",capabilities=frozenset({"backend", "database", "api"}),tool_names=frozenset({"task_claim","task_checkpoint","task_complete","team_send_message","read_file","edit_file","bash",}),workspace_mode="isolated_write",role_instructions=("负责服务端与 API，交付可验证的接口契约。"),),"frontend": AgentProfile(id="frontend",role="Frontend Engineer",capabilities=frozenset({"frontend", "ui", "accessibility"}),tool_names=frozenset({"task_claim","task_checkpoint","task_complete","team_send_message","read_file","edit_file","bash",}),workspace_mode="isolated_write",role_instructions=("负责界面与交互，并遵守设计和可访问性约束。"),),"qa": AgentProfile(id="qa",role="QA Engineer",capabilities=frozenset({"testing", "verification"}),tool_names=frozenset({"task_claim","task_checkpoint","task_complete","team_send_message","read_file","bash",}),workspace_mode="isolated_write",role_instructions=("验证验收条件，并提供可定位的测试结果。"),),"design": AgentProfile(id="design",role="Product Designer",capabilities=frozenset({"design", "ux", "accessibility"}),tool_names=frozenset({"task_claim","task_checkpoint","task_complete","team_send_message","read_file",}),workspace_mode="read_only",role_instructions=("审查交互与可访问性，并交付明确的设计反馈。"),),}

agent_id是稳定逻辑身份。成员重启后仍然叫backend。

Task 的claim_token则属于某一次运行实例，每次启动重新生成。稳定身份不能替代租约令牌，否则同一个角色的旧进程和新进程会被误认为同一个执行者。

tool_names是服务端权限上限，不是 Prompt 里的建议。Prompt Assembler 只展示允许的工具，实际处理器仍然根据身份、工作区和 Permission Policy 再次校验。

每个 Task 都从一张干净桌面开始

Backend Agent 完成十个 Task 后，如果仍然复用同一条对话，旧日志、旧错误和过期决策迟早会占满上下文。

更稳妥的结构是：

成员级状态Profile、稳定 Memory、邮箱摘要、当前 Task IDTask 级 Session当前任务、依赖结果、相关消息、Skill、文件与工具结果

成员认领 Task 时，Harness 新建 Task Session，只注入：

1. 基础系统规则；

2. 当前 AgentProfile；

3. 当前 Task 与验收条件；

4. 少量依赖结果；

5. 与当前 Task 相关的未读消息；

6. 可用 Artifact、工具和预算。

Task 结束后，完整messages可以丢弃或压缩归档。真正需要跨任务保留的经验进入 Memory，进度进入 checkpoint，产物进入 Artifact Store。

稳定角色带来连续性，干净上下文避免历史把下一项工作拖住。

任务要说明“谁有能力做”

Task 已经有目标和依赖。团队环境还需要能力约束和结构化结果：

from dataclasses import dataclass, field@dataclassclass TaskRouting:required_capabilities: list[str] = field(default_factory=list)@dataclassclass TaskResult:summary: strverification: list[str] = field(default_factory=list)risks: list[str] = field(default_factory=list)handoff: str = ""

然后为上一章的Task增加routing，并把原来的字符串结果升级为结构化TaskResult：

routing: TaskRouting = field(default_factory=TaskRouting)result: TaskResult | None = None

这会改变磁盘上的 Task 结构。实际项目要同步提升schema_version，并在from_dict()中把旧的字符串结果迁移成TaskResult(summary=...)；不能只改类型注解，就假设已经落盘的数据会自动变化。

例如：

task-apirequired: [backend, api]task-uirequired: [frontend, ui]task-api-testrequired: [testing, verification]depends_on: [task-api]task-reviewrequired: [review, integration]depends_on: [task-api-test, task-ui]

能力检查必须发生在 Task Store 的认领锁内：

def ensure_profile_can_claim(task: Task,profile: AgentProfile,) -> None:required = set(task.routing.required_capabilities)missing = required - profile.capabilitiesif missing:raise TaskUnavailableError(f"{profile.id} lacks {sorted(missing)}")def claim_team_task(store: TaskStore,task_id: str,agent_id: str,claim_token: str,lease_seconds: int = 300,) -> Task:profile = TEAM_PROFILES.get(agent_id)if profile is None:raise TaskUnavailableError(f"unknown member: {agent_id}")with store.locked():tasks = store._list_unlocked()task = next((itemfor item in tasksif item.id == task_id),None,)if task is None:raise TaskUnavailableError("task does not exist")ensure_profile_can_claim(task, profile)return claim_task_unlocked(store=store,tasks={item.id: item for item in tasks},task=task,agent_id=agent_id,claim_token=claim_token,lease_seconds=lease_seconds,)

这里把第 13 章的认领代码提取成claim_task_unlocked()，避免持有文件锁时再次进入同一把锁。

模型不能在参数里声明 capabilities。Harness 根据当前运行身份从 Registry 读取，否则任何成员都可以自称拥有deployment或integration能力。

Dispatcher 只负责叫醒，Task Store 决定归属

团队调度可以选择 Push，也可以选择 Pull：

• Push：Lead 指定某个成员，容易让 Lead 成为瓶颈；

• Pull：空闲成员自己找 Task，但会出现认领竞争。

更稳妥的第一版是混合模式：

1. Dispatcher 根据 capabilities 找到可能合适的空闲成员；

2. 它发送“有工作可领”的唤醒事件；

3. 成员醒来后重新读取 Task Store；

4. 最终所有权仍通过task_claim原子竞争。

唤醒只是提示，租约才是事实。

两个 Backend Agent 即使同时被叫醒，也只有一个能认领同一 Task。认领失败不是模型服务故障，不需要指数退避；成员重新查询任务图即可。

第一版最好让每个成员同时只持有一个 Task。一个成员同时维护多个 Task Session、租约和后台 Job，会显著增加恢复难度。需要更多并行时，增加成员通常比让一个成员频繁切换现场更清楚。

Message Bus 不是一间热闹的群聊

Task 依赖能表达“等谁完成”，却无法覆盖所有协作：

• Frontend 想提前确认接口字段；

• QA 发现验收条件不清楚；

• Backend 被外部服务阻塞，需要通知 Lead；

• Design 想查看一张截图。

这些信息适合 Message Bus，但消息应该短、结构化、可追踪：

from dataclasses import asdict, dataclass, fieldfrom typing import LiteralMessageKind = Literal["request","update","result","blocker",]@dataclassclass TeamMessage:id: strsender_id: strrecipient_id: strkind: MessageKindbody: strtask_id: str | None = Nonereply_to: str | None = Noneartifact_paths: list[str] = field(default_factory=list)created_at: str = ""acknowledged_at: str | None = Noneschema_version: int = 1def to_dict(self) -> dict:return asdict(self)

完整日志、代码 diff 和截图放进 Artifact Store，Message 只传摘要与路径。否则团队人数一多，消息会迅速变成另一份无法压缩的大上下文。

写入后响应丢了，不能再送一封新信

工具调用可能已经把消息写入磁盘，却在返回模型前中断。

如果重试时生成随机 ID，收件人会看到两条相同请求。Provider 的tool_call_id通常只保证当前响应里的关联关系，不适合直接充当全局幂等键。Harness 应先为每次工具调用持久化一个稳定的invocation_id，消息 ID 再由发送者和它共同生成：

import uuidfrom queue import Queuedef team_message_id(sender_id: str,invocation_id: str,) -> str:digest = uuid.uuid5(uuid.NAMESPACE_URL,f"{sender_id}:{invocation_id}",).hex[:16]return f"message-{digest}"def send_team_message(bus: MessageStore,wake_queues: dict[str, Queue[dict]],sender_id: str,invocation_id: str,recipient_id: str,kind: MessageKind,body: str,task_id: str | None = None,) -> TeamMessage:if recipient_id not in TEAM_PROFILES:raise ValueError("unknown recipient")message = TeamMessage(id=team_message_id(sender_id,invocation_id,),sender_id=sender_id,recipient_id=recipient_id,kind=kind,body=redact_sensitive_text(body.strip()),task_id=task_id,created_at=utc_now_text(),)stored = bus.put_if_absent(message)wake_queues[recipient_id].put({"kind": "team_message_ready","message_id": stored.id,})return stored

put_if_absent()必须在 Message Store 锁内完成“检查并创建”。相同 ID、相同业务内容返回原消息；内容冲突则报错。

消息先落盘，再唤醒成员。即使进程在两步之间退出，成员启动时扫描未确认邮箱，仍然能找到它。

消息进入上下文，但不会冒充用户

Team Message 不是用户输入，不应该追加成普通user消息。

Prompt Assembler 可以把未确认消息渲染为动态 Section：

Team inbox:- id: message-7f381a0c8d9f4b20from: backendkind: updatetask: task-apibody: API 已完成，接口契约见 artifact。artifacts:- artifact://task-api/openapi.json

一次模型调用成功后，再设置acknowledged_at。如果调用失败，消息保持未确认，下一轮继续投递。

至少一次投递意味着消息可能重复出现。Runtime 用稳定message_id去重，工具副作用则使用各自的幂等键。不能依靠模型说“我好像见过这条消息”。

一条消息不会解锁下游任务

Backend 完成 API 后，正确交接顺序是：

1. 保存 checkpoint 和 Artifact；

2. 调用task_complete写入结构化结果；

3. 依赖task-api的测试 Task 自然变成ready；

4. 给 QA 发送一条简短通知；

5. Dispatcher 或消息事件唤醒 QA；

6. QA 重新读取 Task Store 并认领测试 Task。

实线表示持久化的工作状态，虚线消息只负责沟通和加快发现。

如果消息先到、Backend Task 仍是in_progress，QA 可以提前阅读接口说明，却不能强行认领仍被依赖阻塞的 Task。

反过来，即使消息丢失，只要 Backend Task 已经完成，Dispatcher 仍然能发现下游工作。

结构化结果让交接不必回放整段聊天

成员完成 Task 时，结果应该是固定结构：

{"summary": "实现用户搜索 API，并处理空查询。","verification": ["pytest tests/api/test_search.py -q: passed"],"risks": ["模糊搜索仍依赖数据库 collation"],"handoff": "QA 重点检查分页边界和非 ASCII 查询。"}

Lead 和下游成员只需要读取 Task Result 与 Artifact，不需要回放 Backend 的完整messages。

Task Result 成功落盘后，Message Bus 中的result消息可以发送失败，也不会回滚已经完成的 Task。恢复时，新 Lead 同样可以从 Task Store 重建团队进度。

空闲成员应该休眠，而不是持续调用模型

Team Member 的循环由 Task 和 Message 唤醒：

import secretsfrom queue import Emptydef team_member_loop(state: HarnessState,profile: AgentProfile,) -> None:claim_token = secrets.token_urlsafe(16)wake_queue = state.member_wake_queues[profile.id]wake_queue.put({"kind": "member_started"})while state.accepting_team_work:try:wake_queue.get(timeout=30)except Empty:# 周期性 reconciliation，不调用模型。passunread = state.message_bus.list_unacknowledged(profile.id)priority_messages = [messagefor message in unreadif message.kind in {"request", "blocker"}]if priority_messages:run_coordination_turn(state,profile,priority_messages,)unread = state.message_bus.list_unacknowledged(profile.id)if any(message.kind in {"request", "blocker"}for message in unread):continuetask = claim_next_compatible_task(task_store=state.task_store,profile=profile,claim_token=claim_token,)if task is not None:run_task_session(state=state,profile=profile,task=task,claim_token=claim_token,initial_messages=[messagefor message in unreadif message.task_id in {None, task.id}],)elif unread:run_coordination_turn(state,profile,unread,)

没有工作时，成员阻塞在本地 Queue 上，不消耗 Token。

每 30 秒进行一次不调用模型的 reconciliation，是为了覆盖“Task 或 Message 已经持久化，但进程在发送唤醒前崩溃”的窗口。成员醒来后始终重新读取 Store，不把 Queue 中的通知当作最终事实。

如果一次协调回合后，高优先级消息仍未确认，成员本轮不会认领新 Task；它会回到 Queue 等待新事件或下一次 reconciliation，避免用自唤醒制造紧密的模型调用循环。

run_task_session()每完成一个模型或工具步骤，都应检查新消息和 Task 租约，保存 checkpoint，并在退出前明确 complete、fail 或 release。

成员崩溃后，不必执着于原来那个人

前几章已经准备好了恢复所需的拼图：

• Task 租约过期后变为reclaimable；

• claim_token

  阻止旧实例继续写入；

• checkpoint 说明上一次做到哪里；

• Artifact 保存代码、报告和日志；

• 持久化邮箱保留未确认消息；

• Job Record 让长操作可以查询，而不是盲目重跑。

如果 Backend Agent 长期离线，只要另一个 Profile 具备backend与apicapability，就可以在租约过期后接手。

agent_id用于角色、审计和协作，不应该把工作永久锁死在某个名字上。

如果没有任何 Profile 满足 required capabilities，Dispatcher 应向 Lead 报告“无人可执行”，而不是不停唤醒所有成员。

Lead 管方向，不做团队的传声筒

Lead 的职责是：

• 理解目标并创建 Task DAG；

• 设置能力要求和验收条件；

• 关注 blocked、failed 和无人可执行的 Task；

• 审查结构化结果；

• 完成最终集成与用户汇报。

Lead 不应该手工覆盖claimed_by，也不应该转发每一条成员消息。

Backend 和 Frontend 可以直接确认接口，QA 可以直接把复现步骤发给责任成员。只有优先级冲突、需求变化和最终集成需要回到 Lead。

如果所有沟通都要绕 Lead 一圈，团队虽然有很多 Agent，吞吐量仍然受制于一条对话。

真正危险的冲突发生在文件系统

Task 租约只能阻止两个成员同时执行同一个 Task。

Backend 修改共享类型，Frontend 也可能修改同一文件。它们认领的是两个合法 Task，却仍会在同一个工作目录里互相覆盖。

共享工作区无法清楚回答：

这行修改属于哪个 Task？这份测试结果验证的是哪一版文件？格式化工具改到了谁的现场？一个成员 reset 时会不会毁掉另一个成员的工作？

因此，可写成员需要独立工作区，结果通过 commit、patch 和验证记录交给 Integration Workspace。这个边界会在后续章节落实。

小结

这一章把一次性的 Subagent 扩展成了可持续协作的 Agent Team：

• AgentProfile 固定身份、能力、工具和工作区模式；

• 每个 Task 使用独立上下文，不让长期身份变成长对话；

• Task Routing 决定谁有资格认领；

• Dispatcher 只负责唤醒，Task Store 决定所有权；

• Message Bus 传递短消息，Artifact 保存大内容；

• 稳定消息 ID 和确认状态支持至少一次投递；

• Task Result 让交接不依赖聊天回放；

• 租约、checkpoint 和 Job Record 支持成员故障恢复；

• Lead 管目标与集成，不承包所有沟通。

多个 Agent 现在可以同时工作了，但团队仍缺少一件常被忽视的能力：安全地停下来。

如果 Lead 直接结束 Backend 线程，它可能正写到一半、持有 Task，或等待一个仍有副作用的 Job。