用 Agent Core 自建多智能体

上一篇讲了 Pi 为什么不内置子代理：它只给原语，成品交给生态。这篇反过来——只用 pi-agent-core，你能不能自己拼出一套能用的多智能体？

能，代码量不大。核心一句话：子智能体是一个工具，工具的 execute() 里再开一个 Agent。

但"能跑"和"跑对"之间隔着几道必须想清楚的问题：子代理的上下文从哪来？并发扇出怎么限？子代理的工具权限谁管？成本怎么控？这篇按源码逐层拆开，最后给一份能直接跑的最小骨架。

前置知识：读过本模块第一篇（Pi 的十三个原语），以及第三章（packages/agent/src/ 的用法、事件序列、状态机）。

Agent 类能做什么

packages/agent/src/agent.ts:194 的 Agent 是全部编排能力的载体，公开表面只有六组东西：

没有编排器、没有调度器、没有任务图。 多智能体的结构得你自己用这六样拼出来。

构造有一个隐藏条件：initialState 里的 systemPrompt 和 tools 不是独立字段，会被折叠成 transcript 的首条 system message。官方注释（agent.ts:84）写明："unless messages already starts with one"——做 fork 时你复制了父的 transcript，父的 system message 保留，子代理的 systemPrompt 就不会生效。

干预的两个队列差在注入时机（agent.ts:307）：steer() 是 "to be injected after the current assistant turn finishes"（插队，"停一下，改成查数据库"），followUp() 是 "to run only after the agent would otherwise stop"（续跑，"再补一轮验证"）。配套 steeringMode / followUpMode（默认 "one-at-a-time"）、三个 clear*Queue()、peekQueuedMessages()，后者里 steering 优先（agent.ts:339）。

控制组三个方法：

reset() 三个要点（agent.ts:363）：运行中调用会 throw；保留 system baseline（含工具声明）；不清 state.tools——所以"保留提示词和工具、只换任务"的循环型编排，一行 reset() 就够。signal getter 则给出当前运行的 AbortSignal，用来给子代理接取消链。

十个事件

packages/agent/src/types.ts:528 定义了全部 10 个 AgentEvent：

message_update 只对 assistant 消息发——用户消息和工具结果消息没有增量版本。tool_execution_end 带 durationMs（types.ts:550，"measured with a monotonic clock"），这是子代理编排做慢任务检测的唯一原生信号。

三种时序

纯对话：

prompt("Hello")
├─ agent_start
├─ turn_start
├─ message_start/end  { userMessage }      // 你的输入
├─ message_start      { assistantMessage } // 模型开始响应
├─ message_update...                       // 流式增量
├─ message_end        { assistantMessage } // 完整响应
├─ turn_end        { message, toolResults: [] }
└─ agent_end       { messages: [...] }

带工具调用（message_start/end 合写表示成对事件）：

prompt("Read config.json")
├─ agent_start
├─ turn_start
├─ message_start/end  { userMessage }
├─ message_start      { assistantMessage with toolCall }
├─ message_update...
├─ message_end        { assistantMessage }
├─ tool_execution_start  { toolCallId, toolName, args }
├─ tool_execution_update { partialResult }      // 工具流式时才有
├─ tool_execution_end    { toolCallId, result }
├─ message_start/end  { toolResultMessage }
├─ turn_end           { message, toolResults: [toolResult] }
│
├─ turn_start        // 下一轮：同样的模式再来一遍
└─ agent_end

工具结果消息在 turn_end 之前发出，且在 assistant 消息之后。如果你在 turn_end 里提取子代理结论，要清楚 toolResults 是这一轮的全部工具结果。

订阅结算：

agent_start
  ... 事件流 ...
agent_end          ← 最后一个发出的事件
  ⏳ 等待 awaited 监听器完成   ← 运行仍未结算
  ⏳ 监听器完成
isStreaming = false  ← finishRun()，运行结算
waitForIdle() resolve

packages/agent/README.md:200 说得很明确：

Agent.subscribe() listeners are awaited in registration order. agent_end means no more loop events will be emitted, but await agent.waitForIdle() and await agent.prompt(...) only settle after awaited agent_end listeners finish.

