Streaming

Streaming is key to building responsive applications. There are a few types of data you’ll want to stream:

1. Agent progress — get updates after each node in the agent graph is executed.
2. LLM tokens — stream tokens as they are generated by the language model.
3. Custom updates — emit custom data from tools during execution (e.g., "Fetched 10/100 records")

You can stream more than one type of data at a time.

Waiting is for pigeons.

Agent progress

To stream agent progress, use the stream() or astream() methods with stream_mode="updates". This emits an event after every agent step.

For example, if you have an agent that calls a tool once, you should see the following updates:

• LLM node: AI message with tool call requests
• Tool node: Tool message with execution result
• LLM node: Final AI response

SyncAsync

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)
for chunk in agent.stream(
    {"messages": "what is the weather in sf"},
    stream_mode="updates"
):
    print(chunk)
    print("\n")

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)
async for chunk in agent.astream(
    {"messages": "what is the weather in sf"},
    stream_mode="updates"
):
    print(chunk)
    print("\n")

LLM tokens

To stream tokens as they are produced by the LLM, use stream_mode="messages":

SyncAsync

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)
for token, metadata in agent.stream(
    {"messages": "what is the weather in sf"},
    stream_mode="messages"
):
    print("Token", token)
    print("Metadata", metadata)
    print("\n")

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)
async for token, metadata in agent.astream(
    {"messages": "what is the weather in sf"},
    stream_mode="messages"
):
    print("Token", token)
    print("Metadata", metadata)
    print("\n")

Tool updates

To stream updates from tools as they are executed, you can use get_stream_writer.

SyncAsync

from langgraph.config import get_stream_writer

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    writer = get_stream_writer()
    # stream any arbitrary data
    writer(f"Looking up data for city: {city}")
    return f"It's always sunny in {city}!"

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)

for chunk in agent.stream(
    {"messages": "what is the weather in sf"},
    stream_mode="custom"
):
    print(chunk)
    print("\n")

from langgraph.config import get_stream_writer

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    writer = get_stream_writer()
    # stream any arbitrary data
    writer(f"Looking up data for city: {city}")
    return f"It's always sunny in {city}!"

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)

async for chunk in agent.astream(
    {"messages": "what is the weather in sf"},
    stream_mode="custom"
):
    print(chunk)
    print("\n")

Note

If you add get_stream_writer inside your tool, you won't be able to invoke the tool outside of a LangGraph execution context.

Stream multiple modes

You can specify multiple streaming modes by passing stream mode as a list: stream_mode=["updates", "messages", "custom"]:

SyncAsync

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)

for stream_mode, chunk in agent.stream(
    {"messages": "what is the weather in sf"},
    stream_mode=["updates", "messages", "custom"]
):
    print(chunk)
    print("\n")

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)

async for stream_mode, chunk in agent.astream(
    {"messages": "what is the weather in sf"},
    stream_mode=["updates", "messages", "custom"]
):
    print(chunk)
    print("\n")

Additional resources

• Streaming in LangGraph

Comments