跳到主要内容

多智能体编排:Stack、委派与 A2A

30 秒导读: 一个 Yao agent 在跑的过程中可以再去"叫"别的 agent。本章讲清两件事: (1)这些嵌套/并行的调用如何被一根 Stack(调用栈) 串成一棵树、彼此隔离又能整体追踪; (2)"叫别人"有两种性质完全不同的姿势——委派(delegation) 把活整个交出去、仍算同一场对话; A2A fork 借别的 agent 当工具、结果拿回来但不许污染主对话历史。

本章只讲"多 agent 如何组织与隔离"。一次请求的主管道生命周期见 02-pipeline.md;Hook 与 JSAPI 的 V8 桥机制见 03-hooks-jsapi.md;工具循环见 04-toolloop-mcp.md。

1. 这是什么(零基础也能懂)​

一句话定义: 在处理一个用户请求的过程中,主 agent 可以再启动别的 agent;Yao 用一根 "调用栈"把这些嵌套/并行的 agent 调用记成一棵树,并规定谁的输出进历史、谁负责收尾。

它解决什么问题? 想象你问主 agent 一句话,它自己答不了,于是:

  • 把问题整个转交给一个更专的 agent 来答(比如路由到"客服 agent");或
  • 同时派三个 agent 去查三个数据源,谁先成功用谁的结果。

这时麻烦来了——它们跑在同一个请求里,共享同一个输出通道、同一段聊天历史、同一份追踪。 如果不管,就会出现三种事故:

事故后果
每个子 agent 都发一遍 stream_start/stream_end前端收到多组"开始/结束",UI 错乱
子 agent 把自己的中间对话也写进聊天历史主对话历史被"内心独白"污染
三个并行 agent 同时改同一个 Stack / Logger数据竞争(race),追踪树错乱

Yao 的答卷: 一个扁平的 Stack 树 + 一条铁律——"只有 root 栈才发流首尾、才存历史、才关输出; 子栈只借 ThreadID 在 UI 上分个道"。

一句话直觉: 把 Stack 想成一叠"报账单",每叫一个 agent 就往上摞一张子单,单子记着 "我是谁的孩子、在第几层、花了多久";请求结束时按 TraceID 把整叠单子捞出来,就是完整的调用树。

2. 顶层全景(它大概怎么转)​

2.1 "叫别的 agent"的两种性质​

Yao 里让 agent 调 agent,底下其实分成两条语义完全不同的路,别混:

路子白话上下文进聊天历史吗Referer 标记并行安全
委派 Delegation把活整个交出去,交出去就不管了共享同一个 Context进(算主对话流)agent串行,无需
A2A fork借别的 agent 当工具,要结果fork 出独立子 Context不进(怕污染)agent_fork并行,靠 fork 保证

判断口诀:"我不干了、你接着答用户" = 委派;"你帮我算一下、我拿结果继续" = A2A fork。

2.2 一棵请求内的 agent 树​

下面这张图是一次请求里可能长出的调用树。怎么读:根在最上,每往下一层是一次"叫别的 agent", 方括号里是这条边的性质。

        [root stack · depth 0]  主 agent (Referer=api)
        │  ← 只有它:发 stream_start/end、InitBuffer/FlushBuffer、CloseOutput
        │
   ┌────┴───────────────────────────────┐
   │ 委派 (delegate)                     │ A2A fork (ctx.agent.All)
   │ 同一个 Context                      │ 各自 fork 出独立 Context
   ▼                                     ▼
[depth 1] 客服 agent               [depth 1] 查库A   查库B   查库C  ← 三个并行子栈
Referer=agent                      Referer=agent_fork(每个)
进主历史、共享 Buffer               不进历史、各自 ThreadID 分道

Stack 只是"账本",不是"执行器"。 真正干活的是每个 agent 自己的 Stream() (见第 2 章)。Stack 做的只是:在 Stream() 一进门就往栈上摞一张单、出门时结账并恢复父栈。 所以要理解多 agent 编排,先看这张单子(§3),再看两种"往栈上摞单"的入口(§4 委派、§5 fork)。

2.3 部件与文件​

