Feast 构建 RAG 应用实战基于 Docling 与 Milvus 的文档嵌入存储与向量检索【免费下载链接】feastThe Open Source Feature Store for AI/ML项目地址: https://gitcode.com/GitHub_Trending/fe/feast本篇教程介绍如何用 Feast开源特征存储、Docling文档解析/切块工具与 Milvus向量数据库搭建一个完整的 RAGRetrieval Augmented Generation检索增强生成应用。读完本文你将掌握如何用 Feast 的feature_store.yaml配置 Milvus 向量后端、如何定义带向量索引的 FeatureView 与按需转换on-demand transformation、如何把 PDF 文档切块并嵌入后写入在线存储、如何用retrieve_online_documents_v2做 top-k 相似度检索以及如何使用 Feast 内置的DocEmbedder把「切块-嵌入-写库」一步到位。一、RAG 的整体流程与 Feast 的角色RAG 是把生成式模型如 LLM与检索系统结合的技术让模型针对特定目标例如问答产生上下文相关的输出。典型的 RAG 流程包括获取与业务相关的文本数据将每份文档切分为更小的文本块chunk将文本块转换为嵌入向量embedding将文本块连同块/文档标识符一起写入数据库运行时检索出相关文本块将其注入 LLM 的上下文调用 LLM 推理 API生成上下文相关的输出将输出返回给最终用户。Feast 的价值在于它通过 Milvus 等向量数据库集成让你像管理特征一样管理文档嵌入——统一的write_to_online_store/retrieve_online_documents_v2API、注册表registry与版本治理、批量与实时写入模式。完整的可运行示例位于 examples/rag-docling其中 示例仓库定义 与 feature_store.yaml 可直接参照本文各步骤另有 docling-demo.ipynb 与 docling-quickstart.ipynb 两个 Notebook 版本。二、环境准备PrerequisitesPython 3.10 或更高版本安装带 Milvus 与 NLP 支持的 Feastpip install feast[milvus, nlp]对特征存储和向量嵌入有基本了解。三、Step 0下载、计算并导出 Docling 示例数据集第一步用 Docling 从 PDF 中解析出结构化文档再用HybridChunker切块、用 sentence-transformers 生成嵌入最终导出为两份 Parquetdocling_samples.parquet每行一个 chunk与metadata_samples.parquet文件级元数据含 PDF 原始字节。import os import io import pypdf import logging import hashlib from datetime import datetime import requests import pandas as pd from transformers import AutoTokenizer from sentence_transformers import SentenceTransformer from docling.datamodel.base_models import ConversionStatus, InputFormat from docling.datamodel.pipeline_options import PdfPipelineOptions from docling.document_converter import DocumentConverter, PdfFormatOption from docling.chunking import HybridChunker logging.basicConfig(levellogging.INFO) _log logging.getLogger(__name__) # Base URL for PDFs BASE_URL https://raw.githubusercontent.com/DS4SD/docling/refs/heads/main/tests/data/pdf/ PDF_FILES [ 2203.01017v2.pdf, 2305.03393v1-pg9.pdf, 2305.03393v1.pdf, amt_handbook_sample.pdf, code_and_formula.pdf, picture_classification.pdf, redp5110_sampled.pdf, right_to_left_01.pdf, right_to_left_02.pdf, right_to_left_03.pdf ] INPUT_DOC_PATHS [os.path.join(BASE_URL, pdf_file) for pdf_file in PDF_FILES] # Configure PDF processing pipeline_options PdfPipelineOptions() pipeline_options.generate_page_images True doc_converter DocumentConverter( format_options{InputFormat.PDF: PdfFormatOption(pipeline_optionspipeline_options)} ) # Load tokenizer and embedding model EMBED_MODEL_ID sentence-transformers/all-MiniLM-L6-v2 MAX_TOKENS 64 # Small token limit for demonstration tokenizer AutoTokenizer.from_pretrained(EMBED_MODEL_ID) embedding_model SentenceTransformer(EMBED_MODEL_ID) chunker HybridChunker(tokenizertokenizer, max_tokensMAX_TOKENS, merge_peersTrue) def embed_text(text: str) - list[float]: Generate an embedding for a given text. return embedding_model.encode([text], normalize_embeddingsTrue).tolist()[0] def generate_document_rows(conv_results): Generator that yields one row per chunk from each successfully converted document. Each yielded dict contains: - file_name: Name of the source file. - raw_markdown: Serialized text for the chunk. - chunk_embedding: The embedding vector for that chunk. processed_docs 0 for conv_res in conv_results: if conv_res.status ! ConversionStatus.SUCCESS: continue processed_docs 1 file_name conv_res.input.file.stem # FIX: Use .file.stem instead of .path # Extract the document object (which contains iterate_items) document conv_res.document try: document_markdown document.export_to_markdown() except: document_markdown if document is None: _log.warning(fDocument conversion failed for {file_name}) continue # Process each chunk from the document for chunk in chunker.chunk(dl_docdocument): # Use document here! raw_chunk chunker.serialize(chunkchunk) embedding embed_text(raw_chunk) yield { file_name: file_name, full_document_markdown: document_markdown, raw_chunk_markdown: raw_chunk, chunk_embedding: embedding, } _log.info(fProcessed {processed_docs} documents successfully.) def generate_chunk_id(file_name: str, raw_chunk_markdown: str) - str: Generate a unique chunk ID based on file_name and raw_chunk_markdown. unique_string f{file_name}-{raw_chunk_markdown} return hashlib.sha256(unique_string.encode()).hexdigest() conv_results doc_converter.convert_all(INPUT_DOC_PATHS, raises_on_errorFalse) # Build a DataFrame where each row is a unique chunk record rows list(generate_document_rows(conv_results)) df pd.DataFrame.from_records(rows) output_dict {} for file_name in PDF_FILES: try: r requests.get(BASE_URL file_name) pdf_bytes io.BytesIO(r.content) output_dict[file_name] pdf_bytes.getvalue() except Exception as e: print(ferror with {file_name} \n{e}) odf pd.DataFrame.from_dict(output_dict, orientindex, columns[bytes]).reset_index() odf.rename({index: file_name}, axis1, inplaceTrue) odf[file_name] odf[file_name].str.replace(.pdf, ) finaldf df.merge(odf, onfile_name, howleft) finaldf[chunk_id] finaldf.apply(lambda row: generate_chunk_id(row[file_name], row[raw_chunk_markdown]), axis1) finaldf[created] datetime.now() pdf_example pypdf.PdfReader(io.BytesIO(finaldf[bytes].values[0])) finaldf.drop([full_document_markdown, bytes], axis1).to_parquet(feature_repo/data/docling_samples.parquet, indexFalse) odf.to_parquet(feature_repo/data/metadata_samples.parquet, indexFalse)几个关键点值得注意chunk_id 的生成方式以sha256(file_name raw_chunk_markdown)作为每个块的唯一标识。这个 ID 后续既是 Feast 的实体键entity key也是按需转换on-demand transformation时关联原始 PDF 的依据——同一份文档、同一段切块文本必然得到相同 ID保证幂等。嵌入模型sentence-transformers/all-MiniLM-L6-v2输出 384 维向量与后面 Milvus 配置中的embedding_dim: 384对应并对嵌入做了 L2 归一化normalize_embeddingsTrue这是配合余弦相似度COSINE检索的常规做法。切块策略HybridChunker(tokenizer..., max_tokens64, merge_peersTrue)按 token 上限切块并合并相邻片段MAX_TOKENS 64只是演示用的较小取值生产环境可按语义粒度调大。四、Step 1在 Feast 中配置 Milvus 在线存储创建feature_store.yaml把在线存储指向本地 Milvus本地嵌入式数据库文件并启用向量能力project: docling-rag provider: local registry: data/registry.db online_store: type: milvus path: data/online_store.db vector_enabled: true embedding_dim: 384 index_type: IVF_FLAT offline_store: type: file entity_key_serialization_version: 3 auth: type: no_auth结合 Milvus 在线存储实现sdk/python/feast/infra/online_stores/milvus_online_store/milvus.py可以看出各参数的实际含义path本地模式下的 Milvus 数据文件路径本教程使用嵌入式单文件data/online_store.db无需独立部署 Milvus 服务embedding_dim向量维度必须与嵌入模型输出维度一致本教程为 384。源码中该配置项默认值为 128见 milvus.py 中embedding_dim: Optional[int] 128因此在配置里显式声明 384 是必要的vector_enabled是否启用基于嵌入的向量检索。源码中该开关默认为Truevector_enabled: Optional[bool] True当它为False且未提供query_string时检索会直接报错milvus.py 中有 Either vector_enabled must be True for embedding search or query_string must be provided for keyword search 的校验逻辑。显式声明vector_enabled: true让配置意图更清晰index_typeMilvus 索引类型这里选用IVF_FLAT。仓库自带的 rag-docling 示例配置 则使用FLAT小数据量下暴力检索精度无损、无需建索引参数两者均可按数据规模选择project: docling-rag注册表中的项目名registry: data/registry.db本地 SQLite 注册表文件entity_key_serialization_version: 3实体键序列化版本与示例仓库保持一致offline_store: type: file离线存储使用文件源配合 Parquet 数据文件。五、Step 2定义数据源、实体与 FeatureView创建feature_repo.py定义实体、文件数据源、请求数据源以及一个普通 FeatureView 加一个按需转换 FeatureView。以下代码与 examples/rag-docling/feature_repo/example_repo.py 中的实际定义一致from datetime import timedelta import pandas as pd from feast import ( FeatureView, Field, FileSource, Entity, RequestSource, ) from feast.data_format import ParquetFormat from feast.types import Float64, Array, String, ValueType, PdfBytes from feast.on_demand_feature_view import on_demand_feature_view from sentence_transformers import SentenceTransformer from typing import Dict, Any, List import hashlib from docling.datamodel.base_models import DocumentStream import io from docling.document_converter import DocumentConverter from transformers import AutoTokenizer from sentence_transformers import SentenceTransformer from docling.chunking import HybridChunker # Load tokenizer and embedding model EMBED_MODEL_ID sentence-transformers/all-MiniLM-L6-v2 MAX_TOKENS 64 # Small token limit for demonstration tokenizer AutoTokenizer.from_pretrained(EMBED_MODEL_ID) embedding_model SentenceTransformer(EMBED_MODEL_ID) chunker HybridChunker(tokenizertokenizer, max_tokensMAX_TOKENS, merge_peersTrue) def embed_text(text: str) - list[float]: Generate an embedding for a given text. return embedding_model.encode([text], normalize_embeddingsTrue).tolist()[0] def generate_chunk_id(file_name: str, raw_chunk_markdown: str) - str: Generate a unique chunk ID based on file_name and raw_chunk_markdown. unique_string f{file_name}-{raw_chunk_markdown} if raw_chunk_markdown ! else f{file_name} return hashlib.sha256(unique_string.encode()).hexdigest() # Define entities chunk Entity( namechunk_id, descriptionChunk ID, value_typeValueType.STRING, join_keys[chunk_id], ) document Entity( namedocument_id, descriptionDocument ID, value_typeValueType.STRING, join_keys[document_id], ) source FileSource( file_formatParquetFormat(), path./data/docling_samples.parquet, timestamp_fieldcreated, ) input_request_pdf RequestSource( namepdf_request_source, schema[ Field(namedocument_id, dtypeString), Field(namepdf_bytes, dtypePdfBytes), Field(namefile_name, dtypeString), ], ) # Define the view for retrieval docling_example_feature_view FeatureView( namedocling_feature_view, entities[chunk], schema[ Field(namefile_name, dtypeString), Field(nameraw_chunk_markdown, dtypeString), Field( namevector, dtypeArray(Float64), vector_indexTrue, vector_search_metricCOSINE, ), Field(namechunk_id, dtypeString), ], sourcesource, ttltimedelta(hours2), ) on_demand_feature_view( entities[chunk, document], sources[input_request_pdf], schema[ Field(namedocument_id, dtypeString), Field(namechunk_id, dtypeString), Field(namechunk_text, dtypeString), Field( namevector, dtypeArray(Float64), vector_indexTrue, vector_search_metricL2, ), ], modepython, write_to_online_storeTrue, singletonTrue, ) def docling_transform_docs(inputs: dict[str, Any]): document_ids, chunks, embeddings, chunk_ids [], [], [], [] buf io.BytesIO( inputs[pdf_bytes], ) doc_source DocumentStream(nameinputs[file_name], streambuf) converter DocumentConverter() result converter.convert(doc_source) for i, chunk in enumerate(chunker.chunk(dl_docresult.document)): raw_chunk chunker.serialize(chunkchunk) embedding embed_text(raw_chunk) chunk_id fchunk-{i} document_ids.append(inputs[document_id]) chunks.append(raw_chunk) chunk_ids.append(chunk_id) embeddings.append(embedding) return { document_id: document_ids, chunk_id: chunk_ids, vector: embeddings, chunk_text: chunks, }这份定义中有三处设计是本教程的核心值得结合源码理解5.1 向量字段的声明方式Field(namevector, dtypeArray(Float64), vector_indexTrue, vector_search_metricCOSINE)中vector_indexTrue标记该字段需要建立向量索引Milvus 在线存储会据此创建相应的向量字段与索引vector_search_metric指定相似度度量COSINE / L2。本例中批量预切块视图docling_feature_view用 COSINE因为嵌入已归一化而按需转换视图docling_transform_docs声明为 L2——检索时通过distance_metric参数传入的度量与字段声明应保持一致否则结果语义可能不符合预期。5.2 双实体模型chunk 与 documentchunkchunk_id每个文本块的实体键由 Step 0 的generate_chunk_id生成的 SHA-256 摘要充当documentdocument_id文档级实体键。注意RequestSource中的file_name字段——由于 PDF 原始字节无法直接放入数据文件太大、且不是结构化数据教程把pdf_bytes定义为 Feast 的PdfBytes类型放入请求源按需转换时随请求传入再用sha256(file_name)形式的约定关联文档。5.3 按需转换on-demand feature view写入时即时解析 PDFdocling_transform_docs是modepython的按需转换视图write_to_online_storeTrue表示其输出会写入在线存储singletonTrue表示该文档的转换结果只计算一次并缓存。它的输入来自RequestSourcedocument_id、pdf_bytes、file_name内部流程是用DocumentStream(name..., streambuf)包装请求中的 PDF 字节流Docling 的DocumentConverter现场解析为结构化文档HybridChunker切块chunk-{i}顺序编号逐块生成嵌入返回document_id、chunk_id、vector、chunk_text四个列表列由 Feast 展开为多行写入。从源码结构看这正是 Feast「按需转换」特性的典型用法把昂贵的解析/嵌入计算推迟到写入时刻、只对新文档执行一次避免离线批处理与在线请求两条路径的逻辑分叉。六、Step 3更新注册表将上述 FeatureView 定义应用到注册表feast apply执行后docling_feature_view与docling_transform_docs会注册到data/registry.db并在 Milvus 中创建对应的向量集合与索引。七、Step 4数据摄取Ingestion把预切块的 Parquet 数据与原始 PDF 分别以不同方式写入在线存储——这是本教程「混合摄取」思路的体现import pandas as pd from feast import FeatureStore store FeatureStore(repo_path.) df pd.read_parquet(./data/docling_samples.parquet) mdf pd.read_parquet(./data/metadata_samples.parquet) df[chunk_embedding] df[vector].apply(lambda x: x.tolist()) embedding_length len(df[vector][0]) print(fembedding length {embedding_length}) df[created] pd.Timestamp.now() mdf[created] pd.Timestamp.now() # Ingesting transformed data to the feature view that has no associated transformation store.write_to_online_store(feature_view_namedocling_feature_view, dfdf) # Turning off transformation on writes is as simple as changing the default behavior store.write_to_online_store( feature_view_namedocling_transform_docs, dfdf[df[document_id]!doc-1], transform_on_writeFalse, ) # Now we can transform a raw PDF on the fly store.write_to_online_store( feature_view_namedocling_transform_docs, dfmdf[mdf[document_id]doc-1], transform_on_writeTrue, # this is the default )要点说明docling_feature_view数据已经在 Step 0 完成了切块与嵌入直接批量写入无需任何转换transform_on_writeFalse对docling_transform_docs这个带转换的视图可以显式关闭写入时转换——把已经切好块、算好嵌入的df除 doc-1 外的文档直接写入跳过 Docling 解析省时省算力transform_on_writeTrue默认行为对 doc-1 这一份文档只传入了含 PDF 原始字节的元数据mdfFeast 在写入时触发docling_transform_docs的按需转换函数现场解析 PDF、切块、嵌入后写库。这模拟了「新文档到达时即时处理」的生产场景。八、Step 5检索相关文档写入完成后即可对任意查询做 top-k 向量检索from feast import FeatureStore # Initialize FeatureStore store FeatureStore(.) # Generate query embedding question Who are the authors of the paper? query_embedding embed_text(question) # Retrieve similar documents context_data store.retrieve_online_documents_v2( features[ docling_feature_view:vector, docling_feature_view:file_name, docling_feature_view:raw_chunk_markdown, docling_feature_view:chunk_id, ], queryquery_embedding, top_k3, distance_metricCOSINE, ).to_df() print(context_data)参数约定上需要注意features必须是feature_view:feature形式的字符串引用且其中必须包含嵌入向量字段本身教程中为docling_feature_view:vectorquery传入查询文本经embed_text生成的 384 维嵌入top_k3返回最相似的 3 个块distance_metricCOSINE与 Step 2 中docling_feature_view的vector_search_metric声明一致返回值.to_df()得到包含file_name、raw_chunk_markdown、chunk_id等列的 DataFrame供后续拼装 LLM 上下文。从 API 实现看retrieve_online_documents_v2定义于 feature_store.py除上述文本嵌入检索外还支持query_string关键词/混合检索、query_image_bytesquery_image_model图像相似度检索、以及combine_with_text/text_weight/image_weight/combine_strategy图文多模态组合检索weighted_sum/concatenate/average等能力。也就是说教程里只演示了纯文本检索这一条路径同一套 API 面可以扩展到多模态 RAG。Milvus 在线存储侧的检索实现见 milvus.py 中的retrieve_online_documents_v2。九、Step 6用检索结果驱动 LLM 生成最后把检索到的文档块拼进提示词调用 LLMfrom openai import OpenAI import os client OpenAI( api_keyos.environ.get(OPENAI_API_KEY), ) # Format documents for context def format_documents(context_data, base_prompt): documents \n.join([fDocument {i1}: {row[embedded_documents__sentence_chunks]} for i, row in context_data.iterrows()]) return f{base_prompt}\n\nContext documents:\n{documents} BASE_PROMPT You are a helpful assistant that answers questions based on the provided context. FULL_PROMPT format_documents(context_data, BASE_PROMPT) # Generate response response client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: FULL_PROMPT}, {role: user, content: query_embedding} ], ) print(\n.join([c.message.content for c in response.choices]))两点提示format_documents中引用的列名embedded_documents__sentence_chunks与本教程 Step 5 实际返回的raw_chunk_markdown列并不一致实际使用时请改为按 Step 5 检索结果中真实的文本列名如raw_chunk_markdown拼接上下文用户消息处同样应传入原始问题文本question而非嵌入向量本身。十、替代方案用 DocEmbedder 一步完成摄取上面流程需要手工完成「切块 → 嵌入 → 定义 FeatureView → 写库」多个环节。Feast 提供了DocEmbedder类实现见 sdk/python/feast/doc_embedder.py自动化整个流水线自动生成 FeatureView 定义、应用仓库、切块、生成嵌入并写入在线存储。10.1 安装依赖pip install feast[milvus,rag]10.2 设置并摄取from feast import DocEmbedder import pandas as pd # Prepare your documents as a DataFrame df pd.DataFrame({ id: [doc1, doc2, doc3], text: [ Aaron is a prophet, high priest, and the brother of Moses..., God at Sinai granted Aaron the priesthood for himself..., His rod turned into a snake. Then he stretched out..., ], }) # DocEmbedder handles everything: generates FeatureView, applies repo, # chunks text, generates embeddings, and writes to the online store embedder DocEmbedder( repo_pathfeature_repo/, feature_view_nametext_feature_view, ) result embedder.embed_documents( documentsdf, id_columnid, source_columntext, column_mapping(text, text_embedding), )column_mapping(text, text_embedding)表示从源列text读取文本把生成的嵌入写入名为text_embedding的列再经 schema 转换映射到 FeatureView。10.3 检索与查询摄取之后检索方式与 Step 5 完全相同只是 feature 引用换成自动生成的视图与列名from feast import FeatureStore store FeatureStore(feature_repo/) query_embedding embed_text(Who are the authors of the paper?) context_data store.retrieve_online_documents_v2( features[ text_feature_view:embedding, text_feature_view:text, text_feature_view:source_id, ], queryquery_embedding, top_k3, distance_metricCOSINE, ).to_df()10.4 自定义流水线组件DocEmbedder在每个阶段都可扩展自定义切块器、自定义嵌入器、自定义 schema 转换函数。自定义 Chunker子类化BaseChunker实现load_parse_and_chunk方法接收文档并返回 chunk 字典列表from feast.chunker import BaseChunker, ChunkingConfig from typing import Any, Optional class SentenceChunker(BaseChunker): Chunks text by sentences instead of word count. def load_parse_and_chunk( self, source: Any, source_id: str, source_column: str, source_type: Optional[str] None, ) - list[dict]: import re text str(source) # Split on sentence boundaries sentences re.split(r(?[.!?])\s, text) chunks [] current_chunk [] chunk_index 0 for sentence in sentences: current_chunk.append(sentence) combined .join(current_chunk) if len(combined.split()) self.config.chunk_size: chunks.append({ chunk_id: f{source_id}_{chunk_index}, original_id: source_id, source_column: combined, chunk_index: chunk_index, }) # Keep overlap by retaining the last sentence current_chunk [sentence] chunk_index 1 # Dont forget the last chunk if current_chunk and len( .join(current_chunk).split()) self.config.min_chunk_size: chunks.append({ chunk_id: f{source_id}_{chunk_index}, original_id: source_id, source_column: .join(current_chunk), chunk_index: chunk_index, }) return chunks也可以直接使用内置的TextChunker并通过ChunkingConfig调节切块参数from feast import TextChunker, ChunkingConfig chunker TextChunker(configChunkingConfig( chunk_size200, chunk_overlap50, min_chunk_size30, max_chunk_chars1000, ))自定义 Embedder子类化BaseEmbedder在_register_default_modalities中注册模态处理器并实现embed方法from feast.embedder import BaseEmbedder, EmbeddingConfig from typing import Any, List, Optional import numpy as np class OpenAIEmbedder(BaseEmbedder): Embedder that uses the OpenAI API for text embeddings. def __init__(self, model: str text-embedding-3-small, config: Optional[EmbeddingConfig] None): self.model model self._client None super().__init__(config) def _register_default_modalities(self) - None: self.register_modality(text, self._embed_text) property def client(self): if self._client is None: from openai import OpenAI self._client OpenAI() return self._client def get_embedding_dim(self, modality: str) - Optional[int]: # text-embedding-3-small produces 1536-dim vectors if modality text: return 1536 return None def embed(self, inputs: List[Any], modality: str) - np.ndarray: if modality not in self._modality_handlers: raise ValueError(fUnsupported modality: {modality}) return self._modality_handlersmodality def _embed_text(self, inputs: List[str]) - np.ndarray: response self.client.embeddings.create(inputinputs, modelself.model) return np.array([item.embedding for item in response.data])注意get_embedding_dim必须返回模型的真实输出维度DocEmbedder会用它或vector_length参数确定向量字段长度。自定义 Schema 转换函数把「切块 嵌入」后的 DataFrame 映射成 FeatureView 期望的确切 schema入参pd.DataFrame返回pd.DataFrameimport pandas as pd from datetime import datetime, timezone def my_schema_transform_fn(df: pd.DataFrame) - pd.DataFrame: Map chunked embedded columns to the FeatureView schema. return pd.DataFrame({ passage_id: df[chunk_id], text: df[text], embedding: df[text_embedding], event_timestamp: [datetime.now(timezone.utc)] * len(df), source_id: df[original_id], # Add any extra columns your FeatureView expects chunk_index: df[chunk_index], })这与 doc_embedder.py 中的default_schema_transform_fn输出结构一致passage_id、text、embedding、event_timestamp、source_id五列——DocEmbedder自动生成的 FeatureView 就是按这套 schema 组织的。组合起来把自定义组件传给DocEmbedderfrom feast import DocEmbedder embedder DocEmbedder( repo_pathfeature_repo/, feature_view_nametext_feature_view, chunkerSentenceChunker(configChunkingConfig(chunk_size150, min_chunk_size20)), embedderOpenAIEmbedder(modeltext-embedding-3-small), schema_transform_fnmy_schema_transform_fn, vector_length1536, # Match the OpenAI embedding dimension ) # Embed and ingest result embedder.embed_documents( documentsdf, id_columnid, source_columntext, column_mapping(text, text_embedding), )注意使用自定义schema_transform_fn时确保返回 DataFrame 的列与你的 FeatureView schema 匹配使用不同输出维度的自定义嵌入器时相应设置vector_length或依赖get_embedding_dim自动探测。完整的端到端示例可参考 rag_feast_docembedder.ipynb。十一、为什么用 Feast 做 RAG综合上述实现Feast 为 RAG 系统提供的能力可以归纳为简化向量数据库的配置与管理——feature_store.yaml中几个字段即可接入 Milvus本地嵌入式模式零部署成本对写入与读取嵌入提供一致 API——write_to_online_store写、retrieve_online_documents_v2读向量字段与普通特征字段统一声明同时支持批量与实时数据摄取——预切块数据可transform_on_writeFalse直写新文档可触发按需转换现场解析为文档仓库提供版本化与治理能力——所有视图、实体、schema 均登记在注册表中feast apply统一管理无缝对接多种向量数据库后端源码中 Postgres、SQLite、MongoDB、Elasticsearch、ScyllaDB、远程存储等均实现了retrieve_online_documents_v2接口用统一 API 同时管理特征数据与文档嵌入——RAG 语料库可以与传统 ML 特征共存于同一套基础设施。更多向量数据库在 Feast 中的用法参见 向量数据库参考文档。【免费下载链接】feastThe Open Source Feature Store for AI/ML项目地址: https://gitcode.com/GitHub_Trending/fe/feast创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
