尧图网站建设 
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 2.18 官方 API 参考文档为基础系统讲解如何将云原生向量数据库 Pinecone 接入 Haystack从PineconeDocumentStore的初始化、写入、过滤、删除、元数据统计等完整能力到PineconeEmbeddingRetriever的稠密向量检索、FilterPolicy 过滤策略与异步运行机制并结合仓库源码与配套指南给出可直接落地的代码示例。一、集成概览为什么选择 PineconePinecone 是云托管的向量数据库专为嵌入向量Embedding的存储与相似度检索设计。在 Haystack 的 Document Store 生态中它被归类为Vector Database见 Choosing a Document Store其核心特征是索引采用高效的相似度搜索算法面向高维向量的规模化检索托管服务模式提供可用的免费额度free tier但不能像 Qdrant、Weaviate 那样在本地机器运行支持在向量检索的同时叠加元数据过滤Metadata Filtering在 Haystack 集成中支持同步与异步Async两种运行方式。Haystack 2.18 的 Pinecone 集成由pinecone-haystack包提供包含两个核心类类所属模块职责PineconeDocumentStorehaystack_integrations.document_stores.pinecone连接并管理 Pinecone 索引/命名空间负责文档写入、过滤、删除与元数据统计PineconeEmbeddingRetrieverhaystack_integrations.components.retrievers.pinecone基于稠密向量dense embedding从 Document Store 检索最相似文档二、安装与环境准备2.1 安装集成包Pinecone 集成是独立于 haystack 核心的第三方包通过 pip 安装pip install pinecone-haystack若需要在索引/查询流水线中使用 Sentence Transformers 生成嵌入向量还需额外安装对应集成pip install sentence-transformers-haystack2.2 获取 API Key在 Pinecone 控制台 注册免费账户并获取 API Key。推荐通过环境变量注入这也是PineconeDocumentStore的默认行为import os os.environ[PINECONE_API_KEY] YOUR_PINECONE_API_KEY从 API 参考签名可以看到PineconeDocumentStore.__init__的第一个参数api_key: Secret Secret.from_env_var(PINECONE_API_KEY)见 version-2.18 pinecone API 参考即不传api_key时自动从PINECONE_API_KEY环境变量读取这是官方推荐的安全做法。三、PineconeDocumentStore初始化与参数详解3.1 构造签名PineconeDocumentStore.__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, ) - None3.2 参数含义与使用建议参数类型默认值说明api_keySecret环境变量PINECONE_API_KEYPinecone API Key默认从环境变量读取indexstrdefault要连接的 Pinecone 索引名若索引不存在则自动创建namespacestrdefault索引内的命名空间若不存在首次写入时会创建batch_sizeint100单批次写入的文档数量需参考 Pinecone 配额与限制 设置dimensionint768嵌入向量维度仅在创建新索引时生效specdict \| NoneNone创建新索引时的部署规格可在 serverless 与 pod 之间选择不传则默认使用us-east-1区域的 serverless 部署与免费额度兼容metriccosine \| euclidean \| dotproductcosine相似度度量方式仅在创建新索引时生效show_progressboolTrue批量 upsert 文档时是否显示进度条测试或安静脚本可设为False几点关键实践提醒依据 PineconeDocumentStore 官方指南 与 API 参考dimension与metric只在索引不存在时起作用索引一旦创建其维度与度量方式即被固定后续连接相同索引时传入的参数会被忽略。spec决定部署形态Pinecone 支持 serverless 与 pod 两种部署。默认 serverless 规格与免费额度兼容适合快速起步。例如显式指定 AWS 云上的 serverless 部署document_store PineconeDocumentStore( indexdefault, namespacedefault, dimension5, metriccosine, spec{serverless: {region: us-east-1, cloud: aws}}, )3.3 写入与基础查询示例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实现了 Haystack 的DocumentStore协议定义于 document_stores/types/protocol.py除write_documents外还提供完整的查询、删除与元数据统计能力。绝大部分方法都提供同步与异步_async后缀双版本。4.1 写入write_documentswrite_documents(documents: list[Document], policy: DuplicatePolicy DuplicatePolicy.NONE) - int返回实际写入的文档数量int。policy仅支持DuplicatePolicy.OVERWRITEPinecone 以向量 ID 为主键写入同 ID 文档时以覆盖方式更新。DuplicatePolicy枚举定义于 haystack/document_stores/types/policy.py包含NONE、SKIP、OVERWRITE、FAIL四种取值但 Pinecone 集成仅支持OVERWRITE其余策略会引发错误。在真实 RAG 场景中写入前通常先通过 Document Embedder 生成向量from haystack.document_stores.types import DuplicatePolicy from haystack import Document from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersDocumentEmbedder, ) 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.), ] document_embedder SentenceTransformersDocumentEmbedder() documents_with_embeddings document_embedder.run(documents) document_store.write_documents( documents_with_embeddings.get(documents), policyDuplicatePolicy.OVERWRITE, )4.2 过滤查询filter_documentsfilter_documents(filters: dict[str, Any] | None None) - list[Document]按 Haystack 元数据过滤语法Comparison Logic 两种字典结构筛选文档。过滤语法细节可参考仓库中的 Metadata Filtering 概念文档Comparison比较包含field如meta.type、operator、!、、、、、in、not in、value三个键。Logic逻辑包含operatorAND、OR、NOT与conditions子过滤器列表两个键支持嵌套。示例——筛选type为article且genre属于[economy, politics]的文档filters { operator: AND, conditions: [ {field: meta.type, operator: , value: article}, {field: meta.genre, operator: in, value: [economy, politics]}, ], } results document_store.filter_documents(filtersfilters)4.3 删除delete_documents / delete_all_documents / delete_by_filterdelete_documents(document_ids: list[str]) - None # 按 ID 删除 delete_all_documents() - None # 清空 Document Store delete_by_filter(filters: dict[str, Any]) - int # 按过滤器删除返回删除数量注意Pinecone 不支持服务端按过滤器删除server-side delete by filter因此delete_by_filter的实现是先搜索出匹配的文档再按 ID 逐个删除见 API 参考 中delete_by_filter的说明。4.4 更新update_by_filterupdate_by_filter(filters: dict[str, Any], meta: dict[str, Any]) - int更新所有匹配过滤条件的文档元数据meta字典会与既有元数据合并merge。同样地Pinecone 不支持服务端按过滤器更新实现方式是先搜索匹配文档更新其元数据后重新写回meta合并逻辑见 API 参考。4.5 计数count_documents / count_documents_by_filter / count_unique_metadata_by_filtercount_documents() - int count_documents_by_filter(filters: dict[str, Any]) - int count_unique_metadata_by_filter(filters: dict[str, Any], metadata_fields: list[str]) - dict[str, int]count_documents返回 Document Store 中的文档总数。count_documents_by_filter返回匹配过滤器的文档数。受 Pinecone 限制该方法通过拉取文档后在本地计数结果集上限受 Pinecone 的TOP_K_LIMIT1000 条约束。count_unique_metadata_by_filter统计匹配文档中各元数据字段的唯一值数量同样在 Python 端聚合受 1000 条上限约束。4.6 元数据字段分析get_metadata_fields_info / get_metadata_field_min_max / get_metadata_field_unique_valuesPinecone 不提供 schema 自省introspectionAPI因此以下方法均通过采样索引中已存文档的元数据最多 1000 条来推断字段信息get_metadata_fields_info() - dict[str, dict[str, str]]类型映射关系推断类型含义textDocument 的 content 字段keyword字符串类型的元数据值long数值型元数据int 或 floatboolean布尔型元数据返回示例{ content: {type: text}, category: {type: keyword}, priority: {type: long}, }另外两个方法用于更细粒度的字段分析get_metadata_field_min_max(metadata_field: str) - dict[str, Any] # 返回 {min: ..., max: ...}数值型按数值比较、布尔型 False 为 min/True 为 max、 # 字符串按字母序字段无值或类型不支持时 min/max 均为 None 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] # (唯一值列表, 匹配总数)需要特别留意的一个行为API 参考 有明确说明Pinecone 将数值型元数据存储为float因此写入的 int 值在读取时可能以数值相等的 float 返回但不同类型如 int1与 boolTrue在 Python 中即便比较相等也会被当作两个独立的值返回。五、PineconeEmbeddingRetriever稠密向量检索5.1 初始化PineconeEmbeddingRetriever.__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运行期过滤器与初始化过滤器的组合策略5.2 FilterPolicy初始化过滤器与运行期过滤器的关系FilterPolicy枚举定义于 haystack/document_stores/types/filter_policy.py取值与语义如下FilterPolicy.REPLACE运行期run/run_async传入的过滤器替换初始化时设置的过滤器FilterPolicy.MERGE运行期过滤器与初始化过滤器合并存在重叠键时以运行期过滤器为准。合并的实现由apply_filter_policy()完成filter_policy.py它会根据初始化过滤器与运行期过滤器是 Comparison 还是 Logic 结构分别调用对应的组合函数以AND为默认逻辑操作符将两者拼接成一个嵌套过滤器字典。5.3 run执行检索run( query_embedding: list[float], filters: dict[str, Any] | None None, top_k: int | None None, ) - dict[str, list[Document]]query_embedding查询文本对应的稠密向量list of float是唯一必填参数filters运行期过滤条件其作用方式取决于初始化时选择的filter_policytop_k本次运行覆盖的最大返回文档数返回字典的documents键下是相似度降序排列的Document列表。同时提供异步版本run_async(...)签名与返回结构完全一致用于在异步流水线中调用。5.4 单独使用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)5.5 在 Pipeline 中组合使用官方完整示例Retriever 在 RAG 流水线中通常位于Text Embedder 之后、PromptBuilder 之前也常作为语义搜索流水线的最后一环或位于 Text Embedder 与抽取式阅读理解器之间见 PineconeEmbeddingRetriever 指南。import os from haystack.document_stores.types import DuplicatePolicy from haystack import Document from haystack import Pipeline from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersTextEmbedder, 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 a high level of self-awareness.), Document(contentIn certain places, you can witness the phenomenon of bioluminescent waves.), ] # 1) 索引阶段文档向量化后写入 document_embedder SentenceTransformersDocumentEmbedder() documents_with_embeddings document_embedder.run(documents) document_store.write_documents( documents_with_embeddings.get(documents), policyDuplicatePolicy.OVERWRITE, ) # 2) 查询阶段文本向量化 - 向量检索 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.预期输出示意分数因模型与索引而异Document(idcfe93bc1c274908801e6670440bf2bbba54fad792770d57421f85ffa2a4fcc94, content: There are over 7,000 languages spoken around the world today., score: 0.87717235, embedding: vector of size 768)说明该示例来自 PineconeEmbeddingRetriever 官方指南也是 API 参考文档中的标准用法。assert之所以成立是因为在语义上 How many languages are there? 与 There are over 7,000 languages... 距离最近。5.6 在 Pipeline 中叠加运行期过滤器参考 Metadata Filtering 文档过滤器可以在Pipeline.run()时传入并自动路由到 Retrieverpipeline.run( data{ text_embedder: {text: query}, retriever: { filters: { operator: AND, conditions: [ {field: meta.year, operator: , value: 2024}, {field: meta.company, operator: in, value: [BMW, Mercedes]}, ], } }, }, )配合初始化时的filter_policy可以灵活实现固定范围 临时收窄或完全动态的过滤策略。六、序列化与资源管理6.1 to_dict / from_dict两个类均实现 Haystack 的序列化协议to_dict() - dict[str, Any] # 序列化为字典 from_dict(data: dict[str, Any]) - PineconeDocumentStore # 从字典反序列化 from_dict(data: dict[str, Any]) - PineconeEmbeddingRetriever这使 Document Store 与 Retriever 可以作为 YAML 流水线的一部分被持久化与重建。6.2 close / close_asyncclose() - None close_async() - None分别释放底层 Document Store 的同步与异步资源用于流水线生命周期结束时清理连接。七、常见问题与实现边界FAQ基于 API 参考 中明确的说明汇总以下使用边界重复写入策略受限write_documents只支持DuplicatePolicy.OVERWRITE其余策略不支持见 policy.py 中四种策略定义。按过滤器删除/更新是两步操作Pinecone 没有服务端 delete-by-filter / update-by-filter集成内部先查询再按 ID 操作涉及较大数据集时注意性能。计数与元数据统计受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均通过拉取最多 1000 条文档后在 Python 端计算超出部分不计入。数值元数据以 float 存储int 写入后可能以数值相等的 float 读回不同类型如1与True即使 Python 中相等也保持独立。dimension/metric仅建索引时生效连接既有索引时会被忽略请确保 Embedder 输出维度与索引维度一致。异步版本完备所有读写、过滤、计数、元数据分析方法均有_async变体便于构建异步流水线。八、相关资源API 参考本文基础文档version-2.18 Pinecone 集成 API使用指南PineconeDocumentStore使用指南PineconeEmbeddingRetriever概念Metadata Filtering概念Choosing a Document Store核心源码FilterPolicy与apply_filter_policyhaystack/document_stores/types/filter_policy.py、DuplicatePolicyhaystack/document_stores/types/policy.py、DocumentStore协议haystack/document_stores/types/protocol.py、Document数据类haystack/dataclasses/document.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），仅供参考 Haystack 与 Pinecone 集成实战：PineconeDocumentStore 与 PineconeEmbeddingRetriever 全解析
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 2.18 官方 API 参考文档为基础系统讲解如何将云原生向量数据库 Pinecone 接入 Haystack从PineconeDocumentStore的初始化、写入、过滤、删除、元数据统计等完整能力到PineconeEmbeddingRetriever的稠密向量检索、FilterPolicy 过滤策略与异步运行机制并结合仓库源码与配套指南给出可直接落地的代码示例。一、集成概览为什么选择 PineconePinecone 是云托管的向量数据库专为嵌入向量Embedding的存储与相似度检索设计。在 Haystack 的 Document Store 生态中它被归类为Vector Database见 Choosing a Document Store其核心特征是索引采用高效的相似度搜索算法面向高维向量的规模化检索托管服务模式提供可用的免费额度free tier但不能像 Qdrant、Weaviate 那样在本地机器运行支持在向量检索的同时叠加元数据过滤Metadata Filtering在 Haystack 集成中支持同步与异步Async两种运行方式。Haystack 2.18 的 Pinecone 集成由pinecone-haystack包提供包含两个核心类类所属模块职责PineconeDocumentStorehaystack_integrations.document_stores.pinecone连接并管理 Pinecone 索引/命名空间负责文档写入、过滤、删除与元数据统计PineconeEmbeddingRetrieverhaystack_integrations.components.retrievers.pinecone基于稠密向量dense embedding从 Document Store 检索最相似文档二、安装与环境准备2.1 安装集成包Pinecone 集成是独立于 haystack 核心的第三方包通过 pip 安装pip install pinecone-haystack若需要在索引/查询流水线中使用 Sentence Transformers 生成嵌入向量还需额外安装对应集成pip install sentence-transformers-haystack2.2 获取 API Key在 Pinecone 控制台 注册免费账户并获取 API Key。推荐通过环境变量注入这也是PineconeDocumentStore的默认行为import os os.environ[PINECONE_API_KEY] YOUR_PINECONE_API_KEY从 API 参考签名可以看到PineconeDocumentStore.__init__的第一个参数api_key: Secret Secret.from_env_var(PINECONE_API_KEY)见 version-2.18 pinecone API 参考即不传api_key时自动从PINECONE_API_KEY环境变量读取这是官方推荐的安全做法。三、PineconeDocumentStore初始化与参数详解3.1 构造签名PineconeDocumentStore.__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, ) - None3.2 参数含义与使用建议参数类型默认值说明api_keySecret环境变量PINECONE_API_KEYPinecone API Key默认从环境变量读取indexstrdefault要连接的 Pinecone 索引名若索引不存在则自动创建namespacestrdefault索引内的命名空间若不存在首次写入时会创建batch_sizeint100单批次写入的文档数量需参考 Pinecone 配额与限制 设置dimensionint768嵌入向量维度仅在创建新索引时生效specdict \| NoneNone创建新索引时的部署规格可在 serverless 与 pod 之间选择不传则默认使用us-east-1区域的 serverless 部署与免费额度兼容metriccosine \| euclidean \| dotproductcosine相似度度量方式仅在创建新索引时生效show_progressboolTrue批量 upsert 文档时是否显示进度条测试或安静脚本可设为False几点关键实践提醒依据 PineconeDocumentStore 官方指南 与 API 参考dimension与metric只在索引不存在时起作用索引一旦创建其维度与度量方式即被固定后续连接相同索引时传入的参数会被忽略。spec决定部署形态Pinecone 支持 serverless 与 pod 两种部署。默认 serverless 规格与免费额度兼容适合快速起步。例如显式指定 AWS 云上的 serverless 部署document_store PineconeDocumentStore( indexdefault, namespacedefault, dimension5, metriccosine, spec{serverless: {region: us-east-1, cloud: aws}}, )3.3 写入与基础查询示例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实现了 Haystack 的DocumentStore协议定义于 document_stores/types/protocol.py除write_documents外还提供完整的查询、删除与元数据统计能力。绝大部分方法都提供同步与异步_async后缀双版本。4.1 写入write_documentswrite_documents(documents: list[Document], policy: DuplicatePolicy DuplicatePolicy.NONE) - int返回实际写入的文档数量int。policy仅支持DuplicatePolicy.OVERWRITEPinecone 以向量 ID 为主键写入同 ID 文档时以覆盖方式更新。DuplicatePolicy枚举定义于 haystack/document_stores/types/policy.py包含NONE、SKIP、OVERWRITE、FAIL四种取值但 Pinecone 集成仅支持OVERWRITE其余策略会引发错误。在真实 RAG 场景中写入前通常先通过 Document Embedder 生成向量from haystack.document_stores.types import DuplicatePolicy from haystack import Document from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersDocumentEmbedder, ) 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.), ] document_embedder SentenceTransformersDocumentEmbedder() documents_with_embeddings document_embedder.run(documents) document_store.write_documents( documents_with_embeddings.get(documents), policyDuplicatePolicy.OVERWRITE, )4.2 过滤查询filter_documentsfilter_documents(filters: dict[str, Any] | None None) - list[Document]按 Haystack 元数据过滤语法Comparison Logic 两种字典结构筛选文档。过滤语法细节可参考仓库中的 Metadata Filtering 概念文档Comparison比较包含field如meta.type、operator、!、、、、、in、not in、value三个键。Logic逻辑包含operatorAND、OR、NOT与conditions子过滤器列表两个键支持嵌套。示例——筛选type为article且genre属于[economy, politics]的文档filters { operator: AND, conditions: [ {field: meta.type, operator: , value: article}, {field: meta.genre, operator: in, value: [economy, politics]}, ], } results document_store.filter_documents(filtersfilters)4.3 删除delete_documents / delete_all_documents / delete_by_filterdelete_documents(document_ids: list[str]) - None # 按 ID 删除 delete_all_documents() - None # 清空 Document Store delete_by_filter(filters: dict[str, Any]) - int # 按过滤器删除返回删除数量注意Pinecone 不支持服务端按过滤器删除server-side delete by filter因此delete_by_filter的实现是先搜索出匹配的文档再按 ID 逐个删除见 API 参考 中delete_by_filter的说明。4.4 更新update_by_filterupdate_by_filter(filters: dict[str, Any], meta: dict[str, Any]) - int更新所有匹配过滤条件的文档元数据meta字典会与既有元数据合并merge。同样地Pinecone 不支持服务端按过滤器更新实现方式是先搜索匹配文档更新其元数据后重新写回meta合并逻辑见 API 参考。4.5 计数count_documents / count_documents_by_filter / count_unique_metadata_by_filtercount_documents() - int count_documents_by_filter(filters: dict[str, Any]) - int count_unique_metadata_by_filter(filters: dict[str, Any], metadata_fields: list[str]) - dict[str, int]count_documents返回 Document Store 中的文档总数。count_documents_by_filter返回匹配过滤器的文档数。受 Pinecone 限制该方法通过拉取文档后在本地计数结果集上限受 Pinecone 的TOP_K_LIMIT1000 条约束。count_unique_metadata_by_filter统计匹配文档中各元数据字段的唯一值数量同样在 Python 端聚合受 1000 条上限约束。4.6 元数据字段分析get_metadata_fields_info / get_metadata_field_min_max / get_metadata_field_unique_valuesPinecone 不提供 schema 自省introspectionAPI因此以下方法均通过采样索引中已存文档的元数据最多 1000 条来推断字段信息get_metadata_fields_info() - dict[str, dict[str, str]]类型映射关系推断类型含义textDocument 的 content 字段keyword字符串类型的元数据值long数值型元数据int 或 floatboolean布尔型元数据返回示例{ content: {type: text}, category: {type: keyword}, priority: {type: long}, }另外两个方法用于更细粒度的字段分析get_metadata_field_min_max(metadata_field: str) - dict[str, Any] # 返回 {min: ..., max: ...}数值型按数值比较、布尔型 False 为 min/True 为 max、 # 字符串按字母序字段无值或类型不支持时 min/max 均为 None 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] # (唯一值列表, 匹配总数)需要特别留意的一个行为API 参考 有明确说明Pinecone 将数值型元数据存储为float因此写入的 int 值在读取时可能以数值相等的 float 返回但不同类型如 int1与 boolTrue在 Python 中即便比较相等也会被当作两个独立的值返回。五、PineconeEmbeddingRetriever稠密向量检索5.1 初始化PineconeEmbeddingRetriever.__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运行期过滤器与初始化过滤器的组合策略5.2 FilterPolicy初始化过滤器与运行期过滤器的关系FilterPolicy枚举定义于 haystack/document_stores/types/filter_policy.py取值与语义如下FilterPolicy.REPLACE运行期run/run_async传入的过滤器替换初始化时设置的过滤器FilterPolicy.MERGE运行期过滤器与初始化过滤器合并存在重叠键时以运行期过滤器为准。合并的实现由apply_filter_policy()完成filter_policy.py它会根据初始化过滤器与运行期过滤器是 Comparison 还是 Logic 结构分别调用对应的组合函数以AND为默认逻辑操作符将两者拼接成一个嵌套过滤器字典。5.3 run执行检索run( query_embedding: list[float], filters: dict[str, Any] | None None, top_k: int | None None, ) - dict[str, list[Document]]query_embedding查询文本对应的稠密向量list of float是唯一必填参数filters运行期过滤条件其作用方式取决于初始化时选择的filter_policytop_k本次运行覆盖的最大返回文档数返回字典的documents键下是相似度降序排列的Document列表。同时提供异步版本run_async(...)签名与返回结构完全一致用于在异步流水线中调用。5.4 单独使用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)5.5 在 Pipeline 中组合使用官方完整示例Retriever 在 RAG 流水线中通常位于Text Embedder 之后、PromptBuilder 之前也常作为语义搜索流水线的最后一环或位于 Text Embedder 与抽取式阅读理解器之间见 PineconeEmbeddingRetriever 指南。import os from haystack.document_stores.types import DuplicatePolicy from haystack import Document from haystack import Pipeline from haystack_integrations.components.embedders.sentence_transformers import ( SentenceTransformersTextEmbedder, 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 a high level of self-awareness.), Document(contentIn certain places, you can witness the phenomenon of bioluminescent waves.), ] # 1) 索引阶段文档向量化后写入 document_embedder SentenceTransformersDocumentEmbedder() documents_with_embeddings document_embedder.run(documents) document_store.write_documents( documents_with_embeddings.get(documents), policyDuplicatePolicy.OVERWRITE, ) # 2) 查询阶段文本向量化 - 向量检索 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.预期输出示意分数因模型与索引而异Document(idcfe93bc1c274908801e6670440bf2bbba54fad792770d57421f85ffa2a4fcc94, content: There are over 7,000 languages spoken around the world today., score: 0.87717235, embedding: vector of size 768)说明该示例来自 PineconeEmbeddingRetriever 官方指南也是 API 参考文档中的标准用法。assert之所以成立是因为在语义上 How many languages are there? 与 There are over 7,000 languages... 距离最近。5.6 在 Pipeline 中叠加运行期过滤器参考 Metadata Filtering 文档过滤器可以在Pipeline.run()时传入并自动路由到 Retrieverpipeline.run( data{ text_embedder: {text: query}, retriever: { filters: { operator: AND, conditions: [ {field: meta.year, operator: , value: 2024}, {field: meta.company, operator: in, value: [BMW, Mercedes]}, ], } }, }, )配合初始化时的filter_policy可以灵活实现固定范围 临时收窄或完全动态的过滤策略。六、序列化与资源管理6.1 to_dict / from_dict两个类均实现 Haystack 的序列化协议to_dict() - dict[str, Any] # 序列化为字典 from_dict(data: dict[str, Any]) - PineconeDocumentStore # 从字典反序列化 from_dict(data: dict[str, Any]) - PineconeEmbeddingRetriever这使 Document Store 与 Retriever 可以作为 YAML 流水线的一部分被持久化与重建。6.2 close / close_asyncclose() - None close_async() - None分别释放底层 Document Store 的同步与异步资源用于流水线生命周期结束时清理连接。七、常见问题与实现边界FAQ基于 API 参考 中明确的说明汇总以下使用边界重复写入策略受限write_documents只支持DuplicatePolicy.OVERWRITE其余策略不支持见 policy.py 中四种策略定义。按过滤器删除/更新是两步操作Pinecone 没有服务端 delete-by-filter / update-by-filter集成内部先查询再按 ID 操作涉及较大数据集时注意性能。计数与元数据统计受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均通过拉取最多 1000 条文档后在 Python 端计算超出部分不计入。数值元数据以 float 存储int 写入后可能以数值相等的 float 读回不同类型如1与True即使 Python 中相等也保持独立。dimension/metric仅建索引时生效连接既有索引时会被忽略请确保 Embedder 输出维度与索引维度一致。异步版本完备所有读写、过滤、计数、元数据分析方法均有_async变体便于构建异步流水线。八、相关资源API 参考本文基础文档version-2.18 Pinecone 集成 API使用指南PineconeDocumentStore使用指南PineconeEmbeddingRetriever概念Metadata Filtering概念Choosing a Document Store核心源码FilterPolicy与apply_filter_policyhaystack/document_stores/types/filter_policy.py、DuplicatePolicyhaystack/document_stores/types/policy.py、DocumentStore协议haystack/document_stores/types/protocol.py、Document数据类haystack/dataclasses/document.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），仅供参考