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

FastEmbed

Haystack 的 FastEmbed 集成

模块 haystack_integrations.components.embedders.fastembed.fastembed_document_embedder

FastembedDocumentEmbedder

FastembedDocumentEmbedder 使用 Fastembed 嵌入模型计算 Document 嵌入。每个 Document 的嵌入存储在Document 的 embedding 字段中。

使用示例

# To use this component, install the "fastembed-haystack" package.
# pip install fastembed-haystack

from haystack_integrations.components.embedders.fastembed import FastembedDocumentEmbedder
from haystack.dataclasses import Document

doc_embedder = FastembedDocumentEmbedder(
    model="BAAI/bge-small-en-v1.5",
    batch_size=256,
)

doc_embedder.warm_up()

# Text taken from PubMed QA Dataset (https://hugging-face.cn/datasets/pubmed_qa)
document_list = [
    Document(
        content=("Oxidative stress generated within inflammatory joints can produce autoimmune phenomena and joint "
                 "destruction. Radical species with oxidative activity, including reactive nitrogen species, "
                 "represent mediators of inflammation and cartilage damage."),
        meta={
            "pubid": "25,445,628",
            "long_answer": "yes",
        },
    ),
    Document(
        content=("Plasma levels of pancreatic polypeptide (PP) rise upon food intake. Although other pancreatic "
                 "islet hormones, such as insulin and glucagon, have been extensively investigated, PP secretion "
                 "and actions are still poorly understood."),
        meta={
            "pubid": "25,445,712",
            "long_answer": "yes",
        },
    ),
]

result = doc_embedder.run(document_list)
print(f"Document Text: {result['documents'][0].content}")
print(f"Document Embedding: {result['documents'][0].embedding}")
print(f"Embedding Dimension: {len(result['documents'][0].embedding)}")

FastembedDocumentEmbedder.__init__

def __init__(model: str = "BAAI/bge-small-en-v1.5",
             cache_dir: Optional[str] = None,
             threads: Optional[int] = None,
             prefix: str = "",
             suffix: str = "",
             batch_size: int = 256,
             progress_bar: bool = True,
             parallel: Optional[int] = None,
             local_files_only: bool = False,
             meta_fields_to_embed: Optional[List[str]] = None,
             embedding_separator: str = "\n")

创建一个 FastembedDocumentEmbedder 组件。

参数:

  • model：在 Hugging Face 模型中心中模型的本地路径或名称，例如BAAI/bge-small-en-v1.5.
  • cache_dir: 缓存目录的路径。可以通过FASTEMBED_CACHE_PATH 环境变量设置。默认为系统临时目录中的 fastembed_cache。
  • threads: 单个 onnxruntime 会话可以使用线程数。默认为 None。
  • prefix: 添加到每个文本开头的字符串。
  • suffix: 添加到每个文本末尾的字符串。
  • batch_size: 一次编码的字符串数量。
  • progress_bar: 如果True，在嵌入过程中显示进度条。
  • parallel: 如果 > 1，将使用数据并行编码，推荐用于大型数据集的离线编码。如果为 0，则使用所有可用核心。如果为 None，则不使用数据并行处理，而是使用默认的 onnxruntime 线程。
  • local_files_only: 如果为True，仅使用 cache_dir 中的模型文件cache_dir.
  • meta_fields_to_embed：应与 Document 内容一起嵌入的元字段列表。
  • embedding_separator：用于将元字段连接到 Document 内容的分隔符。

FastembedDocumentEmbedder.to_dict

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

将组件序列化为字典。

返回值:

包含序列化数据的字典。

FastembedDocumentEmbedder.warm_up

def warm_up()

Initializes the component.

FastembedDocumentEmbedder.run

@component.output_types(documents=List[Document])
def run(documents: List[Document]) -> Dict[str, List[Document]]

嵌入 Document 列表。

参数:

  • documents: 要嵌入的 Document 列表。

返回值:

包含以下键的字典

  • documents: Document 列表，其中每个 Document 的embedding 字段设置为计算出的嵌入。

