模块 auto_merging_retriever

AutoMergingRetriever

一个基于阈值设置，返回匹配叶节点文档的父文档的检索器。

AutoMergingRetriever 假定您有一个分层文档树结构，其中叶节点存储在文档存储中。有关如何创建此类结构的更多信息，请参阅 HierarchicalDocumentSplitter。在检索过程中，如果同一父节点下匹配的叶文档数量高于定义的阈值，检索器将返回父文档而不是单个叶文档。

其原理是，如果一个段落被分成多个表示为叶文档的块，并且对于给定的查询，匹配到多个块，那么整个段落可能比单个块本身提供更多信息。

目前 AutoMergingRetriever 只能用于以下 DocumentStore：

from haystack import Document
from haystack.components.preprocessors import HierarchicalDocumentSplitter
from haystack.components.retrievers.auto_merging_retriever import AutoMergingRetriever
from haystack.document_stores.in_memory import InMemoryDocumentStore

# create a hierarchical document structure with 3 levels, where the parent document has 3 children
text = "The sun rose early in the morning. It cast a warm glow over the trees. Birds began to sing."
original_document = Document(content=text)
builder = HierarchicalDocumentSplitter(block_sizes=[10, 3], split_overlap=0, split_by="word")
docs = builder.run([original_document])["documents"]

# store level-1 parent documents and initialize the retriever
doc_store_parents = InMemoryDocumentStore()
for doc in docs["documents"]:
    if doc.meta["children_ids"] and doc.meta["level"] == 1:
        doc_store_parents.write_documents([doc])
retriever = AutoMergingRetriever(doc_store_parents, threshold=0.5)

# assume we retrieved 2 leaf docs from the same parent, the parent document should be returned,
# since it has 3 children and the threshold=0.5, and we retrieved 2 children (2/3 > 0.66(6))
leaf_docs = [doc for doc in docs["documents"] if not doc.meta["children_ids"]]
docs = retriever.run(leaf_docs[4:6])
>> {'documents': [Document(id=538..),
>> content: 'warm glow over the trees. Birds began to sing.',
>> meta: {'block_size': 10, 'parent_id': '835..', 'children_ids': ['c17...', '3ff...', '352...'], 'level': 1, 'source_id': '835...',
>> 'page_number': 1, 'split_id': 1, 'split_idx_start': 45})]}

AutoMergingRetriever.init

def __init__(document_store: DocumentStore, threshold: float = 0.5)

初始化 AutoMergingRetriever。

参数:

document_store: 用于检索父文档的 DocumentStore
threshold: 用于决定是返回父文档还是单个文档的阈值

AutoMergingRetriever.to_dict

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

将组件序列化为字典。

返回值:

包含序列化数据的字典。

AutoMergingRetriever.from_dict

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

从字典反序列化组件。

参数:

data: 包含序列化数据的字典。

返回值:

组件的实例。

AutoMergingRetriever.run

@component.output_types(documents=list[Document])
def run(documents: list[Document])

运行 AutoMergingRetriever。

递归地按父级对文档进行分组，并在满足阈值时合并它们，继续向上层级，直到无法再进行合并。

参数:

documents: 被检索器匹配的叶文档列表

返回值:

文档列表（可以是不同层级的混合）

模块 in_memory/bm25_retriever

InMemoryBM25Retriever

使用基于关键字的算法检索与查询最相似的文档。

将此检索器与 InMemoryDocumentStore 一起使用。

使用示例

from haystack import Document
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.document_stores.in_memory import InMemoryDocumentStore

docs = [
    Document(content="Python is a popular programming language"),
    Document(content="python ist eine beliebte Programmiersprache"),
]

doc_store = InMemoryDocumentStore()
doc_store.write_documents(docs)
retriever = InMemoryBM25Retriever(doc_store)

result = retriever.run(query="Programmiersprache")

print(result["documents"])

InMemoryBM25Retriever.init

def __init__(document_store: InMemoryDocumentStore,
             filters: Optional[dict[str, Any]] = None,
             top_k: int = 10,
             scale_score: bool = False,
             filter_policy: FilterPolicy = FilterPolicy.REPLACE)

