强烈推荐收藏！AI Agent 调试从入门到生产：从 print 到 LangFuse 全链路可观测性——4大调试层级+8个真实Bug案例

Agent 开发的噩梦：LLM 选错工具了、循环调了 14 次、Token 烧了 $15 什么都没做出来。更糟的是——你根本不知道它内部发生了什么。print 打出来的是 Token 序列堆，根本读不懂。本文用 4 层调试架构 + LangFuse 全链路追踪 + 8 个真实 Bug 案例，让你从「盲飞」变成「上帝视角」。

---

一、为什么 print 对 Agent 不管用

1.1 传统调试 vs Agent 调试

# 传统程序：print 够用
def calculate(a, b):
    result = a + b
    print(f"a={a}, b={b}, result={result}")  # 清晰
    return result

# Agent：print 完全不够
def agent_loop(user_input):
    response = llm.chat(messages)   # 1000+ Token 的思维链
    print(response)  # 一大坨文本，关键信息淹没在 3000 字里
    
    tool_result = execute(tool_call)  # 工具返回 500 Token
    print(tool_result)  # 又是 500 字
    
    # 问题：你不知道 LLM 为什么选这个工具、花了多少 Token、这一步的耗时

1.2 Agent 调试的四个维度

一个完整的 Agent 调试体系需要覆盖：

┌─────────────────────────────────────────┐
│          Agent 调试四层金字塔              │
├─────────────────────────────────────────┤
│  L4 成本层：Token 用了多少？花了多少钱？     │
│  L3 循环层：Agent 走了几步？有没有死循环？    │
│  L2 工具层：调了什么工具？参数是什么？结果？   │
│  L1 代码层：传统异常、类型错误、变量状态       │
└─────────────────────────────────────────┘

---

二、L1 代码层：结构化日志替代 print

2.1 给 Agent 加有意义的日志

import logging
import time
from functools import wraps

logger = logging.getLogger("agent")