部件干什么在哪
Stack 结构 + 生命周期记录单次 agent 调用的身份/父子/耗时agent/context/stack.go、agent/context/types.go:366
EnterStack在 Stream() 开头进栈、返回 done() 收尾闭包agent/context/stack.go:200
handleDelegation委派入口:同 Context 调目标 agentagent/assistant/next.go:43
ctx.agent.* JSAPIA2A 入口:Call/All/Any/Raceagent/context/jsapi_agent.go
OrchestratorA2A 实际执行:fork 上下文、并行编排agent/caller/orchestrator.go
root 栈闸门只在 root 栈发流首尾/存历史/关输出agent/assistant/agent.go、agent/assistant/chat.go

3. 核心机制一:Stack 数据结构与生命周期​

3.1 它要解决的小问题​

要把一棵 agent 调用树追踪清楚,每个节点至少得知道:我是谁、我属于哪次追踪、我父亲是谁、 我在第几层、我从根到我的完整路径、我跑了多久、成没成功。 Stack 就是这么一张扁平的记录。

为什么"扁平"? 注释说得很直白:用扁平结构避免循环引用和内存开销——节点不持有父节点指针, 父子关系只靠 ParentID 字符串和 Path 数组表达(agent/context/types.go:364-366)。

3.2 Stack 的字段​

字段类型含义
IDstring本次调用的唯一 ID
TraceIDstring整棵树共享,从 root 继承——按它能捞出全树
AssistantIDstring处理本次调用的 agent
Refererstring调用来源:api / agent(委派) / agent_fork(A2A) 等
Depthint深度,root = 0
ParentIDstring父栈 ID,root 为空(IsRoot() 就靠它判断)
Path[]string从根到本节点的完整路径 [root_id, ..., this_id]
Statusstringpending/running/completed/failed/timeout
CreatedAt / CompletedAt / DurationMsint64时间戳与耗时(毫秒)

依据:agent/context/types.go:366-394(type Stack struct)。

3.3 三种构造函数 + EnterStack 的三条岔路​

栈节点有三种造法,对应三种"我是怎么被叫起来的":

构造函数用于关键行为
NewStackroot(请求入口)Depth=0、ParentID=""、Path=[stackID];traceID 为空则新生成
NewChildStack嵌套调用(委派 / 单次 Call)继承父 TraceID,Depth+1,ParentID=父.ID,Path=父.Path+新ID
NewChildStackFromForkParentfork 出来的子上下文(A2A 并行)从 ForkParentInfo 取父信息建子栈,不共享真实 Stack 引用以免并行 race

依据:agent/context/stack.go:12(NewStack)、:35(NewChildStack)、:61 (NewChildStackFromForkParent)。

这三条岔路由 EnterStack 统一分流。它是每个 Stream() 一进门就调的第一件正事 (agent/assistant/agent.go:63)。怎么读下图:看 ctx 进来时的两个状态位—— 有没有现成的 ctx.Stack、有没有 ctx.ForkParent——决定走哪条造栈路。

EnterStack(ctx, assistantID, opts)
        │
   ctx.Stack == nil ?
        │
   ┌────┴─────────────────────────┐
  是(全新上下文)                 否(上下文里已有栈)
   │                               │
 ctx.ForkParent != nil ?           └─► NewChildStack  ← 委派 / 单次 Call
   │                                     (同 ctx 内嵌套,父子相连)
 ┌─┴──────────────┐
是               否
│                │
NewChildStack-   NewStack        ← 请求真正入口(API 调用)
FromForkParent   (root, 新 traceID)
(A2A fork 子栈,
 继承父 traceID)

依据:agent/context/stack.go:200-235(EnterStack 的三分支)。

3.4 进栈、结账、恢复父栈——都在 done() 里​

EnterStack 造完栈会做两件事:把栈存进 ctx.Stacks[stack.ID](供事后追踪),然后返回一个 done() 闭包给调用方 defer。这个闭包就是"结账 + 恢复现场":

