taolib.testing.multi_agent.agents.main_agent

主智能体。

主智能体负责任务分析、决策和智能体调度。

Attributes

logger

Classes

SubAgentWrapper	子智能体包装器。
MainAgent	主智能体。

Module Contents

taolib.testing.multi_agent.agents.main_agent.logger

class taolib.testing.multi_agent.agents.main_agent.SubAgentWrapper(document: taolib.testing.multi_agent.models.AgentDocument, llm_manager: taolib.testing.multi_agent.llm.LLMManager)

Bases: taolib.testing.multi_agent.agents.base.BaseAgent

子智能体包装器。

_llm_manager

async _handle_message(message: taolib.testing.multi_agent.models.Message) → None

处理接收到的消息。

参数:

message -- 接收到的消息

async execute_task(task: taolib.testing.multi_agent.models.TaskDocument) → None

执行任务。

参数:

task -- 要执行的任务

class taolib.testing.multi_agent.agents.main_agent.MainAgent(document: taolib.testing.multi_agent.models.AgentDocument, llm_manager: taolib.testing.multi_agent.llm.LLMManager | None = None)

Bases: taolib.testing.multi_agent.agents.base.BaseAgent

主智能体。

_llm_manager = None

_sub_agents: Dict[str, taolib.testing.multi_agent.agents.base.BaseAgent]

_task_queue: List[taolib.testing.multi_agent.models.TaskDocument] = []

_running_tasks: Dict[str, asyncio.Task]

_is_running = False

_main_loop_task: asyncio.Task | None = None

async initialize() → None

初始化主智能体。

async _create_default_agents() → None

创建默认子智能体。

async _main_loop() → None

主循环。

async _process_task_queue() → None

处理任务队列。

async _check_completed_tasks() → None

检查已完成的任务。

async _handle_message(message: taolib.testing.multi_agent.models.Message) → None

处理接收到的消息。

参数:

message -- 接收到的消息

async execute_task(task: taolib.testing.multi_agent.models.TaskDocument) → None

执行任务。

参数:

task -- 要执行的任务

async _analyze_task(task: taolib.testing.multi_agent.models.TaskDocument) → Dict[str, Any]

分析任务。

参数:

task -- 任务

返回:

分析结果

返回类型:

Dict[str, Any]

async _decompose_task(task: taolib.testing.multi_agent.models.TaskDocument, analysis: Dict[str, Any]) → List[taolib.testing.multi_agent.models.SubTask]

分解任务为子任务。

参数:

• task -- 任务

• analysis -- 任务分析结果

返回:

子任务列表

返回类型:

List[SubTask]

async _execute_subtasks(subtasks: List[taolib.testing.multi_agent.models.SubTask]) → List[taolib.testing.multi_agent.models.TaskResult]

执行子任务。

参数:

subtasks -- 子任务列表

返回:

子任务结果列表

返回类型:

List[TaskResult]

async _select_agent_for_subtask(subtask: taolib.testing.multi_agent.models.SubTask) → taolib.testing.multi_agent.agents.base.BaseAgent | None

为子任务选择合适的智能体。

参数:

subtask -- 子任务

返回:

选中的智能体,如果没有则返回None

返回类型:

Optional[BaseAgent]

async _aggregate_results(task: taolib.testing.multi_agent.models.TaskDocument, subtask_results: List[taolib.testing.multi_agent.models.TaskResult]) → taolib.testing.multi_agent.models.TaskResult

聚合子任务结果。

参数:

• task -- 主任务

• subtask_results -- 子任务结果列表

返回:

聚合后的结果

返回类型:

TaskResult

async submit_task(task: taolib.testing.multi_agent.models.TaskDocument) → str

提交任务。

参数:

task -- 任务

返回:

任务ID

返回类型:

str

async shutdown() → None

关闭主智能体。