Agents/Streaming/ — LLM Response Streaming Pipeline

Real-time LLM response streaming with provider-specific adapters

The streaming sub-module manages the real-time flow of LLM responses from 12+ providers through a unified pipeline. Each provider has dedicated stream wrappers that normalize vendor-specific SSE formats into a common chunk protocol, handle thinking/reasoning streams, manage cache retention headers, and provide graceful error recovery mid-stream.

12+

Provider Wrappers

6600+

Lines (run.ts)

~80k

Total Lines

Stream Formats

Real-Time Multi-Provider Backpressure-Aware Error-Recovery

📚 What is this & why does it exist?

The streaming pipeline handles the real-time delivery of AI responses. Instead of waiting for the entire response to generate, it delivers text token-by-token as the LLM produces it. Each provider (OpenAI, Anthropic, Google, etc.) sends data in a different format, so this module normalizes everything into a common protocol that the rest of OpenClaw can consume.

💡 Real-world analogy

Imagine a simultaneous translator at the United Nations. Delegates speak in different languages (providers), but listeners all hear through the same headset channel (unified stream). The translator (stream wrapper) converts each language in real-time, sentence-by-sentence, without waiting for the entire speech. If a speaker pauses or stumbles (backpressure/error), the translator smoothly manages the gap, perhaps repeating the last phrase or signaling a brief pause, rather than crashing the entire session.

Architecture

flowchart TD
  subgraph Entry["Stream Initiation"]
    RUN["run.ts\nrunEmbeddedPiAgent()"]
    ATTEMPT["attempt.ts\nTool Execution Loop"]
  end
  subgraph Wrappers["Provider Stream Wrappers"]
    OAI["openai-stream-wrappers.ts"]
    ANT["anthropic-cache-retention.ts"]
    GOO["google-stream-wrappers.ts"]
    BED["bedrock-stream-wrappers.ts"]
    MOON["moonshot-stream-wrappers.ts"]
    MINI["minimax-stream-wrappers.ts"]
    PROXY["proxy-stream-wrappers.ts"]
    ZAI["zai-stream-wrappers.ts"]
  end
  subgraph Processing["Stream Processing"]
    NORM["attempt.tool-call-normalization.ts"]
    REPAIR["attempt.tool-call-argument-repair.ts"]
    TRUNC["tool-result-truncation.ts"]
    GUARD["tool-result-context-guard.ts"]
  end
  subgraph Output["Delivery"]
    PAY["payloads.ts\nbuildEmbeddedRunPayloads()"]
    USAGE["usage-accumulator.ts"]
    REPORT["usage-reporting.ts"]
  end
  RUN --> Wrappers
  Wrappers --> ATTEMPT
  ATTEMPT --> Processing
  Processing --> PAY
  PAY --> USAGE
  USAGE --> REPORT

Data Flow

sequenceDiagram
    participant Runner as run.ts
    participant Wrapper as Stream Wrapper
    participant LLM as LLM Provider API
    participant Attempt as attempt.ts
    participant Payload as payloads.ts

    Runner->>Wrapper: initStream(model, messages)
    Wrapper->>LLM: HTTP SSE request
    loop Streaming chunks
        LLM-->>Wrapper: Provider-specific SSE event
        Wrapper-->>Runner: Normalized chunk
        Runner->>Attempt: Process chunk
        alt Tool call detected
            Attempt->>Attempt: Normalize tool call args
            Attempt->>Attempt: Repair malformed JSON
        end
        alt Thinking/reasoning content
            Attempt->>Attempt: Route to reasoning stream
        end
    end
    Attempt->>Payload: buildEmbeddedRunPayloads()
    Payload-->>Runner: Formatted payloads

Key Classes

Class / Function	File	Purpose
`runEmbeddedPiAgent()`	pi-embedded-runner/run.ts	Main streaming orchestrator; manages the full agent execution lifecycle
`buildEmbeddedRunPayloads()`	run/payloads.ts	Constructs delivery payloads from assistant texts and tool metadata
`attempt.ts` loop	run/attempt.ts	Core tool execution loop; processes streaming chunks, handles tool calls
`openai-stream-wrappers.ts`	pi-embedded-runner/	OpenAI-compatible SSE stream normalization
`google-stream-wrappers.ts`	pi-embedded-runner/	Google Gemini stream normalization with schema sanitization
`anthropic-cache-retention.ts`	pi-embedded-runner/	Anthropic prompt caching and cache-retention header management
`bedrock-stream-wrappers.ts`	pi-embedded-runner/	AWS Bedrock stream adapter with IAM auth integration
`moonshot-stream-wrappers.ts`	pi-embedded-runner/	Moonshot model stream adapter
`moonshot-thinking-stream-wrappers.ts`	pi-embedded-runner/	Moonshot thinking/reasoning stream handling
`minimax-stream-wrappers.ts`	pi-embedded-runner/	Minimax provider stream adapter
`proxy-stream-wrappers.ts`	pi-embedded-runner/	Generic proxy pass-through for custom endpoints
`zai-stream-wrappers.ts`	pi-embedded-runner/	Z AI provider stream adapter
`UsageAccumulator`	usage-accumulator.ts	Tracks input/output token counts across streaming chunks
`usage-reporting.ts`	pi-embedded-runner/	Formats and reports token usage after stream completion

