在 Haystack 中集成 Pinecone 向量数据库PineconeDocumentStore 与 PineconeEmbeddingRetriever 实战指南【免费下载链接】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 官方 Pinecone 集成pinecone-haystack编写的实战指南。它面向需要在 RAG、语义搜索或抽取式问答流水线中使用 Pinecone 托管向量数据库的开发者系统讲解PineconeDocumentStore与PineconeEmbeddingRetriever的初始化参数、读写删改与元数据管理 API、同步/异步两种执行模型以及它们在 Haystack Pipeline 中的完整接入方式。读完本文你将能够独立完成从注册 Pinecone、创建索引到写入带稠密向量的文档、构建向量检索流水线的全流程。集成概览与安装Pinecone 是云托管的向量数据库与本地可运行的其他方案不同它以托管服务的形式提供并带有对新手友好的免费额度。在 Haystack 中Pinecone 的官方集成由两个核心类构成其 API 参考见 integrations-api/pinecone.mdhaystack_integrations.document_stores.pinecone.PineconeDocumentStore负责与 Pinecone 索引/命名空间建立连接承担文档的写入、查询、删除、更新与元数据统计haystack_integrations.components.retrievers.pinecone.PineconeEmbeddingRetriever基于文档稠密向量dense embeddings从PineconeDocumentStore中检索与查询向量最相似的文档。安装只需要一条命令pip install pinecone-haystack如果要在示例中直接使用 Sentence-Transformers 作为嵌入器再安装对应的集成包pip install sentence-transformers-haystack安装完成后注册一个 Pinecone 账号并获取 API Key。推荐将 Key 写入环境变量PINECONE_API_KEY集成默认即从该环境变量读取也可在初始化时显式传入。初始化 PineconeDocumentStore参数逐项解析PineconeDocumentStore的构造函数签名如下来自 integrations-api/pinecone.md__init__( *, 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 ) - None参数类型默认值作用与注意事项api_keySecret从环境变量PINECONE_API_KEY读取Pinecone 的 API Key建议用环境变量管理避免明文写入代码indexstrdefault要连接的 Pinecone 索引名。若索引不存在会在初始化时自动创建namespacestrdefault索引内的命名空间。若不存在会在首次写入文档时创建。每个PineconeDocumentStore实例固定作用于一个索引的一个命名空间batch_sizeint100单批写入的文档数量需参考 Pinecone 官方配额与限制合理设置dimensionint768嵌入向量的维度。仅在创建新索引时生效连接已存在的索引时该参数被忽略specdict \| NoneNone创建新索引时使用的部署规格可控制 serverless / pod 部署方式及区域等参数。不传时使用默认规格us-east-1区域的 serverless 部署与免费额度兼容metricLiteralcosine相似度度量可选cosine、euclidean、dotproduct。仅在创建新索引时生效show_progressboolTrueupsert 文档时是否显示进度条测试或脚本中可设为False静默输出关于dimension与metric的仅创建时生效语义集成文档在 pinecone-document-store.mdx 中做了明确说明如果索引已经存在PineconeDocumentStore直接连接它否则才创建新索引此时dimension、metric与spec才会起作用。初始化示例显式给出 serverless 规格from 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}}, )写入文档DuplicatePolicy 与批量 upsertwrite_documents负责把Document列表写入 Pineconewrite_documents( documents: list[Document], policy: DuplicatePolicy DuplicatePolicy.NONE ) - int参数说明documents待写入的Document列表policy写入重复文档时的策略。PineconeDocumentStore只支持DuplicatePolicy.OVERWRITE覆盖写入这与 Pinecone 的向量 upsert 语义一致返回值实际写入的文档数量。DuplicatePolicy枚举定义于 policy.py包含NONE、SKIP、OVERWRITE、FAIL四种取值由于 Pinecone 本身以 ID 为键执行 upsert集成层将其收敛为OVERWRITE一种行为。Document数据类定义于 document.py包含id、content、blob、meta、score、embedding、sparse_embedding等字段。写入时需要注意文档 ID 在未显式指定的情况下会根据字段内容自动生成 SHA-256 哈希Document._create_id因此对同一内容重复写入会命中相同 ID——这正解释了为什么示例代码普遍使用DuplicatePolicy.OVERWRITE来保证幂等更新。写入示例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())基于稠密向量的检索PineconeEmbeddingRetrieverPineconeEmbeddingRetriever的作用是接收查询向量query_embedding在PineconeDocumentStore中找出与查询向量最相似的文档并返回。它的组件定位说明见 pineconedenseretriever.mdx在 RAG 流水线中通常位于 Text Embedder 之后、PromptBuilder之前也可以作为语义搜索流水线的最后一个组件或在抽取式 QA 流水线中位于 Text Embedder 之后、TransformersExtractiveReader之前。初始化参数__init__( *, document_store: PineconeDocumentStore, filters: dict[str, Any] | None None, top_k: int 10, filter_policy: str | FilterPolicy FilterPolicy.REPLACE ) - None参数类型默认值说明document_storePineconeDocumentStore必填检索所依赖的文档存储必须是PineconeDocumentStore实例否则抛出ValueErrorfiltersdict \| NoneNone对检索结果施加的元数据过滤条件top_kint10最多返回的文档数量filter_policystr \| FilterPolicyFilterPolicy.REPLACE运行时过滤条件与初始化过滤条件如何组合run / run_async 运行参数run( query_embedding: list[float], filters: dict[str, Any] | None None, top_k: int | None None, ) - dict[str, list[Document]]query_embedding查询文本的嵌入向量list[float]是唯一必填的运行参数filters运行时的元数据过滤条件其作用方式取决于初始化时选择的filter_policytop_k本次运行最多返回的文档数不传则回退到初始化时的值返回值形如{documents: [Document, ...]}的字典Document按与查询向量的相似度排序。run_async提供完全一致的语义与签名只是以异步方式执行底层 Pinecone 调用。FilterPolicyREPLACE 与 MERGEFilterPolicy定义于 filter_policy.py取值含义如下FilterPolicy.REPLACEreplace运行时的过滤条件整体替换初始化时设置的过滤条件FilterPolicy.MERGEmerge运行时条件与初始化条件合并运行时值覆盖同名初始化值。合并逻辑由apply_filter_policy()实现当初始化与运行时条件均为比较型field/operator/value或逻辑型operator/conditions过滤器时会按AND逻辑组合字段冲突时以运行时条件为准并输出警告日志。这为同一检索组件服务多个请求的场景提供了灵活性——例如在初始化时限定meta.category article运行时再叠加meta.rating 3的条件。独立使用示例检索器依赖已写入文档的PineconeDocumentStorefrom 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)端到端实战索引流水线与查询流水线将文档嵌入、写入与检索串成一个完整流程分为两步先用SentenceTransformersDocumentEmbedder为文档生成向量并写入 store再构建查询Pipeline把SentenceTransformersTextEmbedder的输出接到PineconeEmbeddingRetriever的query_embedding输入上import os from haystack.document_stores.types import DuplicatePolicy from haystack import Document from haystack import Pipeline # Requires: pip install sentence-transformers-haystack from haystack_integrations.components.embedders.sentence_transformers import SentenceTransformersTextEmbedder from haystack_integrations.components.embedders.sentence_transformers import SentenceTransformersDocumentEmbedder from haystack_integrations.components.retrievers.pinecone import PineconeEmbeddingRetriever from haystack_integrations.document_stores.pinecone import PineconeDocumentStore os.environ[PINECONE_API_KEY] YOUR_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...), Document(contentIn certain places, you can witness the phenomenon of bioluminescent waves.)] document_embedder SentenceTransformersDocumentEmbedder() documents_with_embeddings document_embedder.run(documents) document_store.write_documents(documents_with_embeddings.get(documents), policyDuplicatePolicy.OVERWRITE) 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? res query_pipeline.run({text_embedder: {text: query}}) assert res[retriever][documents][0].content There are over 7,000 languages spoken around the world today.一个典型的返回结果形如示例见 pineconedenseretriever.mdxDocument(idcfe93bc1c274908801e6670440bf2bbba54fad792770d57421f85ffa2a4fcc94, content: There are over 7,000 languages spoken around the world today., score: 0.87717235, embedding: vector of size 768)score即查询向量与文档向量的相似度得分由 Pinecone 依据索引创建时设定的metric默认cosine计算得出。文档管理与元数据操作 API除了基础的读写PineconeDocumentStore还提供了一套面向元数据的管理与统计接口每个方法都配套提供_async异步版本基础管理count_documents()/count_documents_async()返回 store 中的文档总数delete_documents(document_ids: list[str])按 ID 列表删除文档delete_all_documents()清空整个 store作用于当前索引的命名空间。过滤查询filter_documents(filters: dict | None)返回满足过滤条件的文档过滤语法遵循 Haystack 的元数据过滤规范delete_by_filter(filters)删除满足条件的文档。Pinecone 不支持服务端按条件删除因此该方法是先检索匹配文档、再按 ID 删除的两步模拟实现返回被删除的文档数update_by_filter(filters, meta)更新满足条件文档的元数据meta会与既有元数据合并。同样受 Pinecone 服务端能力限制实现为先检索、再更新并重写count_documents_by_filter(filters)统计满足条件的文档数量。由于 Pinecone 限制实现为取回文档再计数count_unique_metadata_by_filter(filters, metadata_fields)统计每个指定元数据字段的唯一值个数返回{字段名: 唯一值个数}字典同样在 Python 侧聚合。元数据 Schema 探查Pinecone 不提供 schema 内省 API因此集成通过采样已存文档最多 1000 条来推断元数据结构get_metadata_fields_info()推断各字段类型类型映射为text文档内容、keyword字符串元数据、long数值元数据、boolean布尔元数据示例返回{ content: {type: text}, category: {type: keyword}, priority: {type: long}, }get_metadata_field_min_max(metadata_field)返回指定字段的最小/最大值支持数值按数值比较、布尔False为 min、True为 max、字符串按字母序三种类型字段无值时返回{min: None, max: None}get_metadata_field_unique_values(metadata_field, search_termNone, from_0, size10, filtersNone)分页获取字段的唯一值列表search_term支持大小写不敏感的子串过滤返回(唯一值列表, 匹配总数)二元组。关于上述方法文档明确标注了两点实现细节一是多数统计方法需要取回全部文档后在 Python 中聚合因此受 Pinecone 单次查询上限TOP_K_LIMIT当前为 1000 条约束二是 Pinecone 会把数值型元数据存为float因此写入的int可能以数值相等的float读回而不同数据类型如int 1与bool True即使数值相等也会作为两个独立值返回。序列化与资源生命周期PineconeDocumentStore与PineconeEmbeddingRetriever都实现了 Haystack 的标准序列化协议to_dict() - dict[str, Any]把组件序列化为字典包括 API Key 以Secret形式安全表示可用于 YAML/JSON 流水线配置的持久化from_dict(data)从字典反序列化出组件实例。同时两者都实现了同步/异步资源释放接口close()释放底层 Document Store 的同步资源close_async()释放底层 Document Store 的异步资源。在长期运行的服务器应用中建议在流水线生命周期结束时调用close释放与 Pinecone 的连接资源。已知限制与注意事项结合集成文档与仓库中的 release note使用时应留意以下几点查询上限约束delete_by_filter、update_by_filter、count_documents_by_filter、count_unique_metadata_by_filter、get_metadata_fields_info、get_metadata_field_min_max、get_metadata_field_unique_values等需要拉取全量文档再在 Python 侧处理的接口均受 Pinecone 单次查询向量上限当前 1000 条约束超出部分无法被统计到按条件更新/删除是模拟实现Pinecone 不提供服务端 delete/update by filter集成采用先查后改策略命中数较大时开销明显元数据数值类型Pinecone 将数值统一存为float读取时int可能变为float索引规格不可热改dimension、metric、spec只在索引首次创建时生效连接已有索引时需保证与索引实际配置一致写入策略只支持DuplicatePolicy.OVERWRITE配合Document基于内容的自动哈希 ID可实现可靠的幂等写入。小结Pinecone 集成让 Haystack 开发者可以借助云托管的向量数据库构建生产级 RAG 与语义搜索应用PineconeDocumentStore屏蔽了索引创建、命名空间、批量 upsert 与元数据管理的细节并提供完整的同步/异步双接口PineconeEmbeddingRetriever则以标准组件的形式接入 Pipeline配合FilterPolicy实现灵活的元数据过滤。本文涉及的完整 API 签名可直接查阅 integrations-api/pinecone.md组件使用指南见 pinecone-document-store.mdx 与 pineconedenseretriever.mdx底层过滤策略实现见 filter_policy.py。【免费下载链接】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），仅供参考