taolib.testing.multi_agent.llm.load_balancer

taolib.testing.multi_agent.llm.load_balancer#

LLM模型负载均衡器。

实现模型自动选择与负载均衡机制。

Classes#

LoadBalancer

LLM模型负载均衡器。

Module Contents#

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