class SecGPTPack(BaseLlamaPack):
"""SecGPT Hub.
一个可信的中央实体，将用户查询路由到适当的隔离应用程序。"""
def __init__(
self,
tools: Sequence[BaseTool],
tool_specs: Sequence[BaseToolSpec],
llm: LLM = None,
memory: BaseMemory = None,
output_parser: Optional[ReActOutputParser] = None,
verbose: bool = False,
handle_reasoning_failure_fn: Optional[
Callable[[CallbackManager, Exception], ToolOutput]
] = None,
user_id: Optional[str] = "0",
) -> None:
"""初始化SecGPTPack。
Args:
tools（Sequence[BaseTool]）：可用工具的序列。
tool_specs（Sequence[BaseToolSpec]）：工具的规格。
llm（LLM，可选）：用于处理的语言模型。默认为Settings.llm。
memory（BaseMemory，可选）：用于跟踪对话历史的内存组件。默认为ChatMemoryBuffer。
output_parser（Optional[ReActOutputParser]，可选）：用于处理输出的解析器。默认为None。
verbose（bool，可选）：启用详细输出的标志。默认为False。
handle_reasoning_failure_fn（Optional[Callable[[CallbackManager, Exception], ToolOutput]]，可选）：处理推理失败的可调用函数。默认为None。
user_id（Optional[str]，可选）：用户标识符。默认为"0"。
"""
self.llm = llm or Settings.llm
self.memory = memory or ChatMemoryBuffer.from_defaults(
chat_history=[], llm=self.llm
)
self.output_parser = output_parser
self.verbose = verbose
self.handle_reasoning_failure_fn = handle_reasoning_failure_fn
self.user_id = user_id
self.planner = HubPlanner(self.llm)
self.tool_importer = ToolImporter(tools, tool_specs)
self.hub_operator = HubOperator(self.tool_importer, self.user_id)
def chat(
self,
query: str,
) -> str:
"""处理用户查询并生成响应。
Args:
query (str): 要处理的用户查询。
Returns:
str: SecGPT 生成的响应。
"""
memory_content = self.memory.get()
self.memory.put(ChatMessage(role=MessageRole.USER, content=(query)))
tool_info = self.tool_importer.get_tool_info()
plan = self.planner.plan_generate(query, tool_info, memory_content)
response = self.hub_operator.run(query, plan)
self.memory.put(ChatMessage(role=MessageRole.CHATBOT, content=(response)))
return response