pi 架构拆解

Deutsch

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

Deutsch

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

Appearance

EventStream: push hinein, Async-Iterator heraus ​

源码版本v0.73.1

utils/event-stream.ts ist das Streaming-Primitiv von pi-ai — eine generische EventStream<T, R>: der Producer pusht Events via push() hinein, der Konsument zieht sie über for await und holt das Endergebnis mit result(). Darunter ist kein WebSocket, kein Observable, sondern eine handgeschriebene Queue plus Waiter-Array, die eine push-basierte Producer-API in einen pull-basierten Async-Iterator übersetzt. Alle Provider pushen am Ende ihre Events in einen AssistantMessageEventStream (die in dieser Datei definierte konkrete Unterklasse von EventStream); der Aufrufer behandelt sie alle gleich.

Verantwortung ​

  1. Queue + Waiter: beim push wird, falls ein Konsument wartet, direkt resolve, sonst in die queue gelegt; siehe packages/ai/src/utils/event-stream.ts:20-35.
  2. Async-Iterator: [Symbol.asyncIterator] implementiert AsyncIterator<T>; erst Queue leeren, dann auf Waiter warten; siehe packages/ai/src/utils/event-stream.ts:49-61.
  3. Abschluss-Signal: isComplete(event) entscheidet, welches Event als „Endzustand" zählt, und triggert resolveFinalResult(extractResult(event)); siehe packages/ai/src/utils/event-stream.ts:23-26.
  4. Endergebnis-Promise: result(): Promise<R> gibt das aus dem End-Event extrahierte Ergebnis zurück und ist vom for await entkoppelt; siehe packages/ai/src/utils/event-stream.ts:63-65.
  5. end-Fallback: end(result?) erzwingt das Ende und leert alle Waiter; siehe packages/ai/src/utils/event-stream.ts:37-47.
  6. AssistantMessage-Unterklasse: AssistantMessageEventStream behandelt done / error als Endzustand und done.message / error.error als Ergebnis; siehe packages/ai/src/utils/event-stream.ts:68-82.

Entwurfsmotivation ​

Warum nicht einfach Nodes EventEmitter oder ein RxJS-Observable? Weil der Konsument die for await-Syntax will — die Pull-Geschwindigkeit wird vom Konsumenten gesteuert, Backpressure ergibt sich von selbst. EventEmitter ist Push ohne Backpressure; ein Observable muss per from zu einem Async-Iterator adaptiert werden. Eine 80-Zeilen-Klasse quetscht beides zusammen und bringt „End-Event triggert result-Promise" mit; die Funktion complete liefert mit stream(...).result() in einer Zeile das AssistantMessage, ohne dass der Konsument selbst akkumulieren muss.

Warum ist AssistantMessageEventStream eine Unterklasse statt eines generischen Parameters? Weil „welches Event ist Endzustand, wie extrahiert man das Ergebnis daraus" eine Provider-unabhängige Strategie ist — alle Provider pushen AssistantMessageEvent, und done.message ist das Endergebnis. Die Unterklasse schreibt die Strategie fest; Provider müssen nicht jeweils isComplete / extractResult übergeben.

Wichtige Dateien ​

push ist der einzige Eingang für den Producer, mit zwei Pfaden — Waiter bedienen oder einreihen:

typescript
// packages/ai/src/utils/event-stream.ts:20-35
push(event: T): void {
	if (this.done) return;

	if (this.isComplete(event)) {
		this.done = true;
		this.resolveFinalResult(this.extractResult(event));
	}

	// Deliver to waiting consumer or queue it
	const waiter = this.waiting.shift();
	if (waiter) {
		waiter({ value: event, done: false });
	} else {
		this.queue.push(event);
	}
}

Der Async-Iterator konsumiert erst bereits eingereihte Events und wartet erst, wenn die Queue leer ist:

typescript
// packages/ai/src/utils/event-stream.ts:49-61
async *[Symbol.asyncIterator](): AsyncIterator<T> {
	while (true) {
		if (this.queue.length > 0) {
			yield this.queue.shift()!;
		} else if (this.done) {
			return;
		} else {
			const result = await new Promise<IteratorResult<T>>((resolve) => this.waiting.push(resolve));
			if (result.done) return;
			yield result.value;
		}
	}
}

AssistantMessageEventStream schreibt die Strategie fest; der Provider muss sich um die Endzustandserkennung nicht kümmern:

typescript
// packages/ai/src/utils/event-stream.ts:68-82
export class AssistantMessageEventStream extends EventStream<AssistantMessageEvent, AssistantMessage> {
	constructor() {
		super(
			(event) => event.type === "done" || event.type === "error",
			(event) => {
				if (event.type === "done") {
					return event.message;
				} else if (event.type === "error") {
					return event.error;
				}
				throw new Error("Unexpected event type for final result");
			},
		);
	}
}

Datenfluss ​

Die Interaktion zwischen Producer und Konsument:

Grenzen und Fehler ​

  • push nach done: if (this.done) return ignoriert still; ein Provider, der nach error weiter pusht, produziert kein Problem; siehe packages/ai/src/utils/event-stream.ts:21-21.
  • end ohne End-Event: wenn end(result?) ohne result aufgerufen wird, kann finalResultPromise nie resolve; der Aufrufer sollte ein End-Event zuvor gepusht haben oder ein result übergeben; siehe packages/ai/src/utils/event-stream.ts:37-41.
  • error als Endzustand: extractResult von AssistantMessageEventStream gibt im error-Fall event.error zurück, sodass result() auch auf einem Fehlerstrom das Fehlerobjekt liefert, statt zu hangen; siehe packages/ai/src/utils/event-stream.ts:75-77.
  • Mehrere Konsumenten: mehrere gleichzeitige for await-Schleifen würden sich um die Waiter-Queue streiten und Events werden über verschiedene Konsumenten verteilt — in der Praxis ist es aber immer ein Konsument; das Code definiert kein Verhalten für mehrere.

Zusammenfassung ​

EventStream ist ein 80-Zeilen-Adapter von Push-zu-Async-Iterator; die Unterklasse AssistantMessageEventStream legt die Endstrategie fest. Provider pushen ausschließlich AssistantMessageEvent, der Aufrufer nutzt ausschließlich for await + result(). Wie ein Provider es nutzt, steht in Anthropic SSE-Implementierung; die obere Verteilung in stream/complete-Fassade.