Pydantic 数据验证库

from pydantic import BaseModel
from typing import Optional, List

# 基本模型定义
class LLMPrompt(BaseModel):
    prompt: str
    max_tokens: int = 100         # 带默认值的字段
    temperature: float = 0.7      # 带默认值的字段
    top_p: float = 1.0

# 实例化
request = LLMPrompt(prompt="解释量子计算")
print(request.prompt)          # 解释量子计算
print(request.max_tokens)      # 100
print(request.temperature)     # 0.7

# 模型字段会自动类型转换（coercion）
request = LLMPrompt(prompt="Hello", max_tokens="200", temperature=0)
print(request.max_tokens)      # 200（str → int 自动转换）
print(type(request.max_tokens)) # <class 'int'>

# 访问模型字段元数据
print(LLMPrompt.model_fields)
# {'prompt': FieldInfo(...), 'max_tokens': FieldInfo(...), ...}

# 模型转字典
print(request.model_dump())
# {'prompt': 'Hello', 'max_tokens': 200, 'temperature': 0.0, 'top_p': 1.0}

# 模型转JSON字符串
print(request.model_dump_json())
# '{"prompt":"Hello","max_tokens":200,"temperature":0.0,"top_p":1.0}'

from pydantic import BaseModel
from typing import Optional, List, Dict, Literal, Union
from datetime import datetime
from uuid import UUID

class LLMConfig(BaseModel):
    # ---- 基本类型 ----
    name: str                                    # 字符串
    version: int                                 # 整数
    score: float                                 # 浮点数
    is_active: bool                              # 布尔值

    # ---- 可选类型 ----
    description: Optional[str] = None            # 可选，默认None
    tags: List[str] = []                         # 列表，默认空列表

    # ---- 复杂类型 ----
    metadata: Dict[str, str] = {}                # 字典
    model_type: Literal['gpt', 'claude', 'llama'] = 'gpt'  # 字面量枚举

    # ---- 日期时间 ----
    created_at: datetime = datetime.now()        # 日期时间

    # ---- 联合类型 ----
    value: Union[str, int] = 0                   # 字符串或整数

# 实例化与验证
config = LLMConfig(
    name="MyModel",
    version="2",      # str → int 自动转换
    score="0.95",     # str → float 自动转换
    is_active=1,      # int → bool 自动转换
    model_type="gpt"
)
print(config.version)    # 2 (int)
print(config.score)      # 0.95 (float)
print(config.is_active)  # True (bool)

from pydantic import BaseModel, Field
from typing import List, Dict
import time

class ModelConfig(BaseModel):
    # 静态默认值
    name: str = "default"
    temperature: float = 0.7

    # 使用 Field 定义默认值（推荐）
    max_tokens: int = Field(default=100, description="最大生成token数")

    # 可变默认值必须使用 default_factory
    tags: List[str] = Field(default_factory=list)
    metadata: Dict[str, str] = Field(default_factory=dict)

    # 动态默认值
    created_at: float = Field(default_factory=time.time)

# ⚠️ 错误示范：可变默认值直接赋值
# class Bad(BaseModel):
#     tags: List[str] = []  # 所有实例共享同一个列表！

# ✅ 正确做法
# class Good(BaseModel):
#     tags: List[str] = Field(default_factory=list)

from pydantic import BaseModel, ValidationError

class APIRequest(BaseModel):
    model: str
    temperature: float
    max_tokens: int

# 正常验证
req = APIRequest(model="gpt-4", temperature=0.7, max_tokens=100)
print(req)  # model='gpt-4' temperature=0.7 max_tokens=100

# 验证失败
try:
    req = APIRequest(model="gpt-4", temperature="not_a_number", max_tokens=100)
except ValidationError as e:
    print(e.error_count())   # 错误数量
    print(e.errors())        # 错误详情列表
    # [{'type': 'float_parsing', 'loc': ('temperature',), 'msg': '...', 'input': 'not_a_number'}]

# 缺少必填字段
try:
    req = APIRequest(model="gpt-4")
