taolib.testing.multi_agent

多智能体系统。

无需API KEY的免费大模型多智能体系统。

Submodules

• taolib.testing.multi_agent.agents
• taolib.testing.multi_agent.api
• taolib.testing.multi_agent.errors
• taolib.testing.multi_agent.llm
• taolib.testing.multi_agent.models
• taolib.testing.multi_agent.skills

Attributes

__version__

Exceptions

AgentError	智能体相关错误。
LLMError	LLM模型相关错误。
ModelUnavailableError	模型不可用错误。
MultiAgentError	多智能体系统基础错误。
SkillError	技能相关错误。
TaskError	任务相关错误。

Classes

AgentFactory	智能体工厂。
BaseAgent	智能体基类。
MainAgent	主智能体。
SubAgentWrapper	子智能体包装器。
LLMManager	LLM模型管理器。
LoadBalancer	LLM模型负载均衡器。
ModelRegistry	LLM模型注册表。
AgentBase	智能体基础字段。
AgentCapability	智能体能力描述。
AgentConfig	智能体配置。
AgentCreate	创建智能体的输入模型。
AgentDocument	智能体的 MongoDB 文档模型。
AgentResponse	智能体的 API 响应模型。
AgentStatus	智能体状态。
AgentTemplate	智能体模板。
AgentType	智能体类型。
AgentUpdate	更新智能体的输入模型(所有字段可选)。
LoadBalanceConfig	负载均衡配置。
LoadBalanceStrategy	负载均衡策略。
Message	智能体间通信消息。
MessagePayload	消息载荷。
MessageType	消息类型。
ModelConfig	模型配置。
ModelInstance	模型实例。
ModelProvider	模型提供商。
ModelStatus	模型状态。
ModelStats	模型统计信息。
SkillBase	技能基础字段。
SkillCreate	创建技能的输入模型。
SkillDocument	技能的 MongoDB 文档模型。
SkillEvaluation	技能评估。
SkillParameter	技能参数。
SkillResponse	技能的 API 响应模型。
SkillStatus	技能状态。
SkillTestResult	技能测试结果。
SkillType	技能类型。
SkillUpdate	更新技能的输入模型(所有字段可选)。
SubTask	子任务。
TaskBase	任务基础字段。
TaskConstraint	任务约束条件。
TaskCreate	创建任务的输入模型。
TaskDocument	任务的 MongoDB 文档模型。
TaskProgress	任务进度。
TaskResponse	任务的 API 响应模型。
TaskResult	任务结果。
TaskStatus	任务状态。
TaskUpdate	更新任务的输入模型(所有字段可选)。
BaseSkill	技能基类。
CodeGenerationSkill	代码生成技能。
DataAnalysisSkill	数据分析技能。
Skill	技能协议。
SkillExecutionContext	技能执行上下文。
SkillManager	技能管理器。
SkillRegistry	技能注册表。
TextSummarizationSkill	文本摘要技能。
TranslationSkill	翻译技能。

Functions

get_agent_factory(→ AgentFactory)	获取全局智能体工厂。
get_all_templates(...)	获取所有预设模板。
get_template(...)	根据ID获取模板。
set_agent_factory(→ None)	设置全局智能体工厂。
get_llm_manager(→ LLMManager)	获取全局LLM管理器。
set_llm_manager(→ None)	设置全局LLM管理器。
get_preset_skills(...)	获取所有预设技能。
get_skill_manager(→ SkillManager)	获取全局技能管理器。
get_skill_registry(→ SkillRegistry)	获取全局技能注册表。
set_skill_manager(→ None)	设置全局技能管理器。
set_skill_registry(→ None)	设置全局技能注册表。

Package Contents

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

class taolib.testing.multi_agent.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.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.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 -- 要执行的任务

taolib.testing.multi_agent.get_agent_factory() → AgentFactory

获取全局智能体工厂。

返回:

智能体工厂

返回类型:

AgentFactory

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

获取所有预设模板。

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

根据ID获取模板。

参数:

template_id -- 模板ID

返回:

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

返回类型:

AgentTemplate | None

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

设置全局智能体工厂。

参数:

factory -- 智能体工厂

exception taolib.testing.multi_agent.AgentError

Bases: MultiAgentError

智能体相关错误。

exception taolib.testing.multi_agent.LLMError

