Pydantic 数据校验与 Settings 管理

项目中的应用位置

使用位置	作用	代码
全局配置	从.env加载并校验配置	Settings(BaseSettings)
API 请求校验	自动校验 JSON 请求体的字段和类型	RetrievalDebugRequest(BaseModel)
LLM 结构化输出	限制 LLM 返回固定字段	QueryVariants(BaseModel)
检索计划	序列化检索参数供诊断使用	RetrievalPlandataclass
数据隔离	封装多租户参数	DataScopedataclass

一、Pydantic 是什么

Pydantic是 Python 最流行的数据校验库。它的核心思想：用 Python 类型注解定义数据结构，运行时自动校验类型。

from pydantic import BaseModel

class User(BaseModel):
    name: str
    age: int
    email: str

# 正确数据
user = User(name="张三", age=28, email="zhangsan@example.com")

# 错误的类型 → 自动报错
user = User(name="张三", age="二十八", email="zhangsan@example.com")
# ValidationError: age: Input should be a valid integer

对比手写校验：

# 手写校验（容易遗漏、重复代码多）
def create_user(data):
    if "name" not in data:
        raise ValueError("name is required")
    if not isinstance(data["name"], str):
        raise ValueError("name must be string")
    if "age" not in data:
        raise ValueError("age is required")
    # ... 20 行校验代码 ...
    return {"name": data["name"], "age": int(data["age"])}

# Pydantic（一行声明，自动校验）
class User(BaseModel):
    name: str
    age: int

二、BaseModel 核心能力

2.1 基础类型校验

from pydantic import BaseModel

class RetrievalDebugRequest(BaseModel):
    query: str                      # 必填，字符串
    session_id: str | None = None   # 可选，默认 None
    scenario_id: str | None = None
    source_filter: str | None = None
    tenant_id: str = "default"      # 可选，默认值 "default"

当 FastAPI 收到一个 JSON 请求体时：

{"query": "入职流程", "tenant_id": 123}

Pydantic 会自动校验： -query是str✅ -tenant_id是int但期望str❌ → 返回 422 错误，附带清晰描述

2.2 Field 约束

from pydantic import BaseModel, Field

class RetrievalDebugRequest(BaseModel):
    query: str = Field(
        ...,                      # ... 表示必填
        min_length=1,             # 至少 1 个字符
        max_length=2000,          # 最多 2000 字符
        description="用户问题"     # 用于生成 OpenAPI 文档
    )
    session_id: str | None = Field(
        default=None,
        max_length=64,
    )
    tenant_id: str = Field(
        default="default",
        pattern=r"^[a-z0-9_]+$",  # 只允许小写字母+数字+下划线
    )

Field 支持的常用约束：

约束	说明
min_length/max_length	字符串长度范围
ge/le/gt/lt	数值范围（≥/≤/>/<）
pattern	正则表达式匹配
description	OpenAPI 文档描述
examples	OpenAPI 示例值

2.3 嵌套模型

class DataScope(BaseModel):
    tenant_id: str = "default"
    dataset_id: str = "default"

class RetrievalDebugRequest(BaseModel):
    query: str
    data_scope: DataScope  # 嵌套模型

# JSON 请求
{
    "query": "入职流程",
    "data_scope": {
        "tenant_id": "company_a",
        "dataset_id": "production"
    }
}

三、BaseSettings — 环境变量管理

这是 Pydantic 的一个特殊子类，专门用于管理配置：

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # 自动从环境变量或本机 .env 文件读取
    llm_api_key: str = ""
    milvus_uri: str = "http://127.0.0.1:19530"
    mysql_port: int = 3306
    api_rate_limit_per_minute: int = 120

    model_config = {
        "env_file": ".env",        # 本机调试读取 .env 文件
        "env_file_encoding": "utf-8",
    }

# 使用
settings = Settings()
print(settings.milvus_uri)  # 优先环境变量，其次 .env，最后默认值

# 优先级
# 1. 环境变量 export MILVUS_URI=http://prod:19530
# 2. 本机 .env 文件中的 MILVUS_URI=http://prod:19530
# 3. 代码中的默认值 "http://127.0.0.1:19530"

上面是 Pydantic 的通用示例。本项目的约定更具体：本机 API 调试时，Settings从进程环境变量和项目根目录.env读取；Docker Compose 部署时，Compose 读取.env.compose并注入到 API 容器的进程环境变量里。因此不要把.env.compose复制成.env使用，也不要把.env里的localhost配置拿去跑 API 容器。

本项目中的 Settings 实例被@lru_cache缓存为全局单例：

# qa_core/config/settings.py
from functools import lru_cache

@lru_cache(maxsize=1)
def get_settings() -> Settings:
    return Settings()

# 任何模块导入即可获取同一份配置
from qa_core.config.settings import get_settings
settings = get_settings()  # 整个进程只加载一次 .env

四、with_structured_output — LLM 输出约束

这是本项目中 Pydantic 最高级的用法：让 LLM 按指定结构返回结果。

from pydantic import BaseModel, Field

class QueryVariants(BaseModel):
    variants: list[str] = Field(
        default_factory=list,
        min_length=1,
        max_length=3,
        description="查询变体列表",
    )

# 使用 LangChain 的 with_structured_output
model = ChatOpenAI(...).with_structured_output(QueryVariants)
decision = model.invoke([...])  # 返回 QueryVariants 对象

# 不是返回一段自由文本
# 而是返回 QueryVariants(variants=["入职流程", "新人入职步骤", ...])

工作原理：LangChain 将 Pydantic 模型的 JSON Schema 嵌入 System Prompt，告诉 LLM 按这些字段回答。LLM 返回的 JSON 会被 Pydantic 自动校验；如果字段类型或数量不符合约束，就会报错。

五、在本项目中的使用速查

Pydantic 类	定义位置	作用
Settings(BaseSettings)	qa_core/config/settings.py	读取运行时配置：本机.env或 Compose 注入的环境变量
RetrievalDebugRequest(BaseModel)	qa_core/schemas.py	/api/retrieval/debug请求体校验
FeedbackRequest(BaseModel)	qa_core/schemas.py	反馈请求校验
QueryVariants(BaseModel)	qa_core/pipeline/query_variants.py	查询变体结构化输出约束

---