# 5.3 中期评审与问题修复 > **第5周 | 第3课 | 中期评审与问题修复 | 预计时长：30分钟** --- ## 学习目标 - 掌握Agent开发中4类常见问题的诊断和修复方法 - 能够使用系统化的代码审查清单自查项目质量 - 完成中期评审报告，明确剩余工作量和风险点 --- ## 一、常见问题诊断与修复 Agent开发过程中，以下4类问题最为常见。每个问题都配有具体的诊断步骤和修复方案。 ### 问题1：Agent陷入无限循环 **症状：** Agent反复调用同一个工具或反复生成相同内容，不返回最终结果。 **诊断步骤：** ``` Step 1: 检查日志输出 -> 查看 logs/app.log 或终端输出 -> 寻找重复的模式（相同工具调用、相同prompt） Step 2: 检查Agent的停止条件 -> 确认Agent有明确的终止判断 -> 确认工具调用次数有限制 Step 3: 检查工具返回值 -> 工具是否返回了Agent期望的格式？ -> 工具返回空数据时Agent如何处理？ ``` **修复方案：** ```python # 方案A：添加工具调用次数限制 class SafeAgent: MAX_TOOL_CALLS = 10 # 最大工具调用次数 def __init__(self): self.tool_call_count = 0 def execute(self, user_input: str) -> dict: self.tool_call_count = 0 while True: if self.tool_call_count >= self.MAX_TOOL_CALLS: return { "success": False, "error": f"达到最大工具调用次数限制 ({self.MAX_TOOL_CALLS})" } # ... 正常执行逻辑 self.tool_call_count += 1 # 方案B：检测重复调用并提前终止 from collections import Counter class LoopDetector: """循环检测器""" def __init__(self, threshold: int = 3): self.history: list[str] = [] self.threshold = threshold def record(self, action: str): """记录一次行动""" self.history.append(action) def is_looping(self) -> bool: """检测是否陷入循环""" if len(self.history) < self.threshold * 2: return False # 检查最近的行动是否有大量重复 recent = self.history[-self.threshold * 2:] counts = Counter(recent) most_common_count = counts.most_common(1)[0][1] return most_common_count >= self.threshold def reset(self): """重置检测器""" self.history.clear() # 使用示例 detector = LoopDetector(threshold=3) def agent_loop(user_input: str): detector.reset() while True: action = decide_next_action() detector.record(action["type"]) if detector.is_looping(): logger.warning("检测到循环，提前终止") return {"success": False, "error": "检测到重复操作，已终止"} result = execute(action) if result.get("done"): return result ``` **预防措施：** - 始终设置最大迭代次数上限 - 工具返回空结果时，Agent应直接返回而非重试 - 在日志中记录每次工具调用的参数，方便事后排查 --- ### 问题2：工具调用静默失败 **症状：** Agent调用了工具，但没有得到预期结果，也不报错。 **诊断步骤：** ``` Step 1: 确认工具是否真的被调用 -> 在工具函数开头添加 print 或 logger 语句 -> 检查是否有拼写错误导致调用了错误的函数 Step 2: 检查工具返回值 -> 打印工具的原始返回值 -> 确认返回值格式与Agent期望一致 Step 3: 检查异常处理 -> 工具内部是否吞掉了异常？ -> 是否有 try/except 返回了空结果？ ``` **修复方案：** ```python # 方案A：工具包装器 - 统一异常处理和日志 from functools import wraps from src.utils.logger import setup_logger logger = setup_logger("tool_wrapper") def tool_wrapper(tool_name: str): """工具装饰器：统一处理异常和日志""" def decorator(func): @wraps(func) def wrapper(*args, **kwargs): logger.info(f"[工具调用] {tool_name} | 参数: {args}, {kwargs}") try: result = func(*args, **kwargs) logger.info(f"[工具结果] {tool_name} | 成功 | 返回类型: {type(result).__name__}") return result except Exception as e: logger.error(f"[工具失败] {tool_name} | 异常: {type(e).__name__}: {e}") # 返回结构化的错误信息，而不是抛出异常 return { "success": False, "tool": tool_name, "error": f"{type(e).__name__}: {str(e)}", "args": str(args)[:100] # 截断过长的参数 } return wrapper return decorator # 使用方式 @tool_wrapper("search") def search_tool(query: str) -> dict: response = httpx.get(f"https://api.example.com/search?q={query}") response.raise_for_status() return {"success": True, "data": response.json()} # 方案B：工具调用结果验证 def validate_tool_result(result: dict, expected_keys: list[str]) -> bool: """验证工具返回结果是否包含必需字段""" if not isinstance(result, dict): logger.warning(f"工具返回非字典类型: {type(result)}") return False missing = [k for k in expected_keys if k not in result] if missing: logger.warning(f"工具返回结果缺少字段: {missing}") return False return True # 使用方式 result = search_tool("AI Agent") if not validate_tool_result(result, ["success", "data"]): logger.error("搜索结果格式不正确，使用降级策略") # 降级：返回友好提示，而不是崩溃 return {"success": True, "content": "抱歉，暂时无法获取相关信息。"} ``` **预防措施：** - 所有工具函数都使用 `@tool_wrapper` 装饰器 - 工具返回结构化的成功/失败结果 - 在Agent侧检查工具返回格式，不信任工具的输出 --- ### 问题3：记忆无法持久化 **症状：** 重启程序后，之前的对话记录、偏好设置全部丢失。 **诊断步骤：** ``` Step 1: 确认数据是否写入磁盘 -> 检查 data/ 目录下是否有 memory.db 或 chroma_db 文件夹 -> 打开SQLite文件查看数据是否存在 sqlite3 data/memory.db "SELECT count(*) FROM messages;" Step 2: 确认数据是否正确读取 -> 重启后打印读取到的消息数量 -> 对比写入数量和读取数量 Step 3: 检查路径是否正确 -> 打印 config.memory_db_path 确认路径 -> 确认工作目录是否一致（相对路径可能在不同目录下指向不同位置） ``` **修复方案：** ```python # 方案A：使用绝对路径 import os from pathlib import Path # 获取项目根目录的绝对路径 PROJECT_ROOT = Path(__file__).parent.parent # src/ -> 项目根目录 DATA_DIR = PROJECT_ROOT / "data" DATA_DIR.mkdir(exist_ok=True) DB_PATH = str(DATA_DIR / "memory.db") VECTOR_STORE_PATH = str(DATA_DIR / "chroma_db") # 方案B：启动时验证持久化 class PersistenceValidator: """持久化验证器""" def __init__(self, store): self.store = store def verify(self) -> bool: """验证持久化是否正常工作""" import uuid # 写入测试数据 test_id = str(uuid.uuid4()) test_content = "persistence_test_data" try: self.store.store(test_content, {"type": "test"}) # 立即读取验证 results = self.store.search(test_content, n_results=1) if not results: return False # 清理测试数据 self.store.delete(test_id) return True except Exception as e: logger.error(f"持久化验证失败: {e}") return False # 方案C：在应用启动时自动加载历史 class SessionManager: """会话管理器""" def __init__(self, store, memory): self.store = store self.memory = memory self.current_session = None def start_session(self, session_id: str = None, topic: str = ""): """开始新会话或恢复已有会话""" import uuid self.current_session = session_id or str(uuid.uuid4()) # 尝试从持久化存储加载历史 messages = self.store.get_session_messages(self.current_session) if messages: logger.info(f"恢复会话 {self.current_session}，加载 {len(messages)} 条历史消息") # 按时间正序加载 for msg in reversed(messages): self.memory.add(msg["role"], msg["content"]) else: logger.info(f"新建会话 {self.current_session}") self.store.create_session(self.current_session, topic) def save_current_state(self): """保存当前会话状态""" if not self.current_session: return for msg in self.memory.get_messages(): self.store.save_message( self.current_session, msg["role"], msg["content"] ) ``` **预防措施：** - 始终使用绝对路径存储数据 - 在应用启动时打印数据文件路径，方便确认 - 定期验证持久化功能（可以写一个单元测试） --- ### 问题4：Agent输出不一致 **症状：** 相同或类似的输入，Agent每次输出差异很大，或者输出格式不固定。 **诊断步骤：** ``` Step 1: 检查temperature设置 -> temperature=0 时输出最稳定 -> temperature=1 时输出变化最大 -> 结构化任务建议 temperature=0.1~0.3 Step 2: 检查Prompt是否足够明确 -> 是否明确指定了输出格式？ -> 是否给出了输出示例？ Step 3: 检查是否有随机因素 -> 工具返回的数据是否每次不同？ -> 记忆中加载的内容是否不一致？ ``` **修复方案：** ```python # 方案A：降低temperature并使用结构化输出 from pydantic import BaseModel class ArticleOutput(BaseModel): """结构化输出模型""" title: str summary: str key_points: list[str] tags: list[str] call_to_action: str def generate_structured(topic: str) -> ArticleOutput: """生成结构化输出""" messages = [ { "role": "system", "content": """你是内容专家。请严格按照JSON格式输出。 输出格式要求： - title: 文章标题（20字以内） - summary: 一句话摘要（50字以内） - key_points: 3-5个要点 - tags: 3-5个标签 - call_to_action: 行动号召（30字以内）""" }, {"role": "user", "content": f"请为主题'{topic}'生成结构化内容"} ] response = client.chat.completions.create( model="gpt-4o-mini", messages=messages, temperature=0.1, # 低temperature保证一致性 response_format={"type": "json_object"} # 强制JSON输出 ) import json data = json.loads(response.choices[0].message.content) return ArticleOutput(**data) # 方案B：输出格式验证和重试 def generate_with_validation(topic: str, max_retries: int = 3) -> dict: """带验证的生成，格式不符则重试""" for attempt in range(max_retries): result = generate_structured(topic) # 验证输出质量 if len(result.key_points) < 3: logger.warning(f"输出要点不足，第{attempt+1}次重试") continue if len(result.title) > 30: logger.warning(f"标题过长，第{attempt+1}次重试") continue return {"success": True, "data": result} return {"success": False, "error": "多次重试后输出仍不符合要求"} # 方案C：输出后处理 - 统一格式 def normalize_output(raw_output: str) -> dict: """将LLM的自由文本输出标准化为统一格式""" # 使用正则或结构化解析提取关键信息 # 即使LLM的输出格式有小偏差，也能正确解析 import re result = { "title": "", "content": "", "tags": [] } # 尝试提取标题（假设以#开头） title_match = re.match(r"#\s*(.+)", raw_output) if title_match: result["title"] = title_match.group(1) # 其余部分作为正文 result["content"] = raw_output.strip() # 尝试提取标签（假设以#标签格式） result["tags"] = re.findall(r"#(\S+)", raw_output) return result ``` **预防措施：** - 结构化任务使用 `response_format={"type": "json_object"}` - 关键任务设置 temperature <= 0.3 - Prompt中包含输出格式示例 - 对输出进行验证和后处理 --- ## 二、代码审查清单 在中期评审时，使用以下清单自查代码质量。 ### 错误处理 - [ ] 所有外部API调用都有 try/except 包裹 - [ ] API超时设置合理（10-30秒） - [ ] 网络请求有重试机制（至少2次） - [ ] 错误信息对用户友好（不暴露内部细节） - [ ] 所有异常分支都有日志记录 ```python # 错误处理模板 def safe_api_call(func): """安全的API调用模板""" from tenacity import retry, stop_after_attempt, wait_exponential @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10), retry=lambda e: isinstance(e, (httpx.TimeoutException, httpx.ConnectError)) ) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except httpx.TimeoutException: logger.error("API调用超时") raise except httpx.HTTPStatusError as e: logger.error(f"API返回错误状态: {e.response.status_code}") raise except Exception as e: logger.error(f"未预期的错误: {type(e).__name__}: {e}") raise return wrapper ``` ### 日志记录 - [ ] 关键操作都有日志（Agent启动、工具调用、记忆读写） - [ ] 日志级别使用正确（INFO正常流程，WARNING异常但可恢复，ERROR不可恢复） - [ ] 日志包含足够上下文（不仅仅是错误信息，还有触发条件） - [ ] 日志文件不会无限增长（考虑日志轮转） ```python # 日志级别使用指南 logger.debug("...") # 开发调试用，生产环境关闭 logger.info("...") # 正常业务流程（开始处理X、完成X、调用工具X） logger.warning("...") # 异常但可恢复（使用降级策略、重试第N次） logger.error("...") # 不可恢复的错误（API不可用、数据损坏） logger.critical("...") # 系统级故障（数据库不可连接） ``` ### 输入验证 - [ ] 所有用户输入都经过验证（长度、格式、类型） - [ ] 外部API返回值在使用前做了格式检查 - [ ] 配置文件缺少必需项时有明确提示 - [ ] 没有直接使用未经验证的数据拼接SQL或命令 ### 速率限制 - [ ] 对API调用有频率控制（避免触发限流） - [ ] 大量请求时有排队机制 - [ ] 用户请求过频时有友好提示 ```python # 简单的速率限制器 import time from collections import deque class RateLimiter: """令牌桶速率限制器""" def __init__(self, max_calls: int = 10, window_seconds: int = 60): self.max_calls = max_calls self.window = window_seconds self.calls: deque = deque() def acquire(self): """获取调用许可，如果超出限制则等待""" now = time.time() # 清理过期记录 while self.calls and self.calls[0] < now - self.window: self.calls.popleft() if len(self.calls) >= self.max_calls: wait_time = self.window - (now - self.calls[0]) if wait_time > 0: logger.info(f"速率限制，等待 {wait_time:.1f} 秒") time.sleep(wait_time) self.acquire() # 递归重试 return self.calls.append(now) ``` --- ## 三、中期评审报告模板 完成以下模板，准备中期评审： ```markdown # 中期评审报告 ## 基本信息 - 项目名称： - 学生姓名： - 评审日期： ## 项目进度 ### 已完成的功能 1. 2. 3. ### 正在开发的功能 1. 2. ### 尚未开始的功能 1. 2. ## 技术状况 ### 能正常工作的部分 - - ### 存在问题的部分 | 问题 | 严重程度 | 当前状态 | 预计修复时间 | |------|---------|---------|-------------| | | 高/中/低 | 排查中/有方案/修复中 | | | | | | | ### 技术难点 1. 2. ## 测试情况 - 核心功能自测：通过 / 未通过 - 工具调用测试：通过 / 未通过 - 记忆持久化测试：通过 / 未通过 ## 下一步计划 1. [ ] 修复问题：... 2. [ ] 完成功能：... 3. [ ] 完善细节：... ## 需要的帮助 - [ ] 技术问题：... - [ ] 架构建议：... - [ ] 其他：... ``` --- ## 四、动手练习 ### 练习1：排查你的项目 依次检查以下项目，标记通过/未通过： ```bash # 1. 检查项目是否能正常运行 python -m src.main --query "测试" # 2. 检查日志是否正常 tail -f logs/app.log # 3. 检查数据文件是否存在 ls -la data/ # 4. 检查依赖是否安装 pip list | grep -E "openai|chromadb|langchain" ``` ### 练习2：模拟常见问题 ```python # 测试循环检测 detector = LoopDetector(threshold=3) for _ in range(7): detector.record("search") print(detector.is_looping()) # 应该返回 True # 测试持久化验证 validator = PersistenceValidator(long_term_memory) print(validator.verify()) # 应该返回 True ``` ### 练习3：填写中期评审报告 根据上面的模板，诚实填写你的项目状态。重点： - 列出所有已知问题 - 标注优先级（高/中/低） - 预估剩余工作量 --- ## 五、问题快速排查表 ``` ┌──────────────┬──────────────────┬────────────────────┐ │ 现象 │ 可能原因 │ 快速修复 │ ├──────────────┼──────────────────┼────────────────────┤ │ Agent不响应 │ API key无效 │ 检查.env配置 │ │ │ 网络不通 │ 检查网络连接 │ │ │ 模型名拼写错误 │ 检查model_name │ ├──────────────┼──────────────────┼────────────────────┤ │ 工具不工作 │ 未正确注册 │ 检查Agent工具列表 │ │ │ 参数格式不对 │ 打印调用参数 │ │ │ 外部服务不可用 │ 单独测试工具 │ ├──────────────┼──────────────────┼────────────────────┤ │ 记忆丢失 │ 路径不一致 │ 使用绝对路径 │ │ │ 未调用save │ 检查保存时机 │ │ │ 数据库被清空 │ 检查初始化逻辑 │ ├──────────────┼──────────────────┼────────────────────┤ │ 输出不一致 │ temperature太高 │ 降至0.1-0.3 │ │ │ Prompt不够明确 │ 添加格式示例 │ │ │ 未用结构化输出 │ 使用JSON模式 │ ├──────────────┼──────────────────┼────────────────────┤ │ 程序崩溃 │ 未处理异常 │ 添加try/except │ │ │ 内存溢出 │ 限制上下文长度 │ │ │ 文件权限问题 │ 检查目录权限 │ └──────────────┴──────────────────┴────────────────────┘ ``` --- ## 小结 本课完成了以下工作： 1. **掌握了4类常见问题的修复方法**：无限循环（设置上限+循环检测）、工具静默失败（统一包装器+结果验证）、记忆无法持久化（绝对路径+验证器）、输出不一致（低temperature+结构化输出）。 2. **使用了代码审查清单**：从错误处理、日志记录、输入验证、速率限制四个维度全面自查。 3. **完成了中期评审报告**：诚实地评估了项目进度，识别了风险和剩余工作量。 **关键要点：** - 无限循环是最危险的问题，必须从一开始就设上限。 - 工具调用不要信任输出，永远验证结果格式。 - 持久化问题多半是路径不对，用绝对路径。 - 结构化输出 + 低temperature = 一致性。 **下一步：** 进入 5.4 项目完善与答辩准备，完善项目并准备最终演示。