大模型应用系列：Query 变换的示例浅析

【引】NLP中的经典组件在大模型应用中还有效么？大模型对自然语言处理中的典型任务有什么影响么？

RAG应用通过分割文档、嵌入向量化并检索高语义相似性的块来响应用户问题，但面临文档块不相关、用户用词不当及结构化查询需求等问题。若RAG无法找到必要信息，则无法正确回答。对此，可采用Query变换技术，包括Query改写以提高检索准确性、后退提示以获取背景信息及子查询分解以全面检索。这些方法常利用大模型生成新query，关键在于生成提示词的选择。

1. Query 变换及其所解决的问题空间

一般地，RAG应用通常将文档分割成块，然后嵌入向量化，并检索与用户问题具有高语义相似性的块。但是，这会带来一些问题:

(1)文档块可能包含不相关的内容，这会降低检索效率;
(2)用户提出的问题可能用词不当，不利于检索;
(3)结构化查询可能需要从用户提出的问题中产生(例如，用元数据过滤或 SQL 数据库查询向量存储)。

也就是说，用户的query可能写得很糟糕，或者表达方式与我们预期的不同。而且，如果我们的 RAG 应用程序不能找到回答这个问题所需的信息，它就不会正确地回答。面对这些问题，一般会采用Query 变换的技术，主要有3种解决方案：

Query改写: 使查询更加具体和详细，提高检索最相关信息的可能性
后退提示: 生成更广泛、更一般的查询，以帮助检索相关的背景信息
子查询分解: 将复杂的查询分解为更简单的子查询，以获得更全面的信息检索。

具体而言，如果考虑一个简单的 RAG流水线，通常的流程是将用户提出的问题直接传递给嵌入模型。然后，将这种嵌入与存储在向量存储中的文档进行比较，返回top-k个最相似的文档，而query变换在传递到嵌入模型之前处理用户query的变换。

随着大模型的普及，这些Query变换的方法都乐意使用大模型来生成新的(或多个新的)query，主要区别在于它们使用的生成提示词。

本文会简述各种Query变换的实现方法示例，当然首先是环境搭建，导入必要的库并设置访问权限。

from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplateimport os
from dotenv import load_dotenv# Load environment variables from a .env file
load_dotenv()# Set the OpenAI API key environment variable
os.environ["OPENAI_API_KEY"] = os.getenv('OPENAI_API_KEY')

2. Query改写

简单地说，Query改写意味着我们将用自己的语言改写用户的query，以便于我们的 RAG 应用程序将知道如何最好地回答。我们将采用改写-检索-读取的方法，而不仅仅是检索-读取的模式。

from langchain.chat_models import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain# LLM configuration
def get_rewriting_llm(model_name: str = "gpt-4o", temperature: float = 0, max_tokens: int = 4000) -> ChatOpenAI:return ChatOpenAI(temperature=temperature, model_name=model_name, max_tokens=max_tokens)# Prompt Template for Query Rewriting
def create_query_rewrite_prompt() -> PromptTemplate:template = """You are an AI assistant tasked with reformulating user queries to improve retrieval in a RAG system. Given the original query, rewrite it to be more specific, detailed, and likely to retrieve relevant information.Original query: {original_query}Rewritten query:"""return PromptTemplate(input_variables=["original_query"], template=template)# Build Query Rewriting Chain
def build_query_rewriting_chain(llm: ChatOpenAI) -> LLMChain:prompt_template = create_query_rewrite_prompt()return prompt_template | llm# Function to Rewrite the Query
def rewrite_query(original_query: str, query_rewriter_chain: LLMChain) -> str:response = query_rewriter_chain.invoke(original_query)return response.content.strip()if __name__ == "__main__":  llm = get_rewriting_llm()query_rewriter_chain = build_query_rewriting_chain(llm)original_query = "What are the impacts of climate change on the environment?"rewritten_query = rewrite_query(original_query, query_rewriter_chain)print("Original query:", original_query)print("\nRewritten query:", rewritten_query)

