langgraph/docs/docs/agents/streaming.md

search, tags, hide

search	tags	hide
boost 2	agent	tags

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.

{: style="max-height:300px"}

Waiting is for pigeons.

Agent progress

To stream agent progress, use the [stream()][langgraph.graph.state.CompiledStateGraph.stream] or [astream()][langgraph.graph.state.CompiledStateGraph.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

=== "Sync"

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

=== "Async"

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

LLM tokens

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

=== "Sync"

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

=== "Async"

```python
agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)
# highlight-next-line
async for token, metadata in agent.astream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    # highlight-next-line
    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][langgraph.config.get_stream_writer].

=== "Sync"

```python
# highlight-next-line
from langgraph.config import get_stream_writer

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    # highlight-next-line
    writer = get_stream_writer()
    # stream any arbitrary data
    # highlight-next-line
    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": [{"role": "user", "content": "what is the weather in sf"}]},
    # highlight-next-line
    stream_mode="custom"
):
    print(chunk)
    print("\n")
```

=== "Async"

```python
# highlight-next-line
from langgraph.config import get_stream_writer

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    # highlight-next-line
    writer = get_stream_writer()
    # stream any arbitrary data
    # highlight-next-line
    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": [{"role": "user", "content": "what is the weather in sf"}]},
    # highlight-next-line
    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"]:

=== "Sync"

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

for stream_mode, chunk in agent.stream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    # highlight-next-line
    stream_mode=["updates", "messages", "custom"]
):
    print(chunk)
    print("\n")
```

=== "Async"

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

async for stream_mode, chunk in agent.astream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    # highlight-next-line
    stream_mode=["updates", "messages", "custom"]
):
    print(chunk)
    print("\n")
```

Disable streaming

In some applications you might need to disable streaming of individual tokens for a given model. This is useful in multi-agent systems to control which agents stream their output.

See the Models guide to learn how to disable streaming.

Additional resources

• Streaming in LangGraph