class SynthesizerComponent(QueryComponent):
"""合成器组件。"""
synthesizer: BaseSynthesizer = Field(..., description="Synthesizer")
class Config:
arbitrary_types_allowed = True
def set_callback_manager(self, callback_manager: CallbackManager) -> None:
"""设置回调管理器。"""
self.synthesizer.callback_manager = callback_manager
def _validate_component_inputs(self, input: Dict[str, Any]) -> Dict[str, Any]:
"""在运行组件期间验证组件输入。"""
# make sure both query_str and nodes are there
if "query_str" not in input:
raise ValueError("Input must have key 'query_str'")
input["query_str"] = validate_and_convert_stringable(input["query_str"])
if "nodes" not in input:
raise ValueError("Input must have key 'nodes'")
nodes = input["nodes"]
if not isinstance(nodes, list):
raise ValueError("Input nodes must be a list")
for node in nodes:
if not isinstance(node, NodeWithScore):
raise ValueError("Input nodes must be a list of NodeWithScore")
return input
def _run_component(self, **kwargs: Any) -> Dict[str, Any]:
"""运行组件。"""
output = self.synthesizer.synthesize(kwargs["query_str"], kwargs["nodes"])
return {"output": output}
async def _arun_component(self, **kwargs: Any) -> Dict[str, Any]:
"""运行组件。"""
output = await self.synthesizer.asynthesize(
kwargs["query_str"], kwargs["nodes"]
)
return {"output": output}
@property
def input_keys(self) -> InputKeys:
"""输入键。"""
return InputKeys.from_keys({"query_str", "nodes"})
@property
def output_keys(self) -> OutputKeys:
"""输出键。"""
return OutputKeys.from_keys({"output"})