Data Structures

Type	Name	Description
type	`EmbeddedPiRunResult`	Complete result from Pi agent run with payloads, usage, metadata, tool sends
type	`EmbeddedPiRunMeta`	Run metadata: duration, model used, token counts, errors, stop reason
type	`AgentStreamParams`	Stream parameters: temperature, maxTokens, fastMode
type	`RunEmbeddedPiAgentParams`	Full set of parameters for embedded run initiation
type	`SubscribeEmbeddedPiSessionParams`	Callbacks for tool results, reasoning, block replies during streaming
const	`BASE_CHUNK_RATIO`	Base ratio (0.4) for splitting message history during compaction
const	`MIN_CHUNK_RATIO`	Minimum ratio (0.15) for compaction chunk splitting
const	`SAFETY_MARGIN`	20% buffer (1.2) for token estimation inaccuracy

Interactions

Direction	Partner	Interface	Description
→	LLM Providers	HTTP SSE	Sends streaming requests to provider APIs via wrappers
→	Tool System	`attempt.ts` loop	Executes tool calls discovered in streaming chunks
→	Compaction	`compact.ts`	Triggers history summarization on context overflow
←	agent-command.ts	`runEmbeddedPiAgent()`	Receives command parameters and returns results
→	Auth System	`auth-profiles.ts`	Fetches API credentials for provider authentication
→	Session Manager	`SessionManager`	Manages conversation history via pi-coding-agent library
→	Context Engine	`context-engine-maintenance.ts`	Maintains context window boundaries
→	Delivery	`command/delivery.ts`	Routes completed payloads to target channels

Configuration

Knob / Parameter	Default	Description
`streamParams.temperature`	`undefined`	LLM temperature override for response randomness
`streamParams.maxTokens`	`undefined`	Maximum output token count per response
`streamParams.fastMode`	`false`	Enables fast-mode processing with reduced quality
`BASE_CHUNK_RATIO`	`0.4`	Base ratio for splitting messages during compaction
`MIN_CHUNK_RATIO`	`0.15`	Minimum chunk ratio floor for compaction
`SAFETY_MARGIN`	`1.2`	20% safety buffer for token count estimation
`compaction.reserveTokens`	`20000`	Token reserve floor for compaction trigger
`compaction.keepRecentTokens`	—	Recent tokens preserved during history compaction
`thinking`	`off`	Thinking/reasoning level: off, minimal, low, medium, high, xhigh, adaptive
`llmIdleTimeout`	—	Timeout for idle LLM connections during streaming
`STEER_ABORT_SETTLE_TIMEOUT_MS`	`5000`	Abort signal settle timeout for stream termination

Directory Map

Source tree for streaming

pi-embedded-runner/
├── run.ts                                    ← Main orchestrator (6600+ lines)
├── types.ts                                  ← EmbeddedPiRunResult, meta types
├── abort.ts                                  ← Stream abort handling
├── openai-stream-wrappers.ts                 ← OpenAI SSE adapter
├── google-stream-wrappers.ts                 ← Google/Gemini adapter
├── anthropic-cache-retention.ts              ← Anthropic cache headers
├── anthropic-family-tool-payload-compat.ts   ← Anthropic tool compat
├── bedrock-stream-wrappers.ts                ← AWS Bedrock adapter
├── moonshot-stream-wrappers.ts               ← Moonshot adapter
├── moonshot-thinking-stream-wrappers.ts      ← Moonshot thinking
├── minimax-stream-wrappers.ts                ← Minimax adapter
├── proxy-stream-wrappers.ts                  ← Generic proxy
├── zai-stream-wrappers.ts                    ← Z AI adapter
├── usage-accumulator.ts                      ← Token usage tracking
├── usage-reporting.ts                        ← Usage report formatting
├── thinking.ts                               ← Thinking level negotiation
├── session-truncation.ts                     ← Session history truncation
├── tool-result-truncation.ts                 ← Tool result size limiting
├── tool-result-context-guard.ts              ← Context overflow guards
├── compact.ts                                ← Adaptive compaction
├── compact-reasons.ts                        ← Compaction trigger reasons
├── compaction-hooks.ts                       ← Compaction hook system
├── compaction-runtime-context.ts             ← Runtime compaction context
├── compaction-safety-timeout.ts              ← Compaction timeout guards
└── run/
    ├── attempt.ts                            ← Tool execution loop (2300+ lines)
    ├── attempt.tool-call-normalization.ts    ← Tool call normalization
    ├── attempt.tool-call-argument-repair.ts  ← JSON argument repair
    ├── attempt.stop-reason-recovery.ts       ← Stop reason recovery
    ├── payloads.ts                           ← Payload construction
    ├── setup.ts                              ← Model resolution
    └── helpers.ts                            ← Runtime utilities

