taolib.testing.multi_agent.llm

taolib.testing.multi_agent.llm#

LLM模型集成模块。

提供无需API KEY的免费大模型集成。

Submodules#

Classes#

`LoadBalancer`	LLM模型负载均衡器。
`LLMManager`	LLM模型管理器。
`BaseLLMProvider`	LLM模型提供商基类。
`LLMProvider`	LLM模型提供商协议。
`ModelRegistry`	LLM模型注册表。

Functions#

`get_llm_manager`(→ LLMManager)	获取全局LLM管理器。
`set_llm_manager`(→ None)	设置全局LLM管理器。

Package Contents#

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

taolib.testing.multi_agent.llm.get_llm_manager() → LLMManager#

获取全局LLM管理器。

返回:: LLM管理器
返回类型:: LLMManager

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

设置全局LLM管理器。

参数:: manager -- LLM管理器

class taolib.testing.multi_agent.llm.BaseLLMProvider(config: taolib.testing.multi_agent.models.ModelConfig)#

Bases: abc.ABC

LLM模型提供商基类。

_config#

_stats#

property config: taolib.testing.multi_agent.models.ModelConfig#: 获取模型配置。

property stats: taolib.testing.multi_agent.models.ModelStats#: 获取模型统计信息。

abstractmethod health_check() → bool#

Async:

检查模型是否健康可用。

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

Async:

同步生成文本。

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

Async:

流式生成文本。

_update_stats_success(latency_seconds: float, tokens_used: int = 0) → None#

更新成功请求的统计信息。

参数:

latency_seconds -- 请求延迟(秒)
tokens_used -- 使用的token数

_update_stats_failure() → None#: 更新失败请求的统计信息。

class taolib.testing.multi_agent.llm.LLMProvider#

Bases: Protocol

LLM模型提供商协议。

property config: taolib.testing.multi_agent.models.ModelConfig#: 获取模型配置。

property stats: taolib.testing.multi_agent.models.ModelStats#: 获取模型统计信息。

async health_check() → bool#

检查模型是否健康可用。

返回:: 模型是否健康可用
返回类型:: bool

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

同步生成文本。

参数:

prompt -- 用户输入的提示词
temperature -- 温度参数,如果为None则使用默认配置
max_tokens -- 最大生成token数,如果为None则使用默认配置
system_prompt -- 系统提示词,如果为None则使用默认配置
**kwargs -- 其他模型特定参数

返回:

生成的文本

返回类型:

str

抛出:

LLMError -- 生成过程中出错

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

流式生成文本。

参数:

prompt -- 用户输入的提示词
temperature -- 温度参数,如果为None则使用默认配置
max_tokens -- 最大生成token数,如果为None则使用默认配置
system_prompt -- 系统提示词,如果为None则使用默认配置
**kwargs -- 其他模型特定参数

生成器:

str -- 生成的文本片段

抛出:

LLMError -- 生成过程中出错

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

目录

taolib.testing.multi_agent.llm#

Submodules#

Classes#

Functions#

Package Contents#