Query 改写：大模型测试的数据倍增器

一、Query 改写的价值

在大模型应用测试中，测试查询的覆盖面往往捉襟见肘。Query 改写（Query Rewrite/Expansion）作为"数据倍增器"，能从有限的原始查询衍生出语义等价但表达多变的变体，让测试数据集瞬间膨胀，覆盖更多边缘 case 和真实场景。

为什么需要 Query 改写

• 用户表达多样性：同样的意图，用户可能用"北京天气咋样"、"首都今天有雨吗？"或"查查北京当前气象情况"等多种方式表达
• 测试覆盖率提升：改写后的多样查询能提前挖掘模型在模糊、口语化输入下的弱点
• 鲁棒性验证：测试模型对不同表达方式的适应能力

---

二、三种 Query 改写方法

方法一：基于大模型的改写

直接借助大模型能力，基于原始查询生成语义等价的变体。大模型天生懂得捕获上下文和用户意图，能捕捉人类想不到的表达。

class LLMRewriter:
    def __init__(self, llm: LLMBase):
        self.llm = llm
        self.system_prompt = '''
def 资深测试开发专家():
    """
    你是一名从业20年的资深测试开发工程师，一直从事NLP、LLM相关技术的测试。
    """
    能力=["测试分析", "测试设计","NLP性能指标","LLM的性能指标","RAG","大模型应用测试"]
    
def query_rewrite(用户输入):
    """
    分析用户的输入的测试数据，依据数据进行改写并返回，返回格式json
    """
    new_query_list = one_query_rewrite(用户输入["query"], 用户输入["reference"])
    for one_new_query in new_query_list:
        new_data.append({"query": one_new_query, "reference": 用户输入["reference"]})

def one_query_rewrite(query, reference):
    """
    依据reference的上下文，完成query改写，返回一个list，包含10条新query。
    确保每条新query语义锚定reference，避免幻觉；文字结构多样化，如口语/正式/疑问变体。
    """
'''
    
    def rewrite(self, query: Query) -> List[RewrittenQuery]:
        prompt = f'{self.system_prompt}\n\n{{"query":"{query.query}","reference":"{query.reference}"}}'
        response = self.llm.invoke(prompt)
        # JSON解析 + 异常处理
        parsed_response = self.response_parser.loads(response)
        return [RewrittenQuery(**item) for item in parsed_response]

优势：创意丰富，能生成人类想不到的表达变体
适用场景：测试模型的意图识别和鲁棒性

---

方法二：词汇表改写

基于预构建的领域词汇表，扫描原始查询中的关键词，替换成同义备选，批量产出变体。

class GlossaryRewriter:
    def __init__(self, glossary: Glossary, max_combos: int = 100):
        """
        参数:
            glossary: 同义词组列表，格式为 List[List[str]]
            max_combos: 生成重写查询的最大数量（防止组合爆炸）
        """
        self.glossary = glossary
        self.max_combos = max_combos
        self.synonym_map = self._create_synonym_map()
    
    def _create_synonym_map(self) -> dict:
        """创建映射：单词 -> 完整的同义词列表"""
        synonym_map = {}
        for word_list in self.glossary:
            for word in word_list:
                synonym_map[word] = word_list
        return synonym_map
    
    def rewrite(self, query: Query) -> List[RewrittenQuery]:
        words = self._tokenize(query.query)
        rewritten_word_lists = [self.synonym_map.get(word, [word]) for word in words]
        
        # 生成所有组合，如果数量过多则进行采样
        all_combos = list(itertools.product(*rewritten_word_lists))
        if len(all_combos) > self.max_combos:
            all_combos = random.sample(all_combos, self.max_combos)
        
        rewritten_queries = []
        for combination in all_combos:
            joined_query = "".join(combination)  # 中文不加空格
            rewritten_queries.append(RewrittenQuery(query=joined_query, reference=query.reference))
        return rewritten_queries

示例：

• 词汇表：{"买": ["购买", "选购", "入手"], "手机": ["手机", "移动电话", "智能机"]}
• 原始查询："买个新手机"
• 改写结果："选购一部最新款手机"、"入手苹果新机"等

优势：稳定、无幻觉风险，适用于规则明确的垂直场景
适用场景：医疗、法律等领域，避免 LLM “脑补过头”

---

方法三：同义词改写

结合大模型和规则替换的方法：先让大模型生成关键词的备选列表，再用规则替换植入原查询。

class SynonymRewriter:
    def __init__(self, llm: LLMBase, max_synonyms_per_word: int = 5):
        self.llm = llm
        self.max_synonyms_per_word = max_synonyms_per_word
    
    def _get_synonyms(self, word: str, flag: str) -> List[str]:
        """通过LLM获取同义词"""
        skip_flags = ['x', 'wp', 'ws', 'w']  # 跳过标点/未知词
        if flag in skip_flags:
            return [word]
        
        prompt = f"生成'{word}'的最多{self.max_synonyms_per_word}个同义词，以json list的格式返回。"
        response = self.llm.invoke(prompt)
        synonyms = SuperList(response)
        return [s.strip() for s in synonyms[:self.max_synonyms_per_word] if isinstance(s, str) and s]
    
    def rewrite(self, query: Query) -> List[RewrittenQuery]:
        words_pos = self._tokenize_pos(query.query)  # 带词性标注的分词
        rewritten_word_lists = []
        
        for word, flag in words_pos:
            synonyms = self._get_synonyms(word, flag)
            rewritten_word_lists.append(synonyms)
        
        # 生成组合
        all_combos = list(itertools.product(*rewritten_word_lists))
        # ... 组合处理和返回

优势：比 LLM 改写少点"天马行空"，比词汇表改写多点灵活
适用场景：平衡创意与精确控制的场景

---

