Agent Core：状态、工具执行与事件循环

智能体表现不对的时候——工具没执行、提示没生效、消息重复了——你要能指出它卡在状态的哪一步，而不是靠猜。

这一层专门为这件事存在。Agent Core 的公开接口不多，但每个字段都对应运行中一个明确的环节：它持有什么状态、状态怎么变、事件怎么发。

读完这篇你能说出 Agent 构造时必须传什么、可以传什么，每个参数影响哪一段行为；读 agent.state 时能说出每个字段由谁写入、什么时候更新；也能解释"改了 agent.state.tools 但模型不知道"这类现象的原因。

读它需要会用 TypeScript，并且已经跑通过一个带工具的 Agent。你还要有耐心读一类看起来平淡但很关键的代码——事件从模型流到宿主状态的那一段。

最终要跑通什么

1. 能说出 Agent 构造时必须传什么、可以传什么，每个参数影响哪一段行为。
2. 读 agent.state 时，能说出每个字段由谁写入、什么时候更新。
3. 能解释"改了 agent.state.tools 但模型不知道"这类现象的原因。
4. 能对照事件表，说出一次运行里状态变化发生的顺序。

第 3 点是验收点。它是这套设计里最反直觉的地方。

Agent 的构造参数

AgentOptions 定义在 Agent Core 的实现文件里。

必填项

streamFn 的类型是 (model, context, options?) => AssistantMessageEventStream。它是唯一的模型访问接缝，models.streamSimple.bind(models) 直接满足它。

streamFn 收到的是规范化后的 transcript：系统提示和工具声明在 transcript 的系统消息里，不在 context.systemPrompt 或 context.tools 上。

它的契约有三条硬要求：不抛异常、不返回 rejected promise；必须返回 AssistantMessageEventStream；失败必须编码进流——用协议事件加一个 stopReason 为 "error" 或 "aborted" 且带 errorMessage 的最终消息。

不传 streamFn 会走 getDefaultStreamFn()，而它在没人调过 setDefaultStreamFn() 时会抛错。所以显式传。

初始状态

initialState 的类型是 Partial<Omit<AgentState, "pendingToolCalls" | "isStreaming" | "streamingMessage" | "errorMessage">>，也就是五个字段：systemPrompt（系统提示）、model（使用的模型）、thinkingLevel（off / minimal / low / medium / high / xhigh / max）、tools（可执行工具装载）、messages（已有的对话记录）。

systemPrompt 和 tools 会变成首条系统消息，除非 messages 已经以一条系统消息开头。 这条规则决定了后面所有关于"提示和工具归谁管"的行为。

消息转换

convertToLlm（(messages) => Message[] | Promise<Message[]>）过滤 UI 专用消息、把自定义类型转成 LLM 能懂的消息，不传则用默认实现。transformContext（(messages, signal?) => Promise<AgentMessage[]>）在它之前跑，用于裁剪历史、压缩上下文。

处理顺序固定：AgentMessage[] → transformContext() → AgentMessage[] → convertToLlm() → Message[] → LLM。

两个 hook 都有一个共同契约：不得抛异常或 reject。 抛异常会中断低层循环，产生不出正常的事件序列。需要兜底就返回原消息或另一个安全值。

工具执行

beforeToolCall 拿到的 context 有四个字段：assistantMessage（发起这次调用的助手消息）、toolCall（原始工具调用块）、args（按 schema 校验过的参数）、context（当前的 Agent 上下文）。

afterToolCall 的合并语义是逐字段的：content、details、isError、usage、terminate 各自独立，structuredContent 只有在给了 content 而没给它时会被丢弃。没有深合并。

请求与回合控制

prepareRequest（(request, signal?) => Promise<AgentRequestUpdate | undefined>）在每次对话式 provider 请求前运行，含第一次。finishTurn（(turn, signal?) => Promise<AgentTurnDecision | undefined>）在助手消息和所有工具结果定稿后、turn_end 之前运行。prepareNextTurn 和 prepareNextTurnWithContext（后者能拿到上一轮结束的上下文）在决定要开始下一轮之后、下一轮开始之前运行。

prepareRequest 和 finishTurn 的确切调用时机和语义，见本文后面的"请求准备与回合结束"一节。

其它

getApiKey（每次模型调用前动态解析 key，适合会过期的 OAuth token）、sessionId（传给 provider 用于缓存感知）、thinkingBudgets（各思考级别的 token 预算）、transport（默认 "auto"）、maxRetryDelayMs（provider 请求的延迟上限）、steeringMode / followUpMode（两个队列的投递模式，默认都是 "one-at-a-time"）、onPayload / onResponse / onProviderStreamEvent（三个观察 provider 交互的回调）。

getApiKey 与两个消息转换 hook 有同一条契约：不得抛异常或 reject。

