Streaming Agent Output to Users

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", streaming=True)

# Synchronous streaming
for chunk in llm.stream("Write a 3-paragraph essay about AI agents"):
    print(chunk.content, end="", flush=True)

# Or with astream for async
import asyncio

async def stream_response():
    async for chunk in llm.astream("Write a poem"):
        print(chunk.content, end="", flush=True)
        
asyncio.run(stream_response())

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_community.tools import TavilySearchResults

llm = ChatOpenAI(model="gpt-4o", streaming=True)
tools = [TavilySearchResults(max_results=3)]
agent = create_react_agent(llm, tools)

# Stream mode: "values" — emit full state at each step
for event in agent.stream(
    {"messages": [("human", "Research the current state of AI agents in 2025")]},
    stream_mode="values"
):
    last_msg = event["messages"][-1]
    msg_type = last_msg.__class__.__name__
    
    if hasattr(last_msg, 'content') and last_msg.content:
        print(f"[{msg_type}]: {last_msg.content[:200]}")

# Stream mode: "updates" — emit only what changed at each step
for event in agent.stream(
    {"messages": [("human", "Research AI agents")]},
    stream_mode="updates"
):
    for node_name, node_output in event.items():
        print(f"\n[Node: {node_name}]")
        if "messages" in node_output:
            for msg in node_output["messages"]:
                print(f"  {msg.content[:100] if hasattr(msg, 'content') else str(msg)[:100]}")

# stream_mode="events" gives you every event type
for event in agent.astream_events(
    {"messages": [("human", "What's the weather in Tokyo?")]},
    version="v2"
):
    event_type = event["event"]
    
    if event_type == "on_chat_model_stream":
        # LLM is generating tokens
        chunk = event["data"]["chunk"]
        if chunk.content:
            print(chunk.content, end="", flush=True)
    
    elif event_type == "on_tool_start":
        # A tool is being called
        tool_name = event["name"]
        tool_input = event["data"]["input"]
        print(f"\n🔧 Calling tool: {tool_name}")
        print(f"   Input: {tool_input}")
    
    elif event_type == "on_tool_end":
        # Tool finished
        tool_name = event["name"]
        print(f"   ✓ {tool_name} completed")
    
    elif event_type == "on_chain_end" and event["name"] == "LangGraph":
        # Entire agent finished
        print("\n\n✅ Agent completed")

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
import asyncio
import json

app = FastAPI()
llm = ChatOpenAI(model="gpt-4o", streaming=True)
agent = create_react_agent(llm, tools)

@app.get("/agent/stream")
async def stream_agent(task: str):
    async def event_generator():
        try:
            async for event in agent.astream_events(
                {"messages": [("human", task)]},
                version="v2"
            ):
                event_type = event["event"]
                
                if event_type == "on_chat_model_stream":
                    chunk = event["data"]["chunk"]
                    if chunk.content:
                        yield f"data: {json.dumps({'type': 'token', 'content': chunk.content})}\n\n"
                
                elif event_type == "on_tool_start":
                    yield f"data: {json.dumps({'type': 'tool_start', 'tool': event['name'], 'input': str(event['data'].get('input', ''))[:100]})}\n\n"
                
                elif event_type == "on_tool_end":
                    yield f"data: {json.dumps({'type': 'tool_end', 'tool': event['name']})}\n\n"
                
                elif event_type == "on_chain_end" and event["name"] == "LangGraph":
                    yield f"data: {json.dumps({'type': 'done'})}\n\n"
                    
        except Exception as e:
            yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
    
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no"  # Disable Nginx buffering
        }
    )

// React component that streams agent output
function AgentStream({ task }: { task: string }) {
    const [output, setOutput] = useState<string>("");
    const [toolCalls, setToolCalls] = useState<string[]>([]);
    const [isComplete, setIsComplete] = useState(false);
    
    useEffect(() => {
        const eventSource = new EventSource(
            `/agent/stream?task=${encodeURIComponent(task)}`
        );
        
        eventSource.onmessage = (event) => {
            const data = JSON.parse(event.data);
            
            if (data.type === "token") {
                setOutput(prev => prev + data.content);
            } else if (data.type === "tool_start") {
                setToolCalls(prev => [...prev, `🔧 ${data.tool}: ${data.input}`]);
            } else if (data.type === "done") {
                setIsComplete(true);
                eventSource.close();
            } else if (data.type === "error") {
                console.error("Agent error:", data.message);
                eventSource.close();
            }
        };
        
        return () => eventSource.close();
    }, [task]);
    
    return (
        <div>
            {toolCalls.length > 0 && (
                <div className="tool-calls">
                    {toolCalls.map((call, i) => <p key={i} className="text-gray-500 text-sm">{call}</p>)}
                </div>
            )}
            <div className="output">
                {output}
                {!isComplete && <span className="animate-pulse">▊</span>}
            </div>
        </div>
    );
}

from langchain_core.callbacks import BaseCallbackHandler

class StreamingProgressCallback(BaseCallbackHandler):
    """Custom callback for tracking and streaming agent progress."""
    
    def __init__(self, progress_queue: asyncio.Queue):
        self.queue = progress_queue
    
    def on_tool_start(self, serialized, input_str, **kwargs):
        tool_name = serialized.get("name", "unknown")
        asyncio.create_task(
            self.queue.put({"type": "tool_start", "tool": tool_name, "input": input_str[:100]})
        )
    
    def on_tool_end(self, output, **kwargs):
        asyncio.create_task(
            self.queue.put({"type": "tool_end", "output": str(output)[:200]})
        )
    
    def on_llm_new_token(self, token, **kwargs):
        asyncio.create_task(
            self.queue.put({"type": "token", "content": token})
        )

async def buffered_token_stream(agent, task: str):
    buffer = []
    async for event in agent.astream_events({"messages": [("human", task)]}, version="v2"):
        if event["event"] == "on_chat_model_stream":
            buffer.append(event["data"]["chunk"].content)
            if len(buffer) >= 10:
                yield "".join(buffer)
                buffer = []
    if buffer:
        yield "".join(buffer)

headers = {"Last-Event-ID": "0"}  # Client sends this on reconnect
# Server uses this to resume from where it left off