文档API 参考📓 教程🧑‍🍳 食谱🤝 集成💜 Discord🎨 Studio
API 参考

Chroma

Haystack 的 Chroma 集成

模块 haystack_integrations.components.retrievers.chroma.retriever

ChromaQueryTextRetriever

一个组件，用于通过 Chroma 数据库检索文档，使用query API。

示例用法

from haystack import Pipeline
from haystack.components.converters import TextFileToDocument
from haystack.components.writers import DocumentWriter

from haystack_integrations.document_stores.chroma import ChromaDocumentStore
from haystack_integrations.components.retrievers.chroma import ChromaQueryTextRetriever

file_paths = ...

# Chroma is used in-memory so we use the same instances in the two pipelines below
document_store = ChromaDocumentStore()

indexing = Pipeline()
indexing.add_component("converter", TextFileToDocument())
indexing.add_component("writer", DocumentWriter(document_store))
indexing.connect("converter", "writer")
indexing.run({"converter": {"sources": file_paths}})

querying = Pipeline()
querying.add_component("retriever", ChromaQueryTextRetriever(document_store))
results = querying.run({"retriever": {"query": "Variable declarations", "top_k": 3}})

for d in results["retriever"]["documents"]:
    print(d.meta, d.score)

ChromaQueryTextRetriever.__init__

def __init__(document_store: ChromaDocumentStore,
             filters: Optional[Dict[str, Any]] = None,
             top_k: int = 10,
             filter_policy: Union[str, FilterPolicy] = FilterPolicy.REPLACE)

参数:

  • document_store: 实例ChromaDocumentStore.
  • filters: 用于缩小搜索空间的过滤器。
  • top_k: 要检索的最大文档数量。
  • filter_policy: 确定如何应用过滤器的策略。

ChromaQueryTextRetriever.run

@component.output_types(documents=List[Document])
def run(query: str,
        filters: Optional[Dict[str, Any]] = None,
        top_k: Optional[int] = None) -> Dict[str, Any]

在给定的输入数据上运行检索器。

参数:

  • query: 检索器的输入数据。在这种情况下，是纯文本查询。
  • filters: 应用于检索到的文档的过滤器。运行时过滤器的应用方式取决于初始化检索器时选择的filter_policy。有关更多详细信息，请参阅 init 方法文档字符串。
  • top_k: 要检索的最大文档数量。如果未指定，则使用构造函数中的默认值。

引发:

  • ValueError: 如果指定的文档存储未找到或不是 MemoryDocumentStore 实例。

返回值:

包含以下键的字典

  • documents: 搜索引擎返回的文档列表。

ChromaQueryTextRetriever.run_async

@component.output_types(documents=List[Document])
async def run_async(query: str,
                    filters: Optional[Dict[str, Any]] = None,
                    top_k: Optional[int] = None) -> Dict[str, Any]

异步地在给定的输入数据上运行检索器。

异步方法仅支持 HTTP 连接。

参数:

  • query: 检索器的输入数据。在这种情况下，是纯文本查询。
  • filters: 应用于检索到的文档的过滤器。运行时过滤器的应用方式取决于初始化检索器时选择的filter_policy。有关更多详细信息，请参阅 init 方法文档字符串。
  • top_k: 要检索的最大文档数量。如果未指定，则使用构造函数中的默认值。

引发:

  • ValueError: 如果指定的文档存储未找到或不是 MemoryDocumentStore 实例。

返回值:

包含以下键的字典

  • documents: 搜索引擎返回的文档列表。

ChromaQueryTextRetriever.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "ChromaQueryTextRetriever"

从字典反序列化组件。

参数:

  • data: 要反序列化的字典。

返回值:

反序列化后的组件。

ChromaQueryTextRetriever.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

ChromaEmbeddingRetriever

一个组件，用于通过嵌入向量从 Chroma 数据库检索文档。

ChromaEmbeddingRetriever.__init__

def __init__(document_store: ChromaDocumentStore,
             filters: Optional[Dict[str, Any]] = None,
             top_k: int = 10,
             filter_policy: Union[str, FilterPolicy] = FilterPolicy.REPLACE)

参数:

  • document_store: 实例ChromaDocumentStore.
  • filters: 用于缩小搜索空间的过滤器。
  • top_k: 要检索的最大文档数量。
  • filter_policy: 确定如何应用过滤器的策略。

ChromaEmbeddingRetriever.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, Any]

在给定的输入数据上运行检索器。

