第12节：用户查询太模糊？通过查询扩展，提升语义匹配能力

RAG与Agent性能调优第8节打造可配置可扩展的自动化预处理流水线Gitee地址https://gitee.com/agiforgagaplus/OptiRAGAgent文章详情目录RAG与Agent性能调优上一节第11节HNSW参数调优难掌握SQ8量化压缩技术实现速度与准确率平衡下一节为什么要在RAG中进行查询扩展在构建基于RAG的问答系统时用户输入往往存在以下几个问题表达模糊不完整或口语化缺乏上下文信息难以准确命中知识库中的相关文档如果直接使用用户的原始查询进行向量检索可能会导致以下问题召回结果不足命中无关内容最终生成的答案不够准确或全面因此在执行检索前对用户的问题进行语义及扩写是一种有效手段解决方案为了解决这一问题可以采用以下两种策略来增强检索效果问题改写将模糊查询问题转化为更清晰具体的问题提升语义表达能力多步骤检索将复杂问题拆解为多个子任务逐步检索后整合回答提升检索的全面性和准确性优化手段在检索前进行问题改写%pip install langchain faiss-cpu transformers torch sentence-transformers dashscope langchain-community unstructured[md]加载文档并构建FAISS向量库import logging import os from langchain.document_loaders import DirectoryLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.vectorstores import FAISS from langchain_community.embeddings import DashScopeEmbeddings from langchain.chains import RetrievalQA from langchain_community.chat_models.tongyi import ChatTongyi # 配置日志 logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) # 配置 DOCUMENT_PATH ./12/data/ FILE_PATTERN *.md CHUNK_SIZE 1000 CHUNK_OVERLAP 200 # 通义Embedding模型 EMBEDDING_MODEL text-embedding-v4 LLM_MODEL qwen-plus # 支持qwen-plus/qwen-max/qwen-turbo TOP_K 5 # 1. 初始化LLM使用ChatTongyi logging.info(f正在初始化{LLM_MODEL}模型...) llm ChatTongyi( model_nameLLM_MODEL, api_keyos.getenv(DASHSCOPE_API_KEY), # 使用DashScope API Key temperature0.7, streamingTrue # 启用流式输出 ) # 2. 加载文档 logging.info(f从{DOCUMENT_PATH}加载文档...) loader DirectoryLoader(DOCUMENT_PATH, globFILE_PATTERN) docs loader.load() logging.info(f成功加载{len(docs)}个文档) # 3. 文档切片 logging.info(正在进行文档切片...) text_splitter RecursiveCharacterTextSplitter( chunk_sizeCHUNK_SIZE, chunk_overlapCHUNK_OVERLAP ) texts text_splitter.split_documents(docs) logging.info(f文档切片完成获得{len(texts)}个文本块) # 4. 初始化Embedding模型 logging.info(f加载{EMBEDDING_MODEL}嵌入模型...) embeddings DashScopeEmbeddings( modelEMBEDDING_MODEL, dashscope_api_keyos.getenv(DASHSCOPE_API_KEY) ) # 5. 构建向量库 logging.info(构建FAISS向量库...) vectorstore FAISS.from_documents(texts, embeddings) retriever vectorstore.as_retriever(search_kwargs{k: TOP_K}) # 6. 创建RAG查询链 logging.info(创建RAG查询链...) qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrieverretriever ) logging.info(RAG系统初始化完成ready for query!) # 添加查询示例 def query_rag(question): logging.info(f处理查询: {question}) result qa_chain.invoke(question) return result[result]传统检索无优化# 测试查询 if __name__ __main__: test_question 我想去好玩的地方? answer query_rag(test_question) print(f问题: {test_question}\n答案: {answer})使用大语言改写问题增强语言表达from langchain.prompts import PromptTemplate rewrite_prompt PromptTemplate.from_template( 你是一个旅游助手请将以下用户问题改写为更清晰、完整的形式 原始问题 {{question}} 请输出一个改写后的问题要求 - 更具体 - 包含旅行类型亲子/情侣/自驾等 - 能帮助系统准确检索相关信息 输出格式 [改写后] - 你的回答 )from langchain.chains import LLMChain # 初始化改写链 def create_rewrite_chain(llm): return LLMChain(llmllm, promptrewrite_prompt) # 执行问题改写 def rewrite_question(rewrite_chain, question): logging.info(f正在改写问题: {question}) try: rewritten_question rewrite_chain.run(question).strip() if not rewritten_question: logging.warning(改写后的问题为空) return 未获得有效的改写结果 return rewritten_question except Exception as e: logging.error(f改写问题时出错: {str(e)}) return 改写问题时发生错误 rewrite_chain create_rewrite_chain(llm) test_question 我想去好玩的地方? rewritten rewrite_question(rewrite_chain, test_question) print(f改写后问题: {rewritten})多步骤检索对于涉及多个需求的复杂问题单一检索往往难以覆盖所有方面。此时我们可以将拆分为多个子问题分别进行检索后再综合结果处理流程如下先检索亲子友好的景点再根据这些景点推荐附近的餐馆最后将两次检索结果整合形成完整的回答这种分阶段的检索方式能显著提升准确率和全面性尤其是涉及多个维度的复合型查询import logging from langchain.agents import initialize_agent, Tool from langchain.agents.types import AgentType # 配置日志 def setup_logging(): logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) setup_logging() # 定义工具函数 def search_child_friendly_attractions(query): try: logging.info(f正在检索亲子友好型景点: {query}) result qa_chain.run(query) if not result.strip(): logging.warning(未检索到亲子友好型景点的结果) return 未找到相关亲子友好型景点 return result except Exception as e: logging.error(f检索亲子友好型景点时出错: {str(e)}) return 检索亲子友好型景点时发生错误 def search_nearby_restaurants(query): try: logging.info(f正在检索附近的推荐餐厅: {query}) result qa_chain.run(query) if not result.strip(): logging.warning(未检索到附近餐厅的结果) return 未找到附近的推荐餐厅 return result except Exception as e: logging.error(f检索附近餐厅时出错: {str(e)}) return 检索附近餐厅时发生错误 # 初始化工具 def initialize_tools(): return [ Tool( nameSearchChildFriendlyAttractions, funcsearch_child_friendly_attractions, description用于检索亲子友好型景点输入应该是与亲子游玩相关的问题 ), Tool( nameSearchNearbyRestaurants, funcsearch_nearby_restaurants, description用于根据景点检索附近的推荐餐厅输入应该包含景点名称 ) ] # 初始化代理 def initialize_search_agent(llm): tools initialize_tools() return initialize_agent(tools, llm, agentAgentType.ZERO_SHOT_REACT_DESCRIPTION, verboseTrue) # 执行多步检索 def run_multi_step_search(agent, query): try: logging.info(f开始执行多步检索: {query}) response agent.invoke(query) formatted_response f 多步检索结果:\n{response[output]} print(formatted_response) return formatted_response except Exception as e: logging.error(f执行多步检索时出错: {str(e)}) print(执行多步检索时发生错误请检查日志) return None if __name__ __main__: try: if llm is None: logging.info(llm 变量未定义正在使用默认 ChatOpenAI 进行初始化。) llm ChatOpenAI(temperature0, model_namegpt-4o) if qa_chain is None: logging.info(qa_chain 变量未定义正在使用默认 LLMChain 进行初始化。) prompt_template 请根据以下问题给出相关信息{question}\n \ 如果找不到相关信息请明确说明 未找到相关信息。 qa_chain LLMChain(llmllm, promptPromptTemplate.from_template(prompt_template)) except NameError: logging.error(llm 或 qa_chain 变量未定义请确保在使用前正确初始化) raise agent initialize_search_agent(llm) multi_step_query 我想找一个适合带孩子玩的地方附近最好有好吃的餐厅。 先查有哪些亲子友好景点再找附近的餐饮推荐。 run_multi_step_search(agent, multi_step_query)效果与评估评估方法示例构建包含原始问题、改写后问题期望答案的数据集判断最终生成的答案是否包含正确答案统计准确率或召回率等指标import logging from typing import List, Dict, Tuple from difflib import SequenceMatcher # 配置日志 def setup_logging(): logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) setup_logging() class GroundTruthDataset: 管理包含原始问题、改写后问题和期望答案的数据集 def __init__(self): self.data: List[Dict[str, str]] [] def add_example(self, original_q: str, rewritten_q: str, ground_truth: str): 添加数据示例 self.data.append({ original_q: original_q, rewritten_q: rewritten_q, ground_truth: ground_truth }) logging.info(成功添加新的评估示例) def get_all_examples(self) - List[Dict[str, str]]: 获取所有数据示例 return self.data class Evaluator: 评估模型回答的性能 def __init__(self, similarity_threshold: float 0.8): self.similarity_threshold similarity_threshold def _is_answer_correct(self, answer: str, ground_truth: str) - bool: 使用相似度判断答案是否正确 similarity SequenceMatcher(None, answer.lower(), ground_truth.lower()).ratio() return similarity self.similarity_threshold def evaluate_single(self, original_q: str, rewritten_q: str, answer: str, ground_truth: str) - Tuple[int, Dict[str, str]]: 评估单个示例 score 1 if self._is_answer_correct(answer, ground_truth) else 0 result { original_q: original_q, rewritten_q: rewritten_q, answer: answer, ground_truth: ground_truth, is_correct: score 1 } logging.info(f原始问题: {original_q}\n改写问题: {rewritten_q}\n正确答案包含: {✅ if score else ❌}) return score, result def evaluate_dataset(self, dataset: GroundTruthDataset, answer_generator) - Dict[str, float]: 评估整个数据集 total len(dataset.get_all_examples()) correct 0 results [] for example in dataset.get_all_examples(): try: answer answer_generator(example[rewritten_q]) score, result self.evaluate_single( example[original_q], example[rewritten_q], answer, example[ground_truth] ) correct score results.append(result) except Exception as e: logging.error(f评估示例时出错: {str(e)}) continue accuracy correct / total if total 0 else 0 # 可以扩展计算召回率、精确率、F1值等指标 return { accuracy: accuracy, correct_count: correct, total_count: total, results: results } def generate_report(self, metrics: Dict[str, float]) - str: 生成评估报告 report f评估报告:\n report f- 总样本数: {metrics[total_count]}\n report f- 正确样本数: {metrics[correct_count]}\n report f- 准确率: {metrics[accuracy]:.2%}\n return report # 示例使用 if __name__ __main__: # 初始化数据集 dataset GroundTruthDataset() dataset.add_example( original_q我想带孩子玩, rewritten_q推荐适合亲子游玩的目的地, ground_truth推荐了上海迪士尼乐园和北京环球影城两个亲子游目的地 ) # 模拟回答生成函数实际使用时替换为真实的模型调用 def mock_answer_generator(query: str) - str: if 亲子游玩 in query: return 推荐了上海迪士尼乐园和北京环球影城两个亲子游目的地 return 未找到相关信息 # 初始化评估器 evaluator Evaluator(similarity_threshold0.8) # 执行评估 metrics evaluator.evaluate_dataset(dataset, mock_answer_generator) # 生成并打印评估报告 report evaluator.generate_report(metrics) print(report)总结更多优化建议