""" planner.py — Speculative Hybrid Planner (Gap-5: eliminazione bottleneck sequenziale) Architettura: FASE 1 Quick-Start Draft < 500ms Cerebras gpt-oss-120b (o Groq 8B fallback) Genera 2-3 subtask immediati → agente parte istantaneamente. FASE 2 Master Plan background DeepSeek-R1 via OpenRouter Piano architetturale completo; raffina i passi successivi. FASE 3 Parallel Sub-Graphs — Campo `parallel_groups` nel JSON Rami indipendenti (es. Backend vs Frontend) identificati esplicitamente. Compatibilità backward: `create_plan()` restituisce sempre dict con "subtasks". Flag `_speculative: True` → piano quick-start; `_speculative: False` → master plan. """ from __future__ import annotations import asyncio import json import logging import re from models.ai_client import AIClient _logger = logging.getLogger("agente_ai") # ── Prompt master (DeepSeek-R1): piano architetturale completo ─────────────── PLANNER_SYSTEM = """Sei un planner AI avanzato. Dato un obiettivo, decomponilo in subtask concreti. Rispondi SOLO con JSON valido nel formato: { "goal": "obiettivo originale", "complexity": "low|medium|high", "data_model": [ {"entity": "NomeEntità", "fields": ["id: str", "campo1: tipo", "campo2: tipo"]} ], "api_contract": [ {"method": "GET|POST|PUT|DELETE", "path": "/api/risorsa", "body": {}, "response": {"campo": "tipo"}} ], "subtasks": [ { "id": 1, "description": "cosa fare", "tool": "", "requires": [], "risk": "low|medium|high", "priority": "low|medium|high" } ], "parallel_groups": [[1,2],[3,4,5]], "estimated_steps": 3, "impacted_files": [] } data_model: lista di entità dati con i loro campi tipizzati (SOLO per goal con entità persistenti). - Ogni entità: {"entity": "Nome", "fields": ["campo: tipo", ...]} - Esempi di tipi: str, int, float, bool, datetime, list[str], dict. - Lascia [] per task senza entità dati (web search, domande, spiegazioni, singole funzioni). - P25-B5: definisci data_model PRIMA dei subtask di codice — è il contratto condiviso tra tutti i subtask. - I subtask di codice DEVONO referenziare le entità definite qui, MAI inventare nomi diversi on-the-fly. api_contract: endpoint REST/GraphQL con shape request+response (SOLO per goal con interfaccia API). - Ogni endpoint: {"method": "GET", "path": "/api/path", "body": {"campo": "tipo"}, "response": {"campo": "tipo"}} - Lascia [] per task senza endpoint API (script standalone, funzioni pure, task di analisi). - P25-B5: il primo subtask di codice backend DEVE implementare esattamente questo contratto. - MAI aggiungere endpoint non dichiarati qui senza aggiornare api_contract nel piano. parallel_groups: lista di liste di id subtask che possono girare in PARALLELO tra loro. - Ogni lista interna = un gruppo di subtask eseguibili contemporaneamente (nessuna dipendenza reciproca). - Subtask con requires:[] vanno sempre in un gruppo parallelo. - Esempio: backend (id:1,2) e frontend (id:3,4) senza dipendenze reciproche → [[1,2],[3,4]]. - Se tutto è sequenziale: [[1],[2],[3]]. impacted_files: lista di path file VFS che potrebbero essere impattati (vuota se non applicabile). Tool disponibili: web_search — cerca informazioni online in tempo reale code — genera/modifica codice (Python, TS, JS, etc.) read_page — legge una pagina web per URL memory — accede a dati precedentemente memorizzati direct_response — risposta diretta senza tool esterni send_email — invia email via Resend database_query — esegue query su database PostgreSQL/SQLite web_research — ricerca approfondita multi-fonte con sintesi AI execute_sql — esegue SQL su dati in-memory create_pdf — genera un documento PDF call_api — chiama un REST API esterno generate_image — genera un'immagine AI run_python — esegue codice Python in sandbox sicura write_file — scrive un file nel filesystem virtuale (risk: medium) read_file — legge un file dal filesystem virtuale (risk: low) apply_patch — applica una patch unificata a un file (risk: medium) execute_shell — esegue un comando shell in sandbox (risk: high) directory_tree — elenca struttura ad albero di una directory (risk: low) file_search — cerca pattern nei file con grep (risk: low) git_status — mostra branch corrente e file modificati (risk: low) git_clone — clona una repo remota (risk: medium) git_diff — mostra modifiche in sospeso (risk: low) git_commit — esegue add -A + commit (risk: medium) npm_install — installa dipendenze node (risk: medium) npm_run — esegue script node (risk: medium) pip_install — installa pacchetti Python (risk: medium) type_check — type check tsc o mypy (risk: low) scaffold_project — crea struttura progetto da template (risk: medium) delegate_task — delega un sotto-obiettivo a un micro-agente indipendente (risk: low) REGOLA TASK SEMPLICI (S-ROBUSTNESS): Se il task chiede una singola funzione TypeScript pura (sum, add, calculate, map, filter) anche se il prompt ha rumore/distrazioni/noise: → piano con 1 SOLO subtask: tool=direct_response → MAI run_python, type_check o npm_run per funzioni TypeScript di 1-3 righe REGOLA DATA INTEGRITY (S-RECOVERY): Prima di pianificare analisi su dati numerici: - Controlla: conversion rate > 100%? conversioni > utenti? → IMPOSSIBILE - Se dati impossibili → piano con SOLO 1 subtask: tool=direct_response description: "segnala anomalia nei dati: incoerente/impossibile, non calcolare" - MAI pianificare run_python/execute_sql su dati statisticamente impossibili REGOLA ASSOLUTA (S-GAP2): Per qualsiasi richiesta di creazione app/progetto/boilerplate, DEVI verificare se esiste scaffold_project corrispondente. Se esiste → PRIMO subtask. REGOLE GRAFO DI DIPENDENZE: - requires:[] → subtask eseguibile immediatamente in parallelo con altri requires:[] - requires:[N] → subtask che dipende dall'output di subtask id N - priority:high: subtask bloccante; i dipendenti mettono il suo id in requires - priority:low: subtask indipendente; eseguibile in parallelo - Identifica SEMPRE rami indipendenti (es. Backend vs Frontend, Read vs Write diversi file) - Aggiungi entrambi i rami in parallel_groups per massimizzare il parallelismo""" # ── Prompt quick-start (Cerebras/Groq): 2-3 passi immediati ───────────────── PLANNER_QUICK_SYSTEM = """Sei un planner rapido. Dato un obiettivo, genera SOLO i primi 2-3 passi immediati e concreti. Rispondi SOLO con JSON valido: { "goal": "obiettivo", "complexity": "low|medium|high", "subtasks": [ {"id": 1, "description": "primo passo", "tool": "", "requires": [], "risk": "low", "priority": "high"}, {"id": 2, "description": "secondo passo", "tool": "", "requires": [1], "risk": "low", "priority": "medium"} ], "parallel_groups": [[1],[2]], "estimated_steps": 2, "impacted_files": [] } Regole: - MAX 3 subtask — solo le azioni più immediate e ovvie - Scegli tool giusto: web_search/read_page per info, run_python per codice, write_file per file - Non pianificare l'intero progetto — solo il "prossimo passo" logico - requires:[] per passi indipendenti (possono partire subito in parallelo)""" def _extract_json_balanced(raw: str) -> str | None: """P16-B3: depth-counting bilanciato — sostituisce regex greedy r'{[\s\S]+}'. Gestisce JSON annidati correttamente (piani con subtask oggetti complessi). """ depth = 0 start = -1 for i, ch in enumerate(raw): if ch == '{': if depth == 0: start = i depth += 1 elif ch == '}': depth -= 1 if depth == 0 and start != -1: return raw[start:i + 1] return None def _parse_plan(raw: str) -> dict | None: """Estrae e valida il JSON del piano dalla risposta LLM.""" if not raw: return None json_match = _extract_json_balanced(raw) if not json_match: return None try: plan = json.loads(json_match) if not plan.get("subtasks"): return None return plan except (json.JSONDecodeError, ValueError): return None class Planner: def __init__(self, llm_client: AIClient | None = None): if llm_client is not None: self.llm = llm_client else: # Master planner: DeepSeek-R1 per deep reasoning try: from models.role_router import RoleRouter, Role self.llm = RoleRouter.get_client(Role.ARCHITECT) except Exception: self.llm = AIClient() @classmethod def from_ollama(cls, ollama=None) -> "Planner": return cls() def _get_fast_llm(self) -> AIClient: """Gap-5: Cerebras gpt-oss-120b (2000+ tok/s) per quick-start draft. Fallback: Groq llama-3.1-8b-instant se CEREBRAS_API_KEY assente.""" try: from models.role_router import RoleRouter, Role return RoleRouter.get_client(Role.REASONER) # Cerebras 120B except Exception as _exc: _logger.debug("[planner] silenced %s", type(_exc).__name__) # noqa: BLE001 try: from models.role_router import RoleRouter, Role return RoleRouter.get_client(Role.FAST) # Groq 8B fallback except Exception: return AIClient() def _build_messages(self, system: str, goal: str, context: list | None = None) -> list[dict]: msgs = [ {"role": "system", "content": system}, {"role": "user", "content": f"Obiettivo: {goal}"}, ] if context: ctx_str = "\n".join(m.get("content", "")[:500] for m in context[-5:]) msgs[1]["content"] += f"\n\nContesto recente:\n{ctx_str}" return msgs async def create_plan(self, goal: str, context: list | None = None, model: str | None = None) -> dict: """ Gap-5: Speculative Hybrid Planning. Lancia in parallelo: 1. Quick-Start (Cerebras/Groq) — risponde in < 500ms con 2-3 subtask immediati 2. Master Plan (DeepSeek-R1) — risponde in 8-15s con piano architetturale completo Logica: - attende max QUICK_TIMEOUT per il quick-start - se arriva → restituisce subito (flag _speculative=True) così l'agente parte - se DeepSeek-R1 arriva prima → piano completo (flag _speculative=False) - se entrambi timeout → fallback euristico """ QUICK_TIMEOUT = 1.2 # secondi: soglia "fast-first" win MASTER_TIMEOUT = 30.0 # secondi: timeout totale DeepSeek-R1 msgs_quick = self._build_messages(PLANNER_QUICK_SYSTEM, goal, context) msgs_master = self._build_messages(PLANNER_SYSTEM, goal, context) fast_llm = self._get_fast_llm() async def _call_quick() -> dict | None: try: raw = await asyncio.wait_for( fast_llm.chat(msgs_quick, temperature=0.2, max_tokens=512), timeout=QUICK_TIMEOUT, ) plan = _parse_plan(raw) if plan: plan["_speculative"] = True plan["_raw"] = raw[:400] return plan except Exception: return None async def _call_master() -> dict | None: for _attempt in range(3): try: raw = await asyncio.wait_for( self.llm.chat(msgs_master, temperature=0.3, max_tokens=2048), timeout=MASTER_TIMEOUT, ) plan = _parse_plan(raw) if plan: plan["_speculative"] = False plan["_raw"] = raw[:400] return plan except (asyncio.TimeoutError, TimeoutError): if _attempt < 2: await asyncio.sleep(1.0 * (2 ** _attempt)) continue break except Exception: break return None # ── Speculative dual-fire ──────────────────────────────────────────── # Entrambi i modelli partono simultaneamente. # asyncio.wait(FIRST_COMPLETED) con soglia QUICK_TIMEOUT: # - Se quick-start risponde prima → agente parte subito (< 1s) # - Master plan continua in background; il loop lo ignora (non ha callback) # - Se master arriva per primo (es. R1 cold-start veloce) → piano completo quick_task = asyncio.create_task(_call_quick()) master_task = asyncio.create_task(_call_master()) done, pending = await asyncio.wait( {quick_task, master_task}, timeout=QUICK_TIMEOUT, return_when=asyncio.FIRST_COMPLETED, ) # Caso 1: quick-start ha risposto entro QUICK_TIMEOUT if quick_task in done: quick_plan = quick_task.result() if quick_plan: # P25-B3: per goal complessi (app/progetto/sistema), ignora quick-plan con ≤3 subtask # e attendi il master plan architetturale. Il quick-plan 2-3 step causa rework # sistematico su task multi-file che richiedono schema + contratto API. _COMPLEX_APP_RE = re.compile( r'\b(crea\s+(?:una\s+)?(?:app|applicazione|sistema|sito|progetto|piattaforma|servizio)|' r'build\s+(?:a\s+)?(?:app|system|platform|service|website)|' r'sviluppa|realizza|implementa\s+(?:un[ao]\s+)?(?:sistema|app|servizio)|' r'full.?stack|backend\s+e\s+frontend|frontend\s+e\s+backend)\b', re.IGNORECASE, ) _is_complex_app = bool(_COMPLEX_APP_RE.search(goal)) and len(goal) > 60 _n_quick_subtasks = len(quick_plan.get("subtasks", [])) if _is_complex_app and _n_quick_subtasks <= 3: _logger.info( "P25-B3: goal complesso rilevato (%d subtask quick) — attendo master plan", _n_quick_subtasks, ) # Non restituire il quick-plan; lascia cadere al Caso 2 (wait master) else: _logger.info( "Gap-5 speculative: quick-start plan (%d subtask), master in background", _n_quick_subtasks, ) # Master task continua in background; risultato non bloccante master_task.add_done_callback( lambda t: ( _logger.info( "Gap-5 master plan ready (%d subtask) — successiva chiamata beneficerà", len((t.result() or {}).get("subtasks", [])) if not t.cancelled() and t.exception() is None else 0, ) if not t.cancelled() and t.exception() is None else None ) ) return quick_plan # Caso 2: nessuno ha risposto in QUICK_TIMEOUT — aspetta il master plan if master_task in done: master_plan = master_task.result() if master_plan: quick_task.cancel() return master_plan # Caso 3: nessuno ancora pronto — aspetta fino a MASTER_TIMEOUT remaining = {t for t in {quick_task, master_task} if not t.done()} if remaining: done2, _ = await asyncio.wait(remaining, timeout=MASTER_TIMEOUT - QUICK_TIMEOUT) for t in done2: if t.exception() is None and not t.cancelled(): result = t.result() if result: for other in remaining - {t}: other.cancel() return result # Cancella task pendenti for t in {quick_task, master_task}: if not t.done(): t.cancel() # ── Fallback euristico (piano semplice) ────────────────────────────── _logger.warning("Gap-5 planner: tutti i modelli in timeout per goal: %s", goal[:80]) _g = goal.lower() _ft, _fr = "direct_response", "low" if re.search(r"https?://", _g): _ft = "read_page" elif re.search(r"\b(cerca|search|notizie|news)\b", _g): _ft = "web_search" elif re.search(r"\b(git|branch|commit|diff)\b", _g): _ft = "git_status" elif re.search(r"\b(struttura|directory|tree|elenca)\b", _g): _ft = "directory_tree" elif re.search(r"\b(grep|occorrenze|cerca.*codice)\b", _g): _ft = "file_search" elif re.search(r"\b(npm|pnpm|yarn)\b", _g): _ft, _fr = "npm_run", "medium" elif re.search(r"\b(pip |pip3 |installa pacchett)\b", _g): _ft, _fr = "pip_install", "medium" elif re.search(r"\b(codice|python|script)\b", _g): _ft, _fr = "run_python", "medium" elif re.search(r"\b(scaffold|bootstrap)\b|crea.*app|crea.*progetto|nuovo.*progetto", _g): _ft, _fr = "scaffold_project", "medium" elif re.search(r"\b(genera|crea).*immagine\b", _g): _ft, _fr = "generate_image", "medium" return { "goal": goal, "complexity": "medium", "subtasks": [{"id": 1, "description": goal, "tool": _ft, "requires": [], "risk": _fr, "priority": "high"}], "parallel_groups": [[1]], "estimated_steps": 1, "impacted_files": [], "_fallback": True, }