Overview

AI Agent streaming solves user wait time. SSE for unidirectional token streams; WebSocket for bidirectional interactive agents.

SSE (FastAPI)

@app.get("/stream")
async def stream(query: str):
    async def generate():
        async for chunk in llm.astream(query):
            yield f"data: {chunk.content}\n\n"
        yield "data: [DONE]\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")

LangGraph Streaming

async for event in app.astream_events(inputs, version="v2"):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="")

Comparison

References

• AI Token Streaming Guide
• LangChain WebSocket Tutorial
• LangChain.js Streaming SSE
• LangGraph astream_events