test(providers): add ASV performance benchmark suite for the providers module
CI / lint (pull_request) Successful in 21s
CI / build (pull_request) Successful in 17s
CI / helm (pull_request) Successful in 23s
CI / quality (pull_request) Successful in 44s
CI / typecheck (pull_request) Successful in 1m2s
CI / security (pull_request) Successful in 1m2s
CI / unit_tests (pull_request) Successful in 6m54s
CI / docker (pull_request) Successful in 1m50s
CI / coverage (pull_request) Successful in 11m15s
CI / e2e_tests (pull_request) Successful in 22m16s
CI / integration_tests (pull_request) Successful in 23m31s
CI / status-check (pull_request) Successful in 1s
CI / benchmark-publish (pull_request) Has been skipped
CI / benchmark-regression (pull_request) Successful in 58m37s

Implemented 5 new ASV benchmark files under benchmarks/:

- providers_cost_table_bench.py — ProviderCostTable construction (default +
  custom entries), iteration throughput across all providers/models, fallback
  path for unknown providers
- providers_cost_tracker_bench.py — CostTracker construction with various
  budget configurations, accumulation throughput (10/50 calls, mixed
  providers), daily spend tracking, get_cost_entry delegation
- providers_fallback_selector_bench.py — FallbackSelector construction with
  custom order and cost tracker, selection when no providers configured
  (exhausts full list), selection with configured provider at various positions
- providers_registry_bench.py — ProviderRegistry.get_all_providers,
  get_provider_info (by enum and string), is_provider_configured,
  multi-provider initialization
- providers_llm_adapters_bench.py — LangChainChatProvider,
  AnthropicChatProvider, GoogleChatProvider, OpenAIChatProvider,
  OpenRouterChatProvider instantiation with various configurations

Key design decisions:
- Carefully audited existing cost_controls_bench.py and
  provider_selection_bench.py to avoid duplicating any already-covered
  benchmarks
- Used MagicMock for Settings objects to avoid requiring real API keys in
  benchmarks
- LLM adapter benchmarks use mock factories to measure pure instantiation
  cost without network calls
- All benchmark classes use setup() fixtures to isolate measurement from
  fixture construction
- 68 benchmark methods total across 5 files, all verified to execute without
  errors