源码里监听器是串行 await 的（agent.ts:619），一个慢监听器会拖住整个运行的结算。实用结论：监听器里不要做重活——落盘、flush、上报放进工具的 execute()，或者用 waitForIdle() 之后自己做。别把 agent_end 监听器当"收尾钩子"，它是结算的一部分。

AgentState 的四个关键机制

一、拷贝语义。 agent.ts:99-106 的 setter 拷贝顶层数组、getter 直接返回内部数组：

		set tools(nextTools: AgentTool<any>[]) {
			tools = nextTools.slice();
		},
		set messages(nextMessages: AgentMessage[]) {
			messages = nextMessages.slice();
		},

官方 README 确认（:302）："copies the top-level array before storing it. Mutating the returned array mutates the current agent state."

风险点：编排层从 state.messages 取数组、遍历完往里 push 一条"编排注解"，那条注解会直接进子代理的 transcript。这不是 bug 是特性——它让"编排层给子代理追加上下文"变成一行代码。

二、提示词归 transcript 所有。 types.ts:383-389 的注释说得很清楚：systemPrompt 是 "replayed from the transcript's system messages"，Read-only。想改提示词，唯一办法是 append 一条 system message（content 追加指令，sections 具名替换）：

await agent.prompt([
  { role: "system", content: "", sections: { skills: "<skills>...</skills>" }, timestamp: Date.now() },
  { role: "user", content: "Continue", timestamp: Date.now() },
]);

给子代理换提示词不是改字段，是往 transcript 里插一条消息。 fork 场景下 initialState.systemPrompt 赋新值不生效。

三、工具声明差分。 agent-loop.ts:342 的 declareToolChanges：每轮请求前对比"能执行的工具集"和"transcript 声明给模型看的工具集"，不一致就发一条 system message 宣布变更（toolsAdded / toolsRemoved）。官方注释里那句 "so replay always yields exactly context.tools" 是关键——工具清单的最终真相是 context.tools，transcript 只是它的声明。 运行中换工具是官方支持的而且模型会知道；反过来，改了 state.tools 却期望模型看不见，是做不到的。

四、状态归约。 agent.ts:575 的 processEvents 在派发事件前先归约内部状态，其中 pendingToolCalls 用的是"拷贝后修改"（new Set(...) 再 add），因为它在状态类型里是 ReadonlySet<string>（types.ts:418），必须替换引用。编排层的实用价值：state.pendingToolCalls 是一个准确实时的"子代理正在跑哪些工具"视图，可以做超时判断或 UI 展示。

AgentOptions 的二十多个字段

agent.ts:115 起的 AgentOptions：

beforeToolCall 的拦截不抛异常（types.ts:66）：返回 { block: true, reason: "..." }，loop 生成一条错误工具结果喂给模型（agent-loop.ts:755-765）。模型看到的是"这个操作被拒绝了，原因如下"，然后自己换路子。 返回值里还可以带 terminate: true，让被拦截的这次调用参与批次提前停的判定。

finishTurn 是限制子代理轮数的正确位置（types.ts:147）：{ action: "continue" } | { action: "end" }。返回 "end" 在 turn_end 后立刻结束运行。但别无条件返回 "continue"——官方 README 警告：finishTurn runs again after the next request, so returning { action: "continue" } unconditionally creates an endless loop。

sessionId 在 fork 场景要特别处理。 durable 的 fork 语义明确："it keeps the parent's agent as of that entry but receives a fresh provider session identity"——fork 出来的子会话必须拿新的 provider session identity，否则缓存会串。

最小可运行骨架

把上面的东西拼起来，下面这份可以直接跑：

import { Agent, type AgentTool, type AgentToolResult } from "@earendil-works/pi-agent-core";
import { createModels } from "@earendil-works/pi-ai";
import { anthropicProvider } from "@earendil-works/pi-ai/providers/anthropic";

const models = createModels();
models.setProvider(anthropicProvider());
const model = models.getModel("anthropic", "claude-sonnet-4-6")!;

async function runResearcher(task: string, signal?: AbortSignal): Promise<AgentToolResult> {
	// 独立上下文：子 Agent 不继承父对话
	const child = new Agent({
		initialState: {
			systemPrompt: "你是一个只做调研的子智能体。输出简明的事实清单，不做实现。",
			model,
		},
		streamFn: models.streamSimple.bind(models),
	});
	await child.prompt(task);
	const last = child.state.messages.findLast((m) => m.role === "assistant");
	const text = last?.role === "assistant"
		? last.content.filter((c) => c.type === "text").map((c) => c.text).join("")
		: "";
	return { content: [{ type: "text", text }], details: { chars: text.length } };
}

