| """Planner Agent — breaks goals into actionable steps. |
| |
| Takes a goal description and generates a step-by-step execution plan. |
| Can also replan when steps fail. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import logging |
| from typing import Any |
|
|
| from .agent_base import BaseAgent |
| from ..memory.goal_memory import Goal |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| class PlannerAgent(BaseAgent): |
| """Plans goals — breaks them into steps and sub-goals.""" |
|
|
| def __init__(self, goal_memory, persistent_memory=None, generate_fn=None): |
| super().__init__( |
| name="planner", |
| role="Goal Planner", |
| description="Breaks goals into actionable steps and sub-goals", |
| goal_memory=goal_memory, |
| persistent_memory=persistent_memory, |
| generate_fn=generate_fn, |
| poll_interval_s=3.0, |
| ) |
|
|
| def _can_handle(self, goal: Goal) -> bool: |
| """Planner handles goals that need planning.""" |
| return goal.status in ("pending", "planning") |
|
|
| def process_goal(self, goal: Goal) -> dict[str, Any]: |
| """Plan a goal by generating steps via LLM.""" |
| if goal.status == "pending" or (goal.status == "planning" and not goal.steps): |
| prompt = ( |
| f"{self.goal_memory.PLANNING_PROMPT}\n\n" |
| f"GOAL:\nTitle: {goal.title}\n" |
| f"Description: {goal.description}\n" |
| f"Priority: {goal.priority}\n" |
| ) |
| response = self._generate(prompt) |
|
|
| |
| plan = self._safe_json_parse(response, {"steps": [], "sub_goals": []}) |
|
|
| if plan.get("steps"): |
| result = self.goal_memory.plan_goal( |
| goal.id, |
| steps=plan["steps"], |
| sub_goals=plan.get("sub_goals", []), |
| plan_type="initial", |
| ) |
| return {"success": True, "output": f"Planned {len(plan['steps'])} steps"} |
| else: |
| |
| self.goal_memory.plan_goal( |
| goal.id, |
| steps=[{"title": "Execute goal", "description": goal.description, "tool": ""}], |
| plan_type="initial", |
| ) |
| return {"success": True, "output": "Planned with single fallback step"} |
|
|
| return {"success": True, "output": "Goal already planned"} |
|
|
| @staticmethod |
| def _safe_json_parse(text: str, fallback: dict) -> dict: |
| try: |
| start = text.find("{") |
| end = text.rfind("}") + 1 |
| if start >= 0 and end > start: |
| return json.loads(text[start:end]) |
| except Exception: |
| pass |
| return fallback |
|
|
