一、检索器（Retrievers）

检索器负责根据用户查询（或聊天消息）获取最相关的上下文。它可以构建在索引之上，但也可以独立定义。 它用作查询引擎（和聊天引擎）中检索相关上下文的关键构建块。它包含高级API和低级API。

高级API：可以通过retrieve_mode选择特定于索引的检索器类。 例如，使用 SummaryIndex：

retriever = summary_index.as_retriever(
    retriever_mode="llm",
    choice_batch_size=5,
)

低级API：如果需要更精细的控制，可以使用低级组合 API。为了达到与上面相同的结果，可以直接导入并构造所需的检索器类：

from llama_index.core.retrievers import SummaryIndexLLMRetriever

retriever = SummaryIndexLLMRetriever(
    index=summary_index,
    choice_batch_size=5,
)

retriever_mode包含以下内容：Vector Index、Summary Index、Tree Index、Keyword Table Index、Knowledge Graph Index、Document Summary Index

高级检索技术例如关键字/混合搜索、重新排名等。 有些是特定于 LLM + RAG 管道的，例如从小到大和自动合并检索。

1.定义自定义检索器

# import QueryBundle
from llama_index.core import QueryBundle

# import NodeWithScore
from llama_index.core.schema import NodeWithScore

# Retrievers
from llama_index.core.retrievers import (
    BaseRetriever,
    VectorIndexRetriever,
    KeywordTableSimpleRetriever,
)

from typing import List

class CustomRetriever(BaseRetriever):
    """Custom retriever that performs both semantic search and hybrid search."""

    def __init__(
        self,
        vector_retriever: VectorIndexRetriever,
        keyword_retriever: KeywordTableSimpleRetriever,
        mode: str = "AND",
    ) -> None:
        """Init params."""

        self._vector_retriever = vector_retriever
        self._keyword_retriever = keyword_retriever
        if mode not in ("AND", "OR"):
            raise ValueError("Invalid mode.")
        self._mode = mode
        super().__init__()

    def _retrieve(self, query_bundle: QueryBundle) -> List[NodeWithScore]:
        """Retrieve nodes given query."""

        vector_nodes = self._vector_retriever.retrieve(query_bundle)
        keyword_nodes = self._keyword_retriever.retrieve(query_bundle)

        vector_ids = {n.node.node_id for n in vector_nodes}
        keyword_ids = {n.node.node_id for n in keyword_nodes}

        combined_dict = {n.node.node_id: n for n in vector_nodes}
        combined_dict.update({n.node.node_id: n for n in keyword_nodes})

        if self._mode == "AND":
            retrieve_ids = vector_ids.intersection(keyword_ids)
        else:
            retrieve_ids = vector_ids.union(keyword_ids)

        retrieve_nodes = [combined_dict[rid] for rid in retrieve_ids]
        return retrieve_nodes

2.BM25 混合检索

RAG提效利器——BM25检索算法原理和Python实现

from llama_index.retrievers.bm25 import BM25Retriever
# We can pass in the index, doctore, or list of nodes to create the retriever
retriever = BM25Retriever.from_defaults(nodes=nodes, similarity_top_k=2)

3. 简单查询融合

在此步骤中，我们将索引融合到单个检索器中。 该检索器还将通过生成与原始问题相关的额外查询来增强我们的查询，并聚合结果。

此设置将查询 4 次，一次使用原始查询，然后再生成 3 次查询。

默认情况下，它使用以下提示来生成额外的查询：

from llama_index.core.retrievers import QueryFusionRetriever

retriever = QueryFusionRetriever(
    [index_1.as_retriever(), index_2.as_retriever()],
    similarity_top_k=2,
    num_queries=4,  # set this to 1 to disable query generation
    use_async=True,
    verbose=True,
    # query_gen_prompt="...",  # we could override the query generation prompt here
)

4. Reciprocal Rerank融合检索

检索到的节点将根据Reciprocal Rerank融合算法进行重排序。 它提供了一种有效的方法来对检索结果进行重新排序，而无需过多的计算或依赖外部模型。

from llama_index.core.retrievers import QueryFusionRetriever

retriever = QueryFusionRetriever(
    [vector_retriever, bm25_retriever],
    similarity_top_k=2,
    num_queries=4,  # set this to 1 to disable query generation
    mode="reciprocal_rerank",
    use_async=True,
    verbose=True,
    # query_gen_prompt="...",  # we could override the query generation prompt here
)

5. 自动合并检索器

AutoMergingRetriever通过查看一组叶节点，并递归地“合并”引用超出给定阈值的父节点的叶节点子集。 这使我们能够将潜在不同的较小上下文合并为可能有助于综合的更大上下文。

