LangGraph Stream 流式调用
本文最后更新于 2026年7月26日
引言
与langchain类似，langgraph支持以迭代器形式进行流式的输出，传递一个或多个流模式来控制接收的数据，可以实时看到过程不需要整个流程走完。相比langchain的流式文字输出，langgraph的流式是能够看到流程实时的状态。
langgraph的流式，有5种常见模式
- values 每步结束后，输出完整当前状态
- updates 每步结束后，仅输出变化部分
- messages 实时输出LLM输出的每个字
- custom 只输出自定义的消息，比如进度信息等
- debug 输出所有细节可用于调试
langgraph采用stream()/astream()进行流式调用，通过参数stream_mode指定流模式，同时，还支持多模式流（multiple modes），支持将一个列表传入stream_mode
本文中所有stream()可改造为异步的astream()
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'}4.debug
debug会显示全部调试信息，一般用于调试
for thunk in app.stream(input=DemoState(msg='xiaomi'), stream_mode='debug'):
print(thunk){'step': 1, 'timestamp': '2026-07-25T11:05:40.796524+00:00', 'type': 'task', 'payload': {'id': 'a52b3195-d88b-26c3-ff9d-640623b831db', 'name': 'hello', 'input': {'msg': 'xiaomi'}, 'triggers': ('branch:to:hello',)}}
{'step': 1, 'timestamp': '2026-07-25T11:05:40.796651+00:00', 'type': 'task_result', 'payload': {'id': 'a52b3195-d88b-26c3-ff9d-640623b831db', 'name': 'hello', 'error': None, 'result': {'msg': 'hello xiaomi'}, 'interrupts': []}}
{'step': 2, 'timestamp': '2026-07-25T11:05:40.796737+00:00', 'type': 'task', 'payload': {'id': 'd8f523cf-1ade-b67c-805b-b379acbba211', 'name': 'hi', 'input': {'msg': 'hello xiaomi'}, 'triggers': ('branch:to:hi',)}}
{'step': 2, 'timestamp': '2026-07-25T11:05:40.796828+00:00', 'type': 'task_result', 'payload': {'id': 'd8f523cf-1ade-b67c-805b-b379acbba211', 'name': 'hi', 'error': None, 'result': {'msg': 'hi hello xiaomi'}, 'interrupts': []}}5.custom
如果langgraph提供的几种模式不能满足需求，例如一个子节点有若干步骤，在全部完成前State不会变化，外部完全看不到进度，希望实时输出节点内部的细节，就可以自定义流式输出，使用get_stream_writer()访问流写入器并发送自定义数据，调用时，需要设置stream_mode='custom'，如果是multiple modes组合模式下，至少有一个是custom
from typing import TypedDict
from langgraph.config import get_stream_writer
from langgraph.constants import START, END
from langgraph.graph import StateGraph
class DemoState(TypedDict):
msg: str
def hello(state: DemoState):
writer = get_stream_writer()
writer({'msg': 'hello 执行了'})
writer({'msg': 'hello 执行了'})
return {'msg': 'hello ' + state['msg']}
def hi(state: DemoState):
writer = get_stream_writer()
writer({'msg': 'hi 执行了'})
writer({'msg': 'hi 执行了'})
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='custom'):
print(thunk){'msg': 'hello 执行了'}
{'msg': 'hello 执行了'}
{'msg': 'hi 执行了'}
{'msg': 'hi 执行了'}"如果文章对您有帮助，可以请作者喝杯咖啡吗？"
微信支付
支付宝
LangGraph Stream 流式调用
https://blog.liuzijian.com/post/2026/07/25/langgraph-stream/