Agent思维链展示：如何通过可视化调试提升LLM应用开发效率

1次阅读

共计 2109 个字符，预计需要花费 6 分钟才能阅读完成。

开发 LLM Agent 应用时，最让人头疼的就是遇到这样的场景：输入提示词后，Agent 输出了一个明显错误的答案，但你完全不知道它中间经历了什么推理步骤。就像调试一个没有 print 语句的复杂函数，这种黑箱体验让优化工作变得异常困难。

传统调试方式主要有两种：

日志输出 ：在代码中插入大量 print 语句，但需要重启服务才能生效
断点调试 ：对异步处理的 Agent 任务几乎不可行

这两种方式都无法实时展示 Agent 的多步推理过程，而思维链（Chain-of-Thought, CoT）可视化正是解决这个痛点的银弹。

我们的解决方案基于 LangChain 的回调机制，核心架构分为三个部分：

数据采集层 ：通过自定义 Handler 捕获 Agent 的中间状态
处理层 ：对原始数据进行清洗和结构化
展示层 ：使用 Streamlit 构建交互式面板

首先安装依赖（建议 Python 3.10+ 环境）：

pip install langchain streamlit pyvis networkx

然后是实现核心的 CallbackHandler：

from langchain.callbacks.base import BaseCallbackHandler
from typing import Any, Dict, List, Optional
import json

class DebugCallbackHandler(BaseCallbackHandler):
    """捕获 Agent 思维链的自定义回调"""

    def __init__(self):
        self.thought_chain = []
        self.current_step = {}

    def on_llm_start(
        self, 
        serialized: Dict[str, Any], 
        prompts: List[str], 
        **kwargs: Any
    ) -> None:
        """记录 LLM 调用开始"""
        self.current_step = {
            "type": "llm",
            "prompt": prompts[0],  # 取首个 prompt
            "children": []}

    def on_llm_end(self, response: Any, **kwargs: Any) -> None:
        """记录 LLM 返回结果"""
        if hasattr(response, 'generations'):
            self.current_step["output"] = response.generations[0][0].text
            self.thought_chain.append(self.current_step)

    # 同样实现 on_tool_start/on_tool_end 等方法...

在长时间运行的服务中，直接存储所有调试数据会消耗大量内存。我们采用滑动窗口策略：

from collections import deque

MAX_HISTORY = 50  # 最多保留 50 条最新记录

class OptimizedDebugHandler(DebugCallbackHandler):
    def __init__(self):
        self.thought_chain = deque(maxlen=MAX_HISTORY)

    # 其他方法保持不变...

当 QPS 较高时，可以采用随机采样：

import random

SAMPLE_RATE = 0.2  # 20% 采样率

class SampledDebugHandler(DebugCallbackHandler):
    def on_llm_start(self, *args, **kwargs):
        if random.random() > SAMPLE_RATE:
            return
        super().on_llm_start(*args, **kwargs)

与传统日志调试相比，我们的方案具有明显优势：

对比项	传统日志	可视化方案
问题定位时间	15-30 分钟	2- 5 分钟
多跳推理支持	❌	✅
历史回溯	困难	即时
协作调试	文本共享	实时共享

Q：可视化面板卡顿怎么办？

检查是否渲染过多节点（建议超过 100 个节点时启用折叠功能）
使用 WebSocket 替代轮询更新
对大型 JSON 数据进行分块加载

Q：如何过滤敏感信息？

在回调处理器中添加过滤逻辑：

def sanitize_output(text: str) -> str:
    """脱敏处理"""
    sensitive_patterns = [r'\b\d{4}-\d{4}-\d{4}-\d{4}\b',  # 银行卡号
        r'\b\d{3}-\d{2}-\d{4}\b'       # 美国 SSN
    ]
    for pattern in sensitive_patterns:
        text = re.sub(pattern, '[REDACTED]', text)
    return text

完整实现代码已放在 GitHub 仓库：agent-debug-visualizer。你可以直接克隆后运行：