pi-agent-core: vòng lặp agent

Vai trò của pi-agent-core#

pi-ai (chương pi-ai) làm đúng một việc: gửi một context tới một mô hình (model) và trả về một luồng (stream) gồm các sự kiện (event) của một câu trả lời. Một agent cần nhiều hơn thế. Khi câu trả lời chứa lời gọi tool, phải có ai đó kiểm tra đối số, chạy tool, gói kết quả thành message, nối vào transcript rồi gọi mô hình lần nữa, cứ thế cho tới khi mô hình trả lời mà không gọi tool. Trong lúc đó giao diện cần biết từng bước đang diễn ra, và người dùng vẫn có thể gõ thêm tin nhắn. Phần “cứ thế cho tới khi” là vòng lặp agent (agent loop), và nó nằm trong gói @earendil-works/pi-agent-core, thư mục packages/agent.

README mô tả gói là “Stateful agent with tool execution and event streaming. Built on @earendil-works/pi-ai” (README.md:3). Gói rất nhỏ: sáu file trong src/, tổng 2.539 dòng. agent-loop.ts (959 dòng) là vòng lặp, agent.ts (623) là lớp bọc có state, types.ts (539) chứa toàn bộ data model, proxy.ts (393) và stream-fn.ts (20) là phần vận chuyển. Gói cung cấp hai tầng API:

• Hàm cấp thấp agentLoop() và agentLoopContinue(): không giữ state, nhận context, cấu hình, một AbortSignal và một hàm stream, trả về luồng AgentEvent.
• Lớp Agent: bọc vòng lặp, sở hữu transcript, hai hàng đợi tin nhắn (steer và follow-up), danh sách subscriber và AbortController của lần chạy hiện tại.

Trong monorepo ở v1.4.0, chỉ pi-coding-agent phụ thuộc vào gói này; Pi Durable có vòng lặp riêng dựng thẳng trên pi-ai (xem Bản đồ monorepo và Pi Durable). Từ 1.0.0 gói chỉ còn Agent, vòng lặp, proxy stream và các kiểu của chúng; harness thử nghiệm cũ (session, compaction, skills, telemetry...) đã bị gỡ ra (CHANGELOG.md:35).

Ranh giới trách nhiệm vì thế rất rõ. agent-core không biết provider (nhà cung cấp) nào, vì hàm stream được truyền từ ngoài vào; nó không ghi file session, không làm compaction (nén ngữ cảnh), không dựng system prompt và không hỏi quyền chạy tool. Nó chỉ biết transcript, bộ tool và các hook. Coding agent lấp các chỗ trống đó bằng cách cấu hình một Agent duy nhất trong createAgentSession():