// 真实源码,agent/context/stack.go:246-256(EnterStack 返回的 done 闭包)
done := func() {
    // 没 panic 的话,标记完成(算耗时)
    if !stack.IsCompleted() {
        stack.Complete()
    }
    // 恢复父栈:把 ctx.Stack 换回进来时的那个
    if parentStack != nil {
        ctx.Stack = parentStack
    }
}

配合 Stream() 里的 stack, traceID, done := context.EnterStack(...); defer done() (agent/assistant/agent.go:63-64),形成对称的进出栈:

  • 进: ctx.Stack 被换成新子栈,后续这段执行"看到的当前栈"就是它。
  • 出: defer done() 触发,Complete() 盖上耗时,再把 ctx.Stack 换回父栈—— 于是父 agent 继续跑时,当前栈又回到自己。

Complete()/Fail(err)/Timeout() 三个收尾方法都盖 CompletedAt、算 DurationMs、 置对应 Status(agent/context/stack.go:85-112)。注意 done() 只在没被显式 Fail 过 (!IsCompleted())时才补 Complete(),所以中途 Fail 的栈不会被覆盖成成功。

3.5 事后检索:按 ID、按 TraceID、找 root​

请求跑完,所有栈都躺在 ctx.Stacks 这张 map 里。Context 上挂了一组检索方法,给追踪日志用:

方法返回用途
GetAllStacks()全部栈请求结束后一把捞出做 trace
GetStackByID(id)单个栈精确查某次调用
GetStacksByTraceID(traceID)同一 trace 的所有栈拼出完整调用树
GetRootStack()IsRoot() 那个找树根

依据:agent/context/stack.go:263-312。辅助:GetPathString() 把 Path 渲染成 root -> parent -> current 便于调试(:132)。

3.6 为什么"仅 root 栈"才收尾?——本章最该记住的一条​

一棵树里有很多栈,但面向用户的那一层只有一个:root。所以凡是"每个请求只该发生一次"的事, Yao 都用 ctx.Stack.IsRoot()(即 ParentID == "",agent/context/stack.go:115)当闸门, 只让 root 栈做:

只在 root 栈做的事若不 gate 会怎样代码
发 stream_start嵌套/多次 LLM 调用各发一遍,前端收到多个"开始"sendAgentStreamStart agent/assistant/agent.go:706
发 stream_end同上,多个"结束"sendAgentStreamEnd agent/assistant/agent.go:728
InitBuffer / FlushBuffer子 agent 重复建/刷聊天缓冲,历史错乱;Buffer 只属于 rootagent/assistant/chat.go:100、:201
BufferUserInput委派的子 agent 会把用户输入重复写一遍agent/assistant/chat.go:142
CloseOutput子 agent 提前关掉共享的输出通道agent/assistant/agent.go:262、:604

sendAgentStreamStart 的守卫就是一行,很能说明问题:

// 真实源码,agent/assistant/agent.go:705-708
func (ast *Assistant) sendAgentStreamStart(ctx *context.Context, handler message.StreamFunc, startTime time.Time) {
    if ctx.Stack == nil || !ctx.Stack.IsRoot() || handler == nil {
        return // 不是 root 栈,直接不发
    }
    // ... 只有 root 才 Marshal 并 handler(ChunkStreamStart, ...)
}

那子栈的输出怎么办?靠 ThreadID。 非 root 栈发消息时,若消息没带 ThreadID, 就自动填成本栈的 ID,让前端把嵌套/并行 agent 的输出分到不同"线程道"上显示:

// 真实源码,agent/context/output.go:86-88
// 为非 root 栈(嵌套 agent 调用)自动设置 ThreadID
if msg.ThreadID == "" && ctx.Stack != nil && !ctx.Stack.IsRoot() {
    msg.ThreadID = ctx.Stack.ID
}

同样的逻辑也出现在 agent/assistant/handlers/stream.go:116(stream_start 的 ThreadID)与 :407(缓冲最终消息时取 ThreadID)。一句话:root 管"整场"的首尾与落库,子栈只管在 UI 上分个道。

4. 核心机制二:委派 Delegation​

4.1 思路:把活整个交出去,仍算同一场对话​

