Router

Bases: QueryComponent

选择器组件。

Source code in llama_index/core/query_pipeline/components/router.py

class SelectorComponent(QueryComponent):
    """选择器组件。"""

    selector: BaseSelector = Field(..., description="Selector")

    class Config:
        arbitrary_types_allowed = True

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

    def _validate_component_inputs(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """在运行组件期间验证组件输入。"""
        if "choices" not in input:
            raise ValueError("Input must have key 'choices'")
        if not isinstance(input["choices"], list):
            raise ValueError("Input choices must be a list")

        for idx, choice in enumerate(input["choices"]):
            # make stringable
            input["choices"][idx] = validate_and_convert_stringable(choice)

        # make sure `query` is stringable
        if "query" not in input:
            raise ValueError("Input must have key 'query'")
        input["query"] = validate_and_convert_stringable(input["query"])

        return input

    def _run_component(self, **kwargs: Any) -> Any:
        """运行组件。"""
        output = self.selector.select(kwargs["choices"], kwargs["query"])
        return {"output": output.selections}

    async def _arun_component(self, **kwargs: Any) -> Any:
        """运行组件（异步）。"""
        # NOTE: no native async for postprocessor
        return self._run_component(**kwargs)

    @property
    def input_keys(self) -> InputKeys:
        """输入键。"""
        return InputKeys.from_keys({"choices", "query"})

    @property
    def output_keys(self) -> OutputKeys:
        """输出键。"""
        return OutputKeys.from_keys({"output"})

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/query_pipeline/components/router.py

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

Bases: QueryComponent

路由器组件。

根据选择器将查询路由到不同的查询组件。

假设只选择一个查询组件。

Source code in llama_index/core/query_pipeline/components/router.py

class RouterComponent(QueryComponent):
    """路由器组件。

根据选择器将查询路由到不同的查询组件。

假设只选择一个查询组件。"""

    selector: BaseSelector = Field(..., description="Selector")
    choices: List[str] = Field(
        ..., description="Choices (must correspond to components)"
    )
    components: List[QueryComponent] = Field(
        ..., description="Components (must correspond to choices)"
    )
    verbose: bool = Field(default=False, description="Verbose")

    _query_keys: List[str] = PrivateAttr()

    class Config:
        arbitrary_types_allowed = True

    def __init__(
        self,
        selector: BaseSelector,
        choices: List[str],
        components: List[QUERY_COMPONENT_TYPE],
        verbose: bool = False,
    ) -> None:
        """初始化。"""
        new_components = []
        query_keys = []
        for component in components:
            if isinstance(component, ChainableMixin):
                new_component = component.as_query_component()
            else:
                new_component = component

            # validate component has one input key
            if len(new_component.free_req_input_keys) != 1:
                raise ValueError("Expected one required input key")
            query_keys.append(next(iter(new_component.free_req_input_keys)))
            new_components.append(new_component)

        self._query_keys = query_keys

        super().__init__(
            selector=selector,
            choices=choices,
            components=new_components,
            verbose=verbose,
        )

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

    def _validate_component_inputs(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """在运行组件期间验证组件输入。"""
        # make sure `query` is stringable
        if "query" not in input:
            raise ValueError("Input must have key 'query'")
        input["query"] = validate_and_convert_stringable(input["query"])

        return input

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

    def _validate_component_outputs(self, output: Dict[str, Any]) -> Dict[str, Any]:
        raise NotImplementedError

    def _run_component(self, **kwargs: Any) -> Any:
        """运行组件。"""
        # for the output selection, run the corresponding component, aggregate into list
        sel_output = self.selector.select(self.choices, kwargs["query"])
        # assume one selection
        if len(sel_output.selections) != 1:
            raise ValueError("Expected one selection")
        component = self.components[sel_output.ind]
        log_str = f"Selecting component {sel_output.ind}: " f"{sel_output.reason}."
        if self.verbose:
            print_text(log_str + "\n", color="pink")
        # run component
        # run with input_keys of component
        return component.run_component(
            **{self._query_keys[sel_output.ind]: kwargs["query"]}
        )

    async def _arun_component(self, **kwargs: Any) -> Any:
        """运行组件（异步）。"""
        # for the output selection, run the corresponding component, aggregate into list
        sel_output = await self.selector.aselect(self.choices, kwargs["query"])
        # assume one selection
        if len(sel_output.selections) != 1:
            raise ValueError("Expected one selection")
        component = self.components[sel_output.ind]
        log_str = f"Selecting component {sel_output.ind}: " f"{sel_output.reason}."
        if self.verbose:
            print_text(log_str + "\n", color="pink")
        # run component
        return await component.arun_component(
            **{self._query_keys[sel_output.ind]: kwargs["query"]}
        )

    @property
    def input_keys(self) -> InputKeys:
        """输入键。"""
        return InputKeys.from_keys({"query"})

    @property
    def output_keys(self) -> OutputKeys:
        """输出键。"""
        # not used
        return OutputKeys.from_keys(set())

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

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

input_keys property

input_keys: InputKeys

输入键。

output_keys property

output_keys: OutputKeys

输出键。

sub_query_components property

sub_query_components: List[QueryComponent]

获取子查询组件。

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

set_callback_manager

set_callback_manager(
    callback_manager: CallbackManager,
) -> None

设置回调管理器。

Source code in llama_index/core/query_pipeline/components/router.py

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

validate_component_outputs

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

在运行组件期间验证组件输入。

Source code in llama_index/core/query_pipeline/components/router.py

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