Bases: MultiAgentError

LLM模型相关错误。

exception taolib.testing.multi_agent.ModelUnavailableError

Bases: LLMError

模型不可用错误。

exception taolib.testing.multi_agent.MultiAgentError

Bases: Exception

多智能体系统基础错误。

exception taolib.testing.multi_agent.SkillError

Bases: MultiAgentError

技能相关错误。

exception taolib.testing.multi_agent.TaskError

Bases: MultiAgentError

任务相关错误。

class taolib.testing.multi_agent.LLMManager(load_balance_config: taolib.testing.multi_agent.models.LoadBalanceConfig | None = None)

LLM模型管理器。

_load_balancer

_default_provider: taolib.testing.multi_agent.llm.protocols.BaseLLMProvider | None = None

add_model(config: taolib.testing.multi_agent.models.ModelConfig, instance_id: str | None = None) → str

添加一个模型。

参数:

• config -- 模型配置

• instance_id -- 实例ID,如果为None则自动生成

返回:

实例ID

返回类型:

str

async generate(prompt: str, *, temperature: float | None = None, max_tokens: int | None = None, system_prompt: str | None = None, instance_id: str | None = None, **kwargs) → str

生成文本。

参数:

• prompt -- 用户输入的提示词

• temperature -- 温度参数

• max_tokens -- 最大生成token数

• system_prompt -- 系统提示词

• instance_id -- 指定实例ID,如果为None则使用负载均衡选择

• **kwargs -- 其他参数

返回:

生成的文本

返回类型:

str

抛出:

• ModelUnavailableError -- 没有可用的模型

• LLMError -- 生成过程出错

async generate_stream(prompt: str, *, temperature: float | None = None, max_tokens: int | None = None, system_prompt: str | None = None, instance_id: str | None = None, **kwargs) → AsyncGenerator[str, None]

流式生成文本。

参数:

• prompt -- 用户输入的提示词

• temperature -- 温度参数

• max_tokens -- 最大生成token数

• system_prompt -- 系统提示词

• instance_id -- 指定实例ID,如果为None则使用负载均衡选择

• **kwargs -- 其他参数

生成器:

str -- 生成的文本片段

抛出:

• ModelUnavailableError -- 没有可用的模型

• LLMError -- 生成过程出错

async health_check(instance_id: str | None = None) → bool

健康检查。

参数:

instance_id -- 指定实例ID,如果为None则检查所有

返回:

是否健康

返回类型:

bool

get_available_models() → list[str]

获取所有可用的模型实例ID。

返回:

可用的模型实例ID列表

返回类型:

list[str]

get_model_stats(instance_id: str)

获取模型统计信息。

参数:

instance_id -- 实例ID

返回:

模型实例信息

返回类型:

Optional[ModelInstance]

get_all_models()

获取所有模型实例信息。

返回:

模型实例列表

返回类型:

list[ModelInstance]

class taolib.testing.multi_agent.LoadBalancer(config: taolib.testing.multi_agent.models.LoadBalanceConfig | None = None)

LLM模型负载均衡器。

_config

_providers: dict[str, taolib.testing.multi_agent.llm.protocols.BaseLLMProvider]

_instances: dict[str, taolib.testing.multi_agent.models.ModelInstance]

_round_robin_index = 0

_circuit_breaker_states: dict[str, dict]

register_provider(instance_id: str, provider: taolib.testing.multi_agent.llm.protocols.BaseLLMProvider) → None

注册模型提供商。

参数:

• instance_id -- 实例ID

• provider -- 模型提供商实例

get_available_providers() → list[tuple[str, taolib.testing.multi_agent.llm.protocols.BaseLLMProvider]]

获取所有可用的提供商。

返回:

可用的提供商列表

返回类型:

list[tuple[str, BaseLLMProvider]]

_select_round_robin(available: list[tuple[str, taolib.testing.multi_agent.llm.protocols.BaseLLMProvider]]) → tuple[str, taolib.testing.multi_agent.llm.protocols.BaseLLMProvider]

轮询策略选择。

参数:

available -- 可用提供商列表

返回:

选中的提供商

返回类型:

tuple[str, BaseLLMProvider]