except ValidationError as e:
    print(e.errors())
    # [{'type': 'missing', 'loc': ('temperature',), 'msg': 'Field required', ...}]
    # [{'type': 'missing', 'loc': ('max_tokens',), 'msg': 'Field required', ...}]

from pydantic import BaseModel, field_validator, ValidationError

class ChatRequest(BaseModel):
    model: str
    temperature: float
    max_tokens: int
    prompt: str

    # 验证单个字段 —— 在类型转换之后执行
    @field_validator('temperature')
    @classmethod
    def temperature_range(cls, v: float) -> float:
        """确保温度在 [0, 2] 范围内"""
        if not 0 <= v <= 2:
            raise ValueError(f'temperature 必须在 0-2 之间，当前值: {v}')
        return v

    @field_validator('max_tokens')
    @classmethod
    def max_tokens_positive(cls, v: int) -> int:
        """确保 max_tokens 为正整数"""
        if v <= 0:
            raise ValueError('max_tokens 必须为正整数')
        return v

    @field_validator('model')
    @classmethod
    def supported_model(cls, v: str) -> str:
        """确保模型名称合法"""
        supported = {'gpt-4', 'gpt-3.5-turbo', 'claude-3', 'llama-3'}
        if v not in supported:
            raise ValueError(f'不支持的模型: {v}，支持的模型: {supported}')
        return v

    # 多字段同时验证
    @field_validator('prompt')
    @classmethod
    def prompt_not_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError('prompt 不能为空')
        return v.strip()

# 测试
try:
    req = ChatRequest(model="gpt-5", temperature=3.0, max_tokens=-1, prompt="  ")
except ValidationError as e:
    for err in e.errors():
        print(f"字段 {err['loc']}: {err['msg']}")

from pydantic import BaseModel, model_validator, ValidationError

class GenerationConfig(BaseModel):
    temperature: float = 0.7
    top_p: float = 1.0
    top_k: int = 50
    frequency_penalty: float = 0.0

    @model_validator(mode='after')
    def check_sampling_params(self) -> 'GenerationConfig':
        """temperature 和 top_p 不能同时设置"""
        if self.temperature < 0.01 and self.top_p < 1.0:
            raise ValueError('不能同时使用低 temperature 和低 top_p 采样')
        return self

    @model_validator(mode='after')
    def check_penalty_range(self) -> 'GenerationConfig':
        """frequency_penalty 范围检查"""
        if not -2.0 <= self.frequency_penalty <= 2.0:
            raise ValueError('frequency_penalty 必须在 -2.0 到 2.0 之间')
        return self

# 测试
try:
    config = GenerationConfig(temperature=0.0, top_p=0.5)
except ValidationError as e:
    print(e)

from pydantic import BaseModel, Field
from typing import Optional

class LLMResponse(BaseModel):
    text: str
    score: float
    model_name: str
    tokens_used: int
    internal_id: Optional[int] = None

response = LLMResponse(
    text="量子计算利用量子叠加和纠缠...",
    score=0.95,
    model_name="gpt-4",
    tokens_used=150,
    internal_id=42
)

# 完整序列化
print(response.model_dump())
# {'text': '量子计算利用量子叠加和纠缠...', 'score': 0.95, 'model_name': 'gpt-4', 'tokens_used': 150, 'internal_id': 42}

# 排除指定字段
print(response.model_dump(exclude={'internal_id'}))
# {'text': '量子计算利用量子叠加和纠缠...', 'score': 0.95, 'model_name': 'gpt-4', 'tokens_used': 150}

# 只包含指定字段
print(response.model_dump(include={'text', 'score'}))
# {'text': '量子计算利用量子叠加和纠缠...', 'score': 0.95}

# 排除未设置的字段（默认值为None且未修改的字段）
print(response.model_dump(exclude_none=True))
# {'text': '...', 'score': 0.95, 'model_name': 'gpt-4', 'tokens_used': 150, 'internal_id': 42}

