Streaming

Agent streams token-by-token from LangGraph agents over Server-Sent Events (SSE). Every update lands directly in Angular Signals — no subscriptions, no manual change detection.

What the demo does

The Run tab hosts one agent and the prebuilt <chat> composition. Send a message and the answer fills in token by token while the typing indicator runs. The composition owns message rendering, the input, and error display, so the demo component is a template plus three fields.

Two welcome suggestions are wired up. "Stream a long answer" asks for a 200-word explanation of LangGraph checkpointing, which is long enough to watch the tokens land one at a time. "How agents pick tools" asks the agent to explain tool selection, and the chat-tool-calls component guide covers how those calls are rendered.

How it is built

Three files carry the visible integration: the graph that streams, the provider that points Angular at it, and the component that renders it. A fourth, small file declares the typed agent ref they share. Open the Code tab to read the three in place.

The streaming graph

The backend is a single node. MessagesState gives the LangGraph SDK a message list it already understands, the model is constructed with streaming=True, and the node prepends a system prompt read from the capability's prompt file before it awaits the model. The compiled graph is exported as graph, which is the symbol langgraph.json points at.

"""
LangGraph Streaming Graph
 
A minimal StateGraph that demonstrates real-time token streaming from an LLM.
Uses LangGraph's MessagesState for compatibility with the LangGraph SDK client.
 
The graph uses LangSmith for observability — every invocation is traced
automatically when LANGCHAIN_TRACING_V2=true is set.
"""
 
from pathlib import Path
from langgraph.graph import StateGraph, MessagesState, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage
 
PROMPTS_DIR = Path(__file__).parent.parent / "prompts"
 
 
def build_streaming_graph():
    """
    Constructs the LangGraph StateGraph for streaming.
 
    The graph has a single node that calls the LLM with the system prompt
    and user message. Uses MessagesState so the LangGraph SDK can send
    and receive messages directly.
 
    Returns:
        A compiled StateGraph ready for invocation
    """
    llm = ChatOpenAI(model="gpt-5-mini", streaming=True)
 
    async def generate(state: MessagesState) -> dict:
        """
        Generate a streaming response from the LLM.
 
        Reads the system prompt and prepends it to the conversation,
        then invokes the LLM with streaming enabled.
        """
        system_prompt = (PROMPTS_DIR / "streaming.md").read_text()
        messages = [SystemMessage(content=system_prompt)] + state["messages"]
        response = await llm.ainvoke(messages)
        return {"messages": [response]}
 
    graph = StateGraph(MessagesState)
    graph.add_node("generate", generate)
    graph.set_entry_point("generate")
    graph.add_edge("generate", END)
 
    return graph.compile()
 
 
# The graph instance — referenced by langgraph.json
graph = build_streaming_graph()

The node awaits a single ainvoke call. streaming=True is what lets the model emit tokens through LangGraph's callbacks, and which of those reach the browser is decided by the stream modes the client asks for.

The agent provider

provideAgent() registers the agent once for the whole application, keyed by the typed ref declared in agent-ref.ts. The ref itself is one line, typed by a state interface declared beside it:

import { createAgentRef } from '@threadplane/chat';
 
export interface StreamingState {
  messages: unknown[];
}
 
export const STREAMING_AGENT = createAgentRef<StreamingState>('streaming');

The example resolves its connection details at runtime from the host that serves the demo, which is why its factory reads them rather than hard-coding them. It is the only provider <chat> requires.

import { injectCockpitRuntimeConnection } from '@threadplane/cockpit-telemetry';
import { ApplicationConfig } from '@angular/core';
import { provideAgent } from '@threadplane/langgraph';
import { STREAMING_AGENT } from './agent-ref';
 
export const appConfig: ApplicationConfig = {
  providers: [
    provideAgent(STREAMING_AGENT, () => {
      const connection = injectCockpitRuntimeConnection();
      if (connection.adapter !== 'langgraph') {
        throw new Error('incompatible runtime');
      }
      return {
        apiUrl: connection.apiUrl,
        assistantId: connection.assistantId,
        clientOptions: connection.clientOptions,
      };
    }),
  ],
};

Your own application does not need the factory. Pass the two values directly:

provideAgent(STREAMING_AGENT, {
  apiUrl: 'https://your-deployment.langgraph.app',
  assistantId: 'streaming',
}),

assistantId must match the graph name in langgraph.json.

The chat component

injectAgent(STREAMING_AGENT) returns the agent registered above, typed by that ref. The component hands it to <chat> and does little else: it renders the two welcome suggestions and forwards the selected one into submit().

import { Component } from '@angular/core';
import { ChatComponent, ChatWelcomeSuggestionComponent } from '@threadplane/chat';
import { injectAgent } from '@threadplane/langgraph';
import { STREAMING_AGENT, type StreamingState } from './agent-ref';
import { ExampleChatLayoutComponent } from '@threadplane/example-layouts';
 
const WELCOME_SUGGESTIONS = [
  {
    label: 'Stream a long answer',
    value: 'Explain LangGraph checkpointing in 200 words.',
    description: 'Watch tokens arrive incrementally — the simplest @threadplane/chat integration.',
  },
  {
    label: 'How agents pick tools',
    value: 'Show me how an agent decides which tool to use.',
    description: 'Explains tool-selection reasoning; good entry point for the tool-calls demo.',
  },
] as const;
 
