当前位置：首页 > news >正文

LangChain Agent 执行图可视化：用 Mermaid 自动生成 Agent 的运行时流程图

news 2026/7/15 22:50:24

LangChain Agent 执行图可视化：用 Mermaid 自动生成 Agent 的运行时流程图

一、深度引言与场景痛点

调试 LangChain Agent 最让人头疼的事，不是它报错，而是它不报错却给了错误答案。你看不到它在每一步调用了什么工具、传了什么参数、工具返回了什么、又是怎么选到最终答案的。日志打了一地，但你得一行行翻，效率极低。

理想的情况是：每次 Agent 跑完，自动生成一张执行流程图，把工具调用链路、决策分支、耗时标注都画出来。一眼就能看出哪一步重试了三回、哪一步超时了、哪一步的结果被忽略掉了。

这件事用 Mermaid 来做，出乎意料地顺手。LangChain 的on_tool_start、on_tool_end、on_agent_action这些回调已经记录了足够多的运行时信息，我们只需要把它们转换成 Mermaid 语法就行了。

二、底层机制与原理深度剖析

flowchart LR A[Agent 启动] --> B[LLM 推理] B --> C{需要调用工具?} C -->|是| D[记录工具名和参数] C -->|否| G[生成最终答案] D --> E[执行工具] E --> F[记录耗时和返回] F --> B G --> H[回调收集器导出 Mermaid] H --> I[生成流程图]

这个流程的关键是回调收集器。每次工具调用开始，记录一个节点和入边；工具返回时，记录出边和耗时。最终把节点列表和边列表拼成 Mermaid 语法，写入文件。整个过程不侵入 Agent 主逻辑，纯粹靠回调完成。

这里有个容易踩的坑：LangChain 的AgentAction和AgentFinish是两个不同的事件类型。工具调用的开始和结束是靠on_tool_start/on_tool_end捕获的，而 LLM 推理的开始和结束要靠on_llm_start/on_llm_end。两种事件都要监听，否则图是不完整的。

另一个要点是循环处理的标注。Agent 经常会有一个"思考-调用-再思考"的循环。如果同一个工具被多次调用，每次都要生成新节点而不是覆盖旧节点。节点的命名规则建议用{工具名}_{序号}，序号从 1 开始递增。这样图里能清楚看到某个工具被调用了 3 次。

三、生产级代码实现

