File size: 2,776 Bytes
0e3d4b8 | 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 | """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)
# Parse the plan
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:
# Fallback: create a single step
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
|