【AI】四.Pydantic模型和LLM高级解析器实战

小滴课堂讲师 2025年09月18日 ai大模型, aigc 预计阅读 13 分钟

#### Python模型管理Pydantic介绍和安装

##### 什么是Pydatic

* Pydantic 是一个在 Python 中用于数据验证和解析的第三方库，是 Python 使用最广泛的数据验证库

* 声明式的方式定义数据模型和，结合Python 类型提示的强大功能来执行数据验证和序列化

* Pydantic 提供了从各种数据格式（例如 JSON、字典）到模型实例的转换功能

* 官方文档：https://pydantic.com.cn/

##### 为什么要用Pydantic

* 处理来自系统外部的数据，如API、用户输入或其他来源时，必须牢记开发中的原则：“永远不要相信用户的输入”
* AI智能体需要处理结构化数据（API请求/响应，配置文件等）
* 必须对这些数据进行严格的检查和验证，确保它们被适当地格式化和标准化
* **解决的问题**
  - **数据验证**：自动验证输入数据的类型和格式
  - **类型提示**：结合Python类型提示系统
  - **序列化**：轻松转换数据为字典/JSON
  - **配置管理**：支持复杂配置项的验证

##### 对比Java开发的模型验证

* 典型Java方式（需手写校验逻辑）

```java
  public class User {
      private String name;
      private int age;
      
      // 需要手写校验方法
      public void validate() throws IllegalArgumentException {
          if (name == null || name.isEmpty()) {
              throw new IllegalArgumentException("姓名不能为空");
          }
          if (age > 150) {
              throw new IllegalArgumentException("年龄不合法");
          }
      }
  }
  ```

* 传统Python方式（样板代码）

```python
  class User:
      def __init__(self, name: str, age: int):
          if not isinstance(name, str):
              raise TypeError("name必须是字符串")
          if not isinstance(age, int):
              raise TypeError("age必须是整数")
          if age > 150:
              raise ValueError("年龄必须在0-150之间")
          self.name = name
          self.age = age
  ```

* pydantic方式（声明式验证）

```python
  # 只需3行代码即可实现完整验证！
  from pydantic import BaseModel, Field
  class User(BaseModel):
      name: str = Field(min_length=1, max_length=50)  # 内置字符串长度验证
      age: int = Field(ge=0, le=150)  # 数值范围验证（类似Java的@Min/@Max）
  ```

##### 案例实战

* 模块安装 Pydantic V2（需要Python 3.10+）

```shell
  pip install pydantic==2.7.4
  ```

* 使用

* Pydantic 的主要方法是创建继承自 BaseModel 的自定义类

```python
  from pydantic import BaseModel
  
  # 类似Java中的POJO，但更强大
  class UserProfile(BaseModel):
      username: str       # 必须字段
      age: int = 18       # 默认值
      email: str | None = None  # 可选字段
  
  # 实例化验证
  user1 = UserProfile(username="Alice")
  print(user1)  # username='Alice' age=18 email=None
  
  user2 = UserProfile(username="Bob", age="20")  # 自动类型转换
  print(user2.age)  # 20（int类型）
  ```

* 创建实例与校验

```python
  try:
      UserProfile(username=123)  # 触发验证错误
  except ValueError as e:
      print(e.errors())
      # [{
      #   'type': 'string_type',
      #   'loc': ('username',),
      #   'msg': 'Input should be a valid string',
      #   'input': 123
      # }]
  ```

* 字段类型验证

```python
  from pydantic import BaseModel,HttpUrl,ValidationError
  
  class WebSite(BaseModel):
      url: HttpUrl #URL格式验证
      visits: int = 0 #默认值
      tags: list[str] = [] #字符串列表
  
  valid_data = {
      "url": "https://www.baidu.com",
      "visits": 100,
      "tags": ["python", "fastapi"]
  }
  
  # try:
  #     website = WebSite(**valid_data)
  #     print(website)
  # except ValidationError as e:
  #     print(e.errors())
  
  try:
      website = WebSite(url="xdclass.net",visits=100)
      print(website)
  except ValidationError as e:
      print(e.errors())
  ```

* 数据解析/序列化

```python
  from pydantic import BaseModel
  
  class Item(BaseModel):
      name: str
      price: float
  
  # 从JSON自动解析（类似Jackson）
  data = '{"name": "Widget", "price": "9.99"}'  # 字符串数字自动转换
  item = Item.model_validate_json(data)
  
  # 导出为字典（类似Java的POJO转Map）
  print(item.model_dump())  # {'name': 'Widget', 'price': 9.99}
  ```

* 调试技巧：打印模型结构

```python
  print(Website.model_json_schema())
  # 输出完整的JSON Schem
  ```

#### Pydantic字段校验Field函数多案例实战

##### Filed函数讲解

