跳到主要内容

调试与故障排查

调试 AI Agent 应用比调试传统软件更具挑战性。Agent 的执行路径不确定，涉及多轮 LLM 调用和工具执行，出问题时往往难以定位。本章将介绍一套完整的调试方法和常见问题的解决方案。

为什么 Agent 调试困难​

Agent 应用具有以下特点，使得调试变得复杂：

  • 不确定性：同样的输入可能产生不同的执行路径
  • 多步骤执行：问题可能出现在任何一步
  • 外部依赖：LLM API、工具调用都可能失败
  • 状态管理：状态在多个节点间传递，难以追踪
  • 异步执行：并行工具调用增加了复杂性

有效的调试策略需要能够回答这些问题：

  • Agent 当前在哪一步？
  • 状态中的数据是什么？
  • 为什么 Agent 做出了某个决策？
  • 工具调用是否成功？返回了什么？

LangGraph 内置调试功能​

启用调试模式​

LangGraph 提供了 debug 参数来启用详细日志：

from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """搜索信息"""
    return f"搜索结果：{query}"

# 启用调试模式
agent = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=[search],
    debug=True  # 启用调试输出
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "搜索 Python 教程"}]
})

启用调试模式后，LangGraph 会输出详细的执行日志，包括：

  • 每个节点的输入和输出
  • 状态变更
  • 工具调用详情

使用 stream 查看执行过程​

stream 方法是调试的核心工具，它让你能看到 Agent 的每一步执行：

def debug_agent(agent, user_input: str):
    """调试 Agent 执行过程"""
    print(f"=== 用户输入 ===\n{user_input}\n")
    
    for i, chunk in enumerate(agent.stream({
        "messages": [{"role": "user", "content": user_input}]
    }), 1):
        print(f"--- 步骤 {i} ---")
        
        if "agent" in chunk:
            for msg in chunk["agent"]["messages"]:
                print(f"[Agent 节点]")
                if hasattr(msg, "tool_calls") and msg.tool_calls:
                    for tc in msg.tool_calls:
                        print(f"  工具调用: {tc['name']}")
                        print(f"  参数: {tc['args']}")
                elif msg.content:
                    print(f"  响应: {msg.content[:200]}...")
        
        elif "tools" in chunk:
            for msg in chunk["tools"]["messages"]:
                print(f"[Tools 节点]")
                print(f"  工具结果: {msg.content[:200]}...")
    
    print("=== 执行完成 ===\n")

# 使用示例
debug_agent(agent, "帮我搜索 Python 教程，然后计算 2 + 2")

获取状态快照​

使用 get_state 方法查看当前状态：

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
agent = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=[search],
    checkpointer=checkpointer
)

config = {"configurable": {"thread_id": "debug-session"}}

# 执行
agent.invoke({
    "messages": [{"role": "user", "content": "你好"}]
}, config=config)

# 获取状态快照
snapshot = agent.get_state(config)

print(f"当前值: {snapshot.values}")
print(f"下一步: {snapshot.next}")
print(f"配置: {snapshot.config}")
print(f"创建时间: {snapshot.created_at}")
print(f"父状态: {snapshot.parent_config}")

查看执行历史​

使用 get_state_history 查看完整的状态历史：

def print_state_history(agent, config):
    """打印状态历史"""
    print("=== 状态历史 ===\n")
    
    for i, state in enumerate(agent.get_state_history(config)):
        print(f"状态 {i}:")
        print(f"  下一步: {state.next}")
        print(f"  消息数: {len(state.values.get('messages', []))}")
        
        if state.values.get("messages"):
            last_msg = state.values["messages"][-1]
            print(f"  最后消息类型: {type(last_msg).__name__}")
            if hasattr(last_msg, "content"):
                print(f"  内容预览: {str(last_msg.content)[:100]}...")
        print()

# 使用示例
agent.invoke({
    "messages": [{"role": "user", "content": "搜索 Python"}]
}, config=config)

