# 核心概念
# 验证和重试 = 确保 LLM 应用的可靠性和数据质量
# 在生产环境中，需要处理三类问题：
# 网络错误 - 临时性连接问题（用 with_retry()）
# 模型故障 - 主模型不可用（用 with_fallbacks()）
# 输出质量 - LLM 输出不符合要求（用验证 + 重试循环）
# 基本用法
# 1. with_retry() - 自动重试
# 处理临时性网络错误：
import os
import sys
from langchain.chat_models import init_chat_model
from pydantic import BaseModel, Field, ValidationError
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from init_model import get_chat_model
chat_model = get_chat_model()
def demo1():
# 添加重试机制
llm_with_retry = chat_model.with_retry(
retry_if_exception_type=(ConnectionError, TimeoutError),
wait_exponential_jitter=True, # 指数退避 + 随机抖动
stop_after_attempt=3, # 最多重试 3 次
)
response = llm_with_retry.invoke("你好")
# 工作原理：
# 第 1 次尝试 → 失败 (ConnectionError)
# ↓ 等待 1s
# 第 2 次尝试 → 失败 (ConnectionError)
# ↓ 等待 2s
# 第 3 次尝试 → 成功 ✓
# 2. with_fallbacks() - 降级方案
# 主模型失败时切换到备用模型：
def demo2():
# 主模型
primary_model = init_chat_model("groq:llama-3.3-70b-versatile")
# 备用模型（更可靠/更便宜）
fallback_model = init_chat_model("groq:llama-3.1-8b-instant")
# 配置降级
llm_with_fallbacks = primary_model.with_fallbacks([fallback_model])
response = llm_with_fallbacks.invoke("介绍 Python")
# 主模型正常 → 使用主模型
# 主模型失败 → 自动切换到备用模型
# 3. Pydantic 验证
# 使用 Pydantic 约束确保数据质量：
def demo3():
from pydantic import BaseModel, Field, field_validator
class User(BaseModel):
name: str = Field(min_length=2, max_length=20)
age: int = Field(ge=0, le=150) # 0-150 岁
email: str
@field_validator("email")
@classmethod
def validate_email(cls, v):
if "@" not in v:
raise ValueError("邮箱必须包含 @")
return v
# 使用
try:
user = User(name="张三", age=200, email="invalid") # 失败
except ValidationError as e:
print(e.errors()) # 查看错误详情
# 核心组件
# 1. with_retry() 参数
# 参数 说明 默认值
# retry_if_exception_type 哪些异常会触发重试 (Exception,)
# stop_after_attempt 最大重试次数 3
# wait_exponential_jitter 是否使用指数退避 + 抖动 False
# 常见重试异常：
# # 网络相关
# (ConnectionError, TimeoutError, httpx.ConnectError)
# # API 限流
# (RateLimitError, )
# # 所有异常（谨慎使用）
# (Exception, )
# 2. with_fallbacks() 参数
# primary_model.with_fallbacks(
# fallbacks=[model1, model2], # 备用模型列表（按顺序尝试）
# exceptions_to_handle=(Exception,) # 触发降级的异常
# )
# 工作流程：
# 尝试 primary_model → 失败
# ↓
# 尝试 model1 → 失败
# ↓
# 尝试 model2 → 成功 ✓
# 3. Pydantic Field 约束
# 约束 用途 示例
# gt / ge 数值 > / >= Field(gt=0)
# lt / le 数值 < / <= Field(le=100)
# min_length / max_length 字符串长度 Field(min_length=2)
# pattern 正则表达式 Field(pattern=r'^\d{11}$')
# class Product(BaseModel):
# name: str = Field(min_length=2, max_length=50)
# price: float = Field(gt=0, description="价格必须 > 0")
# stock: int = Field(ge=0, description="库存 >= 0")
# 实际应用
# 1. LLM 输出验证 + 重试循环
# 当 LLM 输出不符合验证规则时，重新提示：
def demo4():
from pydantic import ValidationError
class Product(BaseModel):
name: str = Field(min_length=2)
price: float = Field(gt=0)
structured_llm = chat_model.with_structured_output(Product)
max_retries = 3
text = "产品价格是 -100 元" # 负价格会触发验证失败
for attempt in range(1, max_retries + 1):
try:
result = structured_llm.invoke(f"提取产品信息：{text}")
# 验证通过
break
except ValidationError as e:
error_msg = e.errors()[0]["msg"]
# 在提示中加入错误信息
text = f"{text}\n注意: {error_msg}。请确保价格 > 0"
if attempt == max_retries:
raise # 重试次数用完
# 第 1 次尝试:
# LLM 输出: price = -100
# 验证: ✗ 失败 (price 必须 > 0)
# 第 2 次尝试（修正提示）:
# 提示: "...注意: price 必须 > 0"
# LLM 输出: price = 100
# 验证: ✓ 通过