Index

管道模式。

InputKeys

Bases: BaseModel

输入键。

Source code in llama_index/core/base/query_pipeline/query.py

class InputKeys(BaseModel):
    """输入键。"""

    required_keys: Set[str] = Field(default_factory=set)
    optional_keys: Set[str] = Field(default_factory=set)

    @classmethod
    def from_keys(
        cls, required_keys: Set[str], optional_keys: Optional[Set[str]] = None
    ) -> "InputKeys":
        """从元组中创建InputKeys。"""
        return cls(required_keys=required_keys, optional_keys=optional_keys or set())

    def validate(self, input_keys: Set[str]) -> None:
        """验证输入的键。"""
        # check if required keys are present, and that keys all are in required or optional
        if not self.required_keys.issubset(input_keys):
            raise ValueError(
                f"Required keys {self.required_keys} are not present in input keys {input_keys}"
            )
        if not input_keys.issubset(self.required_keys.union(self.optional_keys)):
            raise ValueError(
                f"Input keys {input_keys} contain keys not in required or optional keys {self.required_keys.union(self.optional_keys)}"
            )

    def __len__(self) -> int:
        """输入键的长度。"""
        return len(self.required_keys) + len(self.optional_keys)

    def all(self) -> Set[str]:
        """获取所有输入键。"""
        return self.required_keys.union(self.optional_keys)

from_keys classmethod

from_keys(
    required_keys: Set[str],
    optional_keys: Optional[Set[str]] = None,
) -> InputKeys

从元组中创建InputKeys。

Source code in llama_index/core/base/query_pipeline/query.py

@classmethod
def from_keys(
    cls, required_keys: Set[str], optional_keys: Optional[Set[str]] = None
) -> "InputKeys":
    """从元组中创建InputKeys。"""
    return cls(required_keys=required_keys, optional_keys=optional_keys or set())

validate

validate(input_keys: Set[str]) -> None

验证输入的键。

Source code in llama_index/core/base/query_pipeline/query.py

def validate(self, input_keys: Set[str]) -> None:
    """验证输入的键。"""
    # check if required keys are present, and that keys all are in required or optional
    if not self.required_keys.issubset(input_keys):
        raise ValueError(
            f"Required keys {self.required_keys} are not present in input keys {input_keys}"
        )
    if not input_keys.issubset(self.required_keys.union(self.optional_keys)):
        raise ValueError(
            f"Input keys {input_keys} contain keys not in required or optional keys {self.required_keys.union(self.optional_keys)}"
        )

all

all() -> Set[str]

获取所有输入键。

Source code in llama_index/core/base/query_pipeline/query.py

def all(self) -> Set[str]:
    """获取所有输入键。"""
    return self.required_keys.union(self.optional_keys)

OutputKeys

Bases: BaseModel

输出键。

Source code in llama_index/core/base/query_pipeline/query.py

class OutputKeys(BaseModel):
    """输出键。"""

    required_keys: Set[str] = Field(default_factory=set)

    @classmethod
    def from_keys(
        cls,
        required_keys: Set[str],
    ) -> "OutputKeys":
        """从元组中创建OutputKeys。"""
        return cls(required_keys=required_keys)

    def validate(self, input_keys: Set[str]) -> None:
        """验证输入的键。"""
        # validate that input keys exactly match required keys
        if input_keys != self.required_keys:
            raise ValueError(
                f"Input keys {input_keys} do not match required keys {self.required_keys}"
            )

from_keys classmethod

from_keys(required_keys: Set[str]) -> OutputKeys

从元组中创建OutputKeys。

Source code in llama_index/core/base/query_pipeline/query.py

@classmethod
def from_keys(
    cls,
    required_keys: Set[str],
) -> "OutputKeys":
    """从元组中创建OutputKeys。"""
    return cls(required_keys=required_keys)

validate

validate(input_keys: Set[str]) -> None

验证输入的键。