创建 InMemoryBM25Retriever 组件。

参数:

document_store: InMemoryDocumentStore 的一个实例，检索器应在此处搜索相关文档。
filters: 一个包含过滤器字典，用于缩小检索器在文档存储中的搜索范围。
top_k: 要检索的最大文档数。
scale_score: 当True 时，将检索到的文档得分缩放至 0 到 1 的范围，其中 1 表示极度相关。当False 时，使用原始相似度得分。
filter_policy: 检索时要应用的过滤策略。过滤策略决定了检索文档时如何应用过滤器。您可以选择
REPLACE（默认）：使用运行时指定的过滤器覆盖初始化过滤器。使用此策略可以动态更改特定查询的过滤。
MERGE: 将运行时过滤器与初始化过滤器结合起来，以缩小搜索范围。

引发:

ValueError: 如果指定的top_k 不大于 0。

InMemoryBM25Retriever.to_dict

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

将组件序列化为字典。

返回值:

包含序列化数据的字典。

InMemoryBM25Retriever.from_dict

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

从字典反序列化组件。

参数:

data: 要反序列化的字典。

返回值:

反序列化后的组件。

InMemoryBM25Retriever.run

@component.output_types(documents=list[Document])
def run(query: str,
        filters: Optional[dict[str, Any]] = None,
        top_k: Optional[int] = None,
        scale_score: Optional[bool] = None)

在给定的输入数据上运行 InMemoryBM25Retriever。

参数:

query: 检索器的查询字符串。
filters: 一个包含过滤器字典，用于缩小检索文档时的搜索范围。
top_k: 要返回的最大文档数。
scale_score: 当True 时，将检索到的文档得分缩放至 0 到 1 的范围，其中 1 表示极度相关。当False 时，使用原始相似度得分。

引发:

ValueError: 如果找不到指定的 DocumentStore 或它不是 InMemoryDocumentStore 实例。

返回值:

检索到的文档。

InMemoryBM25Retriever.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,
                    scale_score: Optional[bool] = None)

在给定的输入数据上运行 InMemoryBM25Retriever。

参数:

query: 检索器的查询字符串。
filters: 一个包含过滤器字典，用于缩小检索文档时的搜索范围。
top_k: 要返回的最大文档数。
scale_score: 当True 时，将检索到的文档得分缩放至 0 到 1 的范围，其中 1 表示极度相关。当False 时，使用原始相似度得分。

引发:

ValueError: 如果找不到指定的 DocumentStore 或它不是 InMemoryDocumentStore 实例。

返回值:

检索到的文档。

模块 in_memory/embedding_retriever

InMemoryEmbeddingRetriever

检索与查询语义最相似的文档。

将此检索器与 InMemoryDocumentStore 一起使用。

使用此检索器时，请确保它具有查询和文档嵌入。在索引管道中，使用 DocumentEmbedder 嵌入文档。在查询管道中，使用 TextEmbedder 嵌入查询并将其发送给检索器。

使用示例

from haystack import Document
from haystack.components.embedders import SentenceTransformersDocumentEmbedder, SentenceTransformersTextEmbedder
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.document_stores.in_memory import InMemoryDocumentStore

docs = [
    Document(content="Python is a popular programming language"),
    Document(content="python ist eine beliebte Programmiersprache"),
]
doc_embedder = SentenceTransformersDocumentEmbedder()
doc_embedder.warm_up()
docs_with_embeddings = doc_embedder.run(docs)["documents"]

doc_store = InMemoryDocumentStore()
doc_store.write_documents(docs_with_embeddings)
retriever = InMemoryEmbeddingRetriever(doc_store)

query="Programmiersprache"
text_embedder = SentenceTransformersTextEmbedder()
text_embedder.warm_up()
query_embedding = text_embedder.run(query)["embedding"]

result = retriever.run(query_embedding=query_embedding)

print(result["documents"])

InMemoryEmbeddingRetriever.init

def __init__(document_store: InMemoryDocumentStore,
             filters: Optional[dict[str, Any]] = None,
             top_k: int = 10,
             scale_score: bool = False,
             return_embedding: bool = False,
             filter_policy: FilterPolicy = FilterPolicy.REPLACE)