_select_least_connections(available: list[tuple[str, taolib.testing.multi_agent.llm.protocols.BaseLLMProvider]]) → tuple[str, taolib.testing.multi_agent.llm.protocols.BaseLLMProvider]

最少连接策略选择。

参数:

available -- 可用提供商列表

返回:

选中的提供商

返回类型:

tuple[str, BaseLLMProvider]

_select_random(available: list[tuple[str, taolib.testing.multi_agent.llm.protocols.BaseLLMProvider]]) → tuple[str, taolib.testing.multi_agent.llm.protocols.BaseLLMProvider]

随机策略选择。

参数:

available -- 可用提供商列表

返回:

选中的提供商

返回类型:

tuple[str, BaseLLMProvider]

_select_weighted(available: list[tuple[str, taolib.testing.multi_agent.llm.protocols.BaseLLMProvider]]) → tuple[str, taolib.testing.multi_agent.llm.protocols.BaseLLMProvider]

加权随机策略选择。

参数:

available -- 可用提供商列表

返回:

选中的提供商

返回类型:

tuple[str, BaseLLMProvider]

select_provider() → tuple[str, taolib.testing.multi_agent.llm.protocols.BaseLLMProvider]

选择一个提供商。

返回:

选中的实例ID和提供商

返回类型:

tuple[str, BaseLLMProvider]

抛出:

ModelUnavailableError -- 没有可用的提供商

record_success(instance_id: str) → None

记录成功请求。

参数:

instance_id -- 实例ID

record_failure(instance_id: str) → None

记录失败请求。

参数:

instance_id -- 实例ID

async health_check_all() → None

对所有提供商进行健康检查。

get_instance_stats(instance_id: str) → taolib.testing.multi_agent.models.ModelInstance | None

获取实例统计信息。

参数:

instance_id -- 实例ID

返回:

实例信息

返回类型:

Optional[ModelInstance]

get_all_instances() → list[taolib.testing.multi_agent.models.ModelInstance]

获取所有实例信息。

返回:

实例列表

返回类型:

list[ModelInstance]

get_provider(instance_id: str) → taolib.testing.multi_agent.llm.protocols.BaseLLMProvider | None

获取指定实例ID的提供商。

参数:

instance_id -- 实例ID

返回:

提供商实例，如果不存在则返回None

返回类型:

Optional[BaseLLMProvider]

class taolib.testing.multi_agent.ModelRegistry

LLM模型注册表。

_providers: dict[taolib.testing.multi_agent.models.ModelProvider, Type[taolib.testing.multi_agent.llm.protocols.BaseLLMProvider]]

classmethod register(provider_type: taolib.testing.multi_agent.models.ModelProvider, provider_class: Type[taolib.testing.multi_agent.llm.protocols.BaseLLMProvider]) → None

注册模型提供商。

参数:

• provider_type -- 模型提供商类型

• provider_class -- 提供商类

classmethod get_provider_class(provider_type: taolib.testing.multi_agent.models.ModelProvider) → Type[taolib.testing.multi_agent.llm.protocols.BaseLLMProvider]

获取模型提供商类。

参数:

provider_type -- 模型提供商类型

返回:

提供商类

返回类型:

Type[BaseLLMProvider]

抛出:

ValueError -- 提供商未注册

classmethod create_provider(config: taolib.testing.multi_agent.models.ModelConfig) → taolib.testing.multi_agent.llm.protocols.BaseLLMProvider

创建模型提供商实例。

参数:

config -- 模型配置

返回:

提供商实例

返回类型:

BaseLLMProvider

classmethod get_available_providers() → list[taolib.testing.multi_agent.models.ModelProvider]

获取所有已注册的提供商。

返回:

提供商列表

返回类型:

list[ModelProvider]

taolib.testing.multi_agent.get_llm_manager() → LLMManager

获取全局LLM管理器。

返回:

LLM管理器

返回类型:

LLMManager

taolib.testing.multi_agent.set_llm_manager(manager: LLMManager) → None

设置全局LLM管理器。

参数:

manager -- LLM管理器

class taolib.testing.multi_agent.AgentBase

Bases: pydantic.BaseModel

智能体基础字段。

name: str

description: str

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

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

capabilities: list[AgentCapability]

config: AgentConfig

template_id: str | None

skills: list[str]

tags: list[str]

metadata: dict[str, Any]

class taolib.testing.multi_agent.AgentCapability