Source code in llama_index/core/base/query_pipeline/query.py

def validate(self, input_keys: Set[str]) -> None:
    """验证输入的键。"""
    # validate that input keys exactly match required keys
    if input_keys != self.required_keys:
        raise ValueError(
            f"Input keys {input_keys} do not match required keys {self.required_keys}"
        )

ChainableMixin

Bases: ABC

可链接的混合。

一个模块，可以通过as_query_component从一组输入中生成QueryComponent。

如果直接插入到QueryPipeline中，ChainableMixin将被转换为具有默认参数的QueryComponent。

Source code in llama_index/core/base/query_pipeline/query.py

class ChainableMixin(ABC):
    """可链接的混合。

    一个模块，可以通过`as_query_component`从一组输入中生成`QueryComponent`。

    如果直接插入到`QueryPipeline`中，`ChainableMixin`将被转换为具有默认参数的`QueryComponent`。"""

    @abstractmethod
    def _as_query_component(self, **kwargs: Any) -> "QueryComponent":
        """获取查询组件。"""

    def as_query_component(
        self, partial: Optional[Dict[str, Any]] = None, **kwargs: Any
    ) -> "QueryComponent":
        """获取查询组件。"""
        component = self._as_query_component(**kwargs)
        component.partial(**(partial or {}))
        return component

as_query_component

as_query_component(
    partial: Optional[Dict[str, Any]] = None, **kwargs: Any
) -> QueryComponent

获取查询组件。

Source code in llama_index/core/base/query_pipeline/query.py

def as_query_component(
    self, partial: Optional[Dict[str, Any]] = None, **kwargs: Any
) -> "QueryComponent":
    """获取查询组件。"""
    component = self._as_query_component(**kwargs)
    component.partial(**(partial or {}))
    return component

QueryComponent

Bases: BaseModel

查询组件。

表示可以在“QueryPipeline”中运行的组件。

Source code in llama_index/core/base/query_pipeline/query.py

