当前位置：首页 > news >正文

[智能体设计模式] 第12章：异常处理与恢复

news 2025/11/16 5:28:38

智能体在真实环境中需应对突发错误（如工具调用失败、服务不可用），异常处理与恢复模式的核心是：通过错误检测、分层处理、恢复机制，确保智能体在故障时不崩溃，保持功能连续性或优雅降级，提升系统可靠性。

核心逻辑

1. 核心组成

错误检测：识别工具调用失败、API 错误（404/500）、数据格式异常等问题。
错误处理：日志记录、重试、备用方案、优雅降级、通知告警。
恢复机制：状态回滚、自我修正、人工升级，将系统恢复至稳定状态。

2. 典型应用场景

客服机器人：数据库不可用时，告知用户并转人工。
交易机器人：处理“资金不足”“市场关闭”等错误，避免重复无效操作。
数据处理智能体：跳过损坏文件，继续批量任务并记录异常。
爬虫智能体：应对验证码、服务器错误，优雅暂停或报告失败。

实战代码示例（LangChain）

以下基于 LangChain 实现健壮的位置查询系统，核心能力：主工具调用失败时自动触发备用方案，最终统一输出结果，全程记录错误日志。

依赖安装

pip install langchain langchain-google-genai python-dotenv requests

创建 .env 文件配置密钥：

GOOGLE_API_KEY=你的 Gemini API 密钥（从 Google AI Studio 获取）

核心代码

