LangGraph astream_events vs astream_log Stream Mode Errors

# ✅ TOKEN STREAMING for chat UIs — stream_mode="messages"
config = {"configurable": {"thread_id": "chat-1"}}

async for chunk, metadata in app.astream(
    {"messages": [("user", "Explain LangGraph in one paragraph.")]},
    config,
    stream_mode="messages",       # <-- token-by-token
):
    # chunk is an AIMessageChunk; metadata identifies the source node
    if chunk.content:
        print(chunk.content, end="", flush=True)


# ✅ FULL STATE after each step — stream_mode="values"
async for state in app.astream(
    {"messages": [("user", "Hi")]},
    config,
    stream_mode="values",
):
    # state is the full State dict after each step
    print(f"messages so far: {len(state['messages'])}")


# ✅ DELTA UPDATES — stream_mode="updates" (the default)
async for update in app.astream(
    {"messages": [("user", "Hi")]},
    config,
    # stream_mode="updates" (implicit)
):
    # update is {node_name: partial_state_returned_by_that_node}
    for node_name, delta in update.items():
        print(f"{node_name} returned:", delta)


# ✅ MULTIPLE MODES interleaved — pass a list
async for mode, data in app.astream(
    {"messages": [("user", "Hi")]},
    config,
    stream_mode=["updates", "messages"],
):
    if mode == "messages":
        chunk, meta = data
        # forward token to browser
    elif mode == "updates":
        # forward progress event to sidebar
        for node_name, delta in data.items():
            print(f"[progress] {node_name}")


# ✅ CUSTOM events — emit progress signals from inside a node
from langgraph.config import get_stream_writer

async def slow_node(state):
    writer = get_stream_writer()
    writer({"step": "starting", "detail": "loading corpus"})
    # ... slow work ...
    writer({"step": "midway", "progress": 0.5})
    # ... more work ...
    return {"messages": [...]}

# Consumer picks these up with stream_mode="custom"
async for evt in app.astream(inp, config, stream_mode="custom"):
    print("custom:", evt)

# ✅ Full event stream — filter for what you care about
config = {"configurable": {"thread_id": "chat-1"}}

async for event in app.astream_events(
    {"messages": [("user", "Search LangGraph and summarize")]},
    config,
    version="v2",                # <-- REQUIRED in current LangGraph
):
    kind = event["event"]
    name = event.get("name")

    if kind == "on_chat_model_stream":
        # Token-by-token LLM output — the chunk is an AIMessageChunk
        chunk = event["data"]["chunk"]
        if chunk.content:
            print(chunk.content, end="", flush=True)

    elif kind == "on_tool_start":
        print(f"\n[tool {name} starting: {event['data'].get('input')}]")

    elif kind == "on_tool_end":
        # event["data"]["output"] is the tool return value
        print(f"[tool {name} done]")

    elif kind == "on_chain_start" and event.get("metadata", {}).get("langgraph_node"):
        print(f"\n>> node {event['metadata']['langgraph_node']} started")


# ✅ Filter by tags — cleaner than event-type switches for large graphs
from langchain_core.runnables import RunnableConfig

async def responder(state, config: RunnableConfig):
    # Tag the LLM call so events are easy to isolate
    tagged_llm = model.with_config({"tags": ["main_response"]})
    reply = await tagged_llm.ainvoke(state["messages"], config)
    return {"messages": [reply]}


# Consumer filters by tag — only main_response events reach the client
async for event in app.astream_events(inp, config, version="v2"):
    if "main_response" in event.get("tags", []):
        if event["event"] == "on_chat_model_stream":
            yield event["data"]["chunk"].content


# ✅ Custom events — dispatch_custom_event from inside any node
from langchain_core.callbacks import adispatch_custom_event

async def researcher(state, config):
    docs = await search.ainvoke(state["query"])
    await adispatch_custom_event(
        "docs_retrieved",
        {"count": len(docs), "sample_titles": [d.title for d in docs[:3]]},
        config=config,
    )
    return {"docs": docs}

# Consumer sees these as on_custom_event
async for event in app.astream_events(inp, config, version="v2"):
    if event["event"] == "on_custom_event" and event["name"] == "docs_retrieved":
        print("retrieved:", event["data"])


# ❌ DEPRECATED — don't use in new code
# async for log_patch in app.astream_log(inp, config):
#     print(log_patch.ops)         # RunLogPatch — legacy shape

import asyncio
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json


api = FastAPI()


# ✅ PATTERN 1 — bounded queue, backpressure via asyncio.Queue
@api.post("/chat-stream/{thread_id}")
async def chat_stream(thread_id: str, message: str):
    config = {"configurable": {"thread_id": thread_id}}

    # Bounded queue — producer blocks when queue is full
    queue: asyncio.Queue = asyncio.Queue(maxsize=100)

    async def produce():
        try:
            async for chunk, meta in api.state.graph.astream(
                {"messages": [("user", message)]}, config, stream_mode="messages"
            ):
                if chunk.content:
                    await queue.put(chunk.content)   # blocks when queue full
        finally:
            await queue.put(None)                    # sentinel: done

    async def consume():
        producer_task = asyncio.create_task(produce())
        try:
            while True:
                item = await queue.get()
                if item is None:
                    break
                yield f"data: {json.dumps({'chunk': item})}\n\n"
        finally:
            producer_task.cancel()

    return StreamingResponse(consume(), media_type="text/event-stream")


# ✅ PATTERN 2 — drop mid-stream chunks under load
async def rate_limited_stream(app, inp, config, max_chunks_per_sec: int = 30):
    """Emit at most N chunks per second; drop intermediate ones."""
    import time
    last = 0.0
    min_interval = 1.0 / max_chunks_per_sec

    async for chunk, meta in app.astream(inp, config, stream_mode="messages"):
        now = time.perf_counter()
        if now - last < min_interval:
            continue                        # drop this chunk
        last = now
        yield chunk


# ✅ PATTERN 3 — coalesce chunks for lower per-message overhead
async def coalesced_stream(app, inp, config, flush_ms: int = 100):
    """Buffer chunks for up to flush_ms and emit the joined content."""
    buffer: list[str] = []
    last_flush = 0.0

    import time
    async for chunk, meta in app.astream(inp, config, stream_mode="messages"):
        if chunk.content:
            buffer.append(chunk.content)
        now = time.perf_counter() * 1000
        if now - last_flush >= flush_ms and buffer:
            yield "".join(buffer)
            buffer.clear()
            last_flush = now

    if buffer:
        yield "".join(buffer)


# ✅ WebSocket disconnect — cancel the producer
from fastapi import WebSocket, WebSocketDisconnect

@api.websocket("/ws/chat/{thread_id}")
async def ws_chat(websocket: WebSocket, thread_id: str):
    await websocket.accept()
    config = {"configurable": {"thread_id": thread_id}}
    while True:
        try:
            msg = await websocket.receive_text()
        except WebSocketDisconnect:
            return

        task = asyncio.create_task(_stream_reply(websocket, msg, config))
        try:
            await task
        except asyncio.CancelledError:
            pass


async def _stream_reply(ws, msg, config):
    try:
        async for chunk, meta in api.state.graph.astream(
            {"messages": [("user", msg)]}, config, stream_mode="messages"
        ):
            if chunk.content:
                await ws.send_text(chunk.content)
    except Exception as e:
        await ws.send_text(f"[error] {e}")