"""Planner agent that builds structured execution-friendly plans.""" from typing import Any from api.deps import get_logger logger = get_logger("kapo.agent.planner") def _contains_any(text: str, terms: tuple[str, ...]) -> bool: lowered = text.lower() return any(term in lowered for term in terms) class PlannerAgent: def _base_step(self, step_id: str, action: str, user_input: str, tool_hint: str, role: str) -> dict[str, Any]: return { "id": step_id, "action": action, "input": user_input, "tool_hint": tool_hint, "role": role, } def run(self, user_input: str, context: dict[str, Any]) -> list[dict[str, Any]]: try: text = (user_input or "").strip() if not text: return [self._base_step("step-1", "respond", "Empty request", "python", "chat")] context = context or {} steps: list[dict[str, Any]] = [ { **self._base_step("step-1", "analyze", text, "python", "planner"), "context_keys": sorted(context.keys()), } ] research_terms = ("search", "research", "browse", "look up", "find out", "ابحث", "بحث", "دور", "فتش") coding_terms = ( "build", "fix", "debug", "refactor", "implement", "generate", "write code", "api", "fastapi", "python", "react", "repo", "project", "اصلح", "نفذ", "شغل", "عدل", "ابني", "برمجة", "كود", "مشروع", ) planning_terms = ("plan", "roadmap", "architecture", "analyze", "design", "structure", "خطة", "بنية", "معمارية", "حلل") explain_terms = ("explain", "describe", "summarize", "اشرح", "وضح", "لخص", "عرفني") is_research = _contains_any(text, research_terms) is_coding = _contains_any(text, coding_terms) is_planning = _contains_any(text, planning_terms) is_explainer = _contains_any(text, explain_terms) if is_research: steps.extend( [ self._base_step("step-2", "research", text, "web", "chat"), self._base_step("step-3", "synthesize", "Summarize and cite the most relevant findings", "python", "supervisor"), ] ) elif is_coding: steps.extend( [ self._base_step("step-2", "collect_requirements", "Inspect impacted files, dependencies, and constraints", "python", "planner"), self._base_step("step-3", "execute", text, "python", "coding"), self._base_step("step-4", "verify", "Run validation checks and inspect resulting output", "python", "coding"), self._base_step("step-5", "summarize", "Summarize what changed, risks, and verification status", "python", "supervisor"), ] ) elif is_planning: steps.extend( [ self._base_step("step-2", "decompose", "Break the request into phases, dependencies, and risks", "python", "planner"), self._base_step("step-3", "respond", "Provide a structured implementation plan", "python", "supervisor"), ] ) elif is_explainer: steps.extend( [ self._base_step("step-2", "collect_context", "Gather the minimum relevant context for explanation", "python", "chat"), self._base_step("step-3", "respond", "Explain the topic clearly and directly", "python", "supervisor"), ] ) else: steps.append(self._base_step("step-2", "respond", text, "python", "chat")) return steps except Exception as exc: logger.exception("Planner failed") return [{"id": "step-err", "action": "error", "input": str(exc), "tool_hint": "python", "role": "fallback"}]