Bases: pydantic.BaseModel

智能体能力描述。

name: str

description: str

confidence: float

tags: list[str]

class taolib.testing.multi_agent.AgentConfig

Bases: pydantic.BaseModel

智能体配置。

max_concurrent_tasks: int

timeout_seconds: int

preferred_model: str | None

system_prompt: str | None

temperature: float

class taolib.testing.multi_agent.AgentCreate

Bases: AgentBase

创建智能体的输入模型。

class taolib.testing.multi_agent.AgentDocument

Bases: AgentBase

智能体的 MongoDB 文档模型。

id: str

created_at: datetime.datetime

updated_at: datetime.datetime

last_active_at: datetime.datetime | None = None

current_task_id: str | None = None

completed_tasks: int

failed_tasks: int

model_config

to_response() → AgentResponse

转换为 API 响应。

class taolib.testing.multi_agent.AgentResponse

Bases: AgentBase

智能体的 API 响应模型。

id: str

created_at: datetime.datetime

updated_at: datetime.datetime

last_active_at: datetime.datetime | None = None

current_task_id: str | None = None

completed_tasks: int

failed_tasks: int

model_config

class taolib.testing.multi_agent.AgentStatus

Bases: enum.StrEnum

智能体状态。

CREATED = 'created'

IDLE = 'idle'

BUSY = 'busy'

SLEEPING = 'sleeping'

ERROR = 'error'

DESTROYED = 'destroyed'

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

class taolib.testing.multi_agent.AgentType

Bases: enum.StrEnum

智能体类型。

MAIN = 'main'

SUB = 'sub'

SPECIALIZED = 'specialized'

class taolib.testing.multi_agent.AgentUpdate

Bases: pydantic.BaseModel

更新智能体的输入模型(所有字段可选)。

name: str | None = None

description: str | None = None

status: taolib.testing.multi_agent.models.enums.AgentStatus | None = None

capabilities: list[AgentCapability] | None = None

config: AgentConfig | None = None

skills: list[str] | None = None

tags: list[str] | None = None

metadata: dict[str, Any] | None = None

class taolib.testing.multi_agent.LoadBalanceConfig

Bases: pydantic.BaseModel

负载均衡配置。

strategy: taolib.testing.multi_agent.models.enums.LoadBalanceStrategy

fallback_enabled: bool

health_check_interval_seconds: int

circuit_breaker_enabled: bool

circuit_breaker_failure_threshold: int

circuit_breaker_reset_timeout_seconds: int

class taolib.testing.multi_agent.LoadBalanceStrategy

Bases: enum.StrEnum

负载均衡策略。

ROUND_ROBIN = 'round_robin'

LEAST_CONNECTIONS = 'least_connections'

RANDOM = 'random'

WEIGHTED = 'weighted'

class taolib.testing.multi_agent.Message

Bases: pydantic.BaseModel

智能体间通信消息。

id: str

message_type: taolib.testing.multi_agent.models.enums.MessageType

sender_id: str

receiver_id: str | None

payload: MessagePayload

timestamp: datetime.datetime

priority: int

requires_response: bool

response_to: str | None

class taolib.testing.multi_agent.MessagePayload

Bases: pydantic.BaseModel

消息载荷。

task_id: str | None

agent_id: str | None

skill_id: str | None

data: dict[str, Any]

metadata: dict[str, Any]

class taolib.testing.multi_agent.MessageType

Bases: enum.StrEnum

消息类型。

TASK_ASSIGN = 'task_assign'

TASK_UPDATE = 'task_update'

TASK_COMPLETE = 'task_complete'

TASK_ERROR = 'task_error'

SKILL_REQUEST = 'skill_request'

SKILL_RESPONSE = 'skill_response'

INFO = 'info'

WARNING = 'warning'

ERROR = 'error'

class taolib.testing.multi_agent.ModelConfig

Bases: pydantic.BaseModel

模型配置。

provider: taolib.testing.multi_agent.models.enums.ModelProvider

model_name: str

base_url: str | None

api_key: str | None

timeout_seconds: int

max_retries: int

rate_limit_requests_per_minute: int

rate_limit_tokens_per_minute: int | None

max_concurrent_requests: int

temperature: float

max_tokens: int | None

weight: float

metadata: dict[str, Any]