* Field函数通常用于给模型字段添加额外的元数据或者验证条件。
* 例如，title参数用来设置字段的标题，min_length用来限制最小长度。
* 核心
  * **`...` 的本质**：表示“无默认值”，强制字段必填。
  * **`Field` 的常用参数**：
    - `title`：字段标题（用于文档）
    - `description`：详细描述
    - `min_length`/`max_length`（字符串、列表）
    - `gt`/`ge`/`lt`/`le`（数值范围）
    - `regex`（正则表达式）
    - `example`（示例值，常用于 API 文档）

##### 案例实战

* 必填字段（无默认值）

* 必填字段：
    * ... 表示该字段必须显式提供值。若创建模型实例时未传入此字段，Pydantic 会抛出验证错误（ValidationError）。
  * 默认值占位
    * Field 的第一个参数是 default，而 ... 在此处的语义等价于“无默认值”。
    * 若省略 default 参数（如 Field(title="用户名")），Pydantic 会隐式使用 ...，但显式写出更明确

```python
  from pydantic import BaseModel, Field
  class User(BaseModel):
      name: str = Field(..., title="用户名", min_length=2)
  # 正确用法
  user = User(name="Alice")
  
  # 错误用法：缺少 name 字段
  user = User()  # 触发 ValidationError
  ```

* 可选字段（有默认值）

```python
  from pydantic import BaseModel, Field
  class UserOptional(BaseModel):
      name: str = Field("Guest", title="用户名")  # 默认值为 "Guest"
  
  # 可不传 name，自动使用默认值
  user = UserOptional()
  print(user.name)  # 输出 "Guest"
  ```

* 以下两种写法完全等价, 使用 `Field` 的优势在于可以附加额外参数（如 `title`、`min_length`、`description` 等）。

```python
  # 写法 1：省略 Field，直接类型注解
  name: str
  
  # 写法 2：显式使用 Field(...)
  name: str = Field(...)
  ```

* 数值类型必填

```python
  from pydantic import BaseModel, Field, ValidationError
  
  class Product(BaseModel):
      price: float = Field(..., title="价格", gt=0)  # 必须 > 0
      stock: int = Field(..., ge=0)  # 必须 >= 0
  
  # 正确
  product = Product(price=99.9, stock=10)
  
  # 错误：price <= 0
  try:
      Product(price=-5, stock=10)
  except ValidationError as e:
      print(e.json())  # 提示 "price" 必须大于 0
  ```

* 嵌套模型必填

```python
  from pydantic import BaseModel, Field
  class Address(BaseModel):
      city: str = Field(..., min_length=1)
      street: str
  
  class User(BaseModel):
      name: str = Field(...)
      address: Address  # 等效于 address: Address = Field(...)
  
  # 正确
  user = User(name="Alice", address={"city": "Shanghai", "street": "Main St"})
  
  # 错误：缺少 address
  user = User(name="Bob")  # 触发 ValidationError
  ```

* 明确可选字段

```python
  from pydantic import BaseModel, Field
  from typing import Optional
  
  class User(BaseModel):
      name: str = Field(...)
      email: Optional[str] = Field(None, title="邮箱")  # 可选，默认值为 None
  
  # 正确：不传 email
  user = User(name="Alice")
  ```

* 混合使用默认值和必填

```python
  from pydantic import BaseModel, Field
  
  class Config(BaseModel):
      api_key: str = Field(...)  # 必填
      timeout: int = Field(10, ge=1)  # 可选，默认 10，但必须 >=1
  
  # 正确
  config = Config(api_key="secret")
  assert config.timeout == 10
  
  # 错误：未传 api_key
  Config(timeout=5)  # 触发 ValidationError
  ```

#### Pydantic自定义验证器多案例实战

##### @field_validator介绍

* 是 Pydantic 中用于为**单个字段**添加自定义验证逻辑的装饰器

* **适用场景**：当默认验证规则（如 `min_length`、`gt`）无法满足需求时，通过编写代码实现复杂校验。

* **触发时机**：默认在字段通过基础类型和规则校验后执行（可通过 `mode` 参数调整）

* 基础语法

```python
  from pydantic import BaseModel, ValidationError, field_validator
  
  class User(BaseModel):
      username: str
      
      # 带默认值的可选字段
      # int | None表示 age 变量的类型可以是整数 (int) 或 None，旧版本的写法：age: Union[int, None]
      # Python 3.10 开始引入，替代了早期通过 Union[int, None] 的形式（仍兼容）
      age: int | None = Field(
          default=None,
          ge=18,
          description="用户年龄必须≥18岁"
      )
  
      @field_validator("username")
      def validate_username(cls, value: str) -> str:
          # cls: 模型类（可访问其他字段）
          # value: 当前字段的值
          if len(value) < 3:
              raise ValueError("用户名至少 3 个字符")
          return value  # 可修改返回值（如格式化）
  ```

