问题

在AI回答采集系统中，原始回答里混入了大量无效内容：模型拒绝回答、输出无关信息、重复生成、格式异常或回答被截断。如果不先过滤，后续的提及率、推荐率计算都会失真。

这篇文章只解决一个问题：如何在数据处理管道中，系统性地过滤掉无效回答，保证进入统计环节的数据是有效、可分析的。
无效回答类型

根据常见情况，无效回答主要分三类，其中拒绝回答和格式异常出现频率最高：

拒绝回答：模型明确表示无法回答，例如“抱歉，我无法回答这个问题”“I cannot provide that information”。这类回答通常较短，关键词模式明显。
无关回答：回答内容与问题完全不相关，比如问品牌A，回答却一直在讲品牌B。这类回答在实体层面与问题无交集。
格式异常：包括空字符串、纯符号、超长无意义文本、乱码或截断。截断回答虽然可能包含部分有效信息，但信息不完整，不应进入统计。

重复回答在批量采集时也常见，但更多是同一批次内的问题，不同批次间重复较少，因此重复检测的窗口需要控制。
过滤规则设计

过滤规则需要平衡召回率和误杀率。以下是一组常用规则，阈值可根据实际数据调整：

1. 长度过滤

最小长度设为10个字符，低于此值的内容几乎不可能是有效回答。最大长度设为模型最大输出上限的90%，超过此值说明回答被截断。不同模型的输出上限不同，例如GPT-4的max_tokens常见为4096，Claude为8192，因此阈值需要根据模型分别配置。截断回答单独存储，不进入统计，但可人工复核。

2. 关键词匹配

维护一个拒绝关键词列表，例如：

rejection_keywords = [

"抱歉", "无法回答", "不能回答", "I cannot", "I'm sorry",
"无法提供", "cannot provide", "not able to", "拒绝"

]

如果回答命中任意关键词，标记为拒绝回答。注意关键词可能误伤正常回答，例如“这个问题无法简单回答”包含“无法回答”但不一定是拒绝。解决方案：使用正则匹配整句，或结合上下文判断。

3. 重复检测

对同一问题、同一模型、同一时间窗口（如同一任务批次）内的回答，计算文本相似度。常用算法包括编辑距离和Jaccard相似度。相似度阈值设为0.95，超过则视为重复，只保留第一条。窗口大小建议按任务批次分组，避免跨批次误判。

4. 相关性判断

检查回答中是否包含问题中的核心实体词。例如问题包含“品牌A”，回答中必须出现“品牌A”或其别名。如果完全未出现，标记为无关。核心实体需人工定义，且只要回答包含至少一个核心实体，就认为相关。

5. 格式校验

   空字符串或仅含空白字符：过滤。
   纯标点符号或特殊字符：过滤。
   乱码检测：检查非UTF-8字符比例，超过阈值则过滤。乱码通常出现在编码转换错误时，比例阈值可设为10%。

实现流程

以下为过滤模块的整体流程：

flowchart TD

A[原始回答] --> B{长度过滤}
B -->|通过| C{关键词过滤}
B -->|不通过| F[标记无效]
C -->|通过| D{重复检测}
C -->|不通过| F
D -->|通过| E{相关性判断}
D -->|不通过| F
E -->|通过| G{格式校验}
E -->|不通过| F
G -->|通过| H[有效回答]
G -->|不通过| F
F --> I[丢弃或存入无效库]
H --> J[进入统计]

关键代码示例

以下为过滤模块的核心实现片段（示例伪代码，具体字段和阈值请根据实际数据调整）：

class AnswerFilter:

def __init__(self, min_length=10, max_length=4096, rejection_keywords=None):
    self.min_length = min_length
    self.max_length = max_length
    self.rejection_keywords = rejection_keywords or []
    self.seen_answers = []  # 用于重复检测

def is_valid(self, answer: str, question_entities: list) -> bool:
    # 1. 长度过滤
    if len(answer) < self.min_length or len(answer) > self.max_length:
        return False
    # 2. 关键词过滤
    for kw in self.rejection_keywords:
        if kw in answer:
            return False
    # 3. 重复检测
    for prev in self.seen_answers:
        if self._similarity(answer, prev) > 0.95:
            return False
    self.seen_answers.append(answer)
    # 4. 相关性判断
    if not any(entity in answer for entity in question_entities):
        return False
    # 5. 格式校验
    if self._is_garbled(answer):
        return False
    return True

def _similarity(self, a: str, b: str) -> float:
    # 使用Jaccard相似度：交集字符数 / 并集字符数
    set_a, set_b = set(a), set(b)
    intersection = len(set_a & set_b)
    union = len(set_a | set_b)
    return intersection / union if union > 0 else 0.0

def _is_garbled(self, text: str) -> bool:
    # 检查非ASCII且非中文字符的比例
    non_standard = sum(1 for c in text if ord(c) > 127 and not ('u4e00' <= c <= 'u9fff'))
    return non_standard / len(text) > 0.1 if len(text) > 0 else False

代码说明：_similarity 使用Jaccard相似度，计算简单且对短文本有效。_is_garbled 检查非标准字符比例，阈值10%可根据实际数据调整。
验证方法

如何判断过滤是否有效？

抽样人工复核：从过滤后的有效回答中随机抽取100条，人工判断是否确实有效；从无效回答中随机抽取100条，检查是否有误杀。
统计指标：计算过滤率（无效回答数/总回答数），观察趋势是否合理。如果过滤率突然升高或降低，说明规则可能需要调整。
A/B测试：对比使用过滤和不使用过滤时，后续提及率、推荐率等指标的变化。如果过滤后指标更稳定、更符合业务认知，说明过滤有效。

常见问题与坑点

1. 关键词误杀

拒绝关键词列表可能误伤正常回答。例如“无法回答”在模型拒绝时出现，但某些回答可能包含“这个问题无法简单回答，需要从多个角度分析”。解决方案：使用更精确的匹配模式，如正则表达式匹配整句，或结合上下文判断。

2. 重复检测的窗口大小

重复检测如果窗口过大，会将不同时间点的相似回答误判为重复。建议窗口设为同一任务批次内，或按时间戳分组。

3. 相关性判断的实体覆盖

如果问题包含多个实体，而回答只提及部分，可能被误判为无关。建议：只要回答包含至少一个核心实体，就认为相关。核心实体需人工定义。

4. 截断回答的处理

截断回答可能包含有效信息，但信息不完整。建议：截断回答单独存储，不进入统计，但可人工复核。
优化建议

规则可配置：将阈值和关键词列表放到配置文件中，方便调整。
日志记录：每条回答的过滤原因记录到日志，便于排查。
人工复核队列：对边界情况（如长度刚好在阈值附近）的回答，推入人工复核队列。
渐进式过滤：先宽松过滤，再逐步收紧，观察指标变化。

总结

无效回答过滤是AI回答数据管道中不可或缺的一环。通过长度、关键词、重复、相关性和格式五层过滤，可以大幅提升数据质量。但过滤规则需要根据实际数据持续迭代，没有一劳永逸的方案。关键在于：保留原始数据，记录过滤原因，定期人工复核，让过滤效果可测量、可追溯。