您的位置: 首页> Python

LangGraph 系列 · 第 7 讲：流式输出和监控调试（让智能体“可观测”）

匿名上传

发布时间:2026-02-05 11:00:02

1. 导读

在前面几讲里，我们已经让智能体“能记住”“会用工具”“会路由”“会和人协作”。但一旦进入真实项目，你一定会遇到两个非常现实的问题：

我怎么知道它现在在干嘛？卡在哪一步了？
它为什么这样决策？用了哪些工具？状态是怎么一步步变化的？

这就需要“可观测性（Observability）”：

流式输出：让用户和开发者实时看到模型在“思考”和“生成”的过程
监控与调试：把每一步状态变化、节点调用、错误堆栈都“摊在阳光下”

本讲，我们就来解决：如何让你的 LangGraph 智能体“可视、可查、好调试”。

2. 本讲你将学会

理解流式输出的几种粒度：
掌握 LangGraph 中常用的流式 API：
学会给智能体加“监控视角”：
学会配合 LangSmith 做云端可视化调试：
能为自己的项目设计一套最小可用的“观测方案”：

3. 为什么一定要“可观测”？

3.1 LLM 工作流的天然不确定性

传统后端服务：你写完业务逻辑、单测通过、日志简单看看就够了。但 LLM + LangGraph 有几个特点：

路径不固定：不同输入 → 走的节点序列可能完全不同
输出不确定：即便同样输入，也不保证完全一致
错误难复现：用户“说了一句什么”，你没记录下来，之后很难重演

如果没有良好的观测能力，你只会听到用户抱怨：

“它有时候很好用，有时候又突然傻掉了”
“刚才那次回答特别好，怎么现在重来一次就不行了？”

而你自己完全复现不了那次执行路径，只能瞎猜。

3.2 三个层次的“看得见”

可以简单理解为三层：

用户视角：
开发视角（过程）：
运维 / 产品视角（全局）：

LangGraph 的流式 API + 事件流 + LangSmith，就分别对应了这几个层次。

4. 实战一：最基础的流式输出（让用户“看到在想什么”）

4.1 同步流式输出：逐 token 打印

假设你已经有一个最简单的图：

from langgraph.graph import StateGraph, START, MessagesState
from langchain.chat_models import init_chat_model

model = init_chat_model(model="deepseek-chat", model_provider="deepseek")


def call_model(state: MessagesState) -> MessagesState:
    response = model.invoke(state["messages"])
    return {"messages": state["messages"] + [response]}


builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")
graph = builder.compile()

如果希望在控制台里看到模型一点点输出，可以使用graph.stream：

config = {"configurable": {"thread_id": "stream-demo"}}

for chunk in graph.stream({"messages": "请用中文解释一下 LangGraph 是什么？"}, config):
    # 每个 chunk 代表一次状态增量更新
    print(chunk)

这已经是最朴素的“流式可见”，但可读性还不太好。真实项目中更常见的做法是：

只关心AI 最新回复那条消息的 content
在前端（如 WebSocket）里，把每个增量 chunk 追加到页面上

5. 实战二：事件流（stream_events）——看清楚“跑了哪些节点”

仅仅看到“文本流”还是不够，调试时我们更关心：

先执行的是哪个节点？
这个节点里调了哪些工具？
每一步状态更新了什么？

LangGraph 提供了事件流（Events Stream），帮助我们查看整条执行的“时间轴”。

5.1 使用graph.stream_events观察执行过程

下面是一个典型用法（伪代码示意）：

from langgraph.graph import StateGraph, START, MessagesState

# 假设你已经有一个复杂一点的 graph
graph = builder.compile()

config = {"configurable": {"thread_id": "debug-1"}}

events = graph.stream_events(
    {"messages": "帮我查一下订单状态，然后发一封邮件给客户"},
    config,
    version="v1",  # 可选：标记当前版本
)

for event in events:
    # event 里包含：
    # - event["event"]：事件类型（node_started / node_finished / tool_started / tool_finished 等）
    # - event["name"]：节点或工具名称
    # - event["data"]：输入输出、错误信息等
    print(event)

你可以按需筛选：

只打印node_started/node_finished，看整体路径
单独观察tool_started/tool_finished，排查某个工具慢的问题
在错误时（error类型事件）打印堆栈和状态快照

5.2 在本地构建一个“简易可视化”

很多人会在本地做一个小工具：

后端使用stream_events把事件推到 WebSocket
前端用一条时间轴 / 节点树，把每次执行的节点路径画出来

这其实就是一个“简单版 LangSmith UI”，但完全在你本地环境里实现。

6. 实战三：配合 LangSmith 做云端可视化调试

如果你不想自己造可视化工具，LangSmith是一个很现实的选择。

它可以帮你：

存储每一次执行的完整轨迹（包括每一步的 prompt / response / 工具调用）
支持在线回放、Diff 对比不同版本
标注“好 / 坏”样本，配合评估器做自动评估

6.1 打开 LangSmith 集成

大致步骤（略化）：

在环境变量里配置：

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_ENDPOINT="https://api.smith.langchain.com"
export LANGCHAIN_API_KEY="your_api_key"
export LANGCHAIN_PROJECT="your_project_name"

像平时一样运行你的 LangGraph 工作流：

result = graph.invoke({"messages": "帮我规划一个周末北京两日游行程"}, config)

打开 LangSmith 控制台，就能看到：

这次调用走过的每一个节点
每一步用到的 LLM、工具、输入输出
整体耗时、Token 数量、成本估算

6.2 如何利用这些信息调优？

常见优化方式：

找瓶颈：
控成本：
提质量：

7. 实战四：状态级流式监控（Events + State Diff）

在更复杂的系统里，你可能想看到的是：

每一步执行前后，State 具体多了 / 少了什么字段
某个键（如plan/tools_used/errors）是如何变化的

典型做法是：

用stream_events订阅事件流
在每个node_finished事件里，拿到当前的state或state_diff
把这些 diff 写入日志系统（如 Elasticsearch、ClickHouse），做后续分析

伪代码示意：

for event in graph.stream_events(input, config):
    if event["event"] == "node_finished":
        node_name = event["name"]
        state_diff = event["data"].get("state_diff", {})
        logger.info("node=%s diff=%s", node_name, state_diff)

这样你可以回答类似问题：