89 lines
2.6 KiB
Python
89 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
from abc import ABC, abstractmethod
|
|
import itertools
|
|
from dataclasses import dataclass
|
|
from threading import Event
|
|
from typing import Callable
|
|
|
|
|
|
ToolInvoker = Callable[[str], bool]
|
|
|
|
|
|
class Agent(ABC):
|
|
"""Interface for higher-level behaviours that orchestrate tools."""
|
|
|
|
@abstractmethod
|
|
def run(self, *, invoke_tool: ToolInvoker, stop_event: Event) -> None:
|
|
"""Execute the agent strategy until finished or *stop_event* is set."""
|
|
|
|
|
|
@dataclass
|
|
class FixedStrategyAgent(Agent):
|
|
"""Run a comma-separated list of tool specs in order."""
|
|
|
|
plan: str
|
|
|
|
def run(self, *, invoke_tool: ToolInvoker, stop_event: Event) -> None:
|
|
steps = [part.strip() for part in self.plan.split(",") if part.strip()]
|
|
if not steps:
|
|
print("[Agent] No tools configured for fixed strategy", file=sys.stderr)
|
|
return
|
|
|
|
for step in steps:
|
|
if stop_event.is_set():
|
|
break
|
|
success = invoke_tool(step)
|
|
if not success:
|
|
# Abort on first failure to avoid confusing behaviour
|
|
break
|
|
|
|
|
|
@dataclass
|
|
class LoopAgent(Agent):
|
|
"""Continuously execute a fixed strategy until stopped."""
|
|
|
|
plan: str
|
|
delay: float = 0.0
|
|
|
|
def run(self, *, invoke_tool: ToolInvoker, stop_event: Event) -> None:
|
|
steps = [part.strip() for part in self.plan.split(",") if part.strip()]
|
|
if not steps:
|
|
print("[Agent] No tools configured for loop strategy", file=sys.stderr)
|
|
return
|
|
|
|
for step in itertools.cycle(steps):
|
|
if stop_event.is_set():
|
|
break
|
|
success = invoke_tool(step)
|
|
if not success:
|
|
break
|
|
if self.delay > 0:
|
|
if stop_event.wait(self.delay):
|
|
break
|
|
|
|
|
|
def build_agent(spec: str) -> Agent:
|
|
normalized = spec.strip()
|
|
if not normalized:
|
|
raise RuntimeError("Agent specification must not be empty")
|
|
|
|
parts = normalized.split(maxsplit=1)
|
|
kind = parts[0].lower()
|
|
config = parts[1].strip() if len(parts) > 1 else ""
|
|
|
|
if kind in {"fixed", "strategy", "fixedstrategy"}:
|
|
return FixedStrategyAgent(config)
|
|
if kind in {"loop", "cycle"}:
|
|
delay = 0.0
|
|
if ":" in config:
|
|
plan, delay_str = config.split(":", 1)
|
|
config = plan.strip()
|
|
try:
|
|
delay = float(delay_str.strip())
|
|
except ValueError:
|
|
print(f"[Agent] Invalid delay '{delay_str}', defaulting to 0")
|
|
return LoopAgent(config, delay=delay)
|
|
|
|
raise RuntimeError(f"Unknown agent type '{kind}'")
|