class QueryComponent(BaseModel):
    """查询组件。

    表示可以在“QueryPipeline”中运行的组件。"""

    partial_dict: Dict[str, Any] = Field(
        default_factory=dict, description="Partial arguments to run_component"
    )

    # TODO: make this a subclass of BaseComponent (e.g. use Pydantic)

    def partial(self, **kwargs: Any) -> None:
        """使用部分参数进行更新。"""
        self.partial_dict.update(kwargs)

    @abstractmethod
    def set_callback_manager(self, callback_manager: CallbackManager) -> None:
        """设置回调管理器。"""
        # TODO: refactor so that callback_manager is always passed in during runtime.

    @property
    def free_req_input_keys(self) -> Set[str]:
        """获取免费的输入键。"""
        return self.input_keys.required_keys.difference(self.partial_dict.keys())

    @abstractmethod
    def _validate_component_inputs(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """在运行组件期间验证组件输入。"""

    def _validate_component_outputs(self, output: Dict[str, Any]) -> Dict[str, Any]:
        """在运行组件期间验证组件输出。"""
        # override if needed
        return output

    def validate_component_inputs(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """验证组件输入。"""
        # make sure set of input keys == self.input_keys
        self.input_keys.validate(set(input.keys()))
        return self._validate_component_inputs(input)

    def validate_component_outputs(self, output: Dict[str, Any]) -> Dict[str, Any]:
        """验证组件输出。"""
        # make sure set of output keys == self.output_keys
        self.output_keys.validate(set(output.keys()))
        return self._validate_component_outputs(output)

    def run_component(self, **kwargs: Any) -> Dict[str, Any]:
        """运行组件。"""
        kwargs.update(self.partial_dict)
        kwargs = self.validate_component_inputs(kwargs)
        component_outputs = self._run_component(**kwargs)
        return self.validate_component_outputs(component_outputs)

    async def arun_component(self, **kwargs: Any) -> Dict[str, Any]:
        """运行组件。"""
        kwargs.update(self.partial_dict)
        kwargs = self.validate_component_inputs(kwargs)
        component_outputs = await self._arun_component(**kwargs)
        return self.validate_component_outputs(component_outputs)

    @abstractmethod
    def _run_component(self, **kwargs: Any) -> Dict:
        """运行组件。"""

    @abstractmethod
    async def _arun_component(self, **kwargs: Any) -> Any:
        """运行组件（异步）。"""

    @property
    @abstractmethod
    def input_keys(self) -> InputKeys:
        """输入键。"""

    @property
    @abstractmethod
    def output_keys(self) -> OutputKeys:
        """输出键。"""

    @property
    def sub_query_components(self) -> List["QueryComponent"]:
        """获取子查询组件。

某些查询组件可能具有子查询组件，例如，查询管道将具有子查询组件，IfElseComponent 也将具有子查询组件。
"""
        return []

free_req_input_keys property

free_req_input_keys: Set[str]

获取免费的输入键。

input_keys abstractmethod property

input_keys: InputKeys

输入键。

output_keys abstractmethod property

output_keys: OutputKeys

输出键。

sub_query_components property

sub_query_components: List[QueryComponent]

获取子查询组件。

某些查询组件可能具有子查询组件，例如，查询管道将具有子查询组件，IfElseComponent 也将具有子查询组件。

partial

partial(**kwargs: Any) -> None

使用部分参数进行更新。

Source code in llama_index/core/base/query_pipeline/query.py

def partial(self, **kwargs: Any) -> None:
    """使用部分参数进行更新。"""
    self.partial_dict.update(kwargs)

set_callback_manager abstractmethod

set_callback_manager(
    callback_manager: CallbackManager,
) -> None

设置回调管理器。

Source code in llama_index/core/base/query_pipeline/query.py

@abstractmethod
def set_callback_manager(self, callback_manager: CallbackManager) -> None:
    """设置回调管理器。"""

validate_component_inputs

validate_component_inputs(
    input: Dict[str, Any]
) -> Dict[str, Any]

验证组件输入。

Source code in llama_index/core/base/query_pipeline/query.py

def validate_component_inputs(self, input: Dict[str, Any]) -> Dict[str, Any]:
    """验证组件输入。"""
    # make sure set of input keys == self.input_keys
    self.input_keys.validate(set(input.keys()))
    return self._validate_component_inputs(input)

validate_component_outputs

validate_component_outputs(
    output: Dict[str, Any]
) -> Dict[str, Any]

验证组件输出。

Source code in llama_index/core/base/query_pipeline/query.py

def validate_component_outputs(self, output: Dict[str, Any]) -> Dict[str, Any]:
    """验证组件输出。"""
    # make sure set of output keys == self.output_keys
    self.output_keys.validate(set(output.keys()))
    return self._validate_component_outputs(output)

run_component

run_component(**kwargs: Any) -> Dict[str, Any]

运行组件。

Source code in llama_index/core/base/query_pipeline/query.py

def run_component(self, **kwargs: Any) -> Dict[str, Any]:
    """运行组件。"""
    kwargs.update(self.partial_dict)
    kwargs = self.validate_component_inputs(kwargs)
    component_outputs = self._run_component(**kwargs)
    return self.validate_component_outputs(component_outputs)

arun_component async

arun_component(**kwargs: Any) -> Dict[str, Any]

运行组件。

Source code in llama_index/core/base/query_pipeline/query.py

async def arun_component(self, **kwargs: Any) -> Dict[str, Any]:
    """运行组件。"""
    kwargs.update(self.partial_dict)
    kwargs = self.validate_component_inputs(kwargs)
    component_outputs = await self._arun_component(**kwargs)
    return self.validate_component_outputs(component_outputs)

CustomQueryComponent

Bases: QueryComponent

自定义查询组件。

Source code in llama_index/core/base/query_pipeline/query.py

class CustomQueryComponent(QueryComponent):
    """自定义查询组件。"""

    callback_manager: CallbackManager = Field(
        default_factory=CallbackManager, description="Callback manager"
    )

    class Config:
        arbitrary_types_allowed = True

    def set_callback_manager(self, callback_manager: CallbackManager) -> None:
        """设置回调管理器。"""
        self.callback_manager = callback_manager

    def _validate_component_inputs(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """在运行组件期间验证组件输入。"""
        # NOTE: user can override this method to validate inputs
        # but we do this by default for convenience
        return input

    async def _arun_component(self, **kwargs: Any) -> Any:
        """运行组件（异步）。"""
        raise NotImplementedError("This component does not support async run.")

    @property
    def _input_keys(self) -> Set[str]:
        """输入键字典。"""
        raise NotImplementedError("Not implemented yet. Please override this method.")

    @property
    def _optional_input_keys(self) -> Set[str]:
        """可选的输入键字典。"""
        return set()

    @property
    def _output_keys(self) -> Set[str]:
        """输出键字典。"""
        raise NotImplementedError("Not implemented yet. Please override this method.")

    @property
    def input_keys(self) -> InputKeys:
        """输入键。"""
        # NOTE: user can override this too, but we have them implement an
        # abstract method to make sure they do it

        return InputKeys.from_keys(
            required_keys=self._input_keys, optional_keys=self._optional_input_keys
        )

    @property
    def output_keys(self) -> OutputKeys:
        """输出键。"""
        # NOTE: user can override this too, but we have them implement an
        # abstract method to make sure they do it
        return OutputKeys.from_keys(self._output_keys)

input_keys property

input_keys: InputKeys

输入键。

output_keys property

output_keys: OutputKeys

输出键。

set_callback_manager

set_callback_manager(
    callback_manager: CallbackManager,
) -> None

设置回调管理器。

Source code in llama_index/core/base/query_pipeline/query.py

def set_callback_manager(self, callback_manager: CallbackManager) -> None:
    """设置回调管理器。"""
    self.callback_manager = callback_manager

Link

Bases: BaseModel

两个组件之间的链接。

Source code in llama_index/core/base/query_pipeline/query.py

class Link(BaseModel):
    """两个组件之间的链接。"""

    src: str = Field(..., description="Source component name")
    dest: str = Field(..., description="Destination component name")
    src_key: Optional[str] = Field(
        default=None, description="Source component output key"
    )
    dest_key: Optional[str] = Field(
        default=None, description="Destination component input key"
    )

    condition_fn: Optional[Callable] = Field(
        default=None, description="Condition to determine if link should be followed"
    )
    input_fn: Optional[Callable] = Field(
        default=None, description="Input to destination component"
    )

    def __init__(
        self,
        src: str,
        dest: str,
        src_key: Optional[str] = None,
        dest_key: Optional[str] = None,
        condition_fn: Optional[Callable] = None,
        input_fn: Optional[Callable] = None,
    ) -> None:
        """初始化参数。"""
        # NOTE: This is to enable positional args.
        super().__init__(
            src=src,
            dest=dest,
            src_key=src_key,
            dest_key=dest_key,
            condition_fn=condition_fn,
            input_fn=input_fn,
        )

ComponentIntermediates

组件的中间输入和输出。

Source code in llama_index/core/base/query_pipeline/query.py

class ComponentIntermediates:
    """组件的中间输入和输出。"""

    def __init__(
        self,
        inputs: Dict[str, Any],
        outputs: Dict[str, Any],
    ) -> None:
        """初始化。"""
        self.inputs = inputs
        self.outputs = outputs

    def __repr__(self) -> str:
        return (
            f"ComponentIntermediates(inputs={self.inputs!s}, "
            f"outputs={self.outputs!s})"
        )

    def __str__(self) -> str:
        return self.__repr__()