def log_step(step_name: str):
    """装饰器：自动记录每一步的耗时"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            start = time.time()
            try:
                result = await func(*args, **kwargs)
                elapsed = time.time() - start
                logger.info(
                    f"[{step_name}] ✓ {elapsed:.2f}s",
                    extra={"step": step_name, "elapsed": elapsed, "status": "success"}
                )
                return result
            except Exception as e:
                elapsed = time.time() - start
                logger.error(
                    f"[{step_name}] ✗ {elapsed:.2f}s: {e}",
                    extra={"step": step_name, "elapsed": elapsed, "status": "error", "error": str(e)}
                )
                raise
        return wrapper
    return decorator


# 使用
@log_step("LLM推理")
async def llm_think(messages):
    return await client.chat.completions.create(model="gpt-4o", messages=messages)

@log_step("工具执行")
async def execute_tool(name, args):
    func = available_tools[name]
    return await func(**args)

输出效果：

2026-05-20 14:23:01 [LLM推理] ✓ 1.23s
2026-05-20 14:23:03 [工具执行] ✓ 0.45s  → get_weather({"city":"北京"})

---

三、L2 工具层：追踪每一次 Tool Call

3.1 记录完整的工具调用链

from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class ToolTrace:
    tool_name: str
    arguments: dict
    result: str
    start_time: datetime
    end_time: datetime
    success: bool
    error: str = ""

@dataclass
class AgentTrace:
    session_id: str
    steps: list[ToolTrace] = field(default_factory=list)
    total_cost: float = 0.0
    total_tokens: int = 0

class TraceableAgent:
    def __init__(self):
        self.trace = AgentTrace(session_id=generate_session_id())
    
    async def execute_tool(self, tool_call):
        trace = ToolTrace(
            tool_name=tool_call.function.name,
            arguments=json.loads(tool_call.function.arguments),
            start_time=datetime.now(),
            end_time=None,
            success=False,
            result=""
        )
        
        try:
            func = self.tools[trace.tool_name]
            trace.result = await func(**trace.arguments)
            trace.success = True
        except Exception as e:
            trace.success = False
            trace.error = str(e)
            trace.result = f"错误：{e}"
        finally:
            trace.end_time = datetime.now()
        
        self.trace.steps.append(trace)
        
        # 实时输出给人看
        icon = "✓" if trace.success else "✗"
        elapsed = (trace.end_time - trace.start_time).total_seconds()
        print(f"  {icon} {trace.tool_name}({trace.arguments}) → {elapsed:.2f}s")
        
        return trace.result

3.2 可视化工具调用链

def render_agent_trace(trace: AgentTrace):
    """把 Agent 的执行轨迹渲染成可读的文本"""
    print(f"\n{'='*60}")
    print(f"Agent 执行报告 | Session: {trace.session_id}")
    print(f"{'='*60}")
    print(f"总步骤数: {len(trace.steps)}")
    print(f"总 Token:  {trace.total_tokens:,}")
    print(f"总成本:    ${trace.total_cost:.4f}")
    print()
    
    for i, step in enumerate(trace.steps, 1):
        status = "✅" if step.success else "❌"
        print(f"步骤 {i} {status} {step.tool_name}")
        print(f"   参数: {json.dumps(step.arguments, ensure_ascii=False)}")
        print(f"   结果: {step.result[:150]}{'...' if len(step.result)>150 else ''}")
        
        if step.error:
            print(f"   错误: {step.error}")
        
        elapsed = (step.end_time - step.start_time).total_seconds()
        print(f"   耗时: {elapsed:.3f}s")
        print()

---

四、L3 循环层：防止死循环和疯狂烧钱

4.1 循环监控器

class LoopMonitor:
    """监控 Agent 循环，检测异常"""
    
    def __init__(self, max_steps: int = 10, max_cost: float = 1.0):
        self.max_steps = max_steps
        self.max_cost = max_cost
        self.action_history = []
        self.total_cost = 0.0
        self.warnings = []
    
    def record(self, step: int, tool_name: str, args: dict, cost: float):
        self.action_history.append({
            "step": step, "tool": tool_name, "args": args, "cost": cost
        })
        self.total_cost += cost
    
    def check(self, step: int) -> str:
        """返回状态：ok / warn / abort"""
        # 检查 1：步数超限
        if step >= self.max_steps:
            return "abort:steps"
        
        # 检查 2：成本超限
        if self.total_cost > self.max_cost:
            return "abort:cost"
        
        # 检查 3：连续相同动作（死循环检测）
        if len(self.action_history) >= 3:
            recent = self.action_history[-3:]
            if len(set((a["tool"], str(a["args"])) for a in recent)) == 1:
                self.warnings.append(f"步骤 {step}: 连续 3 次调用 {recent[0]['tool']}")
                return "warn:loop"
        
        return "ok"


# 集成到 Agent 中
monitor = LoopMonitor(max_steps=8, max_cost=0.5)

for step in range(max_steps):
    status = monitor.check(step)
    
    if status.startswith("abort"):
        print(f"🛑 {status}")
        messages.append({
            "role": "system",
            "content": "任务已被中止。" + (
                "超过了最大步骤数。" if "steps" in status
                else "Token 成本超过了预算。请立即给出你能给出的最佳答案。"
            )
        })
        break
    
    if status.startswith("warn"):
        print(f"⚠️ {monitor.warnings[-1]}")
        messages.append({
            "role": "system",
            "content": "你连续调用了相同的工具。请尝试不同的方法解决问题。"
        })

4.2 成本实时追踪

# 每次 LLM 调用后更新成本
def estimate_cost(model: str, prompt_tokens: int, completion_tokens: int) -> float:
    PRICES = {
        "gpt-4o":       (0.0025, 0.01),     # 输入/输出 每千Token
        "claude-sonnet": (0.003,  0.015),
        "deepseek-chat": (0.00014, 0.00028),  # ¥/千Token
    }
    input_price, output_price = PRICES.get(model, (0, 0))
    return (prompt_tokens * input_price + completion_tokens * output_price) / 1000

# 每次 LLM 调用后
monitor.record(step, tool_call.function.name, args,
               estimate_cost(model, usage.prompt_tokens, usage.completion_tokens))

---

五、L4 可观测层：LangFuse 全链路追踪

5.1 为什么需要 LangFuse

print → 知道当前值
logging → 知道历史值
LangFuse → 知道"为什么 Agent 这样决策" + "哪个版本的 Agent 最好"

5.2 LangFuse 接入

pip install langfuse openai

from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context

langfuse = Langfuse(
    public_key="pk-lf-...",
    secret_key="sk-lf-...",
    host="https://cloud.langfuse.com"  # 或自部署
)


@observe(as_type="generation")
async def llm_call(messages: list, model: str = "gpt-4o"):
    """LLM 调用 - 自动记录 input/output/tokens/cost"""
    response = await client.chat.completions.create(
        model=model, messages=messages
    )
    return response


@observe()
async def agent_step(step_num: int, user_input: str):
    """Agent 单步 - 自动记录整个步骤的耗时和嵌套调用"""
    langfuse_context.update_current_trace(
        name=f"agent-step-{step_num}",
        metadata={"step": step_num}
    )
    
    # LLM 推理
    response = await llm_call(messages)
    
    # 工具调用
    if response.choices[0].message.tool_calls:
        for tc in response.choices[0].message.tool_calls:
            langfuse_context.update_current_observation(
                metadata={
                    "tool": tc.function.name,
                    "args": tc.function.arguments
                }
            )
            result = await execute_tool(tc)
    
    return response


@observe()
async def agent_session(session_id: str, user_input: str):
    """Agent 完整会话 - 顶层 Trace"""
    langfuse_context.update_current_trace(
        name=f"session-{session_id}",
        user_id="user-123",
        session_id=session_id,
        tags=["production", "agent-v2.1"]
    )
    
    for step in range(MAX_STEPS):
        result = await agent_step(step, user_input)
        if not result.choices[0].message.tool_calls:
            break
    
    # 打分（可选：用于后续评估）
    langfuse.score(name="task_completed", value=1.0)

LangFuse 能看到的：

• 每个 Trace 的完整调用树（LLM→Tool→LLM→Tool→…）
• 每一步的耗时、Token 用量、成本
• 哪个版本 Prompt 效果最好
• 哪些用户遇到了最多的错误

5.3 自部署 LangFuse（无需付费）

# Docker Compose 一键部署
git clone https://github.com/langfuse/langfuse.git
cd langfuse
docker compose up -d
# 访问 http://localhost:3000

---

六、真实 Bug 案例库

Bug 1：LLM 选择了不存在的工具

症状：Agent 返回 "调用 send_sms"，但工具列表里根本没有 send_sms
原因：System Prompt 太长，LLM 混淆了可用工具和示例工具
修复：工具列表不要超过 8 个；如果多，做工具分类路由

# 防御代码
if tool_name not in registered_tools:
    return json.dumps({
        "error": f"工具 {tool_name} 不存在。可用工具：{list(registered_tools.keys())}"
    })

Bug 2：Agent 连续重试同一个失败操作

症状：Agent 连调 14 次 read_file，每次都失败，每次都重试
原因：LLM 对错误信息的理解偏差——它认为是「还没成功」而不是「永远不可能成功」
修复：LoopMonitor 检测 3 次连续相同操作 → 强制注入「请换方法」提示

Bug 3：上下文悄悄溢出

症状：Agent 跑了 20 轮后突然「失忆」，忘记 System Prompt
原因：对话历史 + 工具返回 > Context Window，最前面的 System Prompt 被截断
修复：每 5 轮自动估算 Token 量，超过 80% 阈值时压缩早期对话

threshold = model_context_window * 0.8
if estimate_tokens(messages) > threshold:
    messages = compress_early_messages(messages)

Bug 4：并行工具调用顺序错乱

症状：查了 10 个城市的天气，返回结果和城市对不上
原因：异步并行 + tool_call_id 匹配错误
修复：用 tool_call_id 精确匹配，不要用索引

Bug 5：流式输出中的工具调用被截断

症状：SSE 流中 tool_call 的 arguments JSON 被截断，解析失败
原因：流式传输时，tool_call 的参数可能跨多个 chunk
修复：缓存未完成的 tool_call，等全部到齐后再解析

Bug 6：Token 成本偷偷飙升

症状：月账单从 $60 涨到 $400，代码没改过
原因：新版本 Prompt 多了 500 Token，每天 1000 次调用 = 多 50 万 Token/天
修复：每次改 Prompt 时，估算 Token 增量 × 日均调用量 = 月增量成本

Bug 7：不同模型对 JSON 格式的理解不同

症状：GPT-4 完美输出 JSON，DeepSeek 加了 "```json" 包装
原因：不同模型的输出习惯不同
修复：解析前做格式清洗

import re

def safe_json_parse(text: str) -> dict:
    text = text.strip()
    # 去掉 markdown 代码块包装
    text = re.sub(r'^```(?:json)?\s*', '', text)
    text = re.sub(r'\s*```$', '', text)
    return json.loads(text)

Bug 8：LangFuse 在生产环境的内存泄漏

症状：LangFuse SDK 默认在内存中缓存所有事件，高并发下 OOM
修复：设置 flush 间隔和最大批次大小

langfuse = Langfuse(
    flush_at=5,           # 每 5 个事件 flush 一次
    flush_interval=10,    # 或每 10 秒 flush 一次
    enabled=True
)

---

七、调试 Checklist

□ 代码层
  □ 每个函数入口/出口有结构化日志
  □ 异常被捕获并记录了上下文
  □ 关键变量在日志中可见

□ 工具层
  □ 每个 tool_call 记录了工具名 + 参数 + 结果 + 耗时
  □ 工具不存在时有明确的错误返回
  □ 工具调用之间有依赖关系记录

□ 循环层
  □ max_steps 限制
  □ 成本预算上限
  □ 连续相同操作检测（死循环防护）
  □ Context 溢出监控

□ 可观测层
  □ LangFuse 或 LangSmith 接入
  □ 每次部署有版本标签
  □ 用户反馈有分数回传
  □ Token 用量有日报

---

八、总结

层级	工具	解决什么问题
L1 代码层	结构化日志 + 装饰器	异常定位、变量状态
L2 工具层	ToolTrace + 调用树	工具选对没、参数对没
L3 循环层	LoopMonitor	死循环、成本失控
L4 可观测层	LangFuse	全局视角、A/B 对比、历史回放

Agent 调试和传统程序调试的根本区别：你调试的不是一段代码，而是一个在不确定环境中自主决策的系统。print 能看到代码状态，但看不到 LLM 的决策逻辑。LangFuse 让你同时看到两者。

---

标签：#Agent调试 #LangFuse #可观测性 #AI工程 #生产实践 #程序员必读