综合小项目：用 Python 构建可复核的 AI 用户反馈主题聚合器

用户反馈很多时，逐条阅读容易遗漏共性，但让模型直接修改工单或自动决定产品优先级又过于危险。本篇把结构化输出、结果校验、证据引用和人工复核组合成一个小项目：模型只提出反馈主题，Python 负责检查编号、分组关系和置信度，最终结果仍然保留原始证据。

项目目标与边界

输入是一组带编号的用户反馈，输出是若干主题。每个主题包含名称、反馈编号、代表性证据和 0 到 1 的置信度。为了让示例保持简单，本项目规定一条反馈只归入一个主要主题；这不是普遍正确的业务规则，而是一个可以被程序检查的明确约束。

模型适合做语义归纳，例如判断“搜索结果出来得太慢”和“输入关键词后要等很久”是否属于同一类问题。模型不适合负责编号是否存在、同一条反馈是否被重复分组等确定性判断。因此要把两类职责分开：模型生成候选，代码验证契约。

先定义结果契约

先约定输出结构，再编写调用逻辑，可以避免后面拿一段自然语言去猜主题。示例结果的形状如下：

{
  "themes": [
    {
      "name": "搜索响应速度",
      "feedback_ids": ["F-001", "F-002"],
      "evidence": "用户反映搜索等待时间过长",
      "confidence": 0.86
    }
  ]
}

feedback_ids 只能引用输入中的编号，confidence 必须是 0 到 1 之间的数字，evidence 用来帮助审核者回到原始反馈核对。主题名称可以由模型生成，但不能代替证据，也不能被当成已经确认的产品结论。

最小可运行实现

下面的程序只使用 Python 标准库。demo_model 是离线演示适配器，返回固定的候选结果，不代表真实模型的运行结果；这样可以先验证校验和聚合流程，不需要网络或密钥。

from dataclasses import dataclass
import json


@dataclass(frozen=True)
class Feedback:
    id: str
    text: str


def demo_model(rows: list[Feedback]) -> dict:
    """离线演示：模拟模型提出主题候选。"""
    return {
        "themes": [
            {
                "name": "搜索响应速度",
                "feedback_ids": [rows[0].id, rows[1].id],
                "evidence": "两条反馈都描述了搜索等待时间过长",
                "confidence": 0.86,
            },
            {
                "name": "导出格式支持",
                "feedback_ids": [rows[2].id],
                "evidence": "用户希望导出结果增加新的文件格式",
                "confidence": 0.72,
            },
            {
                "name": "通知设置",
                "feedback_ids": [rows[3].id],
                "evidence": "用户希望能够关闭不需要的提醒",
                "confidence": 0.79,
            },
        ]
    }


def validate(result: dict, rows: list[Feedback]) -> dict:
    allowed = {row.id for row in rows}
    themes = result.get("themes")
    if not isinstance(themes, list) or not themes:
        raise ValueError("themes 必须是非空数组")

    used: set[str] = set()
    for theme in themes:
        name = theme.get("name")
        ids = theme.get("feedback_ids")
        evidence = theme.get("evidence")
        confidence = theme.get("confidence")
        if not isinstance(name, str) or not name.strip():
            raise ValueError("主题必须有非空 name")
        if not isinstance(ids, list) or not ids:
            raise ValueError("feedback_ids 必须是非空数组")
        if not set(ids) <= allowed:
            raise ValueError("主题引用了不存在的反馈编号")
        if used.intersection(ids):
            raise ValueError("一条反馈不能被重复分组")
        if not isinstance(evidence, str) or not evidence.strip():
            raise ValueError("主题必须保留 evidence")
        if not isinstance(confidence, (int, float)) or not 0 <= confidence <= 1:
            raise ValueError("confidence 必须在 0 到 1 之间")
        used.update(ids)

    missing = allowed - used
    if missing:
        raise ValueError(f"存在未分组反馈：{sorted(missing)}")
    return result


