"""ASV benchmarks for tool-calling actor runtime overhead. Measures the performance of: - ToolCallingRuntime construction - Tool schema export - Single-iteration tool-call loop (1 tool call) - Multi-iteration tool-call loop (3 tool calls) - Max-iteration termination overhead - ToolActorContext construction and recording """ from __future__ import annotations import sys from pathlib import Path from typing import Any try: from cleveragents.tool.actor_context import ToolActorContext, ToolCallRecord from cleveragents.tool.actor_runtime import ( LLMResponse, LLMToolCall, ToolCallingRuntime, ) from cleveragents.tool.registry import ToolRegistry 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.actor_context import ToolActorContext, ToolCallRecord from cleveragents.tool.actor_runtime import ( LLMResponse, LLMToolCall, ToolCallingRuntime, ) from cleveragents.tool.registry import ToolRegistry from cleveragents.tool.runner import ToolRunner from cleveragents.tool.runtime import ToolSpec def _echo(inputs: dict[str, Any]) -> dict[str, Any]: return {"echoed": inputs} class _SingleToolLLM: def __init__(self) -> None: self._n = 0 def invoke( self, prompt: str, tool_schemas: list[dict[str, Any]], tool_results: list[dict[str, Any]] | None = None, actor_config: dict[str, Any] | None = None, ) -> LLMResponse: self._n += 1 if self._n == 1: return LLMResponse( tool_calls=[LLMToolCall(name="bench/echo", arguments={"x": 1})] ) return LLMResponse(content="done") class _AlwaysToolLLM: def invoke( self, prompt: str, tool_schemas: list[dict[str, Any]], tool_results: list[dict[str, Any]] | None = None, actor_config: dict[str, Any] | None = None, ) -> LLMResponse: return LLMResponse( tool_calls=[LLMToolCall(name="bench/echo", arguments={"x": 1})] ) class _ImmediateLLM: def invoke( self, prompt: str, tool_schemas: list[dict[str, Any]], tool_results: list[dict[str, Any]] | None = None, actor_config: dict[str, Any] | None = None, ) -> LLMResponse: return LLMResponse(content="immediate") def _make_registry() -> ToolRegistry: reg = ToolRegistry() reg.register(ToolSpec(name="bench/echo", description="Bench echo", handler=_echo)) return reg class ToolCallingRuntimeConstructionSuite: """Benchmark ToolCallingRuntime construction.""" def setup(self) -> None: self.registry = _make_registry() self.runner = ToolRunner(self.registry) self.llm = _ImmediateLLM() def time_construction(self) -> None: ToolCallingRuntime( registry=self.registry, runner=self.runner, llm_caller=self.llm, ) class ToolSchemaExportSuite: """Benchmark tool schema export.""" def setup(self) -> None: self.registry = _make_registry() self.runner = ToolRunner(self.registry) self.llm = _ImmediateLLM() self.runtime = ToolCallingRuntime( registry=self.registry, runner=self.runner, llm_caller=self.llm, ) def time_export_schemas(self) -> None: self.runtime.export_tool_schemas() class SingleToolCallLoopSuite: """Benchmark single-iteration tool-call loop.""" def setup(self) -> None: self.registry = _make_registry() self.runner = ToolRunner(self.registry) def time_single_tool_call(self) -> None: llm = _SingleToolLLM() runtime = ToolCallingRuntime( registry=self.registry, runner=self.runner, llm_caller=llm, ) runtime.run_tool_loop(prompt="bench") class MaxIterationSuite: """Benchmark max-iteration termination overhead.""" def setup(self) -> None: self.registry = _make_registry() self.runner = ToolRunner(self.registry) def time_max_iterations_5(self) -> None: runtime = ToolCallingRuntime( registry=self.registry, runner=self.runner, llm_caller=_AlwaysToolLLM(), max_iterations=5, ) runtime.run_tool_loop(prompt="bench") class ToolActorContextSuite: """Benchmark ToolActorContext operations.""" def time_context_construction(self) -> None: ToolActorContext(plan_id="bench-plan", phase="execute") def time_record_tool_call(self) -> None: ctx = ToolActorContext(plan_id="bench-plan") record = ToolCallRecord( tool_name="bench/tool", inputs={"x": 1}, output={"y": 2}, duration_ms=1.0, ) ctx.record_tool_call(record) def time_context_summary(self) -> None: ctx = ToolActorContext( plan_id="bench-plan", phase="execute", sandbox_root="/tmp/sb", ) ctx.as_summary()