# 排除默认值
print(response.model_dump(exclude_defaults=True))
# {'text': '量子计算利用量子叠加和纠缠...', 'score': 0.95, 'model_name': 'gpt-4', 'tokens_used': 150}

# 序列化为 JSON 字符串
json_str = response.model_dump_json()
print(json_str)
# '{"text":"量子计算利用量子叠加和纠缠...","score":0.95,"model_name":"gpt-4","tokens_used":150,"internal_id":42}'

# 带缩进
json_str = response.model_dump_json(indent=2)

# 排除字段
json_str = response.model_dump_json(exclude={'internal_id'})

from pydantic import BaseModel, ValidationError

class ChatMessage(BaseModel):
    role: str
    content: str

# 从字典创建（带验证）
data = {"role": "user", "content": "Hello!"}
msg = ChatMessage.model_validate(data)
print(msg)  # role='user' content='Hello!'

# 验证失败
try:
    ChatMessage.model_validate({"role": "user"})  # 缺少 content
except ValidationError as e:
    print(e)

# 从 JSON 字符串创建
json_str = '{"role": "assistant", "content": "你好！"}'
msg = ChatMessage.model_validate_json(json_str)
print(msg)  # role='assistant' content='你好！'

# 验证失败
try:
    ChatMessage.model_validate_json('{"role": "user"}')
except ValidationError as e:
    print(e)

from pydantic import BaseModel
from typing import List

class Message(BaseModel):
    role: str
    content: str

class ChatRequest(BaseModel):
    model: str
    messages: List[Message]
    temperature: float = 0.7

# 从嵌套字典创建
req = ChatRequest.model_validate({
    "model": "gpt-4",
    "messages": [
        {"role": "system", "content": "你是一个助手"},
        {"role": "user", "content": "你好"}
    ]
})

# 嵌套序列化
print(req.model_dump())
# {'model': 'gpt-4', 'messages': [{'role': 'system', 'content': '你是一个助手'}, {'role': 'user', 'content': '你好'}], 'temperature': 0.7}

print(req.model_dump_json())
# '{"model":"gpt-4","messages":[{"role":"system","content":"你是一个助手"},{"role":"user","content":"你好"}],"temperature":0.7}'

from pydantic import BaseModel, Field
from typing import Optional

class ModelParams(BaseModel):
    # ---- 数值约束 ----
    temperature: float = Field(
        default=0.7,
        ge=0.0,          # 大于等于 (greater than or equal)
        le=2.0,          # 小于等于 (less than or equal)
        description="采样温度，越高越随机"
    )

    top_p: float = Field(
        default=1.0,
        gt=0.0,          # 大于 (greater than)
        le=1.0,          # 小于等于
        description="核采样概率阈值"
    )

    max_tokens: int = Field(
        default=100,
        gt=0,            # 大于 0
        le=4096,         # 最大 4096
        description="最大生成 token 数"
    )

    # ---- 字符串约束 ----
    prompt: str = Field(
        min_length=1,    # 最小长度
        max_length=10000, # 最大长度
        description="输入提示文本"
    )

    model_name: str = Field(
        default="gpt-4",
        pattern=r'^(gpt-4|gpt-3.5-turbo|claude-3|llama-3)$',  # 正则约束
        description="模型名称"
    )

    # ---- 列表约束 ----
    stop_sequences: list[str] = Field(
        default_factory=list,
        max_length=4,    # 列表最大长度
        description="停止序列"
    )

    # ---- 字段描述（生成JSON Schema时使用） ----
    frequency_penalty: float = Field(
        default=0.0,
        ge=-2.0,
        le=2.0,
        description="频率惩罚，降低重复token的概率"
    )

# 验证约束
from pydantic import ValidationError

try:
    params = ModelParams(
        temperature=3.0,    # 超出 le=2.0
        top_p=0.0,          # 不满足 gt=0.0
        max_tokens=0,       # 不满足 gt=0
        prompt="",          # 不满足 min_length=1
        model_name="unknown"  # 不匹配 pattern
    )
