相关项目代码已上传githubgiteecsdn等等关键词LangGraph、RAG、MinerU、BGE-M3、混合检索、HyDE、RRF、SSE、Milvus一、 承接上篇从“前端交互”深入到“后端大脑”在上一篇《从零构建旅游知识库 RAG 系统上》中我们完成了整套项目的前端交互架构落地实现了大文件无阻塞上传、实时进度可视化、SSE 流式对话、会话持久化等核心能力解决了 RAG 项目前端落地的各类交互痛点。前端页面展示的进度条、状态徽章、逐字流式问答、日志更新等所有交互效果本质都是后端核心流水线的状态反馈。对于初学者而言后端的流程调度、数据处理、检索推理逻辑往往是最难吃透的“黑盒”。本篇将逐层拆解后端核心架构彻底打通前后端联动逻辑深度剖析两大核心业务流水线知识库导入流水线Import Pipeline详解普通 PDF/Markdown 旅游文档如何经过解析、处理、切片、向量化最终结构化存入 Milvus 向量数据库的全流程。智能问答查询流水线Query Pipeline拆解用户提问后系统如何通过多路检索、结果融合、重排序、大模型推理生成高质量旅游问答答案的完整链路。全文不仅提供可直接落地的LangGraph 流程图定义代码还会逐节点拆解核心实现包含MinerU 子进程调用、多模态图片摘要生成、HyDE 假设性文档检索、RRF 倒数排名融合、BGE-reranker 断崖截断算法等核心技术落地细节。二、 项目后端整体架构一览在进入代码细节拆解前我们先搭建全局架构认知。本项目后端采用业务解耦、分层设计、状态驱动的轻量化架构将知识库导入、智能查询两大核心业务完全隔离配套独立的配置、异常、节点、状态管理体系同时通过通用工具层统一封装所有外部依赖极大降低迭代与维护成本。完整后端目录结构及模块职责如下processor/ ├── import_model/ # 知识库导入模块数据入库专属流水线 │ ├── base.py # 节点基类统一日志与异常处理 │ ├── config.py # 导入配置环境变量管理 │ ├── enums.py # 枚举ContentType, FileCheckResult │ ├── exceptions.py # 自定义异常体系 │ ├── import_node_state.py # 状态类型定义TypedDict │ ├── import_main_graph.py # 导入流程图定义核心 │ └── nodes/ # 导入流水线所有业务节点 │ ├── import_file_node.py │ ├── pdf_to_md_node.py │ ├── remake_md_img_node.py │ ├── document_split_node.py │ └── get_name_node.py ├── search_model/ # 智能问答查询模块用户交互专属流水线 │ ├── base.py │ ├── config.py │ ├── exceptions.py │ ├── search_node_state.py │ ├── search_main_graph.py # 查询流程图定义核心 │ └── nodes/ # 查询流水线所有业务节点 │ ├── rewritter_query_node.py │ ├── vector_search_node.py │ ├── hyde_search_node.py │ ├── web_search_node.py │ ├── rrf_search_node.py │ ├── reranker_search_node.py │ └── answer_output_node.py ├── prompts/ # 全局提示词模板 │ ├── import_prompt/ │ │ └── get_name_prompt.py │ └── query/ │ ├── query_rewritter_prompt.py │ └── answer_prompt.py └── utils/ # 基础设施通用层 ├── bge_client_util.py # BGE-M3 稠密稀疏向量生成 ├── bge_rerank_util.py # BGE-reranker 结果重排序 ├── milvus_client_util.py # Milvus 混合检索封装 ├── mongo_client_util.py # MongoDB 对话历史存储 ├── minio_util.py # MinIO 图片对象存储 ├── llm_client_util.py # 大模型统一调用客户端 └── sse_util.py # SSE 流式消息推送队列架构设计核心哲学双流水线物理隔离导入、查询两套独立 StateGraph互不抢占资源、互不干扰运行支持单独扩容、单独迭代适配数据入库、用户问答高并发场景。状态驱动解耦所有业务节点通过统一 state 字典传递数据LangGraph 负责流程编排单个节点仅需实现「输入-处理-输出」逻辑无硬编码耦合。基础设施抽象Milvus、MongoDB、MinIO、大模型等所有外部依赖全部封装在 utils 工具层业务节点仅调用通用接口无需关注底层实现降低改造成本。三、 全局任务状态集成让前端“看见”后端在干什么回顾上篇内容前端通过定时轮询/status/{task_id}接口获取后端任务的done_list已完成节点和running_list运行中节点从而渲染进度条和日志。很多初学者会疑惑LangGraph 自主调度的业务节点如何同步状态到前端可读的任务状态管理器中核心答案通过节点生命周期钩子实现前后端状态联动在每个业务节点执行前后主动调用全局状态工具标记节点运行、完成状态实现前后端进度完全同步。3.1 流水线全局状态调度入口以下是导入流水线的核心调度函数负责初始化任务状态、编译执行流程图、监听节点运行事件是联动task_util的核心入口。# processor/import_model/import_main_graph.py def run_import_pipeline( task_id: str, origin_path_or_url: str, origin_file_name: str , **kwargs ) - ImportNodeState: # 1. 构建初始状态绑定唯一任务ID全局状态关联核心 init_state { task_id: task_id, origin_path_or_url: origin_path_or_url, origin_file_name: origin_file_name, } init_state.update(kwargs) state create_default_state(**init_state) # 2. 编译并执行LangGraph流程图 app create_import_graph() for event in app.stream(state): for node_name, state_data in event.items(): # 3. 核心逻辑监听每一个节点的运行事件 # 节点内部主动调用task_util更新状态实现前端进度同步 print(f运行节点{node_name}) final_state state_data return final_state3.2 单节点状态精准更新实现单纯监听全局事件无法精准区分节点运行状态本项目采用节点内主动上报的优雅方案在每个节点的核心执行方法中手动调用全局状态工具标记节点开始运行、执行完成异常场景可手动标记任务失败。以 PDF 转 Markdown 节点为例完整落地代码如下# processor/import_model/nodes/pdf_to_md_node.py from front.utils.task_util import add_running_task, add_done_task class PdfToMdNode(BaseNode): def process(self, state: ImportNodeState) - ImportNodeState: task_id state.get(task_id) # 1. 节点启动标记当前节点进入运行状态前端实时感知 add_running_task(task_id, self.name) # self.name pdf_to_md try: # 执行 PDF 转 Markdown 的核心业务逻辑 process_code self.execu_minerU(origin_path_or_url, file_output_path) if process_code ! 0: raise PdfConversionError(转换失败) # 2. 节点结束标记当前节点执行完成更新前端进度 add_done_task(task_id, self.name) return state except Exception as e: # 异常兜底可扩展为标记任务整体失败前端展示异常状态 raise 落地踩坑提示所有业务节点的self.name必须和上篇task_util.py中的节点映射字典完全一致否则会出现前端状态不更新、节点名称展示异常的问题这是前后端联调最容易忽略的细节。四、 导入流水线Import Pipeline从 PDF 到向量库的“变形记”导入流水线是整个 RAG 系统的数据源头负责将原始旅游文档转化为可检索的结构化向量数据。整条流水线由 5 个核心节点串联组成支持 PDF、Markdown 双格式适配包含格式校验、格式转换、图片处理、智能切片、实体抽取、向量入库全流程。4.1 导入流程图整体定义本流程图采用条件分支路由设计可根据上传文件格式PDF/MD自动适配不同处理逻辑避免无效流程执行提升入库效率。# processor/import_model/import_main_graph.py def create_import_graph(): graph StateGraph(ImportNodeState) # 注册所有导入流水线核心节点 nodes { import_file: ImportFileNode(), pdf_to_md: PdfToMdNode(), remake_img: RemakeMdImgNode(), split_doc: DocumentSplitNode(), extract_name: GetNameNode(), } for name, node in nodes.items(): graph.add_node(name, node) # 设置流水线入口节点 graph.set_entry_point(import_file) # 条件路由根据文件类型分支执行优化流程效率 graph.add_conditional_edges( import_file, router_after_import, { pdf: pdf_to_md, # PDF文件先走转换流程 md: remake_img, # MD文件直接进入图片处理流程 } ) # 固定顺序流水线分支执行后统一收尾 graph.add_edge(pdf_to_md, remake_img) graph.add_edge(remake_img, split_doc) graph.add_edge(split_doc, extract_name) graph.add_edge(extract_name, END) return graph.compile()4.2 节点一ImportFileNode —— 文件格式检测与校验作为流水线首个执行节点核心职责是文件合法性校验、格式识别、基础参数初始化拦截无效文件避免后续流程无效执行。仅支持 PDF、MD 两种知识库格式适配旅游文档入库场景。# processor/import_model/nodes/import_file_node.py class ImportFileNode(BaseNode): name import_file def process(self, state: ImportNodeState) - ImportNodeState: origin_path state.get(origin_path_or_url) path_obj Path(origin_path) # 通用文件合法性校验基类封装方法 if file_validation(path_obj) ! FileCheckResult.VALID: raise FileProcessingError(文件无效请检查文件完整性) # 识别文件后缀校验格式合法性 file_type path_obj.suffix.lower() if file_type not in [.md, .pdf]: raise FileProcessingError(仅支持 PDF/MD 格式知识库文件) # 初始化状态参数供后续节点调用 state[file_type] file_type state[origin_file_name] path_obj.stem # MD文件无需转换直接初始化最终MD路径 if file_type .md: state[final_md_path] str(path_obj) return state 落地提示此处做了严格格式拦截可根据业务需求拓展 DOCX、TXT 等格式只需修改格式白名单并适配对应转换节点即可。4.3 节点二PdfToMdNode —— MinerU 高精度PDF转换这是导入流水线中最耗时、最核心的重型节点通过子进程调用 MinerU 工具实现 PDF 无损解析精准还原文档文字、图片、排版结构完美适配旅游攻略、景点介绍等图文混合文档。同时内置国内镜像加速配置解决 MinerU 依赖模型下载缓慢、失败的问题适配国内部署环境。# processor/import_model/nodes/pdf_to_md_node.py class PdfToMdNode(BaseNode): name pdf_to_md def process(self, state: ImportNodeState) - ImportNodeState: origin_path state.get(origin_path_or_url) file_name state.get(origin_file_name) # 定义文件输出目录 output_dir Path(origin_path).parent / file_name state[file_output_dir] str(output_dir) # 执行PDF转MD核心逻辑 process_code self.execu_minerU(origin_path, output_dir) if process_code ! 0: raise PdfConversionError(MinerU 转换失败请检查文件格式或工具依赖) # MinerU固定输出路径初始化最终MD文件路径 final_md output_dir / auto / f{file_name}.md state[final_md_path] str(final_md) return state def execu_minerU(self, origin_path, output_dir): # 核心优化配置国内HF镜像加速模型下载解决外网访问失败问题 os.environ[HF_ENDPOINT] http://hf-mirror.com # MinerU执行命令配置适配轻量化CPU部署 cmd [ mineru, -p, str(origin_path), -o, str(output_dir), --source, local, --device, cpu, --backend, pipeline, --batch-size, 1, --no-auto-download ] # 子进程实时流式读取日志便于调试排错 process subprocess.Popen( cmd, stdoutsubprocess.PIPE, stderrsubprocess.STDOUT, textTrue, encodingutf-8, bufsize1, # 行缓冲实时输出日志 ) # 实时打印转换日志便于定位异常 for line in process.stdout: line line.strip() self.logger.info(line) return process.wait() # 返回进程退出码判断转换结果 踩坑重点必须配置HF_ENDPOINT国内镜像环境变量否则 MinerU 会因无法连接外网模型源导致转换失败、模型下载超时这是国内服务器部署的核心必填配置。4.4 节点三RemakeMdImgNode —— 图片处理与多模态摘要生成旅游知识库的核心特点是图文结合单纯文字信息无法满足用户查询需求。本节点专门处理文档图片资源完成图片提取、多模态摘要生成、对象存储上传、链接替换全流程优化。核心处理逻辑扫描 MD 文档中所有本地图片资源提取图片上下文信息调用多模态大模型结合图文上下文生成精准中文摘要将本地图片上传至 MinIO 对象存储实现资源持久化替换 MD 本地图片链接为公网链接更新图片 alt 文本为 AI 摘要。# processor/import_model/nodes/remake_md_img_node.py class RemakeMdImgNode(BaseNode): name remake_img def process(self, state: ImportNodeState) - ImportNodeState: final_md_path state.get(final_md_path) with open(final_md_path, r, encodingutf-8) as f: md_content f.read() # 1. 扫描文档所有图片及上下文信息 img_infos self.scan_images(md_content, final_md_path) if not img_infos: return state # 无图片直接跳过提升执行效率 # 2. 批量生成图片多模态摘要 summaries {} for img_name, img_path, title, pre, post in img_infos: summary self.get_summary_llm(img_path, title, pre, post) summaries[img_name] summary # 3. 初始化MinIO客户端批量上传图片 minio_client get_minio_client() bucket self.config.minio_bucket origin_name state.get(origin_file_name) for img_name, img_path, _, _, _ in img_infos: # 定义对象存储路径避免文件重名冲突 object_name f{origin_name}/{img_name} minio_client.fput_object(bucket, object_name, str(img_path)) # 生成公网访问链接 url fhttp://{self.config.minio_endpoint}/{bucket}/{object_name} # 4. 替换文档本地链接为线上链接 md_content md_content.replace(img_name, url) # 替换图片alt文本为AI生成摘要提升检索语义丰富度 md_content re.sub( r!\[(.*?)\]\(.*? re.escape(img_name) r.*?\), f![{summaries[img_name]}]({url}), md_content ) # 5. 写回优化后的MD文档 with open(final_md_path, w, encodingutf-8) as f: f.write(md_content) return state def get_summary_llm(self, img_path, title, pre, post): # 构建图片上下文辅助大模型生成精准摘要 context f标题{title}\n上文{pre}\n下文{post} # 图片base64编码适配多模态接口入参 with open(img_path, rb) as f: img_b64 base64.b64encode(f.read()).decode() # 调用OpenAI兼容多模态模型 client OpenAI(api_keyself.config.openai_api_key, base_urlself.config.openai_api_base) response client.chat.completions.create( modelself.config.vl_model, messages[{ role: user, content: [ {type: text, text: f结合上下文{context}用一句话描述这张图片不要包含图片二字。}, {type: image_url, image_url: {url: fdata:image/jpeg;base64,{img_b64}}} ] }] ) return response.choices[0].message.content.strip()4.5 节点四DocumentSplitNode —— 语义切分与向量入库文档切片是 RAG 系统检索精度的核心关键。本节点摒弃粗暴的固定长度切分采用标题层级语义切分动态合并拆分策略最大程度保留旅游文档的语义完整性同时完成实体匹配、双向量生成、Milvus 批量入库。核心能力按 Markdown 一、二级标题层级逻辑切分保留文档原有语义结构超大文本递归细切分碎片化小文本自动合并智能匹配旅游实体景点、城市、美食用于检索精准过滤生成 BGE-M3 稠密稀疏双向量适配混合检索模式。# processor/import_model/nodes/document_split_node.py class DocumentSplitNode(BaseNode): name split_doc def process(self, state: ImportNodeState) - ImportNodeState: final_md_path state.get(final_md_path) with open(final_md_path, r, encodingutf-8) as f: lines f.read().split(\n) # 1. 基于标题层级语义切分保证语义完整性 chunks self.get_title_split_chunks(lines, state.get(origin_file_name)) # 2. 大块细分、小块合并统一切片长度标准 chunks_list self.get_split_and_merge_chunks_list(chunks) # 3. 向量化并批量存入Milvus向量库 self.store_chunks(chunks_list) state[chunks_list] chunks_list return state def store_chunks(self, chunks_list): collection self.config.chunks_collection milvus get_milvus_client() # 批量提取文本内容生成双向量 contents [c[content] for c in chunks_list] bge get_bgem3_client() vectors generate_vector_data(bge, contents) # 输出稠密稀疏向量 # 构建入库数据绑定实体标签 insert_data [] for idx, chunk in enumerate(chunks_list): # 智能匹配旅游实体用于后续检索过滤 entity_names self.match_entities(chunk[content], chunk[title], chunk[parent_title]) insert_data.append({ content: chunk[content], title: chunk[title], parent_title: chunk[parent_title], entity_names: entity_names, # 实体数组支持标量过滤 dense_vector: vectors[dense][idx], sparse_vector: vectors[sparse][idx], }) # 批量插入向量库提升入库效率 milvus.insert(collection_namecollection, datainsert_data) 核心优势解析entity_names数组字段建立 Milvus 倒排索引查询时可通过ARRAY_CONTAINS_ANY做实体过滤大幅减少无效检索精准匹配用户查询的旅游目标。4.6 节点五GetNameNode —— 全局旅游实体抽取作为导入流水线最后一个节点核心职责是文档元信息提取与归类。通过大模型智能识别文档中的旅游核心实体涵盖景点、线路、酒店、美食、城市等维度同时判定文档类型为后续问答的意图识别、实体过滤提供数据支撑。# processor/import_model/nodes/get_name_node.py class GetNameNode(BaseNode): name extract_name def process(self, state: ImportNodeState) - ImportNodeState: chunks state.get(chunks_list) # 切片过多时采样处理避免大模型Token超限 if len(chunks) self.config.max_chunks_for_extract: chunks self.select_chunks(chunks) # 合并文档核心内容用于实体抽取 combined \n\n.join([c[content] for c in chunks]) # 调用大模型抽取旅游实体信息 response self.call_llm(combined, state.get(origin_file_name)) data json.loads(response) # 初始化各类实体状态 state[sight_name] data.get(sight_name, []) state[line_name] data.get(line_name, []) state[hotel_name] data.get(hotel_name, []) state[restaurant_name] data.get(restaurant_name, []) state[region_city] data.get(region_city, []) state[content_type] self.CONTENT_TYPE_MAP.get(data.get(content_type)) # 实体向量化入库支撑精准检索过滤 self.store_entities(state) return state五、 查询流水线Query Pipeline多路召回与融合排序查询流水线是用户交互的核心承接前端问答请求完成问题优化、多路并行检索、结果融合、重排序、答案生成、流式推送全链路处理。相比传统单向量检索本项目采用向量检索HyDE假设检索联网搜索三路并行方案兼顾知识库私有数据精度与实时网络信息时效性。5.1 查询流程图整体定义流程图采用并行检索串行精排架构通过虚拟节点实现多任务并行执行最大化检索效率同时支持问题澄清交互能力。# processor/search_model/search_main_graph.py def create_search_graph(): graph StateGraph(SearchNodeState) # 注册所有查询流水线节点 graph.add_node(query_analysis, RewritingQueryNode()) graph.add_node(multi_search, lambda x: x) # 并行分发虚拟节点 graph.add_node(vector_search, VectorSearchNode()) graph.add_node(hyde_search, HydeSearchNode()) graph.add_node(web_search, WebSearchNode()) graph.add_node(join, lambda x: x) # 结果汇合虚拟节点 graph.add_node(rrf_search, RRFSearchNode()) graph.add_node(reranker_search, RerankerSearchNode()) graph.add_node(answer_output, AnswerOutputNode()) # 设置流水线入口 graph.set_entry_point(query_analysis) # 条件路由模糊问题触发用户澄清精准问题直接检索 graph.add_conditional_edges( query_analysis, route_after_analysis, { need_clarification: END, # 需要用户澄清直接终止流程返回候选项 proceed: multi_search, # 问题清晰进入并行检索 } ) # 三路检索并行执行提升检索速度 graph.add_edge(multi_search, vector_search) graph.add_edge(multi_search, hyde_search) graph.add_edge(multi_search, web_search) # 所有检索结果汇合进入后续精排流程 graph.add_edge(vector_search, join) graph.add_edge(hyde_search, join) graph.add_edge(web_search, join) # 结果融合 - 重排序 - 答案生成 串行流程 graph.add_edge(join, rrf_search) graph.add_edge(rrf_search, reranker_search) graph.add_edge(reranker_search, answer_output) graph.add_edge(answer_output, END) return graph.compile()5.2 节点一RewritingQueryNode —— 问题重写与意图澄清用户自然语言提问往往存在模糊、简略、语义不清晰的问题。本节点核心能力上下文补全、问题标准化重写、意图识别、模糊问题澄清从源头提升检索精度。# processor/search_model/nodes/rewritter_query_node.py class RewritingQueryNode(BaseNode): name query_analysis def process(self, state: SearchNodeState) - SearchNodeState: query state.get(original_query) history state.get(history, []) # 调用提示词模板实现问题重写与意图分析 response self.call_llm(query, history) data json.loads(response) # 初始化处理后参数 state[rewritten_query] data.get(rewritten_query, query) state[entities] self.build_entities(data) # 解析旅游实体 state[need_clarification] data.get(need_clarification, False) state[candidate_options] data.get(candidate_options, []) return state业务场景说明用户提问“成都游玩攻略”系统可正常检索若用户提问“那边有什么好玩的”无上下文节点会判定需要澄清向前端推送候选问题避免无效检索。5.3 三路并行检索核心节点1VectorSearchNode —— 知识库向量精准检索基于重写后的标准问题生成双向量检索本地旅游知识库核心亮点是实体标量过滤混合检索精准筛选匹配用户查询的景点、城市、美食等内容过滤无关文档。# processor/search_model/nodes/vector_search_node.py def _build_filter_expr(self, entities): # 提取用户查询中的旅游实体构建过滤条件 names [e[name] for e in entities if e.get(name)] if not names: return # 利用Milvus数组匹配能力精准过滤含目标实体的文档 name_list ,.join(f{name} for name in names) return fARRAY_CONTAINS_ANY(entity_names, [{name_list}])2HydeSearchNode —— 假设性文档检索HyDE 是解决问句与文档语义鸿沟的经典优化方案。核心逻辑不直接用用户短问句检索而是让大模型先生成一段完整的「假设性标准答案」再用标准答案向量检索大幅提升语义匹配度。# processor/search_model/nodes/hyde_search_node.py class HydeSearchNode(BaseNode): name hyde_search def process(self, state): query state.get(rewritten_query) # 1. 大模型生成假设性完整答案 hyde_doc self._generate_hyde_document(query, state.get(entities)) # 2. 拼接原问题假设答案丰富检索语义 embedding_text f{query}\n{hyde_doc} # 3. 生成向量并检索 vectors generate_vector_data(bge_client, [embedding_text]) # 执行Milvus混合检索逻辑 state[hyde_embedding_chunks] result return state3WebSearchNode —— 实时联网检索本地知识库存在信息滞后、更新不及时的问题如景点门票、开放时间、最新活动。本节点基于 MCP 协议调用阿里云百炼搜索工具获取实时旅游资讯弥补静态知识库短板。# processor/search_model/nodes/web_search_node.py async def _execute_web_search(self, query): # 初始化MCP搜索客户端 mcp_client MCPServerStreamableHttp( name通用搜索, params{url: self.config.mcp_dashscope_base_url, headers: headers} ) await mcp_client.connect() # 调用联网搜索工具获取3条高质量实时结果 result await mcp_client.call_tool( tool_namebailian_web_search, arguments{query: query, count: 3} ) # 解析搜索结果提取标题、摘要、链接 return [{snippet: p[snippet], title: p[title], url: p[url]} for p in data[pages]]5.4 RRFSearchNode —— 多路结果倒数排名融合三路检索会产出多批异构结果无法直接对比分数。本项目采用RRF 倒数排名融合算法不依赖绝对分数仅根据排序位置加权融合完美适配多源结果合并场景。核心公式score Σ (1 / (k rank 1))k 为平滑系数避免排名靠前结果分数溢出。# processor/search_model/nodes/rrf_search_node.py class RRFSearchNode(BaseNode): RRF_K 60 # 行业通用平滑系数 def _rrf_merge(self, rrf_inputs): chunk_scores {} # 遍历多路检索结果按排名加权计分 for doc_list, weight in rrf_inputs: for rank, doc in enumerate(doc_list): chunk_id doc.get(chunk_id) or fhash_{hash(doc[content])} score weight / (self.RRF_K rank 1) chunk_scores[chunk_id] chunk_scores.get(chunk_id, 0.0) score # 按融合分数降序排序输出统一结果列表 return sorted(chunk_scores.items(), keylambda x: x[1], reverseTrue)5.5 RerankerSearchNode —— 精排与断崖截断优化RRF 融合仅为粗排结果仍存在冗余内容。本节点通过BGE-reranker 精排模型对候选结果做相关性打分同时独创分数断崖检测算法自动截断低质量内容避免无效文本输入大模型节省 Token 成本、提升答案精度。# processor/search_model/nodes/reranker_search_node.py class RerankerSearchNode(BaseNode): CLIFF_THRESHOLD 0.5 # 断崖分数阈值 def process(self, state): query state.get(rewritten_query) merged self._merge_multi_data(state) # 合并所有检索结果 # 批量计算问答相关性分数 reranker get_reranker_model() pairs [(query, doc[content]) for doc in merged] scores reranker.compute_score(pairs) # 绑定分数并排序 for doc, score in zip(merged, scores): doc[score] score sorted_docs sorted(merged, keylambda x: x[score], reverseTrue) # 核心断崖检测自动截断低质量结果 cut_pos len(sorted_docs) for i in range(2, len(sorted_docs) - 1): # 分数骤降超过阈值判定为质量断层截断后续内容 if sorted_docs[i][score] - sorted_docs[i1][score] self.CLIFF_THRESHOLD: cut_pos i 1 break state[reranked_chunks] sorted_docs[:cut_pos] return state5.6 AnswerOutputNode —— 流式答案生成与SSE推送作为流水线最后一环负责拼接高质量检索上下文、调用大模型生成答案同时兼容流式/非流式双模式。流式模式下通过 SSE 工具逐字推送答案实现上篇提到的 ChatGPT 式打字机效果最终持久化对话记录到 MongoDB。# processor/search_model/nodes/answer_output_node.py class AnswerOutputNode(BaseNode): name answer_output def process(self, state): # 构建最终问答提示词融合所有精排上下文 prompt self._build_prompt(state) is_stream state.get(is_stream, False) task_id state.get(task_id) if is_stream: # 流式生成逐字推送SSE事件 llm get_llm_client() full_answer for chunk in llm.stream(prompt): delta chunk.content if delta: full_answer delta # 推送增量文本前端实时渲染 push_sse_event(task_id, SSEEvent.DELTA, {delta: delta}) # 推送结束事件同步完整答案与配图 push_sse_event(task_id, SSEEvent.FINAL, {answer: full_answer}) else: # 非流式一次性返回完整答案 response get_llm_client().invoke(prompt) full_answer response.content state[answer] full_answer # 持久化对话记录实现会话记忆 save_message(state[session_id], assistant, full_answer, ...) return state六、 基础设施层核心工具类深度封装本项目将所有外部依赖、通用能力统一封装在 utils 层业务节点无需关注底层实现实现高内聚、低耦合同时统一异常处理、参数配置提升项目稳定性。6.1 BGE-M3 双向量生成工具适配 BGE-M3 模型特性同时生成稠密向量稀疏向量适配 Milvus 混合检索模式兼顾语义匹配与关键词匹配能力。def generate_vector_data(bge_m3_client, documents): embedding_result bge_m3_client.encode_documents(documents) final_sparse [] # 解析CSR格式稀疏矩阵适配Milvus入库格式 for i in range(len(documents)): csr embedding_result[sparse] start, end csr.indptr[i], csr.indptr[i1] token_ids csr.indices[start:end].tolist() weights csr.data[start:end].tolist() final_sparse.append(dict(zip(token_ids, weights))) return { dense: [d.tolist() for d in embedding_result[dense]], sparse: final_sparse }6.2 Milvus 混合检索封装封装稠密、稀疏向量检索请求通过加权融合算法合并结果统一检索入参与出参格式适配全项目检索场景。def create_vector_search_request(dense_vector, sparse_vector, exprNone, limit5): # 稠密向量语义检索请求 dense_req AnnSearchRequest( data[dense_vector], anns_fielddense_vector, param{metric_type: COSINE}, exprexpr, limitlimit, ) # 稀疏向量关键词检索请求 sparse_req AnnSearchRequest( data[sparse_vector], anns_fieldsparse_vector, param{metric_type: IP}, exprexpr, limitlimit, ) return [dense_req, sparse_req] def execute_bybrid_search_query(milvus_client, search_request, collection_name, reranker_weight(0.5, 0.5), limit5, output_fieldsNone): # 加权融合双向量结果 reranker WeightedRanker(reranker_weight[0], reranker_weight[1], norm_scoreTrue) return milvus_client.hybrid_search( collection_namecollection_name, reqssearch_request, rankerreranker, limitlimit, output_fieldsoutput_fields, )6.3 SSE 流式推送工具基于队列实现跨线程消息通信解决 FastAPI 主线程与 LangGraph 工作线程的数据同步问题稳定支撑前端流式对话、进度推送。_task_stream {} def push_sse_event(task_id, event, data): # 向任务专属队列推送消息 q _task_stream.get(task_id) if q: q.put({event: event, data: data}) async def sse_generate(task_id, request): q _task_stream.get(task_id) if not q: return # 初始化连接推送就绪状态 yield _sse_pack(SSEEvent.READY, {}) loop asyncio.get_running_loop() # 循环监听队列实时推送消息 while True: if await request.is_disconnected(): break try: msg await loop.run_in_executor(None, q.get, True, 1.0) yield _sse_pack(msg[event], msg[data]) except queue.Empty: continue # 连接关闭清理任务队列释放资源 _task_stream.pop(task_id, None)七、 本篇总结与终章预告本篇核心总结本篇作为系列中篇深度拆解了旅游 RAG 系统的后端核心大脑打通了前后端联动的核心逻辑完整落地两大 LangGraph 流水线知识库导入流水线实现文件校验、MinerU 高精度转换、多模态图片摘要、语义切片、实体抽取、双向量入库全流程解决旅游文档结构化入库难题。智能问答查询流水线实现问题重写、三路并行检索、RRF 结果融合、重排序断崖截断、流式答案生成构建高精度、高时效性的旅游问答能力。同时详解了前后端状态联动机制、国内镜像加速、混合检索、多路结果融合等核心落地技巧补充大量新手易踩坑的细节优化所有代码均可直接商用部署。终章预告第三篇·部署与优化总结本系列最后一篇终章将聚焦项目落地部署、全链路调优、生产级优化完成整套 RAG 系统的闭环落地核心内容如下Docker Compose 一键部署MinerU、Milvus、MongoDB、MinIO 全环境快速搭建国内模型加速落地HF-Mirror 镜像完整配置解决模型下载失败、缓慢问题全链路联调测试双端口服务联动、大文件入库压测、问答精度实测生产级优化升级BackgroundTasks 替换为 Celery 异步队列、内存状态替换为 Redis、Nginx 反向代理配置系列项目整体复盘架构优劣分析、常见报错解决方案、二次开发拓展指南。大家在部署运行过程中遇到 MinerU 转换失败、Milvus 连接超时、SSE 流式中断、模型下载缓慢等问题欢迎评论区留言交流终章我们完成整套项目的收官落地