RAG--Query改写

一、 为什么要进行Query改写？
想象一下，你去图书馆查资料，但问管理员的问题很模糊、有错别字、或者缺少背景信息，管理员就很难找到你真正需要的书。

Query改写就是帮用户“把问题问得更好”。
主要解决以下问题：

1. 表述模糊：用户查询简短、不完整

2. 词汇不匹配：用户用词和文档用词不同

3. 缺少上下文：多轮对话中，当前查询脱离了历史背景

4. 意图不明确：需要推测用户的深层需求

二、 Query改写的主要策略

将策略分为四个层次：

Level 1: 基础改写（适合快速上手）
1. 查询扩展（Query Expansion）
思路：为原始查询添加同义词、相关词

# 使用同义词库扩展
import nltk
from nltk.corpus import wordnetdef expand_query_with_synonyms(query):expanded_terms = []for word in query.split():synonyms = set()for syn in wordnet.synsets(word):for lemma in syn.lemmas():synonyms.add(lemma.name())if synonyms:expanded_terms.append(f"({' OR '.join(list(synonyms)[:3])})")else:expanded_terms.append(word)return ' '.join(expanded_terms)# 示例：将"car"扩展为"(car OR auto OR automobile OR motorcar)"

2.拼写纠正

# 使用TextBlob（需安装：pip install textblob）
from textblob import TextBlobdef correct_spelling(query):blob = TextBlob(query)return str(blob.correct())

Level 2: 上下文感知改写（处理多轮对话）

1. 对话历史整合