except ValidationError as e:
    for err in e.errors():
        print(f"字段 {err['loc']}: {err['msg']}")

# 正常创建
params = ModelParams(prompt="解释量子计算", temperature=0.5, max_tokens=500)
print(params.model_dump())

from pydantic import BaseModel, Field

class APIResponse(BaseModel):
    # Python 用下划线，JSON/API 用驼峰
    model_name: str = Field(alias="modelName")
    token_count: int = Field(alias="tokenCount")
    finish_reason: str = Field(alias="finishReason")

# 使用别名创建
resp = APIResponse(modelName="gpt-4", tokenCount=150, finishReason="stop")

# 序列化时使用别名
print(resp.model_dump(by_alias=True))
# {'modelName': 'gpt-4', 'tokenCount': 150, 'finishReason': 'stop'}

# 不使用别名（默认用 Python 字段名）
print(resp.model_dump())
# {'model_name': 'gpt-4', 'token_count': 150, 'finish_reason': 'stop'}

from pydantic import BaseModel, ConfigDict, Field

class StrictModel(BaseModel):
    model_config = ConfigDict(
        # ---- 额外字段处理 ----
        extra='forbid',          # 禁止额外字段（默认 'ignore'）
        # extra='ignore'        # 忽略额外字段
        # extra='allow'         # 允许额外字段

        # ---- 严格模式 ----
        strict=True,             # 禁止自动类型转换，要求精确类型匹配

        # ---- 验证赋值 ----
        validate_assignment=True, # 属性赋值时也进行验证（默认 False）

        # ---- 别名相关 ----
        populate_by_name=True,   # 允许同时使用字段名和别名创建

        # ---- 字段排序 ----
        # 使用定义顺序而非字母排序
    )

    name: str
    value: int

# 测试 extra='forbid'
try:
    StrictModel(name="test", value=1, extra_field="oops")
except Exception as e:
    print(f"额外字段被禁止: {e}")

# 测试 strict=True
try:
    StrictModel(name="test", value="1")  # str 不能自动转 int
except Exception as e:
    print(f"严格模式: {e}")

# 测试 validate_assignment
class ValidatedModel(BaseModel):
    model_config = ConfigDict(validate_assignment=True)
    score: float

m = ValidatedModel(score=0.9)
try:
    m.score = "not_a_number"  # 赋值时也会验证
except Exception as e:
    print(f"赋值验证: {e}")

from pydantic import BaseModel, ConfigDict

class FrozenConfig(BaseModel):
    model_config = ConfigDict(frozen=True)
    api_key: str
    base_url: str

config = FrozenConfig(api_key="sk-xxx", base_url="https://api.openai.com/v1")

try:
    config.api_key = "new-key"  # 不可修改
except Exception as e:
    print(f"不可变模型: {e}")

# frozen 模型可以作为字典键或放入集合
print(hash(config))  # 可哈希

from pydantic import BaseModel, Field
from typing import List, Optional, Literal
from enum import Enum

class Role(str, Enum):
    SYSTEM = "system"
    USER = "user"
    ASSISTANT = "assistant"

class Message(BaseModel):
    role: Role
    content: str
    name: Optional[str] = None

class ChatCompletionRequest(BaseModel):
    model: str = Field(
        pattern=r'^(gpt-4|gpt-3.5-turbo|claude-3|llama-3)$',
        description="模型ID"
    )
    messages: List[Message] = Field(
        min_length=1,
        description="对话消息列表"
    )
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)
    top_p: float = Field(default=1.0, gt=0.0, le=1.0)
    max_tokens: Optional[int] = Field(default=None, gt=0)
    stop: Optional[List[str]] = Field(default=None, max_length=4)
    stream: bool = False

class Usage(BaseModel):
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int

class Choice(BaseModel):
    index: int
    message: Message
    finish_reason: Literal["stop", "length", "content_filter"]

class ChatCompletionResponse(BaseModel):
    id: str
    object: str = "chat.completion"
    created: int
    model: str
    choices: List[Choice]
    usage: Usage

