Skip to content

Arg pack

Bases: QueryComponent

参数打包组件。

将任意数量的参数打包成一个列表。

Source code in llama_index/core/query_pipeline/components/argpacks.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
class ArgPackComponent(QueryComponent):
    """参数打包组件。

    将任意数量的参数打包成一个列表。"""

    convert_fn: Optional[Callable] = Field(
        default=None, description="Function to convert output."
    )

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

    def validate_component_inputs(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """验证组件输入。"""
        return input

    def _validate_component_outputs(self, output: Dict[str, Any]) -> Dict[str, Any]:
        """验证组件输出。"""
        # make sure output value is a list
        if not isinstance(output["output"], list):
            raise ValueError(f"Output is not a list.")
        return output

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

    def _run_component(self, **kwargs: Any) -> Any:
        """运行组件。"""
        # combine all lists into one
        output = []
        for v in kwargs.values():
            if self.convert_fn is not None:
                v = self.convert_fn(v)

            if isinstance(v, list):
                output.extend(v)
            else:
                output.append(v)
        return {"output": output}

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

    @property
    def input_keys(self) -> InputKeys:
        """输入键。"""
        # NOTE: this shouldn't be used
        return InputKeys.from_keys(set(), optional_keys=set())

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

input_keys property #

input_keys: InputKeys

输入键。

output_keys property #

output_keys: OutputKeys

输出键。

validate_component_inputs #

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

验证组件输入。

Source code in llama_index/core/query_pipeline/components/argpacks.py 
26
27
28
def validate_component_inputs(self, input: Dict[str, Any]) -> Dict[str, Any]:
    """验证组件输入。"""
    return input

set_callback_manager #

set_callback_manager(callback_manager: Any) -> None

设置回调管理器。

Source code in llama_index/core/query_pipeline/components/argpacks.py 
37
38
def set_callback_manager(self, callback_manager: Any) -> None:
    """设置回调管理器。"""

Bases: QueryComponent

关键字参数打包组件。

将任意数量的关键字参数打包成一个字典。

Source code in llama_index/core/query_pipeline/components/argpacks.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class KwargPackComponent(QueryComponent):
    """关键字参数打包组件。

    将任意数量的关键字参数打包成一个字典。"""

    convert_fn: Optional[Callable] = Field(
        default=None, description="Function to convert output."
    )

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

    def validate_component_inputs(self, input: Dict[str, Any]) -> Dict[str, Any]:
        """验证组件输入。"""
        return input

    def _validate_component_outputs(self, output: Dict[str, Any]) -> Dict[str, Any]:
        """验证组件输出。"""
        # make sure output value is a list
        if not isinstance(output["output"], dict):
            raise ValueError(f"Output is not a dict.")
        return output

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

    def _run_component(self, **kwargs: Any) -> Any:
        """运行组件。"""
        if self.convert_fn is not None:
            for k, v in kwargs.items():
                kwargs[k] = self.convert_fn(v)
        return {"output": kwargs}

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

    @property
    def input_keys(self) -> InputKeys:
        """输入键。"""
        # NOTE: this shouldn't be used
        return InputKeys.from_keys(set(), optional_keys=set())

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

input_keys property #

input_keys: InputKeys

输入键。

output_keys property #

output_keys: OutputKeys

输出键。

validate_component_inputs #

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

验证组件输入。

Source code in llama_index/core/query_pipeline/components/argpacks.py 
83
84
85
def validate_component_inputs(self, input: Dict[str, Any]) -> Dict[str, Any]:
    """验证组件输入。"""
    return input

set_callback_manager #

set_callback_manager(callback_manager: Any) -> None

设置回调管理器。

Source code in llama_index/core/query_pipeline/components/argpacks.py 
94
95
def set_callback_manager(self, callback_manager: Any) -> None:
    """设置回调管理器。"""