v1.1.0Pi v1.1.0 源码分析Analysis of Pi v1.1.0 · 2026-10-07

agent-coreagent-core

通用 Agent:状态、事件循环、工具执行与钩子General-purpose agent with transport abstraction, state management, and attachment support

pi-agent-corepi-ai

← → 翻页 · F 全屏 · O 目录 · T 明暗 · L 语言← → navigate · F fullscreen · O overview · T theme · L language

图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.001
01 · 模块结构01 · Module structurepi-agent-core

模块结构Module structure

pi-agent-core modulessrc/6 个文件6 files@earendil-works/pi-ai唯一的运行时依赖the only runtime dependencypi-coding-agentAgentSession · sdk.ts你的应用your appnew Agent({ streamFn })宿主运行时host runtimesetDefaultStreamFn(fn)test/4 个文件 · 77 个用例4 files · 77 casesindex.tsexport *613 Lagent.tsclass Agent · AgentOptionsPendingMessageQueue ×2(私有)PendingMessageQueue ×2 (private)949 Lagent-loop.tsagentLoop() · agentLoopContinue()runAgentLoop*() · runToolCall()20 Lstream-fn.tssetDefaultStreamFn()(导出)setDefaultStreamFn() (exported)getDefaultStreamFn()(内部)getDefaultStreamFn() (internal)393 Lproxy.tsstreamProxy(model, ctx, opts)ProxyAssistantMessageEvent539 Ltypes.tsAgentEvent · AgentTool · AgentStateAgentLoopConfig · StreamFn · hooksutils/transcript.tscreateInitialSystemMessage · getCurrentToolsgetToolStateChanges · toToolDeclarationutils/validation.tsvalidateToolArguments(tool, toolCall)TypeBox 模式校验 · 失败即抛出TypeBox schema check · throws on failureutils/event-stream.tsEventStream<T, R> → agentLoop() 的返回值EventStream<T, R> → what agentLoop() returnsAssistantMessageEventStream → streamProxy()types.tsModel · Message · Tool · UsageSimpleStreamOptions · TranscriptContextAgentagentLoopagentLoopContinuerunAgentLooprunAgentLoopContinuerunToolCallstreamProxysetDefaultStreamFn使用方CONSUMERS入口ENTRYrunAgentLoop()未传 streamFn 时when streamFn is omitted值导出Value exports+ 35 个导出类型+ 35 exported types主路径main pathimport / 再导出import / re-exportpi-agent-corepi-aipi-coding-agent其他other

index.ts 只有 5 行:export * 四个模块,再单独导出 setDefaultStreamFn(getDefaultStreamFn 不导出)。index.ts is five lines: export * from four modules plus setDefaultStreamFn alone (getDefaultStreamFn stays internal).

图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.002
02 · 运行生命周期02 · Run lifecyclepi-agent-core

运行生命周期Run lifecycle

Agent run state machine守卫与约束(agent.ts)GUARDS (agent.ts)abort()中止信号 → 流与工具signal → stream & tools空闲IdleactiveRun = undefinedrunWithLifecycle()isStreaming = truenew AbortController()流式响应Streaming responsestreamingMessage = partialmessage_start · _update收尾Settlingawait agent_end 监听器await agent_end listenersisStreaming 仍为 trueisStreaming still true执行工具Executing toolspendingToolCalls ∪ {id}tool_execution_start · _endfinishRun()isStreaming = false清空状态 · resolve 等待者clear state · resolve waitershandleRunFailure()合成 error 助手消息synthesize error messagemessage_* · turn_end · agent_endprompt · continuetoolCallturn_startagent_end监听器全部完成listeners settledexecutor 抛出executor throwsfinally再次空闲 · waitForIdle() 返回idle again · waitForIdle() resolvesprompt()运行中调用 → 抛错,提示改用 steer() / followUp()while running → throws: use steer() / followUp()reset()运行中调用 → 抛错;空闲时保留 system 基线、清空队列throws while running; idle: keeps the system baseline, clears queuescontinue()末条为 assistant → 先排空 steering,再 follow-up,否则抛错last = assistant → drain steering, else follow-ups, else throwsubscribe()监听器按订阅顺序 await,并拿到本次运行的 signallisteners awaited in order, each gets the run signal12345

prompt() 在运行中调用会抛错,提示改用 steer() 或 followUp()——这是唯一的并发入口约束。prompt() throws while a run is active, pointing at steer() or followUp() — the single concurrency rule at the entry.

图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.003
03 · 钩子调用顺序03 · Hook call orderpi-agent-core