AgentState：运行时可读的那一面

agent.state 返回一个 AgentState。宿主可写的是 model、thinkingLevel、tools、messages；运行时只读维护的是 systemPrompt（从 transcript 回放出来的当前系统提示）、isStreaming（运行期间为 true，直到被 await 的 agent_end 监听器跑完）、streamingMessage（当前的部分助手消息）、pendingToolCalls（正在执行的工具调用 id 集合）、errorMessage（最近一次失败或中止的助手消息的错误文本）。

官方 README 里给的 AgentState 接口示例漏了 systemPrompt（同一篇的正文字段说明里又提到了它）。源码的类型定义里有这个字段，它是只读的回放结果。以源码为准。

tools 和 messages 用访问器属性实现：赋值时复制顶层数组，所以外部再改原数组不会影响 Agent 状态；但读回来的数组改了，会改到 Agent 状态。

状态更新的七个时机

message_start 把 streamingMessage 设为这条消息，message_update 把它更新为部分消息，message_end 清空它并把消息追加进 state.messages。tool_execution_start 往 pendingToolCalls 加入 toolCallId，tool_execution_end 把它移除。turn_end 在助手消息带 errorMessage 时写入 state.errorMessage，agent_end 清空 streamingMessage。

isStreaming 的生命周期比 agent_end 更长。运行真正结束（finishRun()）发生在 executor 返回之后，而 executor 要等所有被 await 的 agent_end 监听器跑完。

一个不常见的失败路径

如果执行器本身抛异常（不是流上的错误），Agent 会自己合成一条失败消息：role 是 "assistant"，content 是空文本，stopReason 按是否被中止取 "aborted" 或 "error"，errorMessage 取异常信息，usage 是全零。然后依次发 message_start、message_end、turn_end、agent_end。宿主从事件流上看到的形状和正常失败完全一致。

transcript 拥有系统提示与工具声明

这是本篇最需要记住的一条规则。

系统提示和工具声明属于 transcript，不属于 Agent 的某个独立字段。

• transcript 的首条系统消息就是系统提示。
• 后续系统消息按名字给提示打补丁（sections），并通过 toolsAdded / toolsRemoved 改工具集。
• agent.state.systemPrompt 是只读的回放结果，由 getCurrentSystemPrompt(messages) 从 transcript 算出来。
• getCurrentSystemMessage(messages) 返回回放后的头部，包含声明的工具集，对任意消息数组都成立——包括带自定义消息角色的 Agent transcript。

于是"可执行装载"和"模型可见声明"就分成了两样东西：agent.state.tools 是真正能跑的 AgentTool 对象，由宿主维护；transcript 里的系统消息是模型看得见的工具清单，由循环在每次请求前维护。

循环怎么维护这个差异

declareToolChanges() 在每次请求前对比两者：一边是 transcript 里通过系统消息声明的工具（getCurrentTools()），另一边是可执行装载（context.tools 映射成的声明）。getToolStateChanges() 的规则是：一个工具的定义变了，等于先移除再加。

差异的处理有两种情况：存在一条待发的系统消息时，它的工具字段被替换成差异（有待发消息时，它的工具字段视为意图，而不是已提交的事实）；不存在时，插入一条新的系统消息，放在第一条非系统消息之前。

所以你改 agent.state.tools 之后，模型不会立刻知道，但下一次请求之前，循环会用一条系统消息把差异播报出去：

// 让模型知道新增了一个工具
agent.state.tools = [...agent.state.tools, myNewTool];
// 下一轮请求前，循环会插入一条带 toolsAdded 的系统消息

反过来，只改 transcript 不改 agent.state.tools，模型看得见但工具跑不了。这是最容易踩的一类不一致。

工具执行如何进入下一轮模型输入

工具结果变成下一轮模型输入，路径是这样的：

# 工具结果进入下一轮模型输入的路径，箭头方向就是执行顺序
助手消息（含 toolCall 块）
  → tool_execution_start
  → [工具执行，可能有 tool_execution_update]
  → tool_execution_end（带 result 和 isError）
  → message_start / message_end（toolResult 消息）
  → turn_end（message + toolResults）
  → 下一轮请求

工具结果消息由 createToolResultMessage() 构造：role 固定 "toolResult"，toolCallId / toolName 来自工具调用块，content 是工具结果的内容数组（缺省成空数组），details 是结构化细节，usage 是工具自身的用量，isError 标记是否按错误处理，durationMs 是执行耗时。

content 的缺省处理值得注意：未类型化的工具（JS 扩展）可能返回没有 content 的结果，循环会把它归一成空数组，避免 null 进入会话历史或 provider 请求体。

这条消息进入下一轮模型输入时和其它消息一样走 transformContext() → convertToLlm()。它的 role 是 "toolResult"，默认的 convertToLlm 会放行它。