const researcherTool: AgentTool<{ task: string }> = {
	name: "researcher",
	label: "Researcher",
	description: "委托一个独立上下文的子智能体做调研，返回简明结论。",
	parameters: {
		type: "object",
		properties: { task: { type: "string", description: "调研任务" } },
		required: ["task"],
	},
	execute: async (_id, params, signal) => runResearcher(params.task, signal),
};

几个细节。 AgentTool 继承 Tool<TParameters>（要求 name / description / parameters），自己加 label（UI 用）和 execute。execute 的签名是 (toolCallId, params, signal?, onUpdate?)（types.ts:495）——signal 是第三个参数。child.prompt(task) 之后读 child.state.messages 是安全的：message_end 时 loop 已经把消息 push 进 state。

content 和 details 的分工是成本控制的核心（types.ts:426-433）：details 是 "Arbitrary structured details for logs or UI rendering"，structuredContent 的注释写明 "Not sent to the model; content remains the model-facing result"。只有 content 进模型上下文。 子代理返回 100KB 原始日志，把冗余部分放 details，content 只留结论，主会话就干净了。

让取消真的能取消。 骨架里 signal 传下去却没用上，正确接法是在 prompt 之后挂一个 abort 监听：

	const run = child.prompt(task);
	// 父级取消时，一并中止子代理
	signal?.addEventListener("abort", () => child.abort(), { once: true });
	await run;

父子取消链要自己接——pi-agent-core 不会自动把工具的 signal 传给子 Agent。另外，要严格等结束用 await child.waitForIdle()：prompt() resolve 时 awaited 的 agent_end 监听器可能还没跑完。

并发扇出

官方示例（packages/coding-agent/examples/extensions/subagent/index.ts:217）给了一个干净的 worker pool：

async function mapWithConcurrencyLimit<TIn, TOut>(
	items: TIn[],
	concurrency: number,
	fn: (item: TIn, index: number) => Promise<TOut>,
): Promise<TOut[]> {
	if (items.length === 0) return [];
	const limit = Math.max(1, Math.min(concurrency, items.length));
	const results: TOut[] = new Array(items.length);
	let nextIndex = 0;
	const workers = new Array(limit).fill(null).map(async () => {
		while (true) {
			const current = nextIndex++;
			if (current >= items.length) return;
			results[current] = await fn(items[current], current);
		}
	});
	await Promise.all(workers);
	return results;
}

"结果按输入顺序排列"在多智能体里很关键：并行 8 个子代理，汇总给模型的文本应该对应请求顺序而不是完成顺序。但 Promise.all 的失败语义要小心：一个 worker 抛错，其它 worker 仍在跑（Promise.all 只是不再等它们），它们持有的资源不会自动清理——一个子代理失败不该让其它子代理的进程泄漏，需要 Promise.allSettled 或显式清理。

官方示例的两个常量（index.ts:34）是 MAX_PARALLEL_TASKS = 8 和 MAX_CONCURRENCY = 4，两个数不一样是有讲究的：前者是"一次最多接受几个并行任务"（超过就拒绝，见 index.ts:603），后者是"同时真正跑几个"。入队上限防止模型一次派 100 个任务撑爆上下文，执行上限防止同时 100 个进程压垮机器。

权限穿透

编排工具内部要调别的工具，正确路径是 runToolCall(toolCall, { tools, assistantMessage, context, signal })——它返回 { toolCall, result, isError, durationMs }。

runToolCall 的行为（agent-loop.ts:812-819 的注释）：走完参数准备 → schema 校验 → beforeToolCall → 执行 → afterToolCall 的完整链路；不发出任何事件，不追加任何消息；永不 reject——未知工具、校验失败、被拦截、抛异常，全部以 isError: true 返回。

"不发出事件"有两个后果。 一是编排层看不到子调用的事件流——父会话的 subscribe() 只看到子智能体工具本身的一次 tool_execution_start / _end，看不到里面 5 次工具调用，想看细节得自己转发。二是子调用不污染 transcript——这是好事，子代理的内部探索不会撑大主会话上下文。"永不 reject"决定了错误处理方式：别 try/catch 期待它抛，要检查 outcome.isError。

