Invoke, Stream and Batch

Capsule overview

You already know how to initialize models with init_chat_model and configure their parameters. But so far you've only used one execution mode: invoke(). In practice, a language model can be run in three fundamental ways — each designed for a different scenario.

invoke() sends a message and waits for the complete response. stream() returns the response token by token, like watching ChatGPT "type" in real time. batch() processes multiple inputs in parallel, ideal for when you need to make many calls to the same model. On top of that, each mode has an asynchronous version (ainvoke, astream, abatch) for applications built on async/await.

Understanding when to use each mode is what separates a prototype from a real application. In the module's project — the Multi-Provider Chat with Fallback — streaming will be essential so the user sees responses appear progressively instead of waiting several seconds for the full text.

invoke(): the complete response

invoke() is the simplest mode: you send a message, you wait, and you get the complete response all at once.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

response = model.invoke("What is LangChain?")
print(response.content)
# Expected output: LangChain is an open-source framework for building
# applications with language models...

The flow is sequential: your code blocks until the model finishes generating the entire response. For short questions that's fine — latency is barely 1-2 seconds. For long responses (500+ tokens), the user might wait 5-10 seconds seeing nothing.

invoke() accepts either a plain string or a list of messages (SystemMessage, HumanMessage) — the list format is more common in real applications because it lets you include a system prompt:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage

model = init_chat_model("openai:gpt-4.1-mini")

response = model.invoke([
    SystemMessage(content="You are a Python expert. Answer in 2 sentences max."),
    HumanMessage(content="What is a decorator?")
])
print(response.content)
# Expected output: A decorator is a function that wraps another function
# to extend its behavior without modifying it directly...

AIMessage: the response structure

Every time you call invoke(), stream() or batch(), the model returns an AIMessage object. It isn't just text — it carries valuable metadata.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")
response = model.invoke("What is Docker, in one sentence?")

# 1. content — the response text
print(response.content)
# Expected output: Docker is a container platform that lets you package
# applications together with all their dependencies.

# 2. response_metadata — information from the provider
print(response.response_metadata)
# Expected output:
# {
#     'token_usage': {'completion_tokens': 25, 'prompt_tokens': 15, 'total_tokens': 40},
#     'model_name': 'gpt-4.1-mini',
#     'finish_reason': 'stop',
#     ...
# }

# 3. usage_metadata — token usage (a format standardized by LangChain)
print(response.usage_metadata)
# Expected output:
# {
#     'input_tokens': 15,
#     'output_tokens': 25,
#     'total_tokens': 40
# }

The three key fields of AIMessage

You can reach the tokens with response.usage_metadata['input_tokens'], ['output_tokens'] and ['total_tokens']. You can also check response.response_metadata.get("finish_reason") to learn why the model stopped.

Why does finish_reason matter?

If you see finish_reason: "length", it means your response was truncated and you need to raise max_tokens.

stream(): progressive tokens

stream() returns the response token by token (or in small chunks), letting you show text to the user progressively.

Why streaming?

Imagine a response that takes 5 seconds to generate:

• Without streaming (invoke): The user stares at a blank screen for 5 seconds, then the whole text appears at once.
• With streaming (stream): The user starts seeing text in ~200ms. The experience feels instant even though the total time is the same.

Basic example

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

# stream() returns an iterator of chunks
for chunk in model.stream("Explain what a REST API is in 3 points"):
    print(chunk.content, end="", flush=True)

print()  # Line break at the end
# Expected output (it appears progressively):
# A REST API is:
# 1. A communication interface...
# 2. It uses standard HTTP methods...
# 3. It returns data in JSON format...

Each chunk is an AIMessageChunk — a partial version of AIMessage. The key bits are end="" and flush=True: end="" avoids line breaks between chunks, and flush=True forces immediate printing.

What's inside each chunk?

Each chunk is an AIMessageChunk. The first chunk usually has an empty content (it carries initial metadata). The ones after it bring small fragments of text — sometimes a word, sometimes a couple of characters. You can inspect the chunks with an enumerate:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

