Streaming Agents

Capsule overview

In the previous capsules you learned to create agents with create_agent, configure system prompts, and manage state and memory. But there's a practical problem: when you call agent.invoke(), your program blocks until the agent finishes all of its internal iterations — tool calls, analyzing results, more tool calls, and finally the answer. If the agent needs 3 rounds of tools, you can sit there for 10-15 seconds watching... nothing.

With streaming, you see every step the agent takes in real time: which tool it's calling, what came back, what the model is thinking. This doesn't just improve the user experience — it gives you visibility into the agent's reasoning process, which is critical for debugging and for building professional interfaces.

agent.stream() is LangChain's streaming interface for agents. It lets you pick how much detail you want with stream_mode: the full state after every step, only the changes, or even the model's individual tokens.

agent.stream() — the streaming interface

Instead of agent.invoke(), which returns the final result, agent.stream() returns an iterator that emits events as the agent makes progress.

Basic example: watching every step

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """Search for information on a topic."""
    data = {
        "langchain": "LangChain is a framework for LLM applications.",
        "python": "Python is a high-level programming language.",
    }
    for key, value in data.items():
        if key in query.lower():
            return value
    return f"No information found about: {query}"

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search])

for step in agent.stream({"messages": [("user", "What is LangChain?")]}):
    print(step)
    print("---")
# Expected output:
# {'agent': {'messages': [AIMessage(content='', tool_calls=[{'name': 'search', 'args': {'query': 'langchain'}, 'id': 'call_abc123'}])]}}
# ---
# {'tools': {'messages': [ToolMessage(content='LangChain is a framework for LLM applications.', tool_call_id='call_abc123')]}}
# ---
# {'agent': {'messages': [AIMessage(content='LangChain is a framework designed for building applications that use language models (LLMs).')]}}
# ---

Each step is a dictionary where the key is the name of the node that produced the event ("agent" for the model, "tools" for tool execution) and the value holds the messages it produced.

stream_mode: controlling the level of detail

The stream_mode parameter defines what information the stream emits. There are three main modes:

stream_mode="updates" (the default)

Emits only the changes produced by each node. It's the most practical mode for most cases.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

@tool
def search(query: str) -> str:
    """Search for information on a topic."""
    return f"Result: {query} is an important concept in AI."

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search])

for step in agent.stream(
    {"messages": [("user", "What is LangChain?")]},
    stream_mode="updates"
):
    for node_name, update in step.items():
        print(f"[{node_name}]")
        if "messages" in update:
            for msg in update["messages"]:
                if isinstance(msg, AIMessage) and msg.tool_calls:
                    for tc in msg.tool_calls:
                        print(f"  → Calling: {tc['name']}({tc['args']})")
                elif isinstance(msg, AIMessage):
                    print(f"  Answer: {msg.content[:100]}")
                elif isinstance(msg, ToolMessage):
                    print(f"  ← Result: {msg.content[:80]}")
# Expected output:
# [agent]
#   → Calling: search({'query': 'LangChain'})
# [tools]
#   ← Result: Result: LangChain is an important concept in AI.
# [agent]
#   Answer: LangChain is an important concept in the field of artificial intelligence...

stream_mode="values"

Emits the agent's full state after every step. That includes the entire accumulated message history.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """Search for information on a topic."""
    return f"Result: {query} is a framework for LLMs."

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search])

for step in agent.stream(
    {"messages": [("user", "What is LangChain?")]},
    stream_mode="values"
):
    messages = step["messages"]
    print(f"Total messages: {len(messages)}")
    last = messages[-1]
    print(f"  Last: {type(last).__name__} → {str(last.content)[:80]}")
    print("---")
# Expected output:
# Total messages: 1
#   Last: HumanMessage → What is LangChain?
# ---
# Total messages: 2
#   Last: AIMessage →
# ---
# Total messages: 3
#   Last: ToolMessage → Result: LangChain is a framework for LLMs.
# ---
# Total messages: 4
#   Last: AIMessage → LangChain is a framework designed for working with language mo
# ---