"""
LangChain 异常处理与恢复示例：健壮位置查询系统
核心特性：
1. 主工具（精确位置查询）优先执行，失败自动触发备用工具（城市级查询）
2. 全程错误日志记录，支持问题追溯
3. 统一结果输出，失败时给出友好提示
4. 状态管理确保流程衔接顺畅
"""import logging
from typing import Dict, Any
from dotenv import load_dotenv
from langchain.agents import AgentType, initialize_agent, Tool
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.callbacks.base import BaseCallbackHandler# --------------------------
# 1. 日志配置：记录错误与流程（便于调试和追溯）
# --------------------------
logging.basicConfig(level=logging.INFO,format="%(asctime)s - %(levelname)s - %(message)s",handlers=[logging.FileHandler("agent_error.log"),  # 错误日志写入文件logging.StreamHandler()  # 同时输出到控制台]
)
logger = logging.getLogger("location_agent")# 加载环境变量
load_dotenv()# --------------------------
# 2. 自定义异常类：统一管理位置查询相关错误
# --------------------------
class LocationToolError(Exception):"""位置查询工具专用异常类，区分不同故障场景"""pass# --------------------------
# 3. 位置查询工具实现（模拟真实场景中的工具/API）
# --------------------------
def get_precise_location_info(address: str) -> str:"""主工具：获取精确位置信息（需完整地址）功能：返回包含经纬度、详细地址的精确结果失败条件：地址缺少街道、门牌号等关键信息"""try:# 模拟业务校验：仅接受包含街道关键词的完整地址street_keywords = ["街", "路", "巷", "弄", "号"]if not any(keyword in address for keyword in street_keywords):raise LocationToolError(f"地址 '{address}' 不完整，缺少街道或门牌号信息")# 模拟成功响应（实际场景中对接地图API）return (f"精确位置结果：\n"f"- 地址：{address}\n"f"- 经度：120.123456\n"f"- 纬度：30.654321\n"f"- 行政区域：XX省XX市XX区")except LocationToolError as e:# 记录业务错误日志logger.error(f"精确位置查询失败：{str(e)}")raise  # 抛出异常，让智能体检测except Exception as e:# 捕获未知错误，统一包装为自定义异常logger.error(f"精确位置查询意外错误：{str(e)}")raise LocationToolError(f"系统异常：{str(e)}") from edef get_general_area_info(city: str) -> str:"""备用工具：获取城市级模糊位置信息（容错性更高）功能：仅需城市名称，返回大致区域信息适用场景：主工具失败时的降级方案"""try:# 模拟基础校验：城市名称不为空if not city.strip():raise LocationToolError("城市名称不能为空")# 模拟成功响应（实际场景中对接简化版地图API）return (f"城市级位置结果：\n"f"- 城市：{city}\n"f"- 经度：120.123\n"f"- 纬度：30.654\n"f"- 行政区域：XX省{city}市")except Exception as e:logger.error(f"城市位置查询错误：{str(e)}")raise LocationToolError(f"获取城市信息失败：{str(e)}") from e# --------------------------
# 4. 回调处理器：检测工具错误并管理状态
# --------------------------
class ErrorDetectionCallback(BaseCallbackHandler):"""自定义回调处理器，核心作用：1. 监听工具调用失败事件2. 维护流程状态（主工具是否失败、查询结果等）3. 为后续智能体提供状态数据"""def __init__(self):# 初始化状态字典，存储流程关键信息self.state: Dict[str, Any] = {"primary_failed": False,  # 主工具是否失败"location_result": None,  # 最终位置结果"original_query": "",     # 用户原始查询"extracted_city": ""      # 提取的城市名称}def on_tool_error(self, error: Exception, **kwargs):"""工具调用失败时触发的回调方法"""if isinstance(error, LocationToolError):self.state["primary_failed"] = Truelogger.info("主工具调用失败，将触发备用方案")def reset(self):"""重置状态（每次处理新查询前调用）"""self.state = {"primary_failed": False,"location_result": None,"original_query": "","extracted_city": ""}# --------------------------
# 5. 构建核心智能体（分层设计，各司其职）
# --------------------------
# 初始化 LLM 模型
llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash",temperature=0.3  # 低温度保证结果稳定性
)# 初始化状态管理回调
state_callback = ErrorDetectionCallback()# 5.1 主智能体：尝试精确位置查询
primary_agent = initialize_agent(tools=[Tool(name="get_precise_location_info",func=get_precise_location_info,description="获取精确位置信息，必须传入包含街道、门牌号的完整地址（示例：上海市浦东新区博云路2号）")],llm=llm,agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,verbose=True,callbacks=[state_callback],  # 绑定状态回调agent_kwargs={"system_message": """你的任务：1. 从用户查询中提取完整地址（含街道、门牌号）2. 调用 get_precise_location_info 工具查询3. 若工具调用成功，将结果存入 state["location_result"]4. 若工具调用失败，无需额外操作，直接结束"""}
)# 5.2 备用智能体：主工具失败时触发
fallback_agent = initialize_agent(tools=[Tool(name="get_general_area_info",func=get_general_area_info,description="获取城市级模糊位置信息，仅需传入城市名称（示例：杭州市、上海市）")],llm=llm,agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,verbose=True,callbacks=[state_callback],agent_kwargs={"system_message": """你的任务：1. 先检查 state["primary_failed"] 是否为 True（主工具是否失败）2. 若是：从用户原始查询中提取城市名称，调用 get_general_area_info 工具3. 将查询结果存入 state["location_result"]4. 若主工具未失败，直接返回空结果，不调用工具"""}
)# 5.3 结果输出智能体：统一响应格式
response_agent = initialize_agent(tools=[],  # 仅处理状态，无需调用工具llm=llm,agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,verbose=True,agent_kwargs={"system_message": """你的任务：1. 检查 state["location_result"] 是否存在且有效2. 若有效：用清晰的语言展示位置信息3. 若无效：向用户致歉，说明无法获取位置，并建议提供完整地址（含街道、门牌号）"""}
)# --------------------------
# 6. 主控制流程：串联各智能体，实现异常处理与恢复
# --------------------------
def robust_location_agent(user_query: str) -> str:"""健壮位置查询主函数：流程：主工具查询 → 检测失败 → 备用工具查询 → 统一结果输出"""# 重置状态（处理新查询前清空历史数据）state_callback.reset()state_callback.state["original_query"] = user_querylogger.info(f"接收用户查询：{user_query}")# 第一步：执行主智能体（精确位置查询）logger.info("开始执行精确位置查询...")try:primary_result = primary_agent.run(f"用户查询：{user_query}，当前状态：{state_callback.state}")if "精确位置结果" in primary_result:state_callback.state["location_result"] = primary_resultexcept Exception as e:# 捕获主智能体执行失败（工具调用失败或其他异常）logger.info(f"精确查询执行失败：{str(e)}")state_callback.state["primary_failed"] = True# 第二步：主工具失败时，执行备用智能体if state_callback.state["primary_failed"] and not state_callback.state["location_result"]:logger.info("开始执行城市级备用查询...")try:fallback_result = fallback_agent.run(f"用户查询：{user_query}，当前状态：{state_callback.state}")if "城市级位置结果" in fallback_result:state_callback.state["location_result"] = fallback_resultexcept Exception as e:logger.error(f"备用查询执行失败：{str(e)}")state_callback.state["location_result"] = None# 第三步：执行结果输出智能体，返回最终响应logger.info("生成最终用户响应...")final_response = response_agent.run(f"当前状态：{state_callback.state}")return final_response# --------------------------
# 测试运行：验证不同场景下的异常处理能力
# --------------------------
if __name__ == "__main__":print("=== 健壮位置查询智能体测试 ===\n")# 测试用例 1：完整地址（主工具成功）print("【测试用例 1】输入完整地址")query1 = "查询上海市浦东新区张江高科技园区博云路2号的位置"print(f"用户输入：{query1}")print(f"智能体响应：{robust_location_agent(query1)}\n")# 测试用例 2：仅城市名称（主工具失败，触发备用工具）print("【测试用例 2】输入仅城市名称")query2 = "查询杭州市的位置"print(f"用户输入：{query2}")print(f"智能体响应：{robust_location_agent(query2)}\n")# 测试用例 3：无效地址（主备工具均失败）print("【测试用例 3】输入无效地址")query3 = "查询火星XXX区域的位置"print(f"用户输入：{query3}")print(f"智能体响应：{robust_location_agent(query3)}")