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)