委派的语义是"我这个 agent 不直接答了,换 X 来答用户"。所以它复用同一个 Context—— 同一个 ID、Space、Writer、Buffer 全不变,只是往栈上再摞一层子栈。因为是"同一场对话的延续", 所以委派的子 agent 的输出要进聊天历史。

4.2 三个委派入口​

Yao 有三处会触发委派,但它们最终都汇到同一个 handleDelegation:

入口时机白话代码
Create hook 委派LLM 调用之前提前路由:不花一次模型调用就把请求转给子 agentagent/assistant/agent.go:246-251
Next hook 委派一轮回答之后Next hook 决定"下一步交给谁"agent/assistant/next.go:19-20
loop_fallback 委派工具循环失败/打满时兜底:把上下文打包成 Markdown 丢给 __yao.loop_fallbackagent/assistant/agent.go:564-567

Create hook 委派的价值在"早"——注释直说是 "early routing to sub-agents without LLM call" (agent/assistant/agent.go:245),即无需先花一次模型调用就能路由。 loop_fallback 则由 buildLoopFallbackDelegate 造出一个指向 __yao.loop_fallback 的 DelegateConfig,把对话与工具结果格式化成一条 user 消息(agent/assistant/loop.go:204-218)。

4.3 handleDelegation:同 Context 调目标 agent​

三个入口的共同落点。它做的事极简:

// 真实源码,agent/assistant/next.go:48-66(handleDelegation 主体,已节选)
targetAssistant, err := Get(delegate.AgentID)          // 1. 加载目标 agent
// ...
ctx.Referer = agentContext.RefererAgent                // 2. 标记为 agent 委派来源
delegateOpts := agentContext.OptionsFromMap(delegate.Options)
return targetAssistant.Stream(ctx, delegate.Messages, delegateOpts)  // 3. 用同一个 ctx 调它

关键在第 3 步传的是同一个 ctx。于是目标 agent 的 Stream() 进门调 EnterStack 时, ctx.Stack != nil,走 NewChildStack 那条岔路(§3.3),自动挂成"父 agent → 委派 agent"的父子栈。

ctx.Referer = RefererAgent(值为 "agent",agent/context/types.go:68)这一步很要紧: 它让 shouldSkipHistory() 判定为"这仍是主对话流,要存历史"——注释原话: "Delegate calls (RefererAgent) still save history as they are part of the main conversation flow" (agent/assistant/agent.go:68)。这正是委派与下面 A2A fork 的分水岭。

DelegateConfig 结构本身很小:目标 AgentID、要发的 Messages、可选 Options (agent/context/types.go:487-491)。

5. 核心机制三:A2A fork(ctx.agent.Call/All/Any/Race)​

5.1 思路:借别的 agent 当"工具",结果拿回来但不许污染历史​

A2A(agent-to-agent)是在 Hook 脚本里用 ctx.agent.* 主动"叫"子 agent 当工具用。 四个方法对标 JS 的 Promise:

JSAPI语义对标
ctx.agent.Call(id, msgs, opts?)单个调用直接 await
ctx.agent.All(reqs)全部完成才返回Promise.all
ctx.agent.Any(reqs)任一成功即返回Promise.any
ctx.agent.Race(reqs)任一完成即返回Promise.race

接口定义:agent/context/jsapi_agent.go:12-24(AgentAPI)。它们和委派的根本区别: 子 agent 的中间对话不能进主聊天历史。

5.2 fork:给并行调用各造一个独立子上下文​

批量调用(All/Any/Race)天然并行,若共享同一个 Context,多个 goroutine 会同时改 Stack、Logger 等,直接 race。Yao 的对策是每个并行调用先 Fork() 出独立上下文:

// 真实源码,agent/caller/orchestrator.go:217-221
func (o *Orchestrator) callAgentWithForkedContext(req *Request) *Result {
    forkedCtx := o.ctx.Fork()                       // 独立 Stack / Logger
    return o.callAgentWithContext(forkedCtx, req)
}

Fork() 造出的子上下文(agent/context/context.go:170-227)遵循一条清晰的"共享 vs 独立"边界:

