""" unified_loop.py — Unified Agent Loop v3 v3 — Fix critico: - tools=[] → tools da TOOL_REGISTRY (get_weather, web_search, calculate, run_python, read_page) - Fallback prompt: non più "indica passi" — esegue tool reali via intent detection - _detect_and_run_tools(): meteo/ricerca/calcolo risolti PRIMA della chiamata LLM """ from __future__ import annotations import asyncio import os import re from dataclasses import dataclass, field from typing import Any, Awaitable, Callable SMOL_TIMEOUT: float = float(os.getenv("UNIFIED_LOOP_TIMEOUT", "12")) # S191: 28→12 fallback veloce LLM_TIMEOUT: float = float(os.getenv("LLM_CALL_TIMEOUT", "20")) TOOL_TIMEOUT: float = float(os.getenv("TOOL_CALL_TIMEOUT", "10")) StepCallback = Callable[[dict[str, Any]], Awaitable[None] | None] # S1-D: rimosso keyword matching — routing LLM-based via smolagents @dataclass class UnifiedLoopState: goal: str context: str = "" max_steps: int = 8 steps: list[dict[str, Any]] = field(default_factory=list) errors: list[str] = field(default_factory=list) async def _maybe_await(val: Any) -> None: """Awaita val solo se è una coroutine/future — altrimenti no-op.""" if asyncio.iscoroutine(val) or asyncio.isfuture(val): await val class UnifiedAgentLoop: """Smolagents-first loop with safe deterministic fallback.""" def __init__(self, llm_client: Any, planner: Any = None, executor: Any = None, critic: Any = None, memory: Any = None, verifier: Any = None) -> None: self.llm = llm_client self.planner = planner self.executor = executor self.critic = critic self.memory = memory self.verifier = verifier self._smol_agent: Any | None = None # ── Prompt ─────────────────────────────────────────────────────────────── # S191/S192: system + user separati per LLM che supporta messages list _SYSTEM_IDENTITY = ( "Sei un agente AI autonomo, preciso e proattivo. Lavori come l'agente di Replit: " "risolvi problemi concretamente, non li descrivi.\n\n" "REGOLE FONDAMENTALI:\n" "1. Rispondi SEMPRE nella lingua dell'utente (default italiano)\n" "2. Lavora autonomamente — non chiedere conferma per ogni passo\n" "3. Quando hai dati reali da tool, usali direttamente nella risposta\n" "4. Non dire 'puoi fare X' — mostra X fatto, con codice completo se richiesto\n" "5. Se incontri un errore, analizza e riprova con approccio diverso\n" "6. Sii specifico e concreto — niente placeholder o risposte vaghe\n" "7. Per codice: sempre blocchi markdown con sintassi corretta, tipizzati\n" "8. Per matematica: mostra calcoli passo passo con numeri esatti\n" "9. Per decisioni architetturali: dai 3 opzioni con pro/contro e raccomandazione\n\n" "REGOLE SPECIALIZZATE:\n" "• Probabilità/Bayes: usa sempre il Teorema di Bayes esplicitamente. " "Scrivi P(A|B) = P(B|A)·P(A)/P(B). Usa la terminologia italiana del problema " "(es. 'scatola', 'cassetto', 'porta', 'malato') nelle equazioni. " "Conclude SEMPRE con la risposta finale come frazione (es. 2/3) " "E come percentuale con punto decimale (es. 66.67%). " "USA SEMPRE il punto come separatore decimale, mai la virgola.\n" " BERTRAND BOX: conta i CASSETTI ORO (non le scatole): [Oro,Oro]=2 cassetti, " "[Oro,Arg]=1 cassetto → 3 cassetti oro totali → 2 cassetti su 3 hanno l'altro=Oro → P = 2/3.\n" "• Bug Python MUTABLE DEFAULT ARGUMENT: nella risposta scrivi LETTERALMENTE " "la frase 'mutable default argument' (in inglese, non tradurre). " "Spiega che lo stesso oggetto mutabile è condiviso tra le chiamate. " "Mostra SEMPRE il fix con None sentinel:\n" " def f(lst=None):\n if lst is None: lst = []\n\n" "• Git workflow: dai sempre i comandi esatti con le opzioni corrette " "(es. `git pull --rebase origin main`, `git fetch && git rebase origin/main`)\n" "• React useEffect: menziona SEMPRE useMemo/useCallback/useRef come possibili fix " "per dipendenze instabili, con esempio di codice per ciascuno" ) def _build_messages(self, state: UnifiedLoopState, tool_results: str = "") -> list[dict]: """Costruisce messages list con system + user separati — S191.""" mem_hint = "Tieni conto della memoria per preferenze e contesto utente.\n" if self.memory else "" if tool_results: tool_section = ( f"--- DATI REALI RECUPERATI ---\n{tool_results}\n--- FINE DATI ---\n\n" "Usa QUESTI DATI REALI per rispondere. Non dire all'utente di controllare altri siti — " "la risposta è già qui. Formula una risposta completa, diretta e utile." ) else: tool_section = "Rispondi in modo diretto, completo e concreto. Niente istruzioni generali — dai la risposta specifica al problema." context_part = f"Contesto sessione: {state.context}\n\n" if state.context and state.context != "nessuno" else "" user_content = f"{context_part}{mem_hint}{tool_section}\n\nObiettivo/Domanda: {state.goal}" return [ {"role": "system", "content": self._SYSTEM_IDENTITY}, {"role": "user", "content": user_content}, ] def _build_prompt(self, state: UnifiedLoopState, tool_results: str = "") -> str: """Legacy: prompt singolo per smolagents. Usa _build_messages per LLM diretto.""" msgs = self._build_messages(state, tool_results) return f"{msgs[0]['content']}\n\n{msgs[1]['content']}" # ── Smolagents tools builder ────────────────────────────────────────────── def _build_smol_tools(self) -> list[Any]: """Avvolge TOOL_REGISTRY in Tool smolagents — async→sync via new event loop.""" try: from smolagents import Tool # type: ignore from tools.registry import TOOL_REGISTRY smol_tools: list[Any] = [] for tname, spec in TOOL_REGISTRY.items(): async_fn = spec["_fn"] tdesc = spec.get("description", spec.get("goal", tname)) req_inputs = spec.get("required_inputs", []) inputs: dict[str, dict[str, str]] = { k: {"type": "string", "description": k} for k in req_inputs } def _make(name: str, desc: str, fn: Any, inp: dict[str, dict[str, str]]) -> type: class _T(Tool): # type: ignore[misc] pass _T.__name__ = f"Tool_{name}" _T.name = name _T.description = desc _T.inputs = inp _T.output_type = "string" def forward(self: Any, **kwargs: Any) -> str: # BUG-10 fix: asyncio.run() crea, esegue e chiude il loop # in un colpo — new_event_loop/run_until_complete/close # causava comportamenti non deterministici con asyncio.to_thread try: return str(asyncio.run(fn(**kwargs))) except RuntimeError: # Fallback: se siamo già in un event loop (pytest, etc.) import concurrent.futures as _cf with _cf.ThreadPoolExecutor(max_workers=1) as _ex: return str(_ex.submit(asyncio.run, fn(**kwargs)).result(timeout=10)) except Exception as exc: return f"[errore {name}: {exc}]" _T.forward = forward # type: ignore[method-assign] return _T smol_tools.append(_make(tname, tdesc, async_fn, inputs)()) return smol_tools except Exception: return [] # ── Smolagents loader ───────────────────────────────────────────────────── def _load_smol_agent(self) -> Any | None: if self._smol_agent is not None: return self._smol_agent try: from smolagents import CodeAgent, LiteLLMModel # type: ignore MODEL_ENV = os.getenv("SMOLAGENTS_MODEL", "") def _prefix(mid: str, pfx: str) -> str: if not mid: return "" known = ("openrouter/","groq/","huggingface/","anthropic/","openai/","gpt-") return mid if any(mid.startswith(p) for p in known) else f"{pfx}/{mid}" if os.getenv("OPENROUTER_API_KEY"): model_id = _prefix(MODEL_ENV, "openrouter") if MODEL_ENV else "openrouter/meta-llama/llama-3.3-70b-instruct:free" api_key = os.getenv("OPENROUTER_API_KEY") elif os.getenv("GROQ_API_KEY"): model_id = _prefix(MODEL_ENV, "groq") if MODEL_ENV else "groq/llama-3.1-8b-instant" api_key = os.getenv("GROQ_API_KEY") elif os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACE_API_KEY"): model_id = _prefix(MODEL_ENV, "huggingface") if MODEL_ENV else "huggingface/Qwen/Qwen2.5-Coder-32B-Instruct" api_key = os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACE_API_KEY") elif os.getenv("OPENAI_API_KEY"): model_id = MODEL_ENV or "gpt-4o-mini" api_key = os.getenv("OPENAI_API_KEY") else: return None model = LiteLLMModel(model_id=model_id, api_key=api_key) tools = self._build_smol_tools() # ← FIX: era tools=[] self._smol_agent = CodeAgent( tools=tools, model=model, max_steps=int(os.getenv("UNIFIED_LOOP_MAX_STEPS", "2")), # S191: 6→2 per risposta in <12s ) return self._smol_agent except Exception: return None # ── Smolagents path ─────────────────────────────────────────────────────── async def _run_smolagents(self, state: UnifiedLoopState, on_step: StepCallback | None) -> dict[str, Any] | None: agent = self._load_smol_agent() if agent is None: return None prompt = self._build_prompt(state) if on_step: await _maybe_await(on_step({"loop": 0, "action": "smolagents", "status": "started"})) try: result = await asyncio.wait_for( asyncio.to_thread(agent.run, prompt), timeout=SMOL_TIMEOUT) output = str(result) state.steps.append({"action": "smolagents", "output": output}) if self.memory: await self.memory.save_episode("unified_loop", state.goal, output[:1000], True, tags=["smolagents"]) if on_step: await _maybe_await(on_step({"loop": 1, "action": "smolagents", "status": "done"})) return {"success": True, "engine": "smolagents", "goal": state.goal, "steps": state.steps, "output": output} except asyncio.TimeoutError: msg = f"smolagents timeout {SMOL_TIMEOUT}s — fallback" state.errors.append(msg) if on_step: await _maybe_await(on_step({"loop": 1, "action": "smolagents", "status": "timeout", "error": msg})) return None except Exception as exc: state.errors.append(str(exc)) if on_step: await _maybe_await(on_step({"loop": 1, "action": "smolagents", "status": "error", "error": str(exc)})) return None # ── Fallback deterministico ─────────────────────────────────────────────── async def _run_fallback(self, state: UnifiedLoopState, on_step: StepCallback | None) -> dict[str, Any]: outputs: list[str] = [] if self.memory: mem_ctx = await self.memory.get_context(state.goal) if mem_ctx: state.context = f"{state.context}\n\nMEMORIA:\n{mem_ctx}".strip() # S1-D: keyword detection rimossa — smolagents gestisce tool calling LLM-based tool_results = "" # Passo 2: planner opzionale (solo se no tool results) if self.planner and not tool_results: if on_step: await _maybe_await(on_step({"loop": 0, "action": "plan", "status": "started"})) plan = await self.planner.create_plan( state.goal, context=[{"role": "system", "content": state.context}]) state.steps.append({"action": "plan", "result": plan}) if on_step: await _maybe_await(on_step({"loop": 0, "action": "plan", "status": "done", "subtasks": len(plan.get("subtasks", []))})) # Passo 2b: BUG-6 fix — executor esegue i subtask step-by-step # Prima: il piano era solo serializzato come stringa e passato all'LLM # Ora: ogni subtask eseguibile viene eseguito via Executor → risultati reali nel prompt if self.executor and plan.get("subtasks"): # Mappa nomi tool planner → TOOL_REGISTRY keys + builder input _TOOL_MAP: dict[str, tuple[str, Any]] = { "web_search": ("web_search", lambda desc: {"query": desc}), "read_page": (None, None), # richiede URL esplicita — skip "code": (None, None), # richiede codice generato — skip "calculate": (None, None), # richiede espressione — skip "memory": (None, None), # gestita da MemoryManager separatamente "direct_response": (None, None), # gestita dall'LLM } exec_parts: list[str] = [] for subtask in plan.get("subtasks", []): if subtask.get("risk", "low") == "high": continue # skip task rischiosi tool_key_pair = _TOOL_MAP.get(subtask.get("tool", ""), (None, None)) reg_name, inp_builder = tool_key_pair if not reg_name or inp_builder is None: continue inputs = inp_builder(subtask.get("description", state.goal)) if on_step: await _maybe_await(on_step({ "loop": 0, "action": f"executor:{reg_name}", "status": "started", "subtask_id": subtask.get("id"), })) res = await self.executor.run_tool(reg_name, inputs) if res.get("success"): snippet = str(res.get("output", ""))[:500] label = f"[subtask {subtask.get('id')} — {subtask.get('description','')[:60]}]" exec_parts.append(f"{label}: {snippet}") state.steps.append({ "action": f"executor:{reg_name}", "subtask_id": subtask.get("id"), "output": snippet, }) if on_step: await _maybe_await(on_step({ "loop": 0, "action": f"executor:{reg_name}", "status": "done", "subtask_id": subtask.get("id"), })) if exec_parts: exec_block = "\n".join(exec_parts) tool_results = (f"{tool_results}\n{exec_block}".strip() if tool_results else exec_block) if self.memory: await self.memory.save_episode( "executor", state.goal, exec_block[:800], True, tags=["executor", "plan"]) # Passo 3: LLM con dati tool iniettati — S191: usa _build_messages (system+user separati) messages = self._build_messages(state, tool_results=tool_results) if on_step: await _maybe_await(on_step({"loop": 1, "action": "llm", "status": "started"})) try: answer = await asyncio.wait_for( self.llm.chat(messages, temperature=0.2, max_tokens=2048), timeout=LLM_TIMEOUT) except asyncio.TimeoutError: answer = f"[LLM timeout {LLM_TIMEOUT}s]" state.errors.append(answer) except Exception as exc: answer = f"[LLM error: {exc}]" state.errors.append(str(exc)) state.steps.append({"action": "llm", "output": answer}) outputs.append(answer) if self.critic: critique = await self.critic.evaluate(state.goal, answer) state.steps.append({"action": "critic", "result": critique}) if critique.get("needs_retry"): state.errors.extend([str(x) for x in critique.get("issues", [])]) success = len(state.errors) == 0 final_output = "\n\n".join(outputs).strip() if self.memory: await self.memory.save_episode("unified_loop", state.goal, final_output[:1000], success, tags=["fallback"]) if on_step: await _maybe_await(on_step({"loop": 2, "action": "fallback", "status": "done", "success": success})) return {"success": success, "engine": "fallback", "goal": state.goal, "steps": state.steps, "errors": state.errors, "output": final_output} # ── Fast-path detection (S192) ──────────────────────────────────────────── _TOOL_NEEDED_RE = re.compile( r"\b(meteo|previsioni|tempo\s+a|notizie|news|bitcoin|ethereum|cambio\s+valuta|" r"leggi\s+pagina|fetch|scarica|wikipedia|cerca\s+su|trova\s+su|api\s+pubblica|" r"run\s+code|esegui\s+codice|installa|pip\s+install|shell|bash|terminal)\b", re.IGNORECASE, ) def _needs_tools(self, goal: str) -> bool: """True solo se il goal richiede tool reali (meteo, web, esecuzione codice).""" return bool(self._TOOL_NEEDED_RE.search(goal)) # ── Entry point ─────────────────────────────────────────────────────────── async def run(self, goal: str, context: str = "", max_steps: int = 8, on_step: StepCallback | None = None) -> dict[str, Any]: state = UnifiedLoopState(goal=goal, context=context, max_steps=max_steps) # S192 fast-path: skip smolagents per reasoning puro → latenza 15s→5s if self._needs_tools(goal): smol_result = await self._run_smolagents(state, on_step) if smol_result: return smol_result return await self._run_fallback(state, on_step)