##### 案例实战

* 字符串格式校验

```python
  from pydantic import BaseModel, Field,field_validator
  
  class User(BaseModel):
      email: str
  
      @field_validator("email")
      def validate_email(cls, v):
          if "@" not in v:
              raise ValueError("邮箱格式无效")
          return v.lower()  # 返回格式化后的值
  
  # 正确
  User(email="ALICE@example.com")  # 自动转为小写：alice@example.com
  
  # 错误
  #User(email="invalid")  # 触发 ValueError
  ```

* 验证用户名和长度

```python
  from pydantic import BaseModel, Field,field_validator
  class User(BaseModel):
      username: str = Field(..., min_length=3)
      
      @field_validator("username")
      def validate_username(cls, v):
          if "admin" in v:
              raise ValueError("用户名不能包含 'admin'")
          return v
  
  # 正确
  User(username="alice123")
  
  # 错误
  #User(username="admin")  # 触发自定义验证错误
  ```

* 密码复杂性验证

```python
  from pydantic import BaseModel, Field,field_validator
  
  class Account(BaseModel):
      password: str
  
      @field_validator("password")
      def validate_password(cls, v):
          errors = []
          if len(v) < 8:
              errors.append("至少 8 个字符")
          if not any(c.isupper() for c in v):
              errors.append("至少一个大写字母")
          if errors:
              raise ValueError("; ".join(errors))
          return v
  
  # 错误：密码不符合规则
  Account(password="weak")  # 提示：至少 8 个字符; 至少一个大写字母
  ```

* 多个字段共享验证器

```python
  from pydantic import BaseModel, Field,field_validator
  class Product(BaseModel):
      price: float
      cost: float
  
      @field_validator("price", "cost")
      def check_positive(cls, v):
          if v <= 0:
              raise ValueError("必须大于 0")
          return v
  
  # 同时验证 price 和 cost 是否为正数
  
  Product(price=1, cost=-2)
  ```

##### 注意事项

* 忘记返回值：验证器必须返回字段的值（除非明确要修改）。

```python
  @field_validator("email")
  def validate_email(cls, v):
      if "@" not in v:
          raise ValueError("Invalid email")
      # ❌ 错误：未返回 v
  ```

##### Pydantic V2 现为 Pydantic 的当前生产发布版本

* 网上不少是V1版本的教程，需要注意
* Pydantic V1和Pydantic V2的API差异

#### 重点-解析器PydanticOutputParser实战

##### 为啥要用为什么需要Pydantic解析

* 结构化输出：将非结构化文本转为可编程对象
* 数据验证：自动验证字段类型和约束条件，单纯json解析器则不会校验
* 开发效率：减少手动解析代码
* 错误处理：内置异常捕获与修复机制

##### 案例实战一

**大模型信息输出提取（  `PydanticOutputParser` 结合Pydantic模型验证输出）**

```python
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field  
from langchain_core.output_parsers import PydanticOutputParser  
#定义模型
model = ChatOpenAI(
    model_name = "qwen-plus",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key="sk-005c3c25f6d042848b29d75f2f020f08",
    temperature=0.7
)

# Step1: 定义Pydantic模型  
class UserInfo(BaseModel):  
    name: str = Field(description="用户姓名")  
    age: int = Field(description="用户年龄", gt=0)  
    hobbies: list[str] = Field(description="兴趣爱好列表")

# Step2: 创建解析器  
parser = PydanticOutputParser(pydantic_object=UserInfo)

# Step3: 构建提示模板  
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template("""  
提取用户信息，严格按格式输出：  
{format_instructions}

输入内容：  
{input}  
""")

# 注入格式指令  
prompt = prompt.partial(  
    format_instructions=parser.get_format_instructions()  
)

# Step4: 组合处理链  
chain = prompt | model | parser

# 执行解析  
result = chain.invoke({  
    "input": """  
   我的名称是张三，年龄是18岁，兴趣爱好有打篮球、看电影。  
   """  
})

print(type(result)) 
print(result) 
```

##### 案例实战二

**电商评论情感分析系统（JsonOutputParser和pydantic结合）**

* `JsonOutputParser`与`PydanticOutputParser`类似
* 新版才支持从pydantic获取约束模型，该参数并非强制要求，而是可选的增强功能
* `JsonOutputParser`可以处理流式返回的部分JSON对象。

