聊一下 LangGraph 的流式打印

发表于 20天前 更新于 20天前

作者 Neo

12~15 分钟 阅读

在把项目从 LangChain 升级到 LangGraph 的过程中，最让人头疼的既不是节点状态（State）的定义，也不是 Redis Checkpointer 的持久化挂载，而是流式打印（Streaming Output）。

明明代码写得毫无 Bug，逻辑边和 HITL 人工中断也都流畅运行，但前端的 SSE（Server-Sent Events）接口就是死活吐不出逐字打字的效果，要么直接白屏卡住，要么等模型全部生成完后一次性“闷爆”出整段文字。甚至在借助 LLM 辅助调试的情况下，折腾几个小时也是常有的事——AI 往往只会带你在外层的 SSE 格式或 astream_events 的逻辑里原地打转。

本文结合实际落地踩坑经验，彻底拆解 LangGraph 流式打印失效的根因与终极解决方案。

一、为什么 LangGraph 的流式这么难调？

在 LangChain 时代，流式输出相对直接，对 Runnable 链调用 .astream() 即可。但在 LangGraph 中，大模型的调用被封装在了各个节点（Node）内部。这种架构的变更带来了两个隐性的“框架契约”：

底层 HTTP 字节流开关：大模型客户端本身必须显式声明开启 Stream 模式，否则底层请求本身就是阻塞式的。
上下文事件总线断裂：LangGraph 的 astream_events 依靠 Python 的 contextvars 在异步协程间传递上下文。如果节点内部在调用 LLM 时断开了上下文透传，外层的事件监听器就成了“聋子”。

二、导致“无流式效果”的三大致命盲点

1. 盲点一：LLM 实例未开启 `streaming=True`

很多开发者在初始化 ChatOpenAI 时，习惯只配置 model_name、api_key 和 temperature：

Python

# ❌ 错误：缺少 streaming=True，底层发起的是一次性阻塞请求
self.llm = ChatOpenAI(
    model_name="qwen2.5:3b",
    openai_api_base="http://localhost:11434/v1",
    temperature=0.1,
).bind_tools(ALL_TOOLS)

只要没写 streaming=True，底层 HTTP 客户端就不会向 Ollama 或 OpenAI 发起 chunk 字节流请求。外层再怎么用 astream_events 监听，也接收不到任何 on_chat_model_stream 事件。

正确做法：

Python

# ✅ 正确：显式开启 streaming=True
self.llm = ChatOpenAI(
    model_name="qwen2.5:3b",
    openai_api_base="http://localhost:11434/v1",
    temperature=0.1,
    streaming=True  # 核心：确保发起字节流请求
).bind_tools(ALL_TOOLS)

2. 盲点二：Node 节点未透传 `config`（最致命）

这是 90% 的开发者（包括大模型推理）最容易忽略的死角。在 LangGraph 节点函数中：

Python

# ❌ 错误：节点没有接收或透传 config
async def _agent_node(self, state: AgentState) -> Dict[str, Any]:
    messages = state.get("messages", [])
    response = await self.llm_with_tools.ainvoke(messages) # 👈 未透传 config
    return {"messages": [response]}

在外层使用 self.app.astream_events(inputs, config=config, version="v2") 时，LangGraph 依赖透传的 config 将图层级的事件监听器挂载到节点内部的 ainvoke 上。一旦节点内部漏传了 config，事件回调链条就会在此处彻底断开！

正确做法：

Python

# ✅ 正确：节点函数接收 config，并在 ainvoke 时显式透传
async def _agent_node(self, state: AgentState, config: RunnableConfig = None) -> Dict[str, Any]:
    messages = state.get("messages", [])
    response = await self.llm_with_tools.ainvoke(messages, config=config) # 👈 关键点
    return {"messages": [response]}

3. 盲点三：混合 Tool Call 时的 Chunk 过滤

当 Agent 绑定了 Tools 时，模型输出的第一个 Response 可能是一个“工具调用指令”（Tool Call），而不是给用户的文本回答。如果不加筛选地推送 chunk，前端极易解析异常或打印出空的 JSON 碎片。

正确做法：

Python

if event_type == "on_chat_model_stream":
    chunk = event["data"]["chunk"]
    text_content = chunk.content if hasattr(chunk, "content") else getattr(chunk, "text", "")

    # 确保只推送非空、且非 ToolCall 的纯文本内容
    if text_content and not getattr(chunk, "tool_call_chunks", None):
        payload = json.dumps({"content": text_content}, ensure_ascii=False)
        yield f"data: {payload}\n\n"

三、整理

在 AI 辅助编码普及的今天，模型往往擅长处理显性的语法糖和单函数逻辑，却极难洞察跨协程上下文传递与底层的隐式约定。

不管是借入Cursor还是Claude-Code或者让在线大模型辅助处理这类隐式问题，都极其浪费资源，要快速解决此类问题，就得依靠架构师与高级研发人员的过往经验了。

未来的软件工程，比拼的不再是谁记住的 API 接口多，而是：

谁能更快在脑海中画出系统的运行全景图。
谁能在 AI 卡壳时，凭经验一枪命中那行藏在深处的关键代码。

这正是高级研发人员与架构师无可替代的硬核价值。

万法归宗

LangGraph AI

许可协议: CC BY-NC 4.0

本文同步发布于个人博客 Neo·元，转载请注明出处。

一、 为什么 LangGraph 的流式这么难调？

二、 导致“无流式效果”的三大致命盲点

1. 盲点一：LLM 实例未开启 streaming=True

2. 盲点二：Node 节点未透传 config（最致命）

3. 盲点三：混合 Tool Call 时的 Chunk 过滤

三、 整理

一、为什么 LangGraph 的流式这么难调？

二、导致“无流式效果”的三大致命盲点

1. 盲点一：LLM 实例未开启 `streaming=True`

2. 盲点二：Node 节点未透传 `config`（最致命）

三、整理