effect-ai-streaming

Effect AI Streaming

When to Use This Skill

• Real-time streaming responses from language models
• Building chat interfaces with incremental updates
• Managing conversation history with streaming
• Protecting concurrent stream operations
• Accumulating stream parts with side effects
• Converting stream responses to prompt history

Import Patterns

CRITICAL: Always use namespace imports:

import * as Stream from 'effect/Stream';
import * as Effect from 'effect/Effect';
import * as Channel from 'effect/Channel';
import * as SubscriptionRef from 'effect/SubscriptionRef';
import * as Match from 'effect/Match';
import * as Response from 'effect/unstable/ai/Response';

StreamPart Protocol

stream := start → delta* → end

StreamPart lifecycle for each content type follows a three-phase protocol:

text      :: text-start → text-delta* → text-end
reasoning :: reasoning-start → reasoning-delta* → reasoning-end
toolParam :: tool-params-start → tool-params-delta* → tool-params-end
finish    :: { type: "finish", reason: FinishReason, usage: Usage }

Each streaming sequence has a unique id field that links start/delta/end parts.

Part Type Matching

Stream parts use a type field (not _tag), so use Match.when with type checks:

import * as Match from 'effect/Match';
import * as Effect from 'effect/Effect';

const processPart = (part: StreamPart) =>
	Match.value(part).pipe(
		Match.when({ type: 'text-delta' }, ({ delta }) =>
			Effect.sync(() => console.log(delta))
		),
		Match.when({ type: 'reasoning-delta' }, ({ delta }) =>
			Effect.sync(() => logReasoning(delta))
		),
		Match.when({ type: 'finish' }, ({ usage, reason }) =>
			Effect.sync(() => recordUsage(usage, reason))
		),
		Match.orElse(() => Effect.void)
	);

Direct type checks also work well for simple branching:

if (part.type === 'text-delta') {
	console.log(part.delta);
}

Accumulation Pattern

Accumulate stream parts incrementally using mutable state for efficiency: