本节学习目标

通过本节课的学习，你将全面掌握：

1. 理解可视化原理：了解LangGraph如何将图结构转换为Mermaid/Graphviz格式
2. 安装可视化工具：配置Graphviz、matplotlib等依赖，为绘图做准备
3. 导出多种格式流程图：生成PNG、SVG、ASCII艺术图等可视化结果
4. 使用在线预览工具：通过Mermaid Live Editor快速查看和分享图结构
5. 配置本地调试日志：设置不同级别的日志输出，追踪工作流执行
6. 实现节点流程追踪：通过回调机制监控每个节点的输入输出和执行时间
7. 运用断点调试技巧：在特定节点暂停、检查状态、逐步执行
8. 排错复杂工作流：分析性能瓶颈、检测死循环、定位状态异常

可视化原理

为什么需要可视化？

随着工作流从3个节点增长到10个、20个甚至更多，单纯靠阅读代码理解图结构变得非常困难。可视化能够：

• 快速理解整体结构：一眼看清节点数量和连接关系
• 发现设计问题：孤立节点、未连接的边、意外的循环等
• 文档化工作流：自动生成的图可以作为设计文档
• 团队协作：用图沟通比用代码描述更直观

LangGraph的可视化架构

LangGraph的可视化基于两个层次：

1. 抽象图表示：编译后的图对象提供了get_graph()方法，返回图的抽象结构（节点、边、条件边）
2. 渲染器：将抽象图转换为具体格式（Mermaid、Graphviz、ASCII）

Mermaid格式简介

Mermaid是一种基于文本的图表绘制工具，无需安装图形软件，只需编写文本即可生成图表。LangGraph默认使用Mermaid。

一个简单的Mermaid图示例：

Graphviz格式

Graphviz是一个更专业的图形渲染引擎，支持更复杂的布局和样式。LangGraph通过draw_png等方法调用Graphviz生成高质量图片。

内置可视化工具安装

基础安装（无需额外依赖）

LangGraph的draw_mermaid和draw_ascii方法不需要任何额外依赖，因为它们是纯Python实现。

# 直接使用，无需安装
from langgraph.graph import StateGraph, START, END

# 构建一个简单图
builder = StateGraph(dict)
builder.add_node("a", lambda s: s)
builder.add_node("b", lambda s: s)
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("b", END)
graph = builder.compile()

# 生成Mermaid代码
mermaid_code = graph.get_graph().draw_mermaid()
print(mermaid_code)

安装图片渲染依赖（可选但推荐）

要生成PNG或SVG图片，需要安装graphviz和matplotlib。

Windows安装

1. 下载Graphviz安装包：https://graphviz.org/download/
2. 运行安装程序，务必勾选“Add Graphviz to system PATH”
3. 验证安装：打开CMD，输入dot -V，应显示版本信息
4. 安装Python包：

pip install graphviz matplotlib

macOS安装

# 使用Homebrew安装Graphviz
brew install graphviz

# 安装Python包
pip install graphviz matplotlib

Linux安装（Ubuntu/Debian）

# 安装Graphviz系统包
sudo apt update
sudo apt install graphviz

# 安装Python包
pip install graphviz matplotlib

验证安装

import graphviz
from matplotlib import pyplot as plt

# 测试graphviz
try:
    dot = graphviz.Digraph()
    dot.node("A", "测试节点")
    dot.render("test", format="png", cleanup=True)
    print("✅ Graphviz安装成功")
except Exception as e:
    print(f"❌ Graphviz安装失败: {e}")

# 测试matplotlib
try:
    plt.plot([1,2,3])
    plt.close()
    print("✅ Matplotlib安装成功")
except:
    print("❌ Matplotlib安装失败")

流程图导出PNG/SVG

基本导出方法

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

# 创建示例图（第4课的文本处理工作流）
class WorkflowState(TypedDict):
    original_text: str
    upper_text: str
    word_count: int

def input_node(state):
    return {"original_text": "Hello world"}

def upper_node(state):
    return {"upper_text": state["original_text"].upper()}

def count_node(state):
    return {"word_count": len(state["upper_text"].split())}

def output_node(state):
    print(f"单词数: {state['word_count']}")
    return {}

builder = StateGraph(WorkflowState)
builder.add_node("input", input_node)
builder.add_node("upper", upper_node)
builder.add_node("count", count_node)
builder.add_node("output", output_node)

builder.add_edge(START, "input")
builder.add_edge("input", "upper")
builder.add_edge("upper", "count")
builder.add_edge("count", "output")
builder.add_edge("output", END)

graph = builder.compile()

# 1. 生成PNG图片文件
png_bytes = graph.get_graph().draw_png()
with open("workflow.png", "wb") as f:
    f.write(png_bytes)
print("PNG已保存为 workflow.png")

# 2. 生成SVG图片文件
svg_bytes = graph.get_graph().draw_svg()
with open("workflow.svg", "w", encoding="utf-8") as f:
    f.write(svg_bytes)
print("SVG已保存为 workflow.svg")

自定义图片样式

可以通过参数调整图片的显示方式：

# 使用draw方法的高级参数
graph.get_graph().draw_png(
    output_file_path="custom.png",  # 直接指定文件路径
    # 以下参数需要graphviz支持
    # prog="dot",  # 布局引擎: dot, neato, fdp, sfdp, circo, twopi
    # format="png",
    # args="-Grankdir=LR"  # 横向布局
)

# 横向布局示例（需Graphviz支持）
# 可以通过设置环境变量或修改参数实现

导出为ASCII艺术图（无需图形库）

# 纯文本图形，适合在终端或日志中快速查看
ascii_art = graph.get_graph().draw_ascii()
print(ascii_art)

输出类似：

+-------+     +-------+     +-------+     +-------+
| input | --> | upper | --> | count | --> | output |
+-------+     +-------+     +-------+     +-------+

导出Mermaid代码

mermaid_code = graph.get_graph().draw_mermaid()
print(mermaid_code)

输出：

批量导出不同格式

def export_all_formats(graph, basename="graph"):
    """导出所有格式的可视化文件"""
    # Mermaid代码
    with open(f"{basename}.mmd", "w") as f:
        f.write(graph.get_graph().draw_mermaid())
    
    # ASCII
    with open(f"{basename}.txt", "w") as f:
        f.write(graph.get_graph().draw_ascii())
    
    # PNG (需要graphviz)
    try:
        png_data = graph.get_graph().draw_png()
        with open(f"{basename}.png", "wb") as f:
            f.write(png_data)
    except Exception as e:
        print(f"PNG导出失败: {e}")
    
    print("导出完成")

export_all_formats(graph, "my_workflow")

在线可视化预览

使用Mermaid Live Editor

Mermaid Live Editor是一个在线工具，可以实时编辑和预览Mermaid图表。

步骤：

1. 打开 https://mermaid.live/
2. 在左侧编辑器中粘贴从draw_mermaid()生成的代码
3. 右侧自动显示流程图
4. 可以导出为PNG、SVG，或分享链接

生成可分享的链接

import urllib.parse

def create_mermaid_link(graph):
    mermaid_code = graph.get_graph().draw_mermaid()
    # 压缩空格，减少URL长度
    compressed = mermaid_code.replace(" ", "").replace("\n", "")
    encoded = urllib.parse.quote(compressed)
    url = f"https://mermaid.live/edit#pako:{encoded}"
    print(f"分享链接: {url}")
    return url

create_mermaid_link(graph)

在Jupyter Notebook中内嵌显示

from IPython.display import Image, display, Markdown

# 方法1：显示PNG（如果已生成）
with open("workflow.png", "rb") as f:
    display(Image(f.read()))

# 方法2：显示Mermaid原生（需要Jupyter扩展）
from IPython.display import display, Markdown
display(Markdown(f"```mermaid\n{mermaid_code}\n```"))

本地调试日志配置

基础日志配置

LangGraph使用Python标准logging模块。通过配置日志级别，可以查看框架内部的执行信息。

import logging

# 配置根日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# 开启LangGraph的详细日志
logging.getLogger("langgraph").setLevel(logging.DEBUG)

执行工作流时，你会看到类似输出：

2025-05-11 10:00:01 - langgraph.graph - INFO - Starting node: input
2025-05-11 10:00:01 - langgraph.graph - DEBUG - State before node: {...}
2025-05-11 10:00:01 - langgraph.graph - INFO - Node input completed

自定义日志级别

节点级别的独立日志

在每个节点内部添加日志，追踪节点执行细节：

import logging
logger = logging.getLogger(__name__)

def process_node(state):
    logger.info(f"进入process_node, 状态键: {state.keys()}")
    try:
        result = do_work(state)
        logger.debug(f"处理结果: {result}")
        logger.info("process_node 成功完成")
        return result
    except Exception as e:
        logger.error(f"process_node 失败: {e}", exc_info=True)
        return {"error": str(e)}

使用日志文件

import logging
from logging.handlers import RotatingFileHandler

# 创建文件Handler
file_handler = RotatingFileHandler(
    "langgraph_debug.log",
    maxBytes=10_485_760,  # 10MB
    backupCount=5
)
file_handler.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
file_handler.setFormatter(formatter)

# 添加到langgraph logger
logger = logging.getLogger("langgraph")
logger.addHandler(file_handler)
logger.setLevel(logging.DEBUG)

节点流程追踪

使用回调追踪执行

LangGraph支持BaseCallbackHandler，可以在节点执行前后插入自定义逻辑。

from langgraph.callbacks import BaseCallbackHandler
from typing import Dict, Any
import time

class TracingCallback(BaseCallbackHandler):
    """追踪每个节点的执行时间和状态变化"""
    
    def __init__(self):
        self.node_times = {}
        self.current_start = None
    
    def on_chain_start(self, serialized: Dict[str, Any], inputs: Any, **kwargs):
        """节点开始执行时调用"""
        node_name = serialized.get("name", "unknown")
        print(f"⏳ 开始执行节点: {node_name}")
        print(f"   输入状态预览: {str(inputs)[:200]}...")
        self.current_start = time.time()
    
    def on_chain_end(self, outputs: Any, **kwargs):
        """节点执行完成时调用"""
        elapsed = time.time() - self.current_start
        node_name = kwargs.get("name", "unknown")
        self.node_times[node_name] = elapsed
        print(f"✅ 节点 {node_name} 完成，耗时 {elapsed:.3f}秒")
        print(f"   输出状态预览: {str(outputs)[:200]}...")
    
    def on_chain_error(self, error: Exception, **kwargs):
        """节点执行出错时调用"""
        node_name = kwargs.get("name", "unknown")
        print(f"❌ 节点 {node_name} 失败: {error}")
    
    def get_summary(self):
        """输出执行摘要"""
        print("\n=== 执行摘要 ===")
        for node, duration in self.node_times.items():
            print(f"  {node}: {duration:.3f}s")
        total = sum(self.node_times.values())
        print(f"  总耗时: {total:.3f}s")

# 使用回调
callback = TracingCallback()
result = graph.invoke(
    initial_state,
    config={"callbacks": [callback]}
)
callback.get_summary()

实时状态可视化

在长时间运行的工作流中，可以定期输出状态快照：

class StateSnapshotCallback(BaseCallbackHandler):
    def __init__(self, interval_nodes=2):
        self.count = 0
        self.interval = interval_nodes
    
    def on_chain_end(self, outputs, **kwargs):
        self.count += 1
        if self.count % self.interval == 0:
            print(f"\n[快照] 第{self.count}个节点完成后状态:")
            # 假设outputs包含完整state（实际需要在回调中捕获）
            print(json.dumps(outputs, indent=2, default=str))

追踪条件边路由决策

通过自定义包装路由函数来追踪路由决策：

def trace_router(router_func):
    """装饰器，打印路由函数的决策"""
    def traced_router(state):
        result = router_func(state)
        print(f"[路由决策] {router_func.__name__} -> {result}")
        return result
    return traced_router

# 在构建图时使用
builder.add_conditional_edges(
    "decision_node",
    trace_router(my_router),
    path_map
)

使用LangSmith集成（高级）

LangSmith是LangChain官方的调试平台，可以记录完整的执行轨迹。需要注册LangSmith账号并设置API密钥。

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "langgraph-debug"

# 之后所有graph.invoke调用会自动上传到LangSmith
graph.invoke(initial_state)

在LangSmith网页上可以看到每个节点的输入输出、耗时、错误等详细信息。

断点调试技巧

使用内置断点（Python breakpoint）

在节点函数中插入breakpoint()，当执行到该节点时会进入Python调试器（pdb）。

def critical_node(state):
    result = complex_calculation(state)
    breakpoint()  # 执行到这里会暂停
    return {"result": result}

运行脚本后，进入pdb交互模式，可以：

• p state 打印当前state
• c 继续执行
• s 单步进入
• n 单步跳过
• q 退出

使用条件断点

只在特定条件下暂停：

def monitored_node(state):
    if state.get("error_count", 0) > 3:
        breakpoint()  # 只在错误计数大于3时暂停
    # 正常逻辑...

通过检查点实现“时间旅行”断点

LangGraph的检查点机制允许我们在执行后回溯到任意节点前的状态。结合持久化，可以实现类似“逆向调试”的效果。

from langgraph.checkpoint import MemorySaver

memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "debug_session"}}
# 执行到某个点
graph.invoke(initial_state, config=config)

# 获取状态历史
history = graph.get_state_history(config)
for i, state_snapshot in enumerate(history):
    print(f"步骤 {i}: 节点={state_snapshot.next}, 状态={state_snapshot.values}")

# 回退到特定历史状态
state_before_node = graph.get_state(config)  # 获取当前状态
# 或者使用update_state手动分支

在VSCode中调试LangGraph

1. 在VSCode中打开项目文件夹
2. 在节点函数中设置断点（点击行号左侧）
3. 创建.vscode/launch.json：

{
    "version": "0.2.0",
    "configurations": [
        {
            "name": "Python: Debug LangGraph",
            "type": "python",
            "request": "launch",
            "program": "${workspaceFolder}/app.py",
            "console": "integratedTerminal",
            "justMyCode": false,
            "env": {
                "PYTHONPATH": "${workspaceFolder}"
            }
        }
    ]
}

4. 按F5启动调试，执行到断点时会暂停，可以查看变量、调用栈、监视表达式。

使用IPython嵌入调试

如果需要更丰富的交互环境（代码补全、魔法命令），可以在节点中嵌入IPython：

from IPython import embed

def interactive_node(state):
    print("进入交互模式，可以手动操作state")
    embed()  # 进入IPython shell
    # 退出后继续执行
    return {}

需要安装：pip install ipython

复杂工作流可视化排错

案例1：检测孤立节点

孤立节点是没有被任何边连接（既没有入边也没有出边）的节点。它永远不会被执行，是一种常见的设计错误。

def detect_orphan_nodes(graph):
    """检测编译后的图中是否有孤立节点"""
    # 获取图的结构
    graph_repr = graph.get_graph()
    # 获取所有节点（包括START和END）
    all_nodes = set(graph_repr.nodes.keys())
    # 获取有边的节点
    nodes_with_edges = set()
    for edge in graph_repr.edges:
        nodes_with_edges.add(edge.source)
        nodes_with_edges.add(edge.target)
    orphans = all_nodes - nodes_with_edges - {"__start__", "__end__"}
    if orphans:
        print(f"⚠️ 发现孤立节点: {orphans}")
    else:
        print("✅ 无孤立节点")
    return orphans

detect_orphan_nodes(graph)

案例2：检测潜在的死循环

虽然LangGraph有递归限制，但提前检测可能无限循环的模式仍有帮助。

def detect_cycles(graph_repr):
    """检测图中是否存在循环（简单的启发式）"""
    # 获取所有边
    edges = [(e.source, e.target) for e in graph_repr.edges if e.source != "__start__" and e.target != "__end__"]
    
    # 构建邻接表
    adj = {}
    for src, tgt in edges:
        adj.setdefault(src, []).append(tgt)
    
    # DFS检测环
    visited = set()
    stack = set()
    
    def has_cycle(node):
        visited.add(node)
        stack.add(node)
        for neighbor in adj.get(node, []):
            if neighbor not in visited:
                if has_cycle(neighbor):
                    return True
            elif neighbor in stack:
                return True
        stack.remove(node)
        return False
    
    cycles = []
    for node in adj:
        if node not in visited:
            if has_cycle(node):
                cycles.append(node)
    
    if cycles:
        print(f"⚠️ 检测到可能的循环，起始于节点: {cycles}")
    else:
        print("✅ 未检测到循环")
    return cycles

注意：LangGraph的循环是合法的，但如果没有退出条件就会无限循环。此检测仅用于提醒。

案例3：性能瓶颈可视化

在回调中记录每个节点的执行时间，然后生成耗时分布图。

import matplotlib.pyplot as plt

class ProfilingCallback(TracingCallback):
    def __init__(self):
        super().__init__()
        self.timeline = []
    
    def on_chain_end(self, outputs, **kwargs):
        elapsed = time.time() - self.current_start
        node_name = kwargs.get("name", "unknown")
        self.timeline.append((node_name, elapsed))
        print(f"{node_name}: {elapsed:.3f}s")
    
    def plot_timeline(self):
        names, times = zip(*self.timeline)
        plt.figure(figsize=(10, 6))
        plt.barh(names, times, color='skyblue')
        plt.xlabel('耗时（秒）')
        plt.title('节点执行时间分布')
        plt.tight_layout()
        plt.savefig('profile.png')
        plt.show()

# 使用
profiler = ProfilingCallback()
graph.invoke(initial_state, config={"callbacks": [profiler]})
profiler.plot_timeline()

案例4：状态突变追踪

当状态在某节点后意外改变，可以记录每个节点后的状态快照，对比变化。

class StateMutationTracker:
    def __init__(self):
        self.history = []
    
    def on_chain_end(self, outputs, **kwargs):
        node_name = kwargs.get("name", "unknown")
        self.history.append((node_name, copy.deepcopy(outputs)))
    
    def diff(self, node_a, node_b):
        """比较两个节点的状态变化"""
        state_a = dict(self.history[node_a][1])
        state_b = dict(self.history[node_b][1])
        added = set(state_b) - set(state_a)
        removed = set(state_a) - set(state_b)
        changed = {k for k in state_a if k in state_b and state_a[k] != state_b[k]}
        print(f"新增字段: {added}")
        print(f"删除字段: {removed}")
        print(f"变更字段: {changed}")
        for k in changed:
            print(f"  {k}: {state_a[k]} -> {state_b[k]}")

案例5：条件边路径覆盖率

测试工作流时，需要确保所有条件分支都被执行过。记录路由函数的决策历史。

class CoverageCallback:
    def __init__(self, router_func):
        self.router = router_func
        self.decisions = []
        self.expected_paths = set()
    
    def on_conditional_edge(self, source, state):
        decision = self.router(state)
        self.decisions.append((source, decision))
        print(f"路径: {source} -> {decision}")
        return decision
    
    def report(self):
        from collections import Counter
        counter = Counter(d for _, d in self.decisions)
        print("路径覆盖统计:")
        for path, count in counter.items():
            print(f"  {path}: {count}次")

