10.5 图的可视化与流式调试

工作流一旦有了十几个节点，光靠想象很容易连错线。LangGraph 内置了两件调试武器：把结构画出来，以及把每一次状态更新流出来。结构图回答“哪些格子可能相连”，运行记录回答“这次实际走了哪条路”。两份对照着看，问题会自己跳出来。

---

1. 把图“画”出来

编译后的图对象自带绘图方法。按“要不要额外依赖”分成两档。

零依赖，开箱即用：

print(app.get_graph().draw_mermaid())
print(app.get_graph().print_ascii())   # 需要 pip install grandalf

Mermaid 源码可以贴到 mermaid.live 预览。嵌了子图时加 xray=True，能透视到子图内部：

print(app.get_graph(xray=True).draw_mermaid())

渲染成图片：

png = app.get_graph().draw_mermaid_png()   # 默认走 Mermaid Ink，需联网
with open("my_graph.png", "wb") as f:
    f.write(png)

# Graphviz 路线：系统要先有 dot 命令
# macOS: brew install graphviz    Debian/Ubuntu: sudo apt install graphviz
# pip install pygraphviz
png2 = app.get_graph().draw_png()

出现 Program dot not found in path 时，先确认系统里的 dot 能用，而不是只装了 Python 包。只想看结构，Mermaid 通常就够。

---

2. stream：看每个节点交了什么卷

invoke() 只给你结局。stream() 每跑完一个节点就吐回更新：

config = {"configurable": {"thread_id": "demo-1"}}
for event in graph.stream({"messages": [("user", "帮我查下周一的机票")]}, config):
    print(event)

{'planner': {'messages': [AIMessage(content='我需要先查询航班...')]}}
{'search_flights': {'messages': [ToolMessage(content='发现 3 个航班...')]}}
{'planner': {'messages': [AIMessage(content='为您找到以下航班...')]}}

异步场景（FastAPI）用 await graph.astream(...)，语义相同。

---

3. stream_mode：换不同倍率的镜头

stream_mode	看到什么	适用场景
"updates"	每个节点返回的增量	流程调试，最常用
"values"	每个超步结束后的完整状态	想看整本交接本
"messages"	逐 token 的 (chunk, 元数据)	打字机式聊天 UI
"custom"	节点用 get_stream_writer() 主动写出的数据	长任务进度条
"checkpoints" / "tasks" / "debug"	检查点、任务起止、底层元数据	排运行时问题

for msg_chunk, metadata in graph.stream(inputs, config, stream_mode="messages"):
    print(msg_chunk.content, end="")

给模型打上 nostream 标签，可以让它参与计算，但不把 token 推到消息流里——适合内部裁判模型。

多个模式一起开：v2 统一成 StreamPart

LangGraph 1.1+ 支持 version="v2"，所有模式都变成 {type, ns, data}：

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode=["updates", "custom"],
    version="v2",
):
    if chunk["type"] == "updates":
        for node_name, state in chunk["data"].items():
            print(f"节点 {node_name} 更新：{state}")
    elif chunk["type"] == "custom":
        print(f"进度：{chunk['data']}")

ns 是命名空间路径。根图为空；子图打开 subgraphs=True 后，能看到类似 ("flight_department:<task_id>",) 的前缀。把子 Agent 当作节点加入父图时，它们也是子图——不打开这个开关，父图的 messages 流收不到内部模型的 token。

1.2 起官方更推荐按投影消费的事件流（typed iterators）。v2 还让 invoke 的返回值带上 .value / .interrupts；字典取值和 __interrupt__ 暂时能用，但属于旧习惯。新代码优先走流式事件。

节点内上报进度

from langgraph.config import get_stream_writer

def heavy_node(state: State):
    writer = get_stream_writer()
    writer({"status": "正在查询第 1/10 家航司..."})
    # ... 耗时工作 ...
    writer({"status": "正在查询第 2/10 家航司..."})
    return {"result": "done"}

Python 3.11 以下的异步节点拿不到 get_stream_writer，改为给节点加 writer: StreamWriter 参数。

---

4. astream_events：镜头再推近一档

stream() 的粒度是节点；astream_events() 的粒度是内部事件（on_chat_model_start / on_chat_model_stream / on_tool_start / on_tool_end）：

async for ev in graph.astream_events(inputs, config, version="v2"):
    if ev["event"] == "on_chat_model_stream":
        print(ev["data"]["chunk"].content, end="")

做前端逐字渲染用事件流；做流程对账用 stream(updates)。HITL 审批循环，13 节会用到 stream_events(..., version="v3") 里的 interrupted 事件。

---

5. 调试时按这个顺序查

1. 画图：draw_mermaid()，确认连线符合预期；
2. 流式观察：stream_mode="updates"，看每一步改了哪些字段；
3. 节点内打日志：确认输入不是你以为的那样。

def search_flights_node(state: State):
    print(">>> 进入 search_flights，收到：", state.get("destination"))
    result = query_flights(state["destination"])
    print(">>> 返回：", result)
    return {"flight_results": result}

日志应带上 thread_id、节点名、耗时和结果摘要，不要直接记录密钥或身份证号。模型响应很长时，可以只存长度和追踪 ID，正文放到受控存储。

还有一层容易看错：前端收到节点更新，不等于整张图已经成功结束。只有到达 END，或收到约定的完成事件，才能把任务标成完成。

本知识库暂不接入 LangSmith / Langfuse（理由见 15 节）。无论以后用不用观测平台，这套“结构图 + 增量流 + 节点日志”都是基本功。

📁 本节示例代码：code/examples/05_streaming_debug_demo.py

---

扩展阅读

• 流式输出（stream_mode、v2 StreamPart、子图）：Streaming
• 可视化你的图：How-to：Graph API
• Pregel.stream() API 参考：reference.langchain.com

---

下一节： 程序跑着跑着断电了怎么办？敏感操作谁来批准？Checkpointer 和静态断点先把“停得住、接得上”讲清楚。