钩子调用顺序Hook call order

  • transformContext 作用于 AgentMessage[](可含自定义消息),convertToLlm 再把它们转成 pi-ai 的 Message[]——压缩、裁剪放前者,格式转换放后者。transformContext works on AgentMessage[] (custom messages included); convertToLlm then turns them into pi-ai Message[] — pruning belongs in the former, format conversion in the latter.
  • getApiKey 每次请求都调用,让短期 OAuth token 在长时间工具执行期间也能刷新。getApiKey is called on every request, so short-lived OAuth tokens can refresh during long tool runs.
Hook call order in one runrunLoop()AgentLoopConfig 钩子AgentLoopConfig hooksstreamFn · pi-aiAgentToolemit → AgentEventagent_start · turn_start · message_*(user)message_start (assistant)message_update ×nmessage_end (assistant)tool_execution_starttool_execution_update ×ntool_execution_end · message_*(toolResult)turn_endturn_startmessage_start … message_endturn_end · agent_end1 · 开始1 · Start2 · 请求 12 · Request 13 · 工具3 · Tools4 · 轮次结束4 · End of turn5 · 请求 2 → 结束5 · Request 2 → enddeclareToolChanges(prompts)getSteeringMessages()prepareRequest(req, signal)transformContext(messages)convertToLlm(messages)getApiKey(provider)streamFn(model, normalizeContext(llm), opts)text_* · thinking_* · toolcall_*done(toolUse)prepareArguments() → validatebeforeToolCall(ctx, signal)execute(id, args, signal, onUpdate)onUpdate(partial) … resultafterToolCall(ctx, signal)finishTurn(turn, signal)getSteeringMessages()prepareNextTurn(turn)prepareRequest → … → getApiKeystreamFn(…)done(stop)finishTurn → getFollowUpMessageserror / aborted → finishTurn → turn_end → agent_end,直接退出error / aborted → finishTurn → turn_end → agent_end, hard exit123456789101112131415161718192021调用call返回 / 流事件return / stream event1顺序编号,颜色 = 发起方order · colour = caller
图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.004
04 · AgentEvent 事件流04 · The AgentEvent streampi-agent-core

AgentEvent 事件流The AgentEvent stream

AgentEvent streamagent_start … agent_end一次运行one runturn_start … turn_end第 1 轮:一次响应 + 其工具turn 1 · one response + its toolsturn_start … turn_end第 2 轮:最终回答turn 2 · final answerAgent.processEvents()先归约,再 await 监听器reduce, then await listenerspi-ai 事件 → AgentEvent(streamAssistantResponse)pi-ai events → AgentEvent (streamAssistantResponse)message_startmessage_endmessage_startmessage_update ×nmessage_endtool_execution_starttool_execution_update ×ntool_execution_endmessage_startmessage_endmessage_startmessage_update ×nmessage_endmessage_start · _updatemessage_endtool_execution_starttool_execution_endturn_endagent_endstarttext_starttext_deltatext_endthinking_startthinking_deltathinking_endtoolcall_starttoolcall_deltatoolcall_enddoneerrormessage_startpush(partial)message_update{ message: {...partial}, assistantMessageEvent }message_endresult() + thinkingLeveluser 提示user promptassistant每个工具调用per tool calltoolResultassistantstopReason: "stop"→ streamingMessage = message→ messages.push(message)→ pendingToolCalls ∪ {id}→ pendingToolCalls ∖ {id}→ errorMessage ← message.errorMessage→ streamingMessage = undefinedpendingToolCalls 每次替换为新 Set(便于 UI 比较)pendingToolCalls is replaced with a new Set each time若流在 start 之前就以 error 结束,循环会补发 message_start;最终消息替换上下文中的 partialif the stream ends before start, the loop emits message_start itself; the final message replaces the partial in context映射maps touser 消息user messageassistant 消息assistant message工具执行tool executiontoolResult 消息toolResult message

message_update 同时携带 assistantMessageEvent(原始 pi-ai 事件),UI 可以按 delta 精细渲染。message_update also carries assistantMessageEvent (the raw pi-ai event), so a UI can render per delta.

图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.005
05 · Steering 与 Follow-up05 · Steering and follow-uppi-coding-agent

Steering 与 Follow-upSteering and follow-up