class taolib.testing.multi_agent.ModelInstance

Bases: pydantic.BaseModel

模型实例。

id: str

config: ModelConfig

status: taolib.testing.multi_agent.models.enums.ModelStatus

stats: ModelStats

created_at: datetime.datetime

updated_at: datetime.datetime

class taolib.testing.multi_agent.ModelProvider

Bases: enum.StrEnum

模型提供商。

OLLAMA = 'ollama'

HUGGINGFACE = 'huggingface'

GEMINI = 'gemini'

class taolib.testing.multi_agent.ModelStatus

Bases: enum.StrEnum

模型状态。

AVAILABLE = 'available'

BUSY = 'busy'

ERROR = 'error'

UNAVAILABLE = 'unavailable'

class taolib.testing.multi_agent.ModelStats

Bases: pydantic.BaseModel

模型统计信息。

total_requests: int

successful_requests: int

failed_requests: int

total_tokens_used: int

average_latency_seconds: float

current_concurrent_requests: int

requests_this_minute: int

tokens_this_minute: int

last_health_check_at: datetime.datetime | None

last_error_at: datetime.datetime | None

last_error_message: str

class taolib.testing.multi_agent.SkillBase

Bases: pydantic.BaseModel

技能基础字段。

name: str

description: str

skill_type: taolib.testing.multi_agent.models.enums.SkillType

status: taolib.testing.multi_agent.models.enums.SkillStatus

version: str

content: str

parameters: list[SkillParameter]

tags: list[str]

categories: list[str]

dependencies: list[str]

evaluation: SkillEvaluation | None

usage_count: int

success_count: int

metadata: dict[str, Any]

class taolib.testing.multi_agent.SkillCreate

Bases: SkillBase

创建技能的输入模型。

class taolib.testing.multi_agent.SkillDocument

Bases: SkillBase

技能的 MongoDB 文档模型。

id: str

created_by: str = 'system'

updated_by: str = 'system'

created_at: datetime.datetime

updated_at: datetime.datetime

last_used_at: datetime.datetime | None = None

model_config

to_response() → SkillResponse

转换为 API 响应。

class taolib.testing.multi_agent.SkillEvaluation

Bases: pydantic.BaseModel

技能评估。

overall_score: float

accuracy: float

efficiency: float

reliability: float

test_results: list[SkillTestResult]

evaluated_at: datetime.datetime

evaluator: str

comments: str

class taolib.testing.multi_agent.SkillParameter

Bases: pydantic.BaseModel

技能参数。

name: str

type: str

description: str

required: bool

default: Any

constraints: dict[str, Any]

class taolib.testing.multi_agent.SkillResponse

Bases: SkillBase

技能的 API 响应模型。

id: str

created_by: str

updated_by: str

created_at: datetime.datetime

updated_at: datetime.datetime

last_used_at: datetime.datetime | None = None

model_config

class taolib.testing.multi_agent.SkillStatus

Bases: enum.StrEnum

技能状态。

DRAFT = 'draft'

TESTING = 'testing'

APPROVED = 'approved'

DEPRECATED = 'deprecated'

class taolib.testing.multi_agent.SkillTestResult

Bases: pydantic.BaseModel

技能测试结果。

test_name: str

success: bool

input_data: dict[str, Any]

expected_output: dict[str, Any]

actual_output: dict[str, Any]

error_message: str

execution_time_seconds: float

class taolib.testing.multi_agent.SkillType

Bases: enum.StrEnum

技能类型。

CODE = 'code'

PROMPT = 'prompt'

HYBRID = 'hybrid'

class taolib.testing.multi_agent.SkillUpdate

Bases: pydantic.BaseModel

更新技能的输入模型(所有字段可选)。

name: str | None = None

description: str | None = None

skill_type: taolib.testing.multi_agent.models.enums.SkillType | None = None

status: taolib.testing.multi_agent.models.enums.SkillStatus | None = None

version: str | None = None

content: str | None = None

parameters: list[SkillParameter] | None = None

tags: list[str] | None = None

categories: list[str] | None = None

dependencies: list[str] | None = None

evaluation: SkillEvaluation | None = None

metadata: dict[str, Any] | None = None

class taolib.testing.multi_agent.SubTask

Bases: pydantic.BaseModel

子任务。

id: str

name: str

