综合小项目：用 Python 构建带脱敏与复核的 AI 文本流水线

前面的系列已经覆盖了模型调用、提示词、结构化输出、错误处理、评估、安全边界和人工审批。本篇用一个综合小项目把这些知识串起来：读取一段待处理文本，先在本地遮盖明显的邮箱和手机号，再请求模型生成结构化摘要，最后把结果保存为“待复核”记录。项目不自动发布、不调用外部工具，重点是理解一条 AI 流水线如何划分信任边界。

先定义数据流和边界

程序分成五步：输入校验、脱敏、模型调用、输出校验、人工复核。脱敏发生在离开本机之前，模型只接收脱敏后的文本；模型输出只是候选摘要，不能直接当成最终内容；写入结果文件是唯一的副作用，并且只保存待复核状态。

这里的脱敏规则只覆盖示例中的邮箱和中国大陆手机号，不能宣称已经满足完整合规要求。真实项目还要根据数据类型、法律要求和业务流程设计规则，并用脱敏测试集验证。安全的第一步不是增加复杂框架，而是明确哪些数据可以发送、哪些动作必须由程序或人控制。

准备环境和输入文件

在独立虚拟环境中安装官方 Python SDK：

1
2
3

python -m venv .venv
source .venv/bin/activate
python -m pip install openai

密钥和模型名只从环境变量读取：

1 2	export OPENAI_API_KEY="替换为你的真实密钥" export MODEL_NAME="替换为你可用的模型名称"

新建 input.txt，写入一段普通文本。不要把真实敏感资料直接复制到练习项目；可以使用下面的测试内容：

1	客户李明（邮箱 liming@example.com，电话 13812345678）反馈：昨天导出的报表缺少三列，希望今天确认处理计划。

编写最小流水线

将下面代码保存为 pipeline.py：

import json
import os
import re
import sys
from pathlib import Path

from openai import OpenAI

EMAIL = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
PHONE = re.compile(r"(?<!\d)1[3-9]\d{9}(?!\d)")


def redact(text: str) -> str:
    text = EMAIL.sub("[邮箱已脱敏]", text)
    return PHONE.sub("[手机号已脱敏]", text)


def validate_input(text: str) -> None:
    if not text.strip():
        raise ValueError("输入文本不能为空")
    if len(text) > 6000:
        raise ValueError("输入文本超过 6000 个字符")


def ask_model(client: OpenAI, text: str) -> dict:
    response = client.responses.create(
        model=os.environ["MODEL_NAME"],
        instructions=(
            "你是文本整理助手。只返回 JSON 对象，不要 Markdown。"
            "字段必须是 summary（不超过80字的中文摘要）、"
            "priority（low、normal 或 high）、needs_review（布尔值）。"
        ),
        input=f"请整理下面的文本，不要补充原文没有的事实：\n{text}",
        temperature=0,
    )
    raw = response.output_text
    try:
        result = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise ValueError("模型返回的内容不是合法 JSON") from exc

    expected = {"summary", "priority", "needs_review"}
    if set(result) != expected:
        raise ValueError("模型输出字段不符合约定")
    if not isinstance(result["summary"], str) or not result["summary"].strip():
        raise ValueError("summary 必须是非空字符串")
    if len(result["summary"]) > 80:
        raise ValueError("summary 超过 80 个字符")
    if result["priority"] not in {"low", "normal", "high"}:
        raise ValueError("priority 不是允许的值")
    if not isinstance(result["needs_review"], bool):
        raise ValueError("needs_review 必须是布尔值")
    return result


def main() -> None:
    if len(sys.argv) != 3:
        raise SystemExit("用法：python pipeline.py input.txt review.json")
    source_path, output_path = map(Path, sys.argv[1:])
    original = source_path.read_text(encoding="utf-8")
    validate_input(original)
    safe_text = redact(original)

    client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
    result = ask_model(client, safe_text)
    record = {
        "status": "待人工复核",
        "input_was_redacted": safe_text != original,
        "source_text": safe_text,
        "candidate": result,
    }
    output_path.write_text(
        json.dumps(record, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    print(f"已生成待复核记录：{output_path}")


if __name__ == "__main__":
    main()

运行：

1	python pipeline.py input.txt review.json

示例不会预先写死模型的摘要，因此不能凭空给出固定运行结果。实际可验证的结果是：模型收到的请求文本中应出现脱敏标记，review.json 应包含 status、脱敏后的输入和经过校验的候选结果。若网络、认证或模型配置不正确，程序会在请求阶段失败，而不是生成一份看似成功的假记录。

为什么要分成这些函数

redact 是发送前的本地边界。它减少了明显敏感信息泄露的机会，但不是万能的隐私检测器；姓名、地址、订单号等内容可能仍然存在，所以业务上仍需决定哪些文本允许离开本机。

ask_model 同时完成请求和结构化校验，但没有执行任何外部动作。模型返回的 JSON 即使格式正确，也只代表“能被解析”，不代表摘要事实正确。生产代码可把校验函数单独拆出，并加入事实抽检、引用检查或离线评估。

record 明确保存 待人工复核 状态。后续如果要发布、发送通知或写入业务系统，应增加独立的确认步骤，不能因为 needs_review 为 false 就绕过审批。模型可以提出候选结果，程序负责验证格式和权限，人负责处理无法自动判断的事实与风险。

常见问题

脱敏后模型还能理解内容吗？ 通常仍能理解事件、时间和处理诉求，但被替换的身份信息无法恢复。若某个任务必须使用原始标识，应采用受控映射和最小权限方案，而不是把原文无条件发送给模型。

为什么还要校验 JSON？ 提示词中的“只返回 JSON”是请求，不是安全保证。模型可能输出解释文字、缺字段或错误枚举值。解析、字段集合、类型和长度检查，才能把不符合契约的结果挡在副作用之前。

失败时应该自动重试吗？ 超时或限流可以在有上限的情况下重试，并使用指数退避；认证失败、输入超限和输出契约错误通常不应盲目重试。重试记录还应包含次数和错误类型，避免重复计费却无法追踪。

能不能直接把 review.json 当作最终结果？ 不能。它保存的是候选内容和脱敏输入，status 明确表示尚未批准。真正的发布流程应由人查看摘要是否忠实、优先级是否合理、是否包含不应公开的信息，再由确定性的程序完成后续动作。

小结

这个小项目把一条 AI 功能拆成了可检查的边界：数据先在本地脱敏，模型只生成候选结构，程序严格校验，结果进入人工复核而不是直接产生副作用。它没有解决所有隐私、事实和合规问题，却展示了一个可靠的起点。继续扩展时，可以加入脱敏单元测试、固定评估样例、有限重试和审批日志，但每次都应保持“模型建议、程序约束、人来确认”的职责划分。

搜索时光笔记