Skip to main content

Spring AI Streaming Source Code Analysis

Streaming is the de facto interaction mode for modern LLM-powered applications. Rather than waiting for the entire response to be generated, streaming delivers tokens to the client as they are produced, enabling real-time typing indicators, lower perceived latency, and the ability to abort long generations. Spring AI integrates streaming deeply into its architecture using Project Reactor, making it a first-class citizen alongside synchronous call().

This chapter examines the source code that powers streaming in Spring AI. We will trace a streaming request from the ChatClient fluent API down through the reactive ChatModel abstraction, provider-specific adapters, and the Flux<ChatResponse> pipeline. By the end, you will understand how the framework achieves non-blocking, backpressure-aware token delivery while remaining provider-agnostic.

What Is Spring AI Streaming?​

Spring AI Streaming refers to the capability of requesting and consuming a large language model response as a stream of partial ChatResponse objects, each typically containing a single token or a small group of tokens. This contrasts with the synchronous call() method, which returns the complete response only after the entire generation is finished.

Key design goals:

  • Incremental delivery – Emit each piece of the response as soon as the model produces it.
  • Reactive programming – Leverage Project Reactor's Flux to provide a standard reactive-streams API.
  • Non-blocking I/O – Keep server threads free while waiting for model output.
  • Unified API – The streaming contract is defined at the ChatModel level, so all providers can implement it uniformly.

Streaming dramatically improves user experience and enables scenarios like server‑sent events (SSE), real‑time dashboards, and early cancellation of expensive generations.

Streaming in Spring AI Architecture​

Streaming is not bolted onto the framework; it is a parallel execution path that follows the same layered architecture as synchronous calls but uses reactive types.

  • Application – Calls chatClient.prompt().stream().chatResponse() to obtain a Flux<ChatResponse>.
  • ChatClient – Orchestrator; it builds the prompt, applies advisors, and then delegates to the StreamingChatModel.
  • Prompt – Immutable request object, identical to the one used in synchronous calls.
  • Advisor Chain – Pre‑processes the prompt; in streaming, response advisors are applied per‑emitted item or at the end.
  • StreamingChatModel – An extension of ChatModel that declares Flux<ChatResponse> stream(Prompt prompt).
  • Provider Adapter – Translates the prompt into a streaming HTTP request (using SSE or WebSocket) and maps each chunk to a ChatResponse.
  • Flux – The reactive stream that the application subscribes to; it respects backpressure and supports cancellation.

The architecture ensures that the same advisors and tooling work for both streaming and non‑streaming paths, with minimal branching.

Core Interfaces and Classes​

Class / InterfaceResponsibility
ChatClientTop‑level entry point; its response spec offers stream() to initiate streaming.
StreamingChatModelInterface extending ChatModel with Flux<ChatResponse> stream(Prompt prompt).
DefaultChatClientImplementation of ChatClient; internally decides whether to call call() or stream().
ChatResponseNormalized response object; in streaming, each chunk is a ChatResponse containing a delta.
Flux<ChatResponse>Project Reactor type representing the stream of partial responses.
PromptThe request object; identical for synchronous and streaming.
AdvisorRequestResponseAdvisor interface; applied around both synchronous and streaming calls.
OpenAiChatModelExample provider adapter; implements StreamingChatModel for OpenAI's streaming API.
StreamingChatClientResponseSpecThe internal spec that returns Flux<ChatResponse> from the stream() call.

These components collaborate to deliver a consistent reactive programming model to the developer.

Source Code Structure​

Streaming-related source files are spread across the core and provider modules:

org.springframework.ai.chat.client
├── ChatClient.java                   // defines stream() in response spec
├── DefaultChatClient.java            // implements streaming delegation

org.springframework.ai.chat.model
├── StreamingChatModel.java           // the reactive extension interface
├── ChatModel.java                    // base model interface (call())

org.springframework.ai.openai
├── OpenAiChatModel.java              // implements StreamingChatModel for OpenAI
├── OpenAiStreamingChatModel.java     // (older variant) may still exist

org.springframework.ai.azure.openai
├── AzureOpenAiChatModel.java         // streaming support via Azure

org.springframework.ai.ollama
├── OllamaChatModel.java              // streaming support for Ollama

The key separation: the core defines the streaming contract (StreamingChatModel), and each provider module implements it. The ChatClient remains unaware of the provider specifics, interacting only with the interface.

Streaming Execution Lifecycle​

The following sequence diagram shows a complete streaming request from the application through to the LLM provider and back.

  1. Request creation – The application calls chatClient.prompt().user(...).stream().chatResponse(). The ChatClient builds the prompt and prepares the advisor chain.
  2. Advisor pre-processing – Advisors modify the prompt just as in a synchronous call.
  3. Streaming invocation – The ChatClient calls streamingChatModel.stream(prompt). The provider adapter constructs a streaming HTTP request (e.g., with "stream": true).
  4. Token emission – The provider’s API sends server‑sent events or chunked responses. The adapter parses each chunk into a ChatResponse containing a content delta and metadata (e.g., tool calls, finish reason).
  5. Reactive propagation – The Flux<ChatResponse> flows back through the framework. Advisors that need to process the response can apply per‑item transformations or attach a doFinally callback.
  6. Client consumption – The application subscribes to the Flux and processes each partial response, often forwarding it to a UI via SSE or WebSockets.
  7. Stream completion – The provider signals the end of the stream, and the Flux completes.