print_state_history(agent, config)

LangSmith 追踪​

LangSmith 是调试 Agent 的强大工具，提供了可视化追踪界面。

配置 LangSmith​

import os

# 启用 LangSmith 追踪
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "your-api-key"
os.environ["LANGSMITH_PROJECT"] = "debug-project"

# 之后所有 Agent 执行都会自动记录

查看追踪信息​

在 LangSmith 界面中，你可以查看：

  • 完整执行链路：每个节点的输入输出
  • Token 消耗：每步调用的 token 数量
  • 延迟分析：每个步骤的耗时
  • 错误堆栈：失败时的详细错误信息

添加自定义标签​

from langchain_core.tracers.context import collect_runs

# 添加自定义标签便于追踪
with collect_runs() as cb:
    result = agent.invoke({
        "messages": [{"role": "user", "content": "测试"}]
    })
    
    # 为这次运行添加标签
    run_id = cb.traced_runs[-1].id
    # 在 LangSmith 界面中可以通过标签筛选

常见问题排查​

1. Agent 陷入循环​

症状：Agent 不断重复调用同一个工具，或重复相同的响应。

原因：

  • 工具返回的信息不明确，Agent 无法判断任务完成
  • 提示词缺少明确的终止条件
  • 模型参数（如 temperature）设置不当

解决方案：

# 方案一：设置递归限制
from langgraph.prebuilt import create_react_agent

agent = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=[...],
)

# 修改递归限制（默认 25）
agent.step_timeout = 60  # 设置超时（秒）

# 方案二：在自定义状态中添加步骤计数
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from langgraph.prebuilt.chat_agent_executor import AgentState

class LimitedState(AgentState):
    step_count: int = 0
    max_steps: int = 10

# 方案三：优化提示词，明确终止条件
agent = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=[...],
    prompt="""你是一个助手。请注意：

1. 如果任务已完成，直接给出答案，不要再调用工具
2. 如果工具无法解决问题，说明原因，不要重复尝试
3. 最多尝试 3 次工具调用

在给出最终答案前，先确认任务是否真正完成。"""
)

2. 工具调用参数错误​

症状：工具调用时参数格式不正确或缺少必要参数。

原因：

  • 工具的参数描述不够清晰
  • 参数类型不匹配
  • 模型对参数理解有偏差

解决方案：

from langchain_core.tools import tool
from pydantic import BaseModel, Field, field_validator
import re

# 方案一：使用 Pydantic 进行严格验证
class EmailInput(BaseModel):
    to: str = Field(description="收件人邮箱地址")
    subject: str = Field(description="邮件主题，不超过 100 字符")
    body: str = Field(description="邮件正文")
    
    @field_validator('to')
    @classmethod
    def validate_email(cls, v):
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(pattern, v):
            raise ValueError(f"'{v}' 不是有效的邮箱地址格式")
        return v
    
    @field_validator('subject')
    @classmethod
    def validate_subject(cls, v):
        if len(v) > 100:
            raise ValueError("邮件主题不能超过 100 个字符")
        return v

@tool(args_schema=EmailInput)
def send_email(to: str, subject: str, body: str) -> str:
    """发送电子邮件。
    
    使用示例：
    - to: "[email protected]"
    - subject: "会议通知"
    - body: "请参加明天下午的会议"
    """
    # 发送邮件逻辑
    return f"邮件已发送至 {to}"

# 方案二：在工具描述中添加示例
@tool
def query_database(sql: str) -> str:
    """执行 SQL 查询。
    
    参数说明：
    - sql: SELECT 语句，只支持查询操作
    
    正确示例：
    - "SELECT * FROM users WHERE id = 1"
    - "SELECT name, email FROM customers LIMIT 10"
    
    错误示例（不支持）：
    - "INSERT INTO users VALUES (...)"
    - "UPDATE users SET ..."
    """
    if not sql.strip().upper().startswith("SELECT"):
        return "错误：只支持 SELECT 查询"
    # 执行查询
    return "查询结果..."