const agent = new Agent({	initialState: {		systemPrompt: "",		model,		thinkingLevel,		tools: [],		messages: existingSession.messages,	},	convertToLlm: convertToLlmWithBlockImages,	streamFn: async (model, context, options) => {		const requestOptions = buildRequestOptions(options);		// Compaction and summaries use their own routing ids; only session requests		// replace the cache entry, so warming restarts from them. Keep warming while		// the current transcript still extends the request's prefix. Agent state may		// shallow-copy the messages array or refresh the model object without changing		// the provider request, so top-level object identity is not a valid cache key.		if (options?.sessionId === sessionManager.getSessionId()) {			cacheWarmer.start({ model, context, options: requestOptions }, cacheContextIsCurrent(model));		}		return modelRuntime.streamSimple(model, context, requestOptions);	},	onPayload: transformProviderPayload,	onResponse: handleProviderResponse,	onProviderStreamEvent: handleProviderStreamEvent,	sessionId: sessionManager.getSessionId(),	transformContext: async (messages) => {		const runner = extensionRunnerRef.current;		if (!runner) return messages;		return runner.emitContext(messages);	},	steeringMode: settingsManager.getSteeringMode(),	followUpMode: settingsManager.getFollowUpMode(),	transport: settingsManager.getTransport(),	thinkingBudgets: settingsManager.getThinkingBudgets(),	maxRetryDelayMs: settingsManager.getProviderRetrySettings().maxRetryDelayMs,});

Phần còn lại do AgentSession gắn vào sau khi tạo: một subscriber nhận mọi sự kiện để lưu session, chạy extension, tự compaction và thử lại (agent-session.ts:511–517), cặp hook beforeToolCall/afterToolCall (:652), prepareRequest để thay context bằng phép chiếu chuẩn của session (:787), finishTurn (:886) và prepareNextTurnWithContext để compaction và cập nhật system prompt giữa các lượt (:898). Chương này giải thích các hook đó có nghĩa gì với vòng lặp; nội dung bên trong chúng thuộc về chương AgentSession.

Mô hình tư duy: lần chạy, lượt và hai vòng lặp#

Phần còn lại của chương dùng các từ sau với nghĩa chính xác như trong code:

Lần chạy (run)

Một lần gọi prompt() hoặc continue(), mở bằng sự kiện agent_start và đóng bằng agent_end. Một lần chạy có thể gồm nhiều lượt.

Lượt (turn)

Một câu trả lời của mô hình cộng với các tool mà câu trả lời đó gọi, mở bằng turn_start, đóng bằng turn_end (types.ts:520).

Batch tool

Mọi lời gọi tool nằm trong một assistant message. Mô hình có thể gọi nhiều tool cùng lúc; vòng lặp chạy xong cả batch rồi mới gọi lại mô hình.

Transcript

Mảng AgentMessage[] theo thời gian: system, user, assistant, toolResult và các message tuỳ biến của ứng dụng.

Tin nhắn steer

Tin nhắn chen vào lượt đang chạy. Nó không cắt ngang tool nào: nó được chèn sau khi batch tool của lượt hiện tại xong, trước lần gọi mô hình kế tiếp.

Follow-up

Tin nhắn xếp hàng sau, chỉ được chèn khi agent lẽ ra đã dừng.

Cấu trúc điều khiển là hai vòng while lồng nhau trong runLoop(). Vòng trong chạy các lượt chừng nào mô hình còn gọi tool hoặc còn tin nhắn chờ. Vòng ngoài chỉ có một nhiệm vụ: khi vòng trong muốn dừng, hỏi hàng đợi follow-up; có tin thì quay lại vòng trong.

prompt() hoặc continue()
│ agent_start, turn_start, message_start/end cho các message của prompt
│
├─ vòng ngoài: while (true)
│  │
│  ├─ vòng trong: while (còn tool call || còn tin nhắn chờ)
│  │    1. từ lượt thứ hai: prepareNextTurn, hỏi lại steer nếu chưa có, turn_start
│  │    2. chèn tin nhắn chờ (steer hoặc follow-up)   message_start / message_end
│  │    3. prepareRequest
│  │    4. gọi mô hình qua StreamFn                    message_start / update... / end
│  │       stopReason "error" hoặc "aborted"  →  turn_end, agent_end, thoát
│  │    5. chạy batch tool                             tool_execution_*, toolResult
│  │    6. finishTurn                                  turn_end
│  │       {action: "end"}  →  agent_end, thoát
│  │    7. hỏi hàng đợi steer
│  │
│  └─ vòng trong muốn dừng: hỏi hàng đợi follow-up
│       có tin    →  thành tin nhắn chờ, quay lại vòng trong
│       không tin →  thoát vòng ngoài (trừ khi finishTurn đã đòi {action: "continue"})
│
└ agent_end

Data model: AgentMessage, AgentContext và AgentState#

Vấn đề. Transcript của một ứng dụng thực tế chứa nhiều thứ hơn những gì mô hình hiểu. Coding agent lưu kết quả của lệnh !ls mà người dùng tự gõ, bản tóm tắt sau compaction, tin nhắn do extension chèn vào. Nếu kiểu của transcript chỉ là Message của pi-ai (system, user, assistant, toolResult, xem ai/types.ts:652), các thứ đó phải sống ở chỗ khác và transcript không còn là nguồn sự thật duy nhất.

Giải pháp. agent-core làm việc với AgentMessage, là hợp của Message với một interface rỗng mà ứng dụng tự mở rộng:

/** * Extensible interface for custom app messages. * Apps can extend via declaration merging: * * @example * ```typescript * declare module "@mariozechner/agent" { *   interface CustomAgentMessages { *     artifact: ArtifactMessage; *     notification: NotificationMessage; *   } * } * ``` */export interface CustomAgentMessages {	// Empty by default - apps extend via declaration merging}/** * AgentMessage: Union of LLM messages + custom messages. * This abstraction allows apps to add custom message types while maintaining * type safety and compatibility with the base LLM messages. */export type AgentMessage = Message | CustomAgentMessages[keyof CustomAgentMessages];

Mặc định CustomAgentMessages rỗng, nên keyof CustomAgentMessages là never và AgentMessage chỉ là Message. Khi một ứng dụng viết declare module "@earendil-works/pi-agent-core" { interface CustomAgentMessages { ... } }, TypeScript gộp (declaration merging) khai báo đó vào interface gốc, và mỗi thuộc tính thêm một thành viên vào hợp. Tên khoá chỉ là nhãn; điều quan trọng là kiểu giá trị, và trên thực tế mỗi kiểu tuỳ biến cần một role riêng vì cả vòng lặp lẫn convertToLlm phân biệt message bằng role. Coding agent mở rộng đúng theo cách đó với bốn loại message:

// Extend CustomAgentMessages via declaration mergingdeclare module "@earendil-works/pi-agent-core" {	interface CustomAgentMessages {		bashExecution: BashExecutionMessage;		custom: CustomMessage;		branchSummary: BranchSummaryMessage;		compactionSummary: CompactionSummaryMessage;	}}

Mô hình không bao giờ thấy các role này. Trước mỗi lần gọi mô hình, vòng lặp gọi convertToLlm để biến AgentMessage[] thành Message[]. Bản của coding agent chuyển các loại tuỳ biến thành message user và bỏ qua lệnh bash có tiền tố !!:

export function convertToLlm(messages: AgentMessage[]): Message[] {	return messages		.map((m): Message | undefined => {			switch (m.role) {				case "bashExecution":					// Skip messages excluded from context (!! prefix)					if (m.excludeFromContext) {						return undefined;					}					return {						role: "user",						content: [{ type: "text", text: bashExecutionToText(m) }],						timestamp: m.timestamp,					};⋯ lược bớt dòng 162–183				case "system":				case "user":				case "assistant":				case "toolResult":					return m;				default:					// biome-ignore lint/correctness/noSwitchDeclarations: fine					const _exhaustiveCheck: never = m;					return undefined;			}		})		.filter((m) => m !== undefined);

Nếu ứng dụng không truyền convertToLlm, lớp Agent dùng một bộ lọc giữ lại đúng bốn role chuẩn (agent.ts:39–47).

Thứ vòng lặp nhận vào là một AgentContext: transcript và các tool chạy được.

/** Context snapshot passed into the low-level agent loop. */export interface AgentContext {	/** Transcript visible to the model. */	messages: AgentMessage[];	/** Tools available for execution in this run. */	tools?: AgentTool<any>[];}

Lưu ý tools ở đây là các AgentTool có hàm execute, tức những gì runtime chạy được. Những gì mô hình được phép gọi lại được khai báo trong các system message của transcript; mục declareToolChanges giải thích vì sao hai thứ này tách nhau.

Lớp Agent phơi state công khai qua AgentState:

export interface AgentState {	/**	 * Current system prompt, replayed from the transcript's system messages.	 *	 * Read-only: to change the prompt, append a system message with `content` or `sections`.	 * In `initialState`, this seeds the leading system message.	 */	readonly systemPrompt: string;	/** Active model used for future turns. */	model: Model<any>;	/** Requested reasoning level for future turns. */	thinkingLevel: ThinkingLevel;	/**	 * Executable tools. Assigning a new array copies the top-level array.	 *	 * Differences from the tools declared in the transcript are announced to the model	 * with a system message before the next request.	 */	set tools(tools: AgentTool<any>[]);	get tools(): AgentTool<any>[];	/**	 * Conversation transcript. Assigning a new array copies the top-level array.	 *	 * System messages in the transcript carry the prompt and tool declarations.	 */	set messages(messages: AgentMessage[]);	get messages(): AgentMessage[];	/**	 * True while the agent is processing a prompt or continuation.	 *	 * This remains true until awaited `agent_end` listeners settle.	 */	readonly isStreaming: boolean;	/** Partial assistant message for the current streamed response, if any. */	readonly streamingMessage?: AgentMessage;	/** Tool call ids currently executing. */	readonly pendingToolCalls: ReadonlySet<string>;	/** Error message from the most recent failed or aborted assistant turn, if any. */	readonly errorMessage?: string;}

• systemPrompt chỉ đọc. Nó được tính lại bằng cách phát lại (replay) mọi system message trong transcript (getCurrentSystemPrompt). Muốn đổi prompt giữa hội thoại thì nối thêm một system message có content (thêm chỉ dẫn) hoặc sections (thay một mục theo tên).
• model và mức suy nghĩ (thinking level) thinkingLevel chỉ được đọc khi một lần chạy bắt đầu, lúc Agent dựng cấu hình cho vòng lặp. Muốn giá trị mới có hiệu lực ngay từ lượt kế tiếp của lần chạy đang diễn ra, một hook như prepareNextTurn phải trả chúng về; AgentSession làm đúng như vậy.
• tools và messages là accessor: gán một mảng mới thì mảng cấp ngoài cùng được sao chép (agent.ts:95–106), còn push() vào mảng trả về thì sửa thẳng state hiện tại.
• isStreaming, streamingMessage, pendingToolCalls, errorMessage chỉ đọc với bên ngoài; chúng do chính các sự kiện của vòng lặp cập nhật (xem lớp Agent).

Khi tạo Agent với systemPrompt và tools, hai thứ này trở thành system message đầu tiên của transcript, mang toolsAdded và timestamp: 0 (agent.ts:86–87, transcript.ts:10–23). Sau một lượt có gọi tool, transcript trông như sau:

Ví dụ minh hoạ: agent.state.messages sau một prompt có một lời gọi tool
[
  { role: "system", content: "You are helpful.",
    toolsAdded: [{ name: "read", description: "...", parameters: {...} }], timestamp: 0 },
  { role: "user", content: [{ type: "text", text: "Đọc a.txt" }], timestamp: 1761100000000 },
  { role: "assistant", stopReason: "toolUse", thinkingLevel: "off",
    content: [{ type: "toolCall", id: "call_1", name: "read", arguments: { path: "a.txt" } }],
    api: "...", provider: "...", model: "...", usage: {...}, timestamp: 1761100000100 },
  { role: "toolResult", toolCallId: "call_1", toolName: "read", isError: false,
    content: [{ type: "text", text: "nội dung a.txt" }], details: {}, durationMs: 2,
    timestamp: 1761100001200 },
  { role: "assistant", stopReason: "stop", content: [{ type: "text", text: "a.txt chứa ..." }], ... }
]

Data model: AgentTool và kết quả tool#

AgentTool mở rộng Tool của pi-ai (name, description, parameters là một schema TypeBox) bằng những gì runtime cần để chạy tool:

/** Tool definition used by the agent runtime. */export interface AgentTool<TParameters extends TSchema = TSchema, TDetails = any> extends Tool<TParameters> {	/** Human-readable label for UI display. */	label: string;	/**	 * Optional compatibility shim for raw tool-call arguments before schema validation.	 * Must return an object that matches `TParameters`.	 */	prepareArguments?: (args: unknown) => Static<TParameters>;	/**	 * JSON Schema of `structuredContent` in successful results. Tools that declare it should always	 * set `structuredContent`.	 */	outputSchema?: TSchema;	/**	 * Execute the tool call. Throw on failure, or return a result with `isError: true`; do not only	 * describe the failure in `content`.	 */	execute: (		toolCallId: string,		params: Static<TParameters>,		signal?: AbortSignal,		onUpdate?: AgentToolUpdateCallback<TDetails>,	) => Promise<AgentToolResult<TDetails>>;	/** Recovery policy for an effect whose durable intent exists but whose outcome is unknown. */	replay?: "never" | "safe";	/**	 * Per-tool execution mode override.	 * - "sequential": this tool must execute one at a time with other tool calls.	 * - "parallel": this tool can execute concurrently with other tool calls.	 *	 * If omitted, the default execution mode applies.	 */	executionMode?: ToolExecutionMode;}

• label là tên hiển thị cho giao diện, không gửi cho mô hình.
• prepareArguments là lớp tương thích chạy trước khi kiểm tra schema, ví dụ để nhận đối số kiểu cũ từ một session được mở lại sau khi schema đã đổi (CHANGELOG 0.64.0, CHANGELOG.md:481).
• execute(toolCallId, params, signal, onUpdate) nhận đối số đã kiểm tra, tín hiệu huỷ của lần chạy, và một callback để phát kết quả từng phần.
• executionMode cho phép một tool ép cả batch chạy tuần tự (xem thực thi tool).
• replay mô tả chính sách phục hồi cho một hiệu ứng đã có ý định bền (durable intent) nhưng chưa rõ kết quả. Vòng lặp ở v1.4.0 không đọc trường này, và coding agent cũng không; Pi Durable có khái niệm tương tự trong harness riêng.

Kết quả của một tool tách hai phần cho hai người đọc khác nhau:

/** Final or partial result produced by a tool. */export interface AgentToolResult<T = JsonValue | undefined> {	/** Text or image content returned to the model. */	content: (TextContent | ImageContent)[];	/** Arbitrary structured details for logs or UI rendering. */	details: T;	/**	 * Machine-readable result matching the tool's `outputSchema`, for programmatic callers. Not sent	 * to the model; `content` remains the model-facing result.	 */	structuredContent?: JsonValue;	/** Usage from the final tool execution itself, if available. Not used for main LLM context accounting. */	usage?: Usage;	/**	 * Report a failure without throwing. The model sees `content` as an error result, like a thrown	 * error, but `details` and `structuredContent` are kept for the UI and programmatic callers.	 */	isError?: boolean;	/**	 * Hint that the agent should stop after the current tool batch.	 * Early termination only happens when every finalized tool result in the batch sets this to true.	 */	terminate?: boolean;}

content là thứ mô hình đọc. details là dữ liệu có cấu trúc tuỳ ý cho log và giao diện, không bao giờ gửi cho mô hình. Ví dụ tool read của coding agent trả văn bản trong content, còn thông tin cắt bớt có cấu trúc thì nằm trong details: { truncation } (read.ts:40–42) để giao diện vẽ ghi chú riêng. Bài blog đầu tiên của Mario gọi đây là “structured split tool results”: “splitting tool results into a portion handed to the LLM and a portion for UI display” (What I learned building an opinionated and minimal coding agent).

Các trường còn lại phục vụ những người khác. structuredContent dành cho caller lập trình (ví dụ một script codemode gọi tool lồng nhau), khớp với outputSchema. usage là lượng token mà chính việc chạy tool tiêu thụ, không tính vào ngữ cảnh chính. isError báo lỗi mà không cần throw: mô hình vẫn thấy một kết quả lỗi, còn details được giữ lại. terminate là gợi ý dừng sau batch hiện tại. Khi kết quả được đóng gói thành message toolResult cho transcript, chỉ content, details, usage, isError và durationMs được giữ; terminate và structuredContent chỉ sống trong sự kiện tool_execution_end:

function createToolResultMessage(finalized: FinalizedToolCallOutcome): ToolResultMessage {	return {		role: "toolResult",		toolCallId: finalized.toolCall.id,		toolName: finalized.toolCall.name,		// Untyped tools (JS extensions) can return results without content; normalize		// so the null never enters session history or provider payloads.		content: finalized.result.content ?? [],		details: finalized.result.details,		usage: finalized.result.usage,		isError: finalized.isError,		...(finalized.durationMs === undefined ? {} : { durationMs: finalized.durationMs }),		timestamp: Date.now(),	};}

Tool phát kết quả từng phần qua onUpdate, mỗi lần gọi thành một sự kiện tool_execution_update. Callback chỉ có hiệu lực trong lúc execute() còn chạy; gọi sau khi promise đã xong thì bị bỏ qua (types.ts:457–463), một sửa lỗi ở 0.79.2 để giao diện không nhận cập nhật cũ (CHANGELOG.md:295).

Data model: AgentLoopConfig và các hook#

Mọi hành vi tuỳ biến của vòng lặp đi qua một đối tượng cấu hình. AgentLoopConfig kế thừa SimpleStreamOptions của pi-ai (API key, reasoning, sessionId, transport, ...) nên chính nó cũng là options truyền cho hàm stream. Đây là các trường riêng của nó, đã lược chú thích:

export interface AgentLoopConfig extends SimpleStreamOptions {	model: Model<any>;⋯ lược bớt dòng 195–221	convertToLlm: (messages: AgentMessage[]) => Message[] | Promise<Message[]>;⋯ lược bớt dòng 223–243	transformContext?: (messages: AgentMessage[], signal?: AbortSignal) => Promise<AgentMessage[]>;⋯ lược bớt dòng 245–253	getApiKey?: (provider: string) => Promise<string | undefined> | string | undefined;⋯ lược bớt dòng 255–263	finishTurn?: FinishTurn;⋯ lược bớt dòng 265–270	prepareRequest?: PrepareRequest;⋯ lược bớt dòng 272–277	prepareNextTurn?: (		context: PrepareNextTurnContext,	) => AgentLoopTurnUpdate | undefined | Promise<AgentLoopTurnUpdate | undefined>;⋯ lược bớt dòng 281–292	getSteeringMessages?: () => Promise<AgentMessage[]>;⋯ lược bớt dòng 294–305	getFollowUpMessages?: () => Promise<AgentMessage[]>;⋯ lược bớt dòng 307–316	toolExecution?: ToolExecutionMode;⋯ lược bớt dòng 318–325	beforeToolCall?: (context: BeforeToolCallContext, signal?: AbortSignal) => Promise<BeforeToolCallResult | undefined>;⋯ lược bớt dòng 327–340	afterToolCall?: (context: AfterToolCallContext, signal?: AbortSignal) => Promise<AfterToolCallResult | undefined>;}

Hợp đồng “không được throw” có lý do cụ thể: một exception từ các hàm này cắt ngang vòng lặp cấp thấp mà không phát chuỗi sự kiện bình thường (types.ts:203–204). Lớp Agent có một lưới an toàn cho trường hợp này (xem dưới), còn người dùng agentLoop() trực tiếp thì không.

Ba hook điều khiển lịch trình có kiểu riêng:

/** Decision returned by {@link FinishTurn}. Returning undefined preserves normal scheduling. */export type AgentTurnDecision = { action: "continue" } | { action: "end" };/** * Called after a completed assistant turn and all of its tool-result messages, but before `turn_end`. * On a normal turn, `{ action: "continue" }` ensures one next provider request. Tool-result, steering, or * follow-up scheduling can satisfy that request and adds no extra request; otherwise the loop continues once * with the current context. Error and aborted responses remain hard exits. */export type FinishTurn = (	turn: AgentTurnContext,	signal?: AbortSignal,) => AgentTurnDecision | void | Promise<AgentTurnDecision | undefined> | Promise<void>;/** Replacement runtime state used by the agent loop before starting another provider request. */export interface AgentLoopTurnUpdate {	/** Context for the next provider request. */	context?: AgentContext;	/** Messages to append before the next provider request, with normal lifecycle events. */	messages?: AgentMessage[];	/** Model for the next provider request. */	model?: Model<any>;	/** Thinking level for the next provider request. */	thinkingLevel?: ThinkingLevel;}

Ngữ nghĩa của { action: "continue" } tinh tế hơn tên gọi: nó bảo đảm ít nhất một request tiếp theo, chứ không thêm một request. Nếu lượt vừa rồi có tool call, hoặc có tin steer hay follow-up, thì request tự nhiên kế tiếp đã thoả yêu cầu; chỉ khi không có gì, vòng lặp mới chạy thêm đúng một lượt với context hiện tại (types.ts:149–154). Phản hồi lỗi hoặc bị huỷ luôn là lối ra cứng; quyết định của finishTurn khi đó bị bỏ qua. README cảnh báo: trả về continue vô điều kiện sẽ tạo vòng lặp vô tận, vì finishTurn chạy lại sau request kế tiếp (README.md:146).

Data model: AgentEvent#

Vòng lặp không trả về giá trị trung gian nào; mọi thứ nó làm đều được báo qua sự kiện:

/** * Events emitted by the Agent for UI updates. * * `agent_end` is the last event emitted for a run, but awaited `Agent.subscribe()` * listeners for that event are still part of run settlement. The agent becomes * idle only after those listeners finish. */export type AgentEvent =	// Agent lifecycle	| { type: "agent_start" }	| { type: "agent_end"; messages: AgentMessage[] }	// Turn lifecycle - a turn is one assistant response + any tool calls/results	| { type: "turn_start" }	| { type: "turn_end"; message: AgentMessage; toolResults: ToolResultMessage[] }	// Message lifecycle - emitted for system, user, assistant, and toolResult messages	| { type: "message_start"; message: AgentMessage }	// Only emitted for assistant messages during streaming	| { type: "message_update"; message: AgentMessage; assistantMessageEvent: AssistantMessageEvent }	| { type: "message_end"; message: AgentMessage }	// Tool execution lifecycle	| { type: "tool_execution_start"; toolCallId: string; toolName: string; args: any }	| { type: "tool_execution_update"; toolCallId: string; toolName: string; args: any; partialResult: any }	| {			type: "tool_execution_end";			toolCallId: string;			toolName: string;			result: any;			isError: boolean;			/** Milliseconds `execute()` took, measured with a monotonic clock; absent when the tool did not run. */			durationMs?: number;	  };

• Lần chạy: agent_start và agent_end. agent_end.messages là các message mới của lần chạy: với prompt() gồm cả message của prompt, với continue() thì không gồm các message đã có (types.ts:142).
• Lượt: turn_start và turn_end; turn_end mang assistant message và các toolResult của lượt.
• Message: message_start/message_end phát cho mọi message, kể cả system, user và toolResult. message_update chỉ có với assistant message đang stream, kèm assistantMessageEvent gốc của pi-ai (text_delta, toolcall_delta, ...).
• Tool: tool_execution_start, _update, _end. _end mang toàn bộ AgentToolResult (gồm cả structuredContent và terminate) và durationMs đo bằng đồng hồ đơn điệu, không tính thời gian chạy hook (CHANGELOG 1.1.0, CHANGELOG.md:17).

Hai chi tiết đáng nhớ. Message trong message_start và message_update của assistant là bản sao nông { ...partialMessage } chụp tại thời điểm đó, còn message_end mang đối tượng cuối cùng (agent-loop.ts:430, :464). Và agent_end chỉ có nghĩa là vòng lặp sẽ không phát thêm sự kiện nào; lớp Agent chỉ coi là rảnh sau khi các listener của chính sự kiện đó chạy xong (types.ts:509–515).

StreamFn và proxy: lớp vận chuyển#

package.json vẫn mô tả gói là “General-purpose agent with transport abstraction” (package.json:4), nhưng ở v1.4.0 lớp vận chuyển không còn là một interface hay một lớp nào. Ở 0.31.0, các interface ProviderTransport, AppTransport, AgentTransport bị xoá và thay bằng một tuỳ chọn duy nhất, streamFn (CHANGELOG.md:757). Cũng ở phiên bản đó, agentLoop() và agentLoopContinue() chuyển từ pi-ai sang gói này (CHANGELOG.md:770); vì vậy bài blog tháng 11/2025 của Mario còn mô tả vòng lặp dưới mục pi-ai, còn lớp Agent với “transport abstraction” thuộc pi-agent-core. Từ 0.31.0, “vận chuyển” chỉ là một kiểu hàm:

/** * Stream function used by the agent loop. `Models.streamSimple` satisfies * this shape. * * The loop passes a normalized transcript: the system prompt and tool * declarations are carried by the transcript's system messages, never by * `context.systemPrompt` or `context.tools`. * * Contract: * - Must not throw or return a rejected promise for request/model/runtime failures. * - Must return an AssistantMessageEventStream. * - Failures must be encoded in the returned stream via protocol events and a *   final AssistantMessage with stopReason "error" or "aborted" and errorMessage. */export type StreamFn = (	model: Model<Api>,	context: TranscriptContext,	options?: SimpleStreamOptions,) => AssistantMessageEventStream | Promise<AssistantMessageEventStream>;

Hợp đồng có ba điểm. Hàm nhận một TranscriptContext đã chuẩn hoá, trong đó system prompt và khai báo tool nằm trong các system message, không nằm ở context.systemPrompt hay context.tools. Hàm không được throw: mọi lỗi request, lỗi mô hình hay bị huỷ đều phải được mã hoá thành sự kiện trong luồng, kết thúc bằng một assistant message có stopReason là "error" hoặc "aborted". Và Models.streamSimple của pi-ai khớp với kiểu này, nên trường hợp thường gặp chỉ là streamFn: models.streamSimple.bind(models) (README.md:23–29).

Bản thân agent-core không import provider nào. Nếu caller bỏ qua streamFn (các consumer biên dịch từ phiên bản cũ có thể làm vậy, xem agent.ts:239), nó dùng một hàm mặc định do host cài:

let defaultStreamFn: StreamFn | undefined;/** * Configure the fallback used by Agent and low-level loops when callers omit streamFn. * * Hosts that provide a default model runtime can install its stream function here * without making pi-agent-core depend on a provider catalog or compatibility layer. */export function setDefaultStreamFn(streamFn: StreamFn | undefined): void {	defaultStreamFn = streamFn;}export function getDefaultStreamFn(): StreamFn {	if (!defaultStreamFn) {		throw new Error("No default stream function configured. Pass streamFn explicitly or call setDefaultStreamFn().");	}	return defaultStreamFn;}

Coding agent cài streamSimple của pi-ai/compat làm mặc định ngay khi module sdk.ts được nạp, để extension nào tự tạo Agent mà quên streamFn vẫn chạy (sdk.ts:43–46). Cách sắp xếp này có chủ đích: ở 0.81.0, streamFn từng bị bắt buộc để pi-ai/compat và mọi provider dựng sẵn không lọt vào các bundle chỉ chọn vài provider (CHANGELOG.md:193); bản sửa sau đó đưa lại fallback nhưng do host cấu hình (:185).

streamProxy: khi mô hình nằm sau một server#

Ứng dụng trình duyệt không nên giữ API key của provider. streamProxy() là một StreamFn gửi request tới server của chính ứng dụng; server giữ khoá, gọi provider, rồi stream sự kiện ngược lại. Để tiết kiệm băng thông, server bỏ trường partial (assistant message dựng dần) khỏi mỗi sự kiện, nên giao thức proxy là một phiên bản rút gọn của AssistantMessageEvent:

/** * Proxy event types - server sends these with partial field stripped to reduce bandwidth. */export type ProxyAssistantMessageEvent =	| { type: "start" }	| { type: "text_start"; contentIndex: number }	| { type: "text_delta"; contentIndex: number; delta: string }	| { type: "text_end"; contentIndex: number; contentSignature?: string }	| { type: "thinking_start"; contentIndex: number }	| { type: "thinking_delta"; contentIndex: number; delta: string }	| { type: "thinking_end"; contentIndex: number; contentSignature?: string }	| { type: "toolcall_start"; contentIndex: number; id: string; toolName: string }	| { type: "toolcall_delta"; contentIndex: number; delta: string }	| { type: "toolcall_end"; contentIndex: number; toolCall: ToolCall }	| {			type: "done";			reason: Extract<StopReason, "stop" | "length" | "toolUse">;			usage: AssistantMessage["usage"];			providerThinkingLevel?: string;	  }	| {			type: "error";			reason: Extract<StopReason, "aborted" | "error">;			errorMessage?: string;			usage: AssistantMessage["usage"];			providerThinkingLevel?: string;	  };

try {	const response = await fetch(`${options.proxyUrl}/api/stream`, {		method: "POST",		headers: {			Authorization: `Bearer ${options.authToken}`,			"Content-Type": "application/json",		},		body: JSON.stringify({			model,			context,			options: buildProxyRequestOptions(options),		}),		signal: options.signal,	});

Phía client đọc từng dòng data: {...} và dựng lại partial trong processProxyEvent(). Đoạn khó nhất là đối số tool: chúng tới dưới dạng các mảnh JSON chưa hoàn chỉnh, nên client cộng dồn chuỗi và dùng bộ đọc JSON từng phần của pi-ai sau mỗi mảnh:

case "toolcall_delta": {	const content = partial.content[proxyEvent.contentIndex];	if (content?.type === "toolCall") {		(content as any).partialJson += proxyEvent.delta;		content.arguments = parseStreamingJson((content as any).partialJson) || {};		partial.content[proxyEvent.contentIndex] = { ...content }; // Trigger reactivity		return {			type: "toolcall_delta",			contentIndex: proxyEvent.contentIndex,			delta: proxyEvent.delta,			partial,		};	}	throw new Error("Received toolcall_delta for non-toolCall content");}

streamProxy() giữ đúng hợp đồng không-throw của StreamFn: lỗi HTTP, lỗi mạng và abort đều thành một sự kiện error với stopReason tương ứng (proxy.ts:233–243). Nó cũng bắt một trường hợp dễ quên: server đóng kết nối sạch sẽ mà chưa gửi done hay error. Khi đó nó tự phát một sự kiện error với thông báo “Connection closed by proxy server before the response completed” (proxy.ts:219–230); không có kiểm tra này, vòng lặp sẽ chờ mãi một kết quả không bao giờ tới.

runLoop: vòng ngoài và vòng trong#

Hai cửa vào: agentLoop và agentLoopContinue#

agentLoop() thêm prompt mới vào context rồi chạy; agentLoopContinue() chạy tiếp từ context có sẵn mà không thêm gì, dùng cho thử lại. Cả hai trả về ngay một EventStream và để vòng lặp chạy nền:

export function agentLoop(	prompts: AgentMessage[],	context: AgentContext,	config: AgentLoopConfig,	signal: AbortSignal | undefined,	streamFn: StreamFn,): EventStream<AgentEvent, AgentMessage[]> {	const stream = createAgentStream();	void runAgentLoop(		prompts,		context,		config,		async (event) => {			stream.push(event);		},		signal,		streamFn,	).then((messages) => {		stream.end(messages);	});	return stream;}

Luồng kết thúc khi gặp agent_end, và stream.result() trả về danh sách message mới (agent-loop.ts:162–167). Chú ý hàm emit ở dòng 60–62 chỉ push vào luồng rồi trả về ngay, nên vòng lặp không chờ người tiêu thụ xử lý xong một sự kiện. README gọi các luồng cấp thấp này là “observational”: giữ đúng thứ tự, nhưng không phải rào chắn (README.md:565). Lớp Agent thì gọi thẳng runAgentLoop() với một emit có await, nên mỗi sự kiện trở thành một rào chắn.

export async function runAgentLoop(	prompts: AgentMessage[],	context: AgentContext,	config: AgentLoopConfig,	emit: AgentEventSink,	signal: AbortSignal | undefined,	streamFn: StreamFn,): Promise<AgentMessage[]> {	const initialMessages = declareToolChanges(context, prompts);	const newMessages: AgentMessage[] = [...initialMessages];	const currentContext: AgentContext = {		...context,		messages: [...context.messages, ...initialMessages],	};	await emit({ type: "agent_start" });	await emit({ type: "turn_start" });	for (const message of initialMessages) {		await emit({ type: "message_start", message });		await emit({ type: "message_end", message });	}	await runLoop(currentContext, newMessages, config, signal, emit, streamFn ?? getDefaultStreamFn());	return newMessages;}

Các message của prompt đi qua declareToolChanges() trước (dòng 119; giải thích ở mục sau), được nối vào một bản sao của transcript, rồi phát thành cặp message_start/message_end ngay sau agent_start và turn_start đầu tiên. runAgentLoopContinue() giống vậy nhưng không có message mới, và từ chối chạy nếu transcript rỗng hoặc kết thúc bằng assistant message (agent-loop.ts:144–150), vì provider sẽ từ chối một request kết thúc bằng câu trả lời của chính mô hình.

Khởi tạo và hai vòng while#

async function runLoop(	initialContext: AgentContext,	newMessages: AgentMessage[],	initialConfig: AgentLoopConfig,	signal: AbortSignal | undefined,	emit: AgentEventSink,	streamFunction: StreamFn,): Promise<void> {	let currentContext = initialContext;	let config = initialConfig;	let lastCompletedTurn: PrepareNextTurnContext | undefined;	let explicitContinuation = false;	// Check for steering messages at start (user may have typed while waiting)	let pendingMessages: AgentMessage[] = (await config.getSteeringMessages?.()) || [];	// Outer loop: continues when queued follow-up messages arrive after agent would stop	while (true) {		let hasMoreToolCalls = true;		// Inner loop: process tool calls and steering messages		while (hasMoreToolCalls || pendingMessages.length > 0) {

Toàn bộ state của vòng lặp là vài biến cục bộ. lastCompletedTurn nhớ lượt vừa xong để truyền cho prepareNextTurn. explicitContinuation nhớ rằng finishTurn đã đòi thêm một request. pendingMessages là các tin nhắn sẽ được chèn trước request kế tiếp, và nó được khởi tạo bằng một lần hỏi hàng đợi steer ngay từ đầu (dòng 185), vì người dùng có thể đã gõ trong lúc chờ. hasMoreToolCalls bắt đầu là true nên lần lặp đầu tiên luôn gọi mô hình.

Đầu mỗi lượt: chuẩn bị và chèn tin nhắn chờ#

let preparedMessages: AgentMessage[] = [];if (lastCompletedTurn) {	const nextTurnSnapshot = await config.prepareNextTurn?.(lastCompletedTurn);	if (nextTurnSnapshot) {		currentContext = nextTurnSnapshot.context ?? currentContext;		preparedMessages = nextTurnSnapshot.messages ?? [];		config = {			...config,			model: nextTurnSnapshot.model ?? config.model,			reasoning:				nextTurnSnapshot.thinkingLevel === undefined					? config.reasoning					: nextTurnSnapshot.thinkingLevel === "off"						? undefined						: nextTurnSnapshot.thinkingLevel,		};	}	// Preparation can be long-running (for example, compaction). Pick up steering	// queued while it ran. Only poll again if the earlier poll returned nothing;	// otherwise one-at-a-time mode would deliver two messages in this turn.	if (pendingMessages.length === 0) {		pendingMessages = (await config.getSteeringMessages?.()) || [];	}	await emit({ type: "turn_start" });}// Process prepared and queued messages before the next assistant response.for (const message of declareToolChanges(currentContext, [...preparedMessages, ...pendingMessages])) {	await emit({ type: "message_start", message });	await emit({ type: "message_end", message });	currentContext.messages.push(message);	newMessages.push(message);}pendingMessages = [];

Từ lượt thứ hai trở đi (khi lastCompletedTurn đã có), vòng lặp gọi prepareNextTurn. Hook này có thể thay context, model, mức suy nghĩ, hoặc trả về message cần nối thêm; coding agent dùng nó để compaction trước câu trả lời kế tiếp và gắn system message cập nhật prompt. Mức suy nghĩ "off" được dịch thành reasoning: undefined (dòng 202–207). Vì chuẩn bị có thể lâu, vòng lặp hỏi lại hàng đợi steer sau đó, nhưng chỉ khi lần hỏi trước rỗng: chú thích ở dòng 210–212 giải thích rằng hỏi hai lần sẽ khiến chế độ one-at-a-time giao hai tin trong một lượt. Rồi mới phát turn_start.

Dòng 220–224 là chỗ duy nhất tin nhắn chờ, dù là steer, follow-up hay message từ prepareNextTurn, đi vào transcript: mỗi message được phát message_start/message_end, rồi nối vào cả context lẫn newMessages.

Gọi mô hình và lối ra cứng#

const requestUpdate = await config.prepareRequest?.(	{		context: currentContext,		model: config.model,		thinkingLevel: config.reasoning ?? "off",	},	signal,);if (requestUpdate) {	currentContext = requestUpdate.context ?? currentContext;	config = {		...config,		model: requestUpdate.model ?? config.model,		reasoning:			requestUpdate.thinkingLevel === undefined				? config.reasoning				: requestUpdate.thinkingLevel === "off"					? undefined					: requestUpdate.thinkingLevel,	};}// Stream assistant responseconst message = await streamAssistantResponse(currentContext, config, signal, emit, streamFunction);newMessages.push(message);if (message.stopReason === "error" || message.stopReason === "aborted") {	lastCompletedTurn = {		message,		toolResults: [],		context: currentContext,		newMessages,	};	await config.finishTurn?.(lastCompletedTurn, signal);	await emit({ type: "turn_end", message, toolResults: [] });	await emit({ type: "agent_end", messages: newMessages });	return;}

prepareRequest chạy ngay trước mỗi request, sau khi tin nhắn chờ đã được phát. Coding agent dùng nó để thay toàn bộ context bằng phép chiếu chuẩn dựng từ SessionManager, vốn là nguồn có thẩm quyền cho context của mô hình (sdk.md:40, agent-session.ts:787–800). Kết quả của streamAssistantResponse() được nối vào newMessages bất kể thành công hay không. Nếu stopReason là "error" hoặc "aborted", vòng lặp vẫn gọi finishTurn (để hook quan sát), phát turn_end và agent_end, rồi return: không chạy tool nào, không hỏi hàng đợi nào, và giá trị trả về của finishTurn bị bỏ qua.

Cuối lượt: tool, finishTurn và hàng đợi steer#

	// Check for tool calls	const toolCalls = message.content.filter((c) => c.type === "toolCall");	const toolResults: ToolResultMessage[] = [];	hasMoreToolCalls = false;	if (toolCalls.length > 0) {		// A "length" stop means the output was cut off by the token limit, so		// every tool call in the message may carry truncated arguments. Fail		// them all instead of executing potentially borked calls.		const executedToolBatch =			message.stopReason === "length"				? await failToolCallsFromTruncatedMessage(toolCalls, emit)				: await executeToolCalls(currentContext, message, config, signal, emit);		toolResults.push(...executedToolBatch.messages);		hasMoreToolCalls = !executedToolBatch.terminate;		for (const result of toolResults) {			currentContext.messages.push(result);			newMessages.push(result);		}	}	lastCompletedTurn = {		message,		toolResults,		context: currentContext,		newMessages,	};	const decision = await config.finishTurn?.(lastCompletedTurn, signal);	await emit({ type: "turn_end", message, toolResults });	if (decision?.action === "end") {		await emit({ type: "agent_end", messages: newMessages });		return;	}	explicitContinuation = decision?.action === "continue";	pendingMessages = (await config.getSteeringMessages?.()) || [];	if (hasMoreToolCalls || pendingMessages.length > 0) {		explicitContinuation = false;	}}

Batch tool được chạy bởi executeToolCalls(), trừ khi assistant message dừng vì chạm giới hạn token đầu ra (stopReason === "length"); khi đó mọi lời gọi tool trong message bị đánh lỗi mà không chạy (dòng 273–279, xem mục thực thi tool). Sau batch, hasMoreToolCalls là true trừ khi cả batch xin dừng (dòng 281): kết quả tool phải được đưa lại cho mô hình, nên có tool call nghĩa là có thêm một lượt.

Thứ tự ở dòng 295–307 là cố ý: finishTurn chạy trước turn_end nhưng quyết định của nó được áp dụng sau. Với { action: "end" }, vòng lặp phát agent_end ngay, để nguyên hai hàng đợi và không gọi prepareNextTurn (CHANGELOG.md:68). Nếu không, nó hỏi hàng đợi steer (dòng 304). Đây là điểm mà tin nhắn steer đi vào: sau khi mọi tool của lượt đã xong và đã có toolResult, trước lần gọi mô hình kế tiếp. Nếu đã có tool call hoặc tin steer, request tự nhiên kế tiếp đã thoả một { action: "continue" }, nên cờ explicitContinuation được xoá (dòng 305–307).

Khi vòng trong muốn dừng: follow-up#

		// Agent would stop here. Check for follow-up messages.		const followUpMessages = (await config.getFollowUpMessages?.()) || [];		if (followUpMessages.length > 0) {			// Set as pending so inner loop processes them			explicitContinuation = false;			pendingMessages = followUpMessages;			continue;		}		// No natural request was selected, so fulfill the continuation decision with one context-only turn.		if (explicitContinuation) {			explicitContinuation = false;			continue;		}		// No more messages, exit		break;	}	await emit({ type: "agent_end", messages: newMessages });}

Vòng trong thoát khi lượt vừa rồi không có tool call (hoặc cả batch xin dừng) và hàng đợi steer rỗng. Lúc đó, và chỉ lúc đó, hàng đợi follow-up được hỏi. Có tin thì chúng trở thành pendingMessages và vòng ngoài quay lại vòng trong; không có thì xét continuation tường minh từ finishTurn, cuối cùng là break và agent_end. Đây chính là khác biệt giữa steer và follow-up: steer được hỏi sau mỗi lượt, follow-up chỉ được hỏi khi agent đã xong việc.

streamAssistantResponse: nơi AgentMessage thành Message#

Chú thích đầu file nói rõ nguyên tắc: vòng lặp làm việc với AgentMessage ở mọi chỗ và chỉ biến đổi sang Message[] tại ranh giới gọi mô hình (agent-loop.ts:1–4). Ranh giới đó là nửa đầu của streamAssistantResponse():

async function streamAssistantResponse(	context: AgentContext,	config: AgentLoopConfig,	signal: AbortSignal | undefined,	emit: AgentEventSink,	streamFunction: StreamFn,): Promise<AssistantMessage> {	// Apply context transform if configured (AgentMessage[] → AgentMessage[])	let messages = context.messages;	if (config.transformContext) {		messages = await config.transformContext(messages, signal);	}	// Convert to LLM-compatible messages (AgentMessage[] → Message[])	const llmMessages = await config.convertToLlm(messages);	const llmContext = normalizeContext({ messages: llmMessages });	// Resolve API key (important for expiring tokens)	const resolvedApiKey =		(config.getApiKey ? await config.getApiKey(config.model.provider) : undefined) || config.apiKey;	const response = await streamFunction(config.model, llmContext, {		...config,		apiKey: resolvedApiKey,		identity: mergeClientIdentity(DEFAULT_AGENT_IDENTITY, config.identity),		signal,	});	// Record the requested level, whichever stream function answered.	const result = async () => Object.assign(await response.result(), { thinkingLevel: config.reasoning ?? "off" });

1. transformContextTuỳ chọn, trên AgentMessage[]: cắt bớt, chèn ngữ cảnh. Coding agent chuyển nó cho sự kiện context của extension.
2. convertToLlmBắt buộc: lọc message chỉ dành cho UI, đổi message tuỳ biến thành user.
3. normalizeContextGắn nhãn kiểu TranscriptContext; prompt và tool đã nằm trong system message.
4. getApiKeyHỏi key cho từng request, rơi về config.apiKey.
5. streamFunctionGọi với { ...config, apiKey, identity, signal }.

Vì AgentLoopConfig kế thừa SimpleStreamOptions, cả đối tượng cấu hình được rải vào options (dòng 413), nên reasoning, sessionId, transport, thinkingBudgets, onPayload... tự đi tới pi-ai. identity là danh tính client (User-Agent, header attribution), mặc định ghi công cho pi và được gộp từng trường với giá trị của caller (agent-loop.ts:37–41), tính năng mới của 1.4.0. Dòng 419 bọc result() để ghi mức suy nghĩ đã yêu cầu lên assistant message cuối, dù hàm stream nào trả lời.

Nửa sau đọc luồng sự kiện của pi-ai và giữ cho transcript luôn có một assistant message đang lớn dần:

for await (const event of response) {	switch (event.type) {		case "start":			partialMessage = event.partial;			context.messages.push(partialMessage);			addedPartial = true;			await emit({ type: "message_start", message: { ...partialMessage } });			break;		case "text_start":		case "text_delta":		case "text_end":		case "thinking_start":		case "thinking_delta":		case "thinking_end":		case "toolcall_start":		case "toolcall_delta":		case "toolcall_end":			if (partialMessage) {				partialMessage = event.partial;				context.messages[context.messages.length - 1] = partialMessage;				await emit({					type: "message_update",					assistantMessageEvent: event,					message: { ...partialMessage },				});			}			break;		case "done":		case "error": {			const finalMessage = await result();			if (addedPartial) {				context.messages[context.messages.length - 1] = finalMessage;			} else {				context.messages.push(finalMessage);			}			if (!addedPartial) {				await emit({ type: "message_start", message: { ...finalMessage } });			}			await emit({ type: "message_end", message: finalMessage });			return finalMessage;		}

Ở sự kiện start, message dở dang được push thẳng vào context.messages; mỗi sự kiện delta thay phần tử cuối bằng bản partial mới; done hoặc error thay nó bằng message cuối cùng. Nhờ vậy, khi batch tool bắt đầu, context đã chứa đúng assistant message vừa yêu cầu các tool đó, và các hook beforeToolCall thấy nó. Lưu ý message_update chỉ được phát khi đã có start (dòng 442); một luồng chỉ có một sự kiện error (lỗi trước khi sinh) vẫn có đủ cặp message_start/message_end nhờ dòng 461–463.

Thực thi tool: tuần tự, song song, hook và terminate#

Mỗi lời gọi tool đi qua cùng một đường ống bốn bước, bất kể chế độ: chuẩn bị (tìm tool, chuẩn hoá và kiểm tra đối số, beforeToolCall), chạy (execute()), hoàn tất (afterToolCall), rồi phát (tool_execution_end và message toolResult). Chế độ thực thi chỉ quyết định các bước đó của nhiều lời gọi xen kẽ nhau thế nào.

async function executeToolCalls(	currentContext: AgentContext,	assistantMessage: AssistantMessage,	config: AgentLoopConfig,	signal: AbortSignal | undefined,	emit: AgentEventSink,): Promise<ExecutedToolCallBatch> {	const toolCalls = assistantMessage.content.filter((c) => c.type === "toolCall");	const hasSequentialToolCall = toolCalls.some(		(tc) => currentContext.tools?.find((t) => t.name === tc.name)?.executionMode === "sequential",	);	if (config.toolExecution === "sequential" || hasSequentialToolCall) {		return executeToolCallsSequential(currentContext, assistantMessage, toolCalls, config, signal, emit);	}	return executeToolCallsParallel(currentContext, assistantMessage, toolCalls, config, signal, emit);}

Mặc định là song song (Agent.toolExecution mặc định "parallel", agent.ts:261). Chỉ cần một lời gọi trong batch nhắm vào tool có executionMode: "sequential" là cả batch chạy tuần tự.

Chuẩn bị: từ khối toolCall tới đối số hợp lệ#

const tool = tools.find((t) => t.name === toolCall.name);if (!tool) {	return {		kind: "immediate",		result: createErrorToolResult(`Tool ${toolCall.name} not found`),		isError: true,	};}try {	const preparedToolCall = prepareToolCallArguments(tool, toolCall);	const validatedArgs = validateToolArguments(tool, preparedToolCall);	if (config.beforeToolCall) {		const beforeResult = await config.beforeToolCall(			{				assistantMessage,				toolCall,				args: validatedArgs,				context: currentContext,			},			signal,		);		if (signal?.aborted) {			return {				kind: "immediate",				result: createErrorToolResult("Operation aborted"),				isError: true,			};		}		if (beforeResult?.block) {			const result = createErrorToolResult(beforeResult.reason || "Tool execution was blocked");			if (beforeResult.terminate === true) {				result.terminate = true;			}			return {				kind: "immediate",				result,				isError: true,			};		}

Mọi thất bại ở bước này đều thành một kết quả "immediate" có isError: true, không bao giờ là exception: tool không tồn tại (“Tool X not found”), đối số không qua schema, hook chặn, hoặc run đã bị huỷ. Phần cuối hàm (agent-loop.ts:767–786) kiểm tra abort lần nữa, và bọc mọi exception thành kết quả lỗi mang thông báo của exception, nên mô hình đọc được lý do và có thể sửa lời gọi.

Thứ tự bên trong có ý nghĩa. prepareArguments của tool chạy trước (agent-loop.ts:704–716), rồi validateToolArguments() của pi-ai sao chép đối số, ép kiểu theo schema TypeBox và kiểm tra; nếu JSON của mô hình hỏng, pi-ai đã ghi argumentsError lên khối toolCall và hàm này throw ngay (validation.ts:317–323). beforeToolCall nhận đối số đã kiểm tra. Nó có thể sửa đối tượng đó tại chỗ, và tool chạy với bản đã sửa mà không kiểm tra lại; có hẳn một test cho hành vi này (agent-loop.test.ts:528). Riêng sự kiện tool_execution_start luôn mang toolCall.arguments thô, chưa qua chuẩn hoá.

Chạy và hoàn tất#

async function executePreparedToolCall(	prepared: PreparedToolCall,	signal: AbortSignal | undefined,	onUpdate: ToolUpdateSink,): Promise<ExecutedToolCallOutcome> {	const updateEvents: Promise<void>[] = [];	let acceptingUpdates = true;	const startedAt = performance.now();	const elapsed = () => Math.round(performance.now() - startedAt);	try {		const result = await prepared.tool.execute(			prepared.toolCall.id,			prepared.args as never,			signal,			(partialResult) => {				if (!acceptingUpdates) return;				updateEvents.push(Promise.resolve(onUpdate(partialResult)));			},		);		const durationMs = elapsed();		acceptingUpdates = false;		await Promise.all(updateEvents);		return { result, isError: result.isError === true, durationMs };	} catch (error) {		const durationMs = elapsed();		acceptingUpdates = false;		await Promise.all(updateEvents);		return {			result: createErrorToolResult(error instanceof Error ? error.message : String(error)),			isError: true,			durationMs,		};	} finally {		acceptingUpdates = false;	}}

Cờ acceptingUpdates là cơ chế khiến các lời gọi onUpdate muộn bị bỏ qua. Thời gian chỉ đo riêng execute(). Exception thành kết quả lỗi với thông báo của nó, đúng như README dặn: tool hỏng thì cứ throw, đừng trả thông báo lỗi trong content (README.md:495).

if (config.afterToolCall) {	try {		const afterResult = await config.afterToolCall(			{				assistantMessage,				toolCall: prepared.toolCall,				args: prepared.args,				result,				isError,				context: currentContext,			},			signal,		);		if (afterResult) {			// Structured content not replaced along with the content may no longer match it.			const structuredContent =				afterResult.structuredContent ?? (afterResult.content ? undefined : result.structuredContent);			result = {				...result,				content: afterResult.content ?? result.content,				details: afterResult.details ?? result.details,				usage: afterResult.usage ?? result.usage,				terminate: afterResult.terminate ?? result.terminate,			};			if (structuredContent === undefined) delete result.structuredContent;			else result.structuredContent = structuredContent;			isError = afterResult.isError ?? isError;		}	} catch (error) {		result = createErrorToolResult(error instanceof Error ? error.message : String(error));		isError = true;	}

afterToolCall ghi đè theo từng trường, không gộp sâu: trả về content thì thay cả mảng content. Có một ngoại lệ cố ý ở dòng 894–896: nếu hook thay content mà không trả structuredContent, thì structuredContent cũ bị bỏ, vì có thể nó không còn khớp với content mới. Hook throw thì kết quả trở thành lỗi thay vì làm hỏng cả batch, một sửa lỗi ở 0.67.67 (CHANGELOG.md:419).

Tuần tự và song song khác nhau ở thứ tự sự kiện#

for (const toolCall of toolCalls) {	await emit({		type: "tool_execution_start",		toolCallId: toolCall.id,		toolName: toolCall.name,		args: toolCall.arguments,	});	const preparation = await prepareToolCall(currentContext, assistantMessage, toolCall, config, signal);	let finalized: FinalizedToolCallOutcome;	if (preparation.kind === "immediate") {		finalized = {			toolCall,			result: preparation.result,			isError: preparation.isError,		};	} else {		const executed = await executePreparedToolCall(preparation, signal, emitToolExecutionUpdate(toolCall, emit));		finalized = await finalizeExecutedToolCall(			currentContext,			assistantMessage,			preparation,			executed,			config,			signal,		);	}	await emitToolExecutionEnd(finalized, emit);	const toolResultMessage = createToolResultMessage(finalized);	await emitToolResultMessage(toolResultMessage, emit);	finalizedCalls.push(finalized);	messages.push(toolResultMessage);	if (signal?.aborted) {		break;	}}

Ở chế độ tuần tự, mỗi lời gọi phát tool_execution_start, chạy, phát tool_execution_end và message toolResult, rồi mới tới lời gọi sau. Nếu run bị huỷ, vòng for dừng sau lời gọi đang chạy (dòng 585–587); các lời gọi chưa tới không có kết quả nào trong transcript.

for (const toolCall of toolCalls) {	await emit({		type: "tool_execution_start",		toolCallId: toolCall.id,		toolName: toolCall.name,		args: toolCall.arguments,	});	const preparation = await prepareToolCall(currentContext, assistantMessage, toolCall, config, signal);	if (preparation.kind === "immediate") {		const finalized = {			toolCall,			result: preparation.result,			isError: preparation.isError,		} satisfies FinalizedToolCallOutcome;		await emitToolExecutionEnd(finalized, emit);		finalizedCalls.push(finalized);		if (signal?.aborted) {			break;		}		continue;	}

	finalizedCalls.push(async () => {		if (signal?.aborted) {			const finalized = {				toolCall,				result: createErrorToolResult("Operation aborted"),				isError: true,			} satisfies FinalizedToolCallOutcome;			await emitToolExecutionEnd(finalized, emit);			return finalized;		}		const executed = await executePreparedToolCall(preparation, signal, emitToolExecutionUpdate(toolCall, emit));		const finalized = await finalizeExecutedToolCall(			currentContext,			assistantMessage,			preparation,			executed,			config,			signal,		);		await emitToolExecutionEnd(finalized, emit);		return finalized;	});	if (signal?.aborted) {		break;	}}const orderedFinalizedCalls = await Promise.all(	finalizedCalls.map((entry) => (typeof entry === "function" ? entry() : Promise.resolve(entry))),);const messages: ToolResultMessage[] = [];for (const finalized of orderedFinalizedCalls) {	const toolResultMessage = createToolResultMessage(finalized);	await emitToolResultMessage(toolResultMessage, emit);	messages.push(toolResultMessage);}return {	messages,	terminate: shouldTerminateToolBatch(orderedFinalizedCalls),};

Ở chế độ song song có hai pha. Pha 1 chuẩn bị tuần tự: mọi tool_execution_start và mọi beforeToolCall chạy lần lượt theo thứ tự trong assistant message. Lời gọi bị chặn hay lỗi đối số được phát tool_execution_end ngay. Lời gọi hợp lệ được gói thành một closure chưa chạy. Pha 2 chạy mọi closure cùng lúc bằng Promise.all; mỗi closure phát tool_execution_end ngay khi nó xong, tức theo thứ tự hoàn thành. Chỉ khi tất cả đã xong, các message toolResult mới được phát, theo thứ tự nguồn (dòng 660–664). Test ở agent-loop.test.ts:668 khẳng định đúng điều đó: tool_execution_end ra ["tool-2", "tool-1"] còn toolResult ra ["tool-1", "tool-2"].

start #1
end #1    toolResult #1
start #2
end #2    toolResult #2

start #1   start #2      (pha 1)
end #2     end #1        (thứ tự xong)
toolResult #1
toolResult #2            (thứ tự nguồn)

Vì sao chuẩn bị lại tuần tự? beforeToolCall là chỗ coding agent chạy hook tool_call của extension, nơi kiểm tra quyền và hỏi người dùng. Khi chuyển sang chạy song song ở 0.58.0, CHANGELOG của coding agent ghi rõ preflight tuần tự được giữ lại cho việc chặn của extension (coding-agent/CHANGELOG.md:3296). Hệ quả thấy được: các hộp thoại hỏi quyền hiện lần lượt, theo thứ tự lời gọi trong message. Còn ghi transcript theo thứ tự nguồn thì làm cho transcript được lưu ra là tất định, không phụ thuộc tool nào chạy nhanh hơn.

terminate: dừng sau batch#

Một tool có thể trả terminate: true để gợi ý rằng không cần gọi mô hình thêm lần nào sau batch này, ví dụ một tool “báo xong việc”. Hook beforeToolCall (khi chặn) và afterToolCall cũng đặt được cờ này. Quy tắc mang tính cả batch: chỉ khi mọi kết quả đã hoàn tất đều đặt terminate, vòng lặp mới không tự gọi mô hình tiếp (shouldTerminateToolBatch, agent-loop.ts:700–702). Một batch lẫn lộn chạy bình thường. Và terminate chỉ bỏ qua lần gọi tự động; tin steer hay follow-up đang chờ vẫn được xử lý.

Lời gọi tool bị cắt cụt vì giới hạn token#

/** * Fail all tool calls from an assistant message that was truncated by the * output token limit. 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; report each as an error so the model can re-issue them. */async function failToolCallsFromTruncatedMessage(	toolCalls: AgentToolCall[],	emit: AgentEventSink,): Promise<ExecutedToolCallBatch> {	const messages: ToolResultMessage[] = [];	for (const toolCall of toolCalls) {		await emit({			type: "tool_execution_start",			toolCallId: toolCall.id,			toolName: toolCall.name,			args: toolCall.arguments,		});		const finalized: FinalizedToolCallOutcome = {			toolCall,			result: createErrorToolResult(				`Tool call "${toolCall.name}" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments.`,			),			isError: true,		};		await emitToolExecutionEnd(finalized, emit);		const toolResultMessage = createToolResultMessage(finalized);		await emitToolResultMessage(toolResultMessage, emit);		messages.push(toolResultMessage);	}	return { messages, terminate: false };}

Khi mô hình chạm giới hạn token đầu ra giữa chừng, đối số tool đang stream bị cắt. Bộ đọc JSON cứu vãn của pi-ai vẫn có thể dựng ra một đối tượng hợp lệ, chỉ là thiếu nội dung, và nó còn qua được schema. Chạy lời gọi đó (ví dụ một lệnh write với nội dung bị cụt) là nguy hiểm. Vì vậy mọi lời gọi trong message có stopReason === "length" đều bị trả lỗi, kèm lời nhắn bảo mô hình gửi lại lời gọi với đối số đầy đủ, và terminate: false bảo đảm mô hình được gọi lại để làm việc đó. Test ở agent-loop.test.ts:456 kiểm tra rằng tool không bao giờ chạy và vòng lặp gọi mô hình lần thứ hai.

runToolCall: cùng đường ống cho lời gọi lồng nhau#

/** * Run one tool call through the same steps as a model-issued call: argument preparation, schema * validation, `beforeToolCall`, execution, and `afterToolCall`. Emits no events and adds no * messages. Tools that call other tools use this so the hooks (for example permission checks) * apply to those calls too. * * Never rejects for tool failures: unknown tools, validation errors, blocked calls, and thrown * errors come back as `isError: true`. */export async function runToolCall(toolCall: AgentToolCall, options: RunToolCallOptions): Promise<AgentToolCallOutcome> {	const { assistantMessage, context, signal } = options;	const preparation = await prepareToolCall(context, assistantMessage, toolCall, options, signal, options.tools);	if (preparation.kind === "immediate") {		return { toolCall, result: preparation.result, isError: preparation.isError };	}	const executed = await executePreparedToolCall(preparation, signal, options.onUpdate ?? (() => {}));	return finalizeExecutedToolCall(context, assistantMessage, preparation, executed, options, signal);}

Một tool đôi khi gọi tool khác, ví dụ script codemode gọi tools.read(...). Nếu nó gọi thẳng execute(), các hook kiểm tra quyền bị bỏ qua. runToolCall() chạy đúng các bước chuẩn bị, chạy và hoàn tất như lời gọi của mô hình, nhưng không phát sự kiện và không thêm message. Ví dụ examples/mcp-codemode dùng nó cho tool codemode (tools.ts:83–103); chi tiết ở chương Codemode.

declareToolChanges: báo thay đổi bộ tool giữa hội thoại#

Vấn đề. Bộ tool chạy được có thể đổi giữa hội thoại: extension bật một tool, người dùng tắt bash, một server MCP nạp thêm tool. Nếu mỗi request gửi danh sách tool hiện tại ở một trường riêng, thì mọi thay đổi đều làm đổi tiền tố của request và prompt cache bị trượt. Bài “Prompt Caching In Agents” của Earendil nói đúng điều này: thay đổi bộ tool phá cache, và cách sửa là nạp tool kiểu cộng dồn, neo vào transcript (Prompt Caching In Agents). Bài đó (07/2026) mô tả cơ chế lúc ấy là ghi tên tool mới lên kết quả tool (addedToolNames, 0.80.7); từ 0.86.0, prompt và khai báo tool chuyển hẳn vào các system message của transcript (ai/CHANGELOG.md:284), và đó là cơ chế ở v1.4.0.

Giải pháp. SystemMessage của pi-ai có toolsAdded (định nghĩa đầy đủ) và toolsRemoved (chỉ tên) (ai/types.ts:544–570). Phát lại mọi system message theo thứ tự cho ra bộ tool hiện hành (getCurrentTools). Còn context.tools là thứ runtime chạy được. Trước mỗi request, vòng lặp so hai thứ đó và, nếu khác, chèn một system message ghi phần chênh lệch:

/** * Declare tool loadout changes to the model. * * `context.tools` is what the runtime can execute; the transcript's system messages declare * what the model may call. Before each request the difference becomes `toolsAdded` and * `toolsRemoved` on a system message. When a pending system message exists, its tool fields * are treated as intent and replaced with the delta between the committed transcript and * the executable set, so replay always yields exactly `context.tools`. Otherwise a new * system message is inserted before the first non-system pending message. */function declareToolChanges(context: AgentContext, pendingMessages: AgentMessage[]): AgentMessage[] {	let systemIndex = -1;	for (let i = pendingMessages.length - 1; i >= 0; i--) {		if (pendingMessages[i].role === "system") {			systemIndex = i;			break;		}	}	const pending = pendingMessages[systemIndex] as SystemMessage | undefined;	const baseline = pending		? pendingMessages.map((message, index) =>				index === systemIndex ? withToolChanges(pending, NO_CHANGES) : message,			)		: pendingMessages;	const changes = getToolStateChanges(		getCurrentTools([...context.messages, ...baseline]),		(context.tools ?? []).map(toToolDeclaration),	);	const unchanged = changes.toolsAdded.length === 0 && changes.toolsRemoved.length === 0;	if (pending) {		// Keep the caller's message object when it already declares no tool changes.		if (unchanged && !pending.toolsAdded?.length && !pending.toolsRemoved?.length) return pendingMessages;		return baseline.map((message, index) => (index === systemIndex ? withToolChanges(pending, changes) : message));	}	if (unchanged) return pendingMessages;	const update = withToolChanges({ role: "system", content: "", timestamp: Date.now() }, changes);	const insertIndex = pendingMessages.findIndex((message) => message.role !== "system");	const index = insertIndex === -1 ? pendingMessages.length : insertIndex;	return [...pendingMessages.slice(0, index), update, ...pendingMessages.slice(index)];}

Hai trường hợp. Nếu trong các tin nhắn chờ không có system message nào, một system message mới với content: "" và phần chênh lệch được chèn trước tin nhắn chờ đầu tiên không phải system (dòng 367–371). Nếu caller đã tự gửi kèm một system message (chẳng hạn để đổi một mục prompt), phần tool của message đó bị viết lại cho khớp bộ chạy được: bộ chạy được luôn thắng, dù message có “xin” thêm hay bớt tool nào (dòng 351–355, 362–366; xem test ở agent.test.ts:247). Một tool đổi định nghĩa được tính là bỏ rồi thêm (transcript.ts:149).

Hàm được gọi ở hai chỗ: cho các message của prompt (agent-loop.ts:119) và trước mỗi request trong vòng trong (:220), kể cả khi không có tin nhắn chờ nào. Test ở agent.test.ts:164–198 cho thấy hiệu quả: sau khi đổi agent.state.tools từ [first] sang [second], request kế tiếp mang thêm một system message +second, -first, còn các message trước đó, kể cả system message đầu tiên, không bị sửa. Provider nào nhận được system message giữa hội thoại thì gửi tại chỗ; provider khác được pi-ai gộp lại thành một system message đầu (chương pi-ai).

Lớp Agent: state, hàng đợi và vòng đời một lần chạy#

Agent thêm vào vòng lặp ba thứ mà một ứng dụng tương tác cần: state sống lâu hơn một lần chạy, hai hàng đợi tin nhắn, và một vòng đời rõ ràng (đang chạy, huỷ, chờ rảnh).

State chỉ đổi qua sự kiện#

private createContextSnapshot(): AgentContext {	return {		messages: this._state.messages.slice(),		tools: this._state.tools.slice(),	};}private createLoopConfig(options: { skipInitialSteeringPoll?: boolean } = {}): AgentLoopConfig {	let skipInitialSteeringPoll = options.skipInitialSteeringPoll === true;	return {		model: this._state.model,		reasoning: this._state.thinkingLevel === "off" ? undefined : this._state.thinkingLevel,⋯ lược bớt dòng 481–502		convertToLlm: this.convertToLlm,		transformContext: this.transformContext,		getApiKey: this.getApiKey,		getSteeringMessages: async () => {			if (skipInitialSteeringPoll) {				skipInitialSteeringPoll = false;				return [];			}			return this.steeringQueue.drain();		},		getFollowUpMessages: async () => this.followUpQueue.drain(),	};}

Mỗi lần chạy, Agent đưa cho vòng lặp một bản sao của transcript và bộ tool (dòng 469–474). Vòng lặp chỉ sửa bản sao đó. State của Agent thay đổi ở một chỗ duy nhất, hàm processEvents(), như một reducer chạy trên từng sự kiện, rồi mới await từng listener theo thứ tự đăng ký:

private async processEvents(event: AgentEvent): Promise<void> {	switch (event.type) {		case "message_start":			this._state.streamingMessage = event.message;			break;		case "message_update":			this._state.streamingMessage = event.message;			break;		case "message_end":			this._state.streamingMessage = undefined;			this._state.messages.push(event.message);			break;⋯ lược bớt dòng 589–603		case "turn_end":			if (event.message.role === "assistant" && event.message.errorMessage) {				this._state.errorMessage = event.message.errorMessage;			}			break;		case "agent_end":			this._state.streamingMessage = undefined;			break;	}	const signal = this.activeRun?.abortController.signal;	if (!signal) {		throw new Error("Agent listener invoked outside active run");	}	for (const listener of this.listeners) {		await listener(event, signal);	}

Vì emit của Agent await toàn bộ chuỗi trên, mỗi sự kiện là một rào chắn: vòng lặp không đi tiếp cho tới khi state đã cập nhật và mọi listener đã xong. README nêu hệ quả quan trọng nhất: khi beforeToolCall chạy, state đã chứa assistant message yêu cầu tool đó (README.md:122). Listener còn nhận AbortSignal của lần chạy để huỷ việc của chính nó.

prompt() và continue()#

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, or wait for completion.",		);	}	const messages = this.normalizePromptInput(input, images);	await this.runPromptMessages(messages);}/** Continue from the current transcript. The last message must be a user or tool-result message. */async continue(): Promise<void> {	if (this.activeRun) {		throw new Error("Agent is already processing. Wait for completion before continuing.");	}	const lastMessage = this._state.messages[this._state.messages.length - 1];	if (!lastMessage || this._state.messages.every((message) => message.role === "system")) {		throw new Error("No messages to continue from");	}	if (lastMessage.role === "assistant") {		const queuedSteering = this.steeringQueue.drain();		if (queuedSteering.length > 0) {			await this.runPromptMessages(queuedSteering, { skipInitialSteeringPoll: true });			return;		}		const queuedFollowUps = this.followUpQueue.drain();		if (queuedFollowUps.length > 0) {			await this.runPromptMessages(queuedFollowUps);			return;		}		throw new Error("Cannot continue from message role: assistant");	}	await this.runContinuation();}

prompt() nhận một chuỗi (kèm ảnh tuỳ chọn), một message hoặc một mảng message; chuỗi được bọc thành một message user (agent.ts:422–439). Gọi prompt() khi đang chạy là lỗi, với lời nhắn chỉ rõ cách đúng: dùng steer() hoặc followUp(). Quy tắc này có từ 0.32.0 để tránh hai lần chạy giẫm lên cùng một transcript (CHANGELOG.md:749).

continue() có ba nhánh. Transcript rỗng hoặc chỉ có system message thì từ chối. Đuôi không phải assistant (thường là user hoặc toolResult) thì chạy runAgentLoopContinue(), ví dụ để thử lại sau lỗi. Đuôi là assistant thì không thể gửi thẳng, nên nó lấy một lô từ hàng đợi steer, nếu rỗng thì một lô từ hàng đợi follow-up, và chạy lô đó như một prompt. Khi lô đến từ hàng đợi steer, lần hỏi steer đầu tiên của vòng lặp bị bỏ qua một lần (cờ skipInitialSteeringPoll), nếu không chế độ one-at-a-time sẽ giao hai tin steer trong cùng một lượt.

Hai hàng đợi#

class PendingMessageQueue {	private messages: AgentMessage[] = [];	public mode: QueueMode;	constructor(mode: QueueMode) {		this.mode = mode;	}	enqueue(message: AgentMessage): void {		this.messages.push(message);	}	hasItems(): boolean {		return this.messages.length > 0;	}	peek(): AgentMessage[] {		if (this.mode === "all") return this.messages.slice();		const first = this.messages[0];		return first ? [first] : [];	}	drain(): AgentMessage[] {		const drained = this.peek();		this.messages = this.messages.slice(drained.length);		return drained;	}	clear(): void {		this.messages = [];	}}

Cả hai hàng đợi mặc định "one-at-a-time" (agent.ts:255–256): mỗi điểm rút chỉ lấy tin cũ nhất, tin còn lại chờ điểm rút sau. Với "all", mọi tin chờ được chèn cùng lúc. steer() và followUp() chỉ là enqueue đồng bộ (agent.ts:307–315); vòng lặp tự hỏi hàng đợi ở các điểm đã thấy trong runLoop, thông qua hai closure getSteeringMessages/getFollowUpMessages mà createLoopConfig() nối vào drain(). peekQueuedMessages() cho biết lô nào sẽ được chọn kế tiếp mà không lấy ra, với steer được ưu tiên hơn follow-up (agent.ts:338–342).

Vòng đời: chạy, huỷ, chờ rảnh, reset#

private async runWithLifecycle(executor: (signal: AbortSignal) => Promise<void>): Promise<void> {	if (this.activeRun) {		throw new Error("Agent is already processing.");	}	const abortController = new AbortController();	let resolvePromise = () => {};	const promise = new Promise<void>((resolve) => {		resolvePromise = resolve;	});	this.activeRun = { promise, resolve: resolvePromise, abortController };	this._state.isStreaming = true;	this._state.streamingMessage = undefined;	this._state.errorMessage = undefined;	try {		await executor(abortController.signal);	} catch (error) {		await this.handleRunFailure(error, abortController.signal.aborted);	} finally {		this.finishRun();	}}⋯ lược bớt dòng 541–559private finishRun(): void {	this._state.isStreaming = false;	this._state.streamingMessage = undefined;	this._state.pendingToolCalls = new Set<string>();	this.activeRun?.resolve();	this.activeRun = undefined;}

Mỗi lần chạy có một AbortController riêng và một promise được resolve trong finishRun(). abort() chỉ gọi abortController.abort() (agent.ts:349–352); tín hiệu đó tới hàm stream, tới execute() của tool và tới các hook, và mỗi bên tự tôn trọng nó. waitForIdle() trả về chính promise kia (agent.ts:359–361), nên nó resolve sau khi listener của agent_end đã xong, cùng lúc với prompt().

Nếu vòng lặp throw (ví dụ convertToLlm vi phạm hợp đồng), handleRunFailure() dựng một assistant message rỗng với stopReason "aborted" hoặc "error" và phát đủ message_start, message_end, turn_end, agent_end (agent.ts:542–558), để giao diện không bao giờ bị treo ở trạng thái “đang chạy”. reset() từ chối khi đang chạy, xoá hai hàng đợi và thu transcript về đúng một system message phát lại từ mọi system message cũ, nên prompt và bộ tool hiện hành được giữ (agent.ts:363–377).

Trace: một prompt, hai lời gọi tool và một tin nhắn steer#

Ghép mọi mảnh lại bằng một kịch bản cụ thể. Một Agent dùng cấu hình mặc định (tool chạy song song, hàng đợi one-at-a-time, không hook nào), có một tool read. Người dùng gửi một prompt; mô hình gọi read hai lần trong cùng một câu trả lời; trong lúc hai tool đang chạy, người dùng gõ thêm một tin steer.

Ví dụ minh hoạ: cách gọi
const run = agent.prompt("Đọc a.txt và b.txt rồi tóm tắt");
// ... trong lúc hai tool read đang chạy, từ một handler của giao diện:
agent.steer({ role: "user", content: "Chỉ tóm tắt a.txt thôi", timestamp: Date.now() });
await run;

Đây là chuỗi sự kiện chính xác mà subscriber nhận được, cùng với dòng code phát ra mỗi sự kiện. Giả sử b.txt đọc xong trước a.txt.

Ví dụ minh hoạ: chuỗi AgentEvent của lần chạy
 #   sự kiện                                   nguồn (agent-loop.ts)    agent.state sau reducer
 1   agent_start                               runAgentLoop :126        (isStreaming đã true từ trước)
 2   turn_start                                :127                     lượt 1
 3   message_start  user "Đọc a.txt và b.txt…"   :129
 4   message_end    user                       :130                     messages += user
     runLoop: hỏi steer lần đầu → rỗng; declareToolChanges → không đổi; gọi mô hình
 5   message_start  assistant (partial rỗng)   :430                     streamingMessage = partial
 6   message_update toolcall_start/delta/end   :445   (#0 call_1 read {path:"a.txt"})
 7   message_update toolcall_start/delta/end   :445   (#1 call_2 read {path:"b.txt"})
 8   message_end    assistant, stopReason "toolUse"  :464               messages += assistant
     executeToolCallsParallel, pha 1: chuẩn bị tuần tự
 9   tool_execution_start call_1               :607                     pendingToolCalls {call_1}
10   tool_execution_start call_2               :607                     {call_1, call_2}
     pha 2: hai execute() chạy cùng lúc; người dùng gọi agent.steer(...) → vào hàng đợi
11   tool_execution_end   call_2               :648  thứ tự hoàn thành   {call_1}
12   tool_execution_end   call_1               :648                     {}
13   message_start  toolResult call_1          :957  thứ tự nguồn
14   message_end    toolResult call_1          :958                     messages += toolResult 1
15   message_start  toolResult call_2          :957
16   message_end    toolResult call_2          :958                     messages += toolResult 2
17   turn_end       {assistant, toolResults: [call_1, call_2]}  :296
     hỏi hàng đợi steer → [tin steer]; hasMoreToolCalls = true
18   turn_start                                :216                     lượt 2
19   message_start  user "Chỉ tóm tắt a.txt thôi"  :221
20   message_end    user                       :222                     messages += tin steer
21   message_start  assistant (partial)        :430
22   message_update text_start, text_delta…, text_end   :445
23   message_end    assistant, stopReason "stop"  :464                  messages += assistant
24   turn_end       {assistant, toolResults: []}  :296
     hỏi steer → rỗng, vòng trong thoát; hỏi follow-up → rỗng; không có continue → dừng
25   agent_end      messages: [user, asst₁, toolResult₁, toolResult₂, steer, asst₂]  :329
     listener của agent_end chạy xong → finishRun(): isStreaming = false; prompt() resolve

Những điều trace này cho thấy:

• Tin steer không cắt ngang tool. Nó nằm trong hàng đợi suốt lúc hai tool chạy và chỉ được rút ở dòng 304, sau turn_end của lượt 1. Không tool nào bị bỏ qua.
• Tin steer không tốn thêm request. Lượt 2 vốn đã xảy ra vì có tool call (hasMoreToolCalls); tin steer chỉ được chèn vào trước request đó. Mô hình thấy theo thứ tự: hai toolResult, rồi chỉ dẫn mới.
• agent.state.messages lớn dần đúng theo các message_end; lúc listener xử lý sự kiện 9, state đã có assistant message của sự kiện 8.
• Nếu thay steer() bằng followUp(), tin nhắn sẽ chờ tới sau sự kiện 24, khi agent lẽ ra đã dừng, và tạo ra lượt thứ ba. Nếu câu trả lời đầu không gọi tool nào, tin steer vẫn tạo ra lượt 2, vì vòng trong chạy tiếp khi còn tin nhắn chờ.
• Ở chế độ tuần tự, sự kiện 9–16 sẽ thành start 1, end 1, toolResult 1, start 2, end 2, toolResult 2.

Quyết định thiết kế#

Lặp cho tới khi mô hình thôi gọi tool, không có max steps#

Code ở v1.4.0 giữ đúng lựa chọn đó: AgentLoopConfig không có trường đếm bước nào, và điều kiện của vòng trong chỉ là hasMoreToolCalls || pendingMessages.length > 0. Đánh đổi là một mô hình mắc kẹt có thể lặp mãi. Thay vì một con số tuỳ tiện, các điểm dừng là có chủ đích: người dùng huỷ, finishTurn trả end, cả batch xin terminate, hoặc beforeToolCall chặn tool. Một ứng dụng muốn giới hạn số lượt hoàn toàn viết được giới hạn đó trong finishTurn.

Hàng đợi tin nhắn giữa các lượt, không cắt ngang giữa lượt#

Bài blog mô tả ý tưởng gốc: “after each turn, it asks for queued messages and injects them before the next assistant response”, với hai chế độ “one-at-a-time or all-at-once”. Thiết kế này đã đổi hai lần. Ở 0.32.0, một hàng đợi duy nhất tách thành steer() và followUp(), và khi đó steer còn bỏ qua các tool còn lại của batch (CHANGELOG.md:737–739). Ở 0.58.4, steer chuyển sang chờ cả batch tool chạy xong (CHANGELOG.md:507). Hệ quả có thể thấy trực tiếp trong code: mọi lời gọi tool đều có một kết quả thật, và transcript không bao giờ chứa nửa batch. Cái giá là độ trễ: tin steer phải chờ tool chậm nhất của batch. Muốn dừng ngay thì đã có abort().

AgentMessage đi tới tận ranh giới gọi mô hình#

Ở 0.31.0, messageTransformer đổi tên thành convertToLlm, và việc xử lý tệp đính kèm chuyển hẳn sang hàm này (CHANGELOG.md:761, :768). Lý do là transcript phải chứa được mọi thứ ứng dụng cần nhớ, còn mỗi request chỉ cần phần mô hình hiểu. Giữ hai kho riêng sẽ buộc ứng dụng tự đồng bộ thứ tự giữa chúng; một kho với một phép chiếu ở ranh giới thì không. Lệnh !!cmd của coding agent là ví dụ: nó nằm trong transcript và hiện trên giao diện, nhưng convertToLlm bỏ nó khỏi request.

State dựng từ sự kiện, listener là rào chắn#

Ở 0.65.0, listener của subscribe() bắt đầu được await và nhận AbortSignal; agent_end thôi là ranh giới rảnh (CHANGELOG.md:471–475). Đánh đổi rõ ràng: một listener chậm làm chậm cả vòng lặp. Đổi lại, mọi việc listener làm cho một sự kiện, như ghi session hay chạy extension, đã xong trước khi vòng lặp sang bước kế tiếp, và người gọi await prompt() biết chắc mọi việc đó đã xong khi promise resolve.

Song song là mặc định, nhưng chuẩn bị và transcript thì tuần tự#

Chế độ song song có từ 0.58.0 và là mặc định ngay từ đầu (CHANGELOG.md:523); ở 0.68.1, tool_execution_end chuyển sang phát theo thứ tự hoàn thành để giao diện cập nhật ngay khi từng tool xong (:405). Thiết kế ba tầng (chuẩn bị tuần tự, chạy song song, ghi transcript theo thứ tự nguồn) giữ được tốc độ mà không đánh đổi tính tất định của những gì được lưu và gửi lại cho mô hình. Tool nào không an toàn khi chạy đồng thời thì tự khai báo executionMode: "sequential".

Hook là trường public, có thể bọc chồng#

Từ 0.65.0, Agent bỏ các setter và để mọi hook là trường public (CHANGELOG.md:454–466). Nhờ vậy nhiều lớp có thể bọc chồng lên nhau mà không cần kế thừa: AgentSession giữ hook cũ trong previousPrepareRequest và previousFinishTurn, rồi gọi nó bên trong hook mới (agent-session.ts:887–895). agent-core vì thế giữ được kích thước nhỏ: nó chỉ định nghĩa khi nào một hook chạy, còn làm gì trong hook thuộc về ứng dụng. Bài “What is a Harness?” của Earendil đặt vòng lặp agent cạnh system prompt, bộ tool và lớp dịch như các phần của harness (What is a Harness?); trong Pi, chỉ phần vòng lặp nằm ở gói này.

Cạm bẫy và trường hợp biên#

• Throw trong hook cấp thấp làm treo luồng. agentLoop() chạy runAgentLoop(...).then(...) mà không có catch (agent-loop.ts:56–67). Nếu convertToLlm hay transformContext throw, promise bị reject mà không ai bắt và luồng không bao giờ kết thúc. Lớp Agent bắt được trường hợp này nhờ handleRunFailure().
• agentLoopContinue() sửa mảng của bạn. Nó chỉ sao chép nông context (agent-loop.ts:153), nên các message mới được push thẳng vào context.messages mà caller truyền vào. agentLoop() thì tạo mảng mới (dòng 121–124). Agent luôn truyền một bản sao nên không bị ảnh hưởng.
• Đổi state giữa lần chạy không có hiệu lực ngay. Model, mức suy nghĩ và bộ tool được chụp lúc bắt đầu chạy (createContextSnapshot(), createLoopConfig()). Gán agent.state.tools trong lúc chạy chỉ có tác dụng ở lần chạy sau, trừ khi một hook như prepareNextTurn trả về context mới; đó chính là việc AgentSession làm (agent-session.ts:919–925).
• Abort không làm vòng lặp dừng ngay. Vòng lặp chỉ tự kiểm tra tín hiệu quanh các lời gọi tool: khi chuẩn bị, ngay trước khi chạy một closure song song, và sau mỗi tool tuần tự. Nếu huỷ trong lúc tool chạy, nó vẫn phát kết quả, phát turn_end, rồi gửi request kế tiếp và dựa vào hợp đồng của StreamFn để nhận về ngay một message "aborted". Một StreamFn tự viết mà phớt lờ signal sẽ khiến lần chạy tiếp tục. Ở bước đó hàng đợi steer cũng đã bị hỏi (dòng 304), nên một tin steer đang chờ có thể được ghi vào transcript ngay trước câu trả lời bị huỷ.
• Lời gọi tool không có kết quả. Huỷ giữa batch tuần tự để lại các lời gọi chưa chạy mà không có toolResult. pi-ai xử lý việc này khi dựng request: nó bỏ qua assistant message bị lỗi hoặc bị huỷ, và chèn kết quả lỗi “No result provided” cho lời gọi mồ côi (transform-messages.ts:179–224).
• { action: "continue" } vô điều kiện là vòng lặp vô tận, vì finishTurn chạy lại sau request mà chính nó đòi.
• continue() với đuôi assistant chỉ chạy được khi có tin chờ; nếu không, nó throw “Cannot continue from message role: assistant”. Đuôi là message tuỳ biến thì vòng lặp cấp thấp không kiểm tra được; convertToLlm phải biến nó thành user hoặc toolResult, nếu không provider sẽ từ chối (agent-loop.ts:76–78).
• terminate không lưu vào transcript. Nó chỉ là gợi ý lúc chạy; message toolResult được lưu vẫn là kết quả tool chuẩn (README.md:509).
• README còn sót cách nói cũ. README viết “LLMs only understand user, assistant, and toolResult” và ví dụ convertToLlm cấp thấp lọc bỏ cả role system (README.md:49, :546). Ở v1.4.0 system message mang prompt và khai báo tool, nên bộ lọc như vậy làm mất cả hai; bộ lọc mặc định của Agent giữ lại system.

Đọc code ở đâu#

Tóm tắt#

• pi-agent-core là một vòng lặp agent nhỏ (khoảng 2.500 dòng) trên pi-ai: gọi mô hình, chạy batch tool, lặp cho tới khi mô hình thôi gọi tool. Không có max steps.
• Transcript là AgentMessage[], mở rộng được bằng declaration merging; chỉ ở ranh giới gọi mô hình nó mới thành Message[] qua transformContext và convertToLlm.
• Vận chuyển chỉ là một hàm StreamFn không được throw; streamProxy() là một cài đặt qua server.
• Mỗi tool call đi qua chuẩn bị, chạy, hoàn tất; chế độ song song chuẩn bị tuần tự, chạy đồng thời và ghi transcript theo thứ tự nguồn. Lời gọi từ message bị cắt cụt ("length") không bao giờ chạy.
• Tin steer được chèn sau batch tool của lượt hiện tại, follow-up chỉ khi agent sắp dừng; cả hai mặc định one-at-a-time.
• Bộ tool chạy được và bộ tool mô hình thấy được giữ khớp nhau bằng system message toolsAdded/toolsRemoved trong transcript, để thay đổi giữa hội thoại không phá prompt cache.
• Lớp Agent dựng state từ sự kiện, await listener như rào chắn, và cung cấp prompt(), continue(), steer(), followUp(), abort(), waitForIdle(), reset().