多 Agent 协作不是把几个模型连在一起就结束。真正难的地方在主 Agent它要把目标拆成任务板决定哪些信息可以共享哪些信息只能进入某个子 Agent 的私有上下文还要把结果回收到一个可复盘的全局状态里。主从 Agent 协作的架构责任单 Agent 长任务里很多失败看起来像“上下文不够”或“模型不够强”。换成主从 Agent 以后问题没有消失只是变成了更清楚的工程责任谁保留全局状态谁看到哪些背景谁判断结果可不可以进入下一步。单 Agent 长任务的问题不只是上下文长短一个 Agent 一直循环最常见的不是“完全不会做”而是中途开始混乱一段上下文里有用户目标、工具日志、失败尝试、半成品结论、模型自我解释。后续步骤看到这些内容很难判断哪些是事实哪些只是临时推理。单Agent 一轮对话的数据流程图单Agent所有中间过程都进入同一条会话问题单 Agent 里的表现主从调度里的处理方式上下文污染所有中间过程都进入同一条会话子 Agent 只返回公开结果私有过程不回流全局状态发散任务做到哪一步主要靠模型“感觉”主 Agent 把状态写进任务板完成证据不足子任务说完成但没有会话、结果和验收记录回收session_id、结果文本和状态变化失败边界模糊一个子问题失败后主会话继续堆信息失败留在具体任务卡上主 Agent 决定重试或人工确认主 Agent 调度多 Agent 的责任边界共享的是公共任务状态和结果摘要进入子 Agent 之前主 Agent 会把任务卡裁剪成私有任务包。主Agent全局任务信息管理任务拆解选择合适的承接子Agent分发任务信息回收任务信息决定下一步推进动作子Agent承接由主Agent输入的任务 任务信息用“恰当”的方式进行任务处理确保对任务的输出能够反馈给主Agent。两个公开案例给出的边界OpenAI Codex后台任务需要独立环境和可追踪证据。OpenAI 在 Codex 发布说明里把 Codex 描述为可以并行处理多个任务的云端软件工程 Agent每个任务独立运行在隔离环境里完成后向用户提供变更、终端日志和测试输出等证据。这个案例说明多任务不是共享一条巨大上下文而是把任务、环境和证据分开管理。Hermes子 Agent 隔离上下文Kanban 承载跨 Agent 状态。Hermes 的 Subagent Delegation 文档写得很直接子 Agent 从全新 conversation 开始只知道父 Agent 放进goal和context的内容Kanban 文档则把多 Agent 协作放到持久任务板里任务、依赖、评论和结果都成为外部状态。这些案例不能推出“所有任务都应该多 Agent”。它们共同说明的是一旦拆成多个执行者系统就要明确状态、上下文、权限和证据的归属。主从 Agent 方案本文的任务板由主 Agent 根据业务目标动态生成主 Agent 决定几张卡、每张卡叫什么、交给哪个 Codex worker、依赖关系怎么连。宿主代码不补默认任务也不替模型硬排三段流程它只做真实系统需要的护栏校验任务数量范围、子 Agent 白名单、task_id格式、依赖存在性、DAG 无环和至少一条协作依赖边。层负责什么不负责什么主 Agent理解目标、动态生成任务板、回收结果、综合输出不直接执行子任务也不绕过结构护栏TriggerFlow保存board、暴露阶段事件、关闭执行不替模型做语义规划也不保存私有推理过程公共任务板保存任务状态、依赖、会话证据和最终结果不保存子 Agent 的完整私有过程私有任务包给某个子 Agent 的goal context upstream_results不携带完整主会话和其他无关卡片细节ACP 边界连接两个官方 Codex ACP worker收流式 update不拆任务、不验收业务结果子 Agent完成一张任务卡返回可回收摘要不管理全局计划杭州三日游团建从业务输⼊到可回收结果杭州三日游团建案例里主 Agent 先生成任务板再把每张卡片裁剪成私有任务包最后回收结果并综合给组织者。ACP 简介把本地 Agent 接成标准会话主从 Agent 方案里主 Agent 不应该知道 Codex、Claude Agent SDK、Gemini CLI 这些运行时各自怎么启动、怎么流式输出、怎么申请权限。它需要的是一个更窄的连接形状能握手、能建会话、能发 prompt、能收到 update、能保留session_id。ACP 连接本地 Codex主 Agent 只需要⾯对标准会话ACP 在这里像一段标准连接线。Codex 仍然在本机运行认证、模型请求和工具事件仍由 Codex 自己处理主 Agent 侧只面对 ACP 的initialize - new_session新会话 - prompt单次请求 - session/update对会话进行补充生命周期。ACP 只统一连接不替代主 Agent 编排换成工程视角ACP 解决的是“怎么和一个 Agent 说话”不是“怎么决定让谁做什么”。这点一旦分清后面的代码就不会混乱层次ACP 负责主 Agent / Harness 负责连接启动 Agent 进程完成initialize选择要调用哪个 Agent会话创建session_id承载一轮或多轮prompt决定一个任务卡是否复用同一会话输出接收session/update、文本增量、停止原因把公开结果写回任务板权限把工具权限请求转给 Client采用允许、拒绝、人工审批等策略编排不负责 DAG、重试、验收和综合管任务板、依赖、失败边界和最终交付这也是为什么完整实战里主 Agent 仍然要维护TaskCard、WorkerTaskInput和 TriggerFlow state。ACP 只是把本地 Codex 接进来不会替主 Agent 生成任务板。连接本地 Codex 的最简案例先看一段最小代码。它只做一件事启动官方 Codex ACP adapter建立一个 session发一轮很短的 prompt然后打印回收文本和session_id。这段代码不引入任务板也不做上下文裁剪目的只是把 ACP 生命周期跑清楚。# ACP 最简连接启动官方 Codex adapter发送一轮 prompt importjsonimportosimportshutilfrompathlibimportPathfromtypingimportAny,castfromacpimportPROTOCOL_VERSION,Client,spawn_agent_process,text_blockfromacp.schemaimport(ClientCapabilities,DeniedOutcome,Implementation,PermissionOption,ReadTextFileResponse,RequestPermissionResponse,ToolCallUpdate,)# Notebook kernel 的 PATH 不一定包含 Codex.app 内置命令。# 这里把 Codex.app 的资源目录补进子进程环境npx 启动 adapter 时也能找到本地 Codex。INTRO_LESSON_DIRPath.cwd()ifnot(INTRO_LESSON_DIR/materials/product_goal.txt).is_file():INTRO_LESSON_DIRPath(harness_class/lessons/Multi_Agent_Orchestration).resolve()INTRO_ACP_ENVos.environ.copy()INTRO_CODEX_BIN_DIRPath(/Applications/Codex.app/Contents/Resources)ifINTRO_CODEX_BIN_DIR.is_dir():INTRO_ACP_ENV[PATH]f{INTRO_CODEX_BIN_DIR}:{INTRO_ACP_ENV.get(PATH,)}# 像MCP一样进行连接这里直接连接codex的ACPINTRO_CODEX_SPEC{name:codex,command:os.getenv(ACP_NPX_COMMAND,npx),args:(-y,agentclientprotocol/codex-acp1.1.0),}ifshutil.which(str(INTRO_CODEX_SPEC[command]),pathINTRO_ACP_ENV.get(PATH))isNone:raiseRuntimeError(当前 Notebook kernel 找不到 npx请从已配置 Node 的 shell 启动 Jupyter kernel。)print(INTRO_CODEX_SPEC)最小 Client 只做三件事收文本、拒绝副作用权限、暴露最终文本。这里故意不写复杂封装后面的完整实战会在这个基础上增加流式转发、任务板回写和多轮会话。这里必须按照这样写要不然会与codex的ACP Adapter对不上导致没办法正常使用。# ACP 最小 Client收文本拒绝副作用权限 classIntroCodexClient:def__init__(self)-None:self.text_chunks:list[str][]self.update_kinds:list[str][]# 会话的消息更新asyncdefsession_update(self,session_id:str,update:Any,**kwargs:Any)-None:update_kindstr(getattr(update,session_update,))self.update_kinds.append(update_kind)contentgetattr(update,content,None)textgetattr(content,text,None)ifupdate_kindagent_message_chunkandgetattr(content,type,None)textandisinstance(text,str):self.text_chunks.append(text)# 是否请求的权限asyncdefrequest_permission(self,options:list[PermissionOption],session_id:str,tool_call:ToolCallUpdate,**kwargs:Any,)-RequestPermissionResponse:# 这个最简案例只验证连接不允许子 Agent 读写文件或执行命令。returnRequestPermissionResponse(outcomeDeniedOutcome(outcomecancelled))asyncdefwrite_text_file(self,**kwargs:Any)-None:raiseRuntimeError(最简 ACP Client 没有声明文件写入能力)asyncdefread_text_file(self,**kwargs:Any)-ReadTextFileResponse:raiseRuntimeError(最简 ACP Client 没有声明文件读取能力)propertydeftext(self)-str:return.join(self.text_chunks).strip()运行这段代码时spawn_agent_process()会启动npx -y agentclientprotocol/codex-acp1.1.0。initialize()交换协议版本和客户端信息new_session()生成会话prompt()把文本任务发给同一个会话。# ACP 最小运行initialize - new_session - prompt asyncdefrun_intro_codex_prompt(prompt_text:str)-dict[str,object]:# 这是第二节的client代码clientIntroCodexClient()# 这是第一节的代码连接codex的acpcommandstr(INTRO_CODEX_SPEC[command])argscast(tuple[str,...],INTRO_CODEX_SPEC[args])# 创建一个agent连接进程asyncwithspawn_agent_process(cast(Client,client),command,*args,cwdINTRO_LESSON_DIR,envINTRO_ACP_ENV,)as(connection,_process):# 初始化连接initawaitconnection.initialize(protocol_versionPROTOCOL_VERSION,client_capabilitiesClientCapabilities(),client_infoImplementation(namecourse-acp-intro,titleACP 简介最小 Client,version0.1.0),)# 从连接中创建一个新的sessionsessionawaitconnection.new_session(cwdstr(INTRO_LESSON_DIR),mcp_servers[])# 向指定的session id发送新的指令responseawaitconnection.prompt(prompt[text_block(prompt_text)],session_idsession.session_id,)agent_infoinit.agent_info.nameifinit.agent_infoisnotNoneelse(未提供 agentInfo)return{agent_info:agent_info,session_id:session.session_id,stop_reason:str(response.stop_reason),update_kinds:sorted(set(client.update_kinds)),text:client.text,}intro_codex_resultawaitrun_intro_codex_prompt(只回复 ACP_CODEX_OK。不要读取文件不要运行命令不要修改文件。)print(json.dumps(intro_codex_result,ensure_asciiFalse,indent2))PS这段最小代码已经具备 ACP 的核心动作但它还不是多 Agent 协作。它没有任务拆解没有依赖顺序没有上下文裁剪也没有结果验收。后面的完整实战会把这段连接能力放进主 Agent 的任务板调度里。Spec dict 示例与验证状态ACP Registry 里登记了很多可用 Agent但“有启动形状”不等于“本机已经可用”。AgentSpec只描述主 Agent 怎么启动一个外部执行单元真实可用还取决于 CLI 是否安装、是否完成登录、是否有模型额度。先看官方 adapter 或 Registry 里有明确启动形状的 Agent。这类可以写成比较稳定的Spec dict示例Agent来源Spec dict 示例CodexACP Registry / Codex adapter{name: codex, command: npx, args: (-y, agentclientprotocol/codex-acp1.1.0)}Claude AgentACP Registry / Claude adapter{name: claude_agent, command: npx, args: (-y, agentclientprotocol/claude-agent-acp0.55.0)}Gemini CLIACP Registry{name: gemini, command: npx, args: (-y, google/gemini-cli0.49.0, --acp)}Qoder CLIACP Registry{name: qoder, command: npx, args: (-y, qoder-ai/qodercli0.2.14, --acp)}还有一些 Agent 支持 ACP但启动参数依赖具体网关、配置文件、安装方式或运行环境。这里不写成可复制 Spec只给官方文档入口Agent官方文档课程口径OpenClawOpenClaw ACP 文档Gateway-backed bridge通常需要配置网关地址、token、session 等参数Hermes AgentHermes ACP 文档支持hermes acp、hermes-acp、python -m acp_adapter等方式按安装方式选择CursorCursor ACP 文档需要本机安装 Cursor CLI本文完整实战只使用两个 Codex profile是为了让课堂案例可以真实跑通同时保留“两个独立子 Agent 执行域”的效果。其它 Agent 适合课后按文档逐个接入和验证。ACP 和 A2A服务内调用和跨边界协作不是一回事同样是“Agent 之间通信”工程场景其实差很多。主 Agent 在同一个服务模块里维护任务板知道全局目标、依赖关系、上下文裁剪规则和最终验收方式。它只是需要把某张任务卡交给本地 Codex worker 执行再把结果写回任务板。这个 worker 不需要主动发现别的 Agent也不需要和 HR、财务、供应商系统协商任务它只要接收 prompt、输出文本、返回 session/update。这类场景更像“服务内部的主从调度”。主 Agent 是调度中心子 Agent 是执行单元。通信要轻、要直接、要容易观察。普通函数、内部 API、消息队列都可以承担服务内模块通信如果执行单元本身是一个 coding agent 进程ACP 就很贴合启动进程、建 session、发 prompt、收流式 update。A2A 面向的是另一类场景。比如团建任务继续往后推进可能会出现这些独立系统HR Agent确认人数、部门、员工偏好和审批规则财务 Agent确认预算、报销标准和付款限制供应商 Agent提供酒店、餐饮、车辆报价法务 Agent检查合同条款和风险边界。这些 Agent 可能来自不同团队、不同 SaaS、不同供应商彼此不共享进程、数据库、记忆和工具。它们需要先知道“对方能做什么”再交换任务、产物和状态同时保留各自系统边界。A2A 的设计重点更接近这种跨服务、跨团队甚至公开市场里的 Agent 发现与协作。用手机APP来说就像美团点外卖付款会跳转到支付宝APP进行支付支付完成后支付宝会把结果告知美团。问题服务模块内主从调度跨服务 / 跨团队 Agent 协作谁掌握全局任务状态主 Agent / Harness多个独立 Agent 各管一部分子 Agent 是否需要被公开发现不需要主 Agent 已经知道调用谁需要外部 Agent 要能看到能力描述通信对象本地 worker、coding agent、内部执行单元SaaS Agent、企业系统 Agent、供应商 Agent关注重点轻量连接、上下文裁剪、流式输出、结果回收能力发现、身份边界、任务协商、产物交换更贴近的协议ACP 或内部 API / QueueA2A所以不用 A2A不是因为 A2A 没价值而是因为它不是为这种服务模块内的轻量通讯设计的。把 A2A 放到项目内需要引入 agent discovery、capability card、远程服务边界、任务协商等概念反而遮住主线主 Agent 怎样拆任务、裁剪上下文、调度本地 worker、回收结果。海外实际落地也基本沿着这个分工展开。ACP更集中在开发者工具和 coding agent 接入Zed、JetBrains 通过 Registry 发现和安装 coding agentsRegistry 里能看到 Claude Agent、Codex CLI、Gemini CLI、GitHub Copilot、Goose、OpenCode、Qwen Code 等代表项目。A2A更偏企业协作网络Google ADK / A2A samples、LangGraph 或 BeeAI agent 暴露成 A2A server以及 Salesforce、SAP、ServiceNow、Workday、UiPath 这类企业软件或自动化平台围绕跨系统 Agent 协作做互操作。简而言之主 Agent 调本地 Codex worker用 ACP未来如果要让 HR Agent、财务 Agent、供应商 Agent、法务 Agent 跨系统互相发现能力并交换任务就用 A2A。实战用主 Agent 骨架跑通调度 Loop步骤1. 先创建主 Agent 骨架主 Agent 在这节课里由两部分组成Agently 负责模型请求TriggerFlow 负责流程编排和执行状态。先把骨架搭出来后面的任务板、ACP 和综合报告都挂到这两个对象上。业务输入只保留用户目标业务目标只描述杭州三日游团建本身调度规则属于主 Agent 代码不写进用户需求。下面代码还没有 ACP也没有子 Agent。主 Agent 的第一项责任是拥有一个稳定的“模型请求入口”和一个稳定的“流程状态入口”。步骤 2. 定义任务板、私有输入和主 Agent 决策主 Agent 要管理的不是一段聊天记录而是一张公共任务板。每张卡片只保存可共享状态真正发给子 Agent 的内容会裁剪成私有任务包。三类数据契约划清协作边界TaskCard是公共状态WorkerTaskInput是私有输入WorkerResult是回收结果。先定义三类数据。TaskCard留在公共任务板上WorkerTaskInput发给单个子 AgentWorkerResult是回收到主 Agent 的公开结果。decide_board_action() 是主 Agent 判断任务是否完成、是否继续推进、是否重试、是否阻塞的⼊⼝。它只读公共任务板不依赖 ACP也不依赖⼦ Agent的私有过程。有了数据结构以后再定义主 Agent 怎么“读任务板”。这一步还不调用模型也不调用 ACP只做可解释的状态判断。调度判断之外还需要几种状态更新动作。它们都只改当前任务卡不碰其它卡片的私有信息。步骤 3. 主 Agent 动态生成公共任务板杭州三日游团建不是固定三步流程。不同组织者可能更关注预算、交通、活动、天气或安全主 Agent 应该根据业务目标自己决定拆成几张卡、哪些卡先做、哪些卡依赖上游结果。主 Agent 动态⽣成任务板模型负责真实拆解宿主代码只负责把越界输出挡在执行边界之外。任务板生成分三步先确认主 Agent 能调用哪些 worker再让模型返回动态任务板最后做结构护栏校验。这里没有备用任务板模型没有给出可执行任务板时程序应该失败并暴露原因而不是悄悄换成预置流程。边界确认后再让主 Agent 真实拆解任务。输出契约仍然保持简单一组TaskCard建议每张卡说明局部目标、可见上下文、负责的 worker 和依赖关系。模型给出的动态任务板可以直接体现业务判断但进入执行前还要过一层护栏。注意这里的校验只处理结构问题有没有字段、worker 是否存在、依赖是否可执行它不把业务拆解替换成预设答案。现在真正跑一次建板。观察重点不是模型写得多漂亮而是公共任务板是否能支持后续调度。此时还没有调用任何子 Agent。主 Agent 先把杭州三日游团建目标变成可跟踪的公共状态再判断第一张 ready 卡片。步骤 4. 引入 ACP把私有任务包发给官方子 Agent现在主 Agent 已经知道下一张卡片是谁、依赖是否满足、失败后是否该重试。接下来才需要 ACP把一张裁剪后的WorkerTaskInput发给本地 Codex 官方连接器并回收公开文本结果。ACP的位置连接外部子Agent不接管业务编排ACP 负责连接、会话和流式 update任务拆解和完成判断仍属于主 Agent。进入 ACP 前先把“主 Agent 可以调谁”写清楚。这里两个执行者都走官方 Codex ACP 连接器但主 Agent 用不同 profile 承担不同任务。注册表解决“能调谁”私有任务包解决“给它看什么”。这一步最能体现上下文隔离。先不急着调用 ACP。把第一张 ready 卡片裁剪出来直接看子 Agent 会收到哪些信息。任务包确认以后再引入 ACP。ACP 只负责连接和会话事件不负责判断任务是否该派发。ACP 的时序从 Client 开始。这个 Client 不自建子 Agent只负责接收官方连接器推来的 update并把公开文本整理出来。Client 负责接 updateSession 负责完整连接生命周期启动连接器、初始化、建会话、发送 prompt。正式派发业务任务前先用最小 prompt 做一次连接 smoke。这样能把“连接问题”和“业务调度问题”分开排查。ACP 会话可以多轮但主 Agent 不会把完整主会话倒给子 Agent。是否追问、重试或收尾仍由任务板状态和decide_board_action()决定。步骤 5. 用 TriggerFlow 串起主 Agent Loop到这里主 Agent 已经具备四个能力生成任务板、观察任务板、调用外部子 Agent、更新任务状态。现在把它们放进 TriggerFlow 执行里让完整过程可观察、可关闭、可复盘。TriggerFlow的chunk/when/execution结构这段代码按 TriggerFlow 的原生结构展开master_flow.chunk(...)定义阶段master_flow.when(...).to(...)说明阶段关系create_execution()启动一次可观察的执行。长任务的等待体验不靠设置更长的等待上限来解决。主 Agent 要把每个子 Agent 的文本增量转成worker_delta事件让外部观察者持续看到“哪张卡正在推进、子 Agent 已经说了什么”。TriggerFlow 接管的是执行过程里的公共状态。先定义三个状态工具函数读任务板、写任务板、追加 worker 输入/结果记录。后面的 chunk 都只通过这些函数读写state[board]避免每个代码块各自拼装状态结构。第一个 chunk 只做建板。它把ProjectBrief交给主 Agent让模型真实决定任务数量、任务内容、worker 分配和依赖关系然后把公共任务板写入 TriggerFlow state。第二个 chunk 是主 Agent Loop 的核心读任务板、选 ready 卡、裁剪私有任务包、调用 worker、回写结果。这里的循环是业务调度循环TriggerFlow 负责承载它的状态、事件和执行边界。第三个 chunk 不再调用子 Agent。它只读取已完成任务板把多个子 Agent 的公开结果综合成组织者可读的报告。这样最终交付仍由主 Agent 负责而不是把某个子 Agent 的局部回答直接交出去。三个 chunk 定义完以后再用 TriggerFlow 的to()和when()装配流程。这里不把编排藏进 Python 控制流里master_flow.to(plan_board)execution 启动后先建任务板master_flow.when(plan_board).to(dispatch_tasks)建板完成后进入**调度master_flow.when(dispatch_tasks).to(synthesize_report)所有任务回收后综合报告。最后运行完整案例。这里会同时看到 TriggerFlow 阶段事件和 Codex worker 的文本增量。TriggerFlow 的 runtime stream 不是模型输出流它输出的是本次执行里的业务事件所以我们用get_async_runtime_stream(timeoutNone)持续观察到 execution close。TriggerFlow 的结构现在分成四层master_flow.chunk(...)声明阶段state[board]保存跨阶段事实master_flow.when(...).to(...)描述阶段触发关系execution.get_async_runtime_stream(timeoutNone)暴露可观察事件。dispatch_tasks进入时读取一次 board循环内用当前局部 board 推进任务并在每次 running / done / failed 后写回 state。runtime stream 会持续输出board_created、task_started、worker_delta、task_done、dispatch_done、final_report_ready。这里的重点不是给长任务设一个更久的等待时间而是让执行过程持续产生日志级别的业务事件。步骤 6. 检查子 Agent 实际收到的信息完整运行结束后检查worker_inputs最能说明上下文隔离。下游卡片只能看到自己显式依赖的上游公开结果摘要看不到完整主会话也看不到其它卡片的私有过程。派发前裁剪上下文公共任务板里的全局信息不会原样传给子 Agent主 Agent 只发当前任务需要的子集。完整运行之后再回看 worker_inputs能验证上下文隔离是不是实际发生了而不是只停留在设计图上。最后把 Notebook 的运行结果保存成 JSON。这个文件用于对照观察不是主流程依赖。主从 Agent 协作总结主从 Agent 协作的重点不是“多几个 Agent”。真正的工程收益来自几条清楚的边界主 Agent 管全局任务状态不靠对话里的感觉记进度子 Agent 只拿私有任务包不继承完整主会话decide_board_action()让继续、重试、阻塞、结束都有状态依据ACP 负责连接和会话不承担业务编排TriggerFlow state 承载主 Agent Loop让阶段、事件和最终状态可观察最终交付由主 Agent 读任务板后综合而不是把某个子 Agent 的回答直接当最终答案。