掌柜智库项目(RAG)实战7. 检索数据图与状态定义7.1 定义状态 (State)所有节点共享同一个状态对象。我们需要定义它来存储处理过程中的数据如 查询原始问题、改写问题、历史对话、不同方向切片结果等。文件:app/query_process/agent/state.pyfromtyping_extensionsimportTypedDictfromtypingimportListclassQueryGraphState(TypedDict): QueryGraphState 定义了整个查询流程中流转的数据结构。 session_id:str# 会话唯一标识original_query:str# 用户原始问题# 检索过程中的中间数据embedding_chunks:list# 普通向量检索回来的切片hyde_embedding_chunks:list# HyDE 检索回来的切片kg_chunks:list# 图谱检索回来的切片web_search_docs:list# 网络搜索回来的文档# 排序过程中的数据rrf_chunks:list# RRF 融合排序后的切片reranked_docs:list# 重排序后的最终 Top-K 文档# 生成过程中的数据prompt:str# 组装好的 Promptanswer:str# 最终生成的答案# 辅助信息item_names:List[str]# 提取出的商品名称rewritten_query:str# 改写后的问题history:list# 历史对话记录is_stream:bool# 是否流式输出标记7.2 定义主图为了确保整体流程设计的科学性与执行连贯性我们采用“Top-Down”自顶向下的开发模式以 “总指挥部” 的全局视角统筹推进具体实施步骤如下搭建节点骨架Stubs优先定义全流程所需的所有功能节点仅保留核心日志打印能力如节点进入 / 退出日志暂不实现内部复杂业务逻辑快速搭建起流程的 “骨架结构”串联主图Graph基于预设的业务流转规则编写主图逻辑将所有节点骨架按序串联明确节点间的输入输出关系、分支判断条件如文件格式分流逻辑形成完整的流程链路验证流程通畅性启动端到端测试验证节点间的调用链路是否通顺、数据流转是否符合预期、分支跳转是否准确确保流程无阻塞、无逻辑漏洞填充节点核心逻辑在流程链路验证通过后再逐一聚焦每个节点的内部实现完成复杂业务逻辑的开发如 查询重写、定向查询、结果重排处理等实现 “骨架” 到 “完整系统” 的落地。该模式的核心优势在于先保障 “流程走得通”再聚焦 “功能做得好”避免因局部逻辑复杂导致整体流程设计偏差大幅提升开发效率与流程稳定性。7.2.1 第一步创建节点骨架我们需要先创建以下 8 个文件每个文件里只写一个最简单的“空函数”确保主图能导入它们。请在app/query_process/agent/nodes目录下创建以下文件并将对应代码复制进去。**(1) 意图确认: **node_item_name_confirm.py这个节点主要干了 4 件事提取与改写 结合历史对话提取商品名并将模糊问题改写为完整独立的精准问题。向量化检索 将提取出的商品名在 Milvus 向量库中进行混合搜索。标准化对齐 根据评分高低自动对齐标准型号或生成反问让用户手动确认。同步历史记录 将改写后的问题、确认的商品名和处理状态实时写入 MongoDB 数据库。importtimeimportsysfromkb.utils.task_utilsimportadd_running_task,add_done_taskdefnode_item_name_confirm(state): 节点功能确认用户问题中的核心商品名称。 输入state[original_query] 输出更新 state[item_names] print(f---node_item_name_confirm---开始处理)# 记录任务开始add_running_task(state[session_id],sys._getframe().f_code.co_name,state[is_stream])# 后面会调用大模型进行逻辑处理time.sleep(1)# 记录任务结束add_done_task(state[session_id],sys._getframe().f_code.co_name,state[is_stream])print(f---node_item_name_confirm---处理结束)return{item_names:[示例商品]}**(2) 向量检索: **node_search_embedding.py这个节点 node_search_embedding 负责根据 改写后的用户问题 在 限定的商品范围内 利用 BGEM3 混合检索稠密稀疏 技术从 Milvus 向量数据库中召回 Top5 最相关的知识切片。from.stateimportImportGraphStateimportsysdefnode_pdf_to_md(state:ImportGraphState)-ImportGraphState: 节点: PDF转Markdown (node_pdf_to_md) 为什么叫这个名字: 核心任务是将 PDF 非结构化数据转换为 Markdown 结构化数据。 未来要实现: 1. 调用 MinerU (magic-pdf) 工具。 2. 将 PDF 转换成 Markdown 格式。 3. 将结果保存到 state[md_content]。 print(f [Stub] 执行节点:{sys._getframe().f_code.co_name})returnstate**(3) HyDE 假设检索节点: **node_search_embedding_hyde.py这个节点 node_search_embedding_hyde 实现了 HyDE (Hypothetical Document Embeddings) 策略核心逻辑是 先让 LLM 虚构一个“理想答案”再用这个答案去向量库检索真实的文档 。一句话总结 它通过“LLM 生成假设性答案”来增强原始问题的语义信息再进行混合向量检索从而大幅提升对“语义匹配但字面不匹配”问题的召回能力。importtimeimportsysfromapp.utils.task_utilsimportadd_done_task,add_running_taskdefnode_search_embedding_hyde(state): 节点功能HyDE (Hypothetical Document Embedding) 先让 LLM 生成假设性答案再对答案进行向量检索提高召回率。 print(---HyDE 开始处理---)add_running_task(state[session_id],sys._getframe().f_code.co_name,state.get(is_stream))# 搜索假设性答案print(搜索架设性答案)time.sleep(1)# ...add_done_task(state[session_id],sys._getframe().f_code.co_name,state.get(is_stream))print(---HyDE 处理结束---)**(4) 网络搜索节点: **node_web_search_mcp.py这个节点 node_web_search_mcp 负责调用 百炼 MCP (Model Context Protocol) 联网搜索服务 获取互联网上的实时信息。一句话总结 它通过 MCP 协议异步调用百炼联网搜索接口将用户的查询转化为实时的、结构化的网络搜索结果包含标题、链接和摘要。importtimeimportsysfromapp.utils.task_utilsimportadd_done_task,add_running_taskdefnode_web_search_mcp(state): 节点功能调用外部搜索引擎补充信息 :param state: :return: add_running_task(state[session_id],sys._getframe().f_code.co_name,state[is_stream])print(---node-web-search-mcp处理---)add_done_task(state[session_id],sys._getframe().f_code.co_name,state[is_stream])time.sleep(1)# 调用mcp外部引擎print(f调用外部mcp引擎)print(---node-web-search-mcp处理结束---)**(5) 图谱查询节点[最后实现]: **node_query_kg.pyimporttimeimportsysfromapp.utils.task_utilsimportadd_running_task,add_done_taskdefnode_query_kg(state): 节点功能在 Neo4j 知识图谱中查询实体关系。 print( node_query_kg 图谱查询处理 )add_running_task(state[session_id],sys._getframe().f_code.co_name,state.get(is_stream))time.sleep(1)# ...add_done_task(state[session_id],sys._getframe().f_code.co_name,state.get(is_stream))**(6) 倒排融合排序节点: **node_rrf.py这个节点 node_rrf 负责 将来自不同检索源如向量检索、HyDE 检索、知识图谱等的文档列表进行统一合并和重新排序 。它消除了不同检索方式的分数差异给出一个综合排名最高的文档列表Top 10。什么是 RRF (Reciprocal Rank Fusion, 倒排排序融合)RRF 是一种 不需要知道具体分数只关心排名 的融合算法。通俗解释 假设你在参加一场全能比赛裁判 A向量检索 觉得你是第 1 名。裁判 BHyDE 检索 觉得你是第 3 名。裁判 C关键字检索 觉得你是第 10 名。RRF 不管裁判打具体多少分因为不同裁判打分标准不同有的满分100有的满分10它只看 排名 。计算公式rank 是你在某个裁判那里的排名第1名就是1第2名就是2。k 是一个常数通常取 60用来防止排名靠前的差距过大。RRF 的核心思想奖励多面手 如果你在多个榜单上都排在前面你的总分就会很高。平滑差异 即使你在某个榜单上稍微落后只要其他榜单表现好总分依然能上去。去重 同一个文档在不同榜单出现多次RRF 会把它合并只算一次总分。举个栗子假设我们有两个检索来源来源 A (向量检索) 找出了前 3 名第 1 名 文档 X第 2 名 文档 Y第 3 名 文档 Z来源 B (HyDE 检索) 找出了前 3 名第 1 名 文档 Y (注意它觉得 Y 比 X 好)第 2 名 文档 Z第 3 名 文档 W (新面孔)我们来算一下每个文档的总分文档 X在 A 中排第 1 → 得分 1 / (60 1) 0.0164在 B 中没出现 → 得分 0总分 0.0164 0 0.0164文档 Y (重点看这个)在 A 中排第 2 → 得分 1 / (60 2) 0.0161在 B 中排第 1 → 得分 1 / (60 1) 0.0164总分 0.0161 0.0164 0.0325文档 Z在 A 中排第 3 → 得分 1 / (60 3) 0.0159在 B 中排第 2 → 得分 1 / (60 2) 0.0161总分 0.0159 0.0161 0.0320文档 W在 A 中没出现 → 得分 0在 B 中排第 3 → 得分 1 / (60 3) 0.0159总分 0 0.0159 0.0159最终排名结果第一名文档 Y (0.0325) (虽然在 A 排第二但在 B 排第一综合实力最强)第二名文档 Z (0.0320) (两个榜单都上榜表现稳定)第三名文档 X (0.0164) (偏科只在 A 表现好)第四名文档 W (0.0159)总结 RRF 就是通过这种方式让 “在多个榜单都靠前” 的文档排到最前面比单看某一个榜单更靠谱importtimeimportsysfromapp.utils.task_utilsimportadd_running_task,add_done_taskdefnode_rrf(state): 节点功能Reciprocal Rank Fusion 将多路召回的结果向量、HyDE、Web、KG进行加权融合排序。 print(---RRF---)add_running_task(state[session_id],sys._getframe().f_code.co_name,state.get(is_stream))time.sleep(1)# ...add_done_task(state[session_id],sys._getframe().f_code.co_name,state.get(is_stream))**(7) 重排序节点: **node_rerank.py这个节点 node_rerank 是知识库检索的“精修师”它对之前所有步骤召回的文档进行 二次精细排序 。核心流程分为三步合并文档 将来自 RRF本地检索和 Web Search联网搜索的文档合并到一个池子中。精确打分 使用重排序模型计算每个文档与用户问题的相关性得分。动态截断 根据得分的“断崖式下跌”点智能截取 TopK最多 10 条只保留高质量结果过滤凑数的低分文档。该节点使用的是 BGE Reranker (BAAI/bge-reranker-large) 模型。模型类型 Cross-Encoder交叉编码器。工作原理 它不像向量检索那样分别计算向量再比对而是直接把“问题”和“文档”拼接在一起扔给模型让模型像阅读理解一样深入分析两者的语义匹配度。特点 精度极高 能识别微小的语义差异如“苹果手机”和“苹果公司”的区别。速度较慢 因为计算量大所以通常只用于对初筛后的少量文档如 Top 20-50进行精排不适合全库检索。importtimeimportsysfromapp.utils.task_utilsimportadd_running_task,add_done_taskdefnode_rerank(state): 节点功能使用 Cross-Encoder 模型对 RRF 后的结果进行精确打分重排。 print(---Rerank处理---)add_running_task(state[session_id],sys._getframe().f_code.co_name,state.get(is_stream))time.sleep(1)# ...add_done_task(state[session_id],sys._getframe().f_code.co_name,state.get(is_stream))**(8) 答案生成: **node_answer_output.py这个节点 node_answer_output 是知识库查询的“最后一公里”负责 生成最终回答 并 交付给用户 。它整合了之前所有步骤的成果通过以下 5 个核心动作完成任务检查前置答案 如果之前步骤如商品名确认节点已经生成了追问或拒绝回答直接输出跳过 LLM 生成。构建 Prompt 将用户问题、历史对话、以及 Rerank 后的 TopK 高质量文档片段包含元数据组装成一段严谨的提示词。LLM 生成与流式推送 调用大模型生成最终答案。如果是流式模式会**逐字推送Delta**给前端实现打字机效果。图片提取与增强 从引用文档中自动提取图片 URL包括网页链接和本地 Markdown 图片为纯文本答案补充视觉信息。收尾与存档 将最终答案和提取的图片写入 MongoDB 历史记录并向前端发送包含完整信息答案图片的 FINAL 信号 。from.stateimportImportGraphStateimportsysdefnode_import_kg(state:ImportGraphState)-ImportGraphState: 节点: 导入知识图谱 (node_import_kg) 为什么叫这个名字: 构建 Knowledge Graph (KG) 并存入 Neo4j。 未来要实现: 1. 调用 LLM 从文本中抽取实体 (Entity) 和关系 (Relation)。 2. 连接 Neo4j 数据库。 3. 执行 Cypher 语句将图谱数据写入数据库。 print(f [Stub] 执行节点:{sys._getframe().f_code.co_name})returnstate7.2.2 第二步编写主图代码文件:app/query_process/agent/main_graph.py1初始化图与节点注册fromlanggraph.graphimportStateGraph,ENDfromkb.query_process.stateimportQueryGraphState# 导入所有节点函数fromkb.query_process.node_item_name_confirmimportnode_item_name_confirmfromkb.query_process.node_query_kgimportnode_query_kgfromkb.query_process.node_answer_outputimportnode_answer_outputfromkb.query_process.node_rerankimportnode_rerankfromkb.query_process.node_rrfimportnode_rrffromkb.query_process.node_search_embeddingimportnode_search_embeddingfromkb.query_process.node_search_embedding_hydeimportnode_search_embedding_hydefromkb.query_process.node_web_search_mcpimportnode_web_search_mcp# 初始化状态图builderStateGraph(QueryGraphState)# 注册所有节点builder.add_node(node_item_name_confirm,node_item_name_confirm)# 确认商品builder.add_node(node_multi_search,lambdax:x)# 虚拟节点多路搜索分叉点builder.add_node(node_search_embedding,node_search_embedding)# 向量搜索builder.add_node(node_search_embedding_hyde,node_search_embedding_hyde)builder.add_node(node_query_kg,node_query_kg)builder.add_node(node_web_search_mcp,node_web_search_mcp)builder.add_node(node_join,lambdax:{})# 虚拟节点多路搜索合并点builder.add_node(node_rrf,node_rrf)# 排序builder.add_node(node_rerank,node_rerank)# 重排builder.add_node(node_answer_output,node_answer_output)# 生成# 虚拟节点的作用作为流程的「分叉 / 合并中转站」解决多分支流程的组织问题本身无业务逻辑# lambda x:x 含义接收 state 并原样返回是最轻便的 “无逻辑传递” 方式# 普通函数替换定义 def 函数名(state): return state 即可完全等价优势是易扩展、易调试# 设置起点builder.set_entry_point(node_item_name_confirm)defroute_after_item_confirm(state:QueryGraphState):# 如果已有答案Branch B/C直接跳到输出ifstate.get(answer): 这主要发生在 node_item_name_confirm 节点无法直接确定唯一的商品型号从而需要“反问用户”或“拒绝回答”的场景。 具体来说有以下两种情况会导致 state 中直接出现 answer 从而跳过后续的检索流程直接输出 1. 多选一反问用户 - 场景 用户问得太模糊比如“华为P60”系统发现数据库里有“华为P60 128G”和“华为P60 Art”两个型号且置信度都不足以直接确认。 - 处理 节点会生成一条反问句作为 answer 例如“您是想问以下哪个产品华为P60 128G、华为P60 Art请明确一下型号。” - 结果 此时不需要再去检索文档了直接把这句话发给用户让他选。 2. 查无此人拒绝回答 - 场景 用户问了一个系统里压根没有的商品比如“小米15”但库里只有华为的数据或者评分过低0.6。 - 处理 节点会生成一条拒绝句作为 answer 例如“抱歉未找到相关产品请提供准确型号以便我为您查询。” - 结果 同样不需要后续检索直接结束流程。 returnnode_answer_output# 否则继续搜索流程returnnode_multi_search# 1. 意图确认 - (条件分叉) - 多路搜索 / 答案输出builder.add_conditional_edges(node_item_name_confirm,route_after_item_confirm)# 2. 并发执行四路搜索builder.add_edge(node_multi_search,node_search_embedding)builder.add_edge(node_multi_search,node_search_embedding_hyde)builder.add_edge(node_multi_search,node_web_search_mcp)builder.add_edge(node_multi_search,node_query_kg)# 3. 四路搜索 - 结果合并builder.add_edge(node_search_embedding,node_join)builder.add_edge(node_search_embedding_hyde,node_join)builder.add_edge(node_web_search_mcp,node_join)builder.add_edge(node_query_kg,node_join)# 4. 合并 - 排序 - 重排 - 生成 - 结束builder.add_edge(node_join,node_rrf)builder.add_edge(node_rrf,node_rerank)builder.add_edge(node_rerank,node_answer_output)builder.add_edge(node_answer_output,END)# 编译生成可执行的 Runnable 应用query_appbuilder.compile()7.2.3 第三步验证图流程在实现具体业务逻辑前我们先跑一个测试脚本看看图能不能跑通路线对不对。创建测试文件:test_graph_flow.py(在项目根目录)fromknowledge_base_0115.kb.import_process.main_graphimportkb_import_appfromknowledge_base_0115.kb.import_process.stateimportImportGraphStateimportsysdeftest_pdf_flow():print(\n测试PDF文件处理流程)# 模拟初始化状态initial_stateImportGraphState(task_idtest_task_001,local_file_pathtest.pdf,local_dir./output,# 确保相关开关被正确初始化根据您的 state 定义有些可能是默认值is_pdf_read_enabledTrue)# 运行图print(开始运行....)try:# 修正点使用 .invoke() 方法resultkb_import_app.invoke(initial_state)print(运行结束最终的状态 keys:,result.keys())exceptExceptionase:print(f运行报错{e})# 打印详细堆栈以便调试importtraceback traceback.print_exc()if__name____main__:print(----,sys.path)test_pdf_flow()预期效果:运行python test_graph_flow.py你应该能看到控制台依次打印出每个节点的 [Stub] 执行节点: ...日志。PDF 流程应包含node_entry-node_pdf_to_md-node_md_img- … -node_import_kgMarkdown 流程应包含node_entry-node_md_img- … -node_import_kg(跳过了node_pdf_to_md)如果能看到这些日志说明我们的图结构搭建成功接下来就可以放心地去填充每个节点的具体代码了。