Spaces:
Sleeping
Sleeping
File size: 4,213 Bytes
06ce7ac | 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 81 82 83 84 85 86 87 88 89 | """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"}]
|