综合小项目：用 Python 构建可复核的 AI 数据字典生成器

前面的系列已经覆盖了模型调用、结构化输出、错误处理、评估和安全边界。现在把这些知识组合成一个小项目：输入一组表头和少量样例行，让模型生成数据字典候选稿，再由 Python 检查字段是否完整、类型是否受限、说明是否为空。程序只输出“待复核”的元数据，不修改原始 CSV，也不把模型的猜测直接当成正式数据标准。

先定义输入和边界

数据字典通常至少要回答四个问题：字段叫什么、可能是什么类型、业务含义是什么、是否包含敏感信息。模型可以根据字段名和样例值提出候选答案，但样例很少时容易误判。例如 id 可能是整数，也可能是带前导零的字符串；邮箱看起来像个人信息，但最终的分级仍应由业务规则确认。

因此本项目只接受“表头 + 样例行”两类输入，输出每个字段的名称、类型、说明、是否可为空、敏感级别和判断依据。字段名称必须与输入完全一致，类型和敏感级别只能从预先定义的枚举中选择。任何解析失败或字段缺失，都进入人工复核，而不是自动补一个默认值。

准备环境

创建虚拟环境并安装官方 Python SDK 与 Pydantic：

python -m venv .venv
source .venv/bin/activate
python -m pip install openai pydantic
export OPENAI_API_KEY="替换为你的真实密钥"
export MODEL_NAME="替换为你可用的模型名称"

密钥只通过环境变量读取，不能写进源码、文章或日志。MODEL_NAME 也放在环境变量中，便于更换模型而不修改业务代码。下面的示例使用当前 OpenAI Python SDK 的 Responses API，并让 SDK 按 Pydantic 模型解析结构化结果；本地仍然要做业务校验。

编写最小程序

新建 data_dictionary.py：

import json
import os
import sys
from typing import Literal

from openai import OpenAI
from pydantic import BaseModel, ConfigDict

DataType = Literal["string", "integer", "number", "boolean", "date", "unknown"]
Sensitivity = Literal["none", "personal", "financial", "secret", "unknown"]


class ColumnInfo(BaseModel):
    model_config = ConfigDict(extra="forbid")

    name: str
    data_type: DataType
    description: str
    nullable: bool
    sensitivity: Sensitivity
    evidence: str


class DictionaryDraft(BaseModel):
    model_config = ConfigDict(extra="forbid")

    columns: list[ColumnInfo]


HEADERS = ["order_id", "customer_email", "amount", "created_at"]
SAMPLE_ROWS = [
    {"order_id": "A001", "customer_email": "demo@example.com", "amount": 19.9, "created_at": "2026-11-04"},
    {"order_id": "A002", "customer_email": "user@example.com", "amount": 8.5, "created_at": "2026-11-05"},
]


def ask_model(client: OpenAI, headers: list[str], rows: list[dict]) -> DictionaryDraft:
    prompt = (
        "根据表头和样例行生成数据字典候选稿。只能描述输入中已有的字段，"
        "不要新增字段，不要修改字段名。样例不足时使用 unknown，并在 evidence 中说明原因。\n"
        f"表头：{json.dumps(headers, ensure_ascii=False)}\n"
        f"样例行：{json.dumps(rows, ensure_ascii=False)}"
    )
    response = client.responses.parse(
        model=os.environ["MODEL_NAME"],
        input=[
            {"role": "system", "content": "你是数据治理助手，只输出符合给定结构的数据字典候选稿。"},
            {"role": "user", "content": prompt},
        ],
        text_format=DictionaryDraft,
    )
    if response.output_parsed is None:
        raise ValueError("模型没有返回可解析的结构化结果")
    return response.output_parsed