3. 上下文过长​

症状：对话进行一段时间后出现 token 超限错误，或响应变得不连贯。

原因：

  • 对话历史累积过多
  • 大量工具调用结果被保存
  • 未做消息裁剪或总结

解决方案：

from langchain_core.messages import trim_messages
from langchain_openai import ChatOpenAI

# 方案一：使用 trim_messages 裁剪消息
def trim_conversation(messages: list, max_tokens: int = 4000):
    """裁剪对话历史"""
    return trim_messages(
        messages,
        max_tokens=max_tokens,
        strategy="last",  # 保留最后的消息
        token_counter=ChatOpenAI(model="gpt-4o-mini").get_num_tokens,
        start_on="human",  # 从人类消息开始
        end_on=("human", "tool"),  # 在人类或工具消息结束
    )

# 方案二：定期总结历史
from langchain_openai import ChatOpenAI

async def summarize_history(messages: list, keep_recent: int = 5):
    """总结历史消息"""
    if len(messages) <= keep_recent:
        return messages
    
    llm = ChatOpenAI(model="gpt-4o-mini")
    old_messages = messages[:-keep_recent]
    recent_messages = messages[-keep_recent:]
    
    summary_prompt = f"""请总结以下对话的关键信息：

{chr(10).join([f"{m.type}: {m.content[:200]}" for m in old_messages])}

请用简洁的语言总结关键事实和上下文。"""
    
    summary = await llm.ainvoke([{"role": "user", "content": summary_prompt}])
    
    from langchain_core.messages import SystemMessage
    return [
        SystemMessage(content=f"[历史摘要] {summary.content}"),
        *recent_messages
    ]

# 方案三：在 Agent 中集成自动裁剪
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

agent = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=[...],
    checkpointer=checkpointer,
    # 可以在自定义状态中跟踪消息数量
)

4. 工具执行超时​

症状：工具调用长时间无响应，最终超时失败。

原因：

  • 外部 API 响应慢
  • 网络问题
  • 工具内部逻辑卡住

解决方案：

import asyncio
from functools import wraps
from langchain_core.tools import tool

