cleveragents-core/benchmarks/acms_backends_bench.py

"""ASV benchmarks for ACMS Backend Abstraction Layer.

Measures the performance of:
- InMemoryTextBackend.search() stub query overhead
- InMemoryVectorBackend.similarity_search() stub query overhead
- InMemoryGraphBackend.sparql_query() / get_triples() / traverse() overhead
- Backend instantiation time
- Result dataclass construction overhead
"""

from __future__ import annotations

import importlib
import sys
from pathlib import Path

# Ensure the local *source* tree is importable even when ASV has an
# older build of the package installed.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

# Force-reload so ASV picks up the source tree version.
import cleveragents  # noqa: E402

importlib.reload(cleveragents)

from cleveragents.domain.models.acms.backends import (  # noqa: E402
    GraphResult,
    TextResult,
    VectorResult,
)
from cleveragents.domain.models.acms.stubs import (  # noqa: E402
    InMemoryGraphBackend,
    InMemoryTextBackend,
    InMemoryVectorBackend,
)

# ---------------------------------------------------------------------------
# Backend instantiation benchmarks
# ---------------------------------------------------------------------------


class BackendInstantiationSuite:
    """Benchmark backend object creation overhead."""

    timeout = 60

    def time_create_text_backend(self) -> None:
        InMemoryTextBackend()

    def time_create_vector_backend(self) -> None:
        InMemoryVectorBackend()

    def time_create_graph_backend(self) -> None:
        InMemoryGraphBackend()


# ---------------------------------------------------------------------------
# Stub query overhead benchmarks
# ---------------------------------------------------------------------------


class TextBackendQuerySuite:
    """Benchmark InMemoryTextBackend.search() stub overhead."""

    timeout = 60

    def setup(self) -> None:
        self.backend = InMemoryTextBackend()
        self.scope: frozenset[str] = frozenset({"RES01", "RES02"})

    def time_search_simple(self) -> None:
        self.backend.search("auth flow", scope=self.scope)

    def time_search_with_max_results(self) -> None:
        self.backend.search("auth flow", scope=self.scope, max_results=5)

    def time_search_large_scope(self) -> None:
        large_scope: frozenset[str] = frozenset(f"RES{i:04d}" for i in range(1000))
        self.backend.search("query", scope=large_scope)


class VectorBackendQuerySuite:
    """Benchmark InMemoryVectorBackend.similarity_search() stub overhead."""

    timeout = 60

    def setup(self) -> None:
        self.backend = InMemoryVectorBackend()
        self.scope: frozenset[str] = frozenset({"RES01", "RES02"})
        self.embedding: list[float] = [0.1] * 768

    def time_similarity_search_simple(self) -> None:
        self.backend.similarity_search(self.embedding, scope=self.scope)

    def time_similarity_search_with_top_k(self) -> None:
        self.backend.similarity_search(self.embedding, scope=self.scope, top_k=5)


class GraphBackendQuerySuite:
    """Benchmark InMemoryGraphBackend query and traversal stub overhead."""

    timeout = 60

    def setup(self) -> None:
        self.backend = InMemoryGraphBackend()
        self.scope: frozenset[str] = frozenset({"RES01"})

    def time_sparql_query(self) -> None:
        self.backend.sparql_query(
            "SELECT ?s WHERE { ?s a uko:Container }", scope=self.scope
        )

    def time_get_triples(self) -> None:
        self.backend.get_triples("uko-py:class/Auth")

    def time_traverse(self) -> None:
        self.backend.traverse("uko-py:class/Auth", depth=3)


# ---------------------------------------------------------------------------
# Result dataclass construction benchmarks
# ---------------------------------------------------------------------------


class ResultConstructionSuite:
    """Benchmark frozen dataclass construction overhead."""

    timeout = 60

    def time_text_result(self) -> None:
        TextResult(uko_uri="uko:test", content="hello", score=0.5)

    def time_vector_result(self) -> None:
        VectorResult(uko_uri="uko:vec", content="embed", score=0.8)

    def time_graph_result_empty(self) -> None:
        GraphResult()

    def time_graph_result_with_triples(self) -> None:
        GraphResult(
            triples=[
                ("uko:A", "uko:contains", "uko:B"),
                ("uko:B", "uko:references", "uko:C"),
            ]
        )