def summarize(result: dict) -> list[dict]:
    """派生统计由程序计算，不让模型直接填写。"""
    return [
        {
            "theme": theme["name"],
            "count": len(theme["feedback_ids"]),
            "confidence": theme["confidence"],
            "evidence": theme["evidence"],
        }
        for theme in result["themes"]
    ]


if __name__ == "__main__":
    rows = [
        Feedback("F-001", "搜索结果出来得太慢"),
        Feedback("F-002", "输入关键词后要等待很久"),
        Feedback("F-003", "导出时希望支持 CSV 格式"),
        Feedback("F-004", "希望可以关闭部分提醒"),
    ]
    result = validate(demo_model(rows), rows)
    output = {"themes": result["themes"], "summary": summarize(result)}
    print(json.dumps(output, ensure_ascii=False, indent=2))

把代码保存为 feedback_theme_aggregator.py 后运行：

1	python feedback_theme_aggregator.py

程序先验证原始反馈编号，再验证主题结构，最后才计算 summary。如果把 F-002 改成不存在的编号、删除一个主题，或者把置信度改成 1.2，程序都会在输出前抛出错误。这样的失败比把错误结果静默交给产品看板更容易发现。

如何接入真实模型

真实接入时，只替换 demo_model 这一层，让它接收反馈列表并返回 Python 字典；validate 和 summarize 不应绕过。模型请求需要使用服务商当前支持的结构化输出能力，解析失败就作为调用失败处理，而不是用字符串切片“修复”半截 JSON。密钥只能从环境变量读取，不能写进代码、示例配置或日志。

如果模型服务返回的不是严格 JSON，先记录可追踪的请求编号和错误类型，再按重试策略处理；不要把未经解析的文本直接存入主题表。生产环境还应保存模型名称、提示词版本、输入样本版本和校验结果，方便复盘某次聚合为什么发生变化。

结果为什么可复核

可复核不只是“把结果打印出来”。首先，主题只引用输入编号，审核者可以通过编号定位原文；其次，evidence 说明模型依据了哪类描述，但它本身仍然只是摘要，不能替代原文；最后，summary 由程序根据已验证的编号计算，因此主题数量不会因为模型漏算或重复计数而漂移。

实际批处理时，可以把每次输入和输出保存为一条不可变记录，并为记录生成请求编号。审核者修改主题时不要覆盖模型原始结果，而是新增人工结论和审核人信息。这样能够区分“模型当时怎么判断”和“后来人为什么改动”，也能把争议样本加入下一轮离线回归测试。

如果某个主题的置信度低于业务设定的阈值，可以直接进入待审核队列；阈值不是模型参数，而是业务策略，应该通过历史样本评估后确定。对于包含个人信息的反馈，发送模型前还应先做脱敏，日志只保存必要的编号、状态和错误代码。

常见问题

置信度高就能自动排优先级吗？ 不能。置信度只表示模型对归类的主观确信程度，不等于用户数量、影响范围或修复价值。优先级应结合业务指标，并保留人工确认。

为什么不让模型直接计算主题数量？ 数量是 feedback_ids 的长度，程序计算更准确，也避免模型同时承担生成和统计两个容易互相矛盾的任务。

一条反馈确实属于多个主题怎么办？ 把“只能出现一次”的规则改成允许重复，并增加 primary_theme 或权重字段；修改契约后要补充对应测试，不能只放宽校验而不改变审核界面。

演示适配器是不是没有真正使用 AI？ 是。它只用于离线验证控制流，没有伪造线上模型结果。真实模型只应替换适配层，输入、校验、证据保留和人工复核边界仍然适用。

小结

这个小项目的重点不是让模型自动决定产品路线，而是把一次开放式归纳限制在可检查的范围内：模型提出主题，程序验证引用关系，统计字段由代码派生，审核者可以沿着 evidence 回到原始反馈。把生成、校验和业务决策分开，才能在更换模型或增加数据量时保持结果可追溯、可回放和可修正。

搜索时光笔记