使用模式指南#

本使用模式指南深入介绍了QueryPipeline的设置和使用方法。

管道设置#

本节将介绍几种不同的查询管道设置方式。

定义顺序链#

某些简单管道本质上是纯线性的——前一模块的输出直接作为下一模块的输入。

典型示例包括： - 提示词 -> 大语言模型 -> 输出解析 - 提示词 -> 大语言模型 -> 提示词 -> 大语言模型 - 检索器 -> 响应合成器

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

from llama_index.core.query_pipeline import QueryPipeline

# 尝试链式基础提示
prompt_str = "Please generate related movies to {movie_name}"
prompt_tmpl = PromptTemplate(prompt_str)
llm = OpenAI(model="gpt-3.5-turbo")

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

定义有向无环图(DAG)#

许多管道需要设置DAG（例如要实现标准RAG管道中的所有步骤时）。

我们提供了底层API来添加模块及其键，并定义前驱模块输出与后继模块输入间的连接关系。

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

# 定义模块
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)

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

运行管道#

单输入/单输出模式#

输入是第一个组件的参数。

如果最后一个组件的输出是单个对象（而非对象字典），则直接返回该对象。

以前文示例管道为例，由于最后一步是TreeSummarize响应合成模块，输出将是Response对象。

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

多输入/多输出模式#

若DAG具有多个根节点或输出节点，可使用run_multi方法。传入包含模块键->输入字典的输入字典。输出为模块键->输出字典。

运行前例：

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

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

定义部分参数#

如需预填充模块的某些输入参数，可使用partial功能！这样DAG只需接入未填充的输入。

可能需要通过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]}}

获取中间输出#

如需获取QueryPipeline中模块的中间输出，可分别使用run_with_intermediates（单输入）或run_multi_with_intermediates（多输入）。

输出将是一个元组，包含常规输出和模块键->ComponentIntermediates字典。ComponentIntermediates包含2个字段：inputs字典和outputs字典。

output, intermediates = p.run_with_intermediates(topic="YC")
print(output)
print(intermediates)

# 输出格式为 (Response, {"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对象。

这使得您可以实现某些原本需要自行编写字符串转换模板的工作流，例如： - 大语言模型 -> 提示词，大语言模型 -> 检索器，大语言模型 -> 查询引擎 - 查询引擎 -> 提示词，查询引擎 -> 检索器

定义自定义组件时，应使用_validate_component_inputs确保输入类型正确，若不符合则抛出错误。