一、 为什么要进行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改写策略，只有最适合数据和场景的策略。