07_错误处理和重试机制

构建健壮的生产级应用！本节课学习如何优雅地处理各种错误情况，实现自动重试、降级策略和完善的日志记录。让你的AI应用在面对网络故障、API限流等问题时依然稳定可靠。

---

🎯 学习目标

1. 掌握常见错误类型和处理方法
2. 实现智能重试机制
3. 设计降级策略
4. 完善日志和监控
5. 构建健壮的生产级应用

---

1. 常见错误类型

1.1 API调用错误

# 网络错误
ConnectionError: 无法连接到API服务器

# 超时错误
TimeoutError: 请求超时

# API限流
RateLimitError: 超过API调用频率限制

# 认证错误
AuthenticationError: API密钥无效

1.2 处理原则

1. 预期错误：可以恢复的错误（网络超时、限流）→ 重试
2. 非预期错误：无法恢复的错误（API密钥错误）→ 快速失败
3. 降级策略：核心功能不可用时，提供降级服务

---

2. 智能重试机制

2.1 指数退避重试

import time
from typing import Callable, Any

def retry_with_backoff(
    func: Callable,
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0
) -> Any:
    """
    指数退避重试
    
    Args:
        func: 要执行的函数
        max_retries: 最大重试次数
        base_delay: 基础延迟（秒）
        max_delay: 最大延迟（秒）
    """
    for attempt in range(max_retries + 1):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries:
                raise  # 达到最大重试次数，抛出异常
            
            # 计算延迟：1s, 2s, 4s, 8s...
            delay = min(base_delay * (2 ** attempt), max_delay)
            print(f"❌ 第{attempt + 1}次失败: {e}")
            print(f"⏳ 等待{delay}秒后重试...")
            time.sleep(delay)

2.2 实战：带重试的LLM调用

创建 01_robust_llm_call.py`：

"""
健壮的LLM调用：错误处理和重试
学习目标：实现生产级的错误处理
"""
import os
import time
import logging
from typing import TypedDict, Optional
from typing_extensions import NotRequired
from langgraph.graph import StateGraph, END
from langchain_community.llms import Tongyi
from dotenv import load_dotenv

load_dotenv()

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


# 定义状态
class State(TypedDict):
    user_input: str
    response: NotRequired[str]
    error: NotRequired[str]
    retry_count: int
    max_retries: int


# 节点1：调用LLM
def call_llm(state: State) -> State:
    """调用LLM，处理各种错误"""
    logger.info(f"[尝试 {state['retry_count'] + 1}] 调用LLM...")
    
    try:
        llm = Tongyi(
            model="qwen-turbo",
            api_key=os.getenv("DASHSCOPE_API_KEY"),
            timeout=10  # 10秒超时
        )
        
        prompt = state["user_input"]
        response = llm.invoke(prompt)
        
        state["response"] = response
        logger.info("✅ LLM调用成功")
        
    except TimeoutError as e:
        logger.error(f"❌ 超时错误: {e}")
        state["error"] = "超时"
        state["retry_count"] += 1
        
    except ConnectionError as e:
        logger.error(f"❌ 连接错误: {e}")
        state["error"] = "连接失败"
        state["retry_count"] += 1
        
    except Exception as e:
        logger.error(f"❌ 未知错误: {e}")
        state["error"] = str(e)
        state["retry_count"] += 1
    
    return state


# 节点2：降级处理
def fallback_response(state: State) -> State:
    """降级策略：返回默认回复"""
    logger.warning("⚠️ 使用降级策略")
    
    state["response"] = """
抱歉，我现在遇到了一些技术问题，无法提供完整的回答。

您可以：
1. 稍后再试
2. 联系人工客服
3. 访问帮助文档

感谢您的理解！
"""
    return state


# 路由函数：决定重试还是降级
def should_retry(state: State) -> str:
    """判断是否需要重试"""
    # 如果成功，直接结束
    if "response" in state and "error" not in state:
        return "end"
    
    # 如果还有重试机会，继续重试
    if state["retry_count"] < state["max_retries"]:
        # 指数退避
        delay = min(2 ** state["retry_count"], 30)
        logger.info(f"⏳ 等待{delay}秒后重试...")
        time.sleep(delay)
        return "retry"
    
    # 超过最大重试次数，使用降级策略
    return "fallback"


# 构建图
def create_graph():
    """创建健壮的LLM调用图"""
    graph = StateGraph(State)
    
    graph.add_node("call_llm", call_llm)
    graph.add_node("fallback", fallback_response)
    
    graph.set_entry_point("call_llm")
    
    # 根据结果决定下一步
    graph.add_conditional_edges(
        "call_llm",
        should_retry,
        {
            "retry": "call_llm",    # 重试
            "fallback": "fallback",  # 降级
            "end": END               # 成功结束
        }
    )
    
    graph.add_edge("fallback", END)
    
    return graph.compile()


def main():
    """测试健壮的LLM调用"""
    app = create_graph()
    
    print("="*60)
    print("健壮的LLM调用系统")
    print("="*60)
    
    result = app.invoke({
        "user_input": "介绍一下人工智能的发展历史",
        "retry_count": 0,
        "max_retries": 3
    })
    
    print(f"\n{'='*60}")
    print("最终回复")
    print(f"{'='*60}")
    print(result.get("response", "无回复"))


if __name__ == "__main__":
    main()