创建 InMemoryEmbeddingRetriever 组件。

参数:

document_store: InMemoryDocumentStore 的一个实例，检索器应在此处搜索相关文档。
filters: 一个包含过滤器字典，用于缩小检索器在文档存储中的搜索范围。
top_k: 要检索的最大文档数。
scale_score: 当True 时，将检索到的文档得分缩放至 0 到 1 的范围，其中 1 表示极度相关。当False 时，使用原始相似度得分。
return_embedding: 当True 时，返回检索到的文档的嵌入。当False 时，只返回文档，不返回它们的嵌入。
filter_policy: 检索时要应用的过滤策略。过滤策略决定了检索文档时如何应用过滤器。您可以选择
REPLACE（默认）：使用运行时指定的过滤器覆盖初始化过滤器。使用此策略可以动态更改特定查询的过滤。
MERGE: 将运行时过滤器与初始化过滤器结合起来，以缩小搜索范围。

引发:

ValueError: 如果指定的 top_k 不大于 0。

InMemoryEmbeddingRetriever.to_dict

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

将组件序列化为字典。

返回值:

包含序列化数据的字典。

InMemoryEmbeddingRetriever.from_dict

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

从字典反序列化组件。

参数:

data: 要反序列化的字典。

返回值:

反序列化后的组件。

InMemoryEmbeddingRetriever.run

@component.output_types(documents=list[Document])
def run(query_embedding: list[float],
        filters: Optional[dict[str, Any]] = None,
        top_k: Optional[int] = None,
        scale_score: Optional[bool] = None,
        return_embedding: Optional[bool] = None)

在给定的输入数据上运行 InMemoryEmbeddingRetriever。

参数:

query_embedding: 查询的嵌入。
filters: 一个包含过滤器字典，用于缩小检索文档时的搜索范围。
top_k: 要返回的最大文档数。
scale_score: 当True 时，将检索到的文档得分缩放至 0 到 1 的范围，其中 1 表示极度相关。当False 时，使用原始相似度得分。
return_embedding: 当True 时，返回检索到的文档的嵌入。当False 时，只返回文档，不返回它们的嵌入。

引发:

ValueError: 如果找不到指定的 DocumentStore 或它不是 InMemoryDocumentStore 实例。

返回值:

检索到的文档。

InMemoryEmbeddingRetriever.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,
                    scale_score: Optional[bool] = None,
                    return_embedding: Optional[bool] = None)

在给定的输入数据上运行 InMemoryEmbeddingRetriever。

参数:

query_embedding: 查询的嵌入。
filters: 一个包含过滤器字典，用于缩小检索文档时的搜索范围。
top_k: 要返回的最大文档数。
scale_score: 当True 时，将检索到的文档得分缩放至 0 到 1 的范围，其中 1 表示极度相关。当False 时，使用原始相似度得分。
return_embedding: 当True 时，返回检索到的文档的嵌入。当False 时，只返回文档，不返回它们的嵌入。

引发:

ValueError: 如果找不到指定的 DocumentStore 或它不是 InMemoryDocumentStore 实例。

返回值:

检索到的文档。

模块 filter_retriever

FilterRetriever

检索与提供的过滤器匹配的文档。

使用示例

from haystack import Document
from haystack.components.retrievers import FilterRetriever
from haystack.document_stores.in_memory import InMemoryDocumentStore

docs = [
    Document(content="Python is a popular programming language", meta={"lang": "en"}),
    Document(content="python ist eine beliebte Programmiersprache", meta={"lang": "de"}),
]

doc_store = InMemoryDocumentStore()
doc_store.write_documents(docs)
retriever = FilterRetriever(doc_store, filters={"field": "lang", "operator": "==", "value": "en"})

# if passed in the run method, filters override those provided at initialization
result = retriever.run(filters={"field": "lang", "operator": "==", "value": "de"})

print(result["documents"])

FilterRetriever.init

def __init__(document_store: DocumentStore,
             filters: Optional[dict[str, Any]] = None)

创建 FilterRetriever 组件。

参数:

document_store: 要与检索器一起使用的 Document Store 实例。
filters：一个包含过滤器的字典，用于缩小搜索范围。

