综合小项目：用 Python 构建可审计的 AI FAQ 生成器

前面的基础路线已经覆盖了 API 调用、消息历史、上下文、生成参数、结构化输出、错误处理和评估。本篇不再引入新的框架，而是把这些知识组合成一个小而完整的工具：读取一组原始问答，让模型生成统一格式的 FAQ 草稿，并把每条记录的输入、结果和错误写入日志。它不是自动发布器，最终内容仍由人审核；重点是让一次模型调用变成可检查、可重试、可追踪的处理流程。

先定义输入、输出和边界

输入文件使用 JSONL，每行是一条原始问答：

1
2

{"id":"q-001","question":"如何修改通知设置？","answer":"打开设置，进入通知页面，选择需要接收的通知类型并保存。"}
{"id":"q-002","question":"导出失败怎么办？","answer":"先确认账号有权限，并检查磁盘空间；仍然失败时记录错误时间，联系管理员。"}

程序输出同样使用 JSONL。成功记录包含稳定的 id、FAQ 标题、答案、关键词和 status；失败记录只保存错误类型，不把原文重复写进日志。id 是幂等键：再次运行时，已经成功的项目会跳过，失败的项目可以重新尝试。

这里把模型定位为“草稿生成器”，而不是事实来源。模型只能重组输入中的信息，不能补充输入没有依据的政策、价格或承诺。程序也不会直接发布结果，审核者需要确认答案是否忠实、是否包含敏感信息，以及标题是否适合公开展示。

准备虚拟环境和依赖。示例使用 OpenAI Python SDK 的 Responses API；如果使用其他服务，应按对应官方文档调整客户端和响应读取方式。

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

密钥只从环境变量读取，不能写在源码、输入文件或结果文件中。

让模型返回固定结构

模型输出要进入程序，就不能只依赖一段自然语言。下面通过提示词要求 JSON，并在本地做第二层校验。提示词不是安全边界，json.loads、字段检查和长度限制才是程序继续执行前的实际门槛。

新建 faq_generator.py：

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

from openai import OpenAI

INPUT = Path("raw_qa.jsonl")
OUTPUT = Path("faq_results.jsonl")
MODEL = os.environ["MODEL_NAME"]
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])


def read_done() -> set[str]:
    done = set()
    if not OUTPUT.exists():
        return done
    for line in OUTPUT.read_text(encoding="utf-8").splitlines():
        try:
            record = json.loads(line)
        except json.JSONDecodeError:
            continue
        if record.get("status") == "ok":
            done.add(record.get("id"))
    return done


def generate_faq(question: str, answer: str) -> dict:
    prompt = f"""根据下面的原始问答生成 FAQ 草稿。
只能使用原始答案中的事实，不要猜测或扩展政策。
只返回一个 JSON 对象，不要 Markdown 代码围栏，字段必须是：
{{"title":"不超过 30 字的标题","answer":"不超过 120 字的答案","keywords":["关键词1","关键词2"]}}

原始问题：{question}
原始答案：{answer}
"""
    response = client.responses.create(
        model=MODEL,
        instructions="你是一个严谨的知识库编辑，只整理输入事实。",
        input=prompt,
    )
    raw = response.output_text.strip()
    result = json.loads(raw)
    if not isinstance(result, dict):
        raise ValueError("模型结果不是 JSON 对象")
    required = {"title", "answer", "keywords"}
    if set(result) != required:
        raise ValueError("JSON 字段不符合约定")
    if not isinstance(result["title"], str) or not result["title"]:
        raise ValueError("title 必须是非空字符串")
    if not isinstance(result["answer"], str) or not result["answer"]:
        raise ValueError("answer 必须是非空字符串")
    if not isinstance(result["keywords"], list):
        raise ValueError("keywords 必须是数组")
    if len(result["title"]) > 30 or len(result["answer"]) > 120:
        raise ValueError("文本超过长度限制")
    if not all(isinstance(word, str) and word for word in result["keywords"]):
        raise ValueError("keywords 中必须都是非空字符串")
    return result


def call_with_retry(question: str, answer: str, attempts: int = 3) -> dict:
    last_error = None
    for attempt in range(attempts):
        try:
            return generate_faq(question, answer)
        except (json.JSONDecodeError, ValueError) as error:
            # 结构不合格时重试有限次数，避免无限请求。
            last_error = error
        except Exception as error:
            last_error = error
        if attempt + 1 < attempts:
            time.sleep(2 ** attempt)
    raise RuntimeError("生成或校验达到重试上限") from last_error