ChatClient Streaming API​

The streaming API is exposed through the ChatClient response spec. The key method is:

ChatClientResponseSpec stream();

Calling stream() returns a StreamingChatClientResponseSpec that offers:

  • Flux<ChatResponse> chatResponse() – raw model response stream.
  • Flux<String> content() – convenience method that extracts the text content from each ChatResponse.

Internally, DefaultChatClient constructs a DefaultStreamingChatClientResponseSpec which holds a reference to the model and the prompt. When chatResponse() is called, it executes the advisor chain and then invokes chatModel.stream(prompt).

The fluent API is designed so that the request spec is identical for both synchronous and streaming paths. This unified model allows the developer to switch between call() and stream() without changing the prompt construction.

ChatModel Streaming Integration​

The StreamingChatModel interface is the backbone of the reactive path. It is defined as:

public interface StreamingChatModel extends ChatModel {
    Flux<ChatResponse> stream(Prompt prompt);
}

All major provider adapters implement this interface:

  • OpenAiChatModel – Opens a streaming connection to the OpenAI /v1/chat/completions endpoint with "stream": true. It uses a reactive WebClient to read SSE events, parse them into ChatResponse deltas, and emit them on a Flux.
  • AzureOpenAiChatModel – Similar to OpenAI but connects to Azure’s streaming endpoint, handling authentication and deployment‑specific URLs.
  • OllamaChatModel – Uses Ollama’s local streaming API; the adapter returns a Flux from the response’s line‑delimited JSON stream.
  • DashScopeChatModel (in Spring AI Alibaba) – Implements DashScope’s streaming SSE protocol.

Each adapter encapsulates the transport and parsing logic, so the rest of the framework sees only a uniform Flux<ChatResponse>.

Reactor Flux Integration​

Spring AI builds its streaming on Project Reactor, the default reactive library in the Spring ecosystem. Key aspects:

  • Flux<ChatResponse> – The type returned by stream(). It is a cold publisher: nothing happens until the application subscribes.
  • Backpressure – The framework respects reactive backpressure. If the consumer is slow, the adapter buffers minimally (typically one chunk) and waits for demand before reading more from the network.
  • Cancellation – When the application unsubscribes or cancels the Flux, the adapter closes the underlying HTTP connection, saving tokens and resources.
  • Error handling – If the streaming connection breaks or the provider returns an error, the Flux signals an onError with a standard Spring AI exception (AiClientException).
  • Completion signals – When the stream ends, the Flux completes with onComplete(). Advisors and the ChatClient can hook into this signal for cleanup.

Reactors Flux provides a rich set of operators that applications can use to transform, filter, or merge streams without blocking threads.

Streaming Response Assembly​

In streaming mode, each emitted ChatResponse represents a partial update, not a full answer. The assembly logic differs from the synchronous path:

  • Content delta – The ChatResponse contains a Generation whose output is a delta, not the final accumulated text. The framework does not automatically concatenate deltas; that is the responsibility of the consumer (or a higher-level utility).
  • Metadata updates – Later chunks may include updated finish reasons or tool call details. The final chunk typically has finishReason = STOP and the aggregated token usage.
  • Tool calls – For streaming with tool calling, the model may emit tool call deltas across multiple chunks. The provider adapter assembles these deltas into a complete ToolCall before emitting a consolidated ChatResponse. Spring AI handles this internally for each provider.

The ChatClient provides a convenience method content() that aggregates the text, offering a simpler API for common use cases.

Advisor Integration​

Advisors are fully compatible with streaming, but their behaviour must be re‑entrant and non‑blocking.

  • Request advisement – adviseRequest() is called synchronously before the stream starts, exactly as in the non‑streaming case.
  • Response advisement – adviseResponse(ChatResponse, Map) is called per emitted item if the advisor is applied as a StreamingResponseAdvisor or if the framework wraps it. Otherwise, it may be called only on the final aggregated response (in the doFinally block). The exact behaviour depends on how the advisor is configured.
  • Streaming‑aware advisors – An advisor can implement StreamingAdvisor (sub‑interface) to explicitly handle per‑item callbacks, or it can use a doOnEach pattern via the Flux.

For example, a logging advisor might log each token as it arrives. A content‑filter advisor might drop tokens that contain sensitive data. The framework’s advisor chain is flexible enough to accommodate both use cases.

Tool Calling and Streaming​

