综合小项目：用 Python 构建可审计的 AI 内容分流器

前面的系列已经覆盖了模型调用、消息历史、结构化输出、错误处理、工具边界、评估和可观测性。本篇把这些能力串成一个小项目：用户输入一段内容，模型判断它属于哪一类、给出理由和建议动作，程序负责校验、记录，并在高风险动作前要求人工确认。重点不是做一个“自动决策者”，而是学习如何把模型放进一条可检查、可追踪的程序流水线。

先定义项目边界

分流器只处理三个类别：question（问题）、feedback（反馈）和 other（其他）。模型必须返回固定 JSON：类别、置信度、简短理由和建议动作。建议动作只能是 reply、review 或 ignore。模型不能发送消息、修改数据库或执行命令；它只产生候选结果，真正的后续动作由普通 Python 代码决定。

这条边界很重要。自然语言模型适合提取和归纳，不适合直接拥有业务系统权限。把“理解”与“执行”分开后，即使模型返回了未知类别，程序也可以拒绝它，而不是把任意字符串传给下游系统。

准备环境和密钥

在独立虚拟环境中安装官方 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="替换为你可用的模型名称"

不要把真实密钥写入源码、测试样例或 Git。下面的程序使用 Responses API 的文本输出接口；如果你使用其他服务，应先对照该服务的官方文档确认客户端地址、模型名称和响应字段。

编写最小分流器

把下面内容保存为 content_router.py。程序把解析、校验、调用和审计记录拆成独立函数，便于以后用固定响应替换网络调用进行测试：

import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

from openai import OpenAI

ALLOWED_CATEGORIES = {"question", "feedback", "other"}
ALLOWED_ACTIONS = {"reply", "review", "ignore"}
AUDIT_FILE = Path("router-audit.jsonl")

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
model = os.environ["MODEL_NAME"]

INSTRUCTIONS = """
你是内容分流助手。只返回一个 JSON 对象，不要使用 Markdown 代码围栏。
字段必须是：category、confidence、reason、action。
category 只能是 question、feedback、other；action 只能是 reply、review、ignore。
confidence 是 0 到 1 之间的数字。无法确定时使用 other、较低置信度和 review。
涉及投诉升级、退款、隐私或安全事件时，action 必须是 review。
""".strip()


def ask_model(text: str) -> dict:
    last_error = None
    for attempt in range(3):
        try:
            response = client.responses.create(
                model=model,
                instructions=INSTRUCTIONS,
                input=text,
            )
            return json.loads(response.output_text)
        except (json.JSONDecodeError, OSError, Exception) as exc:
            last_error = exc
            if attempt < 2:
                time.sleep(2 ** attempt)
    raise RuntimeError(f"模型调用失败：{last_error}") from last_error


def validate(result: dict) -> dict:
    if not isinstance(result, dict):
        raise ValueError("模型结果必须是 JSON 对象")
    if result.get("category") not in ALLOWED_CATEGORIES:
        raise ValueError("category 不在允许集合中")
    if result.get("action") not in ALLOWED_ACTIONS:
        raise ValueError("action 不在允许集合中")
    confidence = result.get("confidence")
    if not isinstance(confidence, (int, float)) or not 0 <= confidence <= 1:
        raise ValueError("confidence 必须在 0 到 1 之间")
    if not isinstance(result.get("reason"), str) or not result["reason"].strip():
        raise ValueError("reason 必须是非空字符串")
    return {
        "category": result["category"],
        "confidence": float(confidence),
        "reason": result["reason"].strip(),
        "action": result["action"],
    }


def audit(text: str, result: dict | None, error: str | None) -> None:
    record = {
        "time": datetime.now(timezone.utc).isoformat(),
        "input_length": len(text),
        "result": result,
        "error": error,
    }
    with AUDIT_FILE.open("a", encoding="utf-8") as file:
        file.write(json.dumps(record, ensure_ascii=False) + "\n")


def main() -> int:
    text = " ".join(sys.argv[1:]).strip()
    if not text:
        print("用法：python content_router.py '待分流内容'")
        return 2
    try:
        result = validate(ask_model(text))
    except Exception as exc:
        audit(text, None, str(exc))
        print(f"处理失败：{exc}")
        return 1

    audit(text, result, None)
    print(json.dumps(result, ensure_ascii=False, indent=2))
    if result["action"] == "review":
        print("提示：该结果需要人工复核，程序不会自动执行后续动作。")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

这里的 responses.create 只负责请求模型，validate 才是程序信任结果的入口。即使提示词要求了固定字段，也不能省略校验，因为模型输出仍可能缺字段、类型错误或违反业务规则。audit 使用 JSON Lines 保存每次输入长度、结果和错误，后续可以按行统计失败率；为了减少敏感信息泄露，示例没有保存原文。

代码中的 except 写法是为了让示例保持短小，但在正式项目中不应把所有异常都混在一起。应根据 SDK 文档分别处理超时、认证失败、限流和服务端错误，并为不可重试的错误立即停止。重试也要有上限，且不能因为重试而重复执行真实业务动作。

增加人工确认和测试边界

当前程序只打印结果，因此没有真正的外部副作用。若下一步要把 reply 交给消息系统，应先加一道明确的审批函数：当动作是 review、置信度低于阈值，或输入命中隐私与安全关键词时，只写入待审核队列，不调用发送接口。审批记录还应包含操作者、时间和原始模型结果，便于追责和回滚。

网络调用不适合直接写进单元测试。可以把 ask_model 改成接收一个调用函数，并在测试中注入固定返回值，至少覆盖四类情况：合法结果、缺少字段、未知类别，以及无效置信度。这样测试验证的是业务边界，而不是某一次模型恰好返回了什么文字。

运行程序时，命令行参数会作为一段完整文本发送给模型：

1	python content_router.py "我想了解退款进度，但订单已经超过承诺时间"

不要把这里可能出现的输出当作固定结果。模型版本、提示词和服务端策略变化都会影响分类；真正可靠的验证方式是查看退出码、审计文件和测试集统计，而不是复制某次回答作为断言。

常见问题

为什么已经要求 JSON，还要 json.loads 和字段校验？ 因为提示词是约束，不是类型系统。解析保证它至少是 JSON，业务校验保证它符合程序允许的集合和类型，两者缺一不可。

为什么不让模型直接发送回复？ 因为发送消息是有副作用的动作。模型可以提出建议，但权限、审批和执行应该由确定性的代码控制，尤其要对投诉、退款、隐私和安全事件设置人工边界。

审计日志是否会泄露隐私？ 会有风险。示例只保存长度和结果；真实项目还要做字段脱敏、访问控制、保留期限和加密，并评估是否有必要保存模型原始输出。

低置信度是不是一定代表错误？ 不是。它只是一个需要复核的信号，不能代替真实评估。应准备带人工标签的测试集，按类别统计准确率、漏分流和误分流，再调整阈值。

小结

这个小项目的核心不是某个 SDK 调用，而是四层职责分离：模型负责理解，校验负责守住数据边界，审计负责留下证据，人工确认负责拦截高风险动作。沿着这条链路继续扩展时，可以替换检索器、增加评估集或接入真实队列，但不要跳过输入校验和副作用隔离。只要每一步都有明确输入、输出和失败路径，AI 功能就更容易测试、排错和逐步上线。

搜索时光笔记