pi 架构拆解

English

简体中文
日本語
Español
Français
Deutsch
繁體中文

English

简体中文
日本語
Español
Français
Deutsch
繁體中文

Appearance

Anthropic SSE implementation: from bytes to AssistantMessage ​

源码版本v0.73.1

providers/anthropic.ts is the heaviest of the 9 built-in providers — it implements SSE incremental decoding itself, converts Anthropic events to AssistantMessageEvent, injects OAuth/Claude Code identity, applies cache_control policy, and handles block-level incremental assembly for thinking and tool calls. It does not rely on the Anthropic SDK's streaming parser; it reads the byte stream of response.body directly and runs the SSE protocol itself. The final output is a sequence of events pushed into AssistantMessageEventStream, which consumers iterate with for await to get increments.

Responsibilities ​

  1. SSE byte-stream decoding: iterateSseMessages turns a ReadableStream<Uint8Array> into a ServerSentEvent async generator, supporting \r\n / \n and cross-chunk reassembly. See packages/ai/src/providers/anthropic.ts:321-369.
  2. Single-line decoding: decodeSseLine handles event: / data: / : comments / empty-line flush per the SSE spec. See packages/ai/src/providers/anthropic.ts:266-290.
  3. Event → typed object: iterateAnthropicEvents parses SSE data with parseJsonWithRepair into RawMessageStreamEvent, filters non-Anthropic events, and validates message_start / message_stop pairing. See packages/ai/src/providers/anthropic.ts:380-419.
  4. Streaming main entry: streamAnthropic constructs the client, builds params, calls client.messages.create({stream:true}), and pushes events to AssistantMessageEventStream. See packages/ai/src/providers/anthropic.ts:421-510.
  5. Simple version: streamSimpleAnthropic translates SimpleStreamOptions into the full AnthropicOptions, distinguishing adaptive thinking (Opus 4.6 / Sonnet 4.6) from budget-based thinking. See packages/ai/src/providers/anthropic.ts:717-756.
  6. Param construction: buildParams assembles messages / system / tools / thinking, and forces Claude Code identity injection for OAuth tokens. See packages/ai/src/providers/anthropic.ts:868-925.
  7. Message conversion: convertMessages turns the internal Message[] into Anthropic MessageParam[], handling text/image/tool blocks, cache_control, and surrogate cleanup. See packages/ai/src/providers/anthropic.ts:978-1070.

Design motivation ​

Why not just use @anthropic-ai/sdk's client.messages.stream()? Because pi-ai wants to control the event shape itself — all providers ultimately push the same AssistantMessageEvent, and the consumer side handles every provider with one for await code path. The SDK's built-in stream helper gives you the SDK's own event types, and converting them is an extra hop. Running SSE ourselves also lets us attach sse.raw to error messages when parsing fails, which is useful for tracking down CDN truncation or proxies rewriting responses.

Why split convertMessages and buildParams? Because OAuth tokens and plain API keys take different system prompt construction paths — buildParams decides whether to inject Claude Code identity, while convertMessages only handles per-message formatting. The two each own a layer: changing cache policy does not touch message conversion, and changing message conversion does not touch OAuth logic.

Key files ​

The core of SSE decoding, where an empty line triggers a flush that returns the full event:

typescript
// packages/ai/src/providers/anthropic.ts:266-290
function decodeSseLine(line: string, state: SseDecoderState): ServerSentEvent | null {
	if (line === "") {
		return flushSseEvent(state);
	}
	state.raw.push(line);
	if (line.startsWith(":")) {
		return null;
	}
	const delimiterIndex = line.indexOf(":");
	const fieldName = delimiterIndex === -1 ? line : line.slice(0, delimiterIndex);
	let value = delimiterIndex === -1 ? "" : line.slice(delimiterIndex + 1);
	if (value.startsWith(" ")) {
		value = value.slice(1);
	}
	if (fieldName === "event") {
		state.event = value;
	} else if (fieldName === "data") {
		state.data.push(value);
	}
	return null;
}

The Anthropic event iterator throws on pairing failure, rather than silently swallowing a truncated stream:

typescript
// packages/ai/src/providers/anthropic.ts:416-418
	if (sawMessageStart && !sawMessageEnd) {
		throw new Error("Anthropic stream ended before message_stop");
	}

OAuth tokens go through a special system prompt that forces Claude Code identity:

typescript
// packages/ai/src/providers/anthropic.ts:883-890
	if (isOAuthToken) {
		params.system = [
			{
				type: "text",
				text: "You are Claude Code, Anthropic's official CLI for Claude.",
				...(cacheControl ? { cache_control: cacheControl } : {}),
			},
		];

Data flow ​

The full chain from byte stream to event stream:

Boundaries and failure modes ​

Summary ​

anthropic.ts runs SSE itself, chaining four transformations — byte stream → ServerSentEvent → RawMessageStreamEvent → AssistantMessageEvent — and handles OAuth identity, cache_control, and thinking policy in buildParams / convertMessages. For the event stream data structure, see EventStream async iterator; for how providers register into the table, see Provider abstraction and built-ins.