Tool calling during streaming introduces additional complexity because tool call requests are often assembled incrementally. Key points:

  • Delta assembly – The provider adapter buffers tool call deltas across chunks, assembling them into complete ToolCall objects before emitting a ChatResponse that contains the tool call.
  • Execution pause – When a tool call is detected, the stream typically pauses (the model stops emitting tokens), and the tool is executed. The tool result is then injected back into the conversation, and a new stream may be started for the subsequent model call.
  • Compatibility – Not all providers support tool calling in streaming mode identically. Spring AI’s adapters handle these differences, presenting a uniform experience.
  • Current limitations – Some providers may require special handling (e.g., parallel_tool_calls enabled). The framework adapters are continuously updated to keep pace with provider APIs.

The ChatClient and agent runtime manage tool‑calling loops transparently; streaming consumers receive the final answer chunks after all tool calls have resolved.

Design Patterns Used​

  • Strategy Pattern – StreamingChatModel is the strategy interface; each provider adapter is a concrete strategy.
  • Adapter Pattern – Provider adapters translate the uniform stream() call into provider‑specific protocols and back.
  • Publisher‑Subscriber (Reactive Streams) – The Flux<ChatResponse> implements the publisher‑subscriber pattern, decoupling data production from consumption.
  • Builder Pattern – The ChatClient uses a builder for configuration and a fluent request/response spec for execution.
  • Decorator Pattern – The advisor chain can be seen as a decorator around the ChatModel, adding behavior to the streaming call.
  • Dependency Injection – The ChatModel and advisors are injected, allowing easy composition and testing.

Extension Points​

Developers can extend streaming in several ways:

  • Custom StreamingChatModel – Implement the interface for a new provider. The adapter should return a Flux that emits ChatResponse objects.
  • Custom advisor – Create an advisor that operates on each stream element (by wrapping the Flux with operators) or that acts as a StreamingAdvisor.
  • Response transformation – Use Reactor operators (map, filter, buffer) in the application layer or inside a custom advisor to modify the stream.
  • Streaming aggregation – Provide a custom StreamingResponseExtractor (if the framework exposes such an SPI) to control how deltas are combined.

All extensions leverage standard Spring and Reactor mechanisms.

Enterprise Best Practices​

  • Streaming APIs – Expose streaming endpoints using Spring WebFlux and application/x-ndjson or SSE. Forward the Flux directly to the HTTP response.
  • Server‑Sent Events (SSE) – Use spring-webflux to return a Flux<ServerSentEvent<String>> that maps each token to an event.
  • WebFlux integration – Leverage reactive controllers (@RestController returning Flux) to avoid blocking threads.
  • Scalability – Reactive streams use fewer threads; the framework can handle thousands of concurrent streaming connections with a small thread pool.
  • Timeout handling – Set network timeouts on the provider adapter to avoid hung connections. Reactor’s timeout() operator can be applied to the Flux.
  • Error recovery – Use onErrorResume() or retry() operators in the application layer to gracefully handle transient failures.
  • Observability – Each streaming chunk can be traced individually; Spring AI’s Micrometer integration records token counts and latency for streaming calls.

Performance Considerations​

  • Latency – Streaming reduces time‑to‑first‑token, but network round‑trips remain the dominant factor.
  • Throughput – Reactive non‑blocking I/O allows a single application instance to manage many concurrent streams without thread exhaustion.
  • Memory usage – Streaming avoids buffering the entire response in memory, reducing heap pressure.
  • Reactive scheduling – The default Reactor scheduler uses a small number of worker threads; for CPU‑heavy token processing, consider offloading to a dedicated scheduler.
  • Backpressure – The framework respects demand; slow consumers will cause the provider adapter to pause reading from the socket, preventing memory overruns.
  • High‑concurrency scenarios – Use connection pooling and HTTP client reuse; each provider adapter typically uses a shared WebClient or RestClient instance.

Source Code Reading Guide​

To understand the streaming internals, follow this order:

  1. StreamingChatModel.java – The core reactive interface.
  2. DefaultChatClient.java – Look for the stream() method and the inner class DefaultStreamingChatClientResponseSpec.
  3. OpenAiChatModel.java – Trace the stream() implementation; see how it constructs the request and parses SSE events.
  4. OllamaChatModel.java – A simpler streaming implementation; good for understanding the pattern.
  5. AzureOpenAiChatModel.java – Similar to OpenAI, with Azure-specific authentication.
  6. ChatClientResponseSpec and StreamingChatClientResponseSpec – Interfaces that define the streaming return types.
  7. Test classes – OpenAiChatModelTests and ChatClientStreamingTests show end‑to‑end streaming scenarios.

Summary​

Spring AI’s streaming support is a first‑class citizen built on Project Reactor and the StreamingChatModel interface. The framework provides a unified programming model across all providers, encapsulating the complexity of SSE, chunk parsing, and backpressure. Understanding the source code reveals how the ChatClient, provider adapters, and the reactive Flux<ChatResponse> pipeline collaborate to deliver low‑latency, resource‑efficient token streaming. This knowledge is essential for building high‑performance, real‑time AI applications and for extending the framework with custom providers or streaming behaviors.

Proceed to ChatModel Source Code Analysis for a deeper understanding of the model abstraction that underpins both synchronous and streaming execution.