# 方案一：添加超时控制
def with_timeout(timeout_seconds: int):
    """超时装饰器"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            try:
                return await asyncio.wait_for(
                    func(*args, **kwargs),
                    timeout=timeout_seconds
                )
            except asyncio.TimeoutError:
                return f"操作超时（{timeout_seconds}秒），请稍后重试"
        return wrapper
    return decorator

@tool
@with_timeout(30)
async def slow_api_call(query: str) -> str:
    """调用可能较慢的 API"""
    import aiohttp
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.example.com/search?q={query}") as resp:
            return await resp.text()

# 方案二：添加重试机制
from tenacity import retry, stop_after_attempt, wait_exponential

@tool
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def reliable_api_call(query: str) -> str:
    """带重试的 API 调用"""
    import requests
    response = requests.get(f"https://api.example.com/search?q={query}", timeout=10)
    response.raise_for_status()
    return response.text

# 方案三：使用 ToolNode 的超时设置
from langgraph.prebuilt import ToolNode

tool_node = ToolNode([slow_api_call, reliable_api_call])

# 在 Agent 中设置步骤超时
agent = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=tool_node,
)
agent.step_timeout = 60  # 整个步骤最多 60 秒

5. 状态管理问题​

症状：多轮对话中状态丢失或不一致。

原因：

  • 未正确使用 checkpointer
  • thread_id 配置错误
  • 状态序列化问题

解决方案：

from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import create_react_agent

# 正确的状态管理
checkpointer = MemorySaver()

agent = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=[...],
    checkpointer=checkpointer
)

# 每个会话使用唯一的 thread_id
def get_thread_id(user_id: str, conversation_id: str) -> str:
    """生成唯一的线程 ID"""
    return f"{user_id}:{conversation_id}"

config = {
    "configurable": {
        "thread_id": get_thread_id("user-123", "conv-001")
    }
}

# 验证状态是否正确保存
result = agent.invoke({
    "messages": [{"role": "user", "content": "你好"}]
}, config=config)

# 检查状态
state = agent.get_state(config)
print(f"消息数量: {len(state.values.get('messages', []))}")

6. 结构化输出失败​

症状：使用 response_format 时返回格式不符合预期。

原因：

  • 模型不支持结构化输出
  • Schema 定义有问题
  • 任务本身不适合结构化输出

解决方案：

from pydantic import BaseModel, Field
from typing import List
from langgraph.prebuilt import create_react_agent

# 正确的结构化输出定义
class AnalysisResult(BaseModel):
    """分析结果结构"""
    summary: str = Field(description="内容摘要，不超过 100 字")
    key_points: List[str] = Field(
        description="关键点列表，每点不超过 50 字",
        max_length=5  # 限制数量
    )
    sentiment: str = Field(
        description="情感倾向：positive、neutral 或 negative"
    )

# 确保模型支持结构化输出
agent = create_react_agent(
    model="openai:gpt-4o-mini",  # 确保使用支持结构化输出的模型
    tools=[...],
    response_format=AnalysisResult
)

# 添加验证和容错
def get_structured_output(agent, user_input: str):
    """获取结构化输出，带容错处理"""
    result = agent.invoke({
        "messages": [{"role": "user", "content": user_input}]
    })
    
    if "structured_response" in result:
        return result["structured_response"]
    else:
        # 如果结构化输出失败，尝试解析文本响应
        last_msg = result["messages"][-1].content
        try:
            import json
            return AnalysisResult(**json.loads(last_msg))
        except:
            # 返回默认值
            return AnalysisResult(
                summary="解析失败",
                key_points=["无法解析"],
                sentiment="neutral"
            )

调试技巧总结​

日志策略​

import logging
from datetime import datetime

# 配置详细日志
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("agent-debug")

def log_agent_step(step_name: str, data: dict):
    """记录 Agent 步骤"""
    logger.info(f"[{step_name}] {datetime.now().isoformat()}")
    logger.debug(f"数据: {data}")

断点调试​

def debug_node(state):
    """调试用的节点"""
    print("=== 断点 ===")
    print(f"当前状态: {state}")
    print(f"消息数量: {len(state.get('messages', []))}")
    
    # 可以在这里设置断点（使用 IDE 调试功能）
    import pdb; pdb.set_trace()
    
    return state

性能分析​

import time
from functools import wraps

def measure_time(func):
    """测量函数执行时间"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} 执行时间: {end - start:.2f}秒")
        return result
    return wrapper

@measure_time
def run_agent(agent, user_input: str):
    return agent.invoke({
        "messages": [{"role": "user", "content": user_input}]
    })

错误代码参考​

LangGraph 定义了以下错误代码：

错误代码说明解决方案
GRAPH_RECURSION_LIMIT超过递归限制检查 Agent 是否陷入循环
INVALID_CONCURRENT_GRAPH_UPDATE并发更新冲突检查并行节点状态更新
INVALID_GRAPH_NODE_RETURN_VALUE节点返回值无效确保节点返回字典格式
MULTIPLE_SUBGRAPHS子图冲突检查子图配置
INVALID_CHAT_HISTORY聊天历史格式错误验证消息格式

小结​

调试 AI Agent 需要系统的方法：

  • 使用 debug=True 启用详细日志
  • 使用 stream() 查看每步执行
  • 使用 get_state() 和 get_state_history() 追踪状态
  • 配置 LangSmith 进行可视化追踪
  • 了解常见问题的解决方案
  • 建立完善的日志和监控体系

有效的调试能大幅提高开发效率，帮助快速定位和解决问题。

参考资料​