多agent系统重构:进程内 agent + codex V2 架构
背景
当前多agent系统(packages/core/src/tools/tools/collab.ts)是进程级 subagent:每次 spawn_agent 启动一个独立 memo --dangerous 子进程(stdin 单条消息 → stdout 收集输出),模块级全局 Map 注册表,5 个工具(spawn_agent / send_input / resume_agent / wait / close_agent),状态机 running/completed/errored/closed,wait 每 100ms 轮询。
痛点
- 无状态通信:每次 submission = 全新进程全新会话,resume 不恢复上下文
- 无上下文继承:子 agent 只能从消息里获得上下文
- wait 轮询而非事件驱动
- 注册表无回收:模块级 Map 只增不减、跨 session 共享
- agent 状态不进事件流、不进 TUI:用户看不到子 agent 活动
参考:OpenAI Codex 的 multi-agent 演进
当前实现与 codex 的 V0 subagents 阶段同构。codex 经 V0(子进程) → V1(多agent) → V2(生命周期重构) 演进,最终架构洞见:agent 不是特殊对象,而是"带 mailbox 的 run loop 实例",复杂度收敛在三个机制:
- 并发预留(
AgentRegistry + SpawnReservation RAII:树作用域注册表,spawn 异常自动回滚)
- 消息投递(mailbox + watch channel:wait 事件驱动而非轮询,
InterAgentCommunication{author, recipient, trigger_turn})
- 状态派生(
AgentStatus 由事件推导而非手写更新:TurnStarted→Running、TurnComplete→Completed、TurnAborted(Interrupted)→Interrupted、Error→Errored、ShutdownComplete→Shutdown)
关键经验:V2 用 interrupt_agent 取代 close_agent(打断当前 turn,agent 保留可恢复);fork 上下文继承(子 agent 从父历史复制 system/user/最终答案,剥掉工具调用与推理);begin/end 成对的结构化 collab 事件(带 call_id、End 带最终 status 快照,TUI 可直接渲染)。
已确认的决策
- 进程内 agent:子 agent = 进程内
AgentSessionImpl 实例(复用现有 ReAct loop,核心零改动)
- codex V2 工具集:spawn_agent / send_message / followup_task / wait / interrupt_agent / list_agents
- 范围:core 完整重构 + TUI 展示 agent 状态
目标架构
AgentControl(每 root 树一个,克隆共享)
├── registry: AgentRegistry(agent_tree by path + thread_paths by id,树作用域)
├── limiter: ExecutionLimiter(非阻塞并发上限,超限报错)
├── activity: ActivityBus(watch channel,wait 事件驱动)
├── depthLimit(默认 3,MEMO_SUBAGENT_MAX_DEPTH)
└── shutdownSubtree(path, reason) ← 根 close 时级联
AgentRuntime(每个子 agent 一个,包装 AgentSessionImpl)
└── submissionLoop: while(true) {
mailbox.takeTrigger() → slot = limiter.tryAcquire() → session.runTurn(joined)
→ 状态映射 → activity.publish() }
生命周期状态机
spawn_agent(预扣 slot)→ pending_init → running → completed / interrupted / errored
▲ │
└── send_message(trigger_turn) / followup_task(同会话续跑,历史保留)
任何状态 → 根 close / shutdownSubtree → shutdown(loop 退出 + session.close)
- interrupt 不退出 loop:
cancelCurrentTurn() → turn 以 cancelled 结束 → 状态 interrupted → loop 继续等消息;resume = 下一条 trigger 消息,同进程同 history
- is_final:completed | errored | shutdown
- 状态映射:
turn_end.ok → completed、turn_end.cancelled → interrupted、turn_end.error/prompt_limit → errored、collab_spawn_begin → pending_init、collab_shutdown → shutdown
- 预留 RAII 化:
slot.release() + try/finally,spawn 中途异常自动释放
文件级改动
新建(core)
| 文件 |
职责 |
packages/core/src/agent/status.ts |
AgentStatus(6 态)、isFinalAgentStatus、deriveAgentStatusFromEvents(JSONL 回放用) |
packages/core/src/agent/registry.ts |
AgentPath 工具函数(join/depth/resolve)、AgentRegistry(register/listSubtree/allocChildName)、ExecutionLimiter |
packages/core/src/agent/communication.ts |
Mailbox(trigger/queue-only + shutdown)、ActivityBus(事件驱动 wait) |
packages/core/src/agent/fork.ts |
buildForkHistory(上下文继承过滤)、buildSubagentSystemPrompt |
packages/core/src/agent/runtime.ts |
AgentControl + AgentRuntime(submission loop) |
packages/core/src/agent/subagent.ts |
createSubagentSession(共享父 deps 直接构造) |
重写 / 修改(core)
tools/tools/collab.ts:整体重写为 6 个 V2 工具,状态读自 experimental_context.collab(不再 spawn 进程、无模块级 Map)
types.ts:CollabEventType 加入 HistoryEventType;onAssistantStep/onReasoningChunk 加可选第 3 参 sessionId?
agent/loop.ts:构造器加 collab?: CollabSessionBinding;toolContext.collab 注入;emitEvent 合并 eventMeta(agent_id/agent_path);closeInternal 级联 shutdownSubtree
agent/defaults.ts:导出 ResolvedSessionDeps
agent/session.ts:根会话创建 root AgentControl(MEMO_ENABLE_COLLAB_TOOLS 门控)
tools/index.ts、tools/approval/constants.ts、index.ts、prompt/prompt.md:注册与文档更新
修改(TUI)
shared/types/index.ts:AgentActivityView
features/timeline/chatTimeline.ts:state 加 agents + agent_status action
features/timeline/Cells.tsx:AgentCell(状态 glyph + path + lastMessage 预览)
features/timeline/ChatWidget.tsx、app/App.tsx(hooks 按 sessionId 路由)、shared/ui/Footer.tsx
runtimeState.ts / approvalQueue.ts 零改动(审批队列已支持多 agent 排队)。
V2 工具语义
| 工具 |
语义 |
| spawn_agent |
{message, agent_type?, fork_turns?}:深度检查 → collab_spawn_begin → limiter 预扣 → fork 上下文 → 注册(pending_init)→ runtime start |
| send_message |
{agent_id, message, trigger_turn?}:QueueOnly(入 mailbox 累积) |
| followup_task |
{agent_id, task}:等价 trigger_turn=true |
| wait |
{ids, timeout_ms?}:订阅 ActivityBus 事件驱动(全 final / 新消息 steered / 超时) |
| interrupt_agent |
cancelCurrentTurn() → interrupted(幂等,不销毁) |
| list_agents |
{path?}:按 path 前缀列子树 |
全部 ALWAYS_AUTO_APPROVE_TOOLS;wait/list_agents 只读可并行,其余 mutating 串行化(step_gate)。
上下文继承(fork)
buildForkHistory 过滤规则(对齐 codex keep_forked_rollout_item):
- system 丢弃(子用自己的 prompt);user 保留(forkTurns=N 截取最后 N 轮;compact summary 保留)
- assistant:含 tool-call 的丢弃;纯 text/reasoning 保留(最终答案通常为最后一个);reasoning part 剥掉
- tool role 丢弃
实施步骤(每步独立可测)
agent/status.ts + types.ts 事件 type → status 单测
agent/registry.ts(path 解析、allocChildName、limiter)→ 单测
agent/communication.ts(Mailbox、ActivityBus)→ 单测
agent/fork.ts → 单测
- loop.ts binding + session.ts root control +
subagent.ts + runtime.ts → 集成测试(脚本化 callLLM + 进程内子会话跑一轮)
- 重写
collab.ts + approval/constants.ts + tools/index.ts → 重写 collab.test.ts(fake callLLM 集成式,保留语义:spawn+wait 到 completed、超限报错、wait 未知 id、timeout、interrupt 后恢复)
- prompt.md 指南更新
- TUI:types + chatTimeline + AgentCell + App.tsx 路由 + Footer → chatTimeline.test.ts 补 agent_status 用例
- 端到端:
pnpm run format:check && pnpm run test:coverage && pnpm run build + 手动 TUI 验证
兼容与迁移
| 旧 |
新 |
| spawn_agent |
spawn_agent(同名前向兼容,+fork_turns) |
| send_input |
send_message(改名,+trigger_turn) |
| resume_agent |
删除(resume = send_message(trigger_turn=true) 同会话续跑) |
| wait |
wait(同名字段兼容,事件驱动,+steered) |
| close_agent |
interrupt_agent(保留 agent 可恢复;根 close 才真正 shutdown) |
| — |
followup_task / list_agents(新增) |
| MEMO_SUBAGENT_COMMAND |
删除(不再 spawn 进程) |
| MEMO_SUBAGENT_MAX_AGENTS |
保留(语义变为树内并发 running turn 数) |
| MEMO_ENABLE_COLLAB_TOOLS |
保留(门控不变) |
| — |
MEMO_SUBAGENT_MAX_DEPTH 新增(默认 3) |
已知取舍(接受)
- interrupt 时已执行的工具结果仍会回流历史(只 abort LLM 调用)
- limiter 非阻塞"超限即报错"(规避父 wait 持 slot 死锁;与旧行为一致)
- 子事件双写子 JSONL + 父 JSONL(根日志体积膨胀;回放按 sessionId 分流,historyParser v1 只渲染 root 为已知限制)
agent_type 参数接受但忽略(为 codex schema 兼容预留)
多agent系统重构:进程内 agent + codex V2 架构
背景
当前多agent系统(
packages/core/src/tools/tools/collab.ts)是进程级 subagent:每次spawn_agent启动一个独立memo --dangerous子进程(stdin 单条消息 → stdout 收集输出),模块级全局 Map 注册表,5 个工具(spawn_agent / send_input / resume_agent / wait / close_agent),状态机running/completed/errored/closed,wait 每 100ms 轮询。痛点
参考:OpenAI Codex 的 multi-agent 演进
当前实现与 codex 的 V0 subagents 阶段同构。codex 经 V0(子进程) → V1(多agent) → V2(生命周期重构) 演进,最终架构洞见:agent 不是特殊对象,而是"带 mailbox 的 run loop 实例",复杂度收敛在三个机制:
AgentRegistry+SpawnReservationRAII:树作用域注册表,spawn 异常自动回滚)InterAgentCommunication{author, recipient, trigger_turn})AgentStatus由事件推导而非手写更新:TurnStarted→Running、TurnComplete→Completed、TurnAborted(Interrupted)→Interrupted、Error→Errored、ShutdownComplete→Shutdown)关键经验:V2 用
interrupt_agent取代close_agent(打断当前 turn,agent 保留可恢复);fork 上下文继承(子 agent 从父历史复制 system/user/最终答案,剥掉工具调用与推理);begin/end 成对的结构化 collab 事件(带 call_id、End 带最终 status 快照,TUI 可直接渲染)。已确认的决策
AgentSessionImpl实例(复用现有 ReAct loop,核心零改动)目标架构
生命周期状态机
cancelCurrentTurn()→ turn 以 cancelled 结束 → 状态 interrupted → loop 继续等消息;resume = 下一条 trigger 消息,同进程同 historyturn_end.ok → completed、turn_end.cancelled → interrupted、turn_end.error/prompt_limit → errored、collab_spawn_begin → pending_init、collab_shutdown → shutdownslot.release()+ try/finally,spawn 中途异常自动释放文件级改动
新建(core)
packages/core/src/agent/status.tsAgentStatus(6 态)、isFinalAgentStatus、deriveAgentStatusFromEvents(JSONL 回放用)packages/core/src/agent/registry.tsAgentPath工具函数(join/depth/resolve)、AgentRegistry(register/listSubtree/allocChildName)、ExecutionLimiterpackages/core/src/agent/communication.tsMailbox(trigger/queue-only + shutdown)、ActivityBus(事件驱动 wait)packages/core/src/agent/fork.tsbuildForkHistory(上下文继承过滤)、buildSubagentSystemPromptpackages/core/src/agent/runtime.tsAgentControl+AgentRuntime(submission loop)packages/core/src/agent/subagent.tscreateSubagentSession(共享父 deps 直接构造)重写 / 修改(core)
tools/tools/collab.ts:整体重写为 6 个 V2 工具,状态读自experimental_context.collab(不再 spawn 进程、无模块级 Map)types.ts:CollabEventType加入HistoryEventType;onAssistantStep/onReasoningChunk加可选第 3 参sessionId?agent/loop.ts:构造器加collab?: CollabSessionBinding;toolContext.collab注入;emitEvent合并 eventMeta(agent_id/agent_path);closeInternal级联shutdownSubtreeagent/defaults.ts:导出ResolvedSessionDepsagent/session.ts:根会话创建 rootAgentControl(MEMO_ENABLE_COLLAB_TOOLS 门控)tools/index.ts、tools/approval/constants.ts、index.ts、prompt/prompt.md:注册与文档更新修改(TUI)
shared/types/index.ts:AgentActivityViewfeatures/timeline/chatTimeline.ts:state 加agents+agent_statusactionfeatures/timeline/Cells.tsx:AgentCell(状态 glyph + path + lastMessage 预览)features/timeline/ChatWidget.tsx、app/App.tsx(hooks 按 sessionId 路由)、shared/ui/Footer.tsxruntimeState.ts/approvalQueue.ts零改动(审批队列已支持多 agent 排队)。V2 工具语义
{message, agent_type?, fork_turns?}:深度检查 → collab_spawn_begin → limiter 预扣 → fork 上下文 → 注册(pending_init)→ runtime start{agent_id, message, trigger_turn?}:QueueOnly(入 mailbox 累积){agent_id, task}:等价 trigger_turn=true{ids, timeout_ms?}:订阅 ActivityBus 事件驱动(全 final / 新消息 steered / 超时)cancelCurrentTurn()→ interrupted(幂等,不销毁){path?}:按 path 前缀列子树全部
ALWAYS_AUTO_APPROVE_TOOLS;wait/list_agents 只读可并行,其余 mutating 串行化(step_gate)。上下文继承(fork)
buildForkHistory过滤规则(对齐 codexkeep_forked_rollout_item):实施步骤(每步独立可测)
agent/status.ts+ types.ts 事件 type → status 单测agent/registry.ts(path 解析、allocChildName、limiter)→ 单测agent/communication.ts(Mailbox、ActivityBus)→ 单测agent/fork.ts→ 单测subagent.ts+runtime.ts→ 集成测试(脚本化 callLLM + 进程内子会话跑一轮)collab.ts+ approval/constants.ts + tools/index.ts → 重写 collab.test.ts(fake callLLM 集成式,保留语义:spawn+wait 到 completed、超限报错、wait 未知 id、timeout、interrupt 后恢复)pnpm run format:check && pnpm run test:coverage && pnpm run build+ 手动 TUI 验证兼容与迁移
已知取舍(接受)
agent_type参数接受但忽略(为 codex schema 兼容预留)