模块 haystack_integrations.components.embedders.fastembed.fastembed_text_embedder

FastembedTextEmbedder

FastembedTextEmbedder 使用 fastembed 嵌入模型计算字符串嵌入。

使用示例

from haystack_integrations.components.embedders.fastembed import FastembedTextEmbedder

text = ("It clearly says online this will work on a Mac OS system. "
        "The disk comes and it does not, only Windows. Do Not order this if you have a Mac!!")

text_embedder = FastembedTextEmbedder(
    model="BAAI/bge-small-en-v1.5"
)
text_embedder.warm_up()

embedding = text_embedder.run(text)["embedding"]

FastembedTextEmbedder.__init__

def __init__(model: str = "BAAI/bge-small-en-v1.5",
             cache_dir: Optional[str] = None,
             threads: Optional[int] = None,
             prefix: str = "",
             suffix: str = "",
             progress_bar: bool = True,
             parallel: Optional[int] = None,
             local_files_only: bool = False)

创建一个 FastembedTextEmbedder 组件。

参数:

  • model: Fastembed 模型中心的本地路径或模型名称，例如BAAI/bge-small-en-v1.5
  • cache_dir: 缓存目录的路径。可以通过FASTEMBED_CACHE_PATH 环境变量设置。默认为系统临时目录中的 fastembed_cache。
  • threads: 单个 onnxruntime 会话可以使用线程数。默认为 None。
  • prefix: 添加到每个文本开头的字符串。
  • suffix: 添加到每个文本末尾的字符串。
  • progress_bar: 如果True，在嵌入过程中显示进度条。
  • parallel: 如果 > 1，将使用数据并行编码，推荐用于大型数据集的离线编码。如果为 0，则使用所有可用核心。如果为 None，则不使用数据并行处理，而是使用默认的 onnxruntime 线程。
  • local_files_only: 如果为True，仅使用 cache_dir 中的模型文件cache_dir.

FastembedTextEmbedder.to_dict

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

将组件序列化为字典。

返回值:

包含序列化数据的字典。

FastembedTextEmbedder.warm_up

def warm_up()

Initializes the component.

FastembedTextEmbedder.run

@component.output_types(embedding=List[float])
def run(text: str) -> Dict[str, List[float]]

使用 Fastembed 模型嵌入文本。

参数:

  • text: 要嵌入的字符串。

引发:

  • TypeError: 如果输入不是字符串。
  • RuntimeError: 如果嵌入模型尚未加载。

返回值:

包含以下键的字典

  • embedding: 一个浮点数列表，表示输入文本的嵌入。

模块 haystack_integrations.components.embedders.fastembed.fastembed_sparse_document_embedder

FastembedSparseDocumentEmbedder

FastembedSparseDocumentEmbedder 使用 Fastembed 稀疏模型计算 Document 嵌入。

使用示例

from haystack_integrations.components.embedders.fastembed import FastembedSparseDocumentEmbedder
from haystack.dataclasses import Document

sparse_doc_embedder = FastembedSparseDocumentEmbedder(
    model="prithivida/Splade_PP_en_v1",
    batch_size=32,
)

sparse_doc_embedder.warm_up()

# Text taken from PubMed QA Dataset (https://hugging-face.cn/datasets/pubmed_qa)
document_list = [
    Document(
        content=("Oxidative stress generated within inflammatory joints can produce autoimmune phenomena and joint "
                 "destruction. Radical species with oxidative activity, including reactive nitrogen species, "
                 "represent mediators of inflammation and cartilage damage."),
        meta={
            "pubid": "25,445,628",
            "long_answer": "yes",
        },
    ),
    Document(
        content=("Plasma levels of pancreatic polypeptide (PP) rise upon food intake. Although other pancreatic "
                 "islet hormones, such as insulin and glucagon, have been extensively investigated, PP secretion "
                 "and actions are still poorly understood."),
        meta={
            "pubid": "25,445,712",
            "long_answer": "yes",
        },
    ),
]