description: str

status: taolib.testing.multi_agent.models.enums.TaskStatus

assigned_agent_id: str | None

result: TaskResult | None

created_at: datetime.datetime

completed_at: datetime.datetime | None

class taolib.testing.multi_agent.TaskBase

Bases: pydantic.BaseModel

任务基础字段。

name: str

description: str

user_input: str

status: taolib.testing.multi_agent.models.enums.TaskStatus

constraints: TaskConstraint

progress: TaskProgress

result: TaskResult | None

subtasks: list[SubTask]

assigned_agent_id: str | None

tags: list[str]

metadata: dict[str, Any]

class taolib.testing.multi_agent.TaskConstraint

Bases: pydantic.BaseModel

任务约束条件。

max_duration_seconds: int | None

max_cost: float | None

required_skills: list[str]

forbidden_skills: list[str]

priority: int

class taolib.testing.multi_agent.TaskCreate

Bases: TaskBase

创建任务的输入模型。

class taolib.testing.multi_agent.TaskDocument

Bases: TaskBase

任务的 MongoDB 文档模型。

id: str

created_at: datetime.datetime

updated_at: datetime.datetime

started_at: datetime.datetime | None = None

completed_at: datetime.datetime | None = None

execution_seconds: float | None = None

model_config

to_response() → TaskResponse

转换为 API 响应。

class taolib.testing.multi_agent.TaskProgress

Bases: pydantic.BaseModel

任务进度。

current_step: str

step_description: str

progress_percent: float

completed_steps: list[str]

estimated_remaining_seconds: int | None

class taolib.testing.multi_agent.TaskResponse

Bases: TaskBase

任务的 API 响应模型。

id: str

created_at: datetime.datetime

updated_at: datetime.datetime

started_at: datetime.datetime | None = None

completed_at: datetime.datetime | None = None

execution_seconds: float | None

model_config

class taolib.testing.multi_agent.TaskResult

Bases: pydantic.BaseModel

任务结果。

success: bool

summary: str

details: dict[str, Any]

output_files: list[str]

artifacts: dict[str, Any]

warnings: list[str]

errors: list[str]

class taolib.testing.multi_agent.TaskStatus

Bases: enum.StrEnum

任务状态。

PENDING = 'pending'

ANALYZING = 'analyzing'

ASSIGNED = 'assigned'

IN_PROGRESS = 'in_progress'

WAITING = 'waiting'

COMPLETED = 'completed'

FAILED = 'failed'

CANCELLED = 'cancelled'

class taolib.testing.multi_agent.TaskUpdate

Bases: pydantic.BaseModel

更新任务的输入模型(所有字段可选)。

name: str | None = None

description: str | None = None

status: taolib.testing.multi_agent.models.enums.TaskStatus | None = None

constraints: TaskConstraint | None = None

progress: TaskProgress | None = None

result: TaskResult | None = None

subtasks: list[SubTask] | None = None

assigned_agent_id: str | None = None

tags: list[str] | None = None

metadata: dict[str, Any] | None = None

class taolib.testing.multi_agent.BaseSkill(skill_id: str, name: str, description: str, parameters: list[taolib.testing.multi_agent.models.SkillParameter])

Bases: Skill

技能基类。

_id

_name

_description

_parameters

property id: str

技能ID。

property name: str

技能名称。

property description: str

技能描述。

property parameters: list[taolib.testing.multi_agent.models.SkillParameter]

技能参数定义。

class taolib.testing.multi_agent.CodeGenerationSkill

Bases: taolib.testing.multi_agent.skills.protocols.BaseSkill

代码生成技能。

async execute(context: taolib.testing.multi_agent.skills.protocols.SkillExecutionContext) → Any

执行代码生成。

参数:

context -- 执行上下文

返回:

生成的代码

返回类型:

Any

class taolib.testing.multi_agent.DataAnalysisSkill

Bases: taolib.testing.multi_agent.skills.protocols.BaseSkill

数据分析技能。

async execute(context: taolib.testing.multi_agent.skills.protocols.SkillExecutionContext) → Any

执行数据分析。

参数:

context -- 执行上下文

返回:

分析结果

返回类型:

Any

class taolib.testing.multi_agent.Skill

Bases: abc.ABC

技能协议。

property id: str

Abstractmethod:

