133 lines
4.1 KiB
Python
133 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
import itertools
|
|
import sys
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass
|
|
from threading import Event
|
|
from typing import Callable, Optional, Sequence
|
|
|
|
|
|
ToolInvoker = Callable[[str], bool]
|
|
CommandExecutor = Callable[[str], None]
|
|
|
|
|
|
class Agent(ABC):
|
|
"""Interface for higher-level behaviours that orchestrate tools."""
|
|
|
|
@abstractmethod
|
|
def run(
|
|
self,
|
|
*,
|
|
invoke_tool: ToolInvoker,
|
|
stop_event: Event,
|
|
send_command: Optional[CommandExecutor] = None,
|
|
) -> 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,
|
|
send_command: Optional[CommandExecutor] = None,
|
|
) -> 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,
|
|
send_command: Optional[CommandExecutor] = None,
|
|
) -> 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, *, allowed_tools: Optional[Sequence[str]] = None) -> 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)
|
|
if kind in {"intelligent", "llm"}:
|
|
from intelligent_agent import IntelligentAgent
|
|
|
|
instruction = ""
|
|
model = None
|
|
turn_delay = 0.0
|
|
if config:
|
|
segments = [segment.strip() for segment in config.split("|") if segment.strip()]
|
|
for segment in segments:
|
|
if segment.startswith("model="):
|
|
model = segment.split("=", 1)[1].strip()
|
|
elif segment.startswith("delay="):
|
|
try:
|
|
turn_delay = float(segment.split("=", 1)[1].strip())
|
|
except ValueError:
|
|
print(f"[Agent] Invalid delay '{segment}'", file=sys.stderr)
|
|
else:
|
|
instruction = segment
|
|
|
|
whitelist = tuple(sorted({tool.lower() for tool in (allowed_tools or [])}))
|
|
agent = IntelligentAgent(instruction=instruction, allowed_tools=whitelist)
|
|
if model:
|
|
agent.model = model
|
|
agent.turn_delay = turn_delay if turn_delay > 0 else 1.0
|
|
return agent
|
|
|
|
raise RuntimeError(f"Unknown agent type '{kind}'")
|