LangChain astream_events explained with examples

astream_events is LangChain’s unified streaming API that emits structured events for every step of a chain or agent execution — LLM tokens, tool calls, retriever results, and custom callbacks — as an async iterator of typed event dictionaries. It replaces the fragmented astream, astream_log, and callback handlers with a single, predictable surface that works across all Runnable implementations. If you’re building anything that needs visibility into intermediate steps — debugging, UX streaming, observability — this is the primitive you should standardize on.

How the event stream works

Every Runnable in LangChain (chains, agents, tools, retrievers, prompts) implements the astream_events method. When invoked, it returns an AsyncIterator[dict] where each dictionary follows a consistent schema:

{
    "event": "on_chain_start" | "on_chain_stream" | "on_chain_end" | "on_tool_start" | "on_tool_end" | "on_llm_start" | "on_llm_stream" | "on_llm_end" | "on_retriever_start" | "on_retriever_end" | "on_custom_event",
    "name": "RunnableName",
    "run_id": "uuid",
    "parent_run_id": "uuid | None",
    "tags": ["tag1", "tag2"],
    "metadata": {"key": "value"},
    "data": {...},  # event-specific payload
}

The event field tells you what happened. The data field carries the payload — tokens for on_llm_stream, input/output for chain/tool boundaries, documents for retrievers. The run_id and parent_run_id let you reconstruct the execution tree, which is essential for nested chains and agents.

Events fire in a deterministic order: start → stream* → end. For an LLM call you get on_llm_start, zero or more on_llm_stream (one per token chunk), then on_llm_end. For a tool: on_tool_start, on_tool_end. For a chain: on_chain_start, any child events, on_chain_stream (if the chain yields intermediate values), on_chain_end.

Why it matters compared to the alternatives

Before astream_events (added in LangChain 0.1.0), you had three competing streaming APIs:

astream only gives you the final streamed output — useless if you need to show tool calls, retrieval results, or reasoning steps. astream_log emits JSON patches against a run log, which is powerful but requires understanding the patch format and reconstructing state. Callbacks give you everything but force you to implement a handler class and wire it through config.

astream_events unifies these: one iterator, typed events, no custom classes. You get token streaming and structural visibility. The tradeoff: you consume events sequentially, so backpressure handling is manual (see the example below).

Concrete example: streaming an agent with tool visibility

Here’s a complete, runnable example using a ReAct-style agent with a search tool. This pattern works for any Runnable — chains, custom tools, RAG pipelines.

import asyncio
import json
from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import Tool
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# Simple search tool for demonstration
def search_fn(query: str) -> str:
    # In production, call a real search API
    return f"Search results for: {query}"

search_tool = Tool.from_function(
    func=search_fn,
    name="search",
    description="Search the web for information",
)

prompt = PromptTemplate.from_template("""Answer the question using available tools.

{tools}

Question: {input}
Thought: {agent_scratchpad}""")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, streaming=True)
agent = create_react_agent(llm, [search_tool], prompt)
executor = AgentExecutor(agent=agent, tools=[search_tool], verbose=False)

async def stream_agent(question: str):
    """Stream agent execution with full event visibility."""
    async for event in executor.astream_events(
        {"input": question},
        version="v2",  # required for v2 event schema
    ):
        event_type = event["event"]
        name = event["name"]
        data = event.get("data", {})
        
        # Filter to interesting events for demo
        if event_type == "on_llm_stream":
            chunk = data.get("chunk")
            if chunk and chunk.content:
                print(chunk.content, end="", flush=True)
                
        elif event_type == "on_tool_start":
            print(f"\n[tool: {name}] input: {data.get('input')}")
            
        elif event_type == "on_tool_end":
            output = data.get("output")
            print(f"[tool: {name}] output: {output[:200]}...")
            
        elif event_type == "on_chain_start" and name == "AgentExecutor":
            print(f"\n[agent start] run_id: {event['run_id']}")
            
        elif event_type == "on_chain_end" and name == "AgentExecutor":
            print(f"\n[agent end] output: {data.get('output', {}).get('output', '')[:200]}")

# Run it
asyncio.run(stream_agent("What is the capital of France?"))

Sample output:

[agent start] run_id: 550e8400-e29b-41d4-a716-446655440000
[tool: search] input: capital of France
[tool: search] output: Search results for: capital of France...
The capital of France is Paris.
[agent end] output: The capital of France is Paris.

Key details in this example:

• version="v2" — required. The v1 schema (deprecated) used different event names and nested payloads. Always pass version="v2".
• Token streaming — on_llm_stream fires per provider chunk. OpenAI-compatible endpoints typically emit 1-10 tokens per event. Print with flush=True for real-time UX.
• Tool boundaries — on_tool_start/on_tool_end give you input/output without parsing the agent’s internal scratchpad.
• Run IDs — correlate events across distributed tracing systems. The run_id on on_chain_start for AgentExecutor is the root; all child events carry parent_run_id pointing to it.

Filtering and transforming the event stream

Raw events are noisy. In production you’ll filter, aggregate, or transform. Common patterns:

Only final output tokens (like astream)

async def stream_final_output(runnable, input_data):
    async for event in runnable.astream_events(input_data, version="v2"):
        if event["event"] == "on_chain_stream" and event["name"] == "MyChain":
            yield event["data"]["chunk"]

Collect tool calls for a sidebar UI

