"""ASV benchmarks for Actor YAML schema validation throughput. Measures the performance of: - YAML string parsing + schema validation - YAML file loading + schema validation - Graph topology validation (cycle detection) - Tool definition validation - model_dump() serialization """ from __future__ import annotations import importlib import sys import tempfile from pathlib import Path # Ensure the local *source* tree is importable _SRC = str(Path(__file__).resolve().parents[1] / "src") if _SRC not in sys.path: sys.path.insert(0, _SRC) # Force-reload the top-level package import cleveragents # noqa: E402 importlib.reload(cleveragents) from cleveragents.actor.schema import ActorConfigSchema # noqa: E402 _MINIMAL_LLM_YAML = """\ name: bench/llm type: llm description: Benchmark LLM actor model: gpt-4 """ _FULL_LLM_YAML = """\ name: bench/llm_full type: llm description: Full LLM actor with all features version: "1.0" model: gpt-4-turbo system_prompt: | You are a benchmark testing assistant. Process requests efficiently and accurately. tools: - files/read_file - files/write_file context_view: executor memory: enabled: true max_messages: 100 max_tokens: 8000 summarize_old: true context: include_files: - README.md - pyproject.toml include_dirs: - src/ - tests/ exclude_patterns: - "**/__pycache__/**" - "*.pyc" max_context_tokens: 16000 env_vars: LOG_LEVEL: info WORK_DIR: /tmp/bench """ _TOOL_ACTOR_YAML = """\ name: bench/tools type: tool description: Tool collection for benchmarking tools: - files/read_file - files/write_file - files/delete_file """ _SIMPLE_GRAPH_YAML = """\ name: bench/simple_graph type: graph description: Simple 3-node graph for benchmarking model: gpt-4 route: nodes: - id: extract type: tool name: Extractor description: Extract data config: tool_name: data/extract - id: process type: agent name: Processor description: Process data config: prompt: "Process the data" - id: save type: tool name: Saver description: Save results config: tool_name: data/save edges: - from_node: extract to_node: process - from_node: process to_node: save entry_node: extract exit_nodes: - save """ _COMPLEX_GRAPH_YAML = """\ name: bench/complex_graph type: graph description: Complex graph with 10 nodes and conditionals model: gpt-4 route: nodes: - id: start type: agent name: Starter description: Start node config: prompt: "Begin" - id: node2 type: agent name: Node 2 description: Second node config: prompt: "Continue" - id: node3 type: agent name: Node 3 description: Third node config: prompt: "Process" - id: checker type: conditional name: Checker description: Check condition config: conditions: - check: "state.get('ok') == True" route_to: node4 - check: "state.get('ok') == False" route_to: node5 - id: node4 type: agent name: Node 4 description: Success path config: prompt: "Success" - id: node5 type: agent name: Node 5 description: Retry path config: prompt: "Retry" - id: node6 type: tool name: Tool 6 description: Tool execution config: tool_name: test/tool - id: node7 type: agent name: Node 7 description: Seventh node config: prompt: "Continue" - id: node8 type: subgraph name: Subgraph description: Nested workflow config: actor_path: bench/nested.yaml - id: end type: agent name: End description: Final node config: prompt: "Complete" edges: - from_node: start to_node: node2 - from_node: node2 to_node: node3 - from_node: node3 to_node: checker - from_node: node4 to_node: node6 - from_node: node5 to_node: node7 - from_node: node6 to_node: node8 - from_node: node7 to_node: node8 - from_node: node8 to_node: end entry_node: start exit_nodes: - end """ class TimeActorSchemaMinimalLLM: """Benchmark minimal LLM actor validation.""" def time_parse_minimal_llm(self) -> None: """Time parsing a minimal LLM actor YAML.""" import yaml data = yaml.safe_load(_MINIMAL_LLM_YAML) ActorConfigSchema.model_validate(data) class TimeActorSchemaFullLLM: """Benchmark full LLM actor validation.""" def time_parse_full_llm(self) -> None: """Time parsing a full LLM actor YAML with all features.""" import yaml data = yaml.safe_load(_FULL_LLM_YAML) ActorConfigSchema.model_validate(data) class TimeActorSchemaTool: """Benchmark tool actor validation.""" def time_parse_tool_actor(self) -> None: """Time parsing a tool actor YAML.""" import yaml data = yaml.safe_load(_TOOL_ACTOR_YAML) ActorConfigSchema.model_validate(data) class TimeActorSchemaSimpleGraph: """Benchmark simple graph actor validation.""" def time_parse_simple_graph(self) -> None: """Time parsing a simple 3-node graph actor.""" import yaml data = yaml.safe_load(_SIMPLE_GRAPH_YAML) ActorConfigSchema.model_validate(data) class TimeActorSchemaComplexGraph: """Benchmark complex graph actor validation.""" def time_parse_complex_graph(self) -> None: """Time parsing a complex 10-node graph with conditionals.""" import yaml data = yaml.safe_load(_COMPLEX_GRAPH_YAML) ActorConfigSchema.model_validate(data) class TimeActorSchemaFileIO: """Benchmark file I/O operations.""" def setup(self) -> None: """Create temporary YAML file.""" with tempfile.NamedTemporaryFile( mode="w", suffix=".yaml", delete=False ) as temp_file: temp_file.write(_FULL_LLM_YAML) self.temp_file_name = temp_file.name def teardown(self) -> None: """Remove temporary file.""" Path(self.temp_file_name).unlink(missing_ok=True) def time_load_from_file(self) -> None: """Time loading actor config from YAML file.""" ActorConfigSchema.from_yaml_file(self.temp_file_name) def time_save_to_file(self) -> None: """Time saving actor config to YAML file.""" import yaml data = yaml.safe_load(_FULL_LLM_YAML) config = ActorConfigSchema.model_validate(data) with tempfile.NamedTemporaryFile( mode="w", suffix=".yaml", delete=False ) as temp_out: temp_out_name = temp_out.name try: config.to_yaml_file(temp_out_name) finally: Path(temp_out_name).unlink(missing_ok=True) class TimeActorSchemaSerialization: """Benchmark serialization operations.""" def setup(self) -> None: """Parse actor configs once.""" import yaml self.minimal_config = ActorConfigSchema.model_validate( yaml.safe_load(_MINIMAL_LLM_YAML) ) self.full_config = ActorConfigSchema.model_validate( yaml.safe_load(_FULL_LLM_YAML) ) self.graph_config = ActorConfigSchema.model_validate( yaml.safe_load(_COMPLEX_GRAPH_YAML) ) def time_dump_minimal(self) -> None: """Time model_dump() on minimal config.""" self.minimal_config.model_dump(mode="json") def time_dump_full(self) -> None: """Time model_dump() on full config.""" self.full_config.model_dump(mode="json") def time_dump_graph(self) -> None: """Time model_dump() on complex graph.""" self.graph_config.model_dump(mode="json") class TimeActorSchemaGraphValidation: """Benchmark graph topology validation.""" def setup(self) -> None: """Parse graph configs.""" import yaml self.simple_graph_data = yaml.safe_load(_SIMPLE_GRAPH_YAML) self.complex_graph_data = yaml.safe_load(_COMPLEX_GRAPH_YAML) def time_validate_simple_graph_topology(self) -> None: """Time validation of simple graph topology.""" ActorConfigSchema.model_validate(self.simple_graph_data) def time_validate_complex_graph_topology(self) -> None: """Time validation of complex graph with cycle detection.""" ActorConfigSchema.model_validate(self.complex_graph_data) # Module-level metadata for ASV time_parse_minimal_llm = TimeActorSchemaMinimalLLM() time_parse_full_llm = TimeActorSchemaFullLLM() time_parse_tool_actor = TimeActorSchemaTool() time_parse_simple_graph = TimeActorSchemaSimpleGraph() time_parse_complex_graph = TimeActorSchemaComplexGraph() time_file_io = TimeActorSchemaFileIO() time_serialization = TimeActorSchemaSerialization() time_graph_validation = TimeActorSchemaGraphValidation()