With "values", every step includes all the messages from the start. Useful when you need the full context on each iteration.

stream_mode="messages"

Emits the model's individual tokens, giving you maximum granularity. Each event is a (message_chunk, metadata) tuple.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """Search for information on a topic."""
    return f"LangChain is an open-source framework for LLM applications."

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search])

for chunk, metadata in agent.stream(
    {"messages": [("user", "What is LangChain?")]},
    stream_mode="messages"
):
    if chunk.content:
        print(chunk.content, end="", flush=True)
print()
# Expected output:
# LangChain is an open-source framework designed for building applications that use language models (LLMs). It lets you integrate tools, memory, and processing chains to build intelligent systems.

With "messages", you can show the agent's answer token by token, exactly the way ChatGPT or Claude render their responses.

Processing stream events: model vs tools

When you use stream_mode="updates", you need to tell model events apart from tool events so you can show the user something meaningful.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    weathers = {
        "Madrid": "Sunny, 25°C",
        "Paris": "Cloudy, 16°C",
        "Tokyo": "Rainy, 19°C",
    }
    return weathers.get(city, f"Weather not available for {city}")

@tool
def get_population(city: str) -> str:
    """Get the population of a city."""
    populations = {
        "Madrid": "3.3 million",
        "Paris": "2.1 million",
        "Tokyo": "13.9 million",
    }
    return populations.get(city, f"Population not available for {city}")

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [get_weather, get_population])

step_count = 0
for step in agent.stream(
    {"messages": [("user", "Weather and population of Madrid?")]},
    stream_mode="updates"
):
    step_count += 1
    for node_name, update in step.items():
        if node_name == "agent":
            for msg in update["messages"]:
                if isinstance(msg, AIMessage) and msg.tool_calls:
                    print(f"[Step {step_count}] Agent decides to call tools:")
                    for tc in msg.tool_calls:
                        print(f"  → {tc['name']}({tc['args']})")
                elif isinstance(msg, AIMessage) and msg.content:
                    print(f"[Step {step_count}] Agent answers:")
                    print(f"  {msg.content}")
        elif node_name == "tools":
            for msg in update["messages"]:
                if isinstance(msg, ToolMessage):
                    print(f"[Step {step_count}] Tool returned: {msg.content}")
# Expected output:
# [Step 1] Agent decides to call tools:
#   → get_weather({'city': 'Madrid'})
#   → get_population({'city': 'Madrid'})
# [Step 2] Tool returned: Sunny, 25°C
# [Step 2] Tool returned: 3.3 million
# [Step 3] Agent answers:
#   Madrid has sunny weather at 25°C and a population of 3.3 million.

Streaming the model's individual tokens

The "messages" mode gives you access to every token the model generates. This is essential for chat-style interfaces.

Filtering for only the final answer's tokens

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessageChunk

@tool
def search(query: str) -> str:
    """Search for information on a topic."""
    return f"{query}: a popular framework for AI applications."

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search])

for chunk, metadata in agent.stream(
    {"messages": [("user", "What is LangChain?")]},
    stream_mode="messages"
):
    if isinstance(chunk, AIMessageChunk):
        if chunk.tool_call_chunks:
            pass
        elif chunk.content:
            print(chunk.content, end="", flush=True)
print()
# Expected output:
# LangChain is a popular framework for building artificial intelligence applications that combine language models with external tools.

Using metadata to identify the source node

The metadata includes information about which node of the graph produced the event, which lets you filter precisely.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessageChunk

@tool
def calculate(expression: str) -> str:
    """Evaluate a math expression."""
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"Error: {e}"

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [calculate])

print("Tokens from the final answer:")
for chunk, metadata in agent.stream(
    {"messages": [("user", "What's 15 * 37 + 42?")]},
    stream_mode="messages"
):
    node = metadata.get("langgraph_node", "")
    if isinstance(chunk, AIMessageChunk) and chunk.content and node == "agent":
        print(chunk.content, end="", flush=True)