处理字段为什么
共享(只读/线程安全)Cache、Writer、Capabilities、Authorized、Stacks map线程安全或只读,可放心共享;Stacks 共享才能把子栈汇进同一棵树
独立(避免 race)新 ID、独立 IDGenerator、Logger、fork 的 Memory每个并行分支各写各的,互不干扰
置空(属于 root)Buffer=nil、Interrupt=nil、trace=nil、Stack=nilBuffer/Interrupt 只归 root;Stack 留给 EnterStack 建

Fork() 的收尾一步:若父有 Stack,就把父栈信息拷进 child.ForkParent(ForkParentInfo), 只拷 StackID/TraceID/Depth/Path 四个值,不拷 Stack 指针本身——这样并行分支各自 EnterStack 时走 NewChildStackFromForkParent(§3.3),既能挂对父子关系,又不会几个 goroutine 争同一个 Stack 对象(agent/context/context.go:216-225)。

一个容易踩的细节:单次 Call 并不 fork。 Call 走的是 o.callAgent → callAgentWithContext(o.ctx, req),直接用父自己的 o.ctx(agent/caller/jsapi.go:37、 orchestrator.go:210-212)。因为单次调用是同步的,没有并发 race,所以无需 fork; 此时 ctx.Stack != nil,EnterStack 走 NewChildStack 生成子栈。fork 只发生在 All/Any/Race 的每个并行分支上。

5.3 ForceA2A:强制跳历史,但保留输出​

不管 fork 与否,A2A 调用都会把 ctx.Referer 置成 RefererAgentFork(值 "agent_fork", agent/caller/orchestrator.go:242、agent/context/types.go:71)。这个标记一路传到子 agent 的 Stream(),触发强制跳历史:

// 真实源码,agent/assistant/agent.go:70-75
// 为 fork 的 A2A 调用自动跳历史,避免 fork 消息污染聊天历史。
if ctx.IsForkedA2ACall() {           // Referer == RefererAgentFork ?
    // ...
    opts.ForceA2A()                   // 置 skip.History = true
}

IsForkedA2ACall() 就是判 Referer == RefererAgentFork(agent/context/context.go:543); ForceA2A() 只干一件事——opts.Skip.History = true,但刻意不设 skip.output (agent/context/types.go:341-347)。注释把设计意图写得很清楚:

  • History 跳 → A2A 子 agent 的消息不写进主聊天历史(不污染);
  • Output 不跳 → 子 agent 照常输出,只是靠 ThreadID 在 UI 上分道(§3.6)。

JSAPI 侧还有一道保险:forceSkipForSubAgent 在构造请求时也强制 skip.History = true, 同时保留调用方显式设的 skip.output(agent/caller/jsapi.go:117-135)。于是 skip.History 最终会体现在栈的 Options 上,被 InitBuffer(chat.go:110)、 落库判定(handlers/stream.go:422-423)一并读到,真正做到"不进历史"。

5.4 批量 + 回调:为什么要走 channel​

All/Any/Race 若带了 onChunk 回调,会遇到一个 V8 约束:JS 回调必须在主 goroutine 里调, 而并行 agent 跑在各自的后台 goroutine 里。Yao 的解法是 channel 转交:

怎么读下图:后台 goroutine 只管"生产"消息塞进 channel,主 goroutine 专职"消费"并调 JS 回调。

后台 goroutine (并行跑 N 个 agent)          主 goroutine (V8 安全区)
  agent 产出 msg                              for msg := range msgChan {
     └─► goHandler: msgChan <- batchMessage ──►   callJSBatchCallback(cb, ...)  // 调 JS
         (阻塞发送 = 自然背压,不丢消息)         }
  全部跑完 → doneChan <- results  ───────────► return <-doneChan

依据:agent/context/jsapi_agent.go:404-479(executeBatchWithCallback)。要点:

  • msgChan 缓冲 1000,阻塞发送保证"不丢消息"(natural backpressure,:444-453);
  • 真正的批量执行在后台 goroutine 里跑 AllWithHandler/AnyWithHandler/RaceWithHandler (:456-470);
  • 主 goroutine for msg := range msgChan 把每条消息交给 callJSBatchCallback 调 JS—— 只有主 goroutine 碰 V8(:472-478);
  • 没带回调则直接 agentAPI.All/Any/Race,不走 channel(:417-427)。

