Implementing Streaming and Async in Python LangChain

python langchain streaming and async requires a clear understanding of the core syntax, runtime behavior, and practical implementation patterns demonstrated in the examples below.

When building LLM applications with Python LangChain, you often need to stream tokens as they are generated and run calls concurrently. This article explains how to use streaming and async together to build responsive, efficient pipelines.

Understanding Streaming and Async in LangChain

Streaming and async solve different problems. Streaming delivers partial results as they become available, so the user sees tokens appear in real time. Async allows multiple operations to overlap without blocking the event loop, which is essential when you are making many LLM calls or waiting on external services.

In LangChain, these two features are independent but often used together. A model can support streaming, async, or both. The API exposes separate methods for each combination: stream for synchronous streaming, astream for asynchronous streaming, ainvoke for asynchronous single calls, and invoke for synchronous calls. The exact method names have changed across versions, but the pattern remains consistent.

Streaming Tokens from an LLM

To stream tokens synchronously, call stream on a model and iterate over the result. Each chunk contains a piece of the output, typically a token or a small group of tokens.

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")

for chunk in model.stream("Write a short poem about async programming"):
    print(chunk.content, end="", flush=True)

The stream method returns a generator that yields chunks as they arrive. This works well in a script or a synchronous web framework where you can send each chunk to the client immediately.

Making Async Calls to LangChain Models

For a single async call, use ainvoke. This returns a full response once the model finishes, but it does not block the event loop while waiting.

import asyncio
from langchain_openai import ChatOpenAI

async def main():
    model = ChatOpenAI(model="gpt-4o-mini")
    response = await model.ainvoke("Explain async/await in Python")
    print(response.content)

asyncio.run(main())

ainvoke is the preferred method in recent LangChain versions. Older code may use arun, but that method is deprecated and should be replaced with ainvoke where possible.

Combining Streaming and Async

When you need both streaming and non-blocking behavior, use astream. This returns an async iterator, which you consume with async for. The loop yields chunks as they are produced, while the event loop remains free to handle other tasks.

import asyncio
from langchain_openai import ChatOpenAI

async def main():
    model = ChatOpenAI(model="gpt-4o-mini")
    async for chunk in model.astream("Write a short story about a robot learning to code"):
        print(chunk.content, end="", flush=True)

asyncio.run(main())

This pattern is ideal for web applications that serve responses over WebSocket or Server-Sent Events. The server can push each chunk to the client as soon as it arrives, without holding a thread or blocking other requests.

Handling Errors and Cancellation

Async streaming introduces two failure modes: errors from the model and cancellation by the client. Both need explicit handling.

Wrap the async loop in a try/except block to catch provider errors, network timeouts, or rate limits. The exception type depends on the provider, but LangChain wraps many errors in a common LangChainException or a provider-specific subclass.

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.exceptions import LangChainException

async def stream_with_error_handling():
    model = ChatOpenAI(model="gpt-4o-mini")
    try:
        async for chunk in model.astream("Generate a long response"):
            print(chunk.content, end="", flush=True)
    except LangChainException as e:
        print(f"\nStream failed: {e}")
    except asyncio.CancelledError:
        # Client disconnected or task was cancelled
        print("\nStream cancelled")
        raise

When a client disconnects, the async generator may raise CancelledError. You should clean up any resources, such as closing a database connection or releasing a semaphore, but you must re-raise the exception so the cancellation propagates correctly.

Performance and Concurrency Considerations

Using asyncio.gather lets you run multiple async LLM calls concurrently. This is useful when you need to summarize several documents or generate multiple suggestions in parallel.

import asyncio
from langchain_openai import ChatOpenAI

async def generate_summary(text):
    model = ChatOpenAI(model="gpt-4o-mini")
    response = await model.ainvoke(f"Summarize: {text}")
    return response.content

async def main():
    texts = ["First document...", "Second document...", "Third document..."]
    results = await asyncio.gather(*(generate_summary(t) for t in texts))
    print(results)

asyncio.run(main())

Be aware of rate limits. Running many concurrent calls may hit provider limits, so consider using a semaphore to cap the number of simultaneous requests. The exact limit depends on your provider and plan, so you need to tune it based on your usage.

Streaming and async also affect memory usage. Streaming reduces memory pressure because you process chunks incrementally rather than holding the entire response in memory. Async reduces thread usage because you avoid blocking a thread per request, which is critical in high-concurrency web servers.

Choosing Between Streaming and Async

Streaming and async are not mutually exclusive, but you may need to choose one depending on your use case.

Use synchronous streaming (stream) when you are in a simple script or a synchronous framework like Flask and you only need to show progress to the user. Use async streaming (astream) when you are in an async framework like FastAPI or when you need to handle multiple clients concurrently. Use ainvoke when you only need the final result and want to avoid the overhead of streaming, but still want non-blocking behavior.

For a web endpoint that returns a complete JSON response, ainvoke is sufficient. For an endpoint that streams tokens to a chat interface, astream is the right choice. The decision depends on whether the client expects incremental updates or a single final payload.

Cancellation and Resource Cleanup in Async Streaming

When an async stream is cancelled, the generator is closed. You can use finally blocks to release resources such as file handles, database connections, or custom rate-limiters. The async for loop automatically calls aclose() on the async iterator when it exits normally or due to an exception, but you should still handle explicit cancellation carefully.

async def stream_with_cleanup():
    model = ChatOpenAI(model="gpt-4o-mini")
    try:
        async for chunk in model.astream("Long text"):
            yield chunk.content
    finally:
        # Release any resources here
        pass

In a FastAPI endpoint that streams responses, the framework will cancel the generator when the client disconnects. Your finally block runs, allowing you to log the cancellation or clean up state. This prevents resource leaks in long-running applications.