本节总结+思考题

核心知识点速查

最佳实践总结

1. 开发阶段：启用logging.DEBUG + 可视化的ASCII图快速检查结构
2. 代码审查：导出PNG或Mermaid嵌入文档
3. 性能优化：使用回调统计节点耗时，找出瓶颈
4. 调试复杂bug：使用检查点+时间旅行，配合VSCode断点
5. 生产环境：配置LangSmith或自定义日志收集，保留错误上下文

课后思考题

基础题：

1. LangGraph可视化生成的Mermaid代码和Graphviz图片各有什么优缺点？什么场景适合用哪种？
2. 如何在不安装任何额外库的情况下快速查看工作流的节点连接关系？
3. 执行graph.get_graph().draw_ascii()时，如果图结构比较复杂，ASCII输出可能混乱。有什么办法改善可读性？

进阶题：

4. 设计一个回调类，能够自动检测工作流中的“停滞”节点（执行时间超过阈值，如5秒），并记录当前状态和调用栈。
5. 实现一个“单步执行”调试器：允许用户一次执行一个节点，每次执行后打印状态变化，并询问是否继续。提示：使用graph.invoke的config中的recursion_limit配合手动状态管理。
6. 解释LangGraph的检查点（Checkpoint）是如何实现“时间旅行”的？它在底层保存了哪些信息？

实践题：

7. 使用本节课的第4课文本处理工作流，生成它的PNG图片和Mermaid代码，并在Mermaid Live Editor中调整样式（如横向布局、节点颜色）。
8. 编写一个脚本，接收一个LangGraph工作流文件路径，自动输出该工作流的节点数量、边数量、是否有循环、节点依赖深度等信息。
9. 在一个有4个以上节点的工作流中，使用ProfilingCallback找出最慢的节点，并尝试优化该节点的代码（如添加缓存、异步IO）。

思考题：

10. 假设你正在调试一个生产环境中的LangGraph服务，该服务每天处理百万级请求，偶尔会出现状态不一致的问题。你无法在本地复现。结合云日志、LangSmith和分布式追踪（如OpenTelemetry），设计一个调试方案，能够定位问题发生在哪个节点以及状态异常的具体内容。

下节课预告

第8节课：大模型接入 —— OpenAI/本地私有化模型适配LangGraph

可视化让我们看清了工作流的结构，但LangGraph的核心价值在于编排大模型。下一节课我们将学习：

• 大模型接入原理：LangGraph如何与LLM交互，标准接口是什么
• OpenAI接口对接实战：配置API密钥、调用GPT-3.5/GPT-4、处理流式响应
• 本地私有化模型适配：使用Ollama运行Llama、通义千问、智谱等本地模型
• 国内主流模型集成：星火、文心一言、通义千问的API接入
• 模型参数配置：temperature、max_tokens、top_p等参数调优
• 多模型切换方案：通过配置文件动态切换不同模型提供商
• 模型调用异常处理：限流、超时、重试、降级策略
• 生产级模型调用封装：带缓存、监控、令牌桶限流的工程实现

课前准备：

• 注册OpenAI账号并获取API密钥（或准备好其他模型的密钥）
• 安装Ollama（可选）：curl -fsSL https://ollama.com/install.sh | sh
• 拉取一个本地模型：ollama pull llama3.2:1b
• 预习langchain_openai和langchain_community的用法

下一节课后，你的LangGraph工作流将真正拥有“大模型大脑”，能够处理自然语言、推理决策、生成内容！

🔗《20节课 LangGraph 从入门到精通》系列课程导航

去订阅

🌟 感谢您耐心阅读到这里！
💡 如果本文对您有所启发欢迎：
👍 点赞📌 收藏 📤 分享给更多需要的伙伴。
🗣️ 期待在评论区看到您的想法, 共同进步。
🔔 关注我，持续获取更多干货内容～
🤗 我们下篇文章见～