基于 MLflow 与 LlamaIndex Workflow 构建混合检索 RAG 应用从构建、记录到评估与追踪的完整实战【免费下载链接】mlflowThe open source AI engineering platform for agents, LLMs, and ML models. MLflow enables teams of all sizes to debug, evaluate, monitor, and optimize production-quality AI applications while controlling costs and managing access to models and data.项目地址: https://gitcode.com/GitHub_Trending/ml/mlflow本文是一篇围绕仓库中 examples/llama_index/workflow 示例目录展开的实战指南核心场景是使用 LlamaIndex 的事件驱动 Workflow 编排框架搭建一个同时融合向量检索Vector Search、BM25 关键词检索与 Web 搜索的混合检索 RAG 问答系统再借助 MLflow 的模型记录Model-from-code、Experiment 跟踪、Evaluate 评估与Tracing 追踪能力实现从构建、实验对比到质量诊断的完整闭环。阅读本文后你将掌握如何用代码定义可配置的 RAG Workflow、如何用一条命令将其记录为可复现的 MLflow 模型、如何对多种检索策略做量化评估以及如何通过 Trace UI 定位回答质量问题的根因。为什么要做混合检索单一检索方式的局限Retrieval-Augmented GenerationRAG通过给 LLM 注入外部知识来提升回答质量但检索环节常常成为瓶颈基于 embedding 的向量检索未必总能命中语义最相关的内容例如专有名词、缩写、精确术语而纯关键词检索又缺乏语义理解。业界虽有大量检索增强技巧却不存在放之四海皆准的单一方案。因此一个务实策略是把多种检索方法并行组合向量检索负责语义召回BM25 负责精确关键词命中Web 搜索负责补充知识库之外的时效性信息随后将多路结果合并、去重并重排过滤掉无关内容后再交给 LLM 作答。本示例正是按这一思路设计的。LlamaIndex Workflow事件驱动的编排基础在深入代码前先理解 LlamaIndex Workflow 的三大核心抽象来源教程笔记 与 workflow.pyStep步骤执行单元代表工作流中的一个具体动作以step装饰的异步方法实现Event事件触发 Step 的信号在 Step 之间传递数据、控制流转方向Workflow工作流把二者连接成一个 Python 类每个 Step 是类的成员方法显式声明输入与输出事件。这种事件驱动设计天然适合并行/异步执行多个检索 Step 可以同时触发、互不阻塞再通过上下文对象统一汇聚结果这为处理耗时检索任务与生产级扩展性提供了基础。环境准备与依赖安装示例目录包含完整的 Workflow 定义workflow/子目录、手把手教学笔记 Tutorial.ipynb 以及样本数据集mlflow_qa_dataset.csv、urls.txt。克隆仓库并安装依赖按照 README.md 中的说明克隆仓库后进入示例目录并运行安装脚本git clone https://github.com/mlflow/mlflow.gitcd mlflow/examples/llama_index/workflow chmod x install.sh ./install.sh安装完成后在 Poetry 环境内启动 Jupyter Notebookpoetry run jupyter notebook安装脚本做了什么install.sh 的核心逻辑如下将$HOME/.local/bin加入 PATH若未检测到 Poetry 则通过官方安装脚本自动安装检查 Docker由于向量库 Qdrant 通常以 Docker 方式运行脚本会检测 Docker 是否可用并给出提示若你不需要 Qdrant例如只跑 BM25/Web 检索可加--no-qdrant标志跳过该检查通过poetry run pip install jupyter安装 Jupyter Notebook执行poetry install依据 pyproject.toml 安装全部依赖。依赖清单解读pyproject.toml 声明了本示例的完整依赖栈Python 版本要求3.10,3.13mlflow 2.17.0提供实验跟踪、模型记录、评估与追踪能力llama-index 0.11.0Workflow 编排框架核心llama-index-postprocessor-rankgpt-rerank基于 RankGPT 的重排后处理器llama-index-readers-webSimpleWebPageReader用于加载网页文档llama-index-retrievers-bm25BM25 关键词检索器llama-index-tools-tavily-researchTavily Web 搜索工具LLM 场景优化的搜索 APIllama-index-utils-workflowWorkflow 工具集llama-index-vector-stores-qdrantQdrant 向量存储集成。起步创建实验、配置 LLM 与 Embedding创建 MLflow ExperimentExperiment是 MLflow 组织模型开发过程的最小单元记录模型定义、配置、参数、依赖版本等信息。在笔记中创建新实验import mlflow mlflow.set_experiment(LlamaIndex Workflow RAG)配置 LLM 与 EmbeddingLlamaIndex 通过全局Settings对象统一管理 LLM 与 Embedding后续所有 LlamaIndex 组件都会使用这里的配置。MLflow 在记录模型时会自动把Settings配置写入实验从而保证跨环境的可复现性。方案一OpenAI默认LlamaIndex 默认使用 OpenAI API。示例代码截至 2024 年 10 月默认模型为gpt-3.5-turbo与text-embeddings-ada-002推荐换用更新、更高效的模型以获得更好效果与更低成本import getpass import os os.environ[OPENAI_API_KEY] getpass.getpass(Enter your OpenAI API key)from llama_index.core import Settings from llama_index.embeddings.openai import OpenAIEmbedding from llama_index.llms.openai import OpenAI Settings.embed_model OpenAIEmbedding(modeltext-embedding-3-small) Settings.llm OpenAI(modelgpt-4o-mini)方案二其他托管模型以 Databricks 托管的 Llama3.1 70B 为例安装对应提供方的集成包如llama-index-llms-databricks按集成文档设置所需环境变量如DATABRICKS_SERVING_ENDPOINT与DATABRICKS_TOKEN实例化 LLM 与 Embedding 并写入Settingsfrom llama_index.core import Settings from llama_index.embeddings.databricks import DatabricksEmbedding from llama_index.llms.databricks import Databricks Settings.embed_model DatabricksEmbedding(modeldatabricks-gte-large-en) Settings.llm Databricks(modeldatabricks-meta-llama-3-1-70b-instruct)方案三本地模型LlamaIndex 同样支持本地部署的 LLM按 LlamaIndex 本地模型入门教程配置即可。配置 Web 搜索 API本示例的 Web 检索使用Tavily AI——为 LLM 应用优化的搜索 API与 LlamaIndex 原生集成。访问其官网申请免费额度后设置环境变量也可换成 LlamaIndex 支持的其他搜索引擎如 Google Search Toolimport getpass import os os.environ[TAVILY_AI_API_KEY] getpass.getpass(Enter your Tavily AI APi Key)构建检索索引向量索引与 BM25 索引加载文档urls.txt 中存放了一批 MLflow 官方文档页面地址通过SimpleWebPageReader加载为 LlamaIndex 文档对象from llama_index.readers.web import SimpleWebPageReader with open(data/urls.txt) as file: urls [line.strip() for line in file if line.strip()] documents SimpleWebPageReader(html_to_textTrue).load_data(urls)向量索引Qdrant将文档写入向量数据库。教程选用Qdrant自托管免费先用 Docker 启动服务$ docker pull qdrant/qdrant $ docker run -p 6333:6333 -p 6334:6334 \ -v $(pwd)/.qdrant_storage:/qdrant/storage:z \ qdrant/qdrant然后创建连接 Qdrant 的索引对象并摄入文档import qdrant_client from llama_index.vector_stores.qdrant import QdrantVectorStore client qdrant_client.QdrantClient(hostlocalhost, port6333) vector_store QdrantVectorStore(clientclient, collection_namemlflow_doc) from llama_index.core import StorageContext, VectorStoreIndex storage_context StorageContext.from_defaults(vector_storevector_store) index VectorStoreIndex.from_documents(documentsdocuments, storage_contextstorage_context)当然也可以换成 FAISS、Chroma、Databricks Vector Search 等 LlamaIndex 支持的任意向量库——若更换需要按对应文档调整 workflow.py 中的向量存储接入代码。BM25 关键词索引BM25 检索基于本地节点文件先对文档分块chunk_size512后构建检索器并持久化到.bm25_retriever目录供工作流运行时加载from llama_index.core.node_parser import SentenceSplitter from llama_index.retrievers.bm25 import BM25Retriever splitter SentenceSplitter(chunk_size512) nodes splitter.get_nodes_from_documents(documents) bm25_retriever BM25Retriever.from_defaults(nodesnodes) bm25_retriever.persist(.bm25_retriever)深入工作流实现events、prompts 与 workflow 类workflow/目录包含三份核心代码events.py事件定义、prompts.py提示词模板与 workflow.py主工作流类。事件定义数据如何在步骤间流转events.py 中的事件都是携带字段的 Pydantic 模型。例如VectorSearchRetrieveEvent携带用户查询触发向量检索步骤class VectorSearchRetrieveEvent(Event): Event for triggering VectorStore index retrieval step. query: str完整的 7 种事件及其职责事件携带字段职责VectorSearchRetrieveEventquery触发向量库索引检索步骤BM25RetrieveEventquery触发 BM25 检索步骤TransformQueryEventquery将用户问题改写为搜索友好查询WebsearchEventsearch_query触发 Web 搜索工具步骤RetrievalResultEventnodes,retriever把各路检索结果含来源标识送回汇聚步骤RerankEventnodes把合并后的节点送往重排步骤QueryEventcontext触发最终问答步骤注意RetrievalResultEvent.retriever的类型被限定为Literal[vector_search, bm25, web_search]与工作流支持的三类检索器一一对应。提示词模板prompts.py 定义了两处 LLM 调用提示TRANSFORM_QUERY_TEMPLATE把用户原始问题提炼成更适合搜索引擎的查询串只输出优化后的查询FINAL_QUERY_TEMPLATE要求 LLM只依据给定的上下文作答、不得使用先验知识只输出答案。工作流类可配置的混合检索编排workflow.py 定义了HybridRAGWorkflow(Workflow)。构造函数通过retrievers参数声明要启用的检索方式支持{vector_search, bm25, web_search}的任意子集class HybridRAGWorkflow(Workflow): VALID_RETRIEVERS {vector_search, bm25, web_search} def __init__(self, retrieversNone, **kwargs): super().__init__(**kwargs) self.llm Settings.llm self.retrievers retrievers or [] if invalid_retrievers : set(self.retrievers) - self.VALID_RETRIEVERS: raise ValueError(fInvalid retrievers specified: {invalid_retrievers}) self._use_vs_retriever vector_search in self.retrievers self._use_bm25_retriever bm25 in self.retrievers self._use_web_search web_search in self.retrievers if self._use_vs_retriever: qd_client qdrant_client.QdrantClient(host_QDRANT_HOST, port_QDRANT_PORT) vector_store QdrantVectorStore(clientqd_client, collection_name_QDRANT_COLLECTION_NAME) index VectorStoreIndex.from_vector_store(vector_storevector_store) self.vs_retriever index.as_retriever() if self._use_bm25_retriever: self.bm25_retriever BM25Retriever.from_persist_dir(_BM25_PERSIST_DIR) if self._use_web_search: self.tavily_tool TavilyToolSpec(api_keyos.environ.get(TAVILY_AI_API_KEY))文件头部定义了外部依赖的环境变量均有默认值便于本地快速启动QDRANT_HOST默认localhostQDRANT_PORT默认6333QDRANT_COLLECTION_NAME默认mlflow_doc动态决定检索器是本设计的精髓只需改变retrievers列表就能实验不同检索组合而不必为几乎相同的模型代码做多份复制。第一步路由检索route_retrieval工作流以接收StartEvent的route_retrieval步骤为入口它把查询写入上下文然后根据配置并行派发事件——这是事件驱动框架实现并行异步执行的关键# If no retriever is specified, proceed directly to the final query step with an empty context if len(self.retrievers) 0: return QueryEvent(context) # Trigger the retrieval steps based on the configuration if self._use_vs_retriever: ctx.send_event(VectorSearchRetrieveEvent(queryquery)) if self._use_bm25_retriever: ctx.send_event(BM25RetrieveEvent(queryquery)) if self._use_web_search: ctx.send_event(TransformQueryEvent(queryquery))当retrievers为空时工作流退化为仅凭 LLM 先验知识直接作答这正好可以作为混合检索的对照基线。三类检索步骤向量检索与 BM25 检索都很直接——调用各自的retrieve()并包装成RetrievalResultEventstep async def query_vector_store(self, ev: VectorSearchRetrieveEvent) - RetrievalResultEvent: nodes self.vs_retriever.retrieve(ev.query) return RetrievalResultEvent(nodesnodes, retrievervector_search) step async def query_bm25(self, ev: BM25RetrieveEvent) - RetrievalResultEvent: nodes self.bm25_retriever.retrieve(ev.query) return RetrievalResultEvent(nodesnodes, retrieverbm25)Web 检索则多一步先用 LLM 把原始问题改写为适合搜索引擎的查询串transform_query再调用 Tavily 工具执行搜索query_web_searchmax_results5step async def transform_query(self, ev: TransformQueryEvent) - WebsearchEvent: prompt TRANSFORM_QUERY_TEMPLATE.format(queryev.query) transformed_query self.llm.complete(prompt).text return WebsearchEvent(search_querytransformed_query)汇聚与重排gather_retrieval_results/rerankgather_retrieval_results用ctx.collect_events()异步轮询各检索步骤的结果——在工作流框架下只要还有检索器未返回collect_events就返回None步骤等待下一轮轮询全部到齐后再决定走哪条路results ctx.collect_events(ev, [RetrievalResultEvent] * len(self.retrievers))只有一个检索器跳过重排直接把节点文本拼接为上下文进入最终问答多个检索器合并所有节点并给每个节点打上来源标记node.metadata[retriever]交给rerank步骤。多路结果直接拼接会造成上下文过大、且混入无关/重复内容。由于 Web 搜索结果没有相似度分数常见的分数排序方案行不通因此这里改用LLM 重排借助 RankGPT 集成的RankGPTRerank按查询相关性排序并取前 5reranker RankGPTRerank(llmself.llm, top_n5) reranked_nodes reranker.postprocess_nodes(ev.nodes, query_strquery) reranked_context \n.join(node.text for node in reranked_nodes)最终问答query_result把重排后的上下文与用户查询填入FINAL_QUERY_TEMPLATE调用 LLM 生成答案以StopEvent(result...)结束整个工作流step async def query_result(self, ctx: Context, ev: QueryEvent) - StopEvent: query await ctx.get(query) prompt FINAL_QUERY_TEMPLATE.format(contextev.context, queryquery) response self.llm.complete(prompt).text return StopEvent(resultresponse)实例化并运行以向量检索 BM25组合为例timeout60秒在 Jupyter 中直接异步运行from workflow.workflow import HybridRAGWorkflow workflow HybridRAGWorkflow(retrievers[vector_search, bm25], timeout60) response await workflow.run(queryWhy use MLflow with LlamaIndex?) print(response)用 Model-from-code 把工作流记录进 MLflow运行评估前先把工作流作为模型记录到 MLflow 中。本示例采用Model-from-code方式模型以独立 Python 脚本的形式记录代码是模型定义的唯一事实来源规避了 pickle 等序列化方式的稳定性风险再结合 MLflow 的依赖环境冻结能力实现可靠的模型持久化。模型入口脚本workflow/model.py 是记录模型的入口它通过mlflow.models.ModelConfig()单例读取记录时传入的model_config从而用同一份代码实例化不同配置的工作流无需重复模型定义from workflow.workflow import HybridRAGWorkflow import mlflow # Get model config from ModelConfig singleton model_config mlflow.models.ModelConfig() retrievers model_config.get(retrievers) # Create the workflow instance. workflow HybridRAGWorkflow(retrieversretrievers, timeout300) # Set the model instance logging. This is mandatory for using model-from-code logging method. mlflow.models.set_model(workflow)其中mlflow.models.set_model(workflow)是 Model-from-code 方式记录模型的必选项。用 mlflow.llama_index.log_model 记录多套配置mlflow.llama_index.log_model 支持记录 Index、Engine、Workflow 对象或指向包含上述类型定义脚本的路径字符串配合model_config参数即可为同一脚本注入不同配置。此处为 4 种检索策略各开一个 Run 并记录模型# 1. No retrievers (prior knowledge in LLM). # 2. Vector search retrieval only. # 3. Vector search and keyword search (BM25) # 4. All retrieval methods including web search. run_name_to_retrievers { none: [], vs: [vector_search], vs bm25: [vector_search, bm25], vs bm25 web: [vector_search, bm25, web_search], } models [] for run_name, retrievers in run_name_to_retrievers.items(): with mlflow.start_run(run_namerun_name): model_info mlflow.llama_index.log_model( # Specify the model Python script. llama_index_modelworkflow/model.py, # Specify retrievers to use. model_config{retrievers: retrievers}, # Define dependency files to save along with the model code_paths[workflow], # Subdirectory to save artifacts (not important) namemodel, ) models.append(model_info)关键参数说明以 model.py 的签名与文档为准llama_index_model模型对象或模型脚本路径model_config记录时保存的配置Model-from-code 方式下通过ModelConfig对象在模型代码内读取实现配置与代码解耦code_paths随模型一起保存的依赖代码目录本示例传入[workflow]以确保workflow包可被加载nameartifacts 保存的子目录名。记录完成后打开 MLflow UI可以看到 4 个 Run 以不同retrievers参数值被记录下来点击 Run 名并进入 Artifacts 页签可查看模型文件、依赖版本与Settings配置等元数据。值得留意的是MLflow 记录Settings时会刻意跳过 API Key避免密钥泄漏与不可序列化的函数对象。用 mlflow.evaluate 量化评估检索策略启用 LlamaIndex Tracing在评估前先开启MLflow Tracing一行命令即可让 MLflow 自动追踪每一次 LlamaIndex 执行其效果下一节详述mlflow.llama_index.autolog()从源码看mlflow/llama_index/autolog.py 目前仅支持 tracing 自动记录log_tracesTrue时安装 LlamaIndex tracerdisable/silent分别用于禁用与静默模式。加载评估数据集示例仓库自带含30 组问答对的评估数据 mlflow_qa_dataset.csv字段为query与ground_truth问题覆盖 MLflow 的各类用法例如如何用 Amazon S3 存储 MLflow 模型 artifacts及其标准答案import pandas as pd eval_df pd.read_csv(data/mlflow_qa_dataset.csv) display(eval_df.head(3))执行评估mlflow.evaluate()需要三要素数据集、已记录的模型、要计算的指标。本示例对每种配置分别评估采用两项指标Latencylatency()单条查询执行整个工作流所耗时间Answer Correctnessanswer_correctness(openai:/gpt-4o-mini)基于ground_truth由 OpenAI GPT-4o 模型按 1–5 分打分衡量答案正确性。from mlflow.metrics import latency from mlflow.metrics.genai import answer_correctness for model_info in models: with mlflow.start_run(run_idmodel_info.run_id): result mlflow.evaluate( # Pass the URI of the logged model above modelmodel_info.model_uri, dataeval_df, # Specify the column for ground truth answers. targetsground_truth, # Define the metrics to compute. extra_metrics[ latency(), answer_correctness(openai:/gpt-4o-mini), ], # The answer_correctness metric requires inputs column to be # present in the dataset. We have query instead so need to # specify the mapping in evaluator_config parameter. evaluator_config{col_mapping: {inputs: query}}, )注意answer_correctness需要数据集中存在inputs列而本数据集列名是query因此通过evaluator_config{col_mapping: {inputs: query}}完成列映射。这两项指标仅为演示你完全可以补充 toxicity、faithfulness 等指标或自定义指标。阅读评估结果评估耗时几分钟完成后在 MLflow UI 的 Experiment 页面点击 Run 列表上方的图表图标即可看到对比结果第一行是 answer correctness 柱状图第二行是 latency 结果。本例中最优组合是Vector Search BM25有趣的是加入 Web 搜索后不仅延迟显著上升answer correctness 反而下降——例如在回答如何启动 Model Registry时开启 Web 搜索的模型给出了关于模型部署的离题答案而vs bm25组合回答正确。评估结果会因模型配置与随机性略有差异。用 MLflow Tracing 定位回答质量问题的根因由于唯一变化的是检索策略问题大概率出在检索环节但仅看最终答案很难判断各路检索器各自返回了什么。此时MLflow Tracing派上用场它完整记录工作流执行期间每一步的输入、输出、元数据与延迟并与 LlamaIndex 深度集成。在 Experiment 页面切换到Traces 页签找到请求列为该问题、Run 名为 vs bm25 web 的记录点击请求 ID 即可打开 Trace UI 查看各步骤详情本案例中通过检查rerank重排步骤就锁定了问题Web 检索器返回了与模型服务相关的无关上下文而重排器错误地把它排为最相关。有了这一洞察就可以针对性地改进——例如优化重排器对 MLflow 主题的理解、提高 Web 检索精度甚至直接移除 Web 检索器。小结与延伸本示例完整展示了 LlamaIndex 与 MLflow 组合如何提升 RAG 工作流的开发效率与可观测性Experiment Tracking以 Run 维度组织并记录不同工作流配置保证可复现性支持跨 Run 性能追踪MLflow Evaluate对多种检索策略无检索、仅向量、向量BM25、全量混合用 latency 与 answer correctness 做量化对比MLflow UI直观可视化不同检索策略对准确率与延迟的影响帮助选出最优配置MLflow Tracing与 LlamaIndex 集成提供工作流每一步的细粒度可观测性用于诊断诸如重排错误之类的质量问题。由此你获得了一整套构建 → 记录 → 评估 → 诊断 → 优化的 RAG 开发闭环。进一步探索的仓库入口完整教学笔记Tutorial.ipynb工作流核心实现workflow.py、events.py、prompts.py、model.py环境与依赖install.sh、pyproject.toml数据与示例图mlflow_qa_dataset.csv、urls.txtMLflow 集成源码mlflow/llama_index/model.py、mlflow/llama_index/autolog.py。【免费下载链接】mlflowThe open source AI engineering platform for agents, LLMs, and ML models. MLflow enables teams of all sizes to debug, evaluate, monitor, and optimize production-quality AI applications while controlling costs and managing access to models and data.项目地址: https://gitcode.com/GitHub_Trending/ml/mlflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考