Agent

Bases: QueryComponent

代理组件。

用于类型检查的抽象类。

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

class BaseAgentComponent(QueryComponent):
    """代理组件。

    用于类型检查的抽象类。"""

Bases: BaseAgentComponent

代理的功能组件。

旨在让用户轻松修改状态。

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

class AgentFnComponent(BaseAgentComponent):
    """代理的功能组件。

旨在让用户轻松修改状态。"""

    fn: Callable = Field(..., description="Function to run.")
    async_fn: Optional[Callable] = Field(
        None, description="Async function to run. If not provided, will run `fn`."
    )

    _req_params: Set[str] = PrivateAttr()
    _opt_params: Set[str] = PrivateAttr()

    def __init__(
        self,
        fn: Callable,
        async_fn: Optional[Callable] = None,
        req_params: Optional[Set[str]] = None,
        opt_params: Optional[Set[str]] = None,
        **kwargs: Any,
    ) -> None:
        """初始化。"""
        # determine parameters
        default_req_params, default_opt_params = get_parameters(fn)
        # make sure task and step are part of the list, and remove them from the list
        if "task" not in default_req_params or "state" not in default_req_params:
            raise ValueError(
                "AgentFnComponent must have 'task' and 'state' as required parameters"
            )

        default_req_params = default_req_params - {"task", "state"}
        default_opt_params = default_opt_params - {"task", "state"}

        if req_params is None:
            req_params = default_req_params
        if opt_params is None:
            opt_params = default_opt_params

        self._req_params = req_params
        self._opt_params = opt_params
        super().__init__(fn=fn, async_fn=async_fn, **kwargs)

    class Config:
        arbitrary_types_allowed = True

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

    def _validate_component_inputs(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """在运行组件期间验证组件输入。"""
        from llama_index.core.agent.types import Task

        if "task" not in input:
            raise ValueError("Input must have key 'task'")
        if not isinstance(input["task"], Task):
            raise ValueError("Input must have key 'task' of type Task")

        if "state" not in input:
            raise ValueError("Input must have key 'state'")
        if not isinstance(input["state"], dict):
            raise ValueError("Input must have key 'state' of type dict")

        return input

    def validate_component_outputs(self, output: Dict[str, Any]) -> Dict[str, Any]:
        """验证组件输出。"""
        # NOTE: we override this to do nothing
        return output

    def _validate_component_outputs(self, input: Dict[str, Any]) -> Dict[str, Any]:
        return input

    def _run_component(self, **kwargs: Any) -> Dict:
        """运行组件。"""
        output = self.fn(**kwargs)
        # if not isinstance(output, dict):
        #     raise ValueError("Output must be a dictionary")

        return {"output": output}

    async def _arun_component(self, **kwargs: Any) -> Any:
        """运行组件（异步）。"""
        if self.async_fn is None:
            return self._run_component(**kwargs)
        else:
            output = await self.async_fn(**kwargs)
            # if not isinstance(output, dict):
            #     raise ValueError("Output must be a dictionary")
            return {"output": output}

    @property
    def input_keys(self) -> InputKeys:
        """输入键。"""
        return InputKeys.from_keys(
            required_keys={"task", "state", *self._req_params},
            optional_keys=self._opt_params,
        )

    @property
    def output_keys(self) -> OutputKeys:
        """输出键。"""
        # output can be anything, overrode validate function
        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/agent.py

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

validate_component_outputs

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

验证组件输出。

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

def validate_component_outputs(self, output: Dict[str, Any]) -> Dict[str, Any]:
    """验证组件输出。"""
    # NOTE: we override this to do nothing
    return output

Bases: BaseAgentComponent

代理商的自定义组件。

旨在让用户轻松修改状态。

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

class CustomAgentComponent(BaseAgentComponent):
    """代理商的自定义组件。

旨在让用户轻松修改状态。"""

    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
        # TODO: refactor to put this on base class
        for component in self.sub_query_components:
            component.set_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

        input_keys = self._input_keys.union({"task", "state"})
        return InputKeys.from_keys(
            required_keys=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/query_pipeline/components/agent.py

def set_callback_manager(self, callback_manager: CallbackManager) -> None:
    """设置回调管理器。"""
    self.callback_manager = callback_manager
    # TODO: refactor to put this on base class
    for component in self.sub_query_components:
        component.set_callback_manager(callback_manager)

Bases: QueryComponent

接收代理输入并将其转换为期望的输出。

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

class AgentInputComponent(QueryComponent):
    """接收代理输入并将其转换为期望的输出。"""

    fn: Callable = Field(..., description="Function to run.")
    async_fn: Optional[Callable] = Field(
        None, description="Async function to run. If not provided, will run `fn`."
    )

    _req_params: Set[str] = PrivateAttr()
    _opt_params: Set[str] = PrivateAttr()

    def __init__(
        self,
        fn: Callable,
        async_fn: Optional[Callable] = None,
        req_params: Optional[Set[str]] = None,
        opt_params: Optional[Set[str]] = None,
        **kwargs: Any,
    ) -> None:
        """初始化。"""
        # determine parameters
        default_req_params, default_opt_params = get_parameters(fn)
        if req_params is None:
            req_params = default_req_params
        if opt_params is None:
            opt_params = default_opt_params

        self._req_params = req_params
        self._opt_params = opt_params
        super().__init__(fn=fn, async_fn=async_fn, **kwargs)

    class Config:
        arbitrary_types_allowed = True

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

    def _validate_component_inputs(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """在运行组件期间验证组件输入。"""
        from llama_index.core.agent.types import Task

        if "task" not in input:
            raise ValueError("Input must have key 'task'")
        if not isinstance(input["task"], Task):
            raise ValueError("Input must have key 'task' of type Task")

        if "state" not in input:
            raise ValueError("Input must have key 'state'")
        if not isinstance(input["state"], dict):
            raise ValueError("Input must have key 'state' of type dict")

        return input

    def validate_component_outputs(self, output: Dict[str, Any]) -> Dict[str, Any]:
        """验证组件输出。"""
        # NOTE: we override this to do nothing
        return output

    def _validate_component_outputs(self, input: Dict[str, Any]) -> Dict[str, Any]:
        return input

    def _run_component(self, **kwargs: Any) -> Dict:
        """运行组件。"""
        output = self.fn(**kwargs)
        if not isinstance(output, dict):
            raise ValueError("Output must be a dictionary")

        return output

    async def _arun_component(self, **kwargs: Any) -> Any:
        """运行组件（异步）。"""
        if self.async_fn is None:
            return self._run_component(**kwargs)
        else:
            output = await self.async_fn(**kwargs)
            if not isinstance(output, dict):
                raise ValueError("Output must be a dictionary")
            return output

    @property
    def input_keys(self) -> InputKeys:
        """输入键。"""
        return InputKeys.from_keys(
            required_keys={"task", "state", *self._req_params},
            optional_keys=self._opt_params,
        )

    @property
    def output_keys(self) -> OutputKeys:
        """输出键。"""
        # output can be anything, overrode validate function
        return OutputKeys.from_keys(set())

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/agent.py

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

validate_component_outputs

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

验证组件输出。

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

def validate_component_outputs(self, output: Dict[str, Any]) -> Dict[str, Any]:
    """验证组件输出。"""
    # NOTE: we override this to do nothing
    return output