并行与顺序

parallel（默认）模式下，预检阶段串行跑参数校验和 beforeToolCall，然后并发执行允许的工具；sequential 模式则一次一个地准备、执行、收尾。

并行模式下，两类事件的顺序不一样：tool_execution_end 按工具完成顺序发，但工具结果消息和 turn_end.toolResults 仍按助手消息里的原始顺序。这条规则保证了持久化的对话记录顺序稳定。

一条覆盖规则：一批工具调用里只要有任意一个指向 executionMode: "sequential" 的工具，整批都按顺序执行，不管全局设置是什么。

工具出错怎么变成 isError

官方措辞很明确：工具失败时抛异常，不要把错误信息当普通内容返回。

// 工具的失败路径：抛异常，运行时才会把它转成 isError: true 的工具结果
execute: async (toolCallId, params, signal, onUpdate) => {
	if (!existsSync(params.path)) {
		throw new Error(`File not found: ${params.path}`);
	}
	return { content: [{ type: "text", text: "..." }] };
}

运行时捕获异常，生成一条错误工具结果：createErrorToolResult() 的内容是 [{ type: "text", text: <错误文本> }]，details 是空对象，isError 为 true。

模型看到的是一个"这次调用失败了，原因是 X"，可以据此修正参数重试。反过来，如果你把错误写进 content 里返回，模型会当成正常结果，循环也就不会自然纠错。

同样会变成 isError: true 的还有几种情况：工具不存在（Tool <名字> not found）、参数校验失败、beforeToolCall 拦截（拦截时给的 reason，或默认的 Tool execution was blocked）、操作被中止（Operation aborted）、响应撞到输出上限（提示参数可能被截断，请用完整参数重新发起）。

事件序列

AgentEvent 的完整集合：agent_start（一次运行开始）、agent_end（这次运行结束，带 messages）、turn_start（一个回合开始）、turn_end（一个回合结束，带 message 和 toolResults）、message_start / message_end（任意消息开始和完成，带 message）、message_update（仅助手消息，带 message 和 assistantMessageEvent）、tool_execution_start（带 toolCallId、toolName、args）、tool_execution_update（多一个 partialResult）、tool_execution_end（多 result、isError、durationMs?）。

不带工具调用的一次 prompt()：

# 一次 prompt() 的事件序列，缩进表示嵌套层级
agent_start
├─ turn_start
├─ message_start   { message: userMessage }
├─ message_end     { message: userMessage }
├─ message_start   { message: assistantMessage }
├─ message_update  { assistantMessageEvent: text_delta }    ← 可能很多次
├─ message_end     { message: assistantMessage }
├─ turn_end        { message, toolResults: [] }
└─ agent_end       { messages: [...] }

带工具调用时，工具事件插在助手消息结束之后、工具结果消息之前：

  message_end            { message: assistantMessage with toolCall }
  tool_execution_start   { toolCallId, toolName, args }
  tool_execution_update  { partialResult }        ← 工具自己流式上报时才有
  tool_execution_end     { toolCallId, result, isError, durationMs }
  message_start / end    { message: toolResultMessage }
  turn_end               { message, toolResults: [toolResult] }

订阅的语义

agent.subscribe(listener) 返回一个退订函数。监听器有三条例外规则：按注册顺序被 await（不是并发触发）；计入这次运行的结算（await agent.prompt(...) 要等所有被 await 的监听器跑完）；agent_end 是最后一个事件，但它不等于运行结束。

第三条最容易被误读。agent_end 表示不会再有循环事件，但 agent.state.isStreaming 在被 await 的 agent_end 监听器跑完之前仍然是 true。

监听器还会收到当前运行的 abort 信号，所以收尾工作可以感知取消：

const unsubscribe = agent.subscribe(async (event, signal) => {
	if (event.type === "agent_end") {
		// 运行结束前的最后一道收尾工作
		await flushSessionState(signal);
	}
});

processEvents() 里有一处硬失败：如果监听器在活动运行之外被调用，会抛 Agent listener invoked outside active run。所以监听器要在 prompt() 之前注册。

停止、中断与追加输入的契约

prompt(message | messages | string, images?) 开始一次新运行，运行期间再调用会抛错。continue() 从当前 transcript 继续，最后一条必须是 user 或 toolResult。steer(message) 排在当前助手轮次之后注入，followUp(message) 排在智能体做完待办之后注入。peekQueuedMessages() 预览下一轮会被选中的消息但不消费，hasQueuedMessages() 报告队列是否非空，clearSteeringQueue() / clearFollowUpQueue() / clearAllQueues() 丢弃排队消息。abort() 中止当前运行，waitForIdle() 等运行真正结束，reset() 清空对话和队列但保留回放后的提示与工具基线。

