cleveragents-core/benchmarks/tool_router_bench.py

"""ASV benchmarks for ToolCallRouter operations.

Measures the performance of:
- Provider format detection
- Payload normalization
- Stable ID generation
- Tool call routing (OpenAI, Anthropic, LangChain)
- Batch routing
- Schema normalization
- Error classification
"""

from __future__ import annotations

import json
import sys
from pathlib import Path
from typing import Any

try:
    from cleveragents.tool.registry import ToolRegistry
    from cleveragents.tool.router import (
        ProviderFormat,
        ToolCallRouter,
        classify_tool_error,
        detect_provider_format,
        generate_tool_call_id,
        normalize_tool_call,
        normalize_tool_schema_for_provider,
    )
    from cleveragents.tool.runner import ToolRunner
    from cleveragents.tool.runtime import ToolSpec
except ModuleNotFoundError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
    from cleveragents.tool.registry import ToolRegistry
    from cleveragents.tool.router import (
        ProviderFormat,
        ToolCallRouter,
        classify_tool_error,
        detect_provider_format,
        generate_tool_call_id,
        normalize_tool_call,
        normalize_tool_schema_for_provider,
    )
    from cleveragents.tool.runner import ToolRunner
    from cleveragents.tool.runtime import ToolSpec


def _echo(inputs: dict[str, Any]) -> dict[str, Any]:
    return dict(inputs)


class FormatDetectionSuite:
    """Benchmark provider format detection."""

    def setup(self) -> None:
        """Prepare payloads."""
        self.openai_payload = {"name": "bench/echo", "arguments": '{"a": 1}'}
        self.anthropic_payload = {"name": "bench/echo", "input": {"a": 1}}
        self.langchain_payload = {
            "name": "bench/echo",
            "type": "tool_call",
            "args": {"a": 1},
        }

    def time_detect_openai(self) -> None:
        """Benchmark OpenAI format detection."""
        detect_provider_format(self.openai_payload)

    def time_detect_anthropic(self) -> None:
        """Benchmark Anthropic format detection."""
        detect_provider_format(self.anthropic_payload)

    def time_detect_langchain(self) -> None:
        """Benchmark LangChain format detection."""
        detect_provider_format(self.langchain_payload)


class NormalizationSuite:
    """Benchmark payload normalization."""

    def setup(self) -> None:
        """Prepare payloads."""
        self.openai_payload = {"name": "bench/echo", "arguments": '{"a": 1}'}
        self.anthropic_payload = {"name": "bench/echo", "input": {"a": 1}}

    def time_normalize_openai(self) -> None:
        """Benchmark OpenAI payload normalization."""
        normalize_tool_call(self.openai_payload)

    def time_normalize_anthropic(self) -> None:
        """Benchmark Anthropic payload normalization."""
        normalize_tool_call(self.anthropic_payload)


class IDGenerationSuite:
    """Benchmark stable ID generation."""

    def time_generate_id(self) -> None:
        """Benchmark single ID generation."""
        generate_tool_call_id("plan-bench", 42)

    def time_generate_100_ids(self) -> None:
        """Benchmark generating 100 IDs."""
        for i in range(100):
            generate_tool_call_id("plan-bench", i)


class RoutingSuite:
    """Benchmark tool call routing."""

    def setup(self) -> None:
        """Prepare registry and router."""
        self.registry = ToolRegistry()
        spec = ToolSpec(
            name="bench/echo",
            description="Benchmark echo tool",
            handler=_echo,
        )
        self.registry.register(spec)
        self.runner = ToolRunner(self.registry)
        self.router = ToolCallRouter(
            registry=self.registry,
            runner=self.runner,
            plan_id="plan-bench",
        )
        self.openai_payload = {
            "name": "bench/echo",
            "arguments": '{"msg": "hello"}',
        }
        self.anthropic_payload = {
            "name": "bench/echo",
            "input": {"msg": "hello"},
        }
        self.langchain_payload = {
            "name": "bench/echo",
            "type": "tool_call",
            "args": {"msg": "hello"},
        }

    def time_route_openai(self) -> None:
        """Benchmark routing an OpenAI tool call."""
        self.router.route(self.openai_payload)

    def time_route_anthropic(self) -> None:
        """Benchmark routing an Anthropic tool call."""
        self.router.route(self.anthropic_payload)

    def time_route_langchain(self) -> None:
        """Benchmark routing a LangChain tool call."""
        self.router.route(self.langchain_payload)

    def time_route_batch_10(self) -> None:
        """Benchmark routing a batch of 10 tool calls."""
        payloads = [
            {"name": "bench/echo", "arguments": json.dumps({"i": i})} for i in range(10)
        ]
        self.router.route_batch(payloads)


class SchemaSuite:
    """Benchmark schema normalization."""

    def setup(self) -> None:
        """Prepare tool spec."""
        self.spec = ToolSpec(
            name="bench/echo",
            description="Benchmark echo tool",
            handler=_echo,
            input_schema={"type": "object", "properties": {"a": {"type": "integer"}}},
        )

    def time_normalize_openai_schema(self) -> None:
        """Benchmark schema normalization for OpenAI."""
        normalize_tool_schema_for_provider(self.spec, ProviderFormat.OPENAI)

    def time_normalize_anthropic_schema(self) -> None:
        """Benchmark schema normalization for Anthropic."""
        normalize_tool_schema_for_provider(self.spec, ProviderFormat.ANTHROPIC)


class ErrorClassificationSuite:
    """Benchmark error classification."""

    def time_classify_timeout(self) -> None:
        """Benchmark classifying timeout error."""
        classify_tool_error("Operation timed out after 30s")

    def time_classify_permission(self) -> None:
        """Benchmark classifying permission error."""
        classify_tool_error("Permission denied: cannot write")

    def time_classify_generic(self) -> None:
        """Benchmark classifying generic error."""
        classify_tool_error("Something unexpected happened")