使用 HierarchicalNodeParser。 这将输出节点的层次结构，从具有较大块大小的顶级节点到具有较小块大小的子节点，其中每个子节点都有一个具有较大块大小的父节点。

from llama_index.core.retrievers import AutoMergingRetriever
from llama_index.core.node_parser import (
    HierarchicalNodeParser,
    SentenceSplitter,
)
from llama_index.core.node_parser import get_leaf_nodes, get_root_nodes
from llama_index.core import VectorStoreIndex


node_parser = HierarchicalNodeParser.from_defaults()
nodes = node_parser.get_nodes_from_documents(docs)
leaf_nodes = get_leaf_nodes(nodes)
base_index = VectorStoreIndex(
    leaf_nodes,
    storage_context=storage_context,
)

base_retriever = base_index.as_retriever(similarity_top_k=6)
retriever = AutoMergingRetriever(base_retriever, storage_context, verbose=True)

6. 元数据替换

在检索期间，在将检索到的句子传递给 LLM 之前，使用 MetadataReplacementNodePostProcessor 将单个句子替换为包含周围句子的窗口。这对于大型文档/索引最有用，因为它有助于检索更细粒度的详细信息。默认情况下，句子窗口是原始句子两侧各 5 个句子。

from llama_index.core.postprocessor import MetadataReplacementPostProcessor

query_engine = sentence_index.as_query_engine(
    similarity_top_k=2,
    # the target key defaults to `window` to match the node_parser's default
    node_postprocessors=[
        MetadataReplacementPostProcessor(target_metadata_key="window")
    ],
)
window_response = query_engine.query(
    "What are the concerns surrounding the AMOC?"
)
print(window_response)

7. 自动检索

自动检索技术执行半结构化查询，将语义搜索与结构化过滤相结合。VectorIndexAutoRetriever 模块接受 VectorStoreInfo，其中包含矢量存储集合及其支持的元数据过滤器的结构化描述。 然后，该信息将用于自动检索提示，其中 LLM 推断元数据过滤器。

from llama_index.core.retrievers import VectorIndexAutoRetriever
from llama_index.core.vector_stores.types import MetadataInfo, VectorStoreInfo


vector_store_info = VectorStoreInfo(
    content_info="brief biography of celebrities",
    metadata_info=[
        MetadataInfo(
            name="category",
            type="str",
            description=(
                "Category of the celebrity, one of [Sports, Entertainment,"
                " Business, Music]"
            ),
        ),
        MetadataInfo(
            name="country",
            type="str",
            description=(
                "Country of the celebrity, one of [United States, Barbados,"
                " Portugal]"
            ),
        ),
    ],
)
retriever = VectorIndexAutoRetriever(
    index, vector_store_info=vector_store_info
)

8. 递归检索

递归检索的概念是，我们不仅探索直接最相关的节点，而且探索与其他检索器/查询引擎的节点关系并执行它们。 例如，节点可以表示结构化表的简明摘要，并通过该结构化表链接到 SQL/Pandas 查询引擎。 然后，如果检索到该节点，我们还希望向底层查询引擎查询答案。这对于具有层次关系的文档特别有用。

from llama_index.core.retrievers import RecursiveRetriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core import get_response_synthesizer

recursive_retriever = RecursiveRetriever(
    "vector",
    retriever_dict={"vector": vector_retriever},
    query_engine_dict=df_id_query_engine_mapping,
    verbose=True,
)

response_synthesizer = get_response_synthesizer(response_mode="compact")

query_engine = RetrieverQueryEngine.from_args(
    recursive_retriever, response_synthesizer=response_synthesizer
)

二、响应合成器

响应合成器使用用户查询和给定的一组文本块从 LLM 生成响应。 响应合成器的输出是一个 Response 对象。

响应合成器通常通过 response_mode 参数来指定。LlamaIndex 中已经实现了几个响应合成器：

