"""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")