多智能体安全的完整链条：模型调编排工具 → 编排工具内部用 runToolCall() 调子工具 → 子工具的 beforeToolCall 生效（权限在这里把关）→ 子工具执行 → afterToolCall 生效 → 编排工具返回结果给模型。中间那层是你唯一能插入权限检查的地方——直接 tool.execute() 会跳过钩子和参数校验，子代理就成了能绕过父会话所有管控的黑盒。

三种隔离形态

**形态一（进程内）**就是上面的骨架：上下文完全独立，零启动开销，可共享进程内的工具和钩子。缺点是共享进程——一个子代理崩了可能带倒另一个。

**形态二（fork）**继承父会话历史但独立运行。durable 的语义最完整（:710）："A fork sees its parent's entries up to entryId and continues independently. It keeps the parent's agent as of that entry but receives a fresh provider session identity." 生态里 pi-subagents 点了 fork 的 cache 优势（docs/workflows.md:22）："A forked child sends the history it inherited exactly as the parent sent it, so its first request can reuse the parent's prompt cache; it still cannot run subagent."——fork 出来的子代理不能再生子代理，这是能力移除实现的，不是深度计数。

**形态三（子进程）**官方示例的做法（index.ts:298）：args 以 ["--mode", "json", "-p", "--no-session"] 开头，再按需追加 --model、--thinking、--tools，系统提示词写临时文件用 --append-system-prompt <file> 传入（避开命令行长度限制和转义地狱）。四个参数里最关键的是 --tools——子进程是独立进程，父会话的 beforeToolCall 管不到它，权限完全由命令行参数决定。

"权限穿透"只在进程内形态成立。 子进程靠参数白名单，粒度更粗但边界更硬。官方示例的中止处理（index.ts:408-418）是 SIGTERM 后 5 秒再 SIGKILL——宽限期让子进程有机会清理，代价是中止不是即时的，父以为中止了实际可能有 5 秒窗口期。

附件、transport 与 codemode

pi-agent-core 里没有 "attachment" 这个概念（在 packages/agent/src/ 全目录 grep attachment 零命中）。实际载体三个：

多模态内容进对话靠 Agent.prompt() 的重载（agent.ts:381）：prompt(input: string, images?: ImageContent[])，内部把图片塞进 user 消息的 content 数组。"给子代理传一张图"的正确做法是 child.prompt(text, [imageContent])。 工具结果也能带图片，主会话的模型能看到。

Transport 定义在 pi-ai 层（不是 pi-agent-core），packages/ai/src/types.ts:147："sse" | "websocket" | "websocket-cached" | "auto"，auto 是 Agent 的默认值（agent.ts:259）。降级语义有严格边界——唯一真实实现在 packages/ai/src/api/openai-codex-responses.ts:313 附近，失败时先记录诊断（带 phase 和 fallbackTransport），然后 if (websocketStarted) { throw error; }：消息流开始前失败降级到 SSE 重试（before_message_stream_start），消息流开始后失败不降级，直接抛错（after_message_stream_start）。

"消息流开始后不降级"是必须的：流已经开了，模型可能已经吐了一半内容，重发会导致重复内容。多智能体场景下实际影响很小——父子 Agent 各有自己的连接；但子代理数量多时 WebSocket 的连接复用价值会放大，websocket-cached 就是为复用 session 级连接存在的。

codemode 是"编排即代码"的官方答案。 packages/codemode/README.md 开头："Runs model-written JavaScript in a QuickJS VM (compiled to WebAssembly) where the only capability is calling injected tools. Nested tool calls never enter the LLM context; only the script's output and return value do."

三条资源限制（packages/coding-agent/docs/codemode.md:224）：

• A script's VM has 256 MB of memory. Running out throws InternalError: out of memory; filter or aggregate large data instead of accumulating it.
• At most two codemode workers may be alive across all sessions in one pi process; PI_CODEMODE_MAX_WORKERS changes the limit. A cancelled worker occupies its slot until it exits. At capacity, new scripts wait for a free worker in arrival order. A script that gets no worker within 60 seconds, or before its timeout_ms, fails without running.
• Scripts cannot start other codemode scripts.

