掌柜智库项目(RAG)实战9. 检索数据节点实现与测试9.3 假设性文档向量搜索 (node_search_embedding_hyde)文件:app/query_process/agent/nodes/node_search_embedding_hyde.py节点作用和实现思路什么是HyDE: HyDEHypothetical Document Embeddings“假设文档向量化”是一种让检索更“聪明”的技巧。当用户只给出一句短问题时系统先不急着去向量库里硬搜而是先让大模型“脑补”出一段可能的答案或相关描述——就像先写一段“假想的说明书”。这段文字通常更具体、更像一篇小文档包含更多关键词、背景和表达方式。接着把这段“假想文档”做成向量再拿它去向量库里找相似内容。结果往往比直接拿那一句短问题去做向量检索更准因为检索时用的是更“饱满”的语义信号。它的优点是简单、通用、对很多场景都有效缺点是也会“脑补过头”生成的假设文本可能带偏方向所以常见做法是配合过滤、重排或多路检索一起用。实现思路:1利用大模型根据用户查询生成假设性文档Hypothetical Document。2利用“重写问题 假设性文档”生成 embedding并到向量库检索切片。节点代码实现步骤1导入依赖# HyDE节点importsysfromapp.utils.task_utilsimportadd_running_task,add_done_taskfromapp.lm.lm_utilsimport*fromapp.lm.embedding_utilsimport*fromapp.clients.milvus_utilsimport*fromapp.core.loggerimportloggerfromapp.core.load_promptimportload_promptfromdotenvimportload_dotenv,find_dotenv load_dotenv(find_dotenv())步骤2主流程节点入口函数 (node_item_name_confirm)功能串联功能所有步骤作为 LangGraph 的节点入口。defnode_search_embedding_hyde(state): HyDE (Hypothetical Document Embedding) 检索节点 核心思想通过LLM生成假设性答案HyDE文档将其向量化后用于检索以解决短查询语义稀疏问题。 执行步骤 1. 参数提取从会话状态中获取改写后的查询rewritten_query和已确认的商品名item_names。 2. 生成假设文档 (Step 1)调用LLM基于用户问题生成一段假设性的理想回答即HyDE文档。 3. 混合检索 (Step 2) - 将“用户问题 假设文档”合并生成BGE-M3稠密稀疏向量。 - 在Milvus中执行混合检索带商品名过滤召回最相似的知识切片。 4. 结果封装返回检索到的切片列表和生成的假设文档更新会话状态。 :param state: 会话状态字典包含 session_id, rewritten_query, item_names 等 :return: 包含 hyde_embedding_chunks (检索结果) 和 hyde_doc (假设文档) 的字典 logger.info(---HyDE (假设文档检索) 节点开始处理---)# 记录任务开始状态add_running_task(state[session_id],sys._getframe().f_code.co_name,state.get(is_stream))# 1. 参数提取与校验# 优先使用改写后的查询若无则降级使用原始查询rewritten_querystate.get(rewritten_query)ifnotrewritten_query:rewritten_querystate.get(original_query)ifnotrewritten_query:logger.error(HyDE节点错误: 未找到有效的用户查询 (rewritten_query/original_query 均为空))return{}item_namesstate.get(item_names)logger.info(fHyDE检索入参: query{rewritten_query}, item_names{item_names})# 阶段1生成假设性文档hyde_doctry:logger.info(Step 1: 开始生成假设性文档 (HyDE Doc)...)hyde_docstep_1_create_hyde_doc(rewritten_query)logger.info(fStep 1: 假设文档生成成功 (长度:{len(hyde_doc)}))logger.debug(f假设文档预览:{hyde_doc[:100]}...)exceptExceptionase:logger.error(fStep 1 (生成假设文档) 发生异常:{e},exc_infoTrue)# HyDE生成失败属于非阻断性错误可选择直接返回空或降级处理此处直接返回空结果return{}# 阶段2用“重写问题 假设文档”检索切片try:logger.info(Step 2: 基于假设文档执行 Milvus 混合检索...)resstep_2_search_embedding_hyde(rewritten_queryrewritten_query,hyde_dochyde_doc,item_namesitem_names,top_k5,)hit_countlen(res[0])ifresandlen(res)0else0logger.info(fStep 2: 检索完成召回{hit_count}条相关切片)ifhit_count0:# 打印第一条结果用于调试first_hitres[0][0]scorefirst_hit.get(distance)content_previewfirst_hit.get(entity,{}).get(content,)[:30]logger.debug(fTop1 结果: Score{score}, Content{content_preview}...)return{hyde_embedding_chunks:res[0]ifreselse[],hyde_doc:hyde_doc,}exceptExceptionase:logger.error(fStep 2 (向量生成与检索) 发生异常:{e},exc_infoTrue)return{}finally:# 无论成功失败均标记任务结束add_done_task(state[session_id],sys._getframe().f_code.co_name,state.get(is_stream))logger.info(---HyDE 节点处理结束---)步骤3获取假设性答案 (step_1_create_hyde_doc)先提取提示词位置prompts/rewritten_query_and_itemnames.prompt请基于以下用户查询生成一个简洁的回答范文。用户查询:{rewritten_query}要求1.回答要简洁明了包含核心信息即可2.假设你是该领域的专家提供专业的解释3.不要使用假设、可能等不确定的词汇4.保持回答与查询主题高度相关5.使用中文回答且不超过300字方法细化实现defstep_1_create_hyde_doc(rewritten_query:str)-str: 阶段1利用大模型根据用户查询生成假设性文档Hypothetical Document。 HyDE的核心在于利用LLM生成一个“虚构但相关”的文档用该文档的向量去检索真实的文档 从而缓解短查询Query与长文档Document在语义空间不匹配的问题。 :param rewritten_query: 用户改写后的查询语句 :return: LLM生成的假设性文档内容 ifnotrewritten_query:logger.error(Step 1 Error: rewritten_query 为空)raiseValueError(rewritten_query 不能为空)logger.info(fStep 1: 开始生成假设性文档 (HyDE), Query:{rewritten_query})try:llmget_llm_client()# 加载提示词模板生成假设文档# 提示词通常引导LLM请为这个问题写一段专业的回答...hyde_promptload_prompt(hyde_prompt,rewritten_queryrewritten_query)logger.debug(fStep 1: Prompt加载成功, 长度:{len(hyde_prompt)})# 调用LLM生成responsellm.invoke(hyde_prompt)hyde_docresponse.content logger.info(fStep 1: 假设文档生成完成, 长度:{len(hyde_doc)}字符)logger.debug(fStep 1: 文档预览:{hyde_doc[:50]}...)returnhyde_docexceptExceptionase:logger.error(fStep 1: 生成假设文档失败:{e})raisee步骤4假设性答案进行向量查询(step_2_search_embedding_hyde)方法细化实现defstep_2_search_embedding_hyde(rewritten_query:str,hyde_doc:str,item_namesNone,req_limit:int10,top_k:int5,ranker_weights(0.8,0.2),# 调整默认权重以偏向稠密向量 (0.8, 0.2)norm_score:boolTrue,# 默认开启归一化output_fields[chunk_id,content,item_name],): 阶段2利用“重写问题 假设性文档”生成 embedding并到向量库检索切片。 :param rewritten_query: 改写后的查询 :param hyde_doc: Step 1 生成的假设性文档 :param item_names: 商品名称列表用于元数据过滤 (item_name in [...]) :param req_limit: Milvus 搜索时的候选召回数量 :param top_k: 最终返回的 Top K 结果数量 :param ranker_weights: 混合检索权重 (Dense, Sparse) :param norm_score: 是否对分数进行归一化 :param output_fields: 返回结果中包含的字段 :return: 检索结果列表 ifnotrewritten_query:raiseValueError(rewritten_query 不能为空)ifnothyde_doc:raiseValueError(hypothetical_doc 不能为空)# 1. 拼接查询与假设文档形成更丰富的语义上下文combined_textrewritten_query hyde_doc logger.info(fStep 2: 拼接 Query HyDE Doc, 总长度:{len(combined_text)})# 2. 生成向量 (Dense Sparse)logger.info(Step 2: 正在生成混合向量 (Embedding)...)embeddingsgenerate_embeddings([combined_text])# 3. 准备 Milvus 检索collection_nameos.environ.get(CHUNKS_COLLECTION)ifnotcollection_name:logger.error(Step 2 Error: 环境变量 CHUNKS_COLLECTION 未设置)return[]logger.info(fStep 2: 准备在集合 {collection_name} 中执行混合检索)# 构造过滤表达式 (如果有商品名限制)exprNoneifitem_names:# 处理 item_names 中的引号防止注入或语法错误quoted, .join(f{v}forvinitem_names)exprfitem_name in [{quoted}]logger.info(fStep 2: 应用过滤条件:{expr})else:logger.info(Step 2: 未指定商品名过滤将全库检索)try:# 构造搜索请求reqscreate_hybrid_search_requests(dense_vectorembeddings.get(dense)[0],sparse_vectorembeddings.get(sparse)[0],exprexpr,limitreq_limit,)clientget_milvus_client()ifnotclient:logger.error(Step 2 Error: 无法连接到 Milvus)return[]# 执行混合检索logger.info(fStep 2: 执行 Hybrid Search, Weights{ranker_weights}, TopK{top_k})reshybrid_search(clientclient,collection_namecollection_name,reqsreqs,ranker_weightsranker_weights,norm_scorenorm_score,limittop_k,output_fieldslist(output_fields),)hit_countlen(res[0])ifresandlen(res)0else0logger.info(fStep 2: 检索完成, 找到{hit_count}个匹配切片)returnresexceptExceptionase:logger.error(fStep 2: 检索过程发生异常:{e})return[]主流程测试代码if__name____main__:# 本地测试代码print(\n*50)print( 启动 node_search_embedding_hyde 本地测试)print(*50)# 模拟输入状态mock_state{session_id:test_hyde_session_001,original_query:HAK 180 烫金机怎么操作,rewritten_query:HAK 180 烫金机的具体操作步骤是什么,item_names:[HAK 180 烫金机],is_stream:False}try:# 运行节点resultnode_search_embedding_hyde(mock_state)print(\n*50)print( 测试结果摘要:)print(fHyDE Doc Generated:{bool(result.get(hyde_doc))})ifresult.get(hyde_doc):print(fDoc Preview:{result.get(hyde_doc)[:50]}...)chunksresult.get(hyde_embedding_chunks,[])print(fChunks Found:{len(chunks)}, chunks内容{chunks})ifchunks:print(fTop Chunk Score:{chunks[0].get(distance)})print(*50)exceptExceptionase:logger.exception(f测试运行期间发生未捕获异常:{e})