• refine：通过顺序浏览每个检索到的文本块来创建和细化答案。 这会对每个节点/检索到的块进行单独的 LLM 调用。如果块太大而无法容纳在窗口中（考虑提示大小），则使用 TokenTextSplitter 对其进行分割（允许块之间有一些文本重叠），并且（新的）附加块被视为原始块集合的块（并且 因此也可以使用细化模板进行查询）。适合更详细的答案。
• compact（默认）：类似于细化，但预先压缩（连接）块，从而减少 LLM 调用。填充尽可能多的文本（从检索到的块连接/打包），使其适合上下文窗口（考虑text_qa_template和refine_template之间的最大提示大小）。 如果文本太长而无法容纳在一个提示中，则会根据需要将其拆分为多个部分（使用 TokenTextSplitter，从而允许文本块之间存在一些重叠）
• tree_summarize：根据需要多次使用summary_template提示查询LLM，以便查询所有串联的块，从而产生尽可能多的答案，这些答案本身在tree_summarize LLM调用中递归地用作块，依此类推，直到只剩下一个块 ，因此只有一个最终答案。如果只有一个答案（因为只有一大块），那么它就是最终答案。如果有多个答案，这些本身将被视为块并递归发送到 tree_summarize 进程（连接/拆分以适合/查询）。适合总结目的。
• simple_summarize：截断所有文本块以适合单个 LLM 提示。 适合快速总结，但可能会因截断而丢失细节。
• no_text：仅运行检索器来获取已发送到 LLM 的节点，而不实际发送它们。 然后可以通过检查response.source_nodes来检查。
• accumulate：给定一组文本块和查询，将查询应用于每个文本块，同时将响应累积到数组中。 返回所有响应的串联字符串。 适合当您需要对每个文本块单独运行相同的查询时。
• Compact_accumulate：与accumulate 相同，但会像compact 一样“压缩”每个LLM 提示，并对每个文本块运行相同的查询。

自定义响应合成器

每个响应合成器都继承自 llama_index.response_synthesizers.base.BaseSynthesizer。 基本 API 非常简单，这使得可以轻松创建自己的响应合成器。

下面我们展示了 init() 函数，以及每个响应合成器必须实现的两个抽象方法。 基本要求是处理查询和文本块，并返回字符串（或字符串生成器）响应。

from llama_index.core import Settings


