流式对话：把运行事件接入界面
将智能体运行过程接入流式会话界面，区分文本增量、工具状态与最终结果，并定义取消与等待的行为。
流式对话：把运行事件接入界面
把一次智能体运行变成用户看得见的界面，难点不在于渲染文字，而在于把三类性质完全不同的东西分开：
- 文本增量：模型正在逐字吐字，只有一个方向的追加。
- 工具状态：工具还没跑完，但参数已经知道了，结果还不知道。
- 最终结果：消息定稿、运行结束，这时候才有权威结论。
搞混这三类，最典型的表现是：工具卡片先显示了"成功"，几秒后才蹦出真实输出；或者流式文本一直追加，最后却发现模型中途改了主意。
这篇给出三类东西各自的数据源、事件到界面状态的映射表，以及一个能跑的 HTTP + SSE 实现。读完你能写出一个不会提前宣布成功的流式界面。
需要的前置只有一条：知道 Pi 的会话对象可以订阅事件，并区分 message_update 与 message_end。
三类东西，三个不同的数据源
先把它们各自的来源讲清楚，后面映射表才有依据。
文本增量：message_update + text_delta
message_update 事件的类型是：
// message_update 的形状：外层是会话事件，内层是 assistant 消息内部发生了什么
// 只有 assistant 消息会发这个事件，别的角色不会
type MessageUpdateEvent = {
type: "message_update";
message: AgentMessage;
assistantMessageEvent: AssistantMessageEvent;
};只有 assistant 消息会发这个事件。嵌套的 assistantMessageEvent 是一组判别联合：
// 一条 assistant 消息内部会依次发出这些事件
// 三条内容线各自成对：text_* 是正文，thinking_* 是推理，toolcall_* 是工具请求
// 每个事件都带 contentIndex，指向它在 message.content 数组里的位置
export type AssistantMessageEvent =
| { type: "start"; partial: AssistantMessage }
| { type: "text_start"; contentIndex: number; partial: AssistantMessage }
| { type: "text_delta"; contentIndex: number; delta: string; partial: AssistantMessage }
| { type: "text_end"; contentIndex: number; content: string; partial: AssistantMessage }
| { type: "thinking_start"; contentIndex: number; partial: AssistantMessage }
| { type: "thinking_delta"; contentIndex: number; delta: string; partial: AssistantMessage }
| { type: "thinking_end"; contentIndex: number; content: string; partial: AssistantMessage }
| { type: "toolcall_start"; contentIndex: number; partial: AssistantMessage }
| { type: "toolcall_delta"; contentIndex: number; delta: string; partial: AssistantMessage }
| { type: "toolcall_end"; contentIndex: number; toolCall: ToolCall; partial: AssistantMessage }
| {
type: "done";
// stopReason 有四种：正常停、撞长度上限、要调工具、被推迟
reason: Extract<StopReason, "stop" | "length" | "toolUse" | "deferred">;
message: AssistantMessage;
}
| { type: "error"; reason: Extract<StopReason, "aborted" | "error">; error: AssistantMessage };纯文本增量只需要认 text_delta：
// 纯文本增量只需要认 text_delta 一种事件，其它类型可以一概不处理
if (event.type === "message_update" && event.assistantMessageEvent.type === "text_delta") {
process.stdout.write(event.assistantMessageEvent.delta);
}这就是官方 SDK 示例 01-minimal.ts 的全部内容。
注意 contentIndex。 一个 assistant 消息可能有多个内容块（文本、思考、工具调用）。contentIndex 指向块在 content 数组里的位置。做多块渲染时，必须按 contentIndex 分桶，不能把所有 delta 拼成一条。
text_end 带的 content 是权威值。 官方 JSON 文档的建议是：缓冲 delta 用来做实时显示，但要用 text_end 的 content 替换掉重建出来的数据。同理 message_end 到来时，要用整条 message_end.message 替换掉整个流式消息。
为什么？因为 provider 的流可能和最终消息不一致。协议层的"最终一致"要靠 message_end 兜底。
工具状态：三个事件，三个时刻
工具执行有三个事件：
// 工具执行的三个事件，顺序固定：start →（若干 update）→ end
// start 就带 args：参数此时已校验完，界面可以先画出调用行，但结果还未知
type ToolExecutionEvents =
| { type: "tool_execution_start"; toolCallId: string; toolName: string; args: any }
// update 只在工具主动调 onUpdate 时才有；partialResult 是快照还是增量由工具自己约定
| { type: "tool_execution_update"; toolCallId: string; toolName: string; args: any; partialResult: any }
| {
type: "tool_execution_end";
toolCallId: string;
toolName: string;
// 结果只在这里出现：界面的成功/失败必须等这个事件才能定
result: any;
isError: boolean;
// 工具没跑起来时没有这个字段
durationMs?: number;
};三个关键事实：
参数在 tool_execution_start 就已经全都有了。 args 是模型生成的、已通过 schema 校验的参数。官方 TUI 就是靠它在工具真正跑起来之前就把调用行画出来。
结果只在 tool_execution_end 出现。 result 是完整的工具结果，isError 标记成功失败，durationMs 是执行耗时（用单调时钟测量，工具没跑起来时则没有这个字段）。
tool_execution_update 是可选的。 只有工具主动调用 onUpdate 才会发。官方 json.md 的原话是：
partialResultis the latest partial result supplied by the tool. Whether it replaces or extends an earlier update depends on that tool's result contract.
也就是说，增量是替换还是追加，取决于工具自己的约定。你的界面必须能处理两种：内置 bash 每次上报的是完整输出快照（替换），而自己写的流式工具可能上报的是新追加的片段。
一个稳妥的做法是：把 partialResult 当成最新快照来渲染，并在工具定义里明确约定快照语义。
用 toolCallId 关联三个事件。 这是唯一可靠的关联键。不要用工具名关联——一轮里可能有多个同名工具调用。
最终结果：message_end 与 agent_end
message_end 带的是定稿的完整消息，官方 SDK 文档的原话是：
message_endcontains the authoritative completed message.
会话层的事件比核心层多几个，其中最关键的是 agent_end 和 agent_settled 的区别：
// 两个"运行结束"信号，含义完全不同
// agent_end 只标记一次底层运行结束，Pi 还可能自动重试或继续跑排队的工作
// agent_settled 才代表"Pi 不会自动继续了"，界面放开输入框要等它
type RunLifecycleEvents =
| {
type: "agent_end";
messages: AgentMessage[];
// willRetry 为 true 表示后面还有自动重试要接上
willRetry: boolean;
}
| { type: "agent_settled"; aborted: boolean };agent_end 只是一次底层运行结束了。官方 SDK 文档写得很清楚：
agent_endmarks the end of one low-level agent run, but automatic recovery or queued work can still follow. Useagent_settledwhen the host needs to know that Pi will not continue automatically.
也就是说：agent_end 之后可能还有自动重试、溢出恢复、压缩、引导消息、后续消息。只有 agent_settled 才代表"Pi 不会再自动继续了"。
界面上这个区别很实际：agent_end 时按钮应该还能点（可能马上又开始跑），agent_settled 时才真正可以发下一条消息。
agent_settled 的 aborted 字段则告诉你这次是不是被人为取消的。
事件到界面状态的映射表
这是本篇的核心。左边是事件，右边是界面该怎么变：
最小实现：一个 Node HTTP + SSE 服务
有了映射表，实现就是照着填。下面这份代码用 SSE 把事件推给浏览器。
import { createServer } from "node:http";
import {
createAgentSession,
SessionManager,
type AgentSessionEvent,
} from "@earendil-works/pi-coding-agent";
/** 每个连接的界面状态。 */
interface ClientState {
/** 当前 assistant 消息里各内容块的文本，按 contentIndex 分桶。 */
blocks: Map<number, string>;
/** 正在运行的工具调用。 */
tools: Map<string, { toolName: string; args: unknown; output: string; isError: boolean }>;
}
const clients = new Set<{ res: import("node:http").ServerResponse; state: ClientState }>();
/** 把会话事件转换成界面状态变更。 */
function applyEvent(state: ClientState, event: AgentSessionEvent): void {
switch (event.type) {
case "message_update": {
const inner = event.assistantMessageEvent;
if (inner.type === "text_delta") {
const prev = state.blocks.get(inner.contentIndex) ?? "";
state.blocks.set(inner.contentIndex, prev + inner.delta);
} else if (inner.type === "text_end") {
// text_end 带的 content 是权威值，替换掉累积的 delta
state.blocks.set(inner.contentIndex, inner.content);
}
break;
}
case "tool_execution_start": {
// 参数此刻已经确定，结果还没有
state.tools.set(event.toolCallId, {
toolName: event.toolName,
args: event.args,
output: "",
isError: false,
});
break;
}
case "tool_execution_update": {
const tool = state.tools.get(event.toolCallId);
if (tool) {
const partial = event.partialResult as { content?: Array<{ type: string; text?: string }> };
const text = partial?.content?.find((c) => c.type === "text")?.text ?? "";
// 按快照语义替换
tool.output = text;
}
break;
}
case "tool_execution_end": {
const tool = state.tools.get(event.toolCallId);
if (tool) {
const result = event.result as { content?: Array<{ type: string; text?: string }> };
tool.output = result?.content?.find((c) => c.type === "text")?.text ?? tool.output;
tool.isError = event.isError;
}
break;
}
}
}
const { session } = await createAgentSession({
sessionManager: SessionManager.inMemory(),
});
const unsubscribe = session.subscribe((event) => {
for (const client of clients) {
applyEvent(client.state, event);
// SSE 帧格式：data: 后跟单行 JSON，空行结束一帧
client.res.write(`data: ${JSON.stringify(event)}\n\n`);
}
});
const server = createServer(async (req, res) => {
if (req.method === "GET" && req.url === "/events") {
res.writeHead(200, {
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
Connection: "keep-alive",
});
res.write(": connected\n\n");
clients.add({ res, state: { blocks: new Map(), tools: new Map() } });
req.on("close", () => {
for (const client of clients) {
if (client.res === res) clients.delete(client);
}
});
return;
}
if (req.method === "POST" && req.url === "/prompt") {
const body = await readBody(req);
try {
await session.prompt(body);
} catch (error) {
res.writeHead(500, { "Content-Type": "application/json" });
res.end(JSON.stringify({ error: String(error) }));
return;
}
res.writeHead(200, { "Content-Type": "application/json" });
res.end(JSON.stringify({ ok: true }));
return;
}
if (req.method === "POST" && req.url === "/abort") {
// abort 会停掉当前操作并等待会话空闲
await session.abort();
res.writeHead(200, { "Content-Type": "application/json" });
res.end(JSON.stringify({ ok: true }));
return;
}
res.writeHead(404).end();
});
/** 读出请求体。 */
async function readBody(req: import("node:http").IncomingMessage): Promise<string> {
const chunks: Buffer[] = [];
for await (const chunk of req) chunks.push(chunk as Buffer);
return Buffer.concat(chunks).toString("utf-8");
}
// 进程退出时释放资源
const shutdown = async () => {
server.close();
unsubscribe();
session.dispose();
};
process.on("SIGINT", () => {
void shutdown().then(() => process.exit(0));
});
process.on("SIGTERM", () => {
void shutdown().then(() => process.exit(0));
});
server.listen(3000, () => {
console.log("listening on http://localhost:3000");
});三处释放逻辑值得单独说：
unsubscribe() 取消事件订阅。 session.subscribe 返回一个取消函数。不调用它，闭包会一直持有 clients 集合。
session.dispose() 释放会话。 它会中止在跑的工作、断开与 agent 的连接、清空监听器。
server.close() 停掉 HTTP 服务。 否则进程不会退出。
process.on("SIGINT") 是为了保证 Ctrl+C 时这三件事都做掉。少任何一件，要么进程挂住，要么连接泄漏。
取消按钮和等待按钮
界面上的两个按钮对应两个不同的方法，别用混。
import type { AgentSession } from "@earendil-works/pi-coding-agent";
/** 取消按钮：停掉当前操作，并等待会话进入空闲。 */
async function onCancelClick(session: AgentSession): Promise<void> {
// abort 会中断运行中的模型请求和工具执行
await session.abort();
}
/** 等待按钮：只等待，不取消。 */
async function onWaitClick(session: AgentSession): Promise<void> {
// waitForIdle 不打断任何东西：模型会继续跑完，你只是在等
await session.waitForIdle();
}abort() 的实现是三步：设置中止标志、调用 agent.abort() 触发运行中的 AbortSignal、然后 await this.waitForIdle()。
waitForIdle() 的语义更微妙，源码注释写着：
class AgentSession {
/**
* Whether the session has no active agent run, compaction, branch summary, retry, or queued continuation.
*/
get isIdle(): boolean {
return !this._isAgentRunActive && !this.isCompacting;
}
}它不只看"当前运行"，还包括压缩、分支摘要、重试和排队的后续工作。所以"等它跑完"用 waitForIdle() 是对的。
一个常见错误是用 waitForIdle() 来代替取消。 waitForIdle() 不会中断任何东西——模型会继续跑完，工具会继续执行。你只是在等它跑完。
浏览器端的一个最小消费者
SSE 的浏览器端比想象中简单，EventSource 内置支持：
<!doctype html>
<html lang="zh">
<head>
<meta charset="utf-8" />
<title>Pi 流式会话</title>
</head>
<body>
<pre id="log"></pre>
<form id="form">
<input id="input" autocomplete="off" placeholder="输入任务…" />
<button type="submit">发送</button>
<button type="button" id="abort">取消</button>
</form>
<script type="module">
const log = document.getElementById("log");
const input = document.getElementById("input");
const source = new EventSource("/events");
source.onmessage = (event) => {
const payload = JSON.parse(event.data);
// 文本增量
if (payload.type === "message_update" && payload.assistantMessageEvent?.type === "text_delta") {
log.textContent += payload.assistantMessageEvent.delta;
}
// 工具开始：参数已知，结果未知
if (payload.type === "tool_execution_start") {
log.textContent += `\n[工具开始] ${payload.toolName} ${JSON.stringify(payload.args)}\n`;
}
// 工具结束：结果已知
if (payload.type === "tool_execution_end") {
log.textContent += `[工具结束] ${payload.toolName} isError=${payload.isError} 耗时=${payload.durationMs ?? "-"}ms\n`;
}
// 真正结束
if (payload.type === "agent_settled") {
log.textContent += payload.aborted ? "\n[已取消]\n" : "\n[已完成]\n";
}
};
document.getElementById("form").addEventListener("submit", async (e) => {
e.preventDefault();
await fetch("/prompt", { method: "POST", body: input.value });
input.value = "";
});
document.getElementById("abort").addEventListener("click", async () => {
await fetch("/abort", { method: "POST" });
});
</script>
</body>
</html>宿主负责什么
第一行和倒数第二行最容易出问题。订阅必须在 prompt() 之前。 官方 RPC 文档有一段专门的提醒：
Subscribe before sending a prompt to avoid missing a fast completion.
模型响应快的话，prompt() 返回前事件就已经发完了。晚订阅会丢掉整段流式输出。
关于背压：message_update 在长回复里可能每秒几十次。浏览器端应该把 DOM 更新合并到动画帧里，而不是每个 delta 改一次 DOM。
三种场景的可观察结果
取消场景有个容易忽略的细节：已经流出来的文本要留着。 用户取消不代表之前的输出无效。官方 TUI 就是这么做的——它会显示错误信息但保留已渲染的内容。
你现在能做什么
- 写出一个订阅器，按
contentIndex分桶累积text_delta，并在text_end到达时用权威内容替换累积值。 - 在工具卡片上验证三件事：
tool_execution_start时参数已可显示、tool_execution_update只更新输出区、tool_execution_end才定稿成功失败。 - 用
toolCallId关联同一次工具调用的三个事件，跑一轮并行工具调用确认没有串。 - 用
agent_settled（而不是agent_end）决定何时放开输入框，并用它的aborted字段区分正常结束与被取消。 - 把上面的 SSE 服务跑起来，用浏览器端消费者验证流式文本、工具卡片和取消按钮的行为，并确认 Ctrl+C 时进程能干净退出。