def validate(draft: DictionaryDraft, headers: list[str]) -> DictionaryDraft:
    names = [column.name for column in draft.columns]
    if len(names) != len(set(names)):
        raise ValueError("输出包含重复字段")
    if set(names) != set(headers):
        raise ValueError("输出字段与输入表头不一致")
    for column in draft.columns:
        if not column.description.strip():
            raise ValueError(f"字段 {column.name} 缺少说明")
        if not column.evidence.strip():
            raise ValueError(f"字段 {column.name} 缺少判断依据")
    return draft


def main() -> None:
    if "--self-test" in sys.argv:
        sample = DictionaryDraft(
            columns=[
                ColumnInfo(
                    name=name,
                    data_type="unknown",
                    description="待人工确认",
                    nullable=True,
                    sensitivity="unknown",
                    evidence="样例不足，暂不下结论",
                )
                for name in HEADERS
            ]
        )
        print(json.dumps(validate(sample, HEADERS).model_dump(), ensure_ascii=False))
        return

    if not os.environ.get("OPENAI_API_KEY") or not os.environ.get("MODEL_NAME"):
        raise SystemExit("请先设置 OPENAI_API_KEY 和 MODEL_NAME")
    try:
        draft = validate(ask_model(OpenAI(), HEADERS, SAMPLE_ROWS), HEADERS)
    except Exception as exc:
        print(json.dumps({"status": "needs_review", "reason": str(exc)}, ensure_ascii=False))
        return
    print(json.dumps({"status": "ready_for_review", "draft": draft.model_dump()}, ensure_ascii=False))


if __name__ == "__main__":
    main()

这里有三道边界。第一，Pydantic 的 Literal 限制了类型和敏感级别，extra="forbid" 防止模型偷偷添加未定义字段。第二，validate 重新比较输入表头和输出字段，即使结构解析成功，也不会接受字段遗漏、重复或改名。第三，异常只会输出复核状态；生产系统可以把原始文件标识、模型名和错误类型写入受控队列，但不要把整份敏感样例直接写进普通日志。

如何验证

先做不需要网络的语法和规则检查：

1 2	python -m py_compile data_dictionary.py python data_dictionary.py --self-test

--self-test 使用本地构造的未知字段草稿，验证每个输入字段都能通过校验。接着再设置环境变量运行真实调用。第一次运行只应把结果保存为候选稿，由人确认 customer_email 的敏感级别、order_id 的真实类型以及空值规则；不要因为模型给出了完整 JSON 就跳过这一步。

还可以为 validate 写离线测试：把字段名改成不存在的值，期望抛出“字段与输入表头不一致”；把 description 改为空字符串，期望抛出“缺少说明”。这些测试不消耗 API 额度，却能守住最重要的确定性规则。模型调用测试则应使用脱敏样例，并记录提示词版本和模型版本，避免把一次成功响应误认为长期稳定性证明。

常见问题

为什么不让模型直接读取整个 CSV？ 数据字典生成只需要必要的表头和少量脱敏样例。先限制输入范围，可以降低隐私暴露、上下文超限和无关内容干扰。

unknown 是否表示模型失败？ 不一定。它也可能是诚实的结果，说明当前样例不足以支持更具体的判断。相比强行猜测，保留未知状态更适合后续人工确认。

为什么同时保存 description 和 evidence？ 说明是面向使用者的候选定义，依据是帮助审核者回看模型为何这样判断。依据不是事实证明，最终的数据标准仍应由数据负责人确认。

模型返回的字段顺序不同会出错吗？ 不会。本地校验比较的是字段集合和重复情况，不依赖顺序。如果下游需要固定顺序，应在通过校验后按原始 headers 重新排列，而不是要求模型记住顺序。

小结

这个综合项目的核心不是让模型自动建立数据标准，而是把模型限制在“提出候选元数据”的位置。Responses API 和 Pydantic 负责解析结构，Python 负责字段集合、枚举、非空和异常兜底，人工负责最终确认。今后接入真实 CSV、数据库或数据目录时，也应沿用同一原则：最小化输入，保留证据，先校验，再发布。

搜索时光笔记