| """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 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} |
|
|
| |
| 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) |
|
|
| |
| 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]} |
|
|
