Streaming & Real-time Agent Responses

Why Streaming Changes the Agent UX

Non-streaming agent response flow:

flowchart TD
    U([User submits task]) --> S["15 seconds of silence..."]
    S --> R([Full response appears])
    style S fill:#fff0f0,stroke:#f87171,color:#0F172A

Streaming response flow:

flowchart TD
    U([User submits task]) --> T1["Searching the web for...\n< 500ms"]
    T1 --> T2["Found 5 relevant sources...\n< 2s"]
    T2 --> T3["Based on the research...\n< 4s"]
    T3 --> T4(["Tokens streaming → complete\n< 15s"])
    style T1 fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style T2 fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style T3 fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style T4 fill:#f0fdf9,stroke:#0D9488,color:#0F172A

The total time is the same, but the perceived responsiveness is dramatically different.

Streaming OpenAI Tokens

# src/streaming/openai_stream.py
import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

# async generator: the `yield` keyword makes this a generator;
# `async for` in the caller receives each event as it arrives without blocking
async def stream_openai_tokens(messages: list, model: str = "gpt-5.6-sol"):
    
    stream = await client.chat.completions.create(
        model=model,
        messages=messages,
        stream=True,  # stream=True switches from single response to chunked delivery
    )
    
    # Each chunk contains a delta: the incremental addition since the last chunk
    async for chunk in stream:
        delta = chunk.choices[0].delta
        
        # yield chunk.choices[0].delta.content: sends each token to the caller as it arrives
        if delta.content:
            yield {"type": "token", "content": delta.content}
        
        # Tool calls also arrive incrementally: name first, then arguments piece by piece
        elif delta.tool_calls:
            for tc in delta.tool_calls:
                if tc.function.name:
                    yield {"type": "tool_call_start", "name": tc.function.name}
                if tc.function.arguments:
                    # arguments arrive as fragments of a JSON string: accumulate before parsing
                    yield {"type": "tool_call_args", "args": tc.function.arguments}
        
        # finish_reason signals why generation stopped: "stop" = natural end, "tool_calls" = needs tools
        if chunk.choices[0].finish_reason == "stop":
            yield {"type": "done"}
        elif chunk.choices[0].finish_reason == "tool_calls":
            yield {"type": "executing_tools"}

# Usage
async def main():
    async for event in stream_openai_tokens([
        {"role": "user", "content": "Explain how async generators work in Python"}
    ]):
        if event["type"] == "token":
            print(event["content"], end="", flush=True)  # end="" prevents newline between tokens
        elif event["type"] == "done":
            print("\n[Complete]")

Streaming LangGraph Execution

LangGraph’s streaming API gives you fine-grained control over what gets surfaced:

# src/streaming/langgraph_stream.py
from langgraph.graph import StateGraph

async def stream_agent_execution(agent, initial_state: dict, config: dict):
    """Stream all execution events from a LangGraph agent."""
    
    async for event in agent.astream_events(
        initial_state,
        config=config,
        # version="v2" is required to get the new event format with on_chat_model_stream
        # v1 is the legacy format and does not emit per-token streaming events
        version="v2",
    ):
        event_type = event["event"]
        
        # on_chain_start fires for every runnable: filter to meaningful node names only
        if event_type == "on_chain_start" and event.get("name") not in ["LangGraph", "RunnableSequence"]:
            yield {
                "type": "node_start",
                "node": event["name"],
                "run_id": event["run_id"],  # run_id lets the frontend match start/end events
            }
        
        # on_chat_model_stream fires once per token: this is where the live typing effect comes from
        elif event_type == "on_chat_model_stream":
            chunk = event["data"]["chunk"]
            if chunk.content:
                yield {"type": "token", "content": chunk.content}
        
        # on_tool_start fires when a tool call is dispatched: use this to show "Searching..."
        elif event_type == "on_tool_start":
            yield {
                "type": "tool_start",
                "tool": event["name"],
                "input": event["data"].get("input", {}),  # the arguments the model passed to the tool
            }
        
        # on_tool_end fires when the tool returns: truncate the preview to avoid flooding the UI
        elif event_type == "on_tool_end":
            yield {
                "type": "tool_end",
                "tool": event["name"],
                "output_preview": str(event["data"].get("output", ""))[:100],
            }
        
        # on_chain_end with name="LangGraph" signals the entire graph run is complete
        elif event_type == "on_chain_end" and event.get("name") == "LangGraph":
            yield {"type": "complete"}