callJSBatchCallback 给 JS 回调传三个参数:agentID、index(该请求在批次里的下标)、 消息对象——让脚本能分清"这条 chunk 是哪个子 agent、第几个请求"发的(:483-507)。

6. 委派 vs A2A fork:一张对照表带走​

维度委派 DelegationA2A fork(批量)A2A 单次 Call
入口Create/Next hook、loop_fallbackctx.agent.All/Any/Racectx.agent.Call
Context共享同一个每分支 Fork() 独立共享父 ctx(同步,无 race)
建栈函数NewChildStackNewChildStackFromForkParentNewChildStack
Refereragentagent_forkagent_fork
进聊天历史进(主对话流)跳(ForceA2A)跳(forceSkipForSubAgent)
并行串行并行,靠 fork 隔离单个
UI 隔离子栈 ThreadID各 fork 子栈 ThreadID + 批次 index子栈 ThreadID

三者的共同点:都由 EnterStack 挂成 root 的后代栈,都不是 root,所以都不发流首尾、 不 flush buffer、不关 output——那些永远只归 root(§3.6)。

7. 边界与坑​

  • 单次 Call 不 fork,是有意的。 它同步、无并发,复用父 ctx 省一次 fork;但也意味着它在 父的 ctx.Stack 上直接摞子栈并在 done() 里恢复——若你在回调里对父 ctx 有别的假设,记住这点。
  • 委派会存历史,fork 不会——选错语义会污染或丢历史。 想"让子 agent 的回答成为对话一部分" 用委派;想"借它算个中间结果"用 A2A。二者的唯一区分点就是 Referer(agent vs agent_fork)。
  • root 判定纯靠 ParentID == ""。 任何绕过 EnterStack 直接造 Stack 的路径,都可能让 "谁是 root"判断失真,进而错发/漏发 stream 首尾。收尾逻辑全押在 IsRoot() 上。
  • fork 的 Buffer 是 nil。 子上下文没有自己的 Buffer(归 root),所以子 agent 里任何依赖 ctx.Buffer 的落库操作会被 ctx.Buffer == nil 直接短路——这是"子不落库"的物理保证之一。
  • batch 回调的 V8 线程约束。 若绕开 executeBatchWithCallback 直接在后台 goroutine 里 碰 V8,会有线程安全问题;channel 转交不是可选优化,是正确性要求。

8. 代码地图(导航索引)​

主题文件关键符号
Stack 结构agent/context/types.goStack、ForkParentInfo
root 构造agent/context/stack.goNewStack
嵌套子栈agent/context/stack.goNewChildStack
fork 子栈agent/context/stack.goNewChildStackFromForkParent
进/出栈总控agent/context/stack.goEnterStack(返回 done())
收尾agent/context/stack.goComplete、Fail、Timeout
root 判定agent/context/stack.goIsRoot、GetPathString
追踪检索agent/context/stack.goGetAllStacks、GetStacksByTraceID、GetRootStack
root 收尾闸门agent/assistant/agent.gosendAgentStreamStart、sendAgentStreamEnd
Buffer 闸门agent/assistant/chat.goInitBuffer、FlushBuffer、BufferUserInput
委派落点agent/assistant/next.gohandleDelegation、processNextResponse
loop 兜底委派agent/assistant/loop.gobuildLoopFallbackDelegate
A2A 强制跳历史agent/assistant/agent.goStream(IsForkedA2ACall 分支)
A2A JSAPIagent/context/jsapi_agent.goagentCallMethod、executeBatchWithCallback
A2A 编排/forkagent/caller/orchestrator.gocallAgentWithForkedContext、All/Any/Race
fork 上下文agent/context/context.goFork、IsForkedA2ACall
A2A 选项agent/context/types.goForceA2A、RefererAgent/RefererAgentFork
子栈 ThreadIDagent/context/output.go非 root → msg.ThreadID = ctx.Stack.ID