技能ID。

property name: str

Abstractmethod:

技能名称。

property description: str

Abstractmethod:

技能描述。

property parameters: list[taolib.testing.multi_agent.models.SkillParameter]

Abstractmethod:

技能参数定义。

abstractmethod execute(context: SkillExecutionContext) → Any

Async:

执行技能。

参数:

context -- 执行上下文

返回:

执行结果

返回类型:

Any

validate_parameters(parameters: Dict[str, Any]) → tuple[bool, list[str]]

验证参数。

参数:

parameters -- 要验证的参数

返回:

(是否有效, 错误列表)

返回类型:

tuple[bool, list[str]]

class taolib.testing.multi_agent.SkillExecutionContext(parameters: Dict[str, Any], llm_provider: Any = None, agent: Any = None)

技能执行上下文。

parameters

llm_provider = None

agent = None

state: Dict[str, Any]

class taolib.testing.multi_agent.SkillManager(registry: taolib.testing.multi_agent.skills.registry.SkillRegistry | None = None, llm_manager: taolib.testing.multi_agent.llm.LLMManager | None = None)

技能管理器。

_registry = None

_llm_manager = None

_skill_documents: Dict[str, taolib.testing.multi_agent.models.SkillDocument]

_execution_history: List[Dict[str, Any]] = []

register_skill(skill: taolib.testing.multi_agent.skills.protocols.Skill, document: taolib.testing.multi_agent.models.SkillDocument | None = None) → None

注册技能。

参数:

• skill -- 技能实例

• document -- 技能文档,如果为None则自动创建

抛出:

SkillError -- 技能已存在

get_skill(skill_id: str) → taolib.testing.multi_agent.skills.protocols.Skill | None

获取技能。

参数:

skill_id -- 技能ID

返回:

技能实例,如果不存在则返回None

返回类型:

Optional[Skill]

get_skill_document(skill_id: str) → taolib.testing.multi_agent.models.SkillDocument | None

获取技能文档。

参数:

skill_id -- 技能ID

返回:

技能文档,如果不存在则返回None

返回类型:

Optional[SkillDocument]

list_skills(status: taolib.testing.multi_agent.models.SkillStatus | None = None) → List[taolib.testing.multi_agent.models.SkillDocument]

列出技能。

参数:

status -- 技能状态过滤,如果为None则返回所有

返回:

技能文档列表

返回类型:

List[SkillDocument]

async execute_skill(skill_id: str, parameters: Dict[str, Any], agent: Any = None) → Any

执行技能。

参数:

• skill_id -- 技能ID

• parameters -- 技能参数

• agent -- 执行技能的智能体

返回:

执行结果

返回类型:

Any

抛出:

SkillError -- 技能不存在或执行失败

async test_skill(skill_id: str, test_cases: List[Dict[str, Any]]) → taolib.testing.multi_agent.models.SkillTestResult

测试技能。

参数:

• skill_id -- 技能ID

• test_cases -- 测试用例列表,每个用例包含parameters和expected

返回:

测试结果

返回类型:

SkillTestResult

async evaluate_skill(skill_id: str, test_cases: List[Dict[str, Any]] | None = None) → taolib.testing.multi_agent.models.SkillEvaluation

评估技能。

参数:

• skill_id -- 技能ID

• test_cases -- 测试用例,如果为None则使用文档中的测试用例

返回:

评估结果

返回类型:

SkillEvaluation

async discover_skills(task_description: str, existing_skills: List[str] | None = None) → List[str]

发现完成任务所需的技能。

参数:

• task_description -- 任务描述

• existing_skills -- 已有的技能ID列表

返回:

推荐的技能ID列表

返回类型:

List[str]

async create_skill_from_description(skill_create: taolib.testing.multi_agent.models.SkillCreate) → taolib.testing.multi_agent.models.SkillDocument

从描述创建技能。

参数:

skill_create -- 技能创建配置

返回:

创建的技能文档

返回类型:

SkillDocument

备注

这是一个占位实现,实际应该使用LLM生成技能代码

get_execution_history(skill_id: str | None = None, limit: int = 100) → List[Dict[str, Any]]

获取执行历史。

参数:

• skill_id -- 技能ID过滤,如果为None则返回所有

• limit -- 返回记录数量限制

返回:

执行历史记录

