Bases: QueryComponent
Query component that takes in an arbitrary function.
Source code in llama-index-core/llama_index/core/query_pipeline/components/function.py
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
68
69
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 | class FnComponent(QueryComponent):
"""Query component that takes in an arbitrary function."""
fn: Callable = Field(..., description="Function to run.")
async_fn: Optional[Callable] = Field(
None, description="Async function to run. If not provided, will run `fn`."
)
output_key: str = Field(
default="output", description="Output key for component output."
)
_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,
output_key: str = "output",
**kwargs: Any,
) -> None:
"""Initialize."""
# 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, output_key=output_key, **kwargs)
class Config:
arbitrary_types_allowed = True
def set_callback_manager(self, callback_manager: CallbackManager) -> None:
"""Set callback manager."""
# TODO: implement
def _validate_component_inputs(self, input: Dict[str, Any]) -> Dict[str, Any]:
"""Validate component inputs during run_component."""
# check that all required parameters are present
missing_params = self._req_params - set(input.keys())
if missing_params:
raise ValueError(
f"Missing required parameters: {missing_params}. "
f"Input keys: {input.keys()}"
)
# check that no extra parameters are present
extra_params = set(input.keys()) - self._req_params - self._opt_params
if extra_params:
raise ValueError(
f"Extra parameters: {extra_params}. " f"Input keys: {input.keys()}"
)
return input
def _run_component(self, **kwargs: Any) -> Dict:
"""Run component."""
return {self.output_key: self.fn(**kwargs)}
async def _arun_component(self, **kwargs: Any) -> Any:
"""Run component (async)."""
if self.async_fn is None:
return self._run_component(**kwargs)
else:
return {self.output_key: await self.async_fn(**kwargs)}
@property
def input_keys(self) -> InputKeys:
"""Input keys."""
return InputKeys.from_keys(
required_keys=self._req_params, optional_keys=self._opt_params
)
@property
def output_keys(self) -> OutputKeys:
"""Output keys."""
return OutputKeys.from_keys({self.output_key})
|
set_callback_manager
Set callback manager.
Source code in llama-index-core/llama_index/core/query_pipeline/components/function.py
| def set_callback_manager(self, callback_manager: CallbackManager) -> None:
"""Set callback manager."""
|