FastAPI SSE Endpoint

# src/api/agent_stream.py
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from sse_starlette.sse import EventSourceResponse
import json

app = FastAPI()

@app.post("/agent/stream")
async def agent_stream_endpoint(request: Request):
    body = await request.json()
    user_query = body["query"]
    # Use client-provided thread_id for conversation continuity, or generate a new one
    thread_id = body.get("thread_id", str(uuid.uuid4()))
    
    config = {"configurable": {"thread_id": thread_id}}
    initial_state = {
        "messages": [HumanMessage(content=user_query)],
        "iterations": 0,
        "final_answer": None,
    }
    
    # event_generator is an async generator: FastAPI's StreamingResponse pulls from it lazily
    async def event_generator():
        
        try:
            async for event in stream_agent_execution(agent, initial_state, config):
                # SSE wire format: "data: <json>\n\n": double newline signals end of one event
                data = json.dumps(event)
                yield f"data: {data}\n\n"
                
                if event["type"] == "complete":
                    break
        
        except Exception as e:
            # Stream errors back to the client so the UI can display them rather than hanging
            error_event = json.dumps({"type": "error", "message": str(e)})
            yield f"data: {error_event}\n\n"
        
        finally:
            # stream_end always fires: lets the client close the connection cleanly
            yield "data: {\"type\": \"stream_end\"}\n\n"
    
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",  # tells the browser this is an SSE stream
        headers={
            "Cache-Control": "no-cache",   # prevent proxies from buffering the stream
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",     # disable Nginx buffering, critical for real-time delivery
        }
    )

Browser-side SSE Consumer (JavaScript)

// Frontend: consume the SSE stream using the Fetch API (works in all modern browsers)
async function runAgent(query) {
  const outputEl = document.getElementById('output');
  const statusEl = document.getElementById('status');
  
  // Use fetch with stream:true, avoids the 2KB buffer delay of EventSource
  const response = await fetch('/agent/stream', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ query }),
  });
  
  // getReader() gives low-level access to the response body as it arrives byte by byte
  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;  // server closed the connection
    
    // A single read() may contain multiple SSE events, split by newline to process each
    const lines = decoder.decode(value).split('\n');
    for (const line of lines) {
      if (!line.startsWith('data: ')) continue;  // skip blank lines and non-data lines
      const event = JSON.parse(line.slice(6));    // slice(6) removes the "data: " prefix
      
      switch (event.type) {
        case 'node_start':
          // Show which graph node is currently executing
          statusEl.textContent = `Running: ${event.node}`;
          break;
        case 'token':
          // Append each token as it arrives: creates the live typing effect
          outputEl.textContent += event.content;
          break;
        case 'tool_start':
          statusEl.textContent = `🔧 Calling ${event.tool}...`;
          break;
        case 'tool_end':
          statusEl.textContent = `✅ ${event.tool} completed`;
          break;
        case 'complete':
          statusEl.textContent = 'Done';
          break;
        case 'error':
          statusEl.textContent = `Error: ${event.message}`;
          break;
      }
    }
  }
}

Handling Backpressure

For slow clients, use a queue to prevent the generator from running too far ahead:

import asyncio
from asyncio import Queue

async def agent_stream_with_backpressure(agent, initial_state, config):
    # asyncio.Queue(maxsize=50) provides backpressure: prevents the producer from flooding memory
    # if the consumer (slow client) can't keep up, queue.put() will block the producer
    queue = Queue(maxsize=50)
    
    async def producer():
        async for event in stream_agent_execution(agent, initial_state, config):
            await queue.put(event)  # awaits if queue is full: this is the backpressure mechanism
        await queue.put(None)  # sentinel value signals the consumer that production is done
    
    # create_task() runs the producer concurrently while the consumer yields events below
    producer_task = asyncio.create_task(producer())
    
    while True:
        event = await queue.get()  # waits for the next event without busy-looping
        if event is None:          # sentinel received, producer is done
            break
        yield event
    
    await producer_task  # ensure producer cleanup before returning

Streaming is the final piece that turns a backend agent system into a real product. Users tolerate slow processes when they can see progress, they don’t tolerate spinning loaders.