Haystack 的 Pinecone 集成
模块 haystack_integrations.components.retrievers.pinecone.embedding_retriever
PineconeEmbeddingRetriever
从PineconeDocumentStore 中检索文档，依据其密集嵌入。
使用示例
import os
from haystack.document_stores.types import DuplicatePolicy
from haystack import Document
from haystack import Pipeline
from haystack.components.embedders 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(index="my_index", namespace="my_namespace", dimension=768)
documents = [Document(content="There are over 7,000 languages spoken around the world today."),
Document(content="Elephants have been observed to behave in a way that indicates..."),
Document(content="In certain places, you can witness the phenomenon of bioluminescent waves.")]
document_embedder = SentenceTransformersDocumentEmbedder()
document_embedder.warm_up()
documents_with_embeddings = document_embedder.run(documents)
document_store.write_documents(documents_with_embeddings.get("documents"), policy=DuplicatePolicy.OVERWRITE)
query_pipeline = Pipeline()
query_pipeline.add_component("text_embedder", SentenceTransformersTextEmbedder())
query_pipeline.add_component("retriever", PineconeEmbeddingRetriever(document_store=document_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."
PineconeEmbeddingRetriever.__init__
def __init__(*,
document_store: PineconeDocumentStore,
filters: Optional[Dict[str, Any]] = None,
top_k: int = 10,
filter_policy: Union[str, FilterPolicy] = FilterPolicy.REPLACE)
参数:
document_store: Pinecone 文档存储。filters: 应用于检索到的文档的过滤器。top_k: 返回的文档的最大数量。filter_policy: 确定如何应用过滤器的策略。
引发:
ValueError: 如果document_store不是一个实例PineconeDocumentStore.
PineconeEmbeddingRetriever.to_dict
def to_dict() -> Dict[str, Any]
将组件序列化为字典。
返回值:
包含序列化数据的字典。
PineconeEmbeddingRetriever.from_dict
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "PineconeEmbeddingRetriever"
从字典反序列化组件。
参数:
data: 要反序列化的字典。
返回值:
反序列化后的组件。
PineconeEmbeddingRetriever.run
@component.output_types(documents=List[Document])
def run(query_embedding: List[float],
filters: Optional[Dict[str, Any]] = None,
top_k: Optional[int] = None) -> Dict[str, List[Document]]
从PineconeDocumentStore 中检索文档，依据其密集嵌入。
参数:
query_embedding: 查询的嵌入。filters: 应用于检索到的文档的过滤器。运行时过滤器的应用方式取决于初始化检索器时选择的filter_policy。有关更多详细信息，请参阅 init 方法文档字符串。top_k: 返回的Document的最大数量。
返回值:
类似于query_embedding.
PineconeEmbeddingRetriever.run_async
@component.output_types(documents=List[Document])
async def run_async(query_embedding: List[float],
filters: Optional[Dict[str, Any]] = None,
top_k: Optional[int] = None) -> Dict[str, List[Document]]
异步从PineconeDocumentStore 中检索文档，依据其密集嵌入。
参数:
query_embedding: 查询的嵌入。filters: 应用于检索到的文档的过滤器。运行时过滤器的应用方式取决于初始化检索器时选择的filter_policy。有关更多详细信息，请参阅 init 方法文档字符串。top_k: 返回的Document的最大数量。
返回值:
类似于query_embedding.
模块 haystack_integrations.document_stores.pinecone.document_store
METADATA_SUPPORTED_TYPES
List[str] 被支持并单独检查
PineconeDocumentStore
一个使用 Pinecone 向量数据库的文档存储。
PineconeDocumentStore.__init__
def __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: Optional[Dict[str, Any]] = None,
metric: Literal["cosine", "euclidean", "dotproduct"] = "cosine")
创建一个新的 PineconeDocumentStore 实例。
它旨在连接到 Pinecone 索引和命名空间。
参数:
api_key: Pinecone API 密钥。index: 要连接的 Pinecone 索引。如果索引不存在，则会创建它。namespace: 要连接的 Pinecone 命名空间。如果命名空间不存在，则会在第一次写入时创建。batch_size: 单个批次写入的文档数量。设置此参数时，请考虑 文档中规定的 Pinecone 限制。dimension: 嵌入的维度。此参数仅在创建新索引时使用。spec: 创建新索引时使用的 Pinecone spec。允许在无服务器和 pod 部署选项之间进行选择，并设置附加参数。有关更多详细信息，请参阅 Pinecone 文档。如果未提供，将使用在us-east-1区域中具有无服务器部署的默认 spec（与免费套餐兼容）。metric: 用于相似性搜索的度量。此参数仅在创建新索引时使用。
PineconeDocumentStore.close
def close()
关闭相关的同步资源。
PineconeDocumentStore.close_async
async def close_async()
关闭相关的异步资源。当不再需要文档存储时，应手动调用。
PineconeDocumentStore.from_dict
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "PineconeDocumentStore"
从字典反序列化组件。
参数:
data: 要反序列化的字典。
返回值:
反序列化后的组件。
PineconeDocumentStore.to_dict
def to_dict() -> Dict[str, Any]
将组件序列化为字典。
返回值:
包含序列化数据的字典。
PineconeDocumentStore.count_documents
def count_documents() -> int
返回文档存储中存在的文档数量。
PineconeDocumentStore.count_documents_async
async def count_documents_async() -> int
异步返回文档存储中存在的文档数量。
PineconeDocumentStore.write_documents
def write_documents(documents: List[Document],
policy: DuplicatePolicy = DuplicatePolicy.NONE) -> int
将文档写入 Pinecone。
参数:
documents: 要写入文档存储的文档列表。policy: 写入文档时使用的重复策略。PineconeDocumentStore 只支持DuplicatePolicy.OVERWRITE.
返回值:
写入文档存储的文档数量。
PineconeDocumentStore.write_documents_async
async def write_documents_async(
documents: List[Document],
policy: DuplicatePolicy = DuplicatePolicy.NONE) -> int
异步将文档写入 Pinecone。
参数:
documents: 要写入文档存储的文档列表。policy: 写入文档时使用的重复策略。PineconeDocumentStore 只支持DuplicatePolicy.OVERWRITE.
返回值:
写入文档存储的文档数量。
PineconeDocumentStore.filter_documents
def filter_documents(
filters: Optional[Dict[str, Any]] = None) -> List[Document]
返回与提供的过滤器匹配的文档。
有关过滤器的详细规格，请参阅 文档
参数:
filters: 要应用于文档列表的过滤器。
返回值:
与给定过滤器匹配的文档列表。
PineconeDocumentStore.filter_documents_async
async def filter_documents_async(
filters: Optional[Dict[str, Any]] = None) -> List[Document]
异步返回与提供的过滤器匹配的文档。
参数:
filters: 要应用于文档列表的过滤器。
返回值:
与给定过滤器匹配的文档列表。
PineconeDocumentStore.delete_documents
def delete_documents(document_ids: List[str]) -> None
删除与提供的document_ids 匹配的文档。
参数:
document_ids: 要删除的文档 ID
PineconeDocumentStore.delete_documents_async
async def delete_documents_async(document_ids: List[str]) -> None
异步删除与提供的document_ids 匹配的文档。
参数:
document_ids: 要删除的文档 ID