流式输出与实时状态反馈

流式输出与实时状态反馈

学习目标

本章要解决什么问题：

• 如何使用 stream() / astream() 获取 Agent 执行的实时状态
• 不同 stream_mode 的选择策略与适用场景
• 如何通过自定义流模式推送节点内的事件
• 同步流与异步流的差异与注意事项

前置知识

本章涉及流式传输的通用原理，建议先阅读：

• 流式传输架构 — 三层流式模型与分块算法

下文假设你已理解上述概念，直接聚焦 LangGraph 的具体实现。

---

项目实践

流模式选择

LangGraph 支持多种流模式，通过 stream_mode 参数控制：

模式	内容	适用场景
updates	每个节点的状态变化	看每步执行了什么、改变了什么
values	完整的当前状态	实时监控整个状态
messages	LLM 生成的 token	实时展示模型输出
custom	节点内自定义事件	推送业务事件
debug	执行生命周期事件	调试和追踪

基本流式输出

app = graph.compile()

# updates 模式：每个节点的变化
for chunk in app.stream({"messages": ["你好"]}, stream_mode="updates"):
    print(chunk)
    # {"model": {"messages": [AIMessage(content="你好！")]}}
    # {"tools": {"messages": [ToolMessage(content="搜索结果...")]}}

实时展示 LLM 输出

for event in app.astream_events(
    {"messages": ["写一首诗"]},
    version="v2"
):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["content"], end="", flush=True)

astream_events 与 LangChain 的事件系统集成，可以捕获 LLM 生成的每个 token。

自定义流模式

在节点内通过 StreamWriter 推送自定义事件：

from langgraph.types import StreamWriter

def my_node(state, writer: StreamWriter):
    writer({
        "type": "progress",
        "step": "analyzing",
        "detail": "正在分析文档..."
    })
    # ... 执行逻辑
    writer({
        "type": "progress",
        "step": "done",
        "detail": "分析完成"
    })
    return {"status": "done"}

# 消费自定义事件
for mode, chunk in app.stream(
    {"data": "input"},
    stream_mode=["updates", "custom"]
):
    if mode == "custom":
        handle_progress(chunk)

多模式并行

for mode, chunk in app.stream(
    {"messages": ["你好"]},
    stream_mode=["updates", "messages", "debug"]
):
    if mode == "updates":
        log_state_change(chunk)
    elif mode == "messages":
        display_token(chunk)
    elif mode == "debug":
        trace_execution(chunk)

问题与规避

同步流不支持所有模式

问题：stream()（同步版本）可能不支持某些流模式（如 messages）。

规避：使用 astream() 或 astream_events() 获取完整流支持。

astream_events 需要正确的事件循环

问题：astream_events 需要在异步上下文中调用。

规避：使用 async for 循环消费事件，或在 Jupyter Notebook 中使用 asyncio.run()。

大量事件导致性能问题

问题：长时间运行的 Agent 可能产生大量事件，消费端处理不过来。

规避：

• 过滤不需要的事件类型
• 使用缓冲或采样策略
• 在节点内使用 StreamWriter 只推送关键事件

设计取舍

优势

• 灵活：多种流模式覆盖不同使用场景
• 与 LangChain 集成：astream_events 复用 LangChain 事件系统
• 自定义：节点内可推送业务事件

代价

• 异步要求：完整流支持需要异步上下文
• 事件量：长时间运行的 Agent 产生大量事件
• 学习成本：多种流模式和事件类型需要理解

替代方案

• 轮询状态：定期调用 get_state() 获取当前状态，适合低频监控
• 回调机制：通过 callback 接口接收通知，适合集成外部系统

参考来源

• 源码验证: libs/langgraph/langgraph/pregel/main.py — stream / astream 方法
• 文档: LangGraph Streaming, https://docs.langchain.com/oss/python/langgraph/streaming