使用 ZenML 构建 RAG 流水线从数据摄取、向量检索到推理的完整实战指南【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenmlRetrieval-Augmented Generation检索增强生成RAG通过把「检索」与「生成」两种能力结合解决了大语言模型LLM幻觉频发、上下文窗口有限、运行成本高昂等痛点是入门 LLM 应用最务实的起点。本指南以 ZenML 为底座围绕 RAG Pipelines with ZenML 这一官方教程的核心脉络完整讲解 RAG 的原理、最小实现、数据摄取与预处理、Embedding 生成、向量数据库索引管理以及基于 ZenML 的 Artifact 追踪与推理读完后你将掌握一套可复现、可扩展、可观测的 RAG 流水线搭建方案。为什么需要 RAG它解决什么问题LLM 虽然能针对五花八门的提示词生成接近人类的回答但天然存在三个短板容易产生错误或不当的回答尤其是当输入提示词含糊不清或具有误导性时上下文与生成长度受限虽然个别模型如 Google Gemini 1.5 Pro可稳定处理 100 万 token但绝大多数开源模型远达不到这个量级运行成本高昂大规模语料直接喂给生成模型既不经济也不现实。RAG 于 2020 年由 FacebookMeta的研究者首次提出其核心思想是用检索机制retriever从大规模语料库中找出与问题最相关的文档再用生成器generator基于这些文档生成回答。这种「检索 生成」的组合既发挥 LLM 的生成优势又规避其局限缓解上下文限制检索步骤确保生成结果有据可依降低产生错误或不当回答的概率缓解 token 限制生成器只需要聚焦于一小批相关文档而不必处理整个语料库更具成本效益相比纯生成式方案RAG 把生成器资源集中到更少的相关文档上尤其适合大规模语料库或资源受限的部署环境。什么时候该选择 RAGRAG 适合需要生成依赖上下文理解的长文本回答、且手上拥有大规模相关语料库的场景典型任务包括问答question answering、摘要summarization和对话生成dialogue generation。对初学者而言RAG 是踏入 LLM 世界的第一站——它不需要像微调那样消耗大量数据与算力却能让人快速理解整套工作流的运作方式同时能在「享受 LLM 能力」与「规避当前模型局限」之间取得平衡。RAG 在 ZenML 生态中的定位在 ZenML 中你可以把 RAG 流水线组织为「职责清晰、可反复运行」的步骤集合覆盖数据摄取、索引库管理、RAG 相关 Artifact 追踪三大能力。更进一步ZenML 还支持从简单的 RAG 起步平滑演进到微调 Embedding、对检索结果做重排序reranking、甚至微调 LLM 本身等更复杂的方案。ZenML 能够追踪 RAG 流水线产生的全部 Artifact——从超参数、模型权重到元数据与性能指标以及链chain、Agent、Tokenizer、向量存储vector store等 RAG/LLM 专属对象这些对象都可以在 Model Control Plane 中统一管理并可视化于 ZenML Pro 的 Dashboard 上。ZenML 为 RAG 流水线带来的核心价值可总结为可复现性Reproducibility重新运行流水线即可用新文档更新索引库、或调整分块参数历史版本的 Artifact 会被保留方便对比不同版本流水线的性能可扩展性Scalability把流水线部署到云上、配合更可扩展的向量库即可轻松处理更大规模的语料Artifact 追踪与元数据关联流水线产出的 Artifact 可绑定额外元数据并在 Dashboard 中可视化便于监控流水线表现、排查问题可维护性Maintainability清晰、模块化的流水线结构让新增步骤、调整参数、尝试不同配置变得简单协作性Collaboration流水线可与团队共享通过 Dashboard 分享洞察与发现方便协同开发。先看一个 85 行代码的最小 RAG 实现在深入 ZenML 之前先用一段约 85 行的极简代码建立对 RAG 全流程的直观认识。这段代码做了四件事加载一个虚构的「ZenML World」科幻数据集作为语料库处理文本——切分文本并做「分词」按单词切分输入一条查询从语料中找到最相关的文本块用 OpenAI 的 GPT-3.5 模型基于相关文本块回答问题。import os import re import string from openai import OpenAI def preprocess_text(text): text text.lower() text text.translate(str.maketrans(, , string.punctuation)) text re.sub(r\s, , text).strip() return text def tokenize(text): return preprocess_text(text).split() def retrieve_relevant_chunks(query, corpus, top_n2): query_tokens set(tokenize(query)) similarities [] for chunk in corpus: chunk_tokens set(tokenize(chunk)) similarity len(query_tokens.intersection(chunk_tokens)) / len( query_tokens.union(chunk_tokens) ) similarities.append((chunk, similarity)) similarities.sort(keylambda x: x[1], reverseTrue) return [chunk for chunk, _ in similarities[:top_n]] def answer_question(query, corpus, top_n2): relevant_chunks retrieve_relevant_chunks(query, corpus, top_n) if not relevant_chunks: return I dont have enough information to answer the question. context \n.join(relevant_chunks) client OpenAI(api_keyos.environ.get(OPENAI_API_KEY)) chat_completion client.chat.completions.create( messages[ { role: system, content: fBased on the provided context, answer the following question: {query}\n\nContext:\n{context}, }, { role: user, content: query, }, ], modelgpt-3.5-turbo, ) return chat_completion.choices[0].message.content.strip() # Sci-fi themed corpus about ZenML World corpus [ The luminescent forests of ZenML World are inhabited by glowing Zenbots that emit a soft, pulsating light as they roam the enchanted landscape., In the neon skies of ZenML World, Cosmic Butterflies flutter gracefully, their iridescent wings leaving trails of stardust in their wake., Telepathic Treants, ancient sentient trees, communicate through the quantum neural network that spans the entire surface of ZenML World, sharing wisdom and knowledge., Deep within the melodic caverns of ZenML World, Fractal Fungi emit pulsating tones that resonate through the crystalline structures, creating a symphony of otherworldly sounds., Near the ethereal waterfalls of ZenML World, Holographic Hummingbirds hover effortlessly, their translucent wings refracting the prismatic light into mesmerizing patterns., Gravitational Geckos, masters of anti-gravity, traverse the inverted cliffs of ZenML World, defying the laws of physics with their extraordinary abilities., Plasma Phoenixes, majestic creatures of pure energy, soar above the chromatic canyons of ZenML World, their fiery trails painting the sky in a dazzling display of colors., Along the prismatic shores of ZenML World, Crystalline Crabs scuttle and burrow, their transparent exoskeletons refracting the light into a kaleidoscope of hues., ] corpus [preprocess_text(sentence) for sentence in corpus] question1 What are Plasma Phoenixes? answer1 answer_question(question1, corpus) print(fQuestion: {question1}) print(fAnswer: {answer1}) question2 ( What kinds of creatures live on the prismatic shores of ZenML World? ) answer2 answer_question(question2, corpus) print(fQuestion: {question2}) print(fAnswer: {answer2}) irrelevant_question_3 What is the capital of Panglossia? answer3 answer_question(irrelevant_question_3, corpus) print(fQuestion: {irrelevant_question_3}) print(fAnswer: {answer3})运行后输出如下Question: What are Plasma Phoenixes? Answer: Plasma Phoenixes are majestic creatures made of pure energy that soar above the chromatic canyons of Zenml World. They leave fiery trails behind them, painting the sky with dazzling displays of colors. Question: What kinds of creatures live on the prismatic shores of ZenML World? Answer: On the prismatic shores of ZenML World, you can find crystalline crabs scuttling and burrowing with their transparent exoskeletons, which refract light into a kaleidoscope of hues. Question: What is the capital of Panglossia? Answer: The capital of Panglossia is not mentioned in the provided context.值得注意的关键设计这里对查询与文本块的相似度度量使用的是Jaccard 相似系数即「两个集合交集大小 ÷ 并集大小」统计查询与文本块共现的单词数占两者全部去重单词数的比例。这种方式非常朴素且低效真正的生产级方案会用 Embedding 等更先进的技术当查询与语料完全不相关时如第三个问题模型会如实回答「语料中没有相关信息」这正体现了 RAG「让回答有据可依」的初衷该实现只是教学示范接下来的完整 ZenML 方案会在检索质量、可扩展性与工程化程度上全面超越它。完整的 85 行实现见 rag-85-loc.md。第一阶段数据摄取与预处理RAG 流水线的第一步是摄取用于训练与评估检索器、生成器的数据——既包括大规模文档语料也包括相关的元数据或标注信息。官方教程以抓取 ZenML 自己的文档站点为例演示了整个摄取链路。教程中的完整代码可在官方llm-complete-guide项目仓库中找到。用 ZenML Step 抓取 URL 列表借助简单的辅助工具函数如get_all_pages可以轻松实现一个抓取 ZenML 文档相关 URL 的 ZenML stepfrom typing import List from typing import Annotated from zenml import log_artifact_metadata, step from steps.url_scraping_utils import get_all_pages step def url_scraper( docs_url: str https://docs.zenml.io, repo_url: str https://github.com/zenml-io/zenml, website_url: str https://zenml.io, ) - Annotated[List[str], urls]: Generates a list of relevant URLs to scrape. docs_urls get_all_pages(docs_url) log_artifact_metadata( metadata{ count: len(docs_urls), }, ) return docs_urlsget_all_pages函数会爬取文档网站并返回去重后的 URL 集合。教程特意把抓取范围限制在最近发布的文档上避免把旧语法和旧信息与新文档混在一起确保只摄取最新、最相关的信息。同时该 step 用log_artifact_metadata把 URL 数量作为元数据记录下来这在 ZenML Dashboard 中可见为摄取的数据提供了额外可见性。你也可以在此基础上添加更复杂的逻辑比如过滤某些 URL 或记录更多元数据。log_artifact_metadata是 ZenML 提供的 Artifact 元数据记录入口其底层实现位于 src/zenml/artifacts/utils.py它可以把元数据关联到已存在的 Artifact 版本或关联到同一 step 中新创建的 Artifact 版本。值得注意的是从源码看该函数已被标记为 deprecated官方建议改用log_metadata(metadata{...}, infer_artifactTrue, ...)的新接口写法上更统一、能力更强。用 unstructured 解析网页内容拿到 URL 列表后教程使用开源库unstructured来加载并解析页面从而无需关心 HTML 结构和标记的细节直接把网页转成纯文本。这尤其重要因为在 LLM 的受限上下文环境下应尽量让文本内容保持精简from typing import List from unstructured.partition.html import partition_html from zenml import step step def web_url_loader(urls: List[str]) - List[str]: Loads documents from a list of URLs. document_texts [] for url in urls: elements partition_html(urlurl) text \n\n.join([str(el) for el in elements]) document_texts.append(text) return document_texts像unstructured这类框架在数据摄取上提供更多选择支持多种来源、文本预处理、特征提取但对本教程而言自实现的好处是可调试性更强——你能清楚地看到每一步加载了什么、如何被处理这是复杂框架难以提供的可见性。预处理确定 chunk size 与 overlap加载文档后需要把文本切分成适合送入 RAG 流水线的块chunk。当前文本是每条代表一个网页的长字符串直接送入 LLM 过长、影响响应速度因此要拆成更小的块。分块大小存在一个甜蜜点chunk 太小检索/搜索步骤难以找到足够的相关信息传递给 LLMchunk 太大LLM 处理起来吃力。import logging from typing import Annotated, List from utils.llm_utils import split_documents from zenml import ArtifactConfig, log_artifact_metadata, step logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) step(enable_cacheFalse) def preprocess_documents( documents: List[str], ) - Annotated[List[str], ArtifactConfig(namesplit_chunks)]: Preprocesses a list of documents by splitting them into chunks. try: log_artifact_metadata( artifact_namesplit_chunks, metadata{ chunk_size: 500, chunk_overlap: 50 }, ) return split_documents( documents, chunk_size500, chunk_overlap50 ) except Exception as e: logger.error(fError in preprocess_documents: {e}) raise这段代码有几点工程细节值得留意step(enable_cacheFalse)关闭该 step 的缓存确保每次运行都会重新执行分块逻辑避免因输入相同而跳过重跑ArtifactConfig(namesplit_chunks)为输出 Artifact 显式命名后续步骤和元数据记录都可通过该名称引用chunk_size500、chunk_overlap50以 500 字符为块大小、50 字符为重叠区。重叠的设计可以避免切分时遗漏跨边界的重要信息。分块参数并没有「标准答案」关键在于理解你的数据如果数据需要大段文字才能表达完整概念就应使用更大的 chunk size如果是对话式或问答式的数据则适合更小的 chunk size。教程选用 500/50 是因为处理对象是软件库的文档网页。根据业务不同还可以进一步清洗文本、剔除代码片段或确保代码不被切断、提取文本元数据等。第二阶段生成 Embedding 用于检索Embedding 是 RAG 检索机制的核心——它把数据表示在高维空间中语义相近的项在空间中彼此靠近。借助 Embedding推理时能快速找到与查询最相关的小块文本相比简单地在查询中匹配关键词Embedding 能捕捉语义与上下文对复杂查询和长文档更稳健。什么是 EmbeddingEmbedding 是数据的向量表示在高维空间中捕捉数据的语义含义与上下文。它们由机器学习模型如词向量或句向量模型生成编码方式保留了数据的内在结构与关系广泛用于文本分类、情感分析、信息检索等 NLP 任务。教程选用开源库sentence-transformers生成句向量具体模型为sentence-transformers/all-MiniLM-L12-v2输出维度 384from typing import Annotated, List import numpy as np from sentence_transformers import SentenceTransformer from structures import Document from zenml import ArtifactConfig, log_artifact_metadata, step step def generate_embeddings( split_documents: List[Document], ) - Annotated[ List[Document], ArtifactConfig(namedocuments_with_embeddings) ]: try: model SentenceTransformer(sentence-transformers/all-MiniLM-L12-v2) log_artifact_metadata( artifact_nameembeddings, metadata{ embedding_type: sentence-transformers/all-MiniLM-L12-v2, embedding_dimensionality: 384, }, ) document_texts [doc.page_content for doc in split_documents] embeddings model.encode(document_texts) for doc, embedding in zip(split_documents, embeddings): doc.embedding embedding return split_documents except Exception as e: logger.error(fError in generate_embeddings: {e}) raise模型选型要点该模型属于偏小的一侧更看重速度如果更看重检索相关性可换成维度更高的更大模型教程通过log_artifact_metadata把embedding_type与embedding_dimensionality记入元数据让 Embedding 的来源与维度在 Dashboard 中可追溯这里扩展了DocumentPydantic 模型为其增加embedding属性来存储每个文档对应的向量从而把 Embedding 与文档关联起来供后续检索使用。用 UMAP / t-SNE 可视化 Embedding 空间384 维的向量人类无法直接理解教程借助umap与scikit-learn的降维功能把 Embedding 投射到二维空间观察相似块如何按语义聚类。可视化时使用此前存入向量库的「父目录」parent section属性来着色能直观看出数据内在的语义空间结构。from matplotlib.colors import ListedColormap import matplotlib.pyplot as plt import numpy as np from sklearn.manifold import TSNE import umap from zenml.client import Client artifact Client().get_artifact_version(EMBEDDINGS_ARTIFACT_UUID_GOES_HERE) embeddings artifact.load() embeddings np.array([doc.embedding for doc in documents]) parent_sections [doc.parent_section for doc in documents] # Get unique parent sections unique_parent_sections list(set(parent_sections)) # Tol color palette tol_colors [ #4477AA, #EE6677, #228833, #CCBB44, #66CCEE, #AA3377, #BBBBBB, ] # Create a colormap with Tol colors tol_colormap ListedColormap(tol_colors) # Assign colors to each unique parent section section_colors tol_colors[: len(unique_parent_sections)] # Create a dictionary mapping parent sections to colors section_color_dict dict(zip(unique_parent_sections, section_colors)) # Dimensionality reduction using t-SNE def tsne_visualization(embeddings, parent_sections): tsne TSNE(n_components2, random_state42) embeddings_2d tsne.fit_transform(embeddings) plt.figure(figsize(8, 8)) for section in unique_parent_sections: if section in section_color_dict: mask [section ps for ps in parent_sections] plt.scatter( embeddings_2d[mask, 0], embeddings_2d[mask, 1], c[section_color_dict[section]], labelsection, ) plt.title(t-SNE Visualization) plt.legend() plt.show() # Dimensionality reduction using UMAP def umap_visualization(embeddings, parent_sections): umap_2d umap.UMAP(n_components2, random_state42) embeddings_2d umap_2d.fit_transform(embeddings) plt.figure(figsize(8, 8)) for section in unique_parent_sections: if section in section_color_dict: mask [section ps for ps in parent_sections] plt.scatter( embeddings_2d[mask, 0], embeddings_2d[mask, 1], c[section_color_dict[section]], labelsection, ) plt.title(UMAP Visualization) plt.legend() plt.show()从可视化结果可以看出语义相似的块聚集在一起据此可以识别数据中的模式与关系进而改进检索性能。UMAP 与 t-SNE 的降维算法不同、呈现有所差异建议两种都尝试以找到最适合的方案。完整示例见 embeddings-generation.md。架构设计生成与上传解耦教程特意把「生成 Embedding」与「上传到向量库」拆成两个独立步骤Embedding 以 NumPy 数组形式作为 ZenML Artifact 存储在 Artifact Store 中。这种模块化设计意味着未来更换向量数据库时只需替换上传步骤无需重新生成 Embedding既省算力又保持流水线的灵活性。第三阶段把 Embedding 存入向量数据库生成 Embedding 本身并不算耗时尤其在带 GPU 的机器上但我们不可能每次检索文档都重新生成一遍。正确做法是把 Embedding 存入向量数据库通过相似度快速检索最相关的块。教程选用PostgreSQL作为向量数据库它成熟稳定、可扩展且能高效存储与检索高维向量配合pgvector插件提供向量类型与相似度检索能力你也可以换成任何支持高维向量的数据库。连接使用轻量级psycopg2包加原生 SQL 语句。from zenml import step step def index_generator( documents: List[Document], ) - None: try: conn get_db_conn() with conn.cursor() as cur: # Install pgvector if not already installed cur.execute(CREATE EXTENSION IF NOT EXISTS vector) conn.commit() # Create the embeddings table if it doesnt exist table_create_command f CREATE TABLE IF NOT EXISTS embeddings ( id SERIAL PRIMARY KEY, content TEXT, token_count INTEGER, embedding VECTOR({EMBEDDING_DIMENSIONALITY}), filename TEXT, parent_section TEXT, url TEXT ); cur.execute(table_create_command) conn.commit() register_vector(conn) # Insert data only if it doesnt already exist for doc in documents: content doc.page_content token_count doc.token_count embedding doc.embedding.tolist() filename doc.filename parent_section doc.parent_section url doc.url cur.execute( SELECT COUNT(*) FROM embeddings WHERE content %s, (content,), ) count cur.fetchone()[0] if count 0: cur.execute( INSERT INTO embeddings (content, token_count, embedding, filename, parent_section, url) VALUES (%s, %s, %s, %s, %s, %s), ( content, token_count, embedding, filename, parent_section, url, ), ) conn.commit() cur.execute(SELECT COUNT(*) as cnt FROM embeddings;) num_records cur.fetchone()[0] logger.info(fNumber of vector records in table: {num_records}) # calculate the index parameters according to best practices num_lists max(num_records / 1000, 10) if num_records 1000000: num_lists math.sqrt(num_records) # use the cosine distance measure, which is what well later use for querying cur.execute( fCREATE INDEX IF NOT EXISTS embeddings_idx ON embeddings USING ivfflat (embedding vector_cosine_ops) WITH (lists {num_lists}); ) conn.commit() except Exception as e: logger.error(fError in index_generator: {e}) raise finally: if conn: conn.close()这段 step 实际完成六件事连接数据库若不存在则创建vector扩展——这是 PostgreSQL 启用向量数据类型的前提若不存在则创建embeddings表把 Embedding 与文档内容一并插入表中按最佳实践计算索引参数在 Embedding 列上创建索引。表结构设计要点embeddings表不仅存向量还同时存content、token_count、filename、parent_section、url等文档字段。这样做的原因有两个一是后续可以基于 Embedding 把对应的文档内容检索出来二是有利于直接在 Supabase 界面或任何数据库查看工具中排错。增量更新策略「何时更新 Embedding」取决于具体业务如果数据频繁且大幅变化可能需要每次全量重置如果变化轻微或不频繁只需把新文档与新 Embedding 追加进库。上面的代码选择的是只插入库中尚不存在的内容按content查重这是一种兼顾简单与实用的增量策略。索引参数ivfflat 与 cosine 距离教程为 Embedding 创建了ivfflat类型的索引搭配vector_cosine_ops算子这一组合非常适合基于余弦距离的相似度检索。lists列表数根据表中记录数动态计算最小 10最大为记录数的平方根——这是调优索引参数的合理起点实践中可以尝试不同取值观察对检索性能的影响。需要提示的是如果数据集很大、Embedding 数量很多CPU 上运行本 step 可能过慢此时应通过 ZenML 的 step operator 让该步骤跑在 GPU 机器上详见 step-operators 相关文档。向量库能力的官方集成佐证从 ZenML 仓库源码看官方对主流向量存储生态已有直接集成支持。例如 src/zenml/integrations/langchain/materializers/vector_store_materializer.py 中的LangchainVectorStoreMaterializer能把 LangChain 的VectorStore对象涵盖 Chroma、Pinecone、Qdrant、Weaviate 等各类向量库实现作为 ZenML Artifact 序列化与反序列化这意味着你可以把「已构建好的向量索引」本身当作可追踪、可版本化的 Artifact 来管理而不必像教程这样用裸 SQL 操作数据库。教程选择手写 SQL 是为了展示底层原理与保持流水线简单生产环境可按需选择更成熟的集成方式。追踪 RAG 关联的 Artifact贯穿整条流水线的一个核心能力是把 RAG 各阶段产出的 Artifact 与元数据关联起来。从前面各 step 可以看到三种典型用法记录 step 输出元数据如url_scraper记录 URL 数量count记录具名 Artifact 的元数据如preprocess_documents通过ArtifactConfig(namesplit_chunks)命名输出再用log_artifact_metadata(artifact_namesplit_chunks, metadata{...})记录chunk_size与chunk_overlap记录模型与数据特征如generate_embeddings记录embedding_type与embedding_dimensionality。这些元数据连同 Artifact 本身会一并出现在 ZenML Dashboard 中让你能够监控流水线表现、定位问题并对比不同版本流水线的差异。从源码实现看src/zenml/artifacts/utils.pylog_artifact_metadata最终会把元数据通过统一的log_metadata机制写入 ZenML 的元数据存储该函数当前已标记为 deprecated新代码建议直接使用log_metadata(metadata{...}, infer_artifactTrue, ...)。同时RAG 特有的对象链、Agent、Tokenizer、向量存储等也可以纳入 Model Control Plane 统一追踪。第四阶段基础 RAG 推理索引库就绪后就可以基于库中文档回答查询了。官方教程的推理阶段不再是一个 ZenML 流水线而是一个复用流水线产出的函数调用——它使用流水线的输出与组件生成回答。以命令行方式触发python run.py --rag-query how do I use a custom materializer inside my own zenml steps? i.e. how do I set it? inside the step decorator? --modelgpt4完整的推理逻辑如下def process_input_with_retrieval( input: str, model: str OPENAI_MODEL, n_items_retrieved: int 5 ) - str: delimiter # Step 1: Get documents related to the user input from database related_docs get_topn_similar_docs( get_embeddings(input), get_db_conn(), nn_items_retrieved ) # Step 2: Get completion from OpenAI API # Set system message to help set appropriate tone and context for model system_message f You are a friendly chatbot. \ You can answer questions about ZenML, its features and its use cases. \ You respond in a concise, technically credible tone. \ You ONLY use the context from the ZenML documentation to provide relevant answers. \ You do not make up answers or provide opinions that you dont have information to support. \ If you are unsure or dont know, just say so. \ # Prepare messages to pass to model # We use a delimiter to help the model understand the where the user_input # starts and ends messages [ {role: system, content: system_message}, {role: user, content: f{delimiter}{input}{delimiter}}, { role: assistant, content: fRelevant ZenML documentation: \n \n.join(doc[0] for doc in related_docs), }, ] logger.debug(CONTEXT USED\n\n, messages[2][content], \n\n) return get_completion_from_messages(messages, modelmodel)推理流程分两步先用get_topn_similar_docs从数据库取回与输入最相关的文档再调用get_completion_from_messages让 LLM 基于这些文档生成回答。系统提示词system message被精心设计为「只依据 ZenML 文档上下文作答、不编造答案、不确定时如实说明」并用分隔符帮助模型识别用户输入的起止位置。检索借助 pgvector 做余弦相似度排序get_topn_similar_docs使用查询文本的 Embedding在向量库中找出最相似的前 n 个文档def get_topn_similar_docs( query_embedding: List[float], conn: psycopg2.extensions.connection, n: int 5, include_metadata: bool False, only_urls: bool False, ) - List[Tuple]: embedding_array np.array(query_embedding) register_vector(conn) cur conn.cursor() if include_metadata: cur.execute( fSELECT content, url FROM embeddings ORDER BY embedding %s LIMIT {n}, (embedding_array,), ) elif only_urls: cur.execute( fSELECT url FROM embeddings ORDER BY embedding %s LIMIT {n}, (embedding_array,), ) else: cur.execute( fSELECT content FROM embeddings ORDER BY embedding %s LIMIT {n}, (embedding_array,), ) return cur.fetchall()这里的关键是ORDER BY embedding %s是 pgvector 提供的余弦距离算子数据库会按向量与查询向量的余弦距离升序返回记录并限制条数。把「找最相关文档」这个重活交给数据库本身完成效率极高也呼应了前文用vector_cosine_ops建立索引的设计——查询所用的距离度量与建索引时的度量保持一致。函数还支持include_metadata返回内容与 URL和only_urls仅返回 URL两种模式方便在不同场景下复用。生成用 litellm 统一多模型接口def get_completion_from_messages( messages, modelOPENAI_MODEL, temperature0.4, max_tokens1000 ): Generates a completion response from the given messages using the specified model. model MODEL_NAME_MAP.get(model, model) completion_response litellm.completion( modelmodel, messagesmessages, temperaturetemperature, max_tokensmax_tokens, ) return completion_response.choices[0].message.content生成环节使用litellm作为 LLM 统一接口这样就不必为每个想用的模型分别实现一套调用代码。LLM 领域迭代极快新模型层出不穷litellm让你能随时切换新模型而无需重写代码MODEL_NAME_MAP则提供了模型别名映射如命令行里的gpt4映射到具体模型标识。temperature0.4、max_tokens1000控制生成风格与长度上限。至此一条完整的 RAG 推理链路已经打通流水线负责把文档「摄取 → 分块 → 向量化 → 入库」推理函数则「把查询向量化 → 余弦检索 → 组装上下文 → 交给 LLM 生成回答」。完整推理代码见 basic-rag-inference-pipeline.md。从基础 RAG 走向更复杂的方案本教程搭建的是一条清晰的基线 RAG 流水线它本身已具备可复现、可扩展、可观测、可维护、可协作五大优势。在此基础上ZenML 的 LLMOps 指南还提供了向纵深演进的方向见 llmops-guide/README.md检索评估与生成评估用指标量化检索与生成环节的质量见 evaluation重排序Reranking对检索结果做二次精排提升相关性见 reranking微调 Embedding当文档规模很大、或文档与预训练 Embedding 的训练分布差异明显时通过微调提升检索性能见 finetuning-embeddings微调 LLM在需要改变模型行为风格、领域适配时对 LLM 本身进行微调见 finetuning-llms。总结围绕 RAG Pipelines with ZenML 官方指南本篇文章完整梳理了 RAG 在 ZenML 中的落地路径先以 85 行最小实现建立直觉再分阶段构建「URL 抓取 → unstructured 解析 → 分块预处理 → sentence-transformers 向量化 → pgvector 入库建索引」的数据侧流水线最后通过「余弦相似度检索 litellm 统一生成」完成推理。贯穿始终的是 ZenML 的 Artifact 与元数据追踪机制——它把 RAG 流水线的每个环节变成可复现、可对比、可观测的工程资产为后续引入重排序、Embedding 微调乃至 LLM 微调等高级方案打下了坚实基础。仓库中的相关子文档understanding-rag.md、data-ingestion.md、embeddings-generation.md、storing-embeddings-in-a-vector-database.md、basic-rag-inference-pipeline.md提供了各环节更深入的展开可直接按需查阅。【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
使用 ZenML 构建 RAG 流水线：从数据摄取、向量检索到推理的完整实战指南
使用 ZenML 构建 RAG 流水线从数据摄取、向量检索到推理的完整实战指南【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenmlRetrieval-Augmented Generation检索增强生成RAG通过把「检索」与「生成」两种能力结合解决了大语言模型LLM幻觉频发、上下文窗口有限、运行成本高昂等痛点是入门 LLM 应用最务实的起点。本指南以 ZenML 为底座围绕 RAG Pipelines with ZenML 这一官方教程的核心脉络完整讲解 RAG 的原理、最小实现、数据摄取与预处理、Embedding 生成、向量数据库索引管理以及基于 ZenML 的 Artifact 追踪与推理读完后你将掌握一套可复现、可扩展、可观测的 RAG 流水线搭建方案。为什么需要 RAG它解决什么问题LLM 虽然能针对五花八门的提示词生成接近人类的回答但天然存在三个短板容易产生错误或不当的回答尤其是当输入提示词含糊不清或具有误导性时上下文与生成长度受限虽然个别模型如 Google Gemini 1.5 Pro可稳定处理 100 万 token但绝大多数开源模型远达不到这个量级运行成本高昂大规模语料直接喂给生成模型既不经济也不现实。RAG 于 2020 年由 FacebookMeta的研究者首次提出其核心思想是用检索机制retriever从大规模语料库中找出与问题最相关的文档再用生成器generator基于这些文档生成回答。这种「检索 生成」的组合既发挥 LLM 的生成优势又规避其局限缓解上下文限制检索步骤确保生成结果有据可依降低产生错误或不当回答的概率缓解 token 限制生成器只需要聚焦于一小批相关文档而不必处理整个语料库更具成本效益相比纯生成式方案RAG 把生成器资源集中到更少的相关文档上尤其适合大规模语料库或资源受限的部署环境。什么时候该选择 RAGRAG 适合需要生成依赖上下文理解的长文本回答、且手上拥有大规模相关语料库的场景典型任务包括问答question answering、摘要summarization和对话生成dialogue generation。对初学者而言RAG 是踏入 LLM 世界的第一站——它不需要像微调那样消耗大量数据与算力却能让人快速理解整套工作流的运作方式同时能在「享受 LLM 能力」与「规避当前模型局限」之间取得平衡。RAG 在 ZenML 生态中的定位在 ZenML 中你可以把 RAG 流水线组织为「职责清晰、可反复运行」的步骤集合覆盖数据摄取、索引库管理、RAG 相关 Artifact 追踪三大能力。更进一步ZenML 还支持从简单的 RAG 起步平滑演进到微调 Embedding、对检索结果做重排序reranking、甚至微调 LLM 本身等更复杂的方案。ZenML 能够追踪 RAG 流水线产生的全部 Artifact——从超参数、模型权重到元数据与性能指标以及链chain、Agent、Tokenizer、向量存储vector store等 RAG/LLM 专属对象这些对象都可以在 Model Control Plane 中统一管理并可视化于 ZenML Pro 的 Dashboard 上。ZenML 为 RAG 流水线带来的核心价值可总结为可复现性Reproducibility重新运行流水线即可用新文档更新索引库、或调整分块参数历史版本的 Artifact 会被保留方便对比不同版本流水线的性能可扩展性Scalability把流水线部署到云上、配合更可扩展的向量库即可轻松处理更大规模的语料Artifact 追踪与元数据关联流水线产出的 Artifact 可绑定额外元数据并在 Dashboard 中可视化便于监控流水线表现、排查问题可维护性Maintainability清晰、模块化的流水线结构让新增步骤、调整参数、尝试不同配置变得简单协作性Collaboration流水线可与团队共享通过 Dashboard 分享洞察与发现方便协同开发。先看一个 85 行代码的最小 RAG 实现在深入 ZenML 之前先用一段约 85 行的极简代码建立对 RAG 全流程的直观认识。这段代码做了四件事加载一个虚构的「ZenML World」科幻数据集作为语料库处理文本——切分文本并做「分词」按单词切分输入一条查询从语料中找到最相关的文本块用 OpenAI 的 GPT-3.5 模型基于相关文本块回答问题。import os import re import string from openai import OpenAI def preprocess_text(text): text text.lower() text text.translate(str.maketrans(, , string.punctuation)) text re.sub(r\s, , text).strip() return text def tokenize(text): return preprocess_text(text).split() def retrieve_relevant_chunks(query, corpus, top_n2): query_tokens set(tokenize(query)) similarities [] for chunk in corpus: chunk_tokens set(tokenize(chunk)) similarity len(query_tokens.intersection(chunk_tokens)) / len( query_tokens.union(chunk_tokens) ) similarities.append((chunk, similarity)) similarities.sort(keylambda x: x[1], reverseTrue) return [chunk for chunk, _ in similarities[:top_n]] def answer_question(query, corpus, top_n2): relevant_chunks retrieve_relevant_chunks(query, corpus, top_n) if not relevant_chunks: return I dont have enough information to answer the question. context \n.join(relevant_chunks) client OpenAI(api_keyos.environ.get(OPENAI_API_KEY)) chat_completion client.chat.completions.create( messages[ { role: system, content: fBased on the provided context, answer the following question: {query}\n\nContext:\n{context}, }, { role: user, content: query, }, ], modelgpt-3.5-turbo, ) return chat_completion.choices[0].message.content.strip() # Sci-fi themed corpus about ZenML World corpus [ The luminescent forests of ZenML World are inhabited by glowing Zenbots that emit a soft, pulsating light as they roam the enchanted landscape., In the neon skies of ZenML World, Cosmic Butterflies flutter gracefully, their iridescent wings leaving trails of stardust in their wake., Telepathic Treants, ancient sentient trees, communicate through the quantum neural network that spans the entire surface of ZenML World, sharing wisdom and knowledge., Deep within the melodic caverns of ZenML World, Fractal Fungi emit pulsating tones that resonate through the crystalline structures, creating a symphony of otherworldly sounds., Near the ethereal waterfalls of ZenML World, Holographic Hummingbirds hover effortlessly, their translucent wings refracting the prismatic light into mesmerizing patterns., Gravitational Geckos, masters of anti-gravity, traverse the inverted cliffs of ZenML World, defying the laws of physics with their extraordinary abilities., Plasma Phoenixes, majestic creatures of pure energy, soar above the chromatic canyons of ZenML World, their fiery trails painting the sky in a dazzling display of colors., Along the prismatic shores of ZenML World, Crystalline Crabs scuttle and burrow, their transparent exoskeletons refracting the light into a kaleidoscope of hues., ] corpus [preprocess_text(sentence) for sentence in corpus] question1 What are Plasma Phoenixes? answer1 answer_question(question1, corpus) print(fQuestion: {question1}) print(fAnswer: {answer1}) question2 ( What kinds of creatures live on the prismatic shores of ZenML World? ) answer2 answer_question(question2, corpus) print(fQuestion: {question2}) print(fAnswer: {answer2}) irrelevant_question_3 What is the capital of Panglossia? answer3 answer_question(irrelevant_question_3, corpus) print(fQuestion: {irrelevant_question_3}) print(fAnswer: {answer3})运行后输出如下Question: What are Plasma Phoenixes? Answer: Plasma Phoenixes are majestic creatures made of pure energy that soar above the chromatic canyons of Zenml World. They leave fiery trails behind them, painting the sky with dazzling displays of colors. Question: What kinds of creatures live on the prismatic shores of ZenML World? Answer: On the prismatic shores of ZenML World, you can find crystalline crabs scuttling and burrowing with their transparent exoskeletons, which refract light into a kaleidoscope of hues. Question: What is the capital of Panglossia? Answer: The capital of Panglossia is not mentioned in the provided context.值得注意的关键设计这里对查询与文本块的相似度度量使用的是Jaccard 相似系数即「两个集合交集大小 ÷ 并集大小」统计查询与文本块共现的单词数占两者全部去重单词数的比例。这种方式非常朴素且低效真正的生产级方案会用 Embedding 等更先进的技术当查询与语料完全不相关时如第三个问题模型会如实回答「语料中没有相关信息」这正体现了 RAG「让回答有据可依」的初衷该实现只是教学示范接下来的完整 ZenML 方案会在检索质量、可扩展性与工程化程度上全面超越它。完整的 85 行实现见 rag-85-loc.md。第一阶段数据摄取与预处理RAG 流水线的第一步是摄取用于训练与评估检索器、生成器的数据——既包括大规模文档语料也包括相关的元数据或标注信息。官方教程以抓取 ZenML 自己的文档站点为例演示了整个摄取链路。教程中的完整代码可在官方llm-complete-guide项目仓库中找到。用 ZenML Step 抓取 URL 列表借助简单的辅助工具函数如get_all_pages可以轻松实现一个抓取 ZenML 文档相关 URL 的 ZenML stepfrom typing import List from typing import Annotated from zenml import log_artifact_metadata, step from steps.url_scraping_utils import get_all_pages step def url_scraper( docs_url: str https://docs.zenml.io, repo_url: str https://github.com/zenml-io/zenml, website_url: str https://zenml.io, ) - Annotated[List[str], urls]: Generates a list of relevant URLs to scrape. docs_urls get_all_pages(docs_url) log_artifact_metadata( metadata{ count: len(docs_urls), }, ) return docs_urlsget_all_pages函数会爬取文档网站并返回去重后的 URL 集合。教程特意把抓取范围限制在最近发布的文档上避免把旧语法和旧信息与新文档混在一起确保只摄取最新、最相关的信息。同时该 step 用log_artifact_metadata把 URL 数量作为元数据记录下来这在 ZenML Dashboard 中可见为摄取的数据提供了额外可见性。你也可以在此基础上添加更复杂的逻辑比如过滤某些 URL 或记录更多元数据。log_artifact_metadata是 ZenML 提供的 Artifact 元数据记录入口其底层实现位于 src/zenml/artifacts/utils.py它可以把元数据关联到已存在的 Artifact 版本或关联到同一 step 中新创建的 Artifact 版本。值得注意的是从源码看该函数已被标记为 deprecated官方建议改用log_metadata(metadata{...}, infer_artifactTrue, ...)的新接口写法上更统一、能力更强。用 unstructured 解析网页内容拿到 URL 列表后教程使用开源库unstructured来加载并解析页面从而无需关心 HTML 结构和标记的细节直接把网页转成纯文本。这尤其重要因为在 LLM 的受限上下文环境下应尽量让文本内容保持精简from typing import List from unstructured.partition.html import partition_html from zenml import step step def web_url_loader(urls: List[str]) - List[str]: Loads documents from a list of URLs. document_texts [] for url in urls: elements partition_html(urlurl) text \n\n.join([str(el) for el in elements]) document_texts.append(text) return document_texts像unstructured这类框架在数据摄取上提供更多选择支持多种来源、文本预处理、特征提取但对本教程而言自实现的好处是可调试性更强——你能清楚地看到每一步加载了什么、如何被处理这是复杂框架难以提供的可见性。预处理确定 chunk size 与 overlap加载文档后需要把文本切分成适合送入 RAG 流水线的块chunk。当前文本是每条代表一个网页的长字符串直接送入 LLM 过长、影响响应速度因此要拆成更小的块。分块大小存在一个甜蜜点chunk 太小检索/搜索步骤难以找到足够的相关信息传递给 LLMchunk 太大LLM 处理起来吃力。import logging from typing import Annotated, List from utils.llm_utils import split_documents from zenml import ArtifactConfig, log_artifact_metadata, step logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) step(enable_cacheFalse) def preprocess_documents( documents: List[str], ) - Annotated[List[str], ArtifactConfig(namesplit_chunks)]: Preprocesses a list of documents by splitting them into chunks. try: log_artifact_metadata( artifact_namesplit_chunks, metadata{ chunk_size: 500, chunk_overlap: 50 }, ) return split_documents( documents, chunk_size500, chunk_overlap50 ) except Exception as e: logger.error(fError in preprocess_documents: {e}) raise这段代码有几点工程细节值得留意step(enable_cacheFalse)关闭该 step 的缓存确保每次运行都会重新执行分块逻辑避免因输入相同而跳过重跑ArtifactConfig(namesplit_chunks)为输出 Artifact 显式命名后续步骤和元数据记录都可通过该名称引用chunk_size500、chunk_overlap50以 500 字符为块大小、50 字符为重叠区。重叠的设计可以避免切分时遗漏跨边界的重要信息。分块参数并没有「标准答案」关键在于理解你的数据如果数据需要大段文字才能表达完整概念就应使用更大的 chunk size如果是对话式或问答式的数据则适合更小的 chunk size。教程选用 500/50 是因为处理对象是软件库的文档网页。根据业务不同还可以进一步清洗文本、剔除代码片段或确保代码不被切断、提取文本元数据等。第二阶段生成 Embedding 用于检索Embedding 是 RAG 检索机制的核心——它把数据表示在高维空间中语义相近的项在空间中彼此靠近。借助 Embedding推理时能快速找到与查询最相关的小块文本相比简单地在查询中匹配关键词Embedding 能捕捉语义与上下文对复杂查询和长文档更稳健。什么是 EmbeddingEmbedding 是数据的向量表示在高维空间中捕捉数据的语义含义与上下文。它们由机器学习模型如词向量或句向量模型生成编码方式保留了数据的内在结构与关系广泛用于文本分类、情感分析、信息检索等 NLP 任务。教程选用开源库sentence-transformers生成句向量具体模型为sentence-transformers/all-MiniLM-L12-v2输出维度 384from typing import Annotated, List import numpy as np from sentence_transformers import SentenceTransformer from structures import Document from zenml import ArtifactConfig, log_artifact_metadata, step step def generate_embeddings( split_documents: List[Document], ) - Annotated[ List[Document], ArtifactConfig(namedocuments_with_embeddings) ]: try: model SentenceTransformer(sentence-transformers/all-MiniLM-L12-v2) log_artifact_metadata( artifact_nameembeddings, metadata{ embedding_type: sentence-transformers/all-MiniLM-L12-v2, embedding_dimensionality: 384, }, ) document_texts [doc.page_content for doc in split_documents] embeddings model.encode(document_texts) for doc, embedding in zip(split_documents, embeddings): doc.embedding embedding return split_documents except Exception as e: logger.error(fError in generate_embeddings: {e}) raise模型选型要点该模型属于偏小的一侧更看重速度如果更看重检索相关性可换成维度更高的更大模型教程通过log_artifact_metadata把embedding_type与embedding_dimensionality记入元数据让 Embedding 的来源与维度在 Dashboard 中可追溯这里扩展了DocumentPydantic 模型为其增加embedding属性来存储每个文档对应的向量从而把 Embedding 与文档关联起来供后续检索使用。用 UMAP / t-SNE 可视化 Embedding 空间384 维的向量人类无法直接理解教程借助umap与scikit-learn的降维功能把 Embedding 投射到二维空间观察相似块如何按语义聚类。可视化时使用此前存入向量库的「父目录」parent section属性来着色能直观看出数据内在的语义空间结构。from matplotlib.colors import ListedColormap import matplotlib.pyplot as plt import numpy as np from sklearn.manifold import TSNE import umap from zenml.client import Client artifact Client().get_artifact_version(EMBEDDINGS_ARTIFACT_UUID_GOES_HERE) embeddings artifact.load() embeddings np.array([doc.embedding for doc in documents]) parent_sections [doc.parent_section for doc in documents] # Get unique parent sections unique_parent_sections list(set(parent_sections)) # Tol color palette tol_colors [ #4477AA, #EE6677, #228833, #CCBB44, #66CCEE, #AA3377, #BBBBBB, ] # Create a colormap with Tol colors tol_colormap ListedColormap(tol_colors) # Assign colors to each unique parent section section_colors tol_colors[: len(unique_parent_sections)] # Create a dictionary mapping parent sections to colors section_color_dict dict(zip(unique_parent_sections, section_colors)) # Dimensionality reduction using t-SNE def tsne_visualization(embeddings, parent_sections): tsne TSNE(n_components2, random_state42) embeddings_2d tsne.fit_transform(embeddings) plt.figure(figsize(8, 8)) for section in unique_parent_sections: if section in section_color_dict: mask [section ps for ps in parent_sections] plt.scatter( embeddings_2d[mask, 0], embeddings_2d[mask, 1], c[section_color_dict[section]], labelsection, ) plt.title(t-SNE Visualization) plt.legend() plt.show() # Dimensionality reduction using UMAP def umap_visualization(embeddings, parent_sections): umap_2d umap.UMAP(n_components2, random_state42) embeddings_2d umap_2d.fit_transform(embeddings) plt.figure(figsize(8, 8)) for section in unique_parent_sections: if section in section_color_dict: mask [section ps for ps in parent_sections] plt.scatter( embeddings_2d[mask, 0], embeddings_2d[mask, 1], c[section_color_dict[section]], labelsection, ) plt.title(UMAP Visualization) plt.legend() plt.show()从可视化结果可以看出语义相似的块聚集在一起据此可以识别数据中的模式与关系进而改进检索性能。UMAP 与 t-SNE 的降维算法不同、呈现有所差异建议两种都尝试以找到最适合的方案。完整示例见 embeddings-generation.md。架构设计生成与上传解耦教程特意把「生成 Embedding」与「上传到向量库」拆成两个独立步骤Embedding 以 NumPy 数组形式作为 ZenML Artifact 存储在 Artifact Store 中。这种模块化设计意味着未来更换向量数据库时只需替换上传步骤无需重新生成 Embedding既省算力又保持流水线的灵活性。第三阶段把 Embedding 存入向量数据库生成 Embedding 本身并不算耗时尤其在带 GPU 的机器上但我们不可能每次检索文档都重新生成一遍。正确做法是把 Embedding 存入向量数据库通过相似度快速检索最相关的块。教程选用PostgreSQL作为向量数据库它成熟稳定、可扩展且能高效存储与检索高维向量配合pgvector插件提供向量类型与相似度检索能力你也可以换成任何支持高维向量的数据库。连接使用轻量级psycopg2包加原生 SQL 语句。from zenml import step step def index_generator( documents: List[Document], ) - None: try: conn get_db_conn() with conn.cursor() as cur: # Install pgvector if not already installed cur.execute(CREATE EXTENSION IF NOT EXISTS vector) conn.commit() # Create the embeddings table if it doesnt exist table_create_command f CREATE TABLE IF NOT EXISTS embeddings ( id SERIAL PRIMARY KEY, content TEXT, token_count INTEGER, embedding VECTOR({EMBEDDING_DIMENSIONALITY}), filename TEXT, parent_section TEXT, url TEXT ); cur.execute(table_create_command) conn.commit() register_vector(conn) # Insert data only if it doesnt already exist for doc in documents: content doc.page_content token_count doc.token_count embedding doc.embedding.tolist() filename doc.filename parent_section doc.parent_section url doc.url cur.execute( SELECT COUNT(*) FROM embeddings WHERE content %s, (content,), ) count cur.fetchone()[0] if count 0: cur.execute( INSERT INTO embeddings (content, token_count, embedding, filename, parent_section, url) VALUES (%s, %s, %s, %s, %s, %s), ( content, token_count, embedding, filename, parent_section, url, ), ) conn.commit() cur.execute(SELECT COUNT(*) as cnt FROM embeddings;) num_records cur.fetchone()[0] logger.info(fNumber of vector records in table: {num_records}) # calculate the index parameters according to best practices num_lists max(num_records / 1000, 10) if num_records 1000000: num_lists math.sqrt(num_records) # use the cosine distance measure, which is what well later use for querying cur.execute( fCREATE INDEX IF NOT EXISTS embeddings_idx ON embeddings USING ivfflat (embedding vector_cosine_ops) WITH (lists {num_lists}); ) conn.commit() except Exception as e: logger.error(fError in index_generator: {e}) raise finally: if conn: conn.close()这段 step 实际完成六件事连接数据库若不存在则创建vector扩展——这是 PostgreSQL 启用向量数据类型的前提若不存在则创建embeddings表把 Embedding 与文档内容一并插入表中按最佳实践计算索引参数在 Embedding 列上创建索引。表结构设计要点embeddings表不仅存向量还同时存content、token_count、filename、parent_section、url等文档字段。这样做的原因有两个一是后续可以基于 Embedding 把对应的文档内容检索出来二是有利于直接在 Supabase 界面或任何数据库查看工具中排错。增量更新策略「何时更新 Embedding」取决于具体业务如果数据频繁且大幅变化可能需要每次全量重置如果变化轻微或不频繁只需把新文档与新 Embedding 追加进库。上面的代码选择的是只插入库中尚不存在的内容按content查重这是一种兼顾简单与实用的增量策略。索引参数ivfflat 与 cosine 距离教程为 Embedding 创建了ivfflat类型的索引搭配vector_cosine_ops算子这一组合非常适合基于余弦距离的相似度检索。lists列表数根据表中记录数动态计算最小 10最大为记录数的平方根——这是调优索引参数的合理起点实践中可以尝试不同取值观察对检索性能的影响。需要提示的是如果数据集很大、Embedding 数量很多CPU 上运行本 step 可能过慢此时应通过 ZenML 的 step operator 让该步骤跑在 GPU 机器上详见 step-operators 相关文档。向量库能力的官方集成佐证从 ZenML 仓库源码看官方对主流向量存储生态已有直接集成支持。例如 src/zenml/integrations/langchain/materializers/vector_store_materializer.py 中的LangchainVectorStoreMaterializer能把 LangChain 的VectorStore对象涵盖 Chroma、Pinecone、Qdrant、Weaviate 等各类向量库实现作为 ZenML Artifact 序列化与反序列化这意味着你可以把「已构建好的向量索引」本身当作可追踪、可版本化的 Artifact 来管理而不必像教程这样用裸 SQL 操作数据库。教程选择手写 SQL 是为了展示底层原理与保持流水线简单生产环境可按需选择更成熟的集成方式。追踪 RAG 关联的 Artifact贯穿整条流水线的一个核心能力是把 RAG 各阶段产出的 Artifact 与元数据关联起来。从前面各 step 可以看到三种典型用法记录 step 输出元数据如url_scraper记录 URL 数量count记录具名 Artifact 的元数据如preprocess_documents通过ArtifactConfig(namesplit_chunks)命名输出再用log_artifact_metadata(artifact_namesplit_chunks, metadata{...})记录chunk_size与chunk_overlap记录模型与数据特征如generate_embeddings记录embedding_type与embedding_dimensionality。这些元数据连同 Artifact 本身会一并出现在 ZenML Dashboard 中让你能够监控流水线表现、定位问题并对比不同版本流水线的差异。从源码实现看src/zenml/artifacts/utils.pylog_artifact_metadata最终会把元数据通过统一的log_metadata机制写入 ZenML 的元数据存储该函数当前已标记为 deprecated新代码建议直接使用log_metadata(metadata{...}, infer_artifactTrue, ...)。同时RAG 特有的对象链、Agent、Tokenizer、向量存储等也可以纳入 Model Control Plane 统一追踪。第四阶段基础 RAG 推理索引库就绪后就可以基于库中文档回答查询了。官方教程的推理阶段不再是一个 ZenML 流水线而是一个复用流水线产出的函数调用——它使用流水线的输出与组件生成回答。以命令行方式触发python run.py --rag-query how do I use a custom materializer inside my own zenml steps? i.e. how do I set it? inside the step decorator? --modelgpt4完整的推理逻辑如下def process_input_with_retrieval( input: str, model: str OPENAI_MODEL, n_items_retrieved: int 5 ) - str: delimiter # Step 1: Get documents related to the user input from database related_docs get_topn_similar_docs( get_embeddings(input), get_db_conn(), nn_items_retrieved ) # Step 2: Get completion from OpenAI API # Set system message to help set appropriate tone and context for model system_message f You are a friendly chatbot. \ You can answer questions about ZenML, its features and its use cases. \ You respond in a concise, technically credible tone. \ You ONLY use the context from the ZenML documentation to provide relevant answers. \ You do not make up answers or provide opinions that you dont have information to support. \ If you are unsure or dont know, just say so. \ # Prepare messages to pass to model # We use a delimiter to help the model understand the where the user_input # starts and ends messages [ {role: system, content: system_message}, {role: user, content: f{delimiter}{input}{delimiter}}, { role: assistant, content: fRelevant ZenML documentation: \n \n.join(doc[0] for doc in related_docs), }, ] logger.debug(CONTEXT USED\n\n, messages[2][content], \n\n) return get_completion_from_messages(messages, modelmodel)推理流程分两步先用get_topn_similar_docs从数据库取回与输入最相关的文档再调用get_completion_from_messages让 LLM 基于这些文档生成回答。系统提示词system message被精心设计为「只依据 ZenML 文档上下文作答、不编造答案、不确定时如实说明」并用分隔符帮助模型识别用户输入的起止位置。检索借助 pgvector 做余弦相似度排序get_topn_similar_docs使用查询文本的 Embedding在向量库中找出最相似的前 n 个文档def get_topn_similar_docs( query_embedding: List[float], conn: psycopg2.extensions.connection, n: int 5, include_metadata: bool False, only_urls: bool False, ) - List[Tuple]: embedding_array np.array(query_embedding) register_vector(conn) cur conn.cursor() if include_metadata: cur.execute( fSELECT content, url FROM embeddings ORDER BY embedding %s LIMIT {n}, (embedding_array,), ) elif only_urls: cur.execute( fSELECT url FROM embeddings ORDER BY embedding %s LIMIT {n}, (embedding_array,), ) else: cur.execute( fSELECT content FROM embeddings ORDER BY embedding %s LIMIT {n}, (embedding_array,), ) return cur.fetchall()这里的关键是ORDER BY embedding %s是 pgvector 提供的余弦距离算子数据库会按向量与查询向量的余弦距离升序返回记录并限制条数。把「找最相关文档」这个重活交给数据库本身完成效率极高也呼应了前文用vector_cosine_ops建立索引的设计——查询所用的距离度量与建索引时的度量保持一致。函数还支持include_metadata返回内容与 URL和only_urls仅返回 URL两种模式方便在不同场景下复用。生成用 litellm 统一多模型接口def get_completion_from_messages( messages, modelOPENAI_MODEL, temperature0.4, max_tokens1000 ): Generates a completion response from the given messages using the specified model. model MODEL_NAME_MAP.get(model, model) completion_response litellm.completion( modelmodel, messagesmessages, temperaturetemperature, max_tokensmax_tokens, ) return completion_response.choices[0].message.content生成环节使用litellm作为 LLM 统一接口这样就不必为每个想用的模型分别实现一套调用代码。LLM 领域迭代极快新模型层出不穷litellm让你能随时切换新模型而无需重写代码MODEL_NAME_MAP则提供了模型别名映射如命令行里的gpt4映射到具体模型标识。temperature0.4、max_tokens1000控制生成风格与长度上限。至此一条完整的 RAG 推理链路已经打通流水线负责把文档「摄取 → 分块 → 向量化 → 入库」推理函数则「把查询向量化 → 余弦检索 → 组装上下文 → 交给 LLM 生成回答」。完整推理代码见 basic-rag-inference-pipeline.md。从基础 RAG 走向更复杂的方案本教程搭建的是一条清晰的基线 RAG 流水线它本身已具备可复现、可扩展、可观测、可维护、可协作五大优势。在此基础上ZenML 的 LLMOps 指南还提供了向纵深演进的方向见 llmops-guide/README.md检索评估与生成评估用指标量化检索与生成环节的质量见 evaluation重排序Reranking对检索结果做二次精排提升相关性见 reranking微调 Embedding当文档规模很大、或文档与预训练 Embedding 的训练分布差异明显时通过微调提升检索性能见 finetuning-embeddings微调 LLM在需要改变模型行为风格、领域适配时对 LLM 本身进行微调见 finetuning-llms。总结围绕 RAG Pipelines with ZenML 官方指南本篇文章完整梳理了 RAG 在 ZenML 中的落地路径先以 85 行最小实现建立直觉再分阶段构建「URL 抓取 → unstructured 解析 → 分块预处理 → sentence-transformers 向量化 → pgvector 入库建索引」的数据侧流水线最后通过「余弦相似度检索 litellm 统一生成」完成推理。贯穿始终的是 ZenML 的 Artifact 与元数据追踪机制——它把 RAG 流水线的每个环节变成可复现、可对比、可观测的工程资产为后续引入重排序、Embedding 微调乃至 LLM 微调等高级方案打下了坚实基础。仓库中的相关子文档understanding-rag.md、data-ingestion.md、embeddings-generation.md、storing-embeddings-in-a-vector-database.md、basic-rag-inference-pipeline.md提供了各环节更深入的展开可直接按需查阅。【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址: https://gitcode.com/GitHub_Trending/ze/zenml创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