# 使用示例
request = ChatCompletionRequest(
    model="gpt-4",
    messages=[
        Message(role=Role.SYSTEM, content="你是一个Python专家"),
        Message(role=Role.USER, content="解释装饰器")
    ],
    temperature=0.5,
    max_tokens=500
)

print(request.model_dump_json(indent=2))

from pydantic import BaseModel, Field, ValidationError
from typing import List, Optional

# 定义期望的LLM输出结构
class Entity(BaseModel):
    name: str = Field(description="实体名称")
    entity_type: str = Field(description="实体类型，如人名、地名、组织")
    description: str = Field(description="实体描述")

class Relation(BaseModel):
    subject: str = Field(description="主语实体")
    predicate: str = Field(description="关系谓词")
    object: str = Field(description="宾语实体")

class KnowledgeExtraction(BaseModel):
    """从文本中提取的知识三元组"""
    entities: List[Entity] = Field(description="提取的实体列表")
    relations: List[Relation] = Field(description="提取的关系列表")
    summary: str = Field(description="文本摘要")

# 验证LLM输出
llm_output = """
{
    "entities": [
        {"name": "爱因斯坦", "entity_type": "人名", "description": "理论物理学家"},
        {"name": "相对论", "entity_type": "理论", "description": "物理学基础理论"}
    ],
    "relations": [
        {"subject": "爱因斯坦", "predicate": "提出", "object": "相对论"}
    ],
    "summary": "爱因斯坦提出了相对论"
}
"""

try:
    result = KnowledgeExtraction.model_validate_json(llm_output)
    print(f"提取到 {len(result.entities)} 个实体, {len(result.relations)} 个关系")
    print(result.model_dump_json(indent=2))
except ValidationError as e:
    print(f"LLM输出验证失败: {e}")

# pip install instructor openai
import instructor
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import List

# 定义输出模型
class SentimentAnalysis(BaseModel):
    sentiment: str = Field(description="情感极性: positive/negative/neutral")
    confidence: float = Field(ge=0.0, le=1.0, description="置信度")
    keywords: List[str] = Field(description="关键词列表")

# 使用 instructor 包装客户端
client = instructor.from_openai(OpenAI())

# 自动让LLM生成符合模型的结构化输出
result = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {"role": "user", "content": "分析以下文本的情感：这个产品真的太棒了，我非常满意！"}
    ],
    response_model=SentimentAnalysis,  # Pydantic 模型驱动生成
)

print(f"情感: {result.sentiment}")
print(f"置信度: {result.confidence}")
print(f"关键词: {result.keywords}")

from pydantic import BaseModel, Field, field_validator
from pydantic_settings import BaseSettings  # pip install pydantic-settings

class LLMAppSettings(BaseSettings):
    """LLM 应用配置，自动从环境变量读取"""

    # API 配置
    openai_api_key: str = Field(description="OpenAI API Key")
    openai_base_url: str = Field(
        default="https://api.openai.com/v1",
        description="API 基础URL"
    )

    # 模型配置
    default_model: str = Field(default="gpt-4")
    default_temperature: float = Field(default=0.7, ge=0.0, le=2.0)
    default_max_tokens: int = Field(default=1000, gt=0)

    # 应用配置
    debug: bool = Field(default=False)
    log_level: str = Field(default="INFO")

    @field_validator('openai_api_key')
    @classmethod
    def validate_api_key(cls, v: str) -> str:
        if not v.startswith('sk-'):
            raise ValueError('API Key 必须以 sk- 开头')
        return v

    model_config = {
        "env_prefix": "LLM_",        # 环境变量前缀: LLM_OPENAI_API_KEY
        "env_file": ".env",          # 从 .env 文件读取
        "extra": "ignore"
    }

# 使用（自动从环境变量/ .env 文件读取）
# settings = LLMAppSettings()
# print(settings.openai_api_key)
# print(settings.default_model)

from pydantic import BaseModel, Field
from typing import List, Optional