result = sparse_doc_embedder.run(document_list)
print(f"Document Text: {result['documents'][0].content}")
print(f"Document Sparse Embedding: {result['documents'][0].sparse_embedding}")
print(f"Sparse Embedding Dimension: {len(result['documents'][0].sparse_embedding)}")

FastembedSparseDocumentEmbedder.__init__

def __init__(model: str = "prithivida/Splade_PP_en_v1",
             cache_dir: Optional[str] = None,
             threads: Optional[int] = None,
             batch_size: int = 32,
             progress_bar: bool = True,
             parallel: Optional[int] = None,
             local_files_only: bool = False,
             meta_fields_to_embed: Optional[List[str]] = None,
             embedding_separator: str = "\n",
             model_kwargs: Optional[Dict[str, Any]] = None)

创建一个 FastembedDocumentEmbedder 组件。

参数:

  • model：在 Hugging Face 模型中心中模型的本地路径或名称，例如prithivida/Splade_PP_en_v1.
  • cache_dir: 缓存目录的路径。可以通过FASTEMBED_CACHE_PATH 环境变量设置。默认为系统临时目录中的 fastembed_cache。
  • threads: 单个 onnxruntime 会话可以使用线程数。
  • batch_size: 一次编码的字符串数量。
  • progress_bar: 如果True，在嵌入过程中显示进度条。
  • parallel: 如果 > 1，将使用数据并行编码，推荐用于大型数据集的离线编码。如果为 0，则使用所有可用核心。如果为 None，则不使用数据并行处理，而是使用默认的 onnxruntime 线程。
  • local_files_only: 如果为True，仅使用 cache_dir 中的模型文件cache_dir.
  • meta_fields_to_embed：应与 Document 内容一起嵌入的元字段列表。
  • embedding_separator：用于将元字段连接到 Document 内容的分隔符。
  • model_kwargs: 包含模型参数的字典，例如k, b, avg_len, language.

FastembedSparseDocumentEmbedder.to_dict

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

将组件序列化为字典。

返回值:

包含序列化数据的字典。

FastembedSparseDocumentEmbedder.warm_up

def warm_up()

Initializes the component.

FastembedSparseDocumentEmbedder.run

@component.output_types(documents=List[Document])
def run(documents: List[Document]) -> Dict[str, List[Document]]

嵌入 Document 列表。

参数:

  • documents: 要嵌入的 Document 列表。

返回值:

包含以下键的字典

  • documents: Document 列表，其中每个 Document 的sparse_embedding 字段设置为计算出的嵌入。

模块 haystack_integrations.components.embedders.fastembed.fastembed_sparse_text_embedder

FastembedSparseTextEmbedder

FastembedSparseTextEmbedder 使用 fastembed 稀疏模型计算字符串嵌入。

使用示例

from haystack_integrations.components.embedders.fastembed import FastembedSparseTextEmbedder

text = ("It clearly says online this will work on a Mac OS system. "
        "The disk comes and it does not, only Windows. Do Not order this if you have a Mac!!")

sparse_text_embedder = FastembedSparseTextEmbedder(
    model="prithivida/Splade_PP_en_v1"
)
sparse_text_embedder.warm_up()

sparse_embedding = sparse_text_embedder.run(text)["sparse_embedding"]

FastembedSparseTextEmbedder.__init__

def __init__(model: str = "prithivida/Splade_PP_en_v1",
             cache_dir: Optional[str] = None,
             threads: Optional[int] = None,
             progress_bar: bool = True,
             parallel: Optional[int] = None,
             local_files_only: bool = False,
             model_kwargs: Optional[Dict[str, Any]] = None)

创建一个 FastembedSparseTextEmbedder 组件。

