📖 引言：在异常中恢复

在实验室里，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 httpx
import operator
from typing import TypedDict, Annotated, List, Optional
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from 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  # 用于控制重试次数，防止无限循环

# 初始化 LLM
llm = 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']}")

运行结果分析：
这个系统展示了强大的自我修复能力。

1. 如果遇到 Timeout，它会自动重试，直到成功或达到次数上限。
2. 如果遇到 Bad Request，它会调用 LLM 修改查询参数，然后带着新参数重试。
3. 如果遇到 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, SequentialAgent
from google.adk.tools import tool

# --- 模拟工具 ---
@tool
def get_precise_location(user_id: str):
    """获取精确 GPS 坐标 (模拟高故障率)"""
    # 模拟失败
    return {"error": "GPS signal lost", "status": "failed"}

@tool
def 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 经常会因为缺少引号、逗号等导致解析失败。传统的正则修复很难覆盖所有情况。

自我修复流程：

1. 执行：LLM 生成 JSON。
2. 验证：Python json.loads() 尝试解析。
3. 捕获：解析失败，捕获 JSONDecodeError。
4. 修复：将 错误的 JSON 字符串 + 错误信息 作为 Prompt 发送回 LLM：“你生成的 JSON 在第 10 行报错，请修复并只返回正确的 JSON。”
5. 重试：解析修复后的 JSON。

4.2 代码示例 (LangChain LCEL)

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI
from 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
)

# 原始 Prompt
prompt = 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 也会自动触发“生成-报错-修复”循环

---

第五部分：生产环境的最佳实践

1. 永远不要裸奔：所有的工具调用、LLM 调用都必须包裹在重试机制（Retry）和超时控制（Timeout）中。LangChain 的 RunnableRetry 提供了很好的支持。
2. 死信队列 (Dead Letter Queue)：对于最终失败且无法恢复的任务，不要直接丢弃。将它们记录到专门的数据库或日志中（死信队列），以便开发人员后续进行人工分析（Post-Mortem）。
3. 断路器模式 (Circuit Breaker)：如果某个工具（如 Bing Search）连续失败 10 次，应该触发断路器，在接下来的 5 分钟内直接跳过该工具（或返回缓存），避免雪崩效应拖垮整个系统。
4. 状态一致性：在执行多步骤操作（如：扣款 -> 发货）时，如果发货失败，必须有回滚机制（退款）。在智能体设计中，这意味着需要设计“逆向工具”（Compensating Transactions）。

---

结语：拥抱不确定性

构建 AI 智能体的过程，本质上就是与不确定性博弈的过程。

模型是不确定的，世界是不确定的。我们无法消除所有错误，但我们可以通过异常处理与恢复模式，赋予智能体“反脆弱”的能力。

一个优秀的智能体，不是从不犯错，而是能够从错误中恢复，甚至从错误中学习。当你为智能体加上了这道防线，它才真正准备好走出实验室，去面对真实世界的风浪。

参考资料

1.Improving Fault Tolerance and Reliability of Heterogeneous Multi-Agent IoT Systems Using Intelligence Transfer

2.Antonio Gulli 《Agentic Design Patterns》