基于 Haystack 2.23 的 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/haystackHaystack 2.23 通过pinecone-haystack集成包提供了对云向量数据库 Pinecone 的完整支持PineconeDocumentStore负责在 Pinecone 索引与命名空间namespace上写入、过滤、删除与统计文档PineconeEmbeddingRetriever则基于稠密向量从其中检索最相关的文档。本文以本仓库中版本化 API 参考文档 docs-website/reference_versioned_docs/version-2.23/integrations-api/pinecone.md 为骨架结合当前docs目录下的集成指南与核心框架源码完整解析两个组件的全部初始化参数、方法签名、序列化/异步机制与元数据管理 API并给出可直接运行的索引与检索 Pipeline 示例。读完本文你将能在 Haystack 中完整落地一套基于 Pinecone 的语义搜索与 RAG 应用。一、集成概览云原生向量数据库接入 HaystackPinecone 是一种云托管的向量数据库与 Qdrant、Weaviate 等可以本地运行的方案不同它运行在云端但提供对新手友好的免费额度。在 Haystack 中这一集成由两个组件构成组件所属包职责PineconeDocumentStorehaystack_integrations.document_stores.pinecone连接 Pinecone 的 index 与 namespace负责文档写入、删除、过滤、统计与元数据管理PineconeEmbeddingRetrieverhaystack_integrations.components.retrievers.pinecone输入查询的稠密向量query_embedding从 Document Store 中检索最相似的文档两者配合的典型数据流是索引阶段用 Document Embedder 为文档生成向量并写入PineconeDocumentStore查询阶段用 Text Embedder 将用户问题编码为向量交给PineconeEmbeddingRetriever检索。需要说明的是pinecone-haystack属于独立的集成代码库通过 pip 分发本仓库内可直接查阅的是 Haystack 框架本体及其配套文档与类型系统。例如DuplicatePolicy、FilterPolicy以及apply_filter_policy这些集成组件所依赖的核心类型均定义在本仓库的 haystack/document_stores/types/ 目录下本文会结合这些源码深入解释相关参数的行为。二、安装与前置条件安装集成包pip install pinecone-haystack如需在示例中使用 Sentence-Transformers 生成向量再安装pip install sentence-transformers-haystack使用前必须准备 Pinecone API Key推荐通过环境变量注入这也是PineconeDocumentStore的默认取值方式export PINECONE_API_KEYYOUR_PINECONE_API_KEY在 Python 中显式设置亦可import os os.environ[PINECONE_API_KEY] YOUR_PINECONE_API_KEY三、PineconeDocumentStore连接并管理 Pinecone 索引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_keySecret环境变量PINECONE_API_KEYPinecone API Key推荐始终通过环境变量注入避免明文写在代码中indexstrdefault要连接的 Pinecone 索引名。若索引不存在会自动创建namespacestrdefault要连接的命名空间。若不存在会在首次写入文档时自动创建batch_sizeint100单批次写入的文档数量需要参考 Pinecone 官方的配额与限制文档来合理设置dimensionint768向量的维度。仅在创建新索引时生效连接已存在的索引时会被忽略specdict \| NoneNone创建新索引时使用的 Pinecone spec可在 serverless 与 pod 两种部署形态间选择并附加参数不传时默认使用us-east-1区域的 serverless 部署兼容免费额度metriccosine \| euclidean \| dotproductcosine相似度度量方式仅在创建新索引时生效show_progressboolTrueupsert 文档时是否显示进度条在测试或需要安静输出的脚本中可设为False其中几个要点值得强调索引自动创建index参数指向的索引若已存在则直接连接若不存在则按dimension、metric、spec创建。这意味着三个参数必须在首次创建索引前确定后期修改不会影响已有索引。默认 spec 兼容免费额度不传spec时默认使用 serverless 部署、us-east-1区域这正是当前 docs-website/docs/document-stores/pinecone-document-store.mdx 文档所描述的免费层友好配置。namespace 懒创建namespace 不必预先创建首次写入文档时自动建立这为多租户/多数据集隔离提供了便利。3.2 基础使用示例来自本仓库文档的最小可用示例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}}, ) 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())四、文档生命周期管理写入、统计、删除与更新PineconeDocumentStore提供了一套完整的文档管理 API绝大多数方法都同时提供同步与异步_async后缀两个版本。4.1 写入文档与重复策略write_documents( documents: list[Document], policy: DuplicatePolicy DuplicatePolicy.NONE ) - intdocuments要写入的文档列表policy重复文档处理策略返回值是被写入的文档数量。DuplicatePolicy枚举定义在 haystack/document_stores/types/policy.pyclass DuplicatePolicy(Enum): NONE none SKIP skip OVERWRITE overwrite FAIL fail需要注意API 参考文档明确说明PineconeDocumentStore仅支持DuplicatePolicy.OVERWRITE覆盖写入。因此在实践中应显式传入该策略例如from haystack.document_stores.types import DuplicatePolicy document_store.write_documents( documents_with_embeddings.get(documents), policyDuplicatePolicy.OVERWRITE, )对应的异步版本为write_documents_async(documents, policyDuplicatePolicy.NONE)签名与语义一致。4.2 计数与删除方法签名说明count_documents() - int返回存储中的文档总数count_documents_async() - int异步统计文档总数delete_documents(document_ids: list[str]) - None按文档 ID 列表删除delete_documents_async(document_ids: list[str]) - None异步按 ID 删除delete_all_documents() - None清空整个 Document Store 中的所有文档delete_all_documents_async() - None异步清空所有文档delete_by_filter(filters: dict[str, Any]) - int删除满足过滤条件的文档返回删除数量delete_by_filter_async(filters: dict[str, Any]) - int异步按过滤器删除delete_by_filter的底层行为值得注意Pinecone 不支持服务端按过滤器删除因此该方法会先检索出匹配的文档再按 ID 逐个删除。对于大规模数据集这会带来额外的查询开销需要在设计时有所预期。4.3 按过滤器更新元数据update_by_filter(filters: dict[str, Any], meta: dict[str, Any]) - intfilters筛选待更新文档的过滤条件meta要更新的元数字段会与既有元数据合并返回值为更新成功的文档数量。同样由于 Pinecone 不支持服务端按条件更新实现方式是先检索匹配文档更新其元数据后重新写回。异步版本为update_by_filter_async。五、元数据过滤语法与 FilterPolicy 策略5.1 过滤语法速览filter_documents(filters)、delete_by_filter、update_by_filter等方法的过滤条件均遵循 Haystack 统一的元数据过滤语法详细规范见 docs-website/docs/concepts/metadata-filtering.mdx。过滤条件分为两类比较型Comparison过滤器包含三个键{field: meta.type, operator: , value: article}比较运算符包括、!、、、、、in、not in具体支持的运算符以各集成为准。逻辑型Logic过滤器通过AND/OR/NOT组合多个条件filters { operator: AND, conditions: [ {field: meta.type, operator: , value: article}, {field: meta.date, operator: , value: 1420066800}, {field: meta.date, operator: , value: 1609455600}, {field: meta.rating, operator: , value: 3}, { operator: OR, conditions: [ {field: meta.genre, operator: in, value: [economy, politics]}, {field: meta.publisher, operator: , value: nytimes}, ], }, ], }5.2 FilterPolicy初始化过滤器与运行时过滤器的组合规则PineconeEmbeddingRetriever初始化时可传入filters运行时run()也可再传filters两者的组合方式由filter_policy决定。FilterPolicy定义在 haystack/document_stores/types/filter_policy.pyclass FilterPolicy(Enum): # Runtime filters replace init filters during retriever run invocation. REPLACE replace # Runtime filters are merged with init filters, with runtime filters overwriting init values. MERGE mergeREPLACE默认运行时过滤器直接替换初始化时的过滤器MERGE运行时过滤器与初始化过滤器合并同名字段以运行时为准。合并的底层实现是apply_filter_policy函数见 haystack/document_stores/types/filter_policy.py它会根据比较型/逻辑型的四种组合分别调用对应的合并函数并在逻辑运算符不匹配时打印警告后回退到运行时过滤器。框架内置检索器如 InMemoryEmbeddingRetriever在run()中也以相同方式调用apply_filter_policy(self.filter_policy, self.filters, filters)因此PineconeEmbeddingRetriever的过滤策略行为与框架其他检索器保持一致。六、元数据统计与 Schema 推断 API由于 Pinecone 不提供服务端 schema 内省能力PineconeDocumentStore提供了一组基于取样文档、在 Python 侧聚合的元数据统计方法。它们共同的限制是受 Pinecone 的TOP_K_LIMIT1000 条约束大规模结果集会取不全设计查询时应将其视为近似统计工具。6.1 按过滤器计数count_documents_by_filter(filters: dict[str, Any]) - int count_documents_by_filter_async(filters: dict[str, Any]) - int由于 Pinecone 的限制该方法通过拉取文档再计数实现对大规模结果集受 1000 条上限约束。6.2 统计唯一元数据值数量count_unique_metadata_by_filter( filters: dict[str, Any], metadata_fields: list[str] ) - dict[str, int]返回{字段名: 唯一值数量}的映射同样先拉取文档再在 Python 中聚合。6.3 元数据字段类型推断get_metadata_fields_info() - dict[str, dict[str, str]]Pinecone 不提供 schema 内省 API因此该方法通过检查索引中已存文档的元数据最多 1000 条推断字段类型类型映射如下推断类型含义text文档内容字段keyword字符串元数据值long数值元数据值int 或 floatboolean布尔元数据值返回示例{ content: {type: text}, category: {type: keyword}, priority: {type: long}, }6.4 字段最小/最大值get_metadata_field_min_max(metadata_field: str) - dict[str, Any]对数值类型按数值比较返回min/max对布尔类型以False为 min、True为 max对字符串keyword按字母序比较。该方法拉取全部文档并在 Python 中计算同样受 1000 条上限约束。当字段无值空存储、字段不存在或类型不受支持时min与max均为None。6.5 字段唯一值枚举带搜索与分页get_metadata_field_unique_values( metadata_field: str, search_term: str | None None, from_: int 0, size: int 10, filters: dict[str, Any] | None None, ) - tuple[list[Any], int]search_term可选的搜索词用于大小写不敏感的模糊子串匹配from_分页起始偏移默认0size返回值的数量默认10filters限制参与统计的文档范围返回值(唯一值列表, 匹配值总数)的元组。类型陷阱Pinecone 会把数值元数据统一存为float因此写入整数如1的任意字段读回时可能是数值相等但类型不同的float1.0而类型不同的值即使数值相等也会被区分对待例如整数1与布尔True会作为两个独立值返回。这在做唯一值统计、类型判断或下游逻辑时需特别留意。以上统计方法均有对应的_async异步版本语义完全一致。七、PineconeEmbeddingRetriever稠密向量检索7.1 初始化参数PineconeEmbeddingRetriever( *, document_store: PineconeDocumentStore, filters: dict[str, Any] | None None, top_k: int 10, filter_policy: str | FilterPolicy FilterPolicy.REPLACE ) - None参数类型默认值说明document_storePineconeDocumentStore必填检索所依赖的 Document Store 实例filtersdict \| NoneNone初始化时附加的文档过滤条件top_kint10返回的最大文档数filter_policystr \| FilterPolicyFilterPolicy.REPLACE初始化过滤器与运行时过滤器的组合策略见 5.2 节异常行为若传入的document_store不是PineconeDocumentStore实例初始化会抛出ValueError。7.2 run 与 run_asyncrun( query_embedding: list[float], filters: dict[str, Any] | None None, top_k: int | None None, ) - dict[str, list[Document]] run_async( query_embedding: list[float], filters: dict[str, Any] | None None, top_k: int | None None, ) - dict[str, list[Document]]query_embedding查询文本的稠密向量表示float 列表filters运行时过滤条件其生效方式取决于初始化时选择的filter_policytop_k运行时覆盖最大返回文档数返回值{documents: [...]}即与查询向量最相似的文档列表。7.3 独立使用在已有文档的索引上单独调用检索器示例来自 docs-website/docs/pipeline-components/retrievers/pineconedenseretriever.mdxfrom 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)7.4 在 Pipeline 中使用索引 查询全流程参考文档 docs-website/reference_versioned_docs/version-2.23/integrations-api/pinecone.md 提供的完整示例一次性完成生成文档向量 → 写入 → 构建查询 Pipeline → 检索断言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.Pipeline 中的关键连接是text_embedder.embedding → retriever.query_embeddingText Embedder 把用户查询编码为稠密向量注入检索器的query_embedding输入槽检索器返回documents输出。运行结果示例依赖模型略有差异Document(idcfe93bc1c274908801e6670440bf2bbba54fad792770d57421f85ffa2a4fcc94, content: There are over 7,000 languages spoken around the world today., score: 0.87717235, embedding: vector of size 768)在典型 RAG 架构中该检索器位于 Text Embedder 之后、PromptBuilder之前在抽取式问答Extractive QA场景中则可连接在 Text Embedder 与TransformersExtractiveReader之间。八、序列化to_dict 与 from_dict两个组件均实现标准的 Haystack 序列化协议# PineconeDocumentStore to_dict() - dict[str, Any] from_dict(data: dict[str, Any]) - PineconeDocumentStore # PineconeEmbeddingRetriever to_dict() - dict[str, Any] from_dict(data: dict[str, Any]) - PineconeEmbeddingRetrieverto_dict()将组件含初始化参数、API Key 引用方式等序列化为字典便于以 YAML/JSON 形式保存 Pipeline 定义或跨进程传输from_dict()则从字典还原组件实例。这是 Haystack Pipeline 可声明式构建与持久化的基础能力。九、资源释放与异步支持9.1 资源释放组件同步异步行为PineconeDocumentStoreclose()close_async()释放底层关联的同步/异步资源PineconeEmbeddingRetrieverclose()close_async()释放底层 Document Store 的同步/异步资源在应用退出、Pipeline 重建或测试清理阶段调用对应方法可避免连接泄漏。9.2 异步 API 汇总PineconeDocumentStore几乎为每个操作都提供了_async变体write_documents_async、count_documents_async、filter_documents_async、delete_documents_async、delete_all_documents_async、delete_by_filter_async、update_by_filter_async、count_documents_by_filter_async、count_unique_metadata_by_filter_async、get_metadata_fields_info_async、get_metadata_field_min_max_async、get_metadata_field_unique_values_async检索器侧则有run_async。这使集成可以无缝融入基于asyncio的高并发服务例如在异步 Web 框架中直接驱动查询 Pipeline。十、限制与最佳实践小结综合 API 参考文档与仓库配套文档使用该集成时有以下几点需牢记索引参数创建后不可改dimension、metric、spec仅在新建索引时生效规划阶段就要确定向量维度与相似度度量。重复策略仅支持覆盖write_documents请显式传DuplicatePolicy.OVERWRITE这是文档明确标注的受支持策略枚举定义见 haystack/document_stores/types/policy.py。按过滤器删除/更新有额外开销Pinecone 不支持服务端delete by filter与update by filter这两个操作都是先查后改大数据集上成本较高。统计类方法受 1000 条上限约束count_documents_by_filter、count_unique_metadata_by_filter、get_metadata_fields_info、get_metadata_field_min_max、get_metadata_field_unique_values均受 PineconeTOP_K_LIMIT限制适合近似统计与小数据集场景。数值元数据以 float 存储整数可能以 float 形式读回不同来源的同值数据会被区分类型处理。过滤策略默认为替换filter_policy默认REPLACE需要初始化与运行时过滤条件叠加时请显式选择MERGE实现见 haystack/document_stores/types/filter_policy.py。同步/异步配对使用混用同步与异步方法可能导致资源管理混乱建议在异步代码路径中统一使用_async系列方法并在结束时调用close_async。如需进一步查阅本仓库中与本文主题直接相关的配套资料包括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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考