print()
# Expected output:
# Tokens from the final answer:
# The result of 15 × 37 + 42 is **597**.

Showing tool call progress to the user

In a real interface, you want to show the user what the agent is doing: "Searching... Done!" instead of dead silence.

from dotenv import load_dotenv
load_dotenv()

import time
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    time.sleep(0.5)
    return f"Found: {query} has many production applications."

@tool
def get_price(product: str) -> str:
    """Get the price of a product."""
    time.sleep(0.3)
    prices = {"langchain": "Open source (free)", "langsmith": "From $39/month"}
    return prices.get(product.lower(), f"Price not available for {product}")

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search_web, get_price])

active_tools = {}

for step in agent.stream(
    {"messages": [("user", "Look up what LangChain is and how much LangSmith costs")]},
    stream_mode="updates"
):
    for node_name, update in step.items():
        for msg in update.get("messages", []):
            if isinstance(msg, AIMessage) and msg.tool_calls:
                for tc in msg.tool_calls:
                    active_tools[tc["id"]] = tc["name"]
                    print(f"⏳ {tc['name']}({tc['args']})...")
            elif isinstance(msg, ToolMessage):
                tool_name = active_tools.get(msg.tool_call_id, "tool")
                print(f"✅ {tool_name} done → {msg.content[:60]}")
            elif isinstance(msg, AIMessage) and msg.content:
                print(f"\n📝 Final answer:\n{msg.content}")
# Expected output:
# ⏳ search_web({'query': 'LangChain'})...
# ⏳ get_price({'product': 'LangSmith'})...
# ✅ search_web done → Found: LangChain has many production applications.
# ✅ get_price done → From $39/month
#
# 📝 Final answer:
# LangChain is a framework with many production applications. LangSmith is priced from $39/month.

Streaming in an async context (astream)

For web applications or APIs, you need the async version of streaming. agent.astream() works just like agent.stream() but inside an async context.

import asyncio
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

@tool
def search(query: str) -> str:
    """Search for information on a topic."""
    return f"Result for '{query}': relevant information found."

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search])

async def stream_agent():
    async for step in agent.astream(
        {"messages": [("user", "What is RAG?")]},
        stream_mode="updates"
    ):
        for node_name, update in step.items():
            for msg in update.get("messages", []):
                if isinstance(msg, AIMessage) and msg.tool_calls:
                    for tc in msg.tool_calls:
                        print(f"[async] Tool: {tc['name']}")
                elif isinstance(msg, AIMessage) and msg.content:
                    print(f"[async] Answer: {msg.content[:100]}")
                elif isinstance(msg, ToolMessage):
                    print(f"[async] Result: {msg.content[:60]}")

asyncio.run(stream_agent())
# Expected output:
# [async] Tool: search
# [async] Result: Result for 'RAG': relevant information found.
# [async] Answer: RAG (Retrieval-Augmented Generation) is a technique that combines retrieving...

Streaming tokens async

import asyncio
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessageChunk

@tool
def search(query: str) -> str:
    """Search for information on a topic."""
    return f"{query}: an AI technique that combines retrieval with generation."

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search])

async def stream_tokens():
    async for chunk, metadata in agent.astream(
        {"messages": [("user", "Explain RAG briefly")]},
        stream_mode="messages"
    ):
        if isinstance(chunk, AIMessageChunk) and chunk.content:
            if not chunk.tool_call_chunks:
                print(chunk.content, end="", flush=True)
    print()

asyncio.run(stream_tokens())
# Expected output:
# RAG (Retrieval-Augmented Generation) is an artificial intelligence technique that combines searching for information in databases with generating text through language models.

Building UI-friendly output

In a real application, you need to turn the stream's events into a format your frontend can consume.

The pattern: collect structured events

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

@tool
def search(query: str) -> str:
    """Search for information on a topic."""
    return f"Result: {query} is a modern AI technology."

