Spaces:
Sleeping
Sleeping
File size: 4,898 Bytes
763ef0d | 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 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | """
Autonomous task planner.
Produces structured action plans (list of dicts):
{"type": "shell"|"python"|"browser"|"git"|"deploy"|"note", ...}
Strategy:
1. Try LLM-based planning with strict JSON output.
2. If LLM fails or returns invalid JSON, use heuristic fallback.
3. Always produces non-empty plan.
"""
from __future__ import annotations
import json
import logging
import re
from typing import Any, Dict, List
from .llm_router import get_router
logger = logging.getLogger("planner")
PLANNER_SYSTEM = (
"You are an autonomous AI Developer Agent's planner. "
"Decompose a user task into a JSON list of concrete actions. "
"Each action is an object with a 'type' field and supporting fields. "
"Supported types: shell (cmd), python (code), browser (action,url,...), "
"git (op,args), deploy (target), note (msg). "
"Return ONLY a JSON array, no commentary. Keep plan under 12 steps."
)
def plan_task(title: str, description: str, context: Dict[str, Any] | None = None) -> List[Dict[str, Any]]:
"""Generate a concrete action plan."""
router = get_router()
user_prompt = (
f"TASK TITLE: {title}\n"
f"DESCRIPTION:\n{description}\n\n"
f"CONTEXT:\n{json.dumps(context or {}, indent=2)[:2000]}\n\n"
"Output a JSON array of action objects. Example:\n"
'[{"type":"shell","cmd":"echo hi"},{"type":"note","msg":"done"}]'
)
messages = [
{"role": "system", "content": PLANNER_SYSTEM},
{"role": "user", "content": user_prompt},
]
try:
raw = router.chat(messages, temperature=0.1, max_tokens=1200, timeout=45.0)
except Exception as e:
logger.warning("Planner LLM call failed: %s", e)
raw = ""
plan = _parse_plan_json(raw)
if plan:
return plan
logger.info("Planner falling back to heuristic plan")
return heuristic_plan(title, description)
def _parse_plan_json(raw: str) -> List[Dict[str, Any]]:
if not raw:
return []
# Try direct
try:
obj = json.loads(raw)
if isinstance(obj, list):
return [a for a in obj if isinstance(a, dict) and "type" in a]
except Exception:
pass
# Find first JSON array in text
m = re.search(r"\[[\s\S]*\]", raw)
if m:
try:
obj = json.loads(m.group(0))
if isinstance(obj, list):
return [a for a in obj if isinstance(a, dict) and "type" in a]
except Exception:
return []
return []
def heuristic_plan(title: str, description: str) -> List[Dict[str, Any]]:
"""Always-valid fallback plan."""
text = (title + "\n" + description).lower()
plan: List[Dict[str, Any]] = []
if any(k in text for k in ["deploy", "deployment", "huggingface", "hf space", "vercel"]):
plan.append({"type": "note", "msg": "Deployment task detected"})
if "vercel" in text:
plan.append({"type": "deploy", "target": "vercel"})
if "huggingface" in text or "hf" in text:
plan.append({"type": "deploy", "target": "huggingface"})
return plan
if any(k in text for k in ["git", "github", "commit", "push", "pr ", "pull request"]):
plan.append({"type": "git", "op": "status"})
plan.append({"type": "note", "msg": "GitHub task detected"})
return plan
if any(k in text for k in ["browser", "scrape", "navigate", "click", "open url", "http://", "https://"]):
url_match = re.search(r"https?://[\w\.\-/?=&%#]+", description)
url = url_match.group(0) if url_match else "https://example.com"
plan.append({"type": "browser", "action": "navigate", "url": url})
plan.append({"type": "browser", "action": "screenshot"})
return plan
if any(k in text for k in ["run python", "python script", "execute python"]):
plan.append({"type": "python", "code": "print('Hello from AI Developer Agent')"})
return plan
if any(k in text for k in ["install", "pip ", "npm "]):
# Try to extract a package name
m = re.search(r"(?:install|add)\s+([\w\.\-]+)", text)
pkg = m.group(1) if m else ""
if "npm" in text:
plan.append({"type": "shell", "cmd": f"npm install {pkg}".strip()})
else:
plan.append({"type": "shell", "cmd": f"pip install {pkg}".strip()})
return plan
# Generic fallback: echo the task back as an inspection action.
plan.append({"type": "note", "msg": f"Plan-fallback for: {title}"})
plan.append({"type": "shell", "cmd": "uname -a && python3 --version && node --version 2>/dev/null || true"})
return plan
def repair_plan(error_category: str, detail: str = "") -> List[Dict[str, Any]]:
from .classifier import ErrorClass
from .repair import repair_actions
return repair_actions(ErrorClass(category=error_category, detail=detail, suggested_fix=""))
|