参数:

  • query_embedding: 查询的嵌入向量。
  • filters: 应用于检索到的文档的过滤器。运行时过滤器的应用方式取决于初始化检索器时选择的filter_policy。有关更多详细信息，请参阅 init 方法文档字符串。
  • top_k: 要检索的最大文档数量。如果未指定，则使用构造函数中的默认值。

返回值:

一个包含以下键的字典

  • documents: 搜索引擎返回的文档列表。

ChromaEmbeddingRetriever.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, Any]

异步地在给定的输入数据上运行检索器。

异步方法仅支持 HTTP 连接。

参数:

  • query_embedding: 查询的嵌入向量。
  • filters: 应用于检索到的文档的过滤器。运行时过滤器的应用方式取决于初始化检索器时选择的filter_policy。有关更多详细信息，请参阅 init 方法文档字符串。
  • top_k: 要检索的最大文档数量。如果未指定，则使用构造函数中的默认值。

返回值:

一个包含以下键的字典

  • documents: 搜索引擎返回的文档列表。

ChromaEmbeddingRetriever.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "ChromaEmbeddingRetriever"

从字典反序列化组件。

参数:

  • data: 要反序列化的字典。

返回值:

反序列化后的组件。

ChromaEmbeddingRetriever.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

模块 haystack_integrations.document_stores.chroma.document_store

ChromaDocumentStore

一个以 Chroma 作为后端的文档存储。

我们将在一个检索增强生成（RAG）管道中使用collection.get API 来实现文档存储协议，而collection.search API 将在检索器中使用。

ChromaDocumentStore.__init__

def __init__(collection_name: str = "documents",
             embedding_function: str = "default",
             persist_path: Optional[str] = None,
             host: Optional[str] = None,
             port: Optional[int] = None,
             distance_function: Literal["l2", "cosine", "ip"] = "l2",
             metadata: Optional[dict] = None,
             **embedding_function_params: Any)

创建一个新的 ChromaDocumentStore 实例。

它旨在连接到 Chroma 集合。

注意：为了使组件成为可序列化管道的一部分，**init** 参数必须是可序列化的，因此我们使用注册表通过传递字符串来配置嵌入函数。

参数:

  • collection_name: 要在数据库中使用的集合的名称。
  • embedding_function: 用于嵌入查询的嵌入函数的名称
  • persist_path: 本地持久化存储的路径。不能与host 和port 结合使用。如果未指定persist_path, host 和port，则数据库将为内存中的.
  • host: 远程 Chroma HTTP 客户端连接的主机地址。不能与persist_path.
  • port: 远程 Chroma HTTP 客户端连接的端口号。不能与persist_path.
  • distance_function: 嵌入空间的距离度量。
  • "l2" 计算向量之间的欧氏（直线）距离，分数越低表示相似度越高。
  • "cosine" 计算向量之间的余弦相似度，分数越高表示相似度越高。
  • "ip" 代表内积，分数越高表示向量之间的相似度越高。 **注意**distance_function 只能在创建集合时设置。要更改现有集合的距离度量，请考虑克隆该集合。
  • metadata: 一个字典，包含直接传递给 chromadb 的 client 方法的 chromadb 集合参数create_collection。如果它包含键"hnsw:space"，则该值将优先于distance_function 参数。
  • embedding_function_params: 要传递给嵌入函数的附加参数。

ChromaDocumentStore.count_documents

def count_documents() -> int

返回文档存储中存在的文档数量。

返回值:

文档存储中存在的文档数量。

ChromaDocumentStore.count_documents_async

async def count_documents_async() -> int

异步返回文档存储中存在的文档数量。

异步方法仅支持 HTTP 连接。

返回值:

文档存储中存在的文档数量。

ChromaDocumentStore.filter_documents

def filter_documents(
        filters: Optional[Dict[str, Any]] = None) -> List[Document]

返回与提供的过滤器匹配的文档。

有关过滤器的详细规范，请参阅 文档。

参数:

  • filters: 要应用于文档列表的过滤器。

返回值:

匹配给定过滤器的文档列表。

ChromaDocumentStore.filter_documents_async

async def filter_documents_async(
        filters: Optional[Dict[str, Any]] = None) -> List[Document]

异步返回与提供的过滤器匹配的文档。

异步方法仅支持 HTTP 连接。

有关过滤器的详细规范，请参阅 文档。

参数:

  • filters: 要应用于文档列表的过滤器。

返回值:

匹配给定过滤器的文档列表。

ChromaDocumentStore.write_documents

