"""ASV benchmarks for ScoredFragment and ContextFragment pipeline models. Measures the performance of: - ScoredFragment creation (with and without breakdown) - ScoredFragment factory methods - ScoredFragment equality comparison - ScoredFragment set-based deduplication - ContextFragment with strategy_source """ 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.contexts.fragment import ScoredFragment # noqa: E402 from cleveragents.domain.models.core.context_fragment import ( # noqa: E402 ContextFragment, FragmentProvenance, ) _DEFAULT_PROV = FragmentProvenance(resource_uri="bench://default") def _make_fragment(**kwargs: object) -> ContextFragment: """Create a ContextFragment with sensible benchmark defaults.""" defaults: dict[str, object] = { "uko_node": "bench://file", "content": "benchmark content", "token_count": 10, "provenance": _DEFAULT_PROV, } defaults.update(kwargs) return ContextFragment(**defaults) # type: ignore[arg-type] class ScoredFragmentCreationSuite: """Benchmark ScoredFragment creation throughput.""" def setup(self) -> None: """Pre-create fragment for repeated scored-fragment construction.""" self._fragment = _make_fragment() self._fragment_with_relevance = _make_fragment(relevance_score=0.85) self._breakdown = { "relevance": 0.9, "hierarchy": 0.8, "recency": 0.95, } def time_scored_fragment_minimal(self) -> None: """Benchmark creating a ScoredFragment with minimal fields.""" ScoredFragment( fragment=self._fragment, composite_score=0.85, ) def time_scored_fragment_with_breakdown(self) -> None: """Benchmark creating a ScoredFragment with full breakdown.""" ScoredFragment( fragment=self._fragment, composite_score=0.92, score_breakdown=self._breakdown, rank=1, ) def time_scored_fragment_from_relevance(self) -> None: """Benchmark ScoredFragment.from_relevance factory.""" ScoredFragment.from_relevance(self._fragment_with_relevance) def time_scored_fragment_from_factory(self) -> None: """Benchmark ScoredFragment.from_fragment factory.""" ScoredFragment.from_fragment( self._fragment, composite_score=0.6, score_breakdown=self._breakdown, rank=3, ) class ScoredFragmentEqualitySuite: """Benchmark ScoredFragment equality and hashing.""" def setup(self) -> None: """Pre-create scored fragments for comparison benchmarks.""" frag_a = _make_fragment( uko_node="bench://file/a", content="A", detail_depth=3, ) frag_b = _make_fragment( uko_node="bench://file/a", content="B", detail_depth=3, ) self._scored_a = ScoredFragment.from_fragment(frag_a, composite_score=0.5) self._scored_b = ScoredFragment.from_fragment(frag_b, composite_score=0.9) # Build a list of 100 fragments (50 unique uko_node values) self._scored_list = [ ScoredFragment.from_fragment( _make_fragment( uko_node=f"bench://file/{i % 50}", content=f"content {i}", detail_depth=3, ), composite_score=round((i % 10) / 10, 1), ) for i in range(100) ] def time_equality_check(self) -> None: """Benchmark equality comparison.""" _ = self._scored_a == self._scored_b def time_hash_computation(self) -> None: """Benchmark hash computation.""" hash(self._scored_a) def time_set_deduplication_100(self) -> None: """Benchmark set-based deduplication of 100 scored fragments.""" set(self._scored_list) class ContextFragmentStrategySuite: """Benchmark ContextFragment with strategy_source field.""" def time_fragment_with_strategy_source(self) -> None: """Benchmark creating a ContextFragment with strategy_source.""" ContextFragment( uko_node="bench://file", content="benchmark content", token_count=10, provenance=_DEFAULT_PROV, strategy_source="keyword-search", ) def time_fragment_default_strategy_source(self) -> None: """Benchmark creating a ContextFragment with default strategy_source.""" ContextFragment( uko_node="bench://file", content="benchmark content", token_count=10, provenance=_DEFAULT_PROV, )