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