Steering and follow-up queuessteeringQueuePendingMessageQueuefollowUpQueuePendingMessageQueue队列 API(agent.ts)QUEUE API (agent.ts)agent.steer(msg)运行中插话nudge mid-runagent.followUp(msg)做完之后再…after you finish…m1m2m3m1m2m3① 运行开始① run start初次轮询initial poll② turn_end 之后② after turn_end除非 finishTurn → endunless finishTurn → end③ prepareNextTurn之后 · 仅当 ② 为空after it · only if ② empty④ 本该停止时④ would stopgetFollowUpMessages()steer()followUp()clearSteeringQueue()clearFollowUpQueue()clearAllQueues()hasQueuedMessages()peekQueuedMessages()steeringModefollowUpMode← enqueue()one-at-a-time(默认)→ drain() 取 [m1]one-at-a-time (default) → drain() takes [m1]all → drain() 取 [m1, m2, m3]all → drain() takes [m1, m2, m3]← enqueue()one-at-a-time(默认)→ drain() 取 [m1]one-at-a-time (default) → drain() takes [m1]all → drain() 取 [m1, m2, m3]all → drain() takes [m1, m2, m3]getSteeringMessages()外层循环重新开始restarts the outer loop请求 + 工具request + toolsturn_end下一轮next turn一次运行的时间轴timeline of one runcontinue() 从 assistant 末尾恢复且有 steering 时,① 会被跳过一次(skipInitialSteeringPoll)continue() resuming from an assistant tail with steering skips ① once (skipInitialSteeringPoll)peekQueuedMessages() 优先返回 steering,否则 follow-up;reset() 会清空两个队列peekQueuedMessages() returns steering first, else follow-ups; reset() clears both

默认模式都是 one-at-a-time:每个轮询点只取一条,剩下的留到下一轮,避免一次塞入大量插话。Both default to one-at-a-time: each poll takes one message and leaves the rest for later turns, so a burst of nudges doesn't arrive at once.

图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.006
06 · 工具批的并发与顺序06 · Tool batches: concurrency and orderpi-agent-core

工具批的并发与顺序Tool batches: concurrency and order

Timing of a tool batchparallel(默认)parallel (default)准备按顺序 · 执行并发 · 结果按原顺序prepare in order · execute concurrently · results in source ordersequentialtoolExecution = "sequential",或批内任一工具声明 executionMode: "sequential"toolExecution = "sequential", or any tool in the batch declares executionMode: "sequential"A · readB · bashC · grepprepexecute · 120 msprepexecute · 380 mspreptool_execution_end #1tool_execution_end #2tool_execution_end #3消息messagesmsg Amsg Bmsg CPromise.all 之后按 A → B → C 发出after Promise.all · in A → B → C order0 ms100 ms200 ms300 ms400 ms500 ms600 ms700 ms800 msA · read120 msB · bash380 msC · grep≈ 770 ms:每个调用 start → … → message 后才开始下一个≈ 770 ms: each call finishes start → … → message before the next0 ms100 ms200 ms300 ms400 ms500 ms600 ms700 ms800 ms立即结果(未找到 / 校验失败 / 被阻止)在准备阶段就发 tool_execution_end;中止后剩余调用得到 "Operation aborted"immediate outcomes (not found / invalid / blocked) emit tool_execution_end during prepare; after abort the rest get "Operation aborted"terminate:只有批内每个最终结果都设 terminate: true,循环才在本批后停止terminate: the loop stops after the batch only if every finalized result sets terminate: true准备:查找 · prepareArguments · 校验 · beforeToolCall;以及 afterToolCallprepare: find · prepareArguments · validate · beforeToolCall; and afterToolCalltool.execute()ToolResultMessage

立即结果(未知工具、参数校验失败、被 beforeToolCall 阻止)不进入执行,直接得到 error 结果。Immediate outcomes (unknown tool, failed validation, blocked by beforeToolCall) skip execution and get an error result directly.

图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.007
07 · StreamFn 与代理07 · StreamFn and the proxypi-agent-core

StreamFn 与代理StreamFn and the proxy

