LangGraph Stream
本文最后更新于 2026年7月25日
引言
与langchain类似，langgraph支持以迭代器形式进行流式的输出，传递一个或多个流模式来控制接收的数据，可以实时看到过程不需要整个流程走完。相比langchain的流式文字输出，langgraph的流式是能够看到流程实时的状态。
langgraph的流式，有5种常见模式
- values 每步结束后，输出完整当前状态
- updates 每步结束后，仅输出变化部分
- messages 实时输出LLM输出的每个字
- custom 只输出自定义的消息，比如进度信息等
- debug 输出所有细节可用于调试
langgraph采用stream()/astream()进行流式调用，通过参数stream_mode指定流模式，同时，还支持多模式流（multiple modes），支持将一个列表传入stream_mode
1.values
每步结束后，输出完整当前状态
from typing import TypedDict
from langgraph.constants import START, END
from langgraph.graph import StateGraph
class DemoState(TypedDict):
msg: str
def hello(state: DemoState):
return {'msg': 'hello ' + state['msg']}
def hi(state: DemoState):
return {'msg': 'hi ' + state['msg']}
if __name__ == "__main__":
graph = StateGraph(state_schema=DemoState)
graph.add_node('hello', hello)
graph.add_node('hi', hi)
graph.add_edge(START, 'hello')
graph.add_edge('hello', 'hi')
graph.add_edge('hi', END)
app = graph.compile()
for thunk in app.stream(input=DemoState(msg='xiaomi'), stream_mode='values'):
print(thunk){'msg': 'xiaomi'}
{'msg': 'hello xiaomi'}
{'msg': 'hi hello xiaomi'}2.updates
将上面调用部分换为stream_mode='updates'，每步结束后，仅输出变化部分
for thunk in app.stream(input=DemoState(msg='xiaomi'), stream_mode='updates'):
print(thunk){'hello': {'msg': 'hello xiaomi'}}
{'hi': {'msg': 'hi hello xiaomi'}}3.multiple modes
多模式流（multiple modes），支持将一个列表传入stream_mode，返回的是一个(mode, thunk)元组，mode是模式名称，thunk是传输的数据
for mode, thunk in app.stream(input=DemoState(msg='xiaomi'), stream_mode=['values', 'updates']):
print(f'{mode} => {thunk}')values => {'msg': 'xiaomi'}
updates => {'hello': {'msg': 'hello xiaomi'}}
values => {'msg': 'hello xiaomi'}
updates => {'hi': {'msg': 'hi hello xiaomi'}}
values => {'msg': 'hi hello xiaomi'}"如果文章对您有帮助，可以请作者喝杯咖啡吗？"
微信支付
支付宝
LangGraph Stream
https://blog.liuzijian.com/post/2026/07/25/langgraph-stream/