What is .astream()?
.astream() is the async, streaming counterpart to .invoke(). It returns an async generator that yields output chunks as they are produced — token by token for language models, or node by node for LangGraph graphs — without waiting for the full response to complete. This is the foundation of real-time chat UIs, progress indicators, and server-sent event (SSE) endpoints.
Like all Runnable methods, .astream() works across the entire LangChain ecosystem. Chat models yield AIMessageChunk objects whose .content field contains the token text. LCEL chains yield the output type of the final component in the pipeline. LangGraph graphs yield state updates at each node boundary. You consume the generator with async for chunk in runnable.astream(input), and you can pass the same config argument as .ainvoke() — callbacks attached here fire incrementally as each chunk arrives, which is how you wire up logging or token counting in streaming pipelines.
In production SSE endpoints, the idiomatic pattern is to yield each chunk from an async generator function and wrap it in FastAPI's StreamingResponse with media_type="text/event-stream". Keep error handling tight: if the upstream model raises mid-stream, the generator terminates early and the client receives a partial response. For complex pipelines where you need chunk metadata — which step emitted it, the run ID — prefer .astream_events() over .astream(); it includes event, run_id, name, and data fields on every emission.
Use Cases
- • WebSocket connections
- • Async FastAPI
- • Real-time AI chat
- • Event streams
- • Concurrent streaming
- • Async pipelines
Key Features
- ✓ Async streaming
- ✓ Non-blocking
- ✓ WebSocket compatible
- ✓ Async generator
- ✓ Works in async
- ✓ Resource cleanup
When NOT to Use
Outside async context—use stream(). When you need chunk metadata per step—use astream_events().
Notes
Chunk type varies by component
Chat models yield AIMessageChunk objects; access token text via chunk.content. Output parsers yield incrementally parsed output (strings, partial dicts). Retrievers yield full Document objects in a single batch — they do not stream incrementally, so astream on a retriever gives you one chunk containing all documents.
Conversation memory does not capture streamed output automatically
ConversationBufferMemory and similar classes record input/output pairs via save_context(). With streaming, you must accumulate chunks into a full string first, then call save_context after the loop completes. Calling save_context on each chunk writes dozens of partial entries.
Use astream_events for pipeline introspection
For multi-step chains where you need to know which step emitted a chunk, use .astream_events(input, version="v2") instead. Each event includes {"event": "on_chat_model_stream", "name": "ChatOpenAI", "run_id": "...", "data": {"chunk": ...}} so you can route output from different components to different UI elements.
async for is required — sync for gives you a generator object
Calling stream() and forgetting to use async for does not raise an error immediately; it returns an async generator object and silently produces nothing. If your streaming endpoint returns empty responses, check that you used async for, not for.
Method Signature
async for chunk in runnable.astream(input):
print(chunk)Parameters
| Parameter | Type | Required | Purpose |
|---|---|---|---|
| input | Any | Yes | Input data |
| config | RunnableConfig | None | No | Run configuration: callbacks, tags, metadata |
Return Value
Type:
AsyncIterator[Any]
Description:
Async iterator of chunks
Example Output:
async for chunk in model.astream(msg): ...Code Examples
Basic async streaming
async for chunk in model.astream([HumanMessage(content='Explain LangChain')]):
print(chunk.content, end="", flush=True)FastAPI SSE endpoint
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
app = FastAPI()
model = ChatOpenAI(model="gpt-4o")
@app.get("/stream")
async def stream_answer(question: str):
async def token_generator():
async for chunk in model.astream([HumanMessage(content=question)]):
yield chunk.content
return StreamingResponse(token_generator(), media_type="text/event-stream")Collect full text for memory after streaming
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
model = ChatOpenAI(model="gpt-4o")
full_response = ""
async def stream_and_collect(text: str) -> str:
global full_response
full_response = ""
async for chunk in model.astream([HumanMessage(content=text)]):
full_response += chunk.content
print(chunk.content, end="", flush=True)
print() # newline
return full_response
# Save to memory after collecting full text
# memory.save_context({"input": text}, {"output": full_response})Common Mistakes
❌ for chunk in model.astream(msg): # Missing async
✅ async for chunk in model.astream(msg):
Related LangChain References
Browse the full LangChain API reference index to explore more classes, methods, and decorators, or start with the LangChain introduction tutorial for end-to-end context on building with .astream() and the wider framework.
.astream() FAQ
What does .astream() do in LangChain?
Stream output asynchronously. .astream() is the async, streaming counterpart to .invoke(). It returns an async generator that yields output chunks as they are produced — token by token for language models, or node by node for LangGraph graphs — without waiting for the full response to complete. This is the foundation of real-time chat UIs, progress indicators, and server-sent event (SSE) endpoints. Like all Runnable methods, .astream() works across the entire LangChain ecosystem. Chat models yield AIMess…
Which LangChain classes support .astream()?
.astream() is available on All Runnables. Pin your installed LangChain version and verify the method exists in that release before deploying.
When should I use .astream()?
Use .astream() when your LangChain chains, agents, or pipelines need the behavior described in this guide.
What does .astream() return?
.astream() returns a AsyncIterator[Any]. Async iterator of chunks
Does .astream() have an async equivalent?
.astream() does not have a documented async variant. Avoid .astream() Outside async context—use stream(). When you need chunk metadata per step—use astream_events().
Where can I explore more LangChain API reference pages?
Open the LangChain API reference index on DevShelfHub to browse classes, methods, and decorators, each with runnable examples, parameters, common mistakes, and cross-links.