def write_documents(documents: List[Document],
                    policy: DuplicatePolicy = DuplicatePolicy.FAIL) -> int

将文档写入（或覆盖）存储区。

参数:

  • documents: 要写入文档存储的文档列表。
  • policy: 目前不支持。

引发:

  • ValueError: 当输入无效时。

返回值:

写入的文档数量

ChromaDocumentStore.write_documents_async

async def write_documents_async(
        documents: List[Document],
        policy: DuplicatePolicy = DuplicatePolicy.FAIL) -> int

异步地将文档写入（或覆盖）存储区。

异步方法仅支持 HTTP 连接。

参数:

  • documents: 要写入文档存储的文档列表。
  • policy: 目前不支持。

引发:

  • ValueError: 当输入无效时。

返回值:

写入的文档数量

ChromaDocumentStore.delete_documents

def delete_documents(document_ids: List[str]) -> None

从文档存储中删除所有具有匹配 document_ids 的文档。

参数:

  • document_ids: 要删除的文档 ID

ChromaDocumentStore.delete_documents_async

async def delete_documents_async(document_ids: List[str]) -> None

异步地从文档存储中删除所有具有匹配 document_ids 的文档。

异步方法仅支持 HTTP 连接。

参数:

  • document_ids: 要删除的文档 ID

ChromaDocumentStore.search

def search(queries: List[str],
           top_k: int,
           filters: Optional[Dict[str, Any]] = None) -> List[List[Document]]

使用提供的文本查询搜索存储中的文档。

参数:

  • queries: 要搜索的查询列表。
  • top_k: 为每个查询返回的 top_k 文档。
  • filters: 要应用于搜索的过滤器字典。接受 haystack 格式的过滤器。

返回值:

匹配每个查询的文档。

ChromaDocumentStore.search_async

async def search_async(
        queries: List[str],
        top_k: int,
        filters: Optional[Dict[str, Any]] = None) -> List[List[Document]]

异步地使用提供的文本查询搜索存储中的文档。

异步方法仅支持 HTTP 连接。

参数:

  • queries: 要搜索的查询列表。
  • top_k: 为每个查询返回的 top_k 文档。
  • filters: 要应用于搜索的过滤器字典。接受 haystack 格式的过滤器。

返回值:

匹配每个查询的文档。

ChromaDocumentStore.search_embeddings

def search_embeddings(
        query_embeddings: List[List[float]],
        top_k: int,
        filters: Optional[Dict[str, Any]] = None) -> List[List[Document]]

对存储的文档执行向量搜索，传递查询的嵌入向量而不是它们的文本。

参数:

  • query_embeddings: 用作查询的嵌入向量列表。
  • top_k: 要检索的最大文档数量。
  • filters: 要应用于搜索的过滤器字典。接受 haystack 格式的过滤器。

返回值:

匹配给定过滤器的文档列表列表。

ChromaDocumentStore.search_embeddings_async

async def search_embeddings_async(
        query_embeddings: List[List[float]],
        top_k: int,
        filters: Optional[Dict[str, Any]] = None) -> List[List[Document]]

异步地对存储的文档执行向量搜索，传递查询的嵌入向量而不是

它们的文本。

异步方法仅支持 HTTP 连接。

参数:

  • query_embeddings: 用作查询的嵌入向量列表。
  • top_k: 要检索的最大文档数量。
  • filters: 要应用于搜索的过滤器字典。接受 haystack 格式的过滤器。

返回值:

匹配给定过滤器的文档列表列表。

ChromaDocumentStore.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "ChromaDocumentStore"

从字典反序列化组件。

参数:

  • data: 要反序列化的字典。

返回值:

反序列化后的组件。

ChromaDocumentStore.to_dict

def to_dict() -> Dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

模块 haystack_integrations.document_stores.chroma.errors

ChromaDocumentStoreError

ChromaDocumentStore 所有异常的父类。

ChromaDocumentStoreFilterError

当 ChromaDocumentStore 的过滤器无效时引发。

ChromaDocumentStoreConfigError

当 ChromaDocumentStore 的配置无效时引发。

模块 haystack_integrations.document_stores.chroma.utils

get_embedding_function

def get_embedding_function(function_name: str,
                           **kwargs: Any) -> EmbeddingFunction

按名称加载嵌入函数。

参数:

  • function_name: 嵌入函数的名称。
  • kwargs: 要传递给嵌入函数的附加参数。

引发:

  • ChromaDocumentStoreConfigError: 如果函数名称无效。

返回值:

加载的嵌入函数。