def append_record(record: dict) -> None:
    with OUTPUT.open("a", encoding="utf-8") as file:
        file.write(json.dumps(record, ensure_ascii=False) + "\n")
        file.flush()


def main() -> None:
    done = read_done()
    for line in INPUT.read_text(encoding="utf-8").splitlines():
        item = json.loads(line)
        item_id = item["id"]
        if item_id in done:
            print(f"跳过已完成：{item_id}")
            continue
        try:
            faq = call_with_retry(item["question"], item["answer"])
        except Exception as error:
            append_record({
                "id": item_id,
                "status": "error",
                "error_type": type(error).__name__,
                "created_at": datetime.now(timezone.utc).isoformat(),
            })
            print(f"处理失败：{item_id}")
            continue
        append_record({
            "id": item_id,
            "status": "ok",
            "faq": faq,
            "created_at": datetime.now(timezone.utc).isoformat(),
        })
        done.add(item_id)
        print(f"已完成：{item_id}")


if __name__ == "__main__":
    main()

运行前把示例输入保存为 raw_qa.jsonl，然后执行 python faq_generator.py。由于模型输出具有随机性，不能预先保证某个具体标题或关键词；可以验证的是，成功记录必须通过 JSON 解析、字段、类型和长度检查，失败记录不会被误标为成功。再次运行时，程序会根据结果文件中的成功 id 跳过已完成项目。

代码中的几个关键设计

generate_faq 负责一次模型调用和结果校验，main 负责遍历、恢复和落盘。把职责拆开后，可以在不访问网络的测试中替换 generate_faq，也能单独测试 JSON 校验规则。模型输出即使看起来合理，也必须经过本地校验，不能把提示词当成类型系统。

输出采用追加 JSONL，并在每条记录后 flush。进程中断时，之前已保存的记录仍然存在。这里的恢复逻辑只把 status 为 ok 的记录视为完成；临时错误会在下一次运行时重新请求。若输入可能出现重复 id，生产实现应在启动时检测并拒绝重复，或使用带唯一约束的 SQLite 表。

重试只围绕生成和校验，最多三次，等待时间逐步增加。实际应用要进一步区分认证错误、参数错误、超时和限流：密钥错误通常不应重试，限流和暂时网络错误才适合有限退避。无论哪种错误，都应限制总请求数和预算，并避免把原始内容写入普通日志。

如何做最小人工审核

结果文件可以先由一个审核脚本或人工检查处理，而不是直接同步到知识库。审核至少包括四项：标题是否准确概括问题；答案是否只来自原始答案；关键词是否便于检索；内容是否泄露个人信息、内部地址或不应公开的操作细节。审核结果可以写入另一个文件，使用 approved、rejected 和 needs_review 等状态，千万不要用模型自己的判断替代高风险内容的人工确认。

还可以准备一组固定样例做回归测试：正常问答、空答案、超长答案、含换行的答案、重复 ID，以及模型返回缺字段或额外字段的情况。测试重点不是比较模型每个字都相同，而是检查程序是否正确拒绝不合格结构、是否保持输入事实边界、是否能从中断处继续。

常见问题

模型偶尔返回 Markdown 代码围栏怎么办？ 本例会让 json.loads 失败并重试。不要无条件截取首尾字符来“修复”所有结果，这可能把错误内容伪装成合法 JSON；若确实需要修复，应记录原始响应并设置严格规则。

为什么不把原始问答保存到结果里？ 结果文件可能被备份、共享或上传。示例只保留 id 和生成结果，减少敏感信息扩散。需要审计时，应使用受控存储和明确的保留期限。

生成成功是否代表 FAQ 正确？ 不是。结构合法只说明它能被程序读取，不说明事实正确。模型可能遗漏条件、改变语气或把上下文理解错，因此必须进行抽样审核和针对业务的评估。

能不能并发处理提高速度？ 可以，但并发会增加限流、成本和重复写入风险。先用单进程版本验证恢复和校验，再引入有上限的并发，并为结果存储增加唯一约束和可观测指标。

小结

这个 FAQ 生成器把一次 API 调用扩展成了一个可验证的小工作流：输入用稳定 ID 标识，提示词约束输出，Python 做 JSON 和业务字段校验，重试只处理暂时失败，JSONL 让进度可以恢复，人工审核负责最终质量。综合项目的核心不是让模型自动完成全部工作，而是明确模型负责生成候选内容，程序负责边界、状态和副作用，人负责重要事实与发布决策。沿着这个思路，后续增加数据库、队列或网页界面时，基础行为仍然可测试、可审计。

搜索时光笔记