参数:

  • model: Fastembed 模型中心的本地路径或模型名称，例如prithivida/Splade_PP_en_v1
  • cache_dir: 缓存目录的路径。可以通过FASTEMBED_CACHE_PATH 环境变量设置。默认为系统临时目录中的 fastembed_cache。
  • threads: 单个 onnxruntime 会话可以使用线程数。默认为 None。
  • progress_bar: 如果True，在嵌入过程中显示进度条。
  • parallel: 如果 > 1，将使用数据并行编码，推荐用于大型数据集的离线编码。如果为 0，则使用所有可用核心。如果为 None，则不使用数据并行处理，而是使用默认的 onnxruntime 线程。
  • local_files_only: 如果为True，仅使用 cache_dir 中的模型文件cache_dir.
  • model_kwargs: 包含模型参数的字典，例如k, b, avg_len, language.

FastembedSparseTextEmbedder.to_dict

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

将组件序列化为字典。

返回值:

包含序列化数据的字典。

FastembedSparseTextEmbedder.warm_up

def warm_up()

Initializes the component.

FastembedSparseTextEmbedder.run

@component.output_types(sparse_embedding=SparseEmbedding)
def run(text: str) -> Dict[str, SparseEmbedding]

使用 Fastembed 模型嵌入文本。

参数:

  • text: 要嵌入的字符串。

引发:

  • TypeError: 如果输入不是字符串。
  • RuntimeError: 如果嵌入模型尚未加载。

返回值:

包含以下键的字典

  • embedding: 一个浮点数列表，表示输入文本的嵌入。

模块 haystack_integrations.components.rankers.fastembed.ranker

FastembedRanker

使用 Fastembed 模型 根据查询的相似度对 Document 进行排名。

Documents 按与查询的语义相关性从高到低排序。

使用示例

from haystack import Document
from haystack_integrations.components.rankers.fastembed import FastembedRanker

ranker = FastembedRanker(model_name="Xenova/ms-marco-MiniLM-L-6-v2", top_k=2)

docs = [Document(content="Paris"), Document(content="Berlin")]
query = "What is the capital of germany?"
output = ranker.run(query=query, documents=docs)
print(output["documents"][0].content)

# Berlin

FastembedRanker.__init__

def __init__(model_name: str = "Xenova/ms-marco-MiniLM-L-6-v2",
             top_k: int = 10,
             cache_dir: Optional[str] = None,
             threads: Optional[int] = None,
             batch_size: int = 64,
             parallel: Optional[int] = None,
             local_files_only: bool = False,
             meta_fields_to_embed: Optional[List[str]] = None,
             meta_data_separator: str = "\n")

创建 'FastembedRanker' 的一个实例。

参数:

  • model_name: Fastembed 模型名称。请查看 Fastembed 文档 中支持的模型列表。
  • top_k: 要返回的最大文档数。
  • cache_dir: 缓存目录的路径。可以通过FASTEMBED_CACHE_PATH 环境变量设置。默认为系统临时目录中的 fastembed_cache。
  • threads: 单个 onnxruntime 会话可以使用线程数。默认为 None。
  • batch_size: 一次编码的字符串数量。
  • parallel: 如果 > 1，将使用数据并行编码，推荐用于大型数据集的离线编码。如果为 0，则使用所有可用核心。如果为 None，则不使用数据并行处理，而是使用默认的 onnxruntime 线程。
  • local_files_only: 如果为True，仅使用 cache_dir 中的模型文件cache_dir.
  • meta_fields_to_embed: 应与文档内容连接以进行重新排名的元字段列表。
  • meta_data_separator: 用于将元字段连接到 Document 内容的分隔符。

FastembedRanker.to_dict

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

将组件序列化为字典。

返回值:

包含序列化数据的字典。

FastembedRanker.from_dict

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

从字典反序列化组件。

参数:

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

返回值:

反序列化后的组件。

FastembedRanker.warm_up

def warm_up()

Initializes the component.

FastembedRanker.run

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

使用 FastEmbed 返回一个根据与给定查询的相似度排序的文档列表。

参数:

  • query: 用于比较文档的输入查询。
  • documents: 要排名的文档列表。
  • top_k: 要返回的最大文档数。

引发:

  • ValueError: 如果top_k 不大于 0。

返回值:

包含以下键的字典

  • documents: 最接近查询的文档列表，按相似度从高到低排序。