把自然语言整理成固定字段，是 AI 应用中很常见的一类任务：从邮件提取客户、金额和截止日期，从工单提取优先级和问题类型。真正可靠的程序不能只打印模型文本，而要让结果经过解析、校验和复核。本篇把结构化输出、提示词约束和基础安全边界组合成一个最小数据抽取器，并提供离线演示，读者无需密钥也能先验证程序流程。

先定义输入和边界

本项目从一段订单留言中提取三个字段：customer、amount 和 currency。字段必须是字符串、数字和三字母货币代码；缺失信息使用 null，不能根据常识猜测。模型只负责把文字转换成候选 JSON，程序负责校验，最终是否写入业务系统仍由人或后续规则决定。

这个边界很重要。模型返回的是“不可信输入”，即使它看起来像合法 JSON，也可能少字段、类型错误，或者把用户文字中的指令当成程序要求。把解析和校验放在模型外部，才能让失败变得可见、可测试。

用纯 Python 实现最小流水线

先创建文件 extractor.py。下面的 demo_model 模拟模型响应，故意保留一个不合法的金额类型，用来验证校验逻辑。真实项目中，只需要替换这个函数为已有的模型适配器；不要改变校验和复核层。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
import json
import re
from typing import Any

SCHEMA_KEYS = {"customer", "amount", "currency"}


def build_prompt(text: str) -> str:
    return f"""请从订单留言中提取 JSON，只允许包含 customer、amount、currency。
缺失信息填 null，不要猜测；amount 必须是数字，currency 必须是三字母大写代码。
订单留言：\n{text}"""


def demo_model(prompt: str) -> str:
    # 演示模型可能返回的候选结果；生产环境替换为 API 调用。
    return json.dumps({
        "customer": "林晓",
        "amount": "1280",
        "currency": "CNY",
    }, ensure_ascii=False)


def validate_candidate(raw: str) -> tuple[dict[str, Any] | None, list[str]]:
    errors: list[str] = []
    try:
        value = json.loads(raw)
    except json.JSONDecodeError:
        return None, ["响应不是合法 JSON"]

    if not isinstance(value, dict):
        return None, ["顶层结果必须是对象"]
    if set(value) != SCHEMA_KEYS:
        errors.append("字段必须且只能包含 customer、amount、currency")

    customer = value.get("customer")
    amount = value.get("amount")
    currency = value.get("currency")
    if customer is not None and not isinstance(customer, str):
        errors.append("customer 必须是字符串或 null")
    if amount is not None and (isinstance(amount, bool) or not isinstance(amount, (int, float))):
        errors.append("amount 必须是数字或 null")
    if currency is not None and (
        not isinstance(currency, str) or not re.fullmatch(r"[A-Z]{3}", currency)
    ):
        errors.append("currency 必须是三字母大写代码或 null")
    return (value if not errors else None), errors


def extract(text: str, approve: bool = False) -> dict[str, Any]:
    prompt = build_prompt(text)
    raw = demo_model(prompt)
    candidate, errors = validate_candidate(raw)
    if errors:
        return {"status": "rejected", "errors": errors, "raw": raw}
    if not approve:
        return {"status": "pending_review", "data": candidate}
    return {"status": "approved", "data": candidate}


if __name__ == "__main__":
    message = "林晓留言：请按 1280 CNY 处理本月订单。"
    print(json.dumps(extract(message), ensure_ascii=False, indent=2))

运行命令如下：

1
python extractor.py

程序会输出 pending_review，而不是直接声称数据可信。把 extract(message, approve=True) 作为另一次明确操作，才会得到 approved。示例中的模型响应金额是字符串，因此校验会返回 rejected；这正是验证层应该捕获的问题，而不是悄悄转换后继续写库。

接入真实模型时保留适配层

真实调用可以放在独立的 model_adapter.py 中。密钥只能从环境变量读取，模型名称也不要硬编码成项目业务逻辑的一部分：

1
2
3
4
5
6
7
8
9
10
import os

API_KEY = os.environ["AI_API_KEY"]
MODEL_NAME = os.environ["AI_MODEL_NAME"]


def call_model(prompt: str) -> str:
    # 在这里使用已验证过的服务商 SDK，返回模型的文本内容。
    # 不要把 API_KEY 打印到日志，也不要写入仓库。
    raise NotImplementedError("请按服务商官方文档实现调用")

适配器只负责请求和取回文本，validate_candidate 不应依赖某个服务商的响应对象。这样可以在没有网络或密钥时使用 demo_model 做单元测试，也能在更换模型服务时避免修改业务规则。对模型响应设置超时、捕获限流和网络异常，则属于调用层的责任；异常时应返回可识别的失败状态，不要伪造空数据。

人工复核和日志应该记录什么

抽取结果进入数据库前，至少要展示原文、候选 JSON、校验错误和模型版本，供复核者判断。日志可以记录请求编号、耗时、状态和错误类型，但应先脱敏；客户姓名、订单号等个人或业务信息不要无条件写入普通日志。复核通过后再执行写入，写入函数也应再次检查字段，而不是相信上游已经校验过。

还可以为校验函数编写几个不依赖模型的测试：合法数字通过；金额为字符串拒绝；额外字段拒绝；货币代码为小写拒绝；缺失字段返回错误。模型输出具有随机性，测试应固定候选文本，不能把某一次生成结果当作稳定契约。

常见问题

为什么不直接让模型返回一句自然语言？ 因为自然语言难以稳定解析，也无法明确区分缺失值和猜测值。固定 JSON 只是第一步，外部校验才是边界。

校验失败后要不要自动修复？ 简单格式错误可以请求模型重试，但要限制次数，并重新完整校验。金额、身份和权限等高风险字段不要只依赖自动修复，应该转人工。

为什么示例不直接连接 API？ 离线演示能验证核心程序，不需要暴露密钥，也不会把某个服务商的易变参数误当成通用接口。接入时应以所选服务商当前官方文档为准。

小结

一个可用的数据抽取器至少有四层：提示词规定目标，模型产生候选文本，程序解析并校验，人工或业务规则决定是否采纳。把模型当作不可信输入源，把确定性规则放在代码中，才能让结构化输出从“看起来正确”变成“可以复核、可以测试、可以拒绝”。