FilterRetriever.to_dict

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

将组件序列化为字典。

返回值:

包含序列化数据的字典。

FilterRetriever.from_dict

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

从字典反序列化组件。

参数:

data: 要反序列化的字典。

返回值:

反序列化后的组件。

FilterRetriever.run

@component.output_types(documents=list[Document])
def run(filters: Optional[dict[str, Any]] = None)

在给定的输入数据上运行 FilterRetriever。

参数:

filters: 一个包含过滤器字典，用于缩小搜索空间。如果未指定，FilterRetriever 将使用初始化时提供的值。

返回值:

检索到的文档列表。

模块 sentence_window_retriever

SentenceWindowRetriever

从文档存储中检索相邻文档，为查询结果提供上下文。

此组件旨在在检索器（例如 BM25Retriever、EmbeddingRetriever）之后使用。它通过获取相邻的文档块来增强检索结果，为用户提供额外的上下文。

文档必须包含指示其来源和位置的元数据

source_id 用于将属于同一原始文档的句子块分组。
split_id 表示块在文档中的位置/顺序。

可以使用window_size 参数配置在检索到的文档两侧包含的相邻文档数量。您还可以通过source_id_meta_field 和split_id_meta_field.

SentenceWindowRetriever 兼容以下 DocumentStore：

使用示例

from haystack import Document, Pipeline
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.components.retrievers import SentenceWindowRetriever
from haystack.components.preprocessors import DocumentSplitter
from haystack.document_stores.in_memory import InMemoryDocumentStore

splitter = DocumentSplitter(split_length=10, split_overlap=5, split_by="word")
text = (
        "This is a text with some words. There is a second sentence. And there is also a third sentence. "
        "It also contains a fourth sentence. And a fifth sentence. And a sixth sentence. And a seventh sentence"
)
doc = Document(content=text)
docs = splitter.run([doc])
doc_store = InMemoryDocumentStore()
doc_store.write_documents(docs["documents"])


rag = Pipeline()
rag.add_component("bm25_retriever", InMemoryBM25Retriever(doc_store, top_k=1))
rag.add_component("sentence_window_retriever", SentenceWindowRetriever(document_store=doc_store, window_size=2))
rag.connect("bm25_retriever", "sentence_window_retriever")

rag.run({'bm25_retriever': {"query":"third"}})

>> {'sentence_window_retriever': {'context_windows': ['some words. There is a second sentence.
>> And there is also a third sentence. It also contains a fourth sentence. And a fifth sentence. And a sixth
>> sentence. And a'], 'context_documents': [[Document(id=..., content: 'some words. There is a second sentence.
>> And there is ', meta: {'source_id': '...', 'page_number': 1, 'split_id': 1, 'split_idx_start': 20,
>> '_split_overlap': [{'doc_id': '...', 'range': (20, 43)}, {'doc_id': '...', 'range': (0, 30)}]}),
>> Document(id=..., content: 'second sentence. And there is also a third sentence. It ',
>> meta: {'source_id': '74ea87deb38012873cf8c07e...f19d01a26a098447113e1d7b83efd30c02987114', 'page_number': 1,
>> 'split_id': 2, 'split_idx_start': 43, '_split_overlap': [{'doc_id': '...', 'range': (23, 53)}, {'doc_id': '...',
>> 'range': (0, 26)}]}), Document(id=..., content: 'also a third sentence. It also contains a fourth sentence. ',
>> meta: {'source_id': '...', 'page_number': 1, 'split_id': 3, 'split_idx_start': 73, '_split_overlap':
>> [{'doc_id': '...', 'range': (30, 56)}, {'doc_id': '...', 'range': (0, 33)}]}), Document(id=..., content:
>> 'also contains a fourth sentence. And a fifth sentence. And ', meta: {'source_id': '...', 'page_number': 1,
>> 'split_id': 4, 'split_idx_start': 99, '_split_overlap': [{'doc_id': '...', 'range': (26, 59)},
>> {'doc_id': '...', 'range': (0, 26)}]}), Document(id=..., content: 'And a fifth sentence. And a sixth sentence.
>> And a ', meta: {'source_id': '...', 'page_number': 1, 'split_id': 5, 'split_idx_start': 132,
>> '_split_overlap': [{'doc_id': '...', 'range': (33, 59)}, {'doc_id': '...', 'range': (0, 24)}]})]]}}}}