我们使用生成式人工智能模型来改写问题。这个模型是一个大模型，就像我们在最后一步用来回答问题的模型一样。或者，它也可以是一个较小的模型，专门训练来执行这项任务。

3. 后退提示

对基于 RAG 流水线的检索来说，许多问题可能有点过于复杂，以至于无法掌握回答这些问题所需的多层次信息。对于这些情况，生成用于检索的多个附加查询可能很有帮助。这些查询将比原始查询更通用。这将使 RAG 流水线能够在多个级别上检索相关信息。

from langchain.chat_models import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain# LLM Configuration
def get_step_back_llm(model_name: str = "gpt-4o", temperature: float = 0, max_tokens: int = 4000) -> ChatOpenAI:return ChatOpenAI(temperature=temperature, model_name=model_name, max_tokens=max_tokens)# Step-Back Prompt Template
def create_step_back_prompt_template() -> PromptTemplate:template = """You are an AI assistant tasked with generating broader, more general queries to improve context retrieval in a RAG system.Given the original query, generate a step-back query that is more general and can help retrieve relevant background information.Original query: {original_query}Step-back query:"""return PromptTemplate(input_variables=["original_query"], template=template)# Build Step-Back Query Chain
def build_step_back_chain(llm: ChatOpenAI) -> LLMChain:prompt_template = create_step_back_prompt_template()return prompt_template | llm# Function to Generate Step-Back Query
def generate_step_back_query(original_query: str, step_back_chain: LLMChain) -> str:response = step_back_chain.invoke(original_query)return response.content.strip()if __name__ == "__main__":llm = get_step_back_llm()step_back_chain = build_step_back_chain(llm)original_query = "What are the impacts of climate change on the environment?"step_back_query = generate_step_back_query(original_query, step_back_chain)print("Original query:", original_query)print("\nStep-back query:", step_back_query)

4.子查询分解

如果用户Query包含多个问题，这会使上下文检索变得棘手。每个问题可能需要不同的信息，我们不会用所有的问题作为信息检索的基础。为了解决这个问题，我们可以将输入分解为多个子查询，并对每个子查询执行检索。

from langchain.chat_models import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
from typing import List# LLM Configuration
def get_subquery_llm(model_name: str = "gpt-4o", temperature: float = 0, max_tokens: int = 4000) -> ChatOpenAI:return ChatOpenAI(temperature=temperature, model_name=model_name, max_tokens=max_tokens)# Sub-query Decomposition Prompt Template
def create_subquery_decomposition_template() -> PromptTemplate:template = """You are an AI assistant tasked with breaking down complex queries into simpler sub-queries for a RAG system.Given the original query, decompose it into 2-4 simpler sub-queries that, when answered together, would provide a comprehensive response to the original query.Original query: {original_query}example: What are the impacts of climate change on the environment?Sub-queries:1. What are the impacts of climate change on biodiversity?2. How does climate change affect the oceans?3. What are the effects of climate change on agriculture?4. What are the impacts of climate change on human health?"""return PromptTemplate(input_variables=["original_query"], template=template)# Build Sub-query Decomposition Chain
def build_subquery_decomposer_chain(llm: ChatOpenAI) -> LLMChain:prompt_template = create_subquery_decomposition_template()return prompt_template | llm# Function to Decompose Query into Sub-queries
def decompose_query(original_query: str, subquery_chain: LLMChain) -> List[str]:response = subquery_chain.invoke(original_query).content# Parse the sub-queries by splitting lines and removing unwanted textsub_queries = [q.strip() for q in response.split('\n') if q.strip() and not q.strip().startswith('Sub-queries:')]return sub_queriesif __name__ == "__main__":llm = get_subquery_llm()subquery_chain = build_subquery_decomposer_chain(llm)original_query = "What are the impacts of climate change on the environment?"sub_queries = decompose_query(original_query, subquery_chain)print("\nSub-queries:")for i, sub_query in enumerate(sub_queries, 1):print(f"{i}. {sub_query}")