AI Agent 开发中的 5 个 Python 调试技巧 | 避坑备忘录
做 Agent 开发一年多，踩了不少坑。这篇把 5 个最高频的 Python 调试问题写下来，每个都附代码和解决思路。
---
1. 大模型 JSON 输出不稳定的问题
场景： 让 LLM 输出 JSON 格式的 tool call 参数，但它偶尔会在 JSON 前后加 markdown 包裹，或者用单引号代替双引号。
典型报错：
import json
response = llm("给出JSON格式结果")
data = json.loads(response)
# json.decoder.JSONDecodeError: Expecting property name enclosed in double quotes
问题根源： LLM 不是 JSON 序列化器。它可能输出 {key: 'value'}（单引号）、`json {...}`（markdown块）、或者末尾多了逗号 [1, 2, 3,]。
解法：一个容错的 JSON 解析器
import re
import json
def safe_json_parse(text: str) -> dict | list | None:
"""从 LLM 输出中安全提取 JSON 对象"""
if not text:
return None
# 1. 去掉 markdown 代码块包裹
text = text.strip()
if text.startswith("```"):
# 去掉 ```json 或 ``` 开头
text = re.sub(r'^```(?:json|python|)\s*', '', text)
text = re.sub(r'\s*```$', '', text)
# 2. 尝试标准解析
try:
return json.loads(text)
except json.JSONDecodeError:
pass
# 3. 修复单引号问题: 把单引号替换成双引号（但要注意内部字符串里的单引号）
# 先替换被单引号包裹的 key 和 string value
text = re.sub(r"(?<!\\)'([^']*)'(?=\s*:)", r'"\1"', text) # key
text = re.sub(r"(?<=:)\s*'([^']*)'(?=\s*[,}\]])", r'"\1"', text) # value
try:
return json.loads(text)
except json.JSONDecodeError:
pass
# 4. 去掉末尾多余逗号（最后一个元素后的逗号）
text = re.sub(r',\s*\}', '}', text)
text = re.sub(r',\s*\]', ']', text)
try:
return json.loads(text)
except json.JSONDecodeError:
return None # 实在不行就返回 None，外面做兜底
使用方式：
llm_output = '```json\n{\n "action": "search",\n "query": "AI Agent",\n}\n```'
data = safe_json_parse(llm_output)
assert data["action"] == "search" # 正常解析
注意： 这段代码对于嵌套复杂 JSON 或包含大量转义字符的场景会失效。如果发现 LLM 持续输出格式错误，应该：
- 在 system prompt 里加强约束
- 使用 structured output（如果模型支持，如 gpt-4o-mini 的
response_format={"type": "json_object"}） - 用 function calling 代替自由文本输出
---
2. 流式输出：实时展示 Agent 思考过程
场景： Agent 调用工具（tool call）时用户看不到中间过程，以为卡住了。需要把工具调用的中间结果实时推给前端。
错误做法： 等全部完成后一次性返回
正确做法： 用生成器把每一步推出来
import time
from typing import Generator
def agent_stream(question: str) -> Generator[dict, None, dict]:
"""流式 Agent，逐步返回状态"""
yield {"type": "status", "content": "正在理解问题..."}
time.sleep(0.3) # 模拟思考
yield {"type": "thinking", "content": f"用户问: {question}"}
# Step 1: 决定工具
yield {"type": "tool_call", "tool": "search_db", "args": {"query": question}}
time.sleep(0.5) # 模拟工具调用
# Step 2: 工具返回
result = {"products": ["智能网关E100", "边缘计算模块"]}
yield {"type": "tool_result", "content": f"查到 {len(result['products'])} 个结果"}
# Step 3: 生成最终回答
yield {"type": "text", "content": "根据查询结果，你需要的产品如下："}
for p in result["products"]:
yield {"type": "text", "content": f"- {p}"}
time.sleep(0.2)
return {"status": "done", "total_found": len(result["products"])}
# FastAPI 里的用法
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json
app = FastAPI()
@app.get("/chat/{question}")
async def chat(question: str):
async def event_stream():
for chunk in agent_stream(question):
yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
return StreamingResponse(event_stream(), media_type="text/event-stream")
前端（简化版）：
const eventSource = new EventSource(`/chat/${encodeURIComponent(question)}`);
const outputDiv = document.getElementById('output');
eventSource.onmessage = (event) => {
const data = JSON.parse(event.data);
if (data.type === 'thinking' || data.type === 'text') {
outputDiv.innerHTML += `<p>${data.content}</p>`;
} else if (data.type === 'tool_call') {
outputDiv.innerHTML += `<p class="tool">🔧 调用工具: ${data.tool}</p>`;
} else if (data.type === 'tool_result') {
outputDiv.innerHTML += `<p class="result">📊 ${data.content}</p>`;
}
};
效果： 用户能看到 Agent "正在思考→调工具→出结果" 的完整路径，交互感和信任度都会提升。
---
3. 工具调用的超时和重试
场景： Agent 调用外部 API 时，API 可能超时或返回 500。默认的 requests.get() 会一直挂起，导致 Agent 卡死。
问题代码：
import requests
def search_web(query: str):
# 如果百度超时，这个调用会挂住整个 Agent
resp = requests.get(f"https://www.baidu.com/s?wd={query}")
return resp.text
正确做法：带超时和重试的包装器
import time
import requests
from functools import wraps
from typing import Any
def with_retry(
max_retries: int = 3,
timeout: int = 10,
backoff: float = 1.0,
retryable_exceptions: tuple = (requests.Timeout, requests.ConnectionError, ConnectionError)
):
"""装饰器: 自动重试可恢复的错误"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs) -> dict:
last_error = None
for attempt in range(max_retries):
try:
result = func(*args, **kwargs, timeout=timeout)
# 检查 HTTP 状态码
if hasattr(result, 'status_code'):
if result.status_code < 500:
return {"success": True, "data": result.text[:2000]}
else:
raise RuntimeError(f"HTTP {result.status_code}")
return {"success": True, "data": result}
except (requests.Timeout, requests.ConnectionError, RuntimeError) as e:
last_error = str(e)
if attempt < max_retries - 1:
wait = backoff * (2 ** attempt) # 指数退避
time.sleep(wait)
return {"success": False, "error": last_error}
return wrapper
return decorator
# 使用
@with_retry(max_retries=3, timeout=8)
def search_web(query: str, **kwargs):
return requests.get(f"https://www.baidu.com/s?wd={query}", **kwargs)
# Agent 调用时
result = search_web("边缘AI网关")
if result["success"]:
print(f"查询成功: {result['data'][:100]}")
else:
# 告诉 LLM 失败了，让它走 fallback
print(f"查询失败: {result['error']}")
关键点：
timeout参数必须传（默认的不传等于不设超时）- 指数退避：第1次等1秒，第2次等2秒，第3次等4秒
- 返回字典而不是直接抛异常，方便 Agent 判断下一步
---
4. 日志链路追踪
场景： Agent 调用链很长，3 个工具调用嵌套，出错了不知道问题出在哪一步。
错误做法： 每一行都 print()，跑了 200 行日志找不到有用的信息。
正确做法： 每条日志带上链路 ID 和层级标记
import uuid
import datetime
class AgentTracer:
"""简单的 Agent 调用链路追踪器"""
def __init__(self):
self.session_id = str(uuid.uuid4())[:8]
self._depth = 0
def log(self, level: str, message: str, **context):
ts = datetime.datetime.now().strftime("%H:%M:%S.%f")[:-3]
indent = " " * self._depth
ctx_str = f" | {context}" if context else ""
print(f"[{ts}][{self.session_id}] {indent}{level.upper()}: {message}{ctx_str}")
def trace(self, name: str):
"""上下文管理器，自动管理缩进"""
class TraceContext:
def __init__(self, tracer):
self.tracer = tracer
self.name = name
def __enter__(self):
self.tracer.log("ENTER", f"▶ {self.name}")
self.tracer._depth += 1
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.tracer._depth -= 1
if exc_type:
self.tracer.log("ERROR", f"✗ {self.name}: {exc_val}")
else:
self.tracer.log("EXIT", f"◀ {self.name} 完成")
return False # 不吞异常
return TraceContext(self)
# 使用
tracer = AgentTracer()
def process_question(question: str):
tracer.log("INFO", f"收到问题: {question}")
with tracer.trace("理解意图"):
intent = "查询产品"
tracer.log("DEBUG", f"识别到意图: {intent}")
with tracer.trace("调用数据库"):
with tracer.trace("连接数据库"):
db_result = {"data": ["产品A", "产品B"]}
tracer.log("INFO", f"查到 {len(db_result['data'])} 条记录")
with tracer.trace("格式化结果"):
formatted = ", ".join(db_result["data"])
tracer.log("DEBUG", f"格式化后: {formatted}")
tracer.log("INFO", "处理完成")
return formatted
process_question("有哪些产品")
输出效果：
[10:30:15.234][a1b2c3d4] INFO: 收到问题: 有哪些产品
[10:30:15.234][a1b2c3d4] ENTER: ▶ 理解意图
[10:30:15.235][a1b2c3d4] DEBUG: 识别到意图: 查询产品
[10:30:15.235][a1b2c3d4] EXIT: ◀ 理解意图 完成
[10:30:15.235][a1b2c3d4] ENTER: ▶ 调用数据库
[10:30:15.235][a1b2c3d4] ENTER: ▶ 连接数据库
[10:30:15.236][a1b2c3d4] INFO: 查到 2 条记录
[10:30:15.236][a1b2c3d4] EXIT: ◀ 连接数据库 完成
[10:30:15.236][a1b2c3d4] ENTER: ▶ 格式化结果
[10:30:15.236][a1b2c3d4] DEBUG: 格式化后: 产品A, 产品B
[10:30:15.236][a1b2c3d4] EXIT: ◀ 格式化结果 完成
[10:30:15.237][a1b2c3d4] EXIT: ◀ 调用数据库 完成
[10:30:15.237][a1b2c3d4] INFO: 处理完成
缩进和 session_id 一出来，多 Agent 协作时谁调了谁、哪一步异常，一眼就能定位。
---
5. Agent 上下文的 Token 控制
场景： Agent 多次调用工具，历史消息越来越长，最终超出上下文窗口限制。
错误做法： 不处理，直到报错 context_length_exceeded
正确做法： 主动管理上下文窗口
from typing import List
class TokenBudget:
"""Token 预算管理器"""
def __init__(self, max_tokens: int = 128000, reserve_output: int = 4000):
self.max_tokens = max_tokens
self.reserve_output = reserve_output
self.system_tokens = 0
def count_tokens(self, text: str) -> int:
"""估算 token 数量（中文按 1.5 字/token）"""
char_count = len(text)
# 中文占比不敏感时，按约 0.75 字/token 估算
return int(char_count * 1.3)
def trim_messages(self, messages: list, max_tokens: int = None) -> list:
"""裁剪消息列表，保证不超过预算"""
if max_tokens is None:
max_tokens = self.max_tokens - self.reserve_output
# 计算总 token
total = 0
for m in messages:
total += self.count_tokens(str(m.get("content", "")))
if total <= max_tokens:
return messages
# 保留 system prompt + 最近的几条消息
system_msgs = [m for m in messages if m.get("role") == "system"]
history_msgs = [m for m in messages if m.get("role") != "system"]
# 从最早的消息开始丢弃
trimmed = []
trimmed_total = sum(self.count_tokens(str(m.get("content", ""))) for m in system_msgs)
# 保留最近的 5 轮对话
recent = history_msgs[-10:] # 保留最近 10 条
for m in recent:
trimmed_total += self.count_tokens(str(m.get("content", "")))
trimmed = system_msgs + recent
# 如果还是超，继续紧缩
while trimmed_total > max_tokens and len(trimmed) > 1:
# 删掉最早的 assistant 消息
removed = False
for i, m in enumerate(trimmed):
if m.get("role") in ("assistant", "user") and i > 0: # 保留 system
trimmed_total -= self.count_tokens(str(m.get("content", "")))
trimmed.pop(i)
removed = True
break
if not removed:
break
return trimmed
def summarize_middle(self, messages: list) -> list:
"""把中间的历史消息压缩成摘要，保留首尾"""
if len(messages) <= 6:
return messages
system_msgs = [m for m in messages if m.get("role") == "system"]
# 找到第一条 user 和最后三条消息
all_msgs = [m for m in messages if m.get("role") != "system"]
if len(all_msgs) <= 4:
return messages
first = all_msgs[:2] # 最近的用户问题
last = all_msgs[-2:] # 最近的助手回答
middle = all_msgs[2:-2] # 中间的历史
summary_content = f"[省略 {len(middle)} 条历史对话，包含 {sum(len(str(m.get('content',''))))} 字符]"
summary_msg = {"role": "system", "content": f"<summary>{summary_content}</summary>"}
return system_msgs + first + [summary_msg] + last
# 使用示例
budget = TokenBudget(max_tokens=128000)
def advanced_chat(conversation_history: list, new_question: str):
# 加入新消息
conversation_history.append({"role": "user", "content": new_question})
# 裁减超出预算的历史
trimmed = budget.trim_messages(conversation_history)
# 或者用摘要压缩
compressed = budget.summarize_middle(conversation_history)
# 调用 LLM
response = llm(messages=trimmed)
conversation_history.append({"role": "assistant", "content": response})
return response
这两个策略的区别：
| 策略 | 适用场景 | 副作用 |
|---|---|---|
| trim_messages | 对话轮次不多但单条消息很长 | 丢掉历史信息 |
| summarize_middle | 长达数十轮的多轮对话 | 摘要可能丢失细节 |
一般先用 trim_messages，如果发现频繁丢消息导致 Agent 失忆，换成 summarize_middle。
---
总结
这 5 个技巧都是日常开发中每天都会碰到的：
| 问题 | 解法 | 优先级 |
|---|---|---|
| JSON 格式不稳定 | 容错解析器 | ★★★ 必须加 |
| Agent 卡住没反馈 | 流式输出 | ★★★ 用户体验 |
| 外部调用超时 | 重试+超时装饰器 | ★★★ 稳定运行 |
| 多 Agent 调试困难 | 链路追踪日志 | ★★ 开发用 |
| Token 超限 | 上下文窗口管理 | ★★ 长对话场景 |
代码片段可以直接复制到项目里用。有问题欢迎在评论区讨论。
---
这篇是实际踩坑记录，不是理论分析。如果你也有类似的坑，欢迎补充。