class SearchParams(BaseModel):
    """搜索参数"""
    query: str = Field(description="搜索查询关键词")
    max_results: int = Field(default=5, gt=0, description="最大返回结果数")
    language: Optional[str] = Field(default=None, description="语言过滤")

# 生成 JSON Schema（可直接用于 OpenAI Function Calling）
schema = SearchParams.model_json_schema()
print(schema)
# {
#   'properties': {
#     'query': {'description': '搜索查询关键词', 'title': 'Query', 'type': 'string'},
#     'max_results': {'default': 5, 'description': '最大返回结果数', ...},
#     'language': {'anyOf': [{'type': 'string'}, {'type': 'null'}], 'default': None, ...}
#   },
#   'required': ['query'],
#   'title': 'SearchParams',
#   'type': 'object'
# }

# 用于 OpenAI Function Calling
import openai

tools = [
    {
        "type": "function",
        "function": {
            "name": "search",
            "description": "搜索相关信息",
            "parameters": SearchParams.model_json_schema()
        }
    }
]

from pydantic import BaseModel, Field

class Example(BaseModel):
    name: str
    age: int = Field(gt=0, description="年龄")

# model_fields 存储了每个字段的完整元数据
print(Example.model_fields)
# {
#   'name': FieldInfo(annotation=str, required=True, ...),
#   'age': FieldInfo(annotation=int, required=False, default=..., gt=0, description='年龄', ...)
# }

from pydantic import BaseModel, Field
from typing import List, Dict

# ❌ 错误：直接使用可变默认值
# class Bad(BaseModel):
#     tags: List[str] = []          # 所有实例共享同一个列表！
#     meta: Dict[str, str] = {}     # 所有实例共享同一个字典！

# ✅ 正确：使用 default_factory
class Good(BaseModel):
    tags: List[str] = Field(default_factory=list)
    meta: Dict[str, str] = Field(default_factory=dict)

from pydantic import BaseModel, ConfigDict

# 对于高频创建的模型，可以预编译验证器
class HighPerfModel(BaseModel):
    model_config = ConfigDict(
        # 严格模式更快（跳过类型转换）
        # 但要求输入类型精确匹配
        # strict=True,

        # 关闭不需要的验证
        validate_assignment=False,  # 不需要赋值验证时关闭
    )
    name: str
    value: int

# model_validate 比 构造函数 稍慢（因为多一层间接调用）
# 对于性能敏感场景，直接用构造函数
m = HighPerfModel(name="test", value=1)  # 比 model_validate 快

from pydantic import BaseModel

class BaseRequest(BaseModel):
    model: str
    temperature: float = 0.7

class ChatRequest(BaseRequest):
    messages: list[dict]

class CompletionRequest(BaseRequest):
    prompt: str

# 子类会继承父类的所有字段和验证器
# 注意：model_config 也会被继承

# LLM 结构化输出验证最佳实践模板
from pydantic import BaseModel, Field, ConfigDict
from typing import List, Optional, Literal

class LLMOutput(BaseModel):
    """LLM 输出模型的最佳实践模板"""

    model_config = ConfigDict(
        extra='forbid',           # 禁止额外字段（防止幻觉）
        str_strip_whitespace=True # 自动去除空白
    )

    # 使用 Literal 限制枚举
    category: Literal["positive", "negative", "neutral"]

    # 使用 Field 约束和描述
    confidence: float = Field(
        ge=0.0, le=1.0,
        description="分类置信度，0-1之间"
    )

    # 可选字段明确标记
    reasoning: Optional[str] = Field(
        default=None,
        description="分类理由（可选）"
    )

    # 嵌套模型
    keywords: List[str] = Field(
        default_factory=list,
        max_length=10,
        description="关键词列表，最多10个"
    )

# 使用
import json

def parse_llm_output(raw_json: str) -> LLMOutput:
    """安全解析 LLM 输出"""
    try:
        return LLMOutput.model_validate_json(raw_json)
    except Exception as e:
        # 处理验证失败（如回退到默认值或重新请求LLM）
        raise ValueError(f"LLM 输出验证失败: {e}")