TopolearnUsers expect responses to appear token by token, and servers need async to handle many requests.
LLM calls are slow, so perceived latency matters. Every runnable streams:
chain.stream(input)yields output chunks; for a chain ending inStrOutputParserthey are text fragments.chain.astream(input)is the async generator for async servers (FastAPI, etc.).chain.astream_events(input, version="v2")yields structured events for every step (on_chat_model_stream,on_tool_start,on_retriever_end...), so a UI can show retrieval progress or tool use, not just final tokens.
Streaming only works end to end if every step can stream. Parsers like StrOutputParser and JsonOutputParser stream; a plain function in the middle that needs its whole input blocks until it arrives. Use a generator function (taking and yielding chunks) to keep streaming through custom steps.
Async methods (ainvoke, abatch, astream) let one process handle many concurrent requests without threads. Combine with asyncio.gather to run independent calls in parallel. Don't call blocking invoke inside async code; it stalls the event loop.
For agents and LangGraph graphs, stream(..., stream_mode="updates") yields each step's state update, "messages" yields LLM tokens with metadata, and "values" yields the full state after each step. Multiple modes can be combined in one stream.
Example
import asyncio
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_template("Tell me about {x}")
def make_chain(text): # fake model: replace with init_chat_model(...)
return prompt | GenericFakeChatModel(messages=iter([text])) | StrOutputParser()
for chunk in make_chain("Streaming feels fast").stream({"x": "streams"}):
print(repr(chunk), end=" ") # 'Streaming' ' ' 'feels' ' ' 'fast'
print()
async def main():
chains = [make_chain("Answer one"), make_chain("Answer two")]
results = await asyncio.gather(*(c.ainvoke({"x": i}) for i, c in enumerate(chains)))
print(results) # ['Answer one', 'Answer two']
kinds = {e["event"] async for e in make_chain("hi there").astream_events({"x": 1}, version="v2")}
print("on_chat_model_stream" in kinds) # True
asyncio.run(main())Remember
stream/astreamyield chunks;astream_eventsexposes every step's events- Every step must support streaming for tokens to flow end to end
- Use async methods with asyncio.gather for concurrency in servers
Watch out for
- Calling the synchronous invoke inside an async web handler, which blocks the event loop for every other request.