/**
 * Streaming demo — simplest possible @threadplane/chat integration.
 *
 * Injects the singleton agent (configured in app.config.ts) and passes it
 * to the prebuilt <chat> composition. The composition handles message
 * rendering, input, typing indicator, and error display internally.
 */
@Component({
  selector: 'app-streaming',
  standalone: true,
  imports: [ChatComponent, ChatWelcomeSuggestionComponent, ExampleChatLayoutComponent],
  template: `
    <example-chat-layout>
      <chat main [agent]="agent" class="flex-1 min-w-0">
        <div chatWelcomeSuggestions>
          @for (s of suggestions; track s.value) {
            <chat-welcome-suggestion
              [label]="s.label"
              [value]="s.value"
              [description]="s.description"
              (selected)="send($event)"
            />
          }
        </div>
      </chat>
    </example-chat-layout>
  `,
})
export class StreamingComponent {
  protected readonly agent = injectAgent(STREAMING_AGENT);
  // Typed read: prove StreamingState flows through DI.
  protected readonly _typedState: StreamingState = this.agent.value();
  protected readonly suggestions = WELCOME_SUGGESTIONS;
 
  protected send(text: string): void {
    void this.agent.submit({ message: text });
  }
}

submit({ message }) opens the stream, and every signal the composition reads updates as chunks arrive. There is no service layer to write — the agent owns the connection lifecycle, the state, and error recovery.

Stream status

The status() signal reports the current lifecycle state of the SSE connection:

Stream modes

On the server, astream() decides what a run emits, and astream_events() adds raw run events. The three modes you will meet most often:

async def main():
    # "values" — full state snapshot after each node
    async for chunk in graph.astream({"messages": [("user", "Hello")]}, stream_mode="values"):
        print(chunk)
 
    # "messages" — individual message tokens as they are generated
    async for chunk in graph.astream({"messages": [("user", "Hello")]}, stream_mode="messages"):
        print(chunk)
 
    # "events" — raw run events (on_chain_start, on_llm_stream, etc.)
    async for event in graph.astream_events({"messages": [("user", "Hello")]}, version="v2"):
        print(event["event"], event.get("data"))

The client asks for messages-tuple to receive the token tuples that stream_mode="messages" produces on the server; messages is a separate client mode, so the lists below name messages-tuple.

You rarely call those directly from an Angular app. By default, injectAgent() asks LangGraph Platform for the stream modes it needs to populate its public signals: values, messages-tuple, updates, and custom. It also enables streamSubgraphs so namespaced subgraph events can reach the client.

Override streamMode per run when you need a narrower stream. It is a submit option, not a provideAgent() option.

// Receives the full agent state after every node execution.
// Best when you only need state snapshots.
const chat = injectAgent(STREAMING_AGENT);
 
await chat.submit(
  { message: 'Summarize this thread.' },
  { streamMode: ['values'] },
);
 
// chat.messages() always contains the complete message list

Error handling

If the SSE connection drops or the agent throws, status() flips to 'error' and error() is populated. The prebuilt <chat> composition renders the failure for you; a hand-built UI reads the same two signals.

import { computed } from '@angular/core';
import { injectAgent } from '@threadplane/langgraph';
import { STREAMING_AGENT } from './agent-ref';
 
export class ChatComponent {
  protected readonly chat = injectAgent(STREAMING_AGENT);
 
  readonly hasError = computed(() => this.chat.status() === 'error');
 
  // An interrupted run is retryable only when nothing reached the server.
  readonly canCheck = computed(
    () => this.chat.error()?.recovery === 'check' && this.chat.checkStatus !== undefined,
  );
 
  retry() {
    // Clears the error and re-submits the last input on the same thread
    this.chat.retry();
  }
 
  checkStatus() {
    // Read-only: asks the backend what happened, resubmits nothing
    this.chat.checkStatus?.();
  }
}

retry() clears the error, then re-runs the last submission on the same thread; it is a no-op while a run is already in flight or when there is nothing to retry. reload() re-runs the last submission the same way but leaves the error signal alone, which is useful when you want to keep showing the failure state while the retry attempt is in progress. submit(null) opens a stream against the current thread state without a new user message and resumes only pending work, such as an interrupted run or a failed run that left tasks pending; otherwise it is a no-op. Pass submit({ message }) only when you have new input to send.

Throttle configuration

By default Agent coalesces state-like signal updates every 16 ms. That is close to a 60 fps render cadence and prevents fast SSE streams from triggering hundreds of state renders per second.

// app.config.ts
provideAgent(STREAMING_AGENT, {
  apiUrl: '...',
  assistantId: 'streaming',
  // Batch incoming chunks and flush at most once every 50 ms
  throttle: 50,
});

The value is in milliseconds. Pass false or 0 to disable batching and forward state updates immediately. Token message updates are not throttled, so live markdown and typing indicators still receive every token emission.

For me, the 16 ms default is the right starting point — it tracks a 60 fps render and you will rarely feel it. The tradeoff of raising it is straightforward: a larger window costs you a touch of perceived latency on state signals to save renders, so only reach for it when profiling shows state updates are the bottleneck.

What's Next