SentenceWindowRetriever.init

def __init__(document_store: DocumentStore,
             window_size: int = 3,
             *,
             source_id_meta_field: Union[str, list[str]] = "source_id",
             split_id_meta_field: str = "split_id",
             raise_on_missing_meta_fields: bool = True)

创建一个新的 SentenceWindowRetriever 组件。

参数:

document_store: 用于检索周围文档的 Document Store。
window_size: 在相关文档前后检索的文档数量。例如，window_size: 2 提取 2 个前置文档和 2 个后置文档。
source_id_meta_field: 包含文档源 ID 的元数据字段。这可以是一个字段或一个字段列表。如果提供了多个字段，如果所有字段都匹配，则检索器将认为文档属于同一来源。
split_id_meta_field: 包含文档拆分 ID 的元数据字段。
raise_on_missing_meta_fields: 如果为 True，则如果文档不包含所需的元数据字段，则会引发错误。如果为 False，则会跳过检索缺少所需元数据字段的文档的上下文，但仍会将原始文档包含在结果中。

SentenceWindowRetriever.merge_documents_text

@staticmethod
def merge_documents_text(documents: list[Document]) -> str

将文档文本列表合并为单个字符串。

此函数将文档列表的文本内容连接成单个字符串，消除任何重叠内容。

参数:

documents: 要合并的文档列表。

SentenceWindowRetriever.to_dict

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

将组件序列化为字典。

返回值:

包含序列化数据的字典。

SentenceWindowRetriever.from_dict

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

从字典反序列化组件。

返回值:

反序列化后的组件。

SentenceWindowRetriever.run

@component.output_types(context_windows=list[str],
                        context_documents=list[Document])
def run(retrieved_documents: list[Document],
        window_size: Optional[int] = None)

基于source_id 和doc.meta['split_id'] 从文档存储中获取周围的文档。

实现句子窗口技术背后的逻辑，从文档存储中检索给定文档的周围文档。

参数:

retrieved_documents: 从上一个检索器检索到的文档列表。
window_size: 在相关文档前后检索的文档数量。这将覆盖构造函数中设置的window_size 参数。

返回值:

包含以下键的字典

context_windows: 一个字符串列表，其中每个字符串表示retrieved_documents.
context_documents: 一个列表Document 对象，包含检索到的文档以及它们周围的上下文文档。文档按split_idx_start 元字段排序。

模块 auto_merging_retriever

AutoMergingRetriever

AutoMergingRetriever.__init__

AutoMergingRetriever.to_dict

AutoMergingRetriever.from_dict

AutoMergingRetriever.run

模块 in_memory/bm25_retriever

InMemoryBM25Retriever

使用示例

InMemoryBM25Retriever.__init__

InMemoryBM25Retriever.to_dict

InMemoryBM25Retriever.from_dict

InMemoryBM25Retriever.run

InMemoryBM25Retriever.run_async

模块 in_memory/embedding_retriever

InMemoryEmbeddingRetriever

使用示例

InMemoryEmbeddingRetriever.__init__

InMemoryEmbeddingRetriever.to_dict

InMemoryEmbeddingRetriever.from_dict

InMemoryEmbeddingRetriever.run

InMemoryEmbeddingRetriever.run_async

模块 filter_retriever

FilterRetriever

使用示例

FilterRetriever.__init__

FilterRetriever.to_dict

FilterRetriever.from_dict

FilterRetriever.run

模块 sentence_window_retriever

SentenceWindowRetriever

使用示例

SentenceWindowRetriever.__init__

SentenceWindowRetriever.merge_documents_text

SentenceWindowRetriever.to_dict

SentenceWindowRetriever.from_dict

SentenceWindowRetriever.run

AutoMergingRetriever.init

InMemoryBM25Retriever.init

InMemoryEmbeddingRetriever.init

FilterRetriever.init

SentenceWindowRetriever.init