forked from cleveragents/cleveragents-core
117 lines
3.2 KiB
Python
117 lines
3.2 KiB
Python
"""ASV benchmarks for context strategy registry.
|
|
|
|
Measures registry lookup, registration, and can_handle performance
|
|
against spec targets (can_handle < 20ms per strategy,
|
|
specification.md:43245-43261).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Ensure src is on sys.path for ASV
|
|
_SRC = Path(__file__).resolve().parents[1] / "src"
|
|
if str(_SRC) not in sys.path:
|
|
sys.path.insert(0, str(_SRC))
|
|
importlib.invalidate_caches()
|
|
import cleveragents
|
|
|
|
importlib.reload(cleveragents)
|
|
|
|
from cleveragents.application.services.strategy_registry import ( # noqa: E402
|
|
StrategyRegistry,
|
|
)
|
|
from cleveragents.domain.models.acms.crp import ContextRequest # noqa: E402
|
|
from cleveragents.domain.models.acms.strategy import ( # noqa: E402
|
|
BackendSet,
|
|
StrategyConfig,
|
|
)
|
|
from cleveragents.domain.models.acms.strategy_stubs import ( # noqa: E402
|
|
BUILTIN_STRATEGY_CLASSES,
|
|
DEFAULT_ENABLED_STRATEGIES,
|
|
SimpleKeywordStrategy,
|
|
)
|
|
from cleveragents.domain.models.acms.stubs import ( # noqa: E402
|
|
InMemoryGraphBackend,
|
|
InMemoryTextBackend,
|
|
InMemoryVectorBackend,
|
|
)
|
|
|
|
|
|
class TimeRegistration:
|
|
"""Benchmark strategy registration performance."""
|
|
|
|
timeout = 10
|
|
repeat = 3
|
|
number = 100
|
|
|
|
def time_register_single(self) -> None:
|
|
"""Register a single strategy."""
|
|
registry = StrategyRegistry()
|
|
registry.register(SimpleKeywordStrategy(), is_builtin=True)
|
|
|
|
def time_register_all_builtins(self) -> None:
|
|
"""Register all 6 built-in strategies."""
|
|
registry = StrategyRegistry()
|
|
for cls in BUILTIN_STRATEGY_CLASSES:
|
|
inst = cls()
|
|
registry.register(inst, is_builtin=True)
|
|
|
|
|
|
class TimeLookup:
|
|
"""Benchmark registry lookup performance."""
|
|
|
|
timeout = 10
|
|
repeat = 3
|
|
number = 1000
|
|
|
|
def setup(self) -> None:
|
|
self.registry = StrategyRegistry()
|
|
for cls in BUILTIN_STRATEGY_CLASSES:
|
|
inst = cls()
|
|
enabled = inst.name in DEFAULT_ENABLED_STRATEGIES
|
|
self.registry.register(
|
|
inst,
|
|
config=StrategyConfig(enabled=enabled),
|
|
is_builtin=True,
|
|
)
|
|
|
|
def time_get_by_name(self) -> None:
|
|
"""Look up a strategy by name."""
|
|
self.registry.get("simple-keyword")
|
|
|
|
def time_list_enabled(self) -> None:
|
|
"""List enabled strategies."""
|
|
self.registry.list_enabled()
|
|
|
|
def time_list_all(self) -> None:
|
|
"""List all strategies."""
|
|
self.registry.list_all()
|
|
|
|
|
|
class TimeCanHandle:
|
|
"""Benchmark can_handle performance (spec target: < 20ms).
|
|
|
|
Spec: specification.md:43245-43261.
|
|
"""
|
|
|
|
timeout = 10
|
|
repeat = 3
|
|
number = 1000
|
|
|
|
def setup(self) -> None:
|
|
self.request = ContextRequest(query="test query")
|
|
self.backends_all = BackendSet(
|
|
text=InMemoryTextBackend(),
|
|
vector=InMemoryVectorBackend(),
|
|
graph=InMemoryGraphBackend(),
|
|
)
|
|
self.strategies = [cls() for cls in BUILTIN_STRATEGY_CLASSES]
|
|
|
|
def time_can_handle_all_strategies(self) -> None:
|
|
"""Poll can_handle on all 6 built-in strategies."""
|
|
for strategy in self.strategies:
|
|
strategy.can_handle(self.request, self.backends_all)
|