Pi 子代理导览：从官方示例看进程级 Agent 编排
2026-07-29 21:00:00
✨ 心情：study
# Pi
# Agent
# 子代理
# 架构分析
Pi 子代理导览：从官方示例看进程级 Agent 编排
本文是对 Pi 官方
subagent示例扩展的代码导览。Pi 本身不内建子代理能力，但提供了一份相当完整的示例扩展，演示了如何用"进程级隔离 + JSON 流协议"把一个 LLM 编排成多个有专属上下文的协作 Agent。
一、为什么这份示例值得读
Pi 的每个子代理是一次独立的 pi 进程 spawn。父进程不直接调用 LLM，而是通过子进程的 stdout 拿到 JSON 行流，解析出 message_end / tool_result_end 事件再聚合成结果。
这种做法的好处：
- 上下文彻底隔离：子代理的 system prompt、工具集、对话历史都在自己的进程里，父进程不会被污染
- 能力声明式：每个 Agent 是一个
.md文件，YAML frontmatter 声明tools/model，body 是 system prompt - 可独立调试：子代理的
pi进程可以单独跑，跟父进程完全解耦
代价是进程启动开销和跨进程通信的复杂度。
二、目录结构全景
subagent/
├── README.md # 用户文档
├── index.ts # 扩展入口：注册 subagent 工具
├── agents.ts # Agent 发现与加载
├── agents/ # 示例 Agent 定义（Markdown + frontmatter）
│ ├── scout.md # 快速侦察，返回压缩上下文
│ ├── planner.md # 制定实施计划
│ ├── reviewer.md # 代码审查
│ └── worker.md # 通用执行
└── prompts/ # 工作流模板（prompt templates）
├── implement.md # scout → planner → worker
├── scout-and-plan.md # scout → planner（不实施）
└── implement-and-review.md # worker → reviewer → worker
两条核心轴：
agents/*.md— 静态能力声明：一个 Agent 能用什么工具、跑在什么模型上、system prompt 是什么prompts/*.md— 工作流模板：把多个 Agent 串成一条链，用$@接收用户输入
index.ts 是把这两者粘合起来的运行时。
三、Agent 定义：一份 Markdown 就是一个 Agent
先看最简单的 scout.md：
---
name: scout
description: Fast codebase recon that returns compressed context for handoff to other agents
tools: read, grep, find, ls, bash
model: claude-haiku-4-5
---
You are a scout. Quickly investigate a codebase and return structured findings
that another agent can use without re-reading everything.
Your output will be passed to an agent who has NOT seen the files you explored.
Thoroughness (infer from task, default medium):
- Quick: Targeted lookups, key files only
- Medium: Follow imports, read critical sections
- Thorough: Trace all dependencies, check tests/types
Strategy:
1. grep/find to locate relevant code
2. Read key sections (not entire files)
3. Identify types, interfaces, key functions
4. Note dependencies between files
Output format:
## Files Retrieved
List with exact line ranges:
1. `path/to/file.ts` (lines 10-50) - Description of what's here
几个关键点：
- YAML frontmatter 是契约：
name/description必填，tools是逗号分隔的工具白名单，model指定跑哪个模型 - body 就是 system prompt：没有模板变量、没有占位符，纯文本直接喂给子进程
- 输出格式约束在 prompt 里：
## Files Retrieved这种结构化输出约定，是为了让下游 Agent 能直接吃上游的输出
注意 tools 字段。scout 有 bash 但 planner 没有：
---
name: planner
description: Creates implementation plans from context and requirements
tools: read, grep, find, ls
model: claude-sonnet-4-5
---
You are a planning specialist. You receive context (from a scout) and
requirements, then produce a clear implementation plan.
You must NOT make any changes. Only read, analyze, and plan.
planner 只能读不能写——能力约束在声明层就锁死了，不靠 prompt 里的"请不要修改文件"自我约束。这是声明式 Agent 设计的核心收益。
reviewer 进一步演示了"即使给了 bash 也约束为只读"：
---
name: reviewer
description: Code review specialist for quality and security analysis
tools: read, grep, find, ls, bash
model: claude-sonnet-4-5
---
You are a senior code reviewer. Analyze code for quality, security, and
maintainability.
Bash is for read-only commands only: `git diff`, `git log`, `git show`.
Do NOT modify files or run builds.
Assume tool permissions are not perfectly enforceable; keep all bash
usage strictly read-only.
tools 给了 bash，但 prompt 里显式说"假设工具权限不是完美可强制的"——这是一种防御性设计：声明层做最大约束，prompt 层做行为约束，两层不互信。
四、Agent 发现：两级 scope + 安全模型
agents.ts 实现 Agent 发现。核心是两个目录的扫描：
export function discoverAgents(cwd: string, scope: AgentScope): AgentDiscoveryResult {
const userDir = path.join(getAgentDir(), "agents");
const projectAgentsDir = findNearestProjectAgentsDir(cwd);
const userAgents = scope === "project" ? [] : loadAgentsFromDir(userDir, "user");
const projectAgents =
scope === "user" || !projectAgentsDir ? [] : loadAgentsFromDir(projectAgentsDir, "project");
const agentMap = new Map<string, AgentConfig>();
if (scope === "both") {
for (const agent of userAgents) agentMap.set(agent.name, agent);
for (const agent of projectAgents) agentMap.set(agent.name, agent);
} else if (scope === "user") {
for (const agent of userAgents) agentMap.set(agent.name, agent);
} else {
for (const agent of projectAgents) agentMap.set(agent.name, agent);
}
return { agents: Array.from(agentMap.values()), projectAgentsDir };
}
两个目录：
- user scope：
~/.pi/agent/agents/*.md— 用户级，永远加载 - project scope：
.pi/agents/*.md— 项目级，仓库可控
scope === "both" 时，project agent 覆盖同名 user agent（因为 agentMap.set 后写入覆盖前写入）。这是允许项目定制/覆盖用户全局 Agent 的设计。
项目级 Agent 是安全敏感点。因为 .pi/agents/*.md 是仓库 controlled 的 prompt，能指示模型读文件、跑 bash——克隆一个恶意仓库后跑 pi 就可能执行任意指令。所以默认 scope === "user"，要用 project agent 必须显式 agentScope: "both" 或 "project"。
findNearestProjectAgentsDir 还做了向上递归查找：
function findNearestProjectAgentsDir(cwd: string): string | null {
let currentDir = cwd;
while (true) {
const candidate = path.join(currentDir, ".pi", "agents");
if (isDirectory(candidate)) return candidate;
const parentDir = path.dirname(currentDir);
if (parentDir === currentDir) return null;
currentDir = parentDir;
}
}
类似 git 找 .git 目录的逻辑——从当前工作目录向上找最近的 .pi/agents。这意味着子目录工作时会继承父级项目的 Agent 定义。
loadAgentsFromDir 用了 Pi 提供的 parseFrontmatter 工具：
function loadAgentsFromDir(dir: string, source: "user" | "project"): AgentConfig[] {
const agents: AgentConfig[] = [];
if (!fs.existsSync(dir)) return agents;
let entries: fs.Dirent[];
try {
entries = fs.readdirSync(dir, { withFileTypes: true });
} catch {
return agents;
}
for (const entry of entries) {
if (!entry.name.endsWith(".md")) continue;
if (!entry.isFile() && !entry.isSymbolicLink()) continue;
const filePath = path.join(dir, entry.name);
let content: string;
try {
content = fs.readFileSync(filePath, "utf-8");
} catch {
continue;
}
const { frontmatter, body } = parseFrontmatter<Record<string, string>>(content);
if (!frontmatter.name || !frontmatter.description) continue;
const tools = frontmatter.tools
?.split(",")
.map((t: string) => t.trim())
.filter(Boolean);
agents.push({
name: frontmatter.name,
description: frontmatter.description,
tools: tools && tools.length > 0 ? tools : undefined,
model: frontmatter.model,
systemPrompt: body,
source,
filePath,
});
}
return agents;
}
细节：
- 符号链接放行：
entry.isSymbolicLink()也接受——这是 README 里推荐的安装方式（ln -sf把开发目录链接到~/.pi/agent/agents/） - 缺失
name或description静默跳过：不用 throw，避免一个坏文件让整个发现失败 - 每次调用都重新扫：README 里写"Agents discovered fresh on each invocation (allows editing mid-session)"——开发体验优先于性能
五、子进程编排：spawn + JSON line 协议
index.ts 的核心是 runSingleAgent 函数。这是整套示例的心脏。
5.1 构造子进程参数
const args: string[] = ["--mode", "json", "-p", "--no-session"];
if (agent.model) args.push("--model", agent.model);
if (agent.tools && agent.tools.length > 0) args.push("--tools", agent.tools.join(","));
let tmpPromptDir: string | null = null;
let tmpPromptPath: string | null = null;
// ...
if (agent.systemPrompt.trim()) {
const tmp = await writePromptToTempFile(agent.name, agent.systemPrompt);
tmpPromptDir = tmp.dir;
tmpPromptPath = tmp.filePath;
args.push("--append-system-prompt", tmpPromptPath);
}
args.push(`Task: ${task}`);
关键 flag：
--mode json— 让子进程 stdout 输出 JSON 行流，而不是 TTY 渲染-p— non-interactive / pipe 模式--no-session— 不持久化 session，跑完即弃--model— 覆盖默认模型（来自 frontmatter）--tools— 限制工具白名单（来自 frontmatter）--append-system-prompt— 把 frontmatter body 注入为系统提示
system prompt 走临时文件而不是命令行参数，是个细节但很重要：避免 prompt 里的特殊字符触发 shell 解析问题，也绕开命令行长度限制。writePromptToTempFile 用 mkdtemp 建独立目录、mode: 0o600 限制权限、finally 块里清理：
async function writePromptToTempFile(agentName: string, prompt: string): Promise<{ dir: string; filePath: string }> {
const tmpDir = await fs.promises.mkdtemp(path.join(os.tmpdir(), "pi-subagent-"));
const safeName = agentName.replace(/[^\w.-]+/g, "_");
const filePath = path.join(tmpDir, `prompt-${safeName}.md`);
await withFileMutationQueue(filePath, async () => {
await fs.promises.writeFile(filePath, prompt, { encoding: "utf-8", mode: 0o600 });
});
return { dir: tmpDir, filePath };
}
5.2 解析 pi 的运行时入口
getPiInvocation 解决"怎么找到 pi 可执行文件"这个问题：
function getPiInvocation(args: string[]): { command: string; args: string[] } {
const currentScript = process.argv[1];
const isBunVirtualScript = currentScript?.startsWith("/$bunfs/root/");
if (currentScript && !isBunVirtualScript && fs.existsSync(currentScript)) {
return { command: process.execPath, args: [currentScript, ...args] };
}
const execName = path.basename(process.execPath).toLowerCase();
const isGenericRuntime = /^(node|bun)(\.exe)?$/.test(execName);
if (!isGenericRuntime) {
return { command: process.execPath, args };
}
return { command: "pi", args };
}
三种情况：
- 当前进程是从某个脚本文件起的（
process.argv[1]是真实文件路径）：直接复用——node /path/to/pi.js <args>。这是开发模式下最常见的。 - 当前 runtime 不是通用 node/bun（比如打包成了
pi.exe）：直接用process.execPath当命令。 - 是通用 node/bun 但没有可用脚本入口：fallback 到 PATH 里的
pi命令。
这套逻辑保证：开发时跑 ts 源码能 spawn 出同样的 ts 进程；生产环境装了 pi CLI 也能找到。Bun 虚拟脚本检测（/$bunfs/root/）是 Bun 打包模式的特殊处理。
5.3 spawn 与流式解析
const exitCode = await new Promise<number>((resolve) => {
const invocation = getPiInvocation(args);
const proc = spawn(invocation.command, invocation.args, {
cwd: cwd ?? defaultCwd,
shell: false,
stdio: ["ignore", "pipe", "pipe"],
});
let buffer = "";
const processLine = (line: string) => {
if (!line.trim()) return;
let event: any;
try {
event = JSON.parse(line);
} catch {
return;
}
if (event.type === "message_end" && event.message) {
const msg = event.message as Message;
currentResult.messages.push(msg);
if (msg.role === "assistant") {
currentResult.usage.turns++;
const usage = msg.usage;
if (usage) {
currentResult.usage.input += usage.input || 0;
currentResult.usage.output += usage.output || 0;
currentResult.usage.cacheRead += usage.cacheRead || 0;
currentResult.usage.cacheWrite += usage.cacheWrite || 0;
currentResult.usage.cost += usage.cost?.total || 0;
currentResult.usage.contextTokens = usage.totalTokens || 0;
}
if (!currentResult.model && msg.model) currentResult.model = msg.model;
if (msg.stopReason) currentResult.stopReason = msg.stopReason;
if (msg.errorMessage) currentResult.errorMessage = msg.errorMessage;
}
emitUpdate();
}
if (event.type === "tool_result_end" && event.message) {
currentResult.messages.push(event.message as Message);
emitUpdate();
}
};
proc.stdout.on("data", (data) => {
buffer += data.toString();
const lines = buffer.split("\n");
buffer = lines.pop() || "";
for (const line of lines) processLine(line);
});
proc.stderr.on("data", (data) => {
currentResult.stderr += data.toString();
});
proc.on("close", (code) => {
if (buffer.trim()) processLine(buffer);
resolve(code ?? 0);
});
proc.on("error", () => {
resolve(1);
});
if (signal) {
const killProc = () => {
wasAborted = true;
proc.kill("SIGTERM");
setTimeout(() => {
if (!proc.killed) proc.kill("SIGKILL");
}, 5000);
};
if (signal.aborted) killProc();
else signal.addEventListener("abort", killProc, { once: true });
}
});
逐段看：
stdio 配置：["ignore", "pipe", "pipe"] — stdin 不要（子代理不需要交互输入），stdout/stderr 都接管道。
行缓冲处理：data.toString() 拿到的 chunk 不一定是完整一行，所以维护一个 buffer，每次 split("\n") 后把最后一段塞回 buffer 等下次拼接。这是流式 JSON 行协议的标准处理。
事件类型：只关心两种事件
message_end— 一条 assistant 或 user 消息结束。从这里提取 usage 统计、stopReason、errorMessagetool_result_end— 工具调用结果结束。把 toolResult 消息推进 messages 数组
JSON 解析容错：try { event = JSON.parse(line) } catch { return } — 解析失败的行直接丢弃。pi 在非 JSON 模式下可能输出非 JSON 内容，这种容错必要。
Usage 累积：每个 assistant 消息带 usage，逐条累加。contextTokens 用最后一次的 totalTokens（不是累加，因为这是上下文窗口占用，不是流量）。
abort 传播：父进程收到 Ctrl+C 时调 proc.kill("SIGTERM")，5 秒后还没死就 SIGKILL。这是双阶段杀进程的标准做法——给子进程留清理时间，但不容许无限拖延。
stderr 不解析：直接累积到 currentResult.stderr，作为失败时的诊断信息。
5.4 流式更新回调
emitUpdate 是这套架构的精髓：
const emitUpdate = () => {
if (onUpdate) {
onUpdate({
content: [{ type: "text", text: getFinalOutput(currentResult.messages) || "(running...)" }],
details: makeDetails([currentResult]),
});
}
};
onUpdate 是 Pi 工具协议的流式回调——每收到一个事件就推一次更新给父进程的 UI。这意味着用户能看到子代理在跑什么工具、当前输出是什么，而不是等几十秒后一次性返回。
details 字段携带结构化信息（SubagentDetails），UI 层可以渲染成折叠/展开视图；content 是模型可见的纯文本，作为工具返回值喂给父 LLM。
六、三种编排模式
index.ts 的 execute 函数分派三种模式：
6.1 Single 模式
if (params.agent && params.task) {
const result = await runSingleAgent(
ctx.cwd,
agents,
params.agent,
params.task,
params.cwd,
undefined,
signal,
onUpdate,
makeDetails("single"),
);
const isError = isFailedResult(result);
if (isError) {
const errorMsg = getResultOutput(result);
return {
content: [{ type: "text", text: `Agent ${result.stopReason || "failed"}: ${errorMsg}` }],
details: makeDetails("single")([result]),
isError: true,
};
}
return {
content: [{ type: "text", text: getFinalOutput(result.messages) || "(no output)" }],
details: makeDetails("single")([result]),
};
}
最简单的模式：调一个 agent 跑一个 task。失败时 isError: true，父 LLM 看到错误信息可以决定怎么处理。
6.2 Parallel 模式
并行执行多个独立任务，用 mapWithConcurrencyLimit 控制并发：
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;
}
这是个简洁的并发池实现：N 个 worker 共享一个递增的 nextIndex，每个 worker 抢任务跑直到耗尽。
Parallel 模式的关键约束：
const MAX_PARALLEL_TASKS = 8;
const MAX_CONCURRENCY = 4;
const PER_TASK_OUTPUT_CAP = 50 * 1024;
- 最多 8 个任务（避免一次起太多进程）
- 最多 4 个并发（限制同时跑的子进程数）
- 单任务输出上限 50KB（避免某个子代理输出爆炸把父上下文撑爆）
Parallel 的流式更新会聚合所有任务状态：
const emitParallelUpdate = () => {
if (onUpdate) {
const running = allResults.filter((r) => r.exitCode === -1).length;
const done = allResults.filter((r) => r.exitCode !== -1).length;
onUpdate({
content: [
{ type: "text", text: `Parallel: ${done}/${allResults.length} done, ${running} running...` },
],
details: makeDetails("parallel")([...allResults]),
});
}
};
exitCode === -1 是"还在跑"的哨兵值——SingleResult 初始化时把 exitCode 设为 -1，跑完才填真实值。这是个简单但有效的状态机。
输出聚合用 truncateParallelOutput 按字节截断：
function truncateParallelOutput(output: string): string {
const byteLength = Buffer.byteLength(output, "utf8");
if (byteLength <= PER_TASK_OUTPUT_CAP) return output;
let truncated = output.slice(0, PER_TASK_OUTPUT_CAP);
while (Buffer.byteLength(truncated, "utf8") > PER_TASK_OUTPUT_CAP) {
truncated = truncated.slice(0, -1);
}
return `${truncated}\n\n[Output truncated: ${byteLength - Buffer.byteLength(truncated, "utf8")} bytes omitted. Full output preserved in tool details.]`;
}
注意是按字节不是按字符——多字节字符（中文）可能被截断在中间。这里用 while 循环逐字符回退，确保不超出字节上限。完整输出保留在 details 里供 UI 展开，模型可见的 content 才截断。
6.3 Chain 模式
链式执行：上游输出通过 {previous} 占位符传给下游：
if (params.chain && params.chain.length > 0) {
const results: SingleResult[] = [];
let previousOutput = "";
for (let i = 0; i < params.chain.length; i++) {
const step = params.chain[i];
const taskWithContext = step.task.replace(/\{previous\}/g, previousOutput);
const chainUpdate: OnUpdateCallback | undefined = onUpdate
? (partial) => {
const currentResult = partial.details?.results[0];
if (currentResult) {
const allResults = [...results, currentResult];
onUpdate({
content: partial.content,
details: makeDetails("chain")(allResults),
});
}
}
: undefined;
const result = await runSingleAgent(
ctx.cwd,
agents,
step.agent,
taskWithContext,
step.cwd,
i + 1,
signal,
chainUpdate,
makeDetails("chain"),
);
results.push(result);
const isError = isFailedResult(result);
if (isError) {
const errorMsg = getResultOutput(result);
return {
content: [{ type: "text", text: `Chain stopped at step ${i + 1} (${step.agent}): ${errorMsg}` }],
details: makeDetails("chain")(results),
isError: true,
};
}
previousOutput = getFinalOutput(result.messages);
}
return {
content: [{ type: "text", text: getFinalOutput(results[results.length - 1].messages) || "(no output)" }],
details: makeDetails("chain")(results),
};
}
几个设计点：
{previous}是字符串替换：step.task.replace(/\{previous\}/g, previousOutput)— 简单粗暴，但可移植、可在 prompt 模板里预写- 失败即停：链中任何一步失败，整个链停下，返回
Chain stopped at step N - 流式更新合并历史：
chainUpdate回调把当前正在跑的 step 结果和已完成的 step 结果合并推送，UI 能看到整条链的进度 step字段记录步序：i + 1传给runSingleAgent的step参数，UI 渲染时显示Step 1/Step 2
prompts/implement.md 演示了 chain 模板的写法：
---
description: Full implementation workflow - scout gathers context, planner creates plan, worker implements
---
Use the subagent tool with the chain parameter to execute this workflow:
1. First, use the "scout" agent to find all code relevant to: $@
2. Then, use the "planner" agent to create an implementation plan for "$@"
using the context from the previous step (use {previous} placeholder)
3. Finally, use the "worker" agent to implement the plan from the previous step
(use {previous} placeholder)
Execute this as a chain, passing output between steps via {previous}.
$@ 是用户输入占位符（shell 风格），{previous} 是 chain 内部占位符。模板本身是给父 LLM 看的"使用说明"，父 LLM 读到后会构造对应的 chain 参数调用 subagent 工具。
七、UI 渲染：折叠/展开双视图
renderCall 和 renderResult 是 Pi TUI 协议的两个回调。renderCall 渲染工具调用时的预览，renderResult 渲染工具返回后的结果。
renderResult 根据模式分支渲染，但都遵循"折叠/展开"双视图：
renderResult(result, { expanded }, theme, _context) {
const details = result.details as SubagentDetails | undefined;
// ...
if (details.mode === "single" && details.results.length === 1) {
const r = details.results[0];
const isError = isFailedResult(r);
const icon = isError ? theme.fg("error", "✗") : theme.fg("success", "✓");
const displayItems = getDisplayItems(r.messages);
const finalOutput = getFinalOutput(r.messages);
if (expanded) {
const container = new Container();
// ... 完整渲染：header + task + 所有工具调用 + Markdown 输出 + usage
return container;
}
// 折叠视图：header + 最后 N 个 item + usage
let text = `${icon} ${theme.fg("toolTitle", theme.bold(r.agent))}${theme.fg("muted", ` (${r.agentSource})`)}`;
// ...
text += `\n${renderDisplayItems(displayItems, COLLAPSED_ITEM_COUNT)}`;
if (displayItems.length > COLLAPSED_ITEM_COUNT)
text += `\n${theme.fg("muted", "(Ctrl+O to expand)")}`;
return new Text(text, 0, 0);
}
// ... chain / parallel 分支
}
几细节：
- 状态图标：
✓/✗/⏳（运行中）/◐（部分成功）— 一眼可辨 theme.fg(color, text)— 颜色封装，不直接写 ANSI 码，主题切换友好Markdown组件 — final output 用 Markdown 渲染（展开视图），保留格式formatToolCall— 工具调用格式化为$ command/read ~/path:1-10等可读形式，模仿内置工具的显示Ctrl+O to expand— 提示用户怎么展开，UX 细节
formatToolCall 是个长 switch，把工具调用参数转成单行可读字符串：
function formatToolCall(
toolName: string,
args: Record<string, unknown>,
themeFg: (color: any, text: string) => string,
): string {
const shortenPath = (p: string) => {
const home = os.homedir();
return p.startsWith(home) ? `~${p.slice(home.length)}` : p;
};
switch (toolName) {
case "bash": {
const command = (args.command as string) || "...";
const preview = command.length > 60 ? `${command.slice(0, 60)}...` : command;
return themeFg("muted", "$ ") + themeFg("toolOutput", preview);
}
case "read": {
const rawPath = (args.file_path || args.path || "...") as string;
const filePath = shortenPath(rawPath);
const offset = args.offset as number | undefined;
const limit = args.limit as number | undefined;
let text = themeFg("accent", filePath);
if (offset !== undefined || limit !== undefined) {
const startLine = offset ?? 1;
const endLine = limit !== undefined ? startLine + limit - 1 : "";
text += themeFg("warning", `:${startLine}${endLine ? `-${endLine}` : ""}`);
}
return themeFg("muted", "read ") + text;
}
// ...
}
}
shortenPath 把 home 目录替换为 ~，是终端用户的习惯。read ~/path:1-10 这种格式和 pi 内置工具的显示一致——一致性比创新更重要。
八、安全模型：项目 Agent 的确认流程
execute 函数里有一段容易被忽略但很重要的安全逻辑：
if ((agentScope === "project" || agentScope === "both") && confirmProjectAgents && ctx.hasUI) {
const requestedAgentNames = new Set<string>();
if (params.chain) for (const step of params.chain) requestedAgentNames.add(step.agent);
if (params.tasks) for (const t of params.tasks) requestedAgentNames.add(t.agent);
if (params.agent) requestedAgentNames.add(params.agent);
const projectAgentsRequested = Array.from(requestedAgentNames)
.map((name) => agents.find((a) => a.name === name))
.filter((a): a is AgentConfig => a?.source === "project");
if (projectAgentsRequested.length > 0) {
const names = projectAgentsRequested.map((a) => a.name).join(", ");
const dir = discovery.projectAgentsDir ?? "(unknown)";
const ok = await ctx.ui.confirm(
"Run project-local agents?",
`Agents: ${names}\nSource: ${dir}\n\nProject agents are repo-controlled. Only continue for trusted repositories.`,
);
if (!ok)
return {
content: [{ type: "text", text: "Canceled: project-local agents not approved." }],
details: makeDetails(hasChain ? "chain" : hasTasks ? "parallel" : "single")([]),
};
}
}
逻辑：
- 只有
agentScope包含 project 时才检查 - 只有
confirmProjectAgents为 true（默认）且ctx.hasUI（交互模式）才检查 - 找出本次调用涉及的所有 project-source agent
- 弹确认对话框，列出 agent 名字和来源目录
- 用户拒绝则返回 canceled，不执行
非交互模式下不确认——因为没法弹框。这意味着 CI/脚本场景下用 project agent 不会拦截，需要用户自己保证安全。这是合理的折中：交互场景防误触，自动化场景不挡路。
九、Pi 的进程内 Agent 运行时
除了上文 subagent 示例用到的"spawn 子进程"方式，Pi 在 @earendil-works/pi-agent-core 里还提供了进程内的 Agent 类，不开子进程。
agent.ts 的 Agent 类是个有状态封装：
export class Agent {
private _state: MutableAgentState;
private readonly listeners = new Set<(event: AgentEvent, signal: AbortSignal) => Promise<void> | void>();
private readonly steeringQueue: PendingMessageQueue;
private readonly followUpQueue: PendingMessageQueue;
public convertToLlm: (messages: AgentMessage[]) => Message[] | Promise<Message[]>;
public streamFn: StreamFn;
public beforeToolCall?: ...;
public afterToolCall?: ...;
// ...
constructor(options: AgentOptions = {}) {
this._state = createMutableAgentState(options.initialState);
this.streamFn = options.streamFn ?? streamSimple;
this.steeringQueue = new PendingMessageQueue(options.steeringMode ?? "one-at-a-time");
this.followUpQueue = new PendingMessageQueue(options.followUpMode ?? "one-at-a-time");
// ...
}
async prompt(input: string | AgentMessage | AgentMessage[], images?: ImageContent[]): Promise<void> {
if (this.activeRun) {
throw new Error("Agent is already processing a prompt. Use steer() or followUp() to queue messages.");
}
const messages = this.normalizePromptInput(input, images);
await this.runPromptMessages(messages);
}
steer(message: AgentMessage): void { this.steeringQueue.enqueue(message); }
followUp(message: AgentMessage): void { this.followUpQueue.enqueue(message); }
abort(): void { this.activeRun?.abortController.abort(); }
waitForIdle(): Promise<void> { return this.activeRun?.promise ?? Promise.resolve(); }
}
Agent 类之上还有 AgentHarness（agent-harness.ts），进一步提供：
- Session 持久化：跨进程重启恢复对话
- Compaction：上下文超长时自动压缩历史
- Branch summarization：对话分叉时生成摘要
- Hooks：
before_provider_request/before_provider_payload等钩子 - Skills / PromptTemplates：技能和模板系统
// agent-harness.ts 简化示意
export class AgentHarness<TSkill, TPromptTemplate, TTool> {
readonly env: ExecutionEnv;
private session: Session;
private resources: AgentHarnessResources<TSkill, TPromptTemplate>;
private tools = new Map<string, TTool>();
// ...
private async createTurnState(): Promise<AgentHarnessTurnState<...>> {
const context = await this.session.buildContext();
const resources = this.getResources();
const tools = [...this.tools.values()];
// ...
return {
messages: context.messages,
resources,
streamOptions: cloneStreamOptions(this.streamOptions),
sessionId: sessionMetadata.id,
systemPrompt,
model: this.model,
tools,
activeTools,
};
}
}
AgentHarness 是 Pi 自己 coding-agent 用的运行时——Session 是核心，所有状态都围绕 session 组织。Pi 的主对话就是跑在 AgentHarness 上，遇到需要隔离的子任务时通过 subagent 工具 spawn 一次性子进程跑完拿结果。
十、结语
当你需要在自己的项目里实现"把一个 LLM 拆成多个协作角色"时，pi的示例是比较合适的参考。
示例代码位置：位于 earendil-works/pi 仓库 packages/coding-agent/examples/extensions/subagent/
相关文件索引（路径均相对于 earendil-works/pi 仓库根）：
packages/coding-agent/examples/extensions/subagent/index.ts— 扩展入口、subagent 工具实现、三种编排模式packages/coding-agent/examples/extensions/subagent/agents.ts— Agent 发现与加载、两级 scopepackages/coding-agent/examples/extensions/subagent/agents/scout.md/planner.md/reviewer.md/worker.md— 示例 Agent 定义packages/coding-agent/examples/extensions/subagent/prompts/implement.md/scout-and-plan.md/implement-and-review.md— 工作流模板packages/agent/src/agent.ts— 进程内Agent类packages/agent/src/harness/agent-harness.ts—AgentHarness完整运行时packages/agent/src/harness/types.ts—Skill/PromptTemplate/FileSystem等类型定义
本文由 AIGC 调研生成，仅做研究笔记和参考，真实性需要自行考证。