ISSUES CLOSED: #2800
This commit is contained in:
2026-04-05 03:55:26 +00:00
parent bbff42ac9a
commit 254fd07496
5 changed files with 846 additions and 0 deletions
+115
View File
@@ -0,0 +1,115 @@
"""ASV benchmarks for ProviderCostTable construction and iteration.
Measures the performance of:
- ProviderCostTable construction with default entries
- ProviderCostTable construction with custom entries merged in
- Iterating all providers and their models (throughput)
- CostEntry.estimate_cost across the full default table
Note: Basic lookup benchmarks (get_cost_entry, list_providers, list_models)
are already covered by cost_controls_bench.py and are not duplicated here.
"""
from __future__ import annotations
import importlib
import sys
from pathlib import Path
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.providers.cost_table import CostEntry, ProviderCostTable # noqa: E402
# ---------------------------------------------------------------------------
# Shared fixtures
# ---------------------------------------------------------------------------
_CUSTOM_ENTRIES: dict[str, dict[str, CostEntry]] = {
"custom_provider": {
"custom-model-v1": CostEntry(
input_cost_per_token=0.000002,
output_cost_per_token=0.000008,
),
"custom-model-v2": CostEntry(
input_cost_per_token=0.000001,
output_cost_per_token=0.000004,
),
},
"openai": {
"gpt-5": CostEntry(
input_cost_per_token=0.00001,
output_cost_per_token=0.00003,
),
},
}
class TimeCostTableConstruction:
"""Benchmark ProviderCostTable construction cost."""
def time_construct_default(self) -> None:
"""Time construction of a cost table with only default entries."""
ProviderCostTable()
def time_construct_with_custom_entries(self) -> None:
"""Time construction of a cost table with custom entries merged in."""
ProviderCostTable(custom_entries=_CUSTOM_ENTRIES)
def time_construct_with_empty_custom(self) -> None:
"""Time construction with an empty custom entries dict (no-op merge)."""
ProviderCostTable(custom_entries={})
class TimeCostTableIteration:
"""Benchmark iterating all providers and models in the cost table."""
def setup(self) -> None:
self.table = ProviderCostTable()
self.table_with_custom = ProviderCostTable(custom_entries=_CUSTOM_ENTRIES)
def time_iterate_all_providers_and_models(self) -> None:
"""Time iterating every provider and every model in the default table."""
for provider in self.table.list_providers():
for model in self.table.list_models(provider):
self.table.get_cost_entry(provider, model)
def time_iterate_all_providers_and_models_custom(self) -> None:
"""Time iterating every provider and model in a table with custom entries."""
for provider in self.table_with_custom.list_providers():
for model in self.table_with_custom.list_models(provider):
self.table_with_custom.get_cost_entry(provider, model)
def time_estimate_cost_all_entries(self) -> None:
"""Time calling estimate_cost for every known provider/model pair."""
for provider in self.table.list_providers():
for model in self.table.list_models(provider):
entry = self.table.get_cost_entry(provider, model)
entry.estimate_cost(input_tokens=1000, output_tokens=500)
class TimeCostTableFallback:
"""Benchmark the fallback path for unknown provider/model combinations."""
def setup(self) -> None:
self.table = ProviderCostTable()
def time_lookup_unknown_provider(self) -> None:
"""Time lookup for a completely unknown provider (returns DEFAULT_COST)."""
self.table.get_cost_entry("nonexistent_provider", "nonexistent_model")
def time_lookup_known_provider_unknown_model(self) -> None:
"""Time lookup for a known provider but unknown model (returns DEFAULT_COST)."""
self.table.get_cost_entry("anthropic", "claude-99-ultra")
def time_default_cost_estimate(self) -> None:
"""Time estimating cost using the DEFAULT_COST fallback entry."""
ProviderCostTable.DEFAULT_COST.estimate_cost(
input_tokens=10_000, output_tokens=5_000
)
+180
View File
@@ -0,0 +1,180 @@
"""ASV benchmarks for CostTracker accumulation and daily spend tracking.
Measures the performance of:
- CostTracker construction with various budget configurations
- Accumulating many record_usage calls (throughput)
- Daily spend tracking and retrieval
- get_cost_entry delegation to the underlying cost table
- Budget evaluation across multiple accumulated calls
Note: Single-call record_usage, check_plan_budget, check_daily_budget, and
estimate_cost are already covered by cost_controls_bench.py and are not
duplicated here.
"""
from __future__ import annotations
import importlib
import sys
from pathlib import Path
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.domain.models.core.cost_metadata import CostMetadata # noqa: E402
from cleveragents.providers.cost_table import CostEntry, ProviderCostTable # noqa: E402
from cleveragents.providers.cost_tracker import CostTracker # noqa: E402
# ---------------------------------------------------------------------------
# Shared fixtures
# ---------------------------------------------------------------------------
_CUSTOM_TABLE = ProviderCostTable(
custom_entries={
"bench_provider": {
"bench-model": CostEntry(
input_cost_per_token=0.000001,
output_cost_per_token=0.000002,
)
}
}
)
_PROVIDERS_AND_MODELS = [
("openai", "gpt-4o"),
("anthropic", "claude-sonnet-4-20250514"),
("google", "gemini-2.0-flash"),
("groq", "llama-3.1-70b-versatile"),
("cohere", "command-r-plus"),
]
class TimeCostTrackerConstruction:
"""Benchmark CostTracker construction with various budget configurations."""
def time_construct_no_budget(self) -> None:
"""Time construction with no budget limits (unlimited)."""
CostTracker()
def time_construct_plan_budget_only(self) -> None:
"""Time construction with only a per-plan budget."""
CostTracker(budget_per_plan=10.0)
def time_construct_daily_budget_only(self) -> None:
"""Time construction with only a per-day budget."""
CostTracker(budget_per_day=50.0)
def time_construct_both_budgets(self) -> None:
"""Time construction with both plan and daily budgets."""
CostTracker(budget_per_plan=10.0, budget_per_day=50.0)
def time_construct_with_custom_cost_table(self) -> None:
"""Time construction with a pre-built custom cost table injected."""
CostTracker(budget_per_plan=5.0, cost_table=_CUSTOM_TABLE)
class TimeCostTrackerAccumulation:
"""Benchmark CostTracker accumulation throughput across many calls."""
def setup(self) -> None:
self.tracker = CostTracker(budget_per_plan=1000.0, budget_per_day=5000.0)
self.tracker_unlimited = CostTracker()
def time_accumulate_10_calls(self) -> None:
"""Time accumulating 10 record_usage calls on a fresh CostMetadata."""
metadata = CostMetadata()
for _ in range(10):
self.tracker.record_usage(
metadata,
provider="openai",
model="gpt-4o",
input_tokens=500,
output_tokens=250,
)
def time_accumulate_50_calls(self) -> None:
"""Time accumulating 50 record_usage calls on a fresh CostMetadata."""
metadata = CostMetadata()
for _ in range(50):
self.tracker.record_usage(
metadata,
provider="openai",
model="gpt-4o",
input_tokens=500,
output_tokens=250,
)
def time_accumulate_mixed_providers(self) -> None:
"""Time accumulating calls across multiple providers and models."""
metadata = CostMetadata()
for provider, model in _PROVIDERS_AND_MODELS:
self.tracker.record_usage(
metadata,
provider=provider,
model=model,
input_tokens=1000,
output_tokens=500,
)
def time_accumulate_unlimited_budget(self) -> None:
"""Time accumulating 10 calls with no budget limits (no threshold checks)."""
metadata = CostMetadata()
for _ in range(10):
self.tracker_unlimited.record_usage(
metadata,
provider="openai",
model="gpt-4o",
input_tokens=500,
output_tokens=250,
)
class TimeCostTrackerDailySpend:
"""Benchmark daily spend tracking and retrieval."""
def setup(self) -> None:
self.tracker = CostTracker(budget_per_day=100.0)
metadata = CostMetadata()
# Pre-populate with some daily spend
for _ in range(5):
self.tracker.record_usage(
metadata,
provider="openai",
model="gpt-4o",
input_tokens=1000,
output_tokens=500,
)
def time_get_daily_spend(self) -> None:
"""Time retrieving the current day's accumulated spend."""
self.tracker.get_daily_spend()
def time_check_daily_budget_with_spend(self) -> None:
"""Time checking daily budget after some spend has been accumulated."""
self.tracker.check_daily_budget()
class TimeCostTrackerGetCostEntry:
"""Benchmark CostTracker.get_cost_entry delegation."""
def setup(self) -> None:
self.tracker = CostTracker()
def time_get_cost_entry_known(self) -> None:
"""Time get_cost_entry for a known provider/model pair."""
self.tracker.get_cost_entry("openai", "gpt-4o")
def time_get_cost_entry_unknown(self) -> None:
"""Time get_cost_entry for an unknown provider/model (returns default)."""
self.tracker.get_cost_entry("unknown_provider", "unknown_model")
def time_get_cost_entry_all_known(self) -> None:
"""Time get_cost_entry for all known provider/model pairs."""
for provider, model in _PROVIDERS_AND_MODELS:
self.tracker.get_cost_entry(provider, model)
@@ -0,0 +1,166 @@
"""ASV benchmarks for FallbackSelector with varied configurations.
Measures the performance of:
- FallbackSelector construction with custom fallback order
- FallbackSelector construction with an injected CostTracker
- Selection when all providers are unconfigured (exhausts full list)
- Selection with a custom provider order
- Selection with a cost tracker attached
Note: Basic select() calls (no requirements, with tool_calls, with all
requirements) against a default registry are already covered by
cost_controls_bench.py and are not duplicated here.
"""
from __future__ import annotations
import importlib
import sys
from pathlib import Path
from unittest.mock import MagicMock
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.providers.cost_tracker import CostTracker # noqa: E402
from cleveragents.providers.fallback_selector import FallbackSelector # noqa: E402
from cleveragents.providers.registry import ( # noqa: E402
ProviderRegistry,
reset_provider_registry,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_settings(**overrides: object) -> MagicMock:
"""Build a minimal mock Settings object for registry construction."""
settings = MagicMock()
settings.openai_api_key = None
settings.anthropic_api_key = None
settings.google_api_key = None
settings.gemini_api_key = None
settings.azure_api_key = None
settings.openrouter_api_key = None
settings.cohere_api_key = None
settings.groq_api_key = None
settings.together_api_key = None
settings.default_provider = None
settings.default_model = None
settings.azure_openai_endpoint = None
settings.azure_openai_api_version = None
settings.azure_openai_deployment = None
settings.openrouter_organization = None
settings.mock_providers = False
for key, value in overrides.items():
setattr(settings, key, value)
return settings
class TimeFallbackSelectorConstruction:
"""Benchmark FallbackSelector construction with various configurations."""
def setup(self) -> None:
reset_provider_registry()
self.registry = ProviderRegistry(settings=_make_settings())
self.tracker = CostTracker(budget_per_plan=10.0, budget_per_day=50.0)
def time_construct_default_order(self) -> None:
"""Time construction with the default fallback order."""
FallbackSelector(registry=self.registry)
def time_construct_custom_order(self) -> None:
"""Time construction with a custom provider fallback order."""
FallbackSelector(
registry=self.registry,
fallback_providers=["anthropic", "openai", "groq"],
)
def time_construct_with_cost_tracker(self) -> None:
"""Time construction with a CostTracker injected."""
FallbackSelector(registry=self.registry, cost_tracker=self.tracker)
def time_construct_single_provider_order(self) -> None:
"""Time construction with a single-provider fallback list."""
FallbackSelector(
registry=self.registry,
fallback_providers=["openai"],
)
class TimeFallbackSelectorNoProviders:
"""Benchmark selection when no providers are configured (exhausts full list)."""
def setup(self) -> None:
reset_provider_registry()
# No API keys set — all providers unconfigured
self.registry = ProviderRegistry(settings=_make_settings())
self.selector = FallbackSelector(registry=self.registry)
self.selector_short = FallbackSelector(
registry=self.registry,
fallback_providers=["openai", "anthropic"],
)
def time_select_exhausts_all_providers(self) -> None:
"""Time selection that must iterate through all providers and find none."""
self.selector.select()
def time_select_exhausts_short_list(self) -> None:
"""Time selection that exhausts a short 2-provider list."""
self.selector_short.select()
def time_select_with_requirements_no_providers(self) -> None:
"""Time selection with capability requirements when no providers exist."""
self.selector.select(
require_tool_calls=True,
require_streaming=True,
)
class TimeFallbackSelectorWithConfiguredProvider:
"""Benchmark selection when one provider is configured."""
def setup(self) -> None:
reset_provider_registry()
# OpenAI configured, all others not
self.registry = ProviderRegistry(
settings=_make_settings(openai_api_key="sk-bench-key")
)
self.selector_default = FallbackSelector(registry=self.registry)
# Custom order: openai is last — must skip all others first
self.selector_openai_last = FallbackSelector(
registry=self.registry,
fallback_providers=["anthropic", "google", "groq", "openai"],
)
self.tracker = CostTracker(budget_per_plan=100.0)
self.selector_with_tracker = FallbackSelector(
registry=self.registry,
cost_tracker=self.tracker,
)
def time_select_first_provider_matches(self) -> None:
"""Time selection where the first provider in the list is configured."""
self.selector_default.select()
def time_select_last_provider_matches(self) -> None:
"""Time selection where the configured provider is last in the list."""
self.selector_openai_last.select()
def time_select_with_cost_tracker_under_budget(self) -> None:
"""Time selection with a cost tracker when budget is not exceeded."""
self.selector_with_tracker.select()
def time_select_vision_requirement_openai(self) -> None:
"""Time selection requiring vision support (OpenAI supports it)."""
self.selector_default.select(require_vision=True)
def time_select_json_mode_requirement_openai(self) -> None:
"""Time selection requiring JSON mode (OpenAI supports it)."""
self.selector_default.select(require_json_mode=True)
+206
View File
@@ -0,0 +1,206 @@
"""ASV benchmarks for LLM provider adapter instantiation.
Measures the performance of:
- LangChainChatProvider instantiation (base adapter)
- AnthropicChatProvider instantiation
- GoogleChatProvider instantiation
- OpenAIChatProvider instantiation
- OpenRouterChatProvider instantiation
These benchmarks focus on the construction cost of each adapter class,
which is the performance-sensitive path exercised before every LLM call.
No actual network calls are made — all LLM factories are mocked.
"""
from __future__ import annotations
import importlib
import sys
from pathlib import Path
from unittest.mock import MagicMock
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.providers.llm.anthropic_provider import ( # noqa: E402
AnthropicChatProvider,
)
from cleveragents.providers.llm.google_provider import GoogleChatProvider # noqa: E402
from cleveragents.providers.llm.langchain_chat_provider import ( # noqa: E402
LangChainChatProvider,
)
from cleveragents.providers.llm.openai_provider import OpenAIChatProvider # noqa: E402
from cleveragents.providers.llm.openrouter_provider import ( # noqa: E402
OpenRouterChatProvider,
)
# ---------------------------------------------------------------------------
# Shared mock LLM factory
# ---------------------------------------------------------------------------
def _mock_llm_factory(model_id: str) -> MagicMock:
"""Return a MagicMock that satisfies the BaseLanguageModel interface."""
return MagicMock()
class TimeLangChainChatProviderInstantiation:
"""Benchmark LangChainChatProvider (base adapter) instantiation."""
def time_instantiate_default(self) -> None:
"""Time instantiation with default settings."""
LangChainChatProvider(
name="bench_provider",
model_id="bench-model",
llm_factory=_mock_llm_factory,
)
def time_instantiate_with_max_retries(self) -> None:
"""Time instantiation with a custom max_retries value."""
LangChainChatProvider(
name="bench_provider",
model_id="bench-model",
llm_factory=_mock_llm_factory,
max_retries=5,
)
def time_instantiate_no_streaming(self) -> None:
"""Time instantiation with streaming disabled."""
LangChainChatProvider(
name="bench_provider",
model_id="bench-model",
llm_factory=_mock_llm_factory,
supports_streaming=False,
)
def time_instantiate_with_progress_map(self) -> None:
"""Time instantiation with a custom progress map."""
LangChainChatProvider(
name="bench_provider",
model_id="bench-model",
llm_factory=_mock_llm_factory,
progress_map={
"load_context": 10,
"analyze_requirements": 30,
"generate_plan": 60,
"validate": 85,
},
)
class TimeAnthropicChatProviderInstantiation:
"""Benchmark AnthropicChatProvider instantiation."""
def time_instantiate_default_model(self) -> None:
"""Time instantiation with the default Claude model."""
AnthropicChatProvider(api_key="ant-bench-key")
def time_instantiate_custom_model(self) -> None:
"""Time instantiation with a custom model identifier."""
AnthropicChatProvider(
api_key="ant-bench-key",
model="claude-3-5-haiku-20241022",
)
def time_instantiate_custom_retries(self) -> None:
"""Time instantiation with a non-default max_retries value."""
AnthropicChatProvider(
api_key="ant-bench-key",
max_retries=1,
)
class TimeGoogleChatProviderInstantiation:
"""Benchmark GoogleChatProvider instantiation."""
def time_instantiate_default_model(self) -> None:
"""Time instantiation with the default Gemini model."""
GoogleChatProvider(api_key="goog-bench-key")
def time_instantiate_custom_model(self) -> None:
"""Time instantiation with a custom Gemini model identifier."""
GoogleChatProvider(
api_key="goog-bench-key",
model="gemini-1.5-pro",
)
def time_instantiate_custom_retries(self) -> None:
"""Time instantiation with a non-default max_retries value."""
GoogleChatProvider(
api_key="goog-bench-key",
max_retries=2,
)
class TimeOpenAIChatProviderInstantiation:
"""Benchmark OpenAIChatProvider instantiation."""
def time_instantiate_default_model(self) -> None:
"""Time instantiation with the default GPT-4o model."""
OpenAIChatProvider(api_key="sk-bench-key")
def time_instantiate_custom_model(self) -> None:
"""Time instantiation with a custom model identifier."""
OpenAIChatProvider(
api_key="sk-bench-key",
model="gpt-4o-mini",
)
def time_instantiate_with_organization(self) -> None:
"""Time instantiation with an organization identifier."""
OpenAIChatProvider(
api_key="sk-bench-key",
organization="org-bench-12345",
)
def time_instantiate_custom_retries(self) -> None:
"""Time instantiation with a non-default max_retries value."""
OpenAIChatProvider(
api_key="sk-bench-key",
max_retries=1,
)
class TimeOpenRouterChatProviderInstantiation:
"""Benchmark OpenRouterChatProvider instantiation."""
def time_instantiate_default_model(self) -> None:
"""Time instantiation with the default OpenRouter model."""
OpenRouterChatProvider(api_key="or-bench-key")
def time_instantiate_custom_model(self) -> None:
"""Time instantiation with a custom model identifier."""
OpenRouterChatProvider(
api_key="or-bench-key",
model="openai/gpt-4o",
)
def time_instantiate_with_organization(self) -> None:
"""Time instantiation with an organization/referer header."""
OpenRouterChatProvider(
api_key="or-bench-key",
organization="https://bench.example.com",
)
def time_instantiate_with_custom_headers(self) -> None:
"""Time instantiation with custom default headers."""
OpenRouterChatProvider(
api_key="or-bench-key",
default_headers={
"HTTP-Referer": "https://bench.example.com",
"X-Title": "BenchmarkSuite",
},
)
def time_instantiate_custom_retries(self) -> None:
"""Time instantiation with a non-default max_retries value."""
OpenRouterChatProvider(
api_key="or-bench-key",
max_retries=5,
)
+179
View File
@@ -0,0 +1,179 @@
"""ASV benchmarks for ProviderRegistry operations beyond initialization.
Measures the performance of:
- get_all_providers (all known providers, configured or not)
- get_provider_info (by ProviderType and by string name)
- is_provider_configured (by ProviderType and by string name)
- ProviderRegistry with multiple providers configured
Note: Registry initialization, get_default_provider_type,
get_configured_providers, and get_default_model are already covered by
provider_selection_bench.py and are not duplicated here.
"""
from __future__ import annotations
import importlib
import sys
from pathlib import Path
from unittest.mock import MagicMock
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.providers.registry import ( # noqa: E402
ProviderRegistry,
ProviderType,
reset_provider_registry,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_settings(**overrides: object) -> MagicMock:
"""Build a minimal mock Settings object for registry construction."""
settings = MagicMock()
settings.openai_api_key = None
settings.anthropic_api_key = None
settings.google_api_key = None
settings.gemini_api_key = None
settings.azure_api_key = None
settings.openrouter_api_key = None
settings.cohere_api_key = None
settings.groq_api_key = None
settings.together_api_key = None
settings.default_provider = None
settings.default_model = None
settings.azure_openai_endpoint = None
settings.azure_openai_api_version = None
settings.azure_openai_deployment = None
settings.openrouter_organization = None
settings.mock_providers = False
for key, value in overrides.items():
setattr(settings, key, value)
return settings
class TimeRegistryGetAllProviders:
"""Benchmark ProviderRegistry.get_all_providers."""
def setup(self) -> None:
reset_provider_registry()
self.registry_none = ProviderRegistry(settings=_make_settings())
self.registry_one = ProviderRegistry(
settings=_make_settings(openai_api_key="sk-bench-key")
)
self.registry_multi = ProviderRegistry(
settings=_make_settings(
openai_api_key="sk-bench-key",
anthropic_api_key="ant-bench-key",
google_api_key="goog-bench-key",
)
)
def time_get_all_providers_none_configured(self) -> None:
"""Time get_all_providers when no providers are configured."""
self.registry_none.get_all_providers()
def time_get_all_providers_one_configured(self) -> None:
"""Time get_all_providers when one provider is configured."""
self.registry_one.get_all_providers()
def time_get_all_providers_multi_configured(self) -> None:
"""Time get_all_providers when multiple providers are configured."""
self.registry_multi.get_all_providers()
class TimeRegistryGetProviderInfo:
"""Benchmark ProviderRegistry.get_provider_info lookups."""
def setup(self) -> None:
reset_provider_registry()
self.registry = ProviderRegistry(
settings=_make_settings(openai_api_key="sk-bench-key")
)
def time_get_provider_info_by_enum(self) -> None:
"""Time get_provider_info using a ProviderType enum value."""
self.registry.get_provider_info(ProviderType.OPENAI)
def time_get_provider_info_by_string(self) -> None:
"""Time get_provider_info using a string name (requires coercion)."""
self.registry.get_provider_info("openai")
def time_get_provider_info_unknown_string(self) -> None:
"""Time get_provider_info for an unknown provider name (returns None)."""
self.registry.get_provider_info("nonexistent_provider")
def time_get_provider_info_all_types(self) -> None:
"""Time get_provider_info for every known ProviderType."""
for provider_type in ProviderType:
self.registry.get_provider_info(provider_type)
class TimeRegistryIsProviderConfigured:
"""Benchmark ProviderRegistry.is_provider_configured checks."""
def setup(self) -> None:
reset_provider_registry()
self.registry = ProviderRegistry(
settings=_make_settings(
openai_api_key="sk-bench-key",
anthropic_api_key="ant-bench-key",
)
)
def time_is_configured_true_by_enum(self) -> None:
"""Time is_provider_configured for a configured provider (enum)."""
self.registry.is_provider_configured(ProviderType.OPENAI)
def time_is_configured_false_by_enum(self) -> None:
"""Time is_provider_configured for an unconfigured provider (enum)."""
self.registry.is_provider_configured(ProviderType.GROQ)
def time_is_configured_true_by_string(self) -> None:
"""Time is_provider_configured for a configured provider (string)."""
self.registry.is_provider_configured("anthropic")
def time_is_configured_false_by_string(self) -> None:
"""Time is_provider_configured for an unconfigured provider (string)."""
self.registry.is_provider_configured("cohere")
def time_is_configured_all_types(self) -> None:
"""Time is_provider_configured for every known ProviderType."""
for provider_type in ProviderType:
self.registry.is_provider_configured(provider_type)
class TimeRegistryMultiProviderInit:
"""Benchmark ProviderRegistry initialization with many providers configured."""
def setup(self) -> None:
self.settings_all = _make_settings(
openai_api_key="sk-bench-key",
anthropic_api_key="ant-bench-key",
google_api_key="goog-bench-key",
groq_api_key="groq-bench-key",
together_api_key="together-bench-key",
cohere_api_key="cohere-bench-key",
openrouter_api_key="or-bench-key",
)
def time_init_all_providers_configured(self) -> None:
"""Time registry initialization when all major providers are configured."""
reset_provider_registry()
ProviderRegistry(settings=self.settings_all)
def time_get_configured_providers_all(self) -> None:
"""Time get_configured_providers when all major providers are configured."""
reset_provider_registry()
registry = ProviderRegistry(settings=self.settings_all)
registry.get_configured_providers()