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 resultTeam Member 的生命周期更长稳定身份- 等待任务- 认领 Task- 完成或交接- 休眠- 再次被唤醒“长期”并不意味着永远保留同一条messages。长期存在的是成员身份、角色、邮箱和恢复能力每个 Task 仍然使用一段新的上下文。这道区别非常重要Subagent 隔离一次局部思考Agent Team 协调一群可以反复工作的执行者。团队共享事实不共享一团对话一个最小团队需要三类共享记录Task Store目标、依赖、所有权、checkpoint 和最终结果Message Bus请求、进度、阻塞和唤醒通知Artifact Store代码提交、接口文档、截图、报告和日志。成员拥有独立上下文和权限只通过 Task、Message 与 Artifact 交换必要信息。三者不能互相替代。Backend 发来“接口完成了”不会自动把 Task 变成completedTask 完成也不需要把 Backend 的全部对话复制给 QAArtifact 只是产物引用还需要 Task Result 说明它解决了什么、怎样验证。所以团队必须遵守一条底线Task Store 是工作状态的事实来源Message Bus 只负责协调和唤醒。消息可以迟到甚至重复只要 Task 状态已经持久化下游工作就不会永远等在一封没有送达的信上。成员身份由 Harness 注册不要让模型临时发明角色、能力和工具权限。Harness 先维护一份可审计的 Team Registryfrom dataclasses import dataclassfrom typing import LiteralWorkspaceMode Literal[read_only,isolated_write,integration,]dataclass(frozenTrue)class AgentProfile:id: strrole: strcapabilities: frozenset[str]tool_names: frozenset[str]workspace_mode: WorkspaceModerole_instructions: strmax_active_tasks: int 1TEAM_PROFILES {lead: AgentProfile(idlead,roleTeam Lead,capabilitiesfrozenset({planning, review, integration}),tool_namesfrozenset({task_create,task_claim,task_complete,team_send_message,bash,}),workspace_modeintegration,role_instructions(拆解目标、协调依赖、审查结果不要包办所有实现。),),backend: AgentProfile(idbackend,roleBackend Engineer,capabilitiesfrozenset({backend, database, api}),tool_namesfrozenset({task_claim,task_checkpoint,task_complete,team_send_message,read_file,edit_file,bash,}),workspace_modeisolated_write,role_instructions(负责服务端与 API交付可验证的接口契约。),),frontend: AgentProfile(idfrontend,roleFrontend Engineer,capabilitiesfrozenset({frontend, ui, accessibility}),tool_namesfrozenset({task_claim,task_checkpoint,task_complete,team_send_message,read_file,edit_file,bash,}),workspace_modeisolated_write,role_instructions(负责界面与交互并遵守设计和可访问性约束。),),qa: AgentProfile(idqa,roleQA Engineer,capabilitiesfrozenset({testing, verification}),tool_namesfrozenset({task_claim,task_checkpoint,task_complete,team_send_message,read_file,bash,}),workspace_modeisolated_write,role_instructions(验证验收条件并提供可定位的测试结果。),),design: AgentProfile(iddesign,roleProduct Designer,capabilitiesfrozenset({design, ux, accessibility}),tool_namesfrozenset({task_claim,task_checkpoint,task_complete,team_send_message,read_file,}),workspace_moderead_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只注入基础系统规则当前 AgentProfile当前 Task 与验收条件少量依赖结果与当前 Task 相关的未读消息可用 Artifact、工具和预算。Task 结束后完整messages可以丢弃或压缩归档。真正需要跨任务保留的经验进入 Memory进度进入 checkpoint产物进入 Artifact Store。稳定角色带来连续性干净上下文避免历史把下一项工作拖住。任务要说明“谁有能力做”Task 已经有目标和依赖。团队环境还需要能力约束和结构化结果from dataclasses import dataclass, fielddataclassclass TaskRouting:required_capabilities: list[str] field(default_factorylist)dataclassclass TaskResult:summary: strverification: list[str] field(default_factorylist)risks: list[str] field(default_factorylist)handoff: str 然后为上一章的Task增加routing并把原来的字符串结果升级为结构化TaskResultrouting: TaskRouting field(default_factoryTaskRouting)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(funknown 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(storestore,tasks{item.id: item for item in tasks},tasktask,agent_idagent_id,claim_tokenclaim_token,lease_secondslease_seconds,)这里把第 13 章的认领代码提取成claim_task_unlocked()避免持有文件锁时再次进入同一把锁。模型不能在参数里声明 capabilities。Harness 根据当前运行身份从 Registry 读取否则任何成员都可以自称拥有deployment或integration能力。Dispatcher 只负责叫醒Task Store 决定归属团队调度可以选择 Push也可以选择 PullPushLead 指定某个成员容易让 Lead 成为瓶颈Pull空闲成员自己找 Task但会出现认领竞争。更稳妥的第一版是混合模式Dispatcher 根据 capabilities 找到可能合适的空闲成员它发送“有工作可领”的唤醒事件成员醒来后重新读取 Task Store最终所有权仍通过task_claim原子竞争。唤醒只是提示租约才是事实。两个 Backend Agent 即使同时被叫醒也只有一个能认领同一 Task。认领失败不是模型服务故障不需要指数退避成员重新查询任务图即可。第一版最好让每个成员同时只持有一个 Task。一个成员同时维护多个 Task Session、租约和后台 Job会显著增加恢复难度。需要更多并行时增加成员通常比让一个成员频繁切换现场更清楚。Message Bus 不是一间热闹的群聊Task 依赖能表达“等谁完成”却无法覆盖所有协作Frontend 想提前确认接口字段QA 发现验收条件不清楚Backend 被外部服务阻塞需要通知 LeadDesign 想查看一张截图。这些信息适合 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_factorylist)created_at: str acknowledged_at: str | None Noneschema_version: int 1def to_dict(self) - dict:return asdict(self)完整日志、代码 diff 和截图放进 Artifact StoreMessage 只传摘要与路径。否则团队人数一多消息会迅速变成另一份无法压缩的大上下文。写入后响应丢了不能再送一封新信工具调用可能已经把消息写入磁盘却在返回模型前中断。如果重试时生成随机 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 fmessage-{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(idteam_message_id(sender_id,invocation_id,),sender_idsender_id,recipient_idrecipient_id,kindkind,bodyredact_sensitive_text(body.strip()),task_idtask_id,created_atutc_now_text(),)stored bus.put_if_absent(message)wake_queues[recipient_id].put({kind: team_message_ready,message_id: stored.id,})return storedput_if_absent()必须在 Message Store 锁内完成“检查并创建”。相同 ID、相同业务内容返回原消息内容冲突则报错。消息先落盘再唤醒成员。即使进程在两步之间退出成员启动时扫描未确认邮箱仍然能找到它。消息进入上下文但不会冒充用户Team Message 不是用户输入不应该追加成普通user消息。Prompt Assembler 可以把未确认消息渲染为动态 SectionTeam inbox:- id: message-7f381a0c8d9f4b20from: backendkind: updatetask: task-apibody: API 已完成接口契约见 artifact。artifacts:- artifact://task-api/openapi.json一次模型调用成功后再设置acknowledged_at。如果调用失败消息保持未确认下一轮继续投递。至少一次投递意味着消息可能重复出现。Runtime 用稳定message_id去重工具副作用则使用各自的幂等键。不能依靠模型说“我好像见过这条消息”。一条消息不会解锁下游任务Backend 完成 API 后正确交接顺序是保存 checkpoint 和 Artifact调用task_complete写入结构化结果依赖task-api的测试 Task 自然变成ready给 QA 发送一条简短通知Dispatcher 或消息事件唤醒 QAQA 重新读取 Task Store 并认领测试 Task。实线表示持久化的工作状态虚线消息只负责沟通和加快发现。如果消息先到、Backend Task 仍是in_progressQA 可以提前阅读接口说明却不能强行认领仍被依赖阻塞的 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(timeout30)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_storestate.task_store,profileprofile,claim_tokenclaim_token,)if task is not None:run_task_session(statestate,profileprofile,tasktask,claim_tokenclaim_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 租约过期后变为reclaimableclaim_token阻止旧实例继续写入checkpoint 说明上一次做到哪里Artifact 保存代码、报告和日志持久化邮箱保留未确认消息Job Record 让长操作可以查询而不是盲目重跑。如果 Backend Agent 长期离线只要另一个 Profile 具备backend与apicapability就可以在租约过期后接手。agent_id用于角色、审计和协作不应该把工作永久锁死在某个名字上。如果没有任何 Profile 满足 required capabilitiesDispatcher 应向 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 TeamAgentProfile 固定身份、能力、工具和工作区模式每个 Task 使用独立上下文不让长期身份变成长对话Task Routing 决定谁有资格认领Dispatcher 只负责唤醒Task Store 决定所有权Message Bus 传递短消息Artifact 保存大内容稳定消息 ID 和确认状态支持至少一次投递Task Result 让交接不依赖聊天回放租约、checkpoint 和 Job Record 支持成员故障恢复Lead 管目标与集成不承包所有沟通。多个 Agent 现在可以同时工作了但团队仍缺少一件常被忽视的能力安全地停下来。如果 Lead 直接结束 Backend 线程它可能正写到一半、持有 Task或等待一个仍有副作用的 Job。