continue() 的边界：空 transcript 或只有系统消息会拒绝，而且不消费队列。最后一条是助手消息时不能直接续跑，循环会退回到"取一批引导消息，再取一批后续消息"。

reset() 保留基线这一点很实用：调用之后 state.messages 里只剩回放后的那条系统消息，提示和工具声明都还在。

一段完整的追踪代码

这段代码把本篇的每个规则都变成可观察的输出。它不花钱，用的是 fauxProvider 测试替身。

// trace.ts
import { Agent, type AgentTool } from "@earendil-works/pi-agent-core";
import {
	createModels,
	fauxAssistantMessage,
	fauxProvider,
	fauxText,
	fauxToolCall,
	Type,
} from "@earendil-works/pi-ai";

const echoTool: AgentTool<any> = {
	name: "echo",
	label: "回显",
	description: "回显给定文本",
	parameters: Type.Object({ text: Type.String() }),
	execute: async (_toolCallId, params) => {
		// 失败要抛异常，运行时才会把它作为 isError: true 回给模型
		if (params.text === "boom") {
			throw new Error("不允许回显 boom");
		}
		return {
			content: [{ type: "text", text: params.text }],
			details: { length: params.text.length },
		};
	},
};

const faux = fauxProvider();
const models = createModels();
models.setProvider(faux.provider);

// 用 faux 脚本化两轮响应：第一轮要求调工具，第二轮给出最终回答，全程不需要 API key
faux.setResponses([
	fauxAssistantMessage([fauxToolCall("echo", { text: "hello" })], { stopReason: "toolUse" }),
]);
faux.appendResponses([fauxAssistantMessage([fauxText("已回显")])]);

const agent = new Agent({
	initialState: {
		systemPrompt: "你是一个回声助手。",
		model: faux.getModel(),
		tools: [echoTool],
	},
	// streamFn 是唯一的模型访问接缝，绑定后不必再关心 provider 的调用细节
	streamFn: models.streamSimple.bind(models),
	toolExecution: "parallel",
});

// 订阅要早于 prompt，否则快速完成的运行会被整个错过
const unsubscribe = agent.subscribe((event) => {
	switch (event.type) {
		case "message_start":
			console.error(`[message_start] role=${event.message.role}`);
			break;
		case "tool_execution_start":
			console.error(`[tool_start] ${event.toolName} ${JSON.stringify(event.args)}`);
			break;
		case "tool_execution_end":
			// isError 为 true 说明工具抛了异常，循环不会因此停下
			console.error(`[tool_end] ${event.toolName} isError=${event.isError}`);
			break;
		case "turn_end":
			console.error(`[turn_end] toolResults=${event.toolResults.length}`);
			break;
		case "agent_end":
			console.error("[agent_end] 本次运行结束");
			break;
	}
});

try {
	await agent.prompt("回显 hello");
	// state 的每个字段都由具体事件写入，prompt 结束后读到的就是最终值
	console.error("[state] 消息数 =", agent.state.messages.length);
	console.error("[state] 提示回放 =", agent.state.systemPrompt);
	console.error("[state] pendingToolCalls =", agent.state.pendingToolCalls.size);
	console.error("[state] errorMessage =", agent.state.errorMessage ?? "无");
} finally {
	// abort 之后必须等 waitForIdle，否则状态可能还没落定就被读走
	unsubscribe();
	agent.abort();
	await agent.waitForIdle();
}

你应该看到的顺序：

[message_start] role=user
[tool_start] echo {"text":"hello"}
[tool_end] echo isError=false
[turn_end] toolResults=1
[agent_end] 本次运行结束
[state] 消息数 = 4
[state] 提示回放 = 你是一个回声助手。
[state] pendingToolCalls = 0
[state] errorMessage = 无

消息数是 4：用户消息、助手消息（含 toolCall）、工具结果消息、最终助手消息。

想验证错误路径，把脚本改成 fauxToolCall("echo", { text: "boom" })，你会看到 isError=true 而循环没有停。

谁负责什么

你现在能做什么

• 打印一次完整的 agent.state，把每个字段和它的事件来源对上号。
• 在 agent.state.tools 里加一个工具，观察循环在下一次请求前插入的 toolsAdded 系统消息。
• 只改 transcript 不改 agent.state.tools，复现"模型看得见但工具跑不了"的不一致。
• 用 beforeToolCall 拦下一次调用，确认 tool_execution_end.isError 为 true 且循环继续。
• 让工具抛一次异常，观察 state.errorMessage 在什么时机被写入。

排障速查

参考资料

• 官方文档：SDK
• 官方仓库
• Agent Core 的实现
• Agent Core 的类型定义
• Agent Core 的事件循环
• Agent Core 的 README
• pi-ai 的 transcript 工具