async def collect_tool_calls(runnable, input_data):
    tool_calls = []
    async for event in runnable.astream_events(input_data, version="v2"):
        if event["event"] == "on_tool_start":
            tool_calls.append({
                "name": event["name"],
                "input": event["data"]["input"],
                "run_id": event["run_id"],
                "status": "running",
            })
        elif event["event"] == "on_tool_end":
            for tc in tool_calls:
                if tc["run_id"] == event["run_id"]:
                    tc["output"] = event["data"]["output"]
                    tc["status"] = "complete"
    return tool_calls

Reconstruct the execution tree for debugging

from collections import defaultdict

def build_tree(events):
    """Build parent->children mapping from event stream."""
    tree = defaultdict(list)
    nodes = {}
    for event in events:
        run_id = event["run_id"]
        parent = event.get("parent_run_id")
        nodes[run_id] = event
        if parent:
            tree[parent].append(run_id)
    return tree, nodes

Common misconceptions

“Astream_events replaces callbacks entirely”

False. Callbacks still exist and are useful for:

• Cross-cutting concerns (logging, metrics, tracing) that apply to every run without modifying calling code
• Synchronous contexts where you can’t use async for
• Integrations that expect the BaseCallbackHandler interface (LangSmith, LangFuse, custom observability)

astream_events is for consuming events in your application logic. Callbacks are for observing runs you don’t directly control.

“The event schema is stable across versions”

The version="v2" parameter exists because v1 had a different schema. LangChain treats v2 as stable, but minor releases occasionally add new event types (e.g., on_custom_event for user-defined events). Pin your LangChain version and test upgrades. The core events (on_llm_*, on_tool_*, on_chain_*, on_retriever_*) are stable.

“I can parallelize event consumption”

You cannot. astream_events returns a single async iterator. The events are produced sequentially by the Runnable’s execution. If you need parallel processing (e.g., stream to UI and log to disk), you must tee the stream:

import asyncio
from itertools import tee

async def tee_stream(runnable, input_data):
    # This doesn't work directly — async iterators can't be teed safely
    # Instead, collect or use a broadcast pattern:
    queue1, queue2 = asyncio.Queue(), asyncio.Queue()
    
    async def producer():
        async for event in runnable.astream_events(input_data, version="v2"):
            await queue1.put(event)
            await queue2.put(event)
        await queue1.put(None)
        await queue2.put(None)
    
    async def consumer(queue, name):
        while True:
            event = await queue.get()
            if event is None:
                break
            print(f"[{name}] {event['event']}")
    
    await asyncio.gather(producer(), consumer(queue1, "ui"), consumer(queue2, "log"))

“Astream_events works with synchronous code”

No. It returns an AsyncIterator. You must await it inside an async function. If you’re in a sync context (Flask, Django views, scripts), you have options:

# Option 1: Run in event loop (scripts)
import asyncio
events = asyncio.run(collect_events(runnable, input_data))

# Option 2: Run in thread pool (sync web frameworks)
from concurrent.futures import ThreadPoolExecutor

def sync_stream(runnable, input_data):
    def _run():
        return asyncio.run(collect_events(runnable, input_data))
    with ThreadPoolExecutor() as pool:
        return pool.submit(_run).result()

Avoid asyncio.run inside an already-running loop (Jupyter, FastAPI, async workers) — it raises RuntimeError. Use asyncio.get_event_loop().create_task() or the framework’s native async support instead.

“All Runnables emit the same events”

Mostly true, but custom Runnable implementations can emit on_custom_event with arbitrary payloads. If you wrap third-party components, check their docs. The built-in primitives (LLMs, tools, retrievers, prompt templates, output parsers) follow the standard schema.

Integration with observability

If you’re sending traces to LangSmith, LangFuse, or a custom OpenTelemetry backend, you don’t need to manually forward astream_events — the callback system already handles it. But if you are building a custom observability layer, the event stream is your source of truth:

async def trace_to_otel(runnable, input_data, tracer):
    async for event in runnable.astream_events(input_data, version="v2"):
        run_id = event["run_id"]
        parent_id = event.get("parent_run_id")
        event_type = event["event"]
        
        with tracer.start_as_current_span(
            f"langchain.{event_type}",
            context=set_span_in_context(trace.get_current_span()) if parent_id else None,
        ) as span:
            span.set_attribute("langchain.run_id", run_id)
            if parent_id:
                span.set_attribute("langchain.parent_run_id", parent_id)
            span.set_attribute("langchain.event_type", event_type)
            span.set_attribute("langchain.name", event["name"])
            # Add data payload as attributes (truncate large fields)
            for k, v in event.get("data", {}).items():
                if isinstance(v, (str, int, float, bool)):
                    span.set_attribute(f"langchain.data.{k}", v)

This gives you a proper span hierarchy matching the execution tree — something you can’t easily get from astream or astream_log alone.

When to use each streaming API

Performance considerations

• Event overhead: Each event is a Python dict allocation. For high-throughput token streaming (thousands of tokens/sec), astream has lower overhead because it yields raw chunks without wrapping.
• Memory: The iterator holds one event at a time. No buffering unless you collect. Safe for long generations.
• Backpressure: The producer (Runnable) pauses when the consumer (your async for loop) doesn’t await the next iteration. If you do heavy processing per event, consider batching or offloading to a worker queue.

Summary

astream_events is the most versatile streaming API in LangChain. It gives you typed, structured events for every execution step — tokens, tools, retrieval, chain boundaries — with a consistent schema and execution-tree correlation via run_id/parent_run_id. Use it when you need visibility beyond final output. Filter aggressively in your consumer. Remember version="v2". And don’t reach for callbacks unless you need cross-cutting concerns or sync compatibility.