taolib.testing.multi_agent.agents

多智能体系统 - 智能体模块。

包含智能体基类、主智能体、子智能体和工厂类。

Submodules

• taolib.testing.multi_agent.agents.base
• taolib.testing.multi_agent.agents.factory
• taolib.testing.multi_agent.agents.main_agent
• taolib.testing.multi_agent.agents.templates

Classes

BaseAgent	智能体基类。
AgentFactory	智能体工厂。
MainAgent	主智能体。
SubAgentWrapper	子智能体包装器。
AgentTemplate	智能体模板。

Functions

get_agent_factory(→ AgentFactory)	获取全局智能体工厂。
set_agent_factory(→ None)	设置全局智能体工厂。
get_all_templates(...)	获取所有预设模板。
get_template(...)	根据ID获取模板。

Package Contents

class taolib.testing.multi_agent.agents.BaseAgent(document: taolib.testing.multi_agent.models.AgentDocument)

Bases: abc.ABC

智能体基类。

_document

_status

_current_task: taolib.testing.multi_agent.models.TaskDocument | None = None

_message_queue: list[taolib.testing.multi_agent.models.Message] = []

property id: str

获取智能体ID。

property name: str

获取智能体名称。

property status: taolib.testing.multi_agent.models.AgentStatus

获取智能体状态。

property document: taolib.testing.multi_agent.models.AgentDocument

获取智能体文档。

property current_task: taolib.testing.multi_agent.models.TaskDocument | None

获取当前任务。

async initialize() → None

初始化智能体。

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

发送消息。

参数:

message -- 要发送的消息

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

接收消息。

参数:

message -- 接收到的消息

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

消息发送后的回调。

参数:

message -- 发送的消息

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

Async:

处理接收到的消息。

参数:

message -- 接收到的消息

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

Async:

执行任务。

参数:

task -- 要执行的任务

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

分配任务给智能体。

参数:

task -- 要分配的任务

抛出:

AgentError -- 智能体忙碌

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

任务分配后的回调。

参数:

task -- 分配的任务

async complete_task(success: bool, result: Any | None = None) → None

完成当前任务。

参数:

• success -- 是否成功

• result -- 任务结果

async _on_task_completed(task: taolib.testing.multi_agent.models.TaskDocument, success: bool, result: Any | None) → None

任务完成后的回调。

参数:

• task -- 完成的任务

• success -- 是否成功

• result -- 任务结果

async sleep() → None

使智能体休眠。

async wake() → None

唤醒智能体。

async destroy() → None

销毁智能体。

async _on_destroy() → None

智能体销毁前的回调。

class taolib.testing.multi_agent.agents.AgentFactory(llm_manager: taolib.testing.multi_agent.llm.LLMManager | None = None)

智能体工厂。

_llm_manager = None

_templates: Dict[str, taolib.testing.multi_agent.agents.templates.AgentTemplate]

_agent_classes: Dict[taolib.testing.multi_agent.models.AgentType, Type[taolib.testing.multi_agent.agents.base.BaseAgent]]

register_template(template: taolib.testing.multi_agent.agents.templates.AgentTemplate) → None

注册智能体模板。

参数:

template -- 智能体模板

get_template(template_id: str) → taolib.testing.multi_agent.agents.templates.AgentTemplate | None

获取智能体模板。

参数:

template_id -- 模板ID

返回:

模板,如果不存在则返回None

返回类型:

Optional[AgentTemplate]

get_all_templates() → Dict[str, taolib.testing.multi_agent.agents.templates.AgentTemplate]

获取所有模板。

返回:

所有模板的字典

返回类型:

Dict[str, AgentTemplate]

async create_agent(agent_create: taolib.testing.multi_agent.models.AgentCreate, agent_id: str | None = None) → taolib.testing.multi_agent.agents.base.BaseAgent

创建智能体。

参数:

• agent_create -- 智能体创建配置

• agent_id -- 智能体ID,如果为None则自动生成

返回:

创建的智能体

返回类型:

BaseAgent

抛出:

AgentError -- 无法创建智能体

async create_agent_from_template(template_id: str, agent_name: str | None = None, agent_id: str | None = None) → taolib.testing.multi_agent.agents.base.BaseAgent

从模板创建智能体。

参数:

• template_id -- 模板ID

• agent_name -- 智能体名称,如果为None则使用模板名称

• agent_id -- 智能体ID,如果为None则自动生成

返回:

创建的智能体

返回类型:

BaseAgent

抛出:

AgentError -- 模板不存在

async create_main_agent(name: str = '主智能体', agent_id: str | None = None) → taolib.testing.multi_agent.agents.main_agent.MainAgent

创建主智能体。

参数:

• name -- 智能体名称

• agent_id -- 智能体ID,如果为None则自动生成

返回:

创建的主智能体

返回类型:

MainAgent

taolib.testing.multi_agent.agents.get_agent_factory() → AgentFactory

获取全局智能体工厂。

返回:

智能体工厂

返回类型:

AgentFactory

taolib.testing.multi_agent.agents.set_agent_factory(factory: AgentFactory) → None

设置全局智能体工厂。

参数:

factory -- 智能体工厂

class taolib.testing.multi_agent.agents.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

关闭主智能体。

class taolib.testing.multi_agent.agents.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.AgentTemplate

Bases: pydantic.BaseModel

智能体模板。

id: str

name: str

description: str

agent_type: taolib.testing.multi_agent.models.enums.AgentType

capabilities: list[AgentCapability]

config: AgentConfig

skills: list[str]

tags: list[str]

taolib.testing.multi_agent.agents.get_all_templates() → list[taolib.testing.multi_agent.models.AgentTemplate]

获取所有预设模板。

taolib.testing.multi_agent.agents.get_template(template_id: str) → taolib.testing.multi_agent.models.AgentTemplate | None

根据ID获取模板。

参数:

template_id -- 模板ID

返回:

模板实例,如果不存在则返回None

返回类型:

AgentTemplate | None