@tool
def summarize(text: str) -> str:
    """Summarize a piece of text."""
    return f"Summary: {text[:50]}..."

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search, summarize])

ui_events = []

for step in agent.stream(
    {"messages": [("user", "Look up what LangChain is and summarize it")]},
    stream_mode="updates"
):
    for node_name, update in step.items():
        for msg in update.get("messages", []):
            if isinstance(msg, AIMessage) and msg.tool_calls:
                for tc in msg.tool_calls:
                    ui_events.append({
                        "type": "tool_start",
                        "tool": tc["name"],
                        "args": tc["args"],
                    })
            elif isinstance(msg, ToolMessage):
                ui_events.append({
                    "type": "tool_result",
                    "content": msg.content,
                    "tool_call_id": msg.tool_call_id,
                })
            elif isinstance(msg, AIMessage) and msg.content:
                ui_events.append({
                    "type": "response",
                    "content": msg.content,
                })

print("Events for the UI:")
for event in ui_events:
    print(f"  {event['type']}: ", end="")
    if event["type"] == "tool_start":
        print(f"{event['tool']}({event['args']})")
    elif event["type"] == "tool_result":
        print(f"{event['content'][:60]}")
    elif event["type"] == "response":
        print(f"{event['content'][:80]}")
# Expected output:
# Events for the UI:
#   tool_start: search({'query': 'LangChain'})
#   tool_result: Result: LangChain is a modern AI technology.
#   response: LangChain is a modern artificial intelligence technology that lets you buil

These ui_events can be pushed to a frontend over WebSocket or Server-Sent Events to render a chat interface with progress indicators.

Comparison: stream_mode "values" vs "updates" vs "messages"

When to use each one

• ✅ "updates" — Most applications. Shows which tools were called and what they returned
• ✅ "messages" — ChatGPT-style chat interfaces where the answer appears token by token
• ✅ "values" — Debugging, or when you need access to the full state at every step
• ⚠️ Don't mix modes in a single call to stream() — pick one

Combining modes: updates + manual tokens

If you need both tool progress and token streaming, you can do two passes, or use "updates" and detect the final answer to run a second stream:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage, AIMessageChunk

@tool
def search(query: str) -> str:
    """Search for information on a topic."""
    return f"{query}: an AI framework for production."

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search])

for step in agent.stream(
    {"messages": [("user", "What is LangChain?")]},
    stream_mode="updates"
):
    for node_name, update in step.items():
        for msg in update.get("messages", []):
            if isinstance(msg, AIMessage) and msg.tool_calls:
                for tc in msg.tool_calls:
                    print(f"🔧 Calling {tc['name']}...")
            elif isinstance(msg, ToolMessage):
                print(f"✅ Result received")
            elif isinstance(msg, AIMessage) and msg.content:
                print(f"💬 {msg.content}")
# Expected output:
# 🔧 Calling search...
# ✅ Result received
# 💬 LangChain is an artificial intelligence framework designed for production applications...

Connection to the project

In the module project (Capsule 08), you'll use streaming to build a research agent where the user watches in real time: which sources the agent is consulting, what data it gathered, and the final answer appearing token by token. The ui_events pattern will be your foundation for dropping the agent into any web interface.

Troubleshooting

Problem 1: The stream emits nothing

Cause: The agent isn't configured correctly, or the question doesn't trigger any tool. Fix: Check that the agent has tools and that the question actually triggers them:

agent = create_agent(model, [search])
for step in agent.stream({"messages": [("user", "Look up info about Python")]}):
    print(step)

Problem 2: stream_mode="messages" doesn't show tokens

Cause: You're filtering the chunks wrong, or the model returns all the content in a single chunk. Fix: Print every chunk unfiltered to diagnose:

for chunk, metadata in agent.stream(
    {"messages": [("user", "Hello")]},
    stream_mode="messages"
):
    print(f"type={type(chunk).__name__}, content='{chunk.content}', tool_chunks={bool(chunk.tool_call_chunks) if hasattr(chunk, 'tool_call_chunks') else 'N/A'}")

Problem 3: Confusing tool calls with the final answer in updates

Cause: An AIMessage can carry tool_calls (the agent wants to call tools) or content (the final answer), or both. Fix: Check tool_calls first:

if isinstance(msg, AIMessage):
    if msg.tool_calls:
        pass  # The agent wants to call tools
    elif msg.content:
        pass  # The agent's final answer

Problem 4: astream doesn't work outside async

Cause: astream requires an async context (async def + await). Fix: Use asyncio.run() or run it inside an async framework:

import asyncio

async def main():
    async for step in agent.astream({"messages": [("user", "Hello")]}):
        print(step)

asyncio.run(main())

Problem 5: Duplicate events in the stream

Cause: You're iterating over step.items() without filtering by node, and the same message shows up in multiple contexts. Fix: Filter explicitly by node_name:

for step in agent.stream(input_data, stream_mode="updates"):
    for node_name, update in step.items():
        if node_name == "agent":
            pass  # Model events only

Exercises

Exercise 1: Basic stream with updates (Easy)

Create an agent with a get_weather tool and stream it with stream_mode="updates". Print each step, identifying whether it's an agent decision or a tool result.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    weathers = {"Madrid": "Sunny, 24°C", "Lima": "Cloudy, 19°C"}
    return weathers.get(city, f"Weather not available for {city}")

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [get_weather])

for step in agent.stream(
    {"messages": [("user", "What's the weather in Madrid?")]},
    stream_mode="updates"
):
    for node_name, update in step.items():
        for msg in update.get("messages", []):
            if isinstance(msg, AIMessage) and msg.tool_calls:
                print(f"[AGENT] Decides to call:")
                for tc in msg.tool_calls:
                    print(f"  → {tc['name']}({tc['args']})")
            elif isinstance(msg, ToolMessage):
                print(f"[TOOL] Result: {msg.content}")
            elif isinstance(msg, AIMessage) and msg.content:
                print(f"[AGENT] Answer: {msg.content}")
# Expected output:
# [AGENT] Decides to call:
#   → get_weather({'city': 'Madrid'})
# [TOOL] Result: Sunny, 24°C
# [AGENT] Answer: The weather in Madrid is sunny with a temperature of 24°C.

Exercise 2: Token-by-token streaming (Easy)

Use stream_mode="messages" to display the agent's final answer token by token, ignoring the tool call chunks.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessageChunk

@tool
def get_info(topic: str) -> str:
    """Get information about a topic."""
    return f"{topic} is a fundamental tool in modern software development."

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [get_info])

token_count = 0
for chunk, metadata in agent.stream(
    {"messages": [("user", "What is Docker?")]},
    stream_mode="messages"
):
    if isinstance(chunk, AIMessageChunk) and chunk.content:
        if not chunk.tool_call_chunks:
            print(chunk.content, end="", flush=True)
            token_count += 1
print(f"\n\n(Total: {token_count} chunks received)")
# Expected output:
# Docker is a fundamental tool in modern software development that lets you build, ship, and run applications in containers...
#
# (Total: ~30 chunks received)

Exercise 3: Progress indicator for tools (Medium)

Create an agent with two tools and show a progress indicator in the style of "⏳ Searching... ✅ Done" for each tool call.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

@tool
def search_docs(query: str) -> str:
    """Search the documentation."""
    return f"Documentation found: {query} has 3 main methods."

@tool
def search_examples(query: str) -> str:
    """Search for code examples."""
    return f"Example found: use {query} with async/await."

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search_docs, search_examples])

pending_tools = {}

for step in agent.stream(
    {"messages": [("user", "Find documentation and examples for LangChain agents")]},
    stream_mode="updates"
):
    for node_name, update in step.items():
        for msg in update.get("messages", []):
            if isinstance(msg, AIMessage) and msg.tool_calls:
                for tc in msg.tool_calls:
                    pending_tools[tc["id"]] = tc["name"]
                    print(f"⏳ {tc['name']}('{tc['args'].get('query', '')}')...")
            elif isinstance(msg, ToolMessage):
                tool_name = pending_tools.pop(msg.tool_call_id, "unknown")
                print(f"✅ {tool_name} → {msg.content[:50]}...")
            elif isinstance(msg, AIMessage) and msg.content:
                print(f"\n📋 Answer:\n{msg.content}")
# Expected output:
# ⏳ search_docs('LangChain agents')...
# ⏳ search_examples('LangChain agents')...
# ✅ search_docs → Documentation found: LangChain agents has 3 m...
# ✅ search_examples → Example found: use LangChain agents with as...
#
# 📋 Answer:
# Here's what I found about LangChain agents...

Exercise 4: Compare the output of all three stream_modes (Medium)

Run the same question with "values", "updates", and "messages". For each mode, count how many events it emits and show the type of each one.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessageChunk

@tool
def search(query: str) -> str:
    """Search for information."""
    return f"Result: {query} is an AI concept."

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search])
input_data = {"messages": [("user", "What is RAG?")]}

print("=== stream_mode='values' ===")
count_values = 0
for step in agent.stream(input_data, stream_mode="values"):
    count_values += 1
    msgs = step["messages"]
    last = msgs[-1]
    print(f"  Step {count_values}: {len(msgs)} messages, last={type(last).__name__}")
print(f"  Total: {count_values} events\n")

print("=== stream_mode='updates' ===")
count_updates = 0
for step in agent.stream(input_data, stream_mode="updates"):
    count_updates += 1
    for node, update in step.items():
        msg_types = [type(m).__name__ for m in update.get("messages", [])]
        print(f"  Step {count_updates}: node={node}, types={msg_types}")
print(f"  Total: {count_updates} events\n")

print("=== stream_mode='messages' ===")
count_messages = 0
for chunk, metadata in agent.stream(input_data, stream_mode="messages"):
    count_messages += 1
print(f"  Total: {count_messages} chunks")
# Expected output:
# === stream_mode='values' ===
#   Step 1: 1 messages, last=HumanMessage
#   Step 2: 2 messages, last=AIMessage
#   Step 3: 3 messages, last=ToolMessage
#   Step 4: 4 messages, last=AIMessage
#   Total: 4 events
#
# === stream_mode='updates' ===
#   Step 1: node=agent, types=['AIMessage']
#   Step 2: node=tools, types=['ToolMessage']
#   Step 3: node=agent, types=['AIMessage']
#   Total: 3 events
#
# === stream_mode='messages' ===
#   Total: ~35 chunks

Exercise 5: Async agent with astream (Medium)

Rewrite a synchronous agent to use astream with stream_mode="updates". Show the tools' progress and the final answer.

import asyncio
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

@tool
def fetch_data(source: str) -> str:
    """Fetch data from a source."""
    sources = {
        "wikipedia": "RAG combines document retrieval with text generation.",
        "arxiv": "RAG was proposed by Lewis et al. in 2020.",
    }
    return sources.get(source.lower(), f"Source '{source}' not available.")

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [fetch_data])

async def run_async_agent():
    print("Starting async agent...\n")
    async for step in agent.astream(
        {"messages": [("user", "Look up information about RAG on Wikipedia and arXiv")]},
        stream_mode="updates"
    ):
        for node_name, update in step.items():
            for msg in update.get("messages", []):
                if isinstance(msg, AIMessage) and msg.tool_calls:
                    for tc in msg.tool_calls:
                        print(f"⏳ [{node_name}] {tc['name']}({tc['args']})")
                elif isinstance(msg, ToolMessage):
                    print(f"✅ [{node_name}] {msg.content[:60]}")
                elif isinstance(msg, AIMessage) and msg.content:
                    print(f"\n💬 [{node_name}] {msg.content}")

asyncio.run(run_async_agent())
# Expected output:
# Starting async agent...
#
# ⏳ [agent] fetch_data({'source': 'wikipedia'})
# ⏳ [agent] fetch_data({'source': 'arxiv'})
# ✅ [tools] RAG combines document retrieval with text generation.
# ✅ [tools] RAG was proposed by Lewis et al. in 2020.
#
# 💬 [agent] RAG (Retrieval-Augmented Generation) combines document retrieval with text generation. It was proposed by Lewis et al. in 2020.

Exercise 6: Event collector for a frontend (Hard)

Write a collect_ui_events(agent, question) function that returns a list of structured events (tool_start, tool_end, token, done) ready to send to a frontend.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage

@tool
def search(query: str) -> str:
    """Search for information."""
    return f"Result for {query}: relevant information found."

@tool
def translate(text: str, lang: str) -> str:
    """Translate text into another language."""
    return f"[{lang}] {text}"

def collect_ui_events(agent, question: str) -> list[dict]:
    """Collect the agent's events in a UI-ready format."""
    events = []
    pending = {}

    for step in agent.stream(
        {"messages": [("user", question)]},
        stream_mode="updates"
    ):
        for node_name, update in step.items():
            for msg in update.get("messages", []):
                if isinstance(msg, AIMessage) and msg.tool_calls:
                    for tc in msg.tool_calls:
                        pending[tc["id"]] = tc["name"]
                        events.append({
                            "type": "tool_start",
                            "tool": tc["name"],
                            "args": tc["args"],
                            "id": tc["id"],
                        })
                elif isinstance(msg, ToolMessage):
                    tool_name = pending.pop(msg.tool_call_id, "unknown")
                    events.append({
                        "type": "tool_end",
                        "tool": tool_name,
                        "result": msg.content,
                        "id": msg.tool_call_id,
                    })
                elif isinstance(msg, AIMessage) and msg.content:
                    events.append({
                        "type": "response",
                        "content": msg.content,
                    })

    events.append({"type": "done"})
    return events

model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search, translate])

events = collect_ui_events(agent, "Look up what LangChain is and translate it into Spanish")

for e in events:
    if e["type"] == "tool_start":
        print(f"🟡 START: {e['tool']}({e['args']})")
    elif e["type"] == "tool_end":
        print(f"🟢 END: {e['tool']} → {e['result'][:50]}")
    elif e["type"] == "response":
        print(f"💬 RESPONSE: {e['content'][:80]}")
    elif e["type"] == "done":
        print(f"⚫ DONE")

print(f"\nTotal events: {len(events)}")
# Expected output:
# 🟡 START: search({'query': 'LangChain'})
# 🟢 END: search → Result for LangChain: relevant information found
# 🟡 START: translate({'text': '...', 'lang': 'spanish'})
# 🟢 END: translate → [spanish] ...
# 💬 RESPONSE: LangChain is a framework... In Spanish: ...
# ⚫ DONE
#
# Total events: 5

Summary

In this capsule you learned:

• agent.stream() is the interface for watching the agent's process in real time, step by step
• stream_mode="updates" emits only the per-node changes — ideal for showing tool progress
• stream_mode="values" emits the full accumulated state — useful for debugging and logs
• stream_mode="messages" emits individual tokens — perfect for real-time chat interfaces
• To process events, you tell AIMessage with tool_calls (agent decisions) apart from ToolMessage (tool results)
• astream is the async version for web applications and APIs
• The professional pattern collects structured events (tool_start, tool_end, response) ready to consume from a frontend

Next capsule: Structured Output in Agents — how to force the agent to return structured data (Pydantic) instead of free-form text.

Further reading

1. How to stream agent data to the client — Official guide to streaming agents
2. How to stream from a LangGraph agent — Streaming in LangGraph
3. Streaming Conceptual Guide — Conceptual guide to streaming
4. create_agent API Reference — create_agent reference
5. Server-Sent Events with LangChain — SSE for frontends
6. AsyncIO Documentation — For async streaming

Module 3 — LangChain & LangGraph: From Chains to Agents