StreamFn and the proxy streamAgentOptions.streamFn构造时传入passed at constructionagent.streamFunction传给 runAgentLoop()handed to runAgentLoop()getDefaultStreamFn()未设置时抛错throws if none setsetDefaultStreamFn()由宿主安装installed by a hostStreamFn(model, TranscriptContext, SimpleStreamOptions?)→ AssistantMessageEventStream | Promise<…>约定:不抛异常 · 失败编码为 error 事件 + stopReason "error" / "aborted"contract: never throw · failures = error event + stopReason "error" / "aborted"models.streamSimplepi-ai Models · coding-agent 用 ModelRuntimepi-ai Models · coding-agent uses ModelRuntimestreamProxy(model, ctx, opts)proxy.ts · { authToken, proxyUrl }POST {proxyUrl}/api/streamAuthorization: Bearer <authToken>你的代理服务your proxy server运行 pi-airuns pi-aidata: {event}SSE · 无 partialSSE · no partialprocessProxyEvent(e, partial)在客户端重建 partialrebuild partial client-sideAssistantMessageEventStream与直连相同 · 带 durationMs(1.1.0)same as direct · gets durationMs (1.1.0)??直接调用direct经服务端via a serverstreambody: { model, context, options }options = 11 serializable fields:temperature · samplingParams · maxTokensreasoning · cacheRetention · sessionIdheaders · metadata · transportthinkingBudgets · maxRetryDelayMs失败 → error 事件:failures → error event:!response.ok → "Proxy error: …" · 无 done/error 的 EOF → "Connection closed by proxy server…" · abort → reader.cancel() → aborted!response.ok → "Proxy error: …" · EOF without done/error → "Connection closed by proxy server…" · abort → reader.cancel() → aborted1234

解析顺序:AgentOptions.streamFn → setDefaultStreamFn() 安装的全局默认 → 都没有则抛出 “No default stream function configured”。Resolution: AgentOptions.streamFn → the global default from setDefaultStreamFn() → otherwise throw “No default stream function configured”.

图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.008
08 · 数据类型08 · Data typespi-agent-core

数据类型Data types

pi-agent-core data types«interface»AgentState+systemPrompt (readonly)+model · thinkingLevel+tools · messages (copy on set)+isStreaming · streamingMessage?+pendingToolCalls: Set<string>+errorMessage?«interface»AgentContext+messages: AgentMessage[]+tools?: AgentTool[]«union»AgentMessage|Message (pi-ai)|CustomAgentMessages[K]«merge»CustomAgentMessages声明合并扩展extend by declaration merging如 artifact · notificatione.g. artifact · notification«pi-ai»Message|SystemMessage · UserMessage|AssistantMessage|ToolResultMessage«extends Tool»AgentTool+name · description · parameters+label · outputSchema?+executionMode? · replay?+prepareArguments?(args)+execute(id, params, signal, onUpdate) → Promise<Result>AgentToolResult<T>+content: (Text | Image)[]+details: T+structuredContent? · usage?+isError? · terminate?AgentToolCallOutcome+toolCall · result · isError+durationMs? (1.1.0)«interface»AgentLoopConfig+…SimpleStreamOptions+model+convertToLlm (required)+transformContext? · getApiKey?+prepareRequest? · prepareNextTurn?+finishTurn? → end | continue+beforeToolCall? · afterToolCall?+getSteeringMessages?+getFollowUpMessages?+toolExecution? = "parallel"BeforeToolCallResult+block? · reason?+terminate?AfterToolCallResult+content? · details? · isError?+usage? · terminate?+structuredContent?快照snapshottoolsexecute 返回execute returns钩子返回hooks return持有 / 返回owns / returns派生 / 联合成员derived / union memberpi-agent-core由宿主提供provided by the host钩子结果hook resultspi-ai

AgentState 的 tools 与 messages 赋值时复制数组;pendingToolCalls 每次替换为新的 Set,方便 UI 做引用比较。AgentState.tools and messages copy on assignment; pendingToolCalls is replaced with a new Set each time so UIs can compare by reference.

图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.009
扩展点Extension pointspi-coding-agent

agent-core 的扩展点Where to plug into agent-core

1
streamFnstreamFn

替换模型调用:直连、代理、录制回放、测试桩。Swap the model call: direct, proxy, record/replay, test stubs.

2
convertToLlm · transformContextconvertToLlm · transformContext

自定义消息 → LLM 消息;压缩、裁剪、注入上下文。Custom messages → LLM messages; compaction, pruning, context injection.

3
beforeToolCall · afterToolCallbeforeToolCall · afterToolCall

权限审批、阻止、改写结果、终止。Approval, blocking, rewriting results, terminating.

4
prepareRequest · prepareNextTurn · finishTurnprepareRequest · prepareNextTurn · finishTurn

每轮改模型 / 工具 / 选项,决定是否继续。Change model / tools / options per turn and decide whether to continue.

5
getApiKeygetApiKey

每次请求取凭据(短期 OAuth token)。Fetch credentials per request (short-lived OAuth tokens).

图解 Pi Agent · Pi v1.1.0Pi Agent, Illustrated · Pi v1.1.010
←
1 / 10