class BaseSynthesizer(ABC):
    """Response builder class."""

    def __init__(
        self,
        llm: Optional[LLM] = None,
        streaming: bool = False,
    ) -> None:
        """Init params."""
        self._llm = llm or Settings.llm
        self._callback_manager = Settings.callback_manager
        self._streaming = streaming

    @abstractmethod
    def get_response(
        self,
        query_str: str,
        text_chunks: Sequence[str],
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        """Get response."""
        ...

    @abstractmethod
    async def aget_response(
        self,
        query_str: str,
        text_chunks: Sequence[str],
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        """Get response."""
        ...

三、路由器

路由器是接受用户查询和一组“选择”（由元数据定义）并返回一个或多个选定选项的模块。它们可以单独使用（作为“选择器模块”），也可以用作查询引擎或检索器（例如，在其他查询引擎/检索器之上）。

它们是简单但功能强大的模块，使用LLM来提供决策能力。 它们可用于以下用例及更多：

• 在多种数据源中选择正确的数据源
• 决定是否进行摘要（例如使用摘要索引查询引擎）或语义搜索（例如使用向量索引查询引擎）
• 决定是否立即“尝试”一堆选择并组合结果（使用多路由功能）。

核心路由器模块有以下几种形式：

• LLM 选择器将选择作为文本转储放入提示中，并使用 LLM 文本完成端点做出决策
• Pydantic 选择器将选择作为 Pydantic 模式传递到调用端点的函数中，并返回 Pydantic 对象

1. 路由作为查询引擎

from llama_index.core.query_engine import RouterQueryEngine
from llama_index.core.selectors import PydanticSingleSelector
from llama_index.core.selectors.pydantic_selectors import Pydantic
from llama_index.core.tools import QueryEngineTool
from llama_index.core import VectorStoreIndex, SummaryIndex

# define query engines
...

# initialize tools
list_tool = QueryEngineTool.from_defaults(
    query_engine=list_query_engine,
    description="Useful for summarization questions related to the data source",
)
vector_tool = QueryEngineTool.from_defaults(
    query_engine=vector_query_engine,
    description="Useful for retrieving specific context related to the data source",
)

# initialize router query engine (single selection, pydantic)
query_engine = RouterQueryEngine(
    selector=PydanticSingleSelector.from_defaults(),
    query_engine_tools=[
        list_tool,
        vector_tool,
    ],
)
query_engine.query("<query>")

2.路由作为检索器

from llama_index.core.query_engine import RouterQueryEngine
from llama_index.core.selectors import PydanticSingleSelector
from llama_index.core.tools import RetrieverTool

# define indices
...

# define retrievers
vector_retriever = vector_index.as_retriever()
keyword_retriever = keyword_index.as_retriever()

# initialize tools
vector_tool = RetrieverTool.from_defaults(
    retriever=vector_retriever,
    description="Useful for retrieving specific context from Paul Graham essay on What I Worked On.",
)
keyword_tool = RetrieverTool.from_defaults(
    retriever=keyword_retriever,
    description="Useful for retrieving specific context from Paul Graham essay on What I Worked On (using entities mentioned in query)",
)

# define retriever
retriever = RouterRetriever(
    selector=PydanticSingleSelector.from_defaults(llm=llm),
    retriever_tools=[
        list_tool,
        vector_tool,
    ],
)

四、节点后处理器

下面是一些常用的节点后处理器：

相似度后处理器：用于删除低于相似度分数阈值的

from llama_index.core.postprocessor import SimilarityPostprocessor

postprocessor = SimilarityPostprocessor(similarity_cutoff=0.7)

postprocessor.postprocess_nodes(nodes)

关键字节点后处理器：用于确保排除或包含某些关键字。

from llama_index.core.postprocessor import KeywordNodePostprocessor

postprocessor = KeywordNodePostprocessor(
    required_keywords=["word1", "word2"], exclude_keywords=["word3", "word4"]
)

postprocessor.postprocess_nodes(nodes)

元数据替换后处理器：用于将节点内容替换为节点元数据中的字段。 如果元数据中不存在该字段，则节点文本保持不变。 与 SentenceWindowNodeParser 结合使用时最有用。

from llama_index.core.postprocessor import MetadataReplacementPostProcessor

postprocessor = MetadataReplacementPostProcessor(
    target_metadata_key="window",
)

postprocessor.postprocess_nodes(nodes)

句子嵌入优化器：该后处理器通过删除与查询不相关的句子来优化标记的使用（这是使用嵌入完成的）。

from llama_index.core.postprocessor import SentenceEmbeddingOptimizer

postprocessor = SentenceEmbeddingOptimizer(
    embed_model=service_context.embed_model,
    percentile_cutoff=0.5,
    # threshold_cutoff=0.7
)

postprocessor.postprocess_nodes(nodes)

SentenceTransformerRerank：使用SentenceTransformer包中的交叉编码器对节点重新排序，并返回前 N 个节点。同理还有LLMReRank，JinarReRank，CohereReRank

from llama_index.core.postprocessor import SentenceTransformerRerank

# We choose a model with relatively high speed and decent accuracy.
postprocessor = SentenceTransformerRerank(
    model="cross-encoder/ms-marco-MiniLM-L-2-v2", top_n=3
)

postprocessor.postprocess_nodes(nodes)

五、结构化输出

LlamaIndex 提供了各种模块，使LLM能够以结构化格式生成输出。 包括：

• 输出解析器：在 LLM 文本完成对话之前和之后运行的模块。
• Pydantic 程序：将输入提示映射到由 Pydantic 对象表示的结构化输出的通用模块。 他们可能使用函数调用 API 或文本完成 API + 输出解析器。 这些也可以与查询引擎集成。
• 预定义的 Pydantic 程序：利用预定义的 Pydantic 程序，可将输入映射到特定的输出类型（如数据帧）。

1. 输出解析器

LlamaIndex 支持与其他框架提供的输出解析模块集成。Langchain 提供了可以在 LlamaIndex 中使用的输出解析模块。

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.output_parsers import LangchainOutputParser
from llama_index.llms.openai import OpenAI
from langchain.output_parsers import StructuredOutputParser, ResponseSchema


# load documents, build index
documents = SimpleDirectoryReader("../paul_graham_essay/data").load_data()
index = VectorStoreIndex.from_documents(documents)

# define output schema
response_schemas = [
    ResponseSchema(
        name="Education",
        description="Describes the author's educational experience/background.",
    ),
    ResponseSchema(
        name="Work",
        description="Describes the author's work experience/background.",
    ),
]

# define output parser
lc_output_parser = StructuredOutputParser.from_response_schemas(
    response_schemas
)
output_parser = LangchainOutputParser(lc_output_parser)

# Attach output parser to LLM
llm = OpenAI(output_parser=output_parser)

# obtain a structured response
query_engine = index.as_query_engine(llm=llm)
response = query_engine.query(
    "What are a few things the author did growing up?",
)
print(str(response))

2. Pydantic输出

使用 OpenAIPydanitcProgam 或 LLMTextCompletionProgram具体取决于设置的 LLM。 如果存在中间 LLM 响应（即在具有多个 LLM 调用的细化或 tree_summarize 期间），则 pydantic 对象将作为 JSON 对象注入到下一个 LLM 提示中。

from typing import List
from pydantic import BaseModel


class Biography(BaseModel):
    """Data model for a biography."""

    name: str
    best_known_for: List[str]
    extra_info: str

query_engine = index.as_query_engine(
    response_mode="tree_summarize", output_cls=Biography
)

response = query_engine.query("Who is Paul Graham?")

print(response.name)
# > 'Paul Graham'
print(response.best_known_for)
# > ['working on Bel', 'co-founding Viaweb', 'creating the programming language Arc']
print(response.extra_info)
# > "Paul Graham is a computer scientist, entrepreneur, and writer. He is best known      for ..."

六、查询Pipeline

LlamaIndex 提供了查询Pipeline，允许将不同的模块链接在一起，以便在数据上编排从简单到高级的工作流程。以 QueryPipeline 抽象为中心。 加载各种模块（从 LLM 到提示、检索器到其他管道），将它们全部连接到顺序链或 DAG 中，然后端到端运行。

一些简单的管道本质上是纯线性的——前一个模块的输出直接进入下一个模块的输入。下面一些例子：

• 提示->LLM->输出解析
• 提示 -> LLM -> 提示 -> LLM
• 检索器 -> 响应合成器

from llama_index.postprocessor.cohere_rerank import CohereRerank
from llama_index.core.response_synthesizers import TreeSummarize

# define modules
prompt_str = "Please generate a question about Paul Graham's life regarding the following topic {topic}"
prompt_tmpl = PromptTemplate(prompt_str)
llm = OpenAI(model="gpt-3.5-turbo")
retriever = index.as_retriever(similarity_top_k=3)
reranker = CohereRerank()
summarizer = TreeSummarize(llm=llm)

# define query pipeline
p = QueryPipeline(verbose=True)
p.add_modules(
    {
        "llm": llm,
        "prompt_tmpl": prompt_tmpl,
        "retriever": retriever,
        "summarizer": summarizer,
        "reranker": reranker,
    }
)
p.add_link("prompt_tmpl", "llm")
p.add_link("llm", "retriever")
p.add_link("retriever", "reranker", dest_key="nodes")
p.add_link("llm", "reranker", dest_key="query_str")
p.add_link("reranker", "summarizer", dest_key="nodes")
p.add_link("llm", "summarizer", dest_key="query_str")


output = p.run(topic="YC")
# output type is Response
type(output)

七、其他高级检索技巧

以下是构建生产级 RAG 的一些主要注意事项

• 分离用于检索的块与用于合成的块
• 较大文档集的结构化检索
• 根据任务动态检索块（路由）
• 优化上下文嵌入，尝试微调嵌入模型。

1. 分离用于检索的块与用于合成的块

考虑分离用于检索的最佳块表示可能与用于合成的最佳块。 例如，原始文本块可能包含LLM所需的详细信息，以根据查询合成更详细的答案。 然而，它可能包含可能使嵌入表示产生偏差的填充词/信息，或者它可能缺乏全局上下文，并且当相关查询出现时根本不会被检索。

有两种主要方法可以利用这个想法：

1. 嵌入文档摘要：该摘要链接到与文档关联的块。这可以帮助在检索块之前检索高级相关文档，而不是直接检索块（可能在不相关的文档中）。
2. 嵌入一个句子，然后链接到该句子周围的窗口。这允许更细粒度地检索相关上下文（嵌入巨大的块会导致“中间丢失”问题），但也确保了 LLM 合成有足够的上下文。

2. 结构化检索

标准 RAG（top-k 检索 + 基本文本分割）的一个大问题是，它随着文档数量的增加而表现不佳。例如 如果有 100 个不同的 PDF。 我们希望给定一个查询使用结构化信息来帮助更精确的检索； 例如，如果提出一个仅与两个 PDF 相关的问题，使用结构化信息来确保返回的这两个 PDF 超出了与块的原始嵌入相似性。

有几种方法可以为执行更结构化的标记/检索，每种方法都有自己的优缺点。

1. 元数据过滤器+自动检索： 用元数据标记每个文档，然后存储在矢量数据库中。 在推理期间，使用 LLM 推断正确的元数据过滤器，以除了语义查询字符串之外还查询向量数据库。

   • 优点 ：主要矢量数据库支持。 可以通过多个维度过滤文档。
   • 缺点：很难定义正确的标签。 标签可能没有包含足够的相关信息以进行更精确的检索。 此外，标签表示文档级别的关键字搜索，不允许语义查找。

2. 存储文档层次结构（摘要 -> 原始块）+ 递归检索： 嵌入文档摘要并映射到每个文档的块。 首先在文档级别获取，然后在块级别获取。

   • 优点 ：允许在文档级别进行语义查找。
   • 缺点：不允许通过结构化标签进行关键字查找（可以比语义搜索更精确）。 自动生成摘要也可能很昂贵