Deep-Dive Highlights

Core streaming mechanisms have been extracted into dedicated focus pages for in-depth exploration. Each page covers one aspect of the streaming pipeline with diagrams, code walks, and configuration details.

Provider Stream Formats

Provider	Wrapper File	Stream Format	Special Handling
OpenAI	`openai-stream-wrappers.ts`	SSE with `data: {choices}`	Standard delta chunking
Anthropic	`anthropic-cache-retention.ts`	SSE with `content_block_delta`	Cache-retention headers, prompt caching
Google/Gemini	`google-stream-wrappers.ts`	Google-specific event stream	Schema sanitization via `cleanForGemini()`
AWS Bedrock	`bedrock-stream-wrappers.ts`	AWS event stream binary framing	IAM authentication, region routing
Moonshot	`moonshot-stream-wrappers.ts`	OpenAI-compatible SSE	Dedicated thinking stream wrapper
Minimax	`minimax-stream-wrappers.ts`	Provider-specific events	Custom token counting
OpenRouter	Via `openai-stream-wrappers.ts`	OpenAI-compatible SSE	Capability flags from `openrouter-model-capabilities.ts`
Z AI	`zai-stream-wrappers.ts`	Provider-specific events	Custom normalization
Proxy	`proxy-stream-wrappers.ts`	Pass-through	Generic wrapper for custom endpoints

Error Handling

The streaming pipeline handles errors at multiple levels:

Error Type	Detection	Recovery Action
Context Overflow	`estimateTokens() > context_window - reserve`	Trigger adaptive compaction
Rate Limit	HTTP 429 or provider-specific signal	Cascade to fallback model
Auth Failure	HTTP 401/403	Mark auth profile in cooldown, try next
Stream Timeout	`llm-idle-timeout.ts` timer expiry	Abort stream, retry with next provider
Malformed Tool Args	JSON parse failure	Repair via `attempt.tool-call-argument-repair.ts`
Unexpected Stop	Stop reason mismatch	Recovery via `attempt.stop-reason-recovery.ts`
Provider Error	HTTP 5xx or connection reset	Cascade to fallback model
Abort Signal	External cancellation	Graceful shutdown via `abort.ts` (5s settle)

Thinking & Reasoning Levels

The streaming pipeline supports extended thinking/reasoning modes that route cognitive process output to a separate stream:

Level	Description	Token Impact
`off`	No thinking output	None
`minimal`	Brief reasoning hints	~5% overhead
`low`	Key decision points only	~10% overhead
`medium`	Moderate reasoning detail	~20% overhead
`high`	Full reasoning chain	~40% overhead
`xhigh`	Maximum thinking depth	~60% overhead
`adaptive`	Auto-adjust based on complexity	Variable

Code Examples

Example: Configuring Stream Parameters

const streamParams: AgentStreamParams = {
  temperature: 0.7,
  maxTokens: 4096,
  fastMode: false,
};

const opts: AgentCommandOpts = {
  message: "Explain the streaming architecture",
  provider: "anthropic",
  model: "claude-3-5-sonnet",
  streamParams,
  thinking: "medium",
  timeout: "120s",
};

Example: Compaction Token Strategy

// Adaptive compaction splits message history by token share
const BASE_CHUNK_RATIO = 0.4;
const MIN_CHUNK_RATIO = 0.15;
const SAFETY_MARGIN = 1.2; // 20% buffer for estimateTokens() inaccuracy

function splitMessagesByTokenShare(
  messages: AgentMessage[],
  ratio: number,
  contextWindow: number,
): AgentMessage[][] {
  // Splits message history by token ratio for adaptive compaction
  // First chunk gets BASE_CHUNK_RATIO of context window
  // Each subsequent chunk gets at least MIN_CHUNK_RATIO
  // SAFETY_MARGIN accounts for token estimation inaccuracy
}

Example: EmbeddedPiRunResult Structure

// The complete result from a streaming agent run
type EmbeddedPiRunResult = {
  payloads: Array<{
    text?: string;
    mediaUrl?: string;
    mediaUrls?: string[];
    replyToId?: string;
    isError?: boolean;
    isReasoning?: boolean;
  }>;
  usage: {
    inputTokens: number;
    outputTokens: number;
    cacheCreationTokens?: number;
    cacheReadTokens?: number;
  };
  meta: EmbeddedPiRunMeta;
  toolSends: ToolSendEntry[];
  compacted: boolean;
  pendingTools: PendingToolEntry[];
};