langgraph-streaming

用途

LangGraph Streaming 用于为 LangGraph 工作流提供实时更新与进度追踪能力，适用于实现进度指示器、Token 流式输出、自定义事件以及实时用户反馈等场景。

主要能力

五种流模式

通过 graph.stream(inputs, stream_mode=[...]) 可选择以下模式：

values：每步之后的完整状态，用于调试与状态检查。
updates：每步之后的状态增量，用于高效的 UI 更新。
messages：LLM Token 及元数据，用于聊天界面与打字指示器。
custom：用户自定义事件，用于进度条与状态更新。
debug：最大信息量，用于开发与故障排查。

自定义事件（StreamWriter）

使用 from langgraph.config import get_stream_writer 获取 writer，在节点内发出自定义进度事件，例如 {"type": "progress", "current": i + 1, "total": len(state["items"]), "status": f"Processing {item}"}，以及完成事件 {"type": "complete", "message": "All items processed"}。消费时可在 stream_mode=["updates", "custom"] 下按 mode 与 chunk 的 type 字段分别处理。

LLM Token 流式输出

使用 stream_mode="messages" 可逐块输出 LLM Token，例如遍历 (message_chunk, metadata) 并打印 message_chunk.content。可按节点过滤（meta["langgraph_node"] == "writer_agent"），也可按标签过滤（在 init_chat_model 中设置 tags=["main_response"]，再检查 meta.get("tags", [])）。

子图流式输出

设置 subgraphs=True 并配合 stream_mode="updates"，可通过 namespace 查看图层级，例如 ()、("child",)、("child", "grandchild")。

同时使用多种模式

可在 astream 中传入 stream_mode=["updates", "custom", "messages"]，并用 match 分别处理 update_ui_state、show_progress、append_to_chat。

非 LangChain LLM 流式输出

在节点内获取 writer，遍历自定义流式客户端的输出，发出 {"type": "llm_token", "content": chunk.text} 事件。

FastAPI SSE 集成

提供 /stream 端点示例：在 event_generator 中异步遍历 graph.astream(request.inputs, stream_mode=["updates", "custom"])，以 data: {json.dumps({'mode': mode, 'data': chunk})}\n\n 格式产出，并返回 StreamingResponse(..., media_type="text/event-stream")。

Python < 3.11 异步

需手动传播配置：在节点函数中接收 config: RunnableConfig 并传给 model.ainvoke；也可显式注入 writer: StreamWriter 后调用 writer({"status": "processing"})。

使用方式

按需选择流模式：多数 UI 推荐 ["updates", "custom"]；Token 流式输出使用 messages 模式并按节点过滤；进度追踪使用 custom 模式配合 get_stream_writer()；复杂工作流启用 subgraphs=True。

适用场景

聊天界面与打字指示器、进度条与状态更新、复杂嵌套工作流的进度追踪与调试、通过 SSE 向浏览器推送实时事件。

限制与常见错误

忘记设置 stream_mode 参数（默认仅 values）。
在 Python < 3.11 中未正确处理异步。
print 未设置 flush=True，导致无法实时显示。
未按节点或标签过滤 messages，输出嘈杂。

技能介绍

用途