"""ASV benchmarks for actor compiler throughput. Measures the performance of: - compile_actor() on a simple two-node graph - compile_actor() on a larger multi-node graph - Subgraph cycle detection overhead - Compilation metadata serialization """ 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.actor.compiler import compile_actor # noqa: E402 from cleveragents.actor.schema import ( # noqa: E402 ActorConfigSchema, ActorType, EdgeDefinition, NodeDefinition, NodeType, RouteDefinition, ) def _make_graph(node_count: int) -> ActorConfigSchema: """Build a linear graph with *node_count* agent nodes.""" nodes: list[NodeDefinition] = [] edges: list[EdgeDefinition] = [] for i in range(node_count): nodes.append( NodeDefinition( id=f"node-{i}", type=NodeType.AGENT, name=f"Node {i}", description=f"Agent node {i}", config={"agent": f"agent_{i}"}, ) ) if i > 0: edges.append(EdgeDefinition(from_node=f"node-{i - 1}", to_node=f"node-{i}")) route = RouteDefinition( nodes=nodes, edges=edges, entry_node="node-0", exit_nodes=[f"node-{node_count - 1}"], ) return ActorConfigSchema( name="bench/graph", type=ActorType.GRAPH, description="Benchmark graph", model="gpt-4", route=route, ) class CompileSimpleSuite: """Benchmark compile_actor on a simple 2-node graph.""" timeout = 60 def setup(self) -> None: self.config = _make_graph(2) def time_compile_simple(self) -> None: compile_actor(self.config) class CompileLargeSuite: """Benchmark compile_actor on a 20-node linear graph.""" timeout = 60 def setup(self) -> None: self.config = _make_graph(20) def time_compile_large(self) -> None: compile_actor(self.config) class MetadataSerializationSuite: """Benchmark compilation metadata serialization.""" timeout = 60 def setup(self) -> None: self.config = _make_graph(10) self.compiled = compile_actor(self.config) def time_metadata_dump(self) -> None: self.compiled.metadata.model_dump(mode="json") class SubgraphResolutionSuite: """Benchmark compile_actor with subgraph resolution.""" timeout = 60 def setup(self) -> None: inner_nodes = [ NodeDefinition( id="inner", type=NodeType.AGENT, name="Inner", description="Inner agent", config={"agent": "inner"}, ), ] inner_route = RouteDefinition( nodes=inner_nodes, edges=[], entry_node="inner", exit_nodes=["inner"], ) self.inner_config = ActorConfigSchema( name="bench/inner", type=ActorType.GRAPH, description="Inner graph", model="gpt-4", route=inner_route, ) outer_nodes = [ NodeDefinition( id="main", type=NodeType.AGENT, name="Main", description="Main agent", config={"agent": "main"}, ), NodeDefinition( id="sub", type=NodeType.SUBGRAPH, name="Sub", description="Subgraph ref", config={}, actor_ref="bench/inner", ), ] outer_edges = [EdgeDefinition(from_node="main", to_node="sub")] outer_route = RouteDefinition( nodes=outer_nodes, edges=outer_edges, entry_node="main", exit_nodes=["sub"], ) self.outer_config = ActorConfigSchema( name="bench/outer", type=ActorType.GRAPH, description="Outer graph", model="gpt-4", route=outer_route, ) self.resolver = lambda name: ( self.inner_config if name == "bench/inner" else None ) def time_compile_with_subgraph(self) -> None: compile_actor(self.outer_config, actor_resolver=self.resolver)