用法模式#

使用模式指南涵盖了更深入地设置和使用 QueryPipeline。

设置管道#

在这里，我们将介绍设置查询管道的几种不同方式。

定义顺序链#

一些简单的管道在本质上是纯线性的 - 前一个模块的输出直接进入下一个模块的输入。

一些例子：

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

这些工作流可以通过简化的 chain 语法轻松地在 QueryPipeline 中表达。

from llama_index.core.query_pipeline import QueryPipeline

# 尝试链接基本提示
prompt_str = "请生成与 {movie_name} 相关的电影"
prompt_tmpl = PromptTemplate(prompt_str)
llm = OpenAI(model="gpt-3.5-turbo")

p = QueryPipeline(chain=[prompt_tmpl, llm], verbose=True)

定义有向无环图#

许多管道将要求您设置一个有向无环图（例如，如果您想要实现标准 RAG 管道中的所有步骤）。

在这里，我们提供了一个更低级的 API，用于添加模块以及它们的键，并定义前一个模块的输出与下一个模块的输入之间的链接。

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

# 定义模块
prompt_str = "请针对以下主题生成关于保罗·格雷厄姆生活的问题"
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)

# 定义查询管道
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")

运行管道#

单输入/单输出#

输入是第一个组件的 kwargs。

如果最后一个组件的输出是单个对象（而不是对象字典），那么我们直接返回该对象。

以前面示例中的管道为例，输出将是一个 Response 对象，因为最后一步是 TreeSummarize 响应合成模块。

output = p.run(topic="YC")
# 输出类型为 Response
type(output)

多输入/多输出#

如果您的有向无环图具有多个根节点/和/或输出节点，可以尝试使用 run_multi。传入一个包含模块键 -> 输入字典的输入字典。输出是模块键 -> 输出字典的字典。

如果我们运行上一个示例，

output_dict = p.run_multi({"llm": {"topic": "YC"}})
print(output_dict)

# 输出字典为 {"summarizer": {"output": response}}

定义部分#

如果您希望为模块的某些输入预填充，可以使用 partial！然后有向无环图将连接到未填充的输入。

您可能需要通过 as_query_component 将模块转换。

以下是一个例子：

summarizer = TreeSummarize(llm=llm)
summarizer_c = summarizer.as_query_component(partial={"nodes": nodes})
# 可以定义链，因为 llm 输出进入 query_str，nodes 已经预填充
p = QueryPipeline(chain=[prompt_tmpl, llm, summarizer_c])
# 运行管道
p.run(topic="YC")

批量输入#

如果您希望运行多轮单/多输入的管道，请在函数调用中设置 batch=True - 支持 run、arun、run_multi 和 arun_multi。传入您想要运行的单个单/多输入的列表。batch 模式将以与输入相同的顺序返回响应列表。

单输入/单输出的示例：p.run(field=[in1: Any, in2: Any], batch=True) --> [out1: Any, out2: Any]

output = p.run(topic=["YC", "RAG", "LlamaIndex"], batch=True)
# 输出为 [ResponseYC, ResponseRAG, ResponseLlamaIndex]
print(output)

多输入/多输出的示例：p.run_multi("root_node": {"field": [in1: Any, in2, Any]}, batch=True) --> {"output_node": {"field": [out1: Any, out2: Any]}}

output_dict = p.run_multi({"llm": {"topic": ["YC", "RAG", "LlamaIndex"]}})
print(output_dict)

# 输出字典为 {"summarizer": {"output": [ResponseYC, ResponseRAG, ResponseLlamaIndex]}}

中间输出#

如果您希望获取查询管道中模块的中间输出，可以使用 run_with_intermediates 或 run_multi_with_intermediates 分别进行单输入和多输入。

输出将是正常输出的元组和包含模块键 -> ComponentIntermediates 的字典。ComponentIntermediates 有两个字段：inputs 字典和 outputs 字典。

输出, 中间结果 = p.run_with_intermediates(topic="YC")
print(输出)
print(中间结果)

# 输出为 (响应, {"module_key": ComponentIntermediates("inputs": {}, "outputs": {})})

定义自定义查询组件#

您可以轻松地定义一个自定义组件：要么将一个函数传递给 FnComponent，要么继承一个 CustomQueryComponent。

将函数传递给 `FnComponent`#

定义任何函数并将其传递给 FnComponent。位置参数名 (args) 将被转换为必需的输入键，关键字参数名 (kwargs) 将被转换为可选的输入键。

注意：我们假设只有一个输出。

from llama_index.core.query_pipeline import FnComponent


def add(a: int, b: int) -> int:
    """相加两个数字。"""
    return a + b


add_component = FnComponent(fn=add, output_key="output")

# add_component 的输入键为 "a" 和 "b"，输出键为 'output'

继承 `CustomQueryComponent`#

简单地继承 CustomQueryComponent，实现验证/运行函数 + 一些辅助函数，并将其插入。

from llama_index.core.query_pipeline import CustomQueryComponent
from typing import Dict, Any


class MyComponent(CustomQueryComponent):
    """我的组件。"""

    # Pydantic 类，可以在这里放任何属性
    ...

    def _validate_component_inputs(
        self, input: Dict[str, Any]
    ) -> Dict[str, Any]:
        """在运行组件期间验证组件输入。"""
        # 注意：这是可选的，但我们在这里展示了如何进行验证
        return input

    @property
    def _input_keys(self) -> set:
        """输入键字典。"""
        return {"input_key1", ...}

    @property
    def _output_keys(self) -> set:
        # 也可以有多个输出
        return {"output_key"}

    def _run_component(self, **kwargs) -> Dict[str, Any]:
        """运行组件。"""
        # 运行逻辑
        ...
        return {"output_key": result}

更多详情请查看我们的深入查询转换指南。

确保输出兼容性#

通过在 QueryPipeline 中链接模块，一个模块的输出将进入下一个模块的输入。

通常，您必须确保链接有效，即预期的输出和输入类型大致匹配。

我们说“大致”是因为我们对现有模块进行了一些处理，以确保“可字符串化”的输出可以传递到可以查询为“字符串”的输入中。某些输出类型被视为可字符串化 - CompletionResponse、ChatResponse、Response、QueryBundle 等。检索器/查询引擎将自动将 string 输入转换为 QueryBundle 对象。

这使您可以执行某些工作流程，否则，如果您自己编写代码，可能需要繁琐的字符串转换，例如，

LLM -> 提示，LLM -> 检索器，LLM -> 查询引擎
查询引擎 -> 提示，查询引擎 -> 检索器

如果您正在定义自定义组件，应使用 _validate_component_inputs 来确保输入是正确的类型，并在输入不正确时抛出错误。