from __future__ import annotations import time import uuid from dataclasses import dataclass, field from typing import Any, Optional from langchain.callbacks.base import BaseCallbackHandler @dataclass class StepNode: id: str label: str step_type: str # "llm", "tool", "output" tool_name: str = "" args: str = "" result_preview: str = "" duration_ms: float = 0.0 error: str = "" class MermaidTraceCallback(BaseCallbackHandler): """自动收集 Agent 执行轨迹并导出 Mermaid 流程图""" def __init__(self, max_steps: int = 50): self.nodes: list[StepNode] = [] self.edges: list[tuple[str, str, str]] = [] # (from_id, to_id, label) self._pending: dict[str, StepNode] = {} self._step_counter: dict[str, int] = {} self._last_node_id: Optional[str] = None self._max_steps = max_steps self._start_time = time.monotonic() def on_llm_start(self, serialized: dict, prompts: list[str], **kwargs: Any) -> None: if len(self.nodes) >= self._max_steps: return node_id = f"llm_{uuid.uuid4().hex[:6]}" node = StepNode(id=node_id, label=f"LLM推理", step_type="llm") node.args = prompts[0][:200] if prompts else "" self.nodes.append(node) if self._last_node_id: self.edges.append((self._last_node_id, node_id, "")) self._last_node_id = node_id def on_llm_end(self, response, **kwargs: Any) -> None: if self._last_node_id: last = self.nodes[-1] last.duration_ms = (time.monotonic() - self._start_time) * 1000 def on_tool_start( self, serialized: dict, input_str: str, *, run_id: str, **kwargs: Any ) -> None: if len(self.nodes) >= self._max_steps: return tool_name = serialized.get("name", "unknown_tool") count = self._step_counter.get(tool_name, 0) + 1 self._step_counter[tool_name] = count node_id = f"tool_{tool_name}_{count}" node = StepNode( id=node_id, label=f"调用: {tool_name}", step_type="tool", tool_name=tool_name, args=input_str[:200], ) self.nodes.append(node) self._pending[run_id] = node if self._last_node_id: self.edges.append((self._last_node_id, node_id, "调用工具")) def on_tool_end(self, output: str, *, run_id: str, **kwargs: Any) -> None: node = self._pending.pop(run_id, None) if node is None: return node.result_preview = output[:200] node.error = "" self._last_node_id = node.id def on_tool_error(self, error: BaseException, *, run_id: str, **kwargs: Any) -> None: node = self._pending.pop(run_id, None) if node: node.error = str(error)[:200] self._last_node_id = node.id def on_agent_finish(self, finish, **kwargs: Any) -> None: if len(self.nodes) >= self._max_steps: return node_id = f"output_{uuid.uuid4().hex[:6]}" node = StepNode( id=node_id, label="最终回答", step_type="output", result_preview=str(finish.return_values.get("output", ""))[:200], ) self.nodes.append(node) if self._last_node_id: self.edges.append((self._last_node_id, node_id, "汇总")) def to_mermaid(self) -> str: lines = ["flowchart TD"] safe_id = lambda nid: nid.replace("-", "_").replace(" ", "") for node in self.nodes: shape = { "llm": "[\\\"{}\\\"]", "tool": "[({})]", "output": "[({})]", }.get(node.step_type, "[({})]") display = node.label[:40] lines.append(f" {safe_id(node.id)}{shape.format(display)}") for from_id, to_id, label in self.edges: edge_label = f"|{label[:15]}|" if label else "" lines.append(f" {safe_id(from_id)} -->{edge_label} {safe_id(to_id)}") return "\n".join(lines) def save_mermaid(self, filepath: str) -> None: with open(filepath, "w", encoding="utf-8") as f: f.write(self.to_mermaid())

这个实现有几个细节值得展开说。第一，_max_steps防止 Agent 无限循环时图也无限增长，超过 50 步直接停止收集。第二，_step_counter用tool_name做 key 给同名工具递增编号，解决循环调用时的节点重名问题。第三，on_tool_error单独处理工具失败的情况——失败的调用在图里也应显示，标注红色或加错误提示，方便排查。

to_mermaid方法里做了节点 ID 的清理。Mermaid 的节点 ID 不能有连字符和空格，所以用replace统一处理。工具节点的形状用圆角矩形[()]，LLM 节点用梯形[/\]，输出节点用圆角框([\)])，视觉上自然区分三类操作。

四、边界分析与架构权衡

这个方案最大的问题不是技术上的，而是信息量的取舍。Agent 跑一次可能产生上百个事件，全部画在图里，图会变成蜘蛛网。所以max_steps不只是一个保护参数，它其实是信息压缩的决策点：哪些步骤值得画、哪些可以合并。

一个实用的补充策略是：在最终图中只画 LLM 推理和工具调用两层，tool 内部的子步骤不再展开。如果某个工具内部耗时异常，再用单独的诊断图深入。

另一个边界是 Mermaid 的渲染能力。超过 30 个节点时，默认的 flowchart 布局可能很乱。这时候建议切换到flowchart LR（左到右）或stateDiagram-v2，后者对树形布局支持更好。也可以在图生成后手动调整subgraph分组，把推理性节点和执行性节点分别框起来。

性能方面，回调本身是同步执行的，如果to_mermaid和save_mermaid在 Agent 主流程里做，会增加几十毫秒的响应延迟。建议用asyncio.to_thread或一个后台任务异步写文件，不影响 Agent 的正常响应。