136 lines
4.3 KiB
Python
136 lines
4.3 KiB
Python
"""ASV benchmarks for plan actor integration overhead.
|
|
|
|
Measures the cost of strategize and execute stub actor operations,
|
|
PlanExecutor orchestration, and streaming callback overhead.
|
|
"""
|
|
|
|
from typing import ClassVar
|
|
|
|
from ulid import ULID
|
|
|
|
from cleveragents.application.services.plan_executor import (
|
|
ExecuteStubActor,
|
|
PlanExecutor,
|
|
StrategizeStubActor,
|
|
StrategyDecision,
|
|
)
|
|
from cleveragents.application.services.plan_lifecycle_service import (
|
|
PlanLifecycleService,
|
|
)
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.domain.models.core.plan import InvariantSource, PlanInvariant
|
|
from cleveragents.tool.registry import ToolRegistry
|
|
from cleveragents.tool.runner import ToolRunner
|
|
|
|
|
|
class StrategizeStubBench:
|
|
"""Benchmark StrategizeStubActor execution."""
|
|
|
|
params: ClassVar[list[int]] = [1, 5, 20]
|
|
param_names: ClassVar[list[str]] = ["num_steps"]
|
|
|
|
def setup(self, num_steps: int) -> None:
|
|
self.actor = StrategizeStubActor()
|
|
self.plan_id = str(ULID())
|
|
steps = [f"Step {i}" for i in range(num_steps)]
|
|
self.definition = "\n".join(steps)
|
|
self.invariants = [
|
|
PlanInvariant(text=f"Invariant {i}", source=InvariantSource.ACTION)
|
|
for i in range(3)
|
|
]
|
|
|
|
def time_strategize_execute(self, num_steps: int) -> None:
|
|
self.actor.execute(
|
|
plan_id=self.plan_id,
|
|
definition_of_done=self.definition,
|
|
invariants=self.invariants,
|
|
)
|
|
|
|
def time_strategize_with_streaming(self, num_steps: int) -> None:
|
|
events: list[tuple[str, dict[str, object]]] = []
|
|
self.actor.execute(
|
|
plan_id=self.plan_id,
|
|
definition_of_done=self.definition,
|
|
stream_callback=lambda t, d: events.append((t, d)),
|
|
)
|
|
|
|
|
|
class ExecuteStubBench:
|
|
"""Benchmark ExecuteStubActor execution."""
|
|
|
|
params: ClassVar[list[int]] = [1, 5, 20]
|
|
param_names: ClassVar[list[str]] = ["num_decisions"]
|
|
|
|
def setup(self, num_decisions: int) -> None:
|
|
self.actor = ExecuteStubActor()
|
|
self.plan_id = str(ULID())
|
|
root_id = str(ULID())
|
|
self.decisions = [
|
|
StrategyDecision(
|
|
decision_id=root_id if i == 0 else str(ULID()),
|
|
step_text=f"Decision {i}",
|
|
sequence=i,
|
|
parent_id=root_id if i > 0 else None,
|
|
)
|
|
for i in range(num_decisions)
|
|
]
|
|
|
|
def time_execute_stub(self, num_decisions: int) -> None:
|
|
self.actor.execute(
|
|
plan_id=self.plan_id,
|
|
decisions=self.decisions,
|
|
)
|
|
|
|
def time_execute_with_tool_runner(self, num_decisions: int) -> None:
|
|
runner = ToolRunner(registry=ToolRegistry())
|
|
self.actor.execute(
|
|
plan_id=self.plan_id,
|
|
decisions=self.decisions,
|
|
tool_runner=runner,
|
|
)
|
|
|
|
|
|
class PlanExecutorLifecycleBench:
|
|
"""Benchmark full PlanExecutor lifecycle."""
|
|
|
|
def setup(self) -> None:
|
|
settings = Settings()
|
|
self.lifecycle = PlanLifecycleService(settings=settings)
|
|
self.runner = ToolRunner(registry=ToolRegistry())
|
|
self.executor = PlanExecutor(
|
|
lifecycle_service=self.lifecycle,
|
|
tool_runner=self.runner,
|
|
)
|
|
|
|
def time_full_strategize_execute(self) -> None:
|
|
action = self.lifecycle.create_action(
|
|
name=f"local/bench-{ULID()}",
|
|
description="Benchmark action",
|
|
definition_of_done="Step 1\nStep 2\nStep 3",
|
|
strategy_actor="local/stub",
|
|
execution_actor="local/stub",
|
|
)
|
|
plan = self.lifecycle.use_action(
|
|
action_name=str(action.namespaced_name),
|
|
)
|
|
plan_id = plan.identity.plan_id
|
|
self.executor.run_strategize(plan_id)
|
|
self.lifecycle.execute_plan(plan_id)
|
|
self.executor.run_execute(plan_id)
|
|
|
|
|
|
class ParseStepsBench:
|
|
"""Benchmark definition_of_done parsing."""
|
|
|
|
params: ClassVar[list[int]] = [5, 50, 200]
|
|
param_names: ClassVar[list[str]] = ["num_lines"]
|
|
|
|
def setup(self, num_lines: int) -> None:
|
|
self.actor = StrategizeStubActor()
|
|
self.definition = "\n".join(
|
|
f"- Step {i}: do something important" for i in range(num_lines)
|
|
)
|
|
|
|
def time_parse_steps(self, num_lines: int) -> None:
|
|
self.actor._parse_steps(self.definition)
|