返回类型:

List[Dict[str, Any]]

class taolib.testing.multi_agent.SkillRegistry

技能注册表。

_skills: Dict[str, taolib.testing.multi_agent.skills.protocols.Skill]

_skill_classes: Dict[str, Type[taolib.testing.multi_agent.skills.protocols.Skill]]

_skill_paths: Dict[str, pathlib.Path]

register_skill(skill: taolib.testing.multi_agent.skills.protocols.Skill) → None

注册技能实例。

参数:

skill -- 技能实例

抛出:

SkillError -- 技能已存在

register_skill_class(skill_class: Type[taolib.testing.multi_agent.skills.protocols.Skill]) → None

注册技能类。

参数:

skill_class -- 技能类

抛出:

SkillError -- 技能类已存在

get_skill(skill_id: str) → taolib.testing.multi_agent.skills.protocols.Skill | None

获取技能实例。

参数:

skill_id -- 技能ID

返回:

技能实例,如果不存在则返回None

返回类型:

Optional[Skill]

get_skill_class(skill_id: str) → Type[taolib.testing.multi_agent.skills.protocols.Skill] | None

获取技能类。

参数:

skill_id -- 技能ID

返回:

技能类,如果不存在则返回None

返回类型:

Optional[Type[Skill]]

create_skill(skill_id: str, **kwargs) → taolib.testing.multi_agent.skills.protocols.Skill

创建技能实例。

参数:

• skill_id -- 技能ID

• **kwargs -- 技能初始化参数

返回:

创建的技能实例

返回类型:

Skill

抛出:

SkillError -- 技能类不存在

get_all_skills() → List[taolib.testing.multi_agent.skills.protocols.Skill]

获取所有已注册的技能实例。

返回:

技能实例列表

返回类型:

List[Skill]

get_all_skill_classes() → Dict[str, Type[taolib.testing.multi_agent.skills.protocols.Skill]]

获取所有已注册的技能类。

返回:

技能类字典

返回类型:

Dict[str, Type[Skill]]

unregister_skill(skill_id: str) → None

注销技能。

参数:

skill_id -- 技能ID

load_skill_from_file(file_path: pathlib.Path, skill_class_name: str | None = None) → str

从文件加载技能。

参数:

• file_path -- 技能文件路径

• skill_class_name -- 技能类名,如果为None则自动查找

返回:

加载的技能ID

返回类型:

str

抛出:

SkillError -- 加载失败

load_skills_from_directory(directory: pathlib.Path) → List[str]

从目录加载所有技能。

参数:

directory -- 技能目录路径

返回:

加载的技能ID列表

返回类型:

List[str]

clear() → None

清空注册表。

class taolib.testing.multi_agent.TextSummarizationSkill

Bases: taolib.testing.multi_agent.skills.protocols.BaseSkill

文本摘要技能。

async execute(context: taolib.testing.multi_agent.skills.protocols.SkillExecutionContext) → Any

执行文本摘要。

参数:

context -- 执行上下文

返回:

摘要结果

返回类型:

Any

class taolib.testing.multi_agent.TranslationSkill

Bases: taolib.testing.multi_agent.skills.protocols.BaseSkill

翻译技能。

async execute(context: taolib.testing.multi_agent.skills.protocols.SkillExecutionContext) → Any

执行翻译。

参数:

context -- 执行上下文

返回:

翻译结果

返回类型:

Any

taolib.testing.multi_agent.get_preset_skills() → list[taolib.testing.multi_agent.skills.protocols.BaseSkill]

获取所有预设技能。

返回:

预设技能列表

返回类型:

list[BaseSkill]

taolib.testing.multi_agent.get_skill_manager() → SkillManager

获取全局技能管理器。

返回:

技能管理器

返回类型:

SkillManager

taolib.testing.multi_agent.get_skill_registry() → SkillRegistry

获取全局技能注册表。

返回:

技能注册表

返回类型:

SkillRegistry

taolib.testing.multi_agent.set_skill_manager(manager: SkillManager) → None

设置全局技能管理器。

参数:

manager -- 技能管理器

taolib.testing.multi_agent.set_skill_registry(registry: SkillRegistry) → None

设置全局技能注册表。

参数:

registry -- 技能注册表

taolib.testing.multi_agent.__version__ = '0.1.0'