"一个 pi 进程内最多同时存活 2 个 codemode worker"对多智能体影响很大。 如果编排是"父 Agent 起 8 个子 Agent，每个子 Agent 内部都跑 codemode"，worker 池只有 2 个槽——其余排队等，60 秒拿不到就失败。"脚本不能启动其它 codemode 脚本"是防递归的另一道墙，和 durable 的"移除 subagent 工具"同一思路：不是靠计数，是靠能力不存在。

成本控制

多智能体的成本不是"模型贵"，是放大。十个放大源：

第 7 条最反直觉：codemode 让嵌套调用不进 LLM 上下文，但 provider 侧的 token 照样扣。codemode 省的是上下文窗口占用和模型理解噪声，不是钱。第 1 条的推理：子代理用 fork 会继承父的完整历史，父会话有 50KB 上下文，fork 一次就是 50KB 额外输入——子代理通常只需要一段任务简报。

"用小模型 + 移除递归能力"在 durable 的示例里是一行——await configure(tx, created.id, { model: haiku, extensions: { remove: [Subagent] } })，这应该成为子代理配置的默认模板。

失败传播

"输出 token 耗尽 → 所有 tool call 全部失败"这段注释很实在（agent-loop.ts:481）："Streamed tool-call arguments are finalized with a best-effort JSON salvage parser, so a truncated message can yield tool calls whose arguments parse and validate but are silently incomplete. None of them are safe to execute."

"参数能解析、能校验，但内容是残缺的"——流式工具调用的经典陷阱。被截断的 {"path": "/very/long/pat 补全后可能变成合法的短路径，执行了但目标是错的。多智能体场景下这条特别重要：子代理的长任务经常撞上输出 token 上限，如果代理自己实现了"截断后尽力解析并执行"，就会出现执行了错误的目标还报告成功。

谁负责什么

三种场景的可观察结果

场景一：写了子智能体工具，跑起来主会话上下文暴涨。 检查子代理返回的 content 长度。可观察结果是主会话 transcript 里塞满子代理的原始输出——那些被放进了 content。把冗余挪到 details，content 只留结论，模型上下文立刻干净。

场景二：父会话结束了，但子代理还在跑。 检查你用的是 await child.prompt() 还是 await child.waitForIdle()。可观察结果是读结果时 state.isStreaming 还是 true，或者子代理的工具还在执行——前者是结算时序问题，后者是取消链没接。

场景三：子代理执行了你不允许的工具。 检查子代理内部调工具的路径。可观察结果是 beforeToolCall 的日志里没有这次调用——说明子代理绕过了钩子（直接 tool.execute() 或走了子进程）。前者改用 runToolCall()，后者只能靠 --tools 白名单。

你现在能做什么

• 跑一遍上面那份最小骨架，让模型调一次 researcher 工具，确认子代理输出进了 content 而 details 没进上下文。
• 给骨架加一个 beforeToolCall，让子代理也带一个受限工具，验证钩子对嵌套调用生效。
• 把 child.prompt(task) 换成 await child.waitForIdle()，在 agent_end 监听器里加 100ms 延时，观察两者的差别。
• 用 mapWithConcurrencyLimit 扇出 8 个子代理（并发 4），每个打时间戳，验证结果顺序和完成顺序无关。
• 跑官方示例 packages/coding-agent/examples/extensions/subagent/index.ts，观察 --mode json -p --no-session 的事件流和 --tools 白名单怎么传给子进程。
• 把子代理的 initialState.systemPrompt 改一下，再改成"先复制父 transcript 再赋新 prompt"，验证 fork 场景下新 prompt 不生效；再在 finishTurn 里加轮数计数器，超过 5 轮返回 { action: "end" }，验证子代理能被硬性截断。

排障速查

参考资料

• 官方文档：Pi Agent Core
• 官方文档：Extensions
• 官方文档：Codemode
• pi-agent-core 源码
• pi-agent-core 类型定义（事件、状态、工具结果）
• agent-loop 源码（runToolCall、工具声明差分、terminate 规则）
• pi-agent-core README
• 官方 subagent 扩展示例（并发扇出、子进程隔离）
• pi-durable README（fork、ownership、子任务）
• pi-codemode README
• pi-ai 传输层类型定义