File size: 3,090 Bytes
32112fa | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 | """Executor Agent — executes commands, runs tools, deploys.
Handles steps that involve running commands, executing tools, or
performing actions. Uses the tool registry for execution.
"""
from __future__ import annotations
import logging
from typing import Any
from .agent_base import BaseAgent
from ..memory.goal_memory import Goal
logger = logging.getLogger(__name__)
class ExecutorAgent(BaseAgent):
"""Executes commands and runs tools for goal steps."""
def __init__(self, goal_memory, persistent_memory=None, generate_fn=None,
tool_registry=None):
super().__init__(
name="executor",
role="Task Executor",
description="Executes commands, runs tools, and performs actions",
goal_memory=goal_memory,
persistent_memory=persistent_memory,
generate_fn=generate_fn,
poll_interval_s=2.0,
)
self._tool_registry = tool_registry
def _can_handle(self, goal: Goal) -> bool:
"""Executor handles goals related to execution/deployment."""
keywords = ["execute", "run", "deploy", "install", "test", "build", "start",
"stop", "configure", "setup", "launch", "perform", "do"]
text = (goal.title + " " + goal.description).lower()
return any(kw in text for kw in keywords)
def process_goal(self, goal: Goal) -> dict[str, Any]:
"""Execute a step using tools or commands."""
if goal.current_step >= len(goal.steps):
return {"success": True, "output": "No more steps"}
step = goal.steps[goal.current_step]
tool_name = step.get("tool", "")
# If a specific tool is specified, use it
if tool_name and self._tool_registry:
tool = self._tool_registry.get(tool_name)
if tool:
result = self._tool_registry.execute(tool_name, step.get("description", ""))
if result.success:
return {"success": True, "output": result.output[:200]}
else:
return {"success": False, "output": "", "error": result.error}
# Otherwise, use LLM to generate execution plan
prompt = (
f"You are an execution agent. Execute this step:\n"
f"Goal: {goal.title}\n"
f"Step: {step['title']}\n"
f"Description: {step['description']}\n"
f"Execute the step and report the result. Be concise.\n"
)
response = self._generate(prompt)
# Try to extract and execute tool calls from response
if self._tool_registry and "[TOOL:" in response:
from ..harness.tools import tool_loop, parse_tool_calls
final_text, tool_results = tool_loop(response, self._tool_registry, max_rounds=3)
if tool_results:
outputs = [r.output[:100] for r in tool_results if r.success]
if outputs:
return {"success": True, "output": "; ".join(outputs)}
return {"success": True, "output": response[:200]}
|