Haystack 2.20 与 Pinecone 向量数据库集成实践PineconeEmbeddingRetriever 与 PineconeDocumentStore 全指南【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack本文以 Haystack 2.20 的 Pinecone 集成 API 参考文档为主线系统讲解PineconeEmbeddingRetriever检索组件与PineconeDocumentStore文档存储的完整用法从环境准备、索引初始化、文档写入到基于稠密向量的语义检索、元数据过滤再到资源释放与序列化。读者阅读后将掌握在 Haystack Pipeline 中接入 Pinecone 云向量数据库、搭建 RAG 语义检索链路的完整实战方案并能理解过滤策略FilterPolicy、重复文档策略DuplicatePolicy等底层机制的设计取舍。一、集成概览为什么在 Haystack 中使用 PineconePinecone 是一款云端向量数据库具有速度快、上手简单、托管省心的特点。与 Qdrant、Weaviate 等可以本地运行的向量数据库不同Pinecone 无法在用户本机部署但它提供了慷慨的免费额度free tier非常适合从原型验证到生产环境的大规模语义检索场景。在 Haystack 生态中Pinecone 集成由独立的pinecone-haystack包提供包含两个核心类类所属模块职责PineconeDocumentStorehaystack_integrations.document_stores.pinecone连接 Pinecone 索引与命名空间负责文档的写入、删除、更新、过滤与统计PineconeEmbeddingRetrieverhaystack_integrations.components.retrievers.pinecone基于查询向量与文档向量的相似度从PineconeDocumentStore中检索最相关的文档安装方式非常简单pip install pinecone-haystack如果要在 Pipeline 中使用基于 Sentence Transformers 的嵌入组件还需要pip install sentence-transformers二、准备工作账号、API Key 与索引规划在代码层面使用 Pinecone 之前需要完成以下准备工作注册账号并获取 API Key在 Pinecone 官网注册免费账号进入控制台获取 API Key。注入 API Key推荐通过环境变量PINECONE_API_KEY提供PineconeDocumentStore默认会从该环境变量读取密钥import os os.environ[PINECONE_API_KEY] YOUR_PINECONE_API_KEY也可以在初始化时显式传入api_key参数类型为Secret默认值即Secret.from_env_var(PINECONE_API_KEY)。理解 index 与 namespace 的关系在 Haystack 中每个PineconeDocumentStore都作用于某个索引index的特定命名空间namespace。若未指定index 与 namespace 均默认为default。如果索引已存在Document Store 会直接连接如果不存在则自动创建。三、PineconeDocumentStore 详解3.1 初始化参数PineconeDocumentStore通过关键字参数构造完整签名如下PineconeDocumentStore( *, api_key: Secret Secret.from_env_var(PINECONE_API_KEY), index: str default, namespace: str default, batch_size: int 100, dimension: int 768, spec: dict[str, Any] | None None, metric: Literal[cosine, euclidean, dotproduct] cosine, show_progress: bool True )各参数含义与使用要点api_keySecretPinecone API 密钥。默认从环境变量PINECONE_API_KEY读取也可以显式传入Secret.from_token(...)。indexstr默认default要连接的 Pinecone 索引名称。若索引不存在会自动创建。namespacestr默认default要连接的命名空间。若命名空间不存在会在首次写入文档时自动创建。batch_sizeint默认100单批次写入的文档数量。调整该参数时应参考 Pinecone 官方文档中关于请求配额与限制的说明Quotas and Limits过大的批量写入可能导致请求被拒绝或触发限流。dimensionint默认768嵌入向量的维度。仅在创建新索引时生效若索引已存在此参数被忽略。specdict | None创建新索引时使用的 Pinecone spec用于选择serverless无服务器与pod专用实例两种部署形态并可设置附加参数。若不提供默认使用us-east-1区域的 serverless 部署兼容免费额度。例如spec{serverless: {region: us-east-1, cloud: aws}}metricLiteral[cosine, euclidean, dotproduct]默认cosine相似度检索所用的距离度量仅在创建新索引时生效。三种选项分别对应余弦相似度、欧氏距离与点积。show_progressbool默认True写入upsert文档时是否显示进度条。在测试或需要安静输出的脚本中可设为False。一个典型的初始化与写入示例来自 pinecone-document-store.mdxfrom haystack import Document from haystack_integrations.document_stores.pinecone import PineconeDocumentStore # 确保已设置 PINECONE_API_KEY 环境变量 document_store PineconeDocumentStore( indexdefault, namespacedefault, dimension5, metriccosine, spec{serverless: {region: us-east-1, cloud: aws}}, ) document_store.write_documents( [ Document(contentThis is first, embedding[0.1] * 5), Document(contentThis is second, embedding[0.1, 0.2, 0.3, 0.4, 0.5]), ], ) print(document_store.count_documents())3.2 文档写入write_documentswrite_documents( documents: list[Document], policy: DuplicatePolicy DuplicatePolicy.NONE ) - int将Document列表写入 Pinecone返回实际写入的文档数量。关键限制在于PineconeDocumentStore只支持DuplicatePolicy.OVERWRITE。DuplicatePolicy定义在 haystack/document_stores/types/policy.py是一个包含四种取值的枚举取值行为NONE不执行任何去重逻辑默认值SKIP遇到重复 ID 时跳过该文档OVERWRITE用新文档覆盖已有 ID 的文档FAIL遇到重复 ID 时抛出异常由于 Pinecone 的 upsert 语义天然以向量 ID 为主键执行覆盖写入因此该集成将策略固定为OVERWRITE。典型用法from haystack.document_stores.types import DuplicatePolicy document_store.write_documents( documents_with_embeddings.get(documents), policyDuplicatePolicy.OVERWRITE, )对应的异步版本为write_documents_async(documents, policy) - int签名与语义完全一致适用于asyncio环境。3.3 文档过滤filter_documentsfilter_documents(filters: dict[str, Any] | None None) - list[Document]返回满足过滤条件的文档列表。filters遵循 Haystack 的元数据过滤语法field/operator/value比较过滤或operator/conditions逻辑过滤详细规格可参考项目内关于元数据过滤的文档。异步版本为filter_documents_async。3.4 文档删除集成提供了四个删除入口delete_documents(document_ids: list[str]) - None按文档 ID 列表删除异步版为delete_documents_async。delete_all_documents() - None清空 Document Store 中的全部文档异步版为delete_all_documents_async。delete_by_filter(filters: dict[str, Any]) - int按过滤条件删除文档返回删除数量。Pinecone 不支持服务端按过滤条件删除因此该方法内部先检索匹配的文档再按 ID 删除。异步版为delete_by_filter_async。deleted document_store.delete_by_filter({field: meta.category, operator: , value: spam})3.5 元数据更新update_by_filterupdate_by_filter(filters: dict[str, Any], meta: dict[str, Any]) - int更新所有匹配过滤条件的文档的元数据返回更新数量。Pinecone 同样不支持服务端按过滤条件更新因此实现方式是先检索匹配文档合并新的元数据字段meta会与已有元数据做 merge再重新写回 Pinecone。异步版为update_by_filter_async。3.6 统计与计数类方法这一类方法常用于调试、数据质量管理与检索前的探查均提供了同步与异步_async后缀两个版本count_documents() - int返回文档总数。count_documents_by_filter(filters: dict[str, Any]) - int返回匹配过滤条件的文档数量。注意由于 Pinecone 的限制该方法需要先拉取文档再计数对于较大的结果集会受到 PineconeTOP_K_LIMIT最多 1000 条的限制。count_unique_metadata_by_filter(filters, metadata_fields) - dict[str, int]对匹配过滤条件的文档统计每个指定元数据字段的唯一值个数。同样受TOP_K_LIMIT限制且聚合在 Python 侧完成。3.7 元数据探查类方法Pinecone 不提供 schema 内省introspectionAPI因此集成通过**抽样检查索引中的文档元数据最多 1000 条**来推断字段信息get_metadata_fields_info() - dict[str, dict[str, str]]返回字段名到类型信息的映射。类型映射规则为推断类型对应数据text文档content字段keyword字符串元数据值long数值型int/float元数据值boolean布尔元数据值返回示例{ content: {type: text}, category: {type: keyword}, priority: {type: long}, }get_metadata_field_min_max(metadata_field: str) - dict[str, Any]返回某个元数据字段的最小值与最大值返回字典包含min与max两个键。支持三种类型数值型按数值大小取 min/max布尔型以False为 min、True为 max字符串型按字母顺序取 min/max。若字段无任何值空存储、字段不存在或不支持的类型两个值均为None。该方法会拉取全部文档并在 Python 中计算受TOP_K_LIMIT限制。get_metadata_field_unique_values(metadata_field, search_termNone, from_0, size10, filtersNone) - tuple[list[Any], int]获取某个元数据字段的唯一值列表支持搜索与分页。search_term用于大小写不敏感的子串匹配过滤from_为分页起始偏移默认 0size为返回数量默认 10filters用于缩小考察的文档范围。返回(唯一值列表, 匹配总数)的元组。使用该 API 时需要注意 Pinecone 的一个存储特性Pinecone 会将数值元数据存储为float参考实现中的_convert_meta_to_int因此写入的int在读取时可能以数值相等的float返回不同数据类型的值即使数值上相等也会被区分对待例如int的1与bool的True会作为两个独立值返回。四、PineconeEmbeddingRetriever 详解4.1 工作原理与在 Pipeline 中的位置PineconeEmbeddingRetriever是一个基于嵌入向量的检索器它将查询向量与文档向量进行相似度比较从PineconeDocumentStore中取出与查询最相关的文档。它在 Pipeline 中最常见的位置包括RAG Pipeline 中位于 Text Embedder 之后、PromptBuilder之前语义搜索 Pipeline 的末尾作为最后一个组件输出检索结果抽取式问答 Pipeline 中位于 Text Embedder 之后、ExtractiveReader之前。使用它时必须保证查询与文档的嵌入向量已经生成索引 Pipeline 中添加 Document Embedder负责为文档生成向量查询 Pipeline 中添加 Text Embedder负责为查询生成向量。此外影响向量检索效果的关键参数——嵌入dimension与距离metric——需要在初始化PineconeDocumentStore时指定。4.2 初始化参数PineconeEmbeddingRetriever( *, document_store: PineconeDocumentStore, filters: dict[str, Any] | None None, top_k: int 10, filter_policy: str | FilterPolicy FilterPolicy.REPLACE )document_storePineconeDocumentStore必填。检索目标文档存储。若传入的不是PineconeDocumentStore实例构造函数会抛出ValueError。filtersdict | None初始化时设定的过滤条件作用于检索返回的文档。top_kint默认10最多返回的文档数量。filter_policystr | FilterPolicy默认FilterPolicy.REPLACE决定初始化过滤条件与运行时过滤条件如何组合的策略详见下文 4.4 节。4.3 run 与 run_asyncrun( query_embedding: list[float], filters: dict[str, Any] | None None, top_k: int | None None, ) - dict[str, list[Document]]query_embeddinglist[float]查询的嵌入向量必填。filtersdict | None运行时过滤条件。运行时过滤器如何生效取决于初始化时选择的filter_policy。top_kint | None本次运行最多返回的文档数覆盖初始化时的值。返回值dict[str, list[Document]]中键documents对应与query_embedding最相似的文档列表按相似度降序。异步版本run_async签名与返回结构完全一致。单独使用检索器的示例向量仅为演示用from haystack_integrations.components.retrievers.pinecone import PineconeEmbeddingRetriever from haystack_integrations.document_stores.pinecone import PineconeDocumentStore # 确保已设置 PINECONE_API_KEY 环境变量 document_store PineconeDocumentStore( indexmy_index_with_documents, namespacemy_namespace, dimension768, ) retriever PineconeEmbeddingRetriever(document_storedocument_store) # 使用一个虚拟向量保持示例简洁 retriever.run(query_embedding[0.1] * 768)4.4 过滤策略 FilterPolicyREPLACE 与 MERGEfilter_policy决定了检索器初始化时传入的filters与run调用时传入的filters如何组合。FilterPolicy定义在 haystack/document_stores/types/filter_policy.pyFilterPolicy.REPLACEreplace运行时过滤器直接替换初始化过滤器FilterPolicy.MERGEmerge运行时过滤器与初始化过滤器合并存在重叠字段时以运行时值为准。当策略为MERGE且两类过滤器同时存在时底层会调用 apply_filter_policy 及一系列组合函数根据过滤器形态比较过滤{field, operator, value}或逻辑过滤{operator, conditions}执行不同的合并逻辑默认逻辑运算符为AND。例如初始化过滤条件与运行时逻辑过滤同为AND时两边的条件会被拼接合并若运算符不一致则初始化过滤条件会被忽略并给出警告。若仅提供单侧过滤器则直接返回生效的那一侧。from haystack.document_stores.types import FilterPolicy retriever PineconeEmbeddingRetriever( document_storedocument_store, filters{field: meta.type, operator: , value: article}, filter_policyFilterPolicy.MERGE, )五、完整实战从写入索引到 RAG 查询 Pipeline下面是一个完整可运行的最小 RAG 语义检索示例整合了索引写入与查询两个阶段改编自 pineconedenseretriever.mdximport os from haystack import Document, Pipeline from haystack.document_stores.types import DuplicatePolicy from haystack.components.embedders import ( SentenceTransformersTextEmbedder, SentenceTransformersDocumentEmbedder, ) from haystack_integrations.components.retrievers.pinecone import PineconeEmbeddingRetriever from haystack_integrations.document_stores.pinecone import PineconeDocumentStore # 确保已设置 PINECONE_API_KEY 环境变量 document_store PineconeDocumentStore( indexmy_index, namespacemy_namespace, dimension768, ) documents [ Document(contentThere are over 7,000 languages spoken around the world today.), Document( contentElephants have been observed to behave in a way that indicates a high level of self-awareness, such as recognizing themselves in mirrors. ), Document( contentIn certain parts of the world, like the Maldives, Puerto Rico, and San Diego, you can witness the phenomenon of bioluminescent waves. ), ] # 索引阶段为文档生成嵌入向量并写入 Pinecone document_embedder SentenceTransformersDocumentEmbedder() document_embedder.warm_up() documents_with_embeddings document_embedder.run(documents) document_store.write_documents( documents_with_embeddings.get(documents), policyDuplicatePolicy.OVERWRITE, ) # 查询阶段Text Embedder 生成查询向量Retriever 完成向量检索 query_pipeline Pipeline() query_pipeline.add_component(text_embedder, SentenceTransformersTextEmbedder()) query_pipeline.add_component( retriever, PineconeEmbeddingRetriever(document_storedocument_store), ) query_pipeline.connect(text_embedder.embedding, retriever.query_embedding) query How many languages are there? result query_pipeline.run({text_embedder: {text: query}}) print(result[retriever][documents][0])运行后检索器会返回与查询最相关的文档例如Document(idcfe93bc1c274908801e6670440bf2bbba54fad792770d57421f85ffa2a4fcc94, content: There are over 7,000 languages spoken around the world today., score: 0.87717235, embedding: vector of size 768)可以看到Pinecone 返回的相似度得分score: 0.87717235即为所选metric此处默认 cosine下的相似度结果。六、序列化与资源管理6.1 to_dict 与 from_dict两个类都实现了 Haystack 标准的序列化协议便于将组件状态保存到 YAML/JSON 或从配置中恢复to_dict() - dict[str, Any]将组件序列化为字典包含类名与全部初始化参数API Key 以Secret的安全形式存储不会明文落盘。from_dict(data: dict[str, Any]) - PineconeDocumentStore | PineconeEmbeddingRetriever从字典反序列化重建组件实例。这使你可以将完整的 RAG Pipeline含 Pinecone 检索器序列化为 YAML 描述在部署环境中复现参考 marshal/yaml.py。6.2 资源释放close 与 close_asyncclose() - None释放底层 Document Store 持有的同步资源如 HTTP 连接池。close_async() - None释放异步资源。在长期运行的服务中Pipeline 运行结束后调用close()或异步环境中的close_async()可以避免连接泄漏。七、常见问题与使用注意索引已存在时dimension/metric不生效这两个参数只在创建新索引时被使用。若维度不匹配请删除旧索引或在 Pinecone 控制台新建。重复写入策略受限write_documents仅支持DuplicatePolicy.OVERWRITE这是由 Pinecone upsert 语义决定的。过滤删除/更新需要二次操作delete_by_filter与update_by_filter均因 Pinecone 缺少服务端能力采用先查后删/改的实现方式数据量大时耗时较长。TOP_K_LIMIT限制count_documents_by_filter、count_unique_metadata_by_filter、get_metadata_fields_info、get_metadata_field_min_max、get_metadata_field_unique_values等探查类方法都会拉取文档后在 Python 侧计算单次最多覆盖 1000 条文档。数值元数据以 float 存储Pinecone 会把数值元数据转为float读取时int可能显示为数值相等的float不同数据类型即使数值相等也会被视为不同的唯一值。八、参考与延伸阅读Pinecone 集成 API 参考docs-website/reference_versioned_docs/version-2.20/integrations-api/pinecone.mdDocument Store 使用指南docs-website/versioned_docs/version-2.20/document-stores/pinecone-document-store.mdx检索器使用指南docs-website/versioned_docs/version-2.20/pipeline-components/retrievers/pineconedenseretriever.mdxFilterPolicy与过滤器组合逻辑源码haystack/document_stores/types/filter_policy.pyDuplicatePolicy枚举源码haystack/document_stores/types/policy.py元数据过滤语法可参考 haystack/utils/filters.py 中过滤器解析与校验的实现通过本文介绍的PineconeDocumentStore与PineconeEmbeddingRetriever你可以快速在 Haystack 2.20 中搭建一条基于 Pinecone 云端向量库的完整语义检索与 RAG 链路从 API Key 注入、索引与命名空间规划到文档批量写入、向量检索、元数据过滤与统计探查再到 Pipeline 序列化与资源释放每一步都有对应的源码级依据可查。【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考