def rewrite_with_conversation_history(current_query, history, max_turns=3):"""history格式: [{"user": "...", "assistant": "..."}, ...]"""if not history:return current_query# 提取最近的几轮对话recent_history = history[-max_turns:]# 构建带历史的提示词context = "\n".join([f"User: {h['user']}\nAssistant: {h['assistant']}" for h in recent_history])prompt = f"""基于以下对话历史和当前问题，重写当前问题使其更完整、独立。对话历史：
{context}当前问题：{current_query}重写后的问题："""# 调用LLM进行改写（以智谱AI为例）from zai import ZhipuAiClientclient = ZhipuAiClient(api_key="")response = client.chat.completions.create(model="",messages=[{"role": "user", "content": prompt}],temperature=0.3)return response.choices[0].message.content

Level 3: 高级意图理解（使用LLM进行智能改写）

1. 多角度改写（HyDE - Hypothetical Document Embeddings）

def generate_hypothetical_documents(query, n=3):"""让LLM生成假设的答案文档，然后用这些文档的向量进行检索"""prompt = f"""根据以下问题，生成{n}个可能的答案段落。
问题：{query}生成的答案应专业、详细，包含相关术语。
每个答案用'---'分隔："""# 调用LLM生成假设答案# ... (类似上面的LLM调用)# 返回生成的假设答案列表return hypothetical_docs# 检索时，用生成的假设文档的向量来查询

2. 分步思考改写（Step-by-Step）

def stepwise_query_rewrite(query):prompt = f"""请按照以下步骤分析并重写查询：
1. 识别查询的深层意图：用户真正想知道什么？
2. 识别缺失的关键信息：哪些信息对检索是必要的但缺失了？
3. 识别专业术语：哪些词可以替换为更专业的术语？
4. 生成3个不同角度的改写版本。查询：{query}分析步骤1（深层意图）：
分析步骤2（缺失信息）：
分析步骤3（专业术语）：
改写版本1（详细版）：
改写版本2（关键词版）：
改写版本3（疑问解答版）："""# 调用LLM并解析结果# ...return rewritten_queries  # 返回多个版本

Level 4: 专业领域优化

1. 领域术语标准化

def domain_specific_rewrite(query, domain_glossary):"""domain_glossary: 领域术语词典，如{"ML": "机器学习", "NLP": "自然语言处理"}"""# 1. 术语替换for term, standard_term in domain_glossary.items():if term.lower() in query.lower():query = query.replace(term, standard_term)# 2. 添加领域限定词prompt = f"""作为{domain}领域的专家，请重写以下查询，使其更专业：原始查询：{query}重写时请考虑：
- 使用领域标准术语
- 明确问题边界
- 添加必要的技术上下文重写后的查询："""# ...

三、 完整的实现架构

class QueryRewriter:def __init__(self, llm_client, strategy="auto"):self.llm = llm_clientself.strategy = strategy# 预定义的改写提示词模板self.templates = {"general": """请重写以下查询，使其更适合文档检索：
{query}要求：
1. 保持原意不变
2. 更详细、更具体
3. 使用更正式的表述
4. 包含可能的同义词重写后：""","hyde": """请为以下问题生成一个假设性的详细答案段落：
问题：{query}生成的段落应：
1. 直接回答问题
2. 包含相关细节和术语
3. 类似百科全书中的解释
4. 长度在100-200字假设答案：""","conversational": """基于对话历史重写当前查询：对话历史：
{history}当前查询：{query}重写要求：
1. 使查询独立完整
2. 融入必要的上下文
3. 明确指代关系
4. 保持自然流畅重写后："""}def rewrite(self, query, history=None, domain=None):# 1. 基础预处理query = self._preprocess(query)# 2. 根据策略选择改写方式if self.strategy == "hyde":return self._hyde_rewrite(query)elif self.strategy == "multi" and history:return self._multi_turn_rewrite(query, history)else:return self._general_rewrite(query, domain)def _preprocess(self, query):"""基础清洗和纠正"""# 去除多余空格query = ' '.join(query.split())# 可选：拼写纠正# query = correct_spelling(query)return querydef _general_rewrite(self, query, domain=None):"""通用LLM改写"""template = self.templates["general"]if domain:prompt = template.format(query=query) + f"\n领域：{domain}"else:prompt = template.format(query=query)response = self.llm.generate(prompt)return response.strip()def _hyde_rewrite(self, query):"""生成假设文档"""prompt = self.templates["hyde"].format(query=query)hypothetical_doc = self.llm.generate(prompt)# 这里可以返回假设文档用于检索# 或者用假设文档再次生成查询hyde_query_prompt = f"""基于以下问题和生成的假设答案，提炼出最佳的检索查询：原始问题：{query}
假设答案：{hypothetical_doc}最适合向量检索的查询（关键词形式）："""return self.llm.generate(hyde_query_prompt)def generate_multiple_versions(self, query, n=3):"""生成多个改写版本，用于多向量检索"""prompt = f"""请为以下查询生成{n}个不同的改写版本，每个版本侧重不同角度：原始查询：{query}要求：
1. 版本1：详细扩展版（添加细节和上下文）
2. 版本2：简洁关键词版（提取核心关键词）
3. 版本3：疑问解答版（以问题形式表达）输出格式：
版本1：...
版本2：...
版本3：..."""response = self.llm.generate(prompt)# 解析response，提取各个版本versions = []for line in response.split('\n'):if line.startswith('版本'):versions.append(line.split('：', 1)[1])return versions if versions else [query]

四、 评估改写效果

def evaluate_rewrite(original_query, rewritten_query, retrieved_docs):"""简单的评估函数retrieved_docs: 检索到的文档列表"""# 1. 检索相关度评估（需要人工标注或使用LLM评估）# 2. 多样性评估：改写是否带来了新的有效关键词# 3. 忠实度评估：改写是否改变了原意
    evaluation_prompt = f"""请评估查询改写的质量：原始查询：{original_query}
改写后查询：{rewritten_query}
检索到的文档数：{len(retrieved_docs)}请从以下维度评分（1-5分）：
1. 忠实度（保持原意）：
2. 明确性（更具体）：
3. 检索友好度（适合检索）：
4. 自然度（表达自然）：简要说明："""# 用LLM评估或人工评估return scores

 五、 最小简单示例

# 最简化的起点代码
from zai import ZhipuAiClient
import requestsclass SimpleRewriter:def __init__(self, api_key):self.client = ZhipuAiClient(api_key=api_key)def rewrite(self, query):response = self.client.chat.completions.create(model="",messages=[{"role": "system", "content": "你是一个查询改写专家。"},{"role": "user", "content": f"请改进以下查询以便更好地检索相关信息：{query}"}],temperature=0.3,max_tokens=100)return response.choices[0].message.content# 使用
rewriter = SimpleRewriter("api-key")
improved_query = rewriter.rewrite("如何学习Python？")
print(f"原始查询：如何学习Python？")
print(f"改写后：{improved_query}")
# 输出可能是："Python编程语言的学习路径、入门教程和最佳实践"

没有完美的Query改写策略，只有最适合数据和场景的策略。