```python
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field 
from langchain_core.prompts import ChatPromptTemplate  
from langchain_core.output_parsers import JsonOutputParser   
#定义模型
model = ChatOpenAI(
    model_name = "qwen-plus",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key="sk-005c3c25f6d042848b29d75f2f020f08",
    temperature=0.7
)

# 定义JSON结构  
class SentimentResult(BaseModel):  
    sentiment: str  
    confidence: float  
    keywords: list[str]

# 构建处理链  
parser = JsonOutputParser(pydantic_object=SentimentResult)

prompt = ChatPromptTemplate.from_template("""  
分析评论情感：  
{input}  
按以下JSON格式返回：  
{format_instructions}  
""").partial(format_instructions=parser.get_format_instructions())

chain = prompt | model | parser

# 执行分析  
result = chain.invoke({"input": "物流很慢，包装破损严重"})  
print(result)

# 输出：  
# {  
#   "sentiment": "negative",  
#   "confidence": 0.85,  
#   "keywords": ["物流快", "包装破损"]  
# }

# 2. 执行流式调用
#for chunk in chain.stream({"input": "物流很慢，包装破损严重"}):
#    print(chunk)  # 逐词输出
```

#### 重点-大模型修复重试机制OutputFixingParser

##### OutputFixingParser

* 是LangChain中用于修复语言模型（LLM）输出格式错误的工具，通常与`PydanticOutputParser`配合使用。
* 当原始解析器因格式问题（如JSON语法错误、字段缺失等）失败时，它能自动调用LLM修正输出，提升解析的鲁棒性。
* 核心功能：
  * 自动纠错：修复不规范的输出格式（如单引号JSON、字段顺序错误等）。
  * 兼容性：与Pydantic数据模型无缝集成，支持结构化输出验证。
  * 容错机制：避免因模型输出不稳定导致程序中断

##### 核心语法与使用步骤

* 基础语法

```python
from langchain.output_parsers import OutputFixingParser, PydanticOutputParser
from langchain_openai import ChatOpenAI

# 步骤1：定义Pydantic数据模型
class MyModel(BaseModel):
    field1: str = Field(description="字段描述")
    field2: int

# 步骤2：创建原始解析器
parser = PydanticOutputParser(pydantic_object=MyModel)

#定义模型
model = ChatOpenAI(
    model_name = "qwen-plus",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key="sk-005c3c25f6d042848b29d75f2f020f08",
    temperature=0.7
)

# 步骤3：包装为OutputFixingParser
fixing_parser = OutputFixingParser.from_llm(parser=parser, llm=model)
```

* 参数说明：
  * parser: 原始解析器对象（如PydanticOutputParser）。
  * llm: 用于修复错误的语言模型实例。
  * max_retries（可选）: 最大重试次数（默认1）

##### 案例实战

* 修复机制
  * 检测到错误后，将错误信息与原始输入传递给LLM。
  * LLM根据提示生成符合Pydantic模型的修正结果。

```python
from langchain.output_parsers import OutputFixingParser
from langchain_core.output_parsers import PydanticOutputParser
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field  
from typing import List

class Actor(BaseModel):
    name: str = Field(description="演员姓名")
    film_names: List[str] = Field(description="参演电影列表")

parser = PydanticOutputParser(pydantic_object=Actor)

# 包装原始解析器
fixing_parser = OutputFixingParser.from_llm(parser=parser, llm=model)

# 模拟模型输出的错误格式（使用单引号）
misformatted_output = "{'name': '小滴课堂老王', 'film_names': ['A计划','架构大课','一路向西']}"

#在LLM 链中，chain.invoke会将 LLM 返回的文本字符串传入output_parser.invoke
#而`output_parser.invoke`最终会调用到`output_parser.parse`。

# try:
#     parsed_data = parser.parse(misformatted_output)  # 直接解析会失败
# except Exception as e:
#     print(f"解析失败：{e}")  # 抛出JSONDecodeError

# 使用OutputFixingParser修复并解析
fixed_data = fixing_parser.parse(misformatted_output)

print(type(fixed_data))
print(fixed_data.model_dump())
# 输出：{'name': '小滴课堂老王', 'film_names': ['A计划','架构大课','一路向西']}
```

* 完整正常案例

class Actor(BaseModel):
    name: str = Field(description="演员姓名")
    film_names: List[str] = Field(description="参演电影列表")

parser = PydanticOutputParser(pydantic_object=Actor)

prompt = PromptTemplate(
    template="{format_instructions}\n{query}",
    input_variables=["query"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

# 包装原始解析器
fixing_parser = OutputFixingParser.from_llm(parser=parser, llm=model)  
chain = prompt | model | fixing_parser
response = chain.invoke({"query": "说下成龙出演过的5部动作电影？ "})
print(response)
print(type(response))
print(response.model_dump())
```

##### 常见问题与解决方案

* 修复失败的可能原因
  * 模型能力不足：升级LLM版本（如使用更高级的模型和参数量）。
  * 提示词不清晰：在提示模板中明确格式要求。
  * 网络问题：通过代理服务优化API访问

评论区