How to enable streaming in LangChain

Streaming makes a LangChain application show progress while a run is still executing instead of waiting for the final state. It matters when a command-line tool needs live status lines, an API has to flush events to a client, or a chat UI should show model output as it arrives.

Current LangChain agents run on LangGraph, and streaming exposes that runtime as iterators. The stream-mode API returns event chunks when stream() or astream() receives stream_mode, while version="v2" gives every chunk the same type, ns, and data keys.

The smoke-test script uses a small local graph so no provider key or billable model call is needed. Use messages for token chunks from a real chat model, updates for state changes, and custom for progress events written by application code during long work.

1. Open an activated Python project environment.

   Related: How to create a virtual environment for LangChain

2. Install or update LangChain.

   $ python3 -m pip install --upgrade langchain

   LangChain installs LangGraph as the runtime used by current agents and graph streaming. The package requires Python 3.10 or newer.
   Related: How to install LangChain with pip

3. Create a streaming smoke-test script.

   $ cat > stream_langchain.py <<'PY'
   from typing import TypedDict
   
   from langgraph.config import get_stream_writer
   from langgraph.graph import END, START, StateGraph
   
   
   class StreamState(TypedDict):
       prompt: str
       answer: str
   
   
   def draft_answer(state: StreamState) -> dict[str, str]:
       writer = get_stream_writer()
       writer({"status": "reading prompt"})
       writer({"status": "drafting response"})
       writer({"status": "finalizing"})
       return {"answer": f"Streaming enabled for {state['prompt']}"}
   
   
   graph = (
       StateGraph(StreamState)
       .add_node("draft_answer", draft_answer)
       .add_edge(START, "draft_answer")
       .add_edge("draft_answer", END)
       .compile()
   )
   
   for chunk in graph.stream(
       {"prompt": "chat status"},
       stream_mode=["custom", "updates"],
       version="v2",
   ):
       if chunk["type"] == "custom":
           print(f"custom: {chunk['data']['status']}")
       elif chunk["type"] == "updates":
           print(f"updates: {chunk['data']}")
   PY

   get_stream_writer() emits custom stream data while the graph or agent is running. The writer must run inside a LangGraph execution context.

4. Run the streaming smoke test.

   $ python3 stream_langchain.py
   custom: reading prompt
   custom: drafting response
   custom: finalizing
   updates: {'draft_answer': {'answer': 'Streaming enabled for chat status'}}

   The custom chunks arrive before the final updates state. For a provider-backed agent, switch the stream mode to messages when the application needs token chunks from the chat model.

5. Remove the temporary smoke-test script.

   $ rm stream_langchain.py