三、改写 Query 的筛选方法

Query 改写后需要验证环节筛查，确保不丢失原 Query 的核心意图，同时剔除表达雷同或信息贫瘠的 query。

筛选方法一：ROUGE-L 和 BLEU 归一化验证

计算改写 Query 与原始 Query 的 ROUGE-L 和 BLEU 分数，量化变体质量。

def rouge_l_bleu_normalized(rewritten_queries: List[RewrittenQuery], 
                            original_query: str, 
                            rouge_weight: float = 0.7) -> List[RewrittenQuery]:
    """
    通过加权的ROUGE-L和(1-BLEU)分数来选择最佳查询。
    ROUGE-L (越高越好) 代表语义相似度。
    BLEU (越低越好) 代表词汇差异度。我们使用 (1-BLEU) 使其变为越高越好。
    """
    bleu_weight = 1 - rouge_weight
    scored_queries = []
    
    for rq in rewritten_queries:
        rouge_l = calculate_rouge_l(rq["query"], original_query)
        bleu = calculate_bleu(rq["query"], original_query)
        # 综合得分：ROUGE-L越高越好，BLEU越低越好
        score = rouge_weight * rouge_l + bleu_weight * (1 - bleu)
        scored_queries.append((score, rq))
    
    # 返回综合得分最高的查询
    best_query = max(scored_queries, key=lambda item: item[0])
    return [best_query[1]]

解读：高 ROUGE-L + 低 BLEU = 内容抓得准但表达新鲜，是"多样而不乱"的黄金平衡。

---

筛选方法二：帕累托最优

借助帕累托前沿原则，在 ROUGE-L vs BLEU 的坐标系上，挑出那些"无人能敌"的变体。

def pareto_optimal(rewritten_queries: List[RewrittenQuery], original_query: str) -> List[RewrittenQuery]:
    """
    基于 ROUGE-L 和 BLEU 分数找出帕累托最优的改写查询集合。
    一个查询是帕累托最优的，当且仅当没有其他查询在两个维度上都优于它。
    """
    scores = []
    for rq in rewritten_queries:
        rouge_l = calculate_rouge_l(rq["query"], original_query)
        bleu = calculate_bleu(rq["query"], original_query)
        scores.append((rouge_l, bleu, rq))
    
    pareto_front = []
    for i, (r1, b1, q1) in enumerate(scores):
        is_dominated = False
        for j, (r2, b2, q2) in enumerate(scores):
            if i == j: 
                continue
            # 如果另一个查询在ROUGE-L上更高或相等，在BLEU上更低或相等，且至少一个严格更优
            if (r2 >= r1 and b2 <= b1) and (r2 > r1 or b2 < b1):
                is_dominated = True
                break
        if not is_dominated:
            pareto_front.append(q1)
    
    return pareto_front

优势：自动挖掘语义-多样性的"黄金权衡"，避开"一刀切"阈值的主观坑。

---

筛选方法三：阈值过滤

建立门槛，只有双达标的改写 Query 才能被选中。

def filter_by_rouge_l_bleu_thresholds(rewritten_queries: List[RewrittenQuery], 
                                      original_query: str,
                                      rouge_l_threshold: float = 0.4, 
                                      bleu_threshold: float = 0.3) -> List[RewrittenQuery]:
    """
    基于 ROUGE-L 和 BLEU 分数阈值过滤查询。
    - ROUGE-L score > rouge_l_threshold (越高越好)
    - BLEU score < bleu_threshold (越低越好)
    """
    optimal_queries = []
    for rq in rewritten_queries:
        rouge_l_score = calculate_rouge_l(rq["query"], original_query)
        bleu_score = calculate_bleu(rq["query"], original_query)
        if rouge_l_score >= rouge_l_threshold and bleu_score < bleu_threshold:
            optimal_queries.append(rq)
    return optimal_queries

---

筛选方法四：大模型语义相似度

利用"判官"大模型作为语义裁判，评价改写 Query 和原 Query 的相似度。

def llm_semantic_similarity(rewritten_queries: List[RewrittenQuery], 
                            original_query: str, 
                            llm: LLMBase) -> List[RewrittenQuery]:
    """使用LLM寻找语义最相似且词汇差异最大（BLEU最低）的查询。"""
    best_query = None
    highest_similarity = -1.0
    lowest_bleu_at_highest_sim = 2.0
    
    for rq in rewritten_queries:
        prompt = f'''评估以下两个查询的语义相似度，
查询1: {original_query}
查询2: {rq["query"]}
返回一个0到1之间的浮点数，semantic_similarity='''
        
        response = llm.invoke(prompt)
        similarity = SuperFloat(response)
        bleu_score = calculate_bleu(rq["query"], original_query)
        
        # 核心选择逻辑：相似度更高则更新；相似度相等则选BLEU更低的
        if similarity > highest_similarity:
            highest_similarity = similarity
            lowest_bleu_at_highest_sim = bleu_score
            best_query = rq
        elif abs(similarity - highest_similarity) < 1e-9:
            if bleu_score < lowest_bleu_at_highest_sim:
                lowest_bleu_at_highest_sim = bleu_score
                best_query = rq
    
    return [best_query] if best_query else []

---

四、总结

Query 改写遵循"输入-生成-过滤-输出"的模块化流水线：

阶段	方法选择
生成	大模型改写（创意丰富）、词汇表改写（稳定可控）、同义词改写（平衡灵活）
过滤	ROUGE-L/BLEU归一化、帕累托最优、阈值过滤、大模型语义裁判
输出	高质量、多样化、语义等价的测试查询集合

这种方法不仅提升了测试数据构建的效率，更是测试思维的升级——从"手动编查询编到吐"到"AI 帮手一键扩充"，让测试工程师专注于更高阶的测试设计。