RAG附录Pydantic 数据校验与 Settings 管理

附录A：Pydantic 数据校验与 Settings 管理

为什么需要本章

Pydantic 在本项目中被五处关键位置使用，但主体笔记从未系统说明它是什么、怎么工作：

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

一、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	查询变体结构化输出约束

返回笔记开头 ↑