📖 引言：在异常中恢复
在实验室里，AI 智能体总是运行得完美无缺：API 永远响应迅速，数据库永远在线，用户输入永远清晰明确。但在现实世界中，情况截然不同。
外部 API 会突然返回 502 Bad Gateway。
数据库连接会超时。
用户会输入一些让 Prompt 逻辑崩溃的奇怪指令。
LLM 偶尔会产生幻觉，生成不符合 JSON Schema 的“脏数据”。
如果你的智能体一遇到这些问题就抛出 Unhandled Exception 并崩溃，那么它永远无法从玩具变成产品。
第十二篇：“异常处理与恢复”模式，是构建生产级（Production-Ready）智能体的最后一道防线。它不仅仅是 try-except 那么简单，而是赋予智能体一种韧性（Resilience）——在面对不可预见的失败时，能够检测问题、自我修复，或者至少做到“优雅地失败”。
本章将带你深入构建这道防线，结合 Google ADK 和最新的 LangGraph 框架，展示如何让你的智能体在混乱的现实世界中屹立不倒。
第一部分：异常处理的防御纵深
在 AI 智能体架构中，异常处理不是一个点，而是一个分层防御体系。
1.1 错误检测 (Detection)：看见不可见之物
智能体必须能感知到“出问题了”。错误不仅仅是代码层面的 Python Exception，还包括：
工具级错误：API 调用超时、鉴权失败、参数错误。
语义级错误：LLM 生成的内容虽然符合 JSON 格式，但逻辑上行不通（例如：订购了 0 个披萨）。
状态级错误：智能体陷入了死循环，或者在两个步骤之间反复横跳。
1.2 错误处理 (Handling)：战术应对
一旦检测到错误，智能体需要立即采取行动来止损：
重试 (Retry)：针对瞬时故障（如网络抖动），最简单的策略就是“再试一次”。通常配合指数退避 (Exponential Backoff) 策略。
回退 (Fallback)：如果首选工具不可用（如 Google Search 挂了），自动切换到备用工具（如 Bing Search）。
优雅降级 (Graceful Degradation)：如果无法完成全部任务，至少完成核心部分。例如，无法获取实时股价，但仍能提供昨天的收盘价，并明确告知用户数据有延迟。
1.3 恢复 (Recovery)：战略修复
对于更严重的错误，需要更复杂的恢复机制：
自我纠正 (Self-Correction)：利用 LLM 的反思能力。将错误信息（Traceback）喂给 LLM，让它分析原因并生成新的参数或代码。
状态回滚 (State Rollback)：如果一系列操作中的某一步失败，必须撤销之前的操作（如数据库事务回滚），确保系统状态一致性。
人工介入 (Human-in-the-loop)：当所有自动手段都失效时，将控制权转交给人类操作员。
第二部分：实战 LangGraph —— 构建具有“自我修复”能力的智能体
在 LangGraph 中，异常处理不再是散落在代码各处的 try-except，而是图结构中的显式路径。我们可以定义一个专门的 recovery_node，当主逻辑节点失败时，流程会自动流向修复节点。
2.1 场景定义：不稳定的数据查询助手
假设我们正在构建一个查询外部 API 的助手。这个 API 非常不稳定，经常超时或返回错误格式的数据。我们的目标是构建一个能够自动重试、自动修复参数并最终给出结果的智能体。
2.2 基础架构与状态定义
# pip install langgraph langchain-openai httpximport operatorfrom typing import TypedDict, Annotated, List, Optionalfrom langchain_openai import ChatOpenAIfrom langchain_core.messages import BaseMessage, HumanMessage, AIMessagefrom langgraph.graph import StateGraph, END
# 定义状态class AgentState(TypedDict): messages: Annotated[List[BaseMessage], operator.add] query: str tool_output: Optional[str] error: Optional[str] retry_count: int # 用于控制重试次数，防止无限循环
# 初始化 LLMllm = ChatOpenAI(model="gpt-4o", temperature=0)
复制代码
2.3 模拟不稳定的工具 (Simulated Tool)
为了演示，我们编写一个故意会出错的工具。
import random
def unreliable_api_tool(query: str) -> str: """模拟一个不稳定的外部 API""" failure_rate = 0.7 # 70% 概率失败 if random.random() < failure_rate: # 模拟不同类型的错误 error_type = random.choice(["Timeout", "Bad Request", "Auth Error"]) if error_type == "Bad Request": raise ValueError(f"API Error 400: Invalid query format for '{query}'") elif error_type == "Auth Error": raise PermissionError("API Error 403: Token expired") else: raise TimeoutError("API Error 504: Gateway Timeout") return f"Success! Data for '{query}' is: 42"
复制代码
2.4 节点定义：执行与恢复逻辑
我们将逻辑拆分为 Executor（执行者）和 Recoverer（修复者）。
def execution_node(state: AgentState): """尝试执行工具调用""" query = state['query'] current_retries = state.get('retry_count', 0) print(f"🚀 [Executor] 尝试执行查询: '{query}' (重试次数: {current_retries})") try: # 执行工具 result = unreliable_api_tool(query) return {"tool_output": result, "error": None} except Exception as e: error_msg = str(e) print(f"💥 [Executor] 捕获异常: {error_msg}") return { "error": error_msg, "retry_count": current_retries + 1 }
def recovery_node(state: AgentState): """根据错误类型制定恢复策略""" error = state['error'] query = state['query'] print(f"🚑 [Recoverer] 正在分析错误并尝试修复...") # 策略 1: 针对 Timeout，这通常是瞬时故障，只需重试 if "Timeout" in error: print(" -> 策略: 瞬时故障，保持原样重试。") return {"messages": [AIMessage(content="遇到网络超时，正在重试...")]} # 策略 2: 针对 Bad Request，可能是参数问题，利用 LLM 修正 Query if "Invalid query" in error: print(" -> 策略: 查询格式错误，尝试通过 LLM 修正参数。") # 让 LLM 尝试简化查询 fix_prompt = f"查询 '{query}' 导致了错误: {error}。请重写一个更简单、更标准的查询字符串。只返回查询本身。" new_query = llm.invoke(fix_prompt).content print(f" -> 修正后的查询: '{new_query}'") return { "query": new_query, "messages": [AIMessage(content=f"查询格式有误，已自动修正为: {new_query}")] } # 策略 3: 针对 Auth Error，这是无法自动修复的致命错误 # 在这里我们不清除 error，让路由逻辑决定终止 print(" -> 策略: 鉴权失败，无法自动修复。") return {"messages": [AIMessage(content="鉴权失败，请联系管理员更新 Token。")]}
复制代码
2.5 构建具备“韧性”的图
这里的关键是 条件边 (Conditional Edges)，它决定了我们是继续执行，还是进入恢复流程，亦或是彻底放弃。
def route_after_execution(state: AgentState): """执行后的路由逻辑""" if not state['error']: return "success" # 执行成功，结束 # 检查是否达到最大重试次数 if state['retry_count'] > 3: return "give_up" # 检查是否是致命错误 (无法重试) if "Auth Error" in state['error']: return "give_up" return "recover" # 进入恢复节点
# 构建图workflow = StateGraph(AgentState)
workflow.add_node("executor", execution_node)workflow.add_node("recoverer", recovery_node)
workflow.set_entry_point("executor")
workflow.add_conditional_edges( "executor", route_after_execution, { "success": END, "give_up": END, "recover": "recoverer" })
# 恢复后，再次尝试执行workflow.add_edge("recoverer", "executor")
app = workflow.compile()
复制代码
2.6 运行模拟
if __name__ == "__main__": initial_state = { "messages": [], "query": "complex_data_query_v1", "retry_count": 0, "error": None, "tool_output": None } print("--- 开始运行健壮的智能体 ---") final_state = app.invoke(initial_state) print("\n--- 最终结果 ---") if final_state['tool_output']: print(f"✅ 成功: {final_state['tool_output']}") else: print(f"❌ 失败: 最终错误信息 -> {final_state['error']}")
复制代码
运行结果分析：这个系统展示了强大的自我修复能力。
如果遇到 Timeout，它会自动重试，直到成功或达到次数上限。
如果遇到 Bad Request，它会调用 LLM 修改查询参数，然后带着新参数重试。
如果遇到 Auth Error，它会智能地判断这是不可恢复错误，直接终止，避免无意义的重试浪费资源。
第三部分：Google ADK 的回退机制 (Fallback Mechanism)
在 Google ADK 中，异常处理可以通过多智能体协作来优雅实现。我们可以设计一种 Primary/Fallback (主/备) 模式。
3.1 架构设计
Primary Agent：尝试使用高精度的工具（如 get_precise_location），该工具可能因为 GPS 信号弱而失败。
Fallback Agent：当主智能体失败时接管，使用低精度但高可用的工具（如 get_city_from_ip）。
Supervisor (SequentialAgent)：编排这一流程。
3.2 代码实现
from google.adk.agents import Agent, SequentialAgentfrom google.adk.tools import tool
# --- 模拟工具 ---@tooldef get_precise_location(user_id: str): """获取精确 GPS 坐标 (模拟高故障率)""" # 模拟失败 return {"error": "GPS signal lost", "status": "failed"}
@tooldef get_general_location(user_id: str): """获取粗略位置 (基于 IP)""" return {"city": "New York", "status": "success"}
# --- 1. Primary Agent: 尝试精确获取 ---primary_handler = Agent( name="primary_handler", model="gemini-2.5-flash", instruction=""" 尝试获取用户的精确位置。 如果工具返回错误，请将错误信息写入 state['location_error']， 并将 state['is_located'] 设置为 False。 """, tools=[get_precise_location], output_key="primary_logs")
# --- 2. Fallback Agent: 错误恢复 ---fallback_handler = Agent( name="fallback_handler", model="gemini-2.5-flash", instruction=""" 检查 state['is_located']。 - 如果为 True，什么都不做。 - 如果为 False，说明主定位失败。请调用 `get_general_location` 工具作为备选方案。 将最终位置结果写入 state['final_location']。 """, tools=[get_general_location])
# --- 3. Response Agent: 最终回复 ---response_agent = Agent( name="response_agent", model="gemini-2.5-flash", instruction=""" 根据 state['final_location'] 回复用户。 如果使用了备选方案，请告知用户“无法获取精确位置，显示大致位置”。 """)
# --- 4. 编排 ---# SequentialAgent 保证了执行顺序：主 -> 备 -> 回复robust_location_system = SequentialAgent( name="robust_location_system", sub_agents=[primary_handler, fallback_handler, response_agent])
复制代码
这种模式的优势在于解耦。主智能体不需要知道备用方案是什么，它只负责报错。恢复逻辑被封装在专门的 Fallback Agent 中，使得系统易于维护和扩展。
第四部分：高级模式——基于 LLM 的自我修复 (Self-Correction)
除了工具层面的重试，最令人兴奋的是利用 LLM 的认知能力来修复代码或数据错误。
4.1 场景：JSON 格式修复
在大规模数据提取任务中，LLM 生成的 JSON 经常会因为缺少引号、逗号等导致解析失败。传统的正则修复很难覆盖所有情况。
自我修复流程：
执行：LLM 生成 JSON。
验证：Python json.loads() 尝试解析。
捕获：解析失败，捕获 JSONDecodeError。
修复：将 错误的 JSON 字符串 + 错误信息 作为 Prompt 发送回 LLM：“你生成的 JSON 在第 10 行报错，请修复并只返回正确的 JSON。”
重试：解析修复后的 JSON。
4.2 代码示例 (LangChain LCEL)
from langchain_core.output_parsers import JsonOutputParserfrom langchain_core.pydantic_v1 import BaseModel, Fieldfrom langchain_openai import ChatOpenAIfrom langchain_core.prompts import PromptTemplate
# 定义期望的数据结构class UserInfo(BaseModel): name: str = Field(description="用户姓名") age: int = Field(description="用户年龄")
# 初始化model = ChatOpenAI(model="gpt-4o", temperature=0)parser = JsonOutputParser(pydantic_object=UserInfo)
# 构造一个自我修复的 Chain# 注意：LangChain 的 RetryOutputParser 封装了这个逻辑from langchain.output_parsers import RetryOutputParser
retry_parser = RetryOutputParser.from_llm( parser=parser, llm=model, max_retries=3)
# 原始 Promptprompt = PromptTemplate( template="提取信息：{text}\n{format_instructions}", input_variables=["text"], partial_variables={"format_instructions": parser.get_format_instructions()})
chain = prompt | model | retry_parser
# 测试：输入一段极其混乱、包含干扰文本的数据dirty_text = "用户叫张三，年龄大概是...嗯...三十岁吧 (30)。"result = chain.invoke({"text": dirty_text})
print(f"✅ 解析成功: {result}")# 即便模型第一次生成了非标准 JSON，RetryOutputParser 也会自动触发“生成-报错-修复”循环
复制代码
第五部分：生产环境的最佳实践
永远不要裸奔：所有的工具调用、LLM 调用都必须包裹在重试机制（Retry）和超时控制（Timeout）中。LangChain 的 RunnableRetry 提供了很好的支持。
死信队列 (Dead Letter Queue)：对于最终失败且无法恢复的任务，不要直接丢弃。将它们记录到专门的数据库或日志中（死信队列），以便开发人员后续进行人工分析（Post-Mortem）。
断路器模式 (Circuit Breaker)：如果某个工具（如 Bing Search）连续失败 10 次，应该触发断路器，在接下来的 5 分钟内直接跳过该工具（或返回缓存），避免雪崩效应拖垮整个系统。
状态一致性：在执行多步骤操作（如：扣款 -> 发货）时，如果发货失败，必须有回滚机制（退款）。在智能体设计中，这意味着需要设计“逆向工具”（Compensating Transactions）。
结语：拥抱不确定性
构建 AI 智能体的过程，本质上就是与不确定性博弈的过程。
模型是不确定的，世界是不确定的。我们无法消除所有错误，但我们可以通过异常处理与恢复模式，赋予智能体“反脆弱”的能力。
一个优秀的智能体，不是从不犯错，而是能够从错误中恢复，甚至从错误中学习。当你为智能体加上了这道防线，它才真正准备好走出实验室，去面对真实世界的风浪。
参考资料
1.Improving Fault Tolerance and Reliability of Heterogeneous Multi-Agent IoT Systems Using Intelligence Transfer
2.Antonio Gulli 《Agentic Design Patterns》
评论