for i, chunk in enumerate(model.stream("Say hello")):
    print(f"Chunk {i}: '{chunk.content}'")
    if i > 4:
        break
# Expected output:
# Chunk 0: ''
# Chunk 1: 'Hello'
# Chunk 2: '!'
# Chunk 3: ' How'
# Chunk 4: ' can'

Accumulating chunks to get the full response

If you need both streaming and the final response (with token metadata), accumulate the chunks with the + operator. LangChain merges the AIMessageChunks into a complete message:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

accumulated = None
for chunk in model.stream("Say hello"):
    if accumulated is None:
        accumulated = chunk
    else:
        accumulated = accumulated + chunk

print(f"Content: {accumulated.content}")
print(f"Usage: {accumulated.usage_metadata}")
# Expected output:
# Content: Hello! How can I help you?
# Usage: {'input_tokens': 9, 'output_tokens': 12, 'total_tokens': 21}

This pattern is useful when you want streaming for the user but you also need usage_metadata for cost tracking.

batch(): parallel processing

batch() sends multiple inputs to the model and processes them in parallel. Instead of making N sequential calls, it makes N simultaneous ones.

Use it when you have a list of questions to process, need to generate responses for multiple users, or want to classify/summarize/translate multiple texts in a single operation.

Basic example

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

questions = [
    "What is Docker?",
    "What is Kubernetes?",
    "What is Terraform?",
]

responses = model.batch(questions)

for question, response in zip(questions, responses):
    print(f"Q: {question}")
    print(f"A: {response.content[:80]}...")
    print()
# Expected output:
# Q: What is Docker?
# A: Docker is a container platform that lets you package applications with...
#
# Q: What is Kubernetes?
# A: Kubernetes is a container orchestration system that automates the deploy...
#
# Q: What is Terraform?
# A: Terraform is an infrastructure-as-code (IaC) tool that lets you define...

With 5 inputs, batch() is usually 3-5x faster than a loop of invoke(). With 20+ inputs, the difference is dramatic. In Exercise 3 you'll measure that exact difference.

Controlling concurrency with max_concurrency

By default, batch() fires every request in parallel. If you have a lot of inputs, you can cap concurrency so you don't blow through the provider's rate limit:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

questions = [f"How many inhabitants does country #{i} have?" for i in range(20)]

# Max 5 simultaneous calls
responses = model.batch(
    questions,
    config={"max_concurrency": 5}
)

print(f"Processed: {len(responses)} inputs")
# Expected output: Processed: 20 inputs

Recommended values: 5-10 for development, 3-5 for production with a low rate limit, 10-20 for a high rate limit, 1-2 for local APIs (Ollama).

batch() also accepts lists of messages with SystemMessage and HumanMessage, not just strings — you'll see that in Exercise 5.

Async versions: ainvoke, astream, abatch

Every synchronous mode has an asynchronous equivalent. Use the async versions when your application runs inside an event loop (FastAPI, notebooks, web applications).

In a web application (FastAPI, for instance), while you're waiting on an LLM response, the server should be able to serve other requests. The synchronous versions block the thread — the async ones don't.

ainvoke()

import asyncio
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

async def main():
    response = await model.ainvoke("What is FastAPI?")
    print(response.content)

asyncio.run(main())
# Expected output: FastAPI is a modern, high-performance web framework
# for building APIs with Python...

astream()

import asyncio
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

async def main():
    async for chunk in model.astream("Explain what a WebSocket is in 2 sentences"):
        print(chunk.content, end="", flush=True)
    print()

asyncio.run(main())
# Expected output (progressive):
# A WebSocket is a bidirectional communication protocol...

abatch()

abatch() follows the same logic — it takes a list of inputs and processes them in parallel, asynchronously:

responses = await model.abatch(["What is Redis?", "What is Kafka?"])

astream_events(): semantic events

astream_events() goes beyond token streaming — it emits semantic events describing what's happening during execution. It's especially useful in complex pipelines (chains, agents) where you want to know when each step starts and finishes.

import asyncio
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

async def main():
    async for event in model.astream_events(
        "Say hello in 3 languages",
        version="v2"
    ):
        kind = event["event"]
        if kind == "on_chat_model_start":
            print(">>> Model started")
        elif kind == "on_chat_model_stream":
            print(event["data"]["chunk"].content, end="", flush=True)
        elif kind == "on_chat_model_end":
            print("\n>>> Model finished")
            usage = event["data"]["output"].usage_metadata
            if usage:
                print(f">>> Tokens: {usage}")

asyncio.run(main())
# Expected output:
# >>> Model started
# Hello! (English), ¡Hola! (Spanish), Bonjour! (French)
# >>> Model finished
# >>> Tokens: {'input_tokens': 12, 'output_tokens': 18, 'total_tokens': 30}

The main events are on_chat_model_start, on_chat_model_stream and on_chat_model_end. In more complex pipelines you'll also see on_chain_start/end and on_tool_start/end (Module 2). For a model on its own, astream_events behaves much like astream — its real power shows up when you need to tell which component is producing output.

Comparison: when do you use each mode?

Quick decision rules

• ✅ Is it a script or a data pipeline? → invoke() or batch()
• ✅ Does the user see the response in a UI? → stream()
• ✅ Do you have many inputs to process? → batch()
• ✅ Are you inside an async framework (FastAPI)? → ainvoke() or astream()
• ✅ Do you need to know what's happening inside a pipeline? → astream_events()

Connection to the project

In the Multi-Provider Chat with Fallback (Capsule 08), streaming is a central feature. The user will see tokens appear progressively, and when a provider fails, the system will switch to the next one and start streaming immediately. You'll use usage_metadata to show which provider answered and how many tokens it consumed.

The base pattern will be:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

providers = ["openai:gpt-4.1-mini", "anthropic:claude-haiku-4-20250514"]

def chat_with_fallback(question: str):
    for provider_id in providers:
        try:
            model = init_chat_model(provider_id)
            for chunk in model.stream(question):
                print(chunk.content, end="", flush=True)
            print()
            return
        except Exception:
            print(f"\n[Fallback] {provider_id} failed, trying the next one...")
    print("[Error] Every provider failed")

chat_with_fallback("What is Docker?")

Troubleshooting

Problem 1: stream() shows nothing until the end

Cause: flush=True is missing from print(), or the output is being buffered. Fix:

# Make sure you include end="" and flush=True
for chunk in model.stream("question"):
    print(chunk.content, end="", flush=True)

Problem 2: batch() raises RateLimitError

Cause: Too many simultaneous requests exceed the provider's rate limit. Fix:

responses = model.batch(
    questions,
    config={"max_concurrency": 3}
)

Problem 3: asyncio.run() raises "Event loop already running"

Cause: You're in an environment that already has an event loop (a Jupyter notebook, for instance). Fix: In notebooks, use await directly without asyncio.run():

# In a Jupyter notebook
response = await model.ainvoke("Hello")
print(response.content)

# In regular scripts, do use asyncio.run()

Problem 4: usage_metadata is None

Cause: Some providers or modes don't include usage metadata by default. In streaming, the metadata usually arrives in the last chunk. Fix: Accumulate chunks with + to get the complete metadata:

accumulated = None
for chunk in model.stream("question"):
    accumulated = chunk if accumulated is None else accumulated + chunk
print(accumulated.usage_metadata)

Problem 5: stream() returns empty chunks at the start

Cause: That's normal behavior. The first chunk usually carries initial metadata with no text. Fix: Just ignore it — print(chunk.content, end="") already handles it, because it prints an empty string with no visual effect.

Exercises

Exercise 1: Explore AIMessage (Easy)

Call a model with invoke() and print the three key fields of the response: content, response_metadata and usage_metadata. Calculate the approximate cost, knowing that GPT-4.1 Mini costs $0.40 per million input tokens and $1.60 per million output tokens.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")
response = model.invoke("What is a microservice? Answer in 2 sentences.")

print(f"Content: {response.content}")
print(f"Response metadata: {response.response_metadata}")
print(f"Usage metadata: {response.usage_metadata}")

if response.usage_metadata:
    input_cost = response.usage_metadata["input_tokens"] * 0.40 / 1_000_000
    output_cost = response.usage_metadata["output_tokens"] * 1.60 / 1_000_000
    total_cost = input_cost + output_cost
    print(f"\nEstimated cost: ${total_cost:.6f}")
# Expected output:
# Content: A microservice is an architectural pattern...
# Response metadata: {'token_usage': {...}, 'model_name': 'gpt-4.1-mini', ...}
# Usage metadata: {'input_tokens': 18, 'output_tokens': 45, 'total_tokens': 63}
#
# Estimated cost: $0.000079

Exercise 2: Stream with a counter (Easy)

Use stream() to show the response progressively, and at the end print how many chunks you received plus the full accumulated text.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

chunk_count = 0
full_text = ""

for chunk in model.stream("Name 5 popular programming languages"):
    full_text += chunk.content
    chunk_count += 1
    print(chunk.content, end="", flush=True)

print(f"\n\n--- Stats ---")
print(f"Chunks received: {chunk_count}")
print(f"Total characters: {len(full_text)}")
print(f"Full text: {full_text}")
# Expected output:
# 1. Python
# 2. JavaScript
# 3. Java
# 4. TypeScript
# 5. Go
#
# --- Stats ---
# Chunks received: 35
# Total characters: 85
# Full text: 1. Python...

Exercise 3: batch vs sequential (Medium)

Compare the execution time of processing 5 questions with invoke() in a loop versus with batch(). Print the speedup.

import time
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

questions = [
    "What is Python?",
    "What is JavaScript?",
    "What is Rust?",
    "What is Go?",
    "What is TypeScript?",
]

# Sequential
start = time.time()
sequential_results = []
for q in questions:
    r = model.invoke(q)
    sequential_results.append(r)
time_seq = time.time() - start

# Batch
start = time.time()
batch_results = model.batch(questions)
time_batch = time.time() - start

print(f"Sequential: {time_seq:.2f}s")
print(f"Batch:      {time_batch:.2f}s")
print(f"Speedup:    {time_seq / time_batch:.1f}x")

# Check that both produced responses
for q, r in zip(questions, batch_results):
    print(f"  {q} → {r.content[:50]}...")
# Expected output:
# Sequential: 6.10s
# Batch:      1.80s
# Speedup:    3.4x
#   What is Python? → Python is a high-level programming language...
#   ...

Exercise 4: Streaming with formatting (Medium)

Write a function stream_with_border(question) that shows the streamed response inside a visual box. Print a top line before starting, stream the content, and print a bottom line when it finishes. Include the token count at the end.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

def stream_with_border(question: str):
    print(f"\n{'=' * 60}")
    print(f"Question: {question}")
    print(f"{'-' * 60}")

    accumulated = None
    for chunk in model.stream(question):
        print(chunk.content, end="", flush=True)
        if accumulated is None:
            accumulated = chunk
        else:
            accumulated = accumulated + chunk

    print(f"\n{'-' * 60}")
    if accumulated and accumulated.usage_metadata:
        tokens = accumulated.usage_metadata
        print(f"Tokens: {tokens['input_tokens']} in / {tokens['output_tokens']} out")
    print(f"{'=' * 60}")

stream_with_border("What is a container, in 2 sentences?")
# Expected output:
# ============================================================
# Question: What is a container, in 2 sentences?
# ------------------------------------------------------------
# A container is a lightweight, portable package that holds
# an application and all of its dependencies...
# ------------------------------------------------------------
# Tokens: 15 in / 35 out
# ============================================================

Exercise 5: Batch translator (Medium)

Build a translator that takes a text and a list of target languages, and uses batch() to translate the text into all of them in parallel. Use messages with a system prompt for each translation.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage

model = init_chat_model("openai:gpt-4.1-mini")

def translate_batch(text: str, target_languages: list[str]) -> dict[str, str]:
    """Translates a text into multiple languages in parallel."""
    inputs = [
        [
            SystemMessage(content=f"Translate the following text into {lang}. "
                                  f"Respond ONLY with the translation, no explanations."),
            HumanMessage(content=text)
        ]
        for lang in target_languages
    ]

    responses = model.batch(inputs)

    return {
        lang: response.content
        for lang, response in zip(target_languages, responses)
    }

text = "Artificial intelligence is transforming the world"
languages = ["Spanish", "French", "Japanese", "Portuguese"]

translations = translate_batch(text, languages)
for lang, translation in translations.items():
    print(f"{lang:>12}: {translation}")
# Expected output:
#      Spanish: La inteligencia artificial está transformando el mundo
#       French: L'intelligence artificielle transforme le monde
#     Japanese: 人工知能が世界を変革している
#   Portuguese: A inteligência artificial está transformando o mundo

Exercise 6: async stream with gather (Hard)

Write an async function astream_and_collect(model, question, label) that streams while printing each chunk with a [label] prefix, accumulates the chunks, and returns a dict with text, chunks (the count), and usage. Run 2 questions in parallel with asyncio.gather.

import asyncio
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

async def astream_and_collect(model, question: str, label: str) -> dict:
    accumulated = None
    chunk_count = 0
    async for chunk in model.astream(question):
        if chunk.content:
            print(f"[{label}] {chunk.content}", end="", flush=True)
        chunk_count += 1
        accumulated = chunk if accumulated is None else accumulated + chunk
    print()
    return {
        "text": accumulated.content if accumulated else "",
        "chunks": chunk_count,
        "usage": accumulated.usage_metadata if accumulated else None,
    }

async def main():
    results = await asyncio.gather(
        astream_and_collect(model, "What is Redis?", "Q1"),
        astream_and_collect(model, "What is Kafka?", "Q2"),
    )
    print("\n--- Summary ---")
    for i, r in enumerate(results, 1):
        print(f"Q{i}: {r['chunks']} chunks, {len(r['text'])} chars, usage={r['usage']}")

asyncio.run(main())
# Expected output (interleaved, because they run in parallel):
# [Q1] Redis is an in-memory...[Q2] Kafka is a platform...
#
# --- Summary ---
# Q1: 28 chunks, 120 chars, usage={'input_tokens': 10, 'output_tokens': 35, ...}
# Q2: 32 chunks, 140 chars, usage={'input_tokens': 10, 'output_tokens': 40, ...}

Summary

In this capsule you learned:

• invoke() sends a message and waits for the complete response — ideal for scripts and backend processing
• AIMessage holds three key fields: content (the text), response_metadata (provider info), usage_metadata (tokens used)
• stream() returns the response token by token — essential for chat UIs where user experience matters
• batch() processes multiple inputs in parallel — significantly faster than a loop of invoke()
• The async versions (ainvoke, astream, abatch) are necessary in frameworks like FastAPI
• astream_events() emits semantic events — its real power shows up in pipelines and agents
• Use end="" and flush=True for streaming, and max_concurrency to control batch
• Accumulating chunks with + gives you the complete metadata after streaming

Next capsule: Structured Output — you'll learn to get responses in typed formats (Pydantic, TypedDict, JSON Schema) instead of free-form text, so your code never has to parse strings.

Additional resources

1. Chat Models: invoke, stream, batch - Official documentation for the execution modes
2. How to stream chat model responses - Streaming guide
3. How to stream events from a Runnable - astream_events in detail
4. AIMessage API Reference - Complete AIMessage reference
5. Async Programming with LangChain - async/await guide
6. How to batch calls to a Runnable - Batch processing and parallelism
7. LangChain Runnable Interface - The unified interface that enables invoke/stream/batch
8. OpenAI Streaming Guide - How streaming works at the API level

Module 1 — LangChain & LangGraph: From Chains to Agents