""" research.py — Ricerca web multi-URL con ARL (Adaptive Research Loop). Endpoint: POST /api/web/research — ricerca multi-round con raffinamento automatico della query Flusso (V5 — ARL + parallel multi-query + GF-7 smart retry): 1. Round 0: _gen_alt_queries() → 3 query parallele → corpus unificato (GAP-RESEARCH-PARALLEL) 2. Calcola goal_coverage: frazione di token-goal presenti nel corpus 3. Se coverage >= MIN_COVERAGE → esci 4. Se coverage < MIN_COVERAGE → _refine_query() → round successivo (max MAX_ROUNDS) 5. GF-7: se new_urls vuoto (tutti già visitati) O ok_pages vuoto → cambia query strategy invece di uscire: prova variante lessicale non ancora tentata, poi esci solo se esaurite 6. Sintesi finale Groq (se key disponibile) Budget (iPhone free tier): - Round 0: 3 query × N URL = max 3×6=18 fetch in parallelo (≈ stessa latenza di 6 sequenziali) - Round 1+: max 4 round × 6 URL = max 24 fetch totali - Hard timeout: MIN_COVERAGE=0.70 o page budget esaurito """ import asyncio, os, re, logging, time import httpx from fastapi import APIRouter, HTTPException, Request from pydantic import BaseModel router = APIRouter(prefix="/api/web", tags=["web-research"]) _logger = logging.getLogger("research") _UA = "Mozilla/5.0 (compatible; AgenteAI/3.0)" _MAX_SRC = 8 # max URL totali per ricerca _MAX_ROUNDS = 5 # QF-1: 3→5 — più coverage per task complessi _MAX_URLS_PER_ROUND = 6 # QF-1: 4→6 URL per round — maggiore coverage fonti _MIN_COVERAGE = 0.70 # token goal coverage per uscita anticipata _TIMEOUT_S = 35.0 # QF-1: 20→35s timeout ARL — supporta 5 round × 6 URL class ResearchRequest(BaseModel): topic: str depth: int = 4 # number of sources per round lang: str = "it" # language for results synthesize: bool = True # use LLM to synthesize if key available # ─── ARL helpers ───────────────────────────────────────────────────────────── _STOP_WORDS = { "the","and","for","with","that","this","from","into","have","are","was","were", "che","del","della","per","con","una","uno","gli","dei","nel","nelle","alla", "sulle","degli","sono","come","quando","dove","anche","quindi","essere", } def _tokenize(txt: str) -> set[str]: return {t for t in re.split(r'\W+', txt.lower()) if len(t) > 3 and t not in _STOP_WORDS} def _goal_coverage(goal: str, corpus: str) -> float: """Calcola la fraction di token-goal presenti nel corpus (ARL coverage check).""" goal_toks = _tokenize(goal) if not goal_toks: return 1.0 found_toks = _tokenize(corpus) return sum(1 for t in goal_toks if t in found_toks) / len(goal_toks) def _refine_query(goal: str, corpus: str) -> str | None: """Genera query di refinement basata sui token-goal mancanti nel corpus. Zero LLM — solo analisi lessicale. """ goal_list = [t for t in re.split(r'\W+', goal.lower()) if len(t) > 3 and t not in _STOP_WORDS] found_toks = _tokenize(corpus) missing = [t for t in goal_list if t not in found_toks] if not missing: return None # Suffix: ultimi 2 token del goal NON già nei missing (evita duplicati) missing_set = set(missing) raw_suffix = [w for w in goal.strip().split()[-2:] if w.lower().rstrip("?!.,;:") not in missing_set] raw_query = f"{' '.join(missing[:3])} {' '.join(raw_suffix)}".strip() # Dedup mantenendo l'ordine (evita "finanziamento finanziamento") seen: set[str] = set() parts = [] for w in raw_query.split(): k = w.lower() if k not in seen: seen.add(k) parts.append(w) return " ".join(parts) if parts else None # U1: mappatura IT→EN per variante inglese nelle query italiane. # ~30 termini tecnici comuni; nomi propri (React, Vue, Docker…) invariati. _IT_EN_MAP: dict[str, str] = { 'autenticazione': 'authentication', 'accesso': 'login', 'registrazione': 'registration', 'database': 'database', 'applicazione': 'application', 'componente': 'component', 'pagina': 'page', 'schermata': 'screen', 'utente': 'user', 'errore': 'error', 'configurazione': 'configuration', 'installazione': 'installation', 'distribuzione': 'deployment', 'sviluppo': 'development', 'funzione': 'function', 'classe': 'class', 'metodo': 'method', 'variabile': 'variable', 'importazione': 'import', 'modulo': 'module', 'libreria': 'library', 'integrazione': 'integration', 'servizio': 'service', 'richiesta': 'request', 'risposta': 'response', 'connessione': 'connection', 'sicurezza': 'security', 'ottimizzazione': 'optimization', 'creare': 'create', 'costruire': 'build', 'implementare': 'implement', 'aggiungere': 'add', 'aggiornare': 'update', 'chiamata': 'call', 'interfaccia': 'interface', 'progetto': 'project', 'struttura': 'structure', } # Indicatori lessicali italiani (articoli, preposizioni, verbi comuni) _IT_INDICATORS: frozenset[str] = frozenset({ 'come', 'cosa', 'perché', 'quando', 'dove', 'quale', 'quali', 'creare', 'fare', 'usare', 'avere', 'essere', 'per', 'con', 'una', 'uno', 'del', 'della', 'degli', 'dal', 'dalla', 'degli', }) def _is_italian(text: str) -> bool: """Rileva se il testo è principalmente italiano — euristica lessicale leggera.""" toks = {w.lower() for w in re.split(r'\W+', text) if w} return len(toks & _IT_INDICATORS) >= 2 def _translate_it_en(goal: str) -> str: """Traduzione IT→EN zero-LLM: sostituisce termini tecnici mappati, rimuove articoli/preposizioni italiani senza corrispondente inglese.""" _it_stopwords = {'il', 'lo', 'la', 'le', 'gli', 'i', 'un', "un'", 'uno', 'una', 'di', 'da', 'in', 'con', 'su', 'per', 'tra', 'fra', 'che', 'come'} parts = [] for w in goal.split(): clean = w.lower().strip("'!?.,;:") if clean in _it_stopwords: continue # articoli/preposizioni → drop en = _IT_EN_MAP.get(clean) if en: parts.append(en) elif not re.match(r'^[a-z]+', clean) or len(w) > 2: parts.append(w) # parola non-IT o nome proprio → mantieni return ' '.join(parts).strip() def _gen_fallback_queries(goal: str, tried: set[str]) -> list[str]: """GF-7: genera varianti di query non ancora tentate quando new_urls/ok_pages è vuoto. Chiamata solo quando il loop si troverebbe ad uscire con 0 nuovi URL o 0 pagine leggibili — invece di arrendersi, prova angolazioni diverse: v1 — inversione ordine parole chiave (cerca complemento prima del soggetto) v2 — aggiunge "tutorial" / "guida" / "come" (disambigua intent informativo) v3 — singola keyword più specifica (narrow search su termine principale) Zero LLM. Restituisce solo varianti non già in `tried`. """ words = [w for w in re.split(r'\W+', goal) if len(w) > 2 and w.lower() not in _STOP_WORDS] candidates: list[str] = [] # v1 — inversione ultime/prime keyword if len(words) >= 3: v1 = " ".join(words[len(words)//2:] + words[:len(words)//2]) candidates.append(v1) # v2 — intent informativo esplicito kw_core = " ".join(words[:4]) for prefix in ("come funziona", "guida", "spiegazione"): v2 = f"{prefix} {kw_core}".strip() candidates.append(v2) break # un solo prefisso # v3 — termine più specifico (seconda keyword, spesso più discriminante) if len(words) >= 2: v3 = words[1] if len(words[1]) > 4 else (words[0] if len(words[0]) > 4 else "") if v3: candidates.append(v3) # D8: v4 — inversione completa delle keyword (garantisce query diversa da _gen_alt_queries) if len(words) >= 3: v4 = ' '.join(reversed(words)) if v4 not in set(candidates): candidates.append(v4) return [c for c in candidates if c and c.lower() not in tried][:3] # era :2, ora :3 per v4 # ─── Search: usa pipeline web_search.py (Brave → Tavily → Wikipedia → HN) ──── async def _pipeline_search(query: str, n: int) -> list[dict]: """Usa la pipeline condivisa web_search.py — stessa logica di _run_direct_tools.""" try: from tools.web_search import web_search as _ws result = await _ws(query, max_results=n) hits = result.get("results", []) if hits: return [{"url": h["url"], "title": h.get("title", "")} for h in hits if h.get("url")] except Exception as exc: _logger.debug("pipeline_search fallback: %s", exc) return [] async def _ddg_fallback_search(query: str, n: int) -> list[dict]: """Fallback DDG HTML parse quando nessuna chiave API è configurata.""" try: _ddg_kl = "it-it" if _is_italian(query) else "en-us" # B-GAP-D: locale EN-aware _ddg_al = "it-IT,it;q=0.9,en;q=0.8" if _ddg_kl == "it-it" else "en-US,en;q=0.9" async with httpx.AsyncClient(timeout=10, headers={"User-Agent": _UA, "Accept-Language": _ddg_al}) as c: r = await c.get("https://html.duckduckgo.com/html/", params={"q": query, "kl": _ddg_kl}) if r.status_code != 200: return [] html = r.text link_pattern = re.compile(r']+class="result__url"[^>]*href="([^"]+)"[^>]*>([^<]*)', re.DOTALL) title_pattern = re.compile(r']+class="result__a"[^>]*href="[^"]+"[^>]*>([^<]+)', re.DOTALL) links = link_pattern.findall(html) titles = [re.sub(r"\s+", " ", t).strip() for t in title_pattern.findall(html)] results = [] for i, (url, _) in enumerate(links[:n]): if url.startswith("http"): results.append({"url": url, "title": titles[i] if i < len(titles) else url}) return results[:n] except Exception: return [] # ─── Page content extraction ────────────────────────────────────────────────── async def _fetch_page(url: str, max_chars: int = 2000) -> dict: try: async with httpx.AsyncClient(timeout=10, follow_redirects=True, headers={"User-Agent": _UA}) as c: r = await c.get(url) if r.status_code != 200: return {"url": url, "ok": False, "error": f"HTTP {r.status_code}"} html = r.text try: import trafilatura text = trafilatura.extract( html, include_comments=False, include_tables=False, favor_recall=True, deduplicate=True, ) or "" except ImportError: text = re.sub(r"<[^>]+>", " ", html) text = re.sub(r"\s{2,}", " ", text).strip() noise = {"cookie","accept all cookies","privacy policy","terms of service","subscribe","follow us on"} lines = [l for l in text.split("\n") if len(l.strip()) > 4 and not any(n in l.lower() for n in noise)] text = "\n".join(lines) title_m = re.search(r"]*>([^<]+)", html, re.IGNORECASE) title = title_m.group(1).strip()[:120] if title_m else url return {"url": url, "title": title, "text": text[:max_chars], "ok": True} except Exception as e: return {"url": url, "ok": False, "error": str(e)[:100]} # ─── LLM synthesis (Groq) ───────────────────────────────────────────────────── async def _synthesize(topic: str, sources: list[dict]) -> str: groq_key = os.getenv("GROQ_API_KEY", "") if not groq_key: return "" context = "\n\n".join( f"[{i+1}] {s['title']}\n{s['text'][:800]}" for i, s in enumerate(sources) if s.get("ok") and s.get("text") )[:6000] try: async with httpx.AsyncClient(timeout=30) as c: r = await c.post( "https://api.groq.com/openai/v1/chat/completions", headers={"Authorization": f"Bearer {groq_key}", "Content-Type": "application/json"}, json={ "model": "llama-3.1-8b-instant", "max_tokens": 700, "messages": [ {"role": "system", "content": "Sei un assistente che sintetizza informazioni web. Rispondi sempre in italiano. Sii conciso e preciso."}, {"role": "user", "content": ( f"Argomento: **{topic}**\n\n" f"Fonti trovate:\n{context}\n\n" "Sintetizza le informazioni principali in 3-5 punti chiave, citando le fonti [N]." )}, ], }, ) if r.status_code == 200: _chs = r.json().get("choices") or [] return (_chs[0].get("message", {}).get("content") or "") if _chs else "" except Exception as exc: _logger.debug("synthesis error: %s", exc) return "" # ─── Main endpoint ──────────────────────────────────────────────────────────── @router.post("/research") async def web_research(req: ResearchRequest, request: Request): # S-GAP23: X-Internal-Token guard — protegge consumi Groq API da abusi esterni. _internal_token = os.getenv('INTERNAL_TOKEN', '') if _internal_token and request.headers.get('X-Internal-Token') != _internal_token: raise HTTPException(401, 'Unauthorized') n = min(max(int(req.depth), 1), _MAX_URLS_PER_ROUND) # ── ARL loop ─────────────────────────────────────────────────────────────── _t0 = time.monotonic() visited = set() corpus = "" query = req.topic all_pages: list[dict] = [] rounds = 0 # GF-7: traccia tutte le query tentate per evitare duplicati nei fallback tried_queries: set[str] = {query.lower()} while rounds < _MAX_ROUNDS and (time.monotonic() - _t0) < _TIMEOUT_S: # 1. Search ───────────────────────────────────────────────────────────── if rounds == 0: # GAP-RESEARCH-PARALLEL: round 0 usa 3 query in parallelo per massimizzare # la coverage iniziale senza penalità di latenza (asyncio.gather). alt_queries = _gen_alt_queries(query) if alt_queries: search_batches = await asyncio.gather( _pipeline_search(query, n), *[_pipeline_search(q, n) for q in alt_queries], ) # Dedup preservando ordine: priorità alla query principale _seen_u: set[str] = set() results: list[dict] = [] for batch in search_batches: for r in batch: if r["url"] not in _seen_u: _seen_u.add(r["url"]) results.append(r) _logger.debug( "ARL round 0 parallel: %d queries → %d unique URLs", 1 + len(alt_queries), len(results), ) # GF-7: registra le alt_queries come già tentate for aq in alt_queries: tried_queries.add(aq.lower()) else: results = await _pipeline_search(query, n) else: results = await _pipeline_search(query, n) if not results: results = await _ddg_fallback_search(query, n) if not results: # GF-7: search completamente vuota → prova query alternativa non ancora tentata fb_queries = _gen_fallback_queries(req.topic, tried_queries) if fb_queries: query = fb_queries[0] tried_queries.add(query.lower()) _logger.debug("GF-7: search vuota → fallback query: %r", query) rounds += 1 continue break # 2. Fetch new URLs only ──────────────────────────────────────────────── new_urls = [r["url"] for r in results if r["url"] not in visited][:n] if not new_urls: # GF-7: tutti gli URL già visitati → cambia query invece di arrendersi fb_queries = _gen_fallback_queries(req.topic, tried_queries) if fb_queries: query = fb_queries[0] tried_queries.add(query.lower()) _logger.debug("GF-7: new_urls vuoto → fallback query: %r", query) rounds += 1 continue break for u in new_urls: visited.add(u) pages = await asyncio.gather(*[_fetch_page(u) for u in new_urls]) ok_pages = [p for p in pages if p.get("ok") and p.get("text")] # GF-7: pagine fetch tutte fallite (bloccate/vuote) → cambia query if not ok_pages: fb_queries = _gen_fallback_queries(req.topic, tried_queries) if fb_queries: query = fb_queries[0] tried_queries.add(query.lower()) _logger.debug("GF-7: ok_pages vuoto → fallback query: %r", query) rounds += 1 continue break all_pages.extend(ok_pages) # 3. Build corpus ─────────────────────────────────────────────────────── for p in ok_pages: corpus += f"\n\n[{p['url']}]\n{p['text'][:1500]}" # 4. Coverage check ───────────────────────────────────────────────────── coverage = _goal_coverage(req.topic, corpus) _logger.debug("ARL round %d: %d pages, coverage=%.2f", rounds, len(all_pages), coverage) if coverage >= _MIN_COVERAGE: break # 5. Refine query for next round ───────────────────────────────────────── refined = _refine_query(req.topic, corpus) if not refined or refined.lower() in tried_queries: # GF-7: _refine_query non produce nulla di nuovo → prova fallback fb_queries = _gen_fallback_queries(req.topic, tried_queries) if fb_queries: query = fb_queries[0] tried_queries.add(query.lower()) _logger.debug("GF-7: refine esaurito → fallback query: %r", query) else: break else: query = refined tried_queries.add(query.lower()) rounds += 1 # ── Response ─────────────────────────────────────────────────────────────── if not all_pages: return { "ok": False, "error": "Nessuna pagina leggibile trovata (tutte bloccate o vuote).", } synthesis = "" if req.synthesize: synthesis = await _synthesize(req.topic, all_pages) final_coverage = _goal_coverage(req.topic, corpus) elapsed_ms = round((time.monotonic() - _t0) * 1000) return { "ok": True, "topic": req.topic, "sources": [ {"url": p["url"], "title": p.get("title", ""), "excerpt": p["text"][:500]} for p in all_pages ], "synthesis": synthesis, "count": len(all_pages), # ARL metadata (for debugging / monitoring) "arl": { "rounds": rounds + 1, "rounds_to_converge": rounds + 1, "coverage": round(final_coverage, 3), "elapsed_ms": elapsed_ms, "sources_total": len(all_pages), }, } def _gen_alt_queries(goal: str) -> list[str]: """GAP-RESEARCH-PARALLEL: genera varianti lessicali per multi-angle search al round 0. Zero LLM — analisi sintattica pura. Produce query diverse per ampliare la coverage iniziale senza aspettare i round successivi (3× fonti in parallelo ≈ stessa latenza). Strategie: alt1 — top-3 keyword (sostituisce frasi lunghe con le parole chiave essenziali) alt2 — tail context (ultime 3 parole, cattura il complemento tematico) alt3 — traduzione EN (U1): per query italiane, aggiunge variante inglese per raggiungere documentazione tecnica (prevalentemente in inglese). """ words = [w for w in re.split(r'\W+', goal) if len(w) > 2 and w.lower() not in _STOP_WORDS] alts: list[str] = [] if len(words) >= 4: alt1 = ' '.join(words[:3]) if alt1.lower() != goal.lower(): alts.append(alt1) if len(words) >= 5: alt2 = ' '.join(words[-3:]) if alt2.lower() != goal.lower() and alt2 not in alts: alts.append(alt2) # U1: alt3 — traduzione EN se query italiana (zero LLM, +30% coverage documentazione tecnica) if _is_italian(goal): alt3 = _translate_it_en(goal) if alt3 and alt3.lower() not in {a.lower() for a in alts} and alt3.lower() != goal.lower(): alts.append(alt3) return alts[:3] def _gen_fallback_queries(goal: str, tried: set[str]) -> list[str]: """GF-7: genera varianti di query non ancora tentate quando new_urls/ok_pages è vuoto. Chiamata solo quando il loop si troverebbe ad uscire con 0 nuovi URL o 0 pagine leggibili — invece di arrendersi, prova angolazioni diverse: v1 — inversione ordine parole chiave (cerca complemento prima del soggetto) v2 — aggiunge "tutorial" / "guida" / "come" (disambigua intent informativo) v3 — singola keyword più specifica (narrow search su termine principale) Zero LLM. Restituisce solo varianti non già in `tried`. """ words = [w for w in re.split(r'\W+', goal) if len(w) > 2 and w.lower() not in _STOP_WORDS] candidates: list[str] = [] # v1 — inversione ultime/prime keyword if len(words) >= 3: v1 = " ".join(words[len(words)//2:] + words[:len(words)//2]) candidates.append(v1) # v2 — intent informativo esplicito (B-GAP-D: EN-aware prefix) kw_core = " ".join(words[:4]) _v2_prefix = "how to" if not _is_italian(goal) else "come funziona" v2 = f"{_v2_prefix} {kw_core}".strip() candidates.append(v2) # v3 — termine più specifico (seconda keyword, spesso più discriminante) if len(words) >= 2: v3 = words[1] if len(words[1]) > 4 else (words[0] if len(words[0]) > 4 else "") if v3: candidates.append(v3) # D8: v4 — inversione completa (B-GAP-D: porta D8 nella definizione effettiva) if len(words) >= 3: v4 = ' '.join(reversed(words)) if v4 not in set(candidates): candidates.append(v4) return [c for c in candidates if c and c.lower() not in tried][:3] # B-GAP-D: era :2, ora :3 per v4 # ─── Search: usa pipeline web_search.py (Brave → Tavily → Wikipedia → HN) ──── async def _pipeline_search(query: str, n: int) -> list[dict]: """Usa la pipeline condivisa web_search.py — stessa logica di _run_direct_tools.""" try: from tools.web_search import web_search as _ws result = await _ws(query, max_results=n) hits = result.get("results", []) if hits: return [{"url": h["url"], "title": h.get("title", "")} for h in hits if h.get("url")] except Exception as exc: _logger.debug("pipeline_search fallback: %s", exc) return [] async def _ddg_fallback_search(query: str, n: int) -> list[dict]: """Fallback DDG HTML parse quando nessuna chiave API è configurata.""" try: _ddg_kl = "it-it" if _is_italian(query) else "en-us" # B-GAP-D: locale EN-aware _ddg_al = "it-IT,it;q=0.9,en;q=0.8" if _ddg_kl == "it-it" else "en-US,en;q=0.9" async with httpx.AsyncClient(timeout=10, headers={"User-Agent": _UA, "Accept-Language": _ddg_al}) as c: r = await c.get("https://html.duckduckgo.com/html/", params={"q": query, "kl": _ddg_kl}) if r.status_code != 200: return [] html = r.text link_pattern = re.compile(r']+class="result__url"[^>]*href="([^"]+)"[^>]*>([^<]*)', re.DOTALL) title_pattern = re.compile(r']+class="result__a"[^>]*href="[^"]+"[^>]*>([^<]+)', re.DOTALL) links = link_pattern.findall(html) titles = [re.sub(r"\s+", " ", t).strip() for t in title_pattern.findall(html)] results = [] for i, (url, _) in enumerate(links[:n]): if url.startswith("http"): results.append({"url": url, "title": titles[i] if i < len(titles) else url}) return results[:n] except Exception: return [] # ─── Page content extraction ────────────────────────────────────────────────── async def _fetch_page(url: str, max_chars: int = 2000) -> dict: try: async with httpx.AsyncClient(timeout=10, follow_redirects=True, headers={"User-Agent": _UA}) as c: r = await c.get(url) if r.status_code != 200: return {"url": url, "ok": False, "error": f"HTTP {r.status_code}"} html = r.text try: import trafilatura text = trafilatura.extract( html, include_comments=False, include_tables=False, favor_recall=True, deduplicate=True, ) or "" except ImportError: text = re.sub(r"<[^>]+>", " ", html) text = re.sub(r"\s{2,}", " ", text).strip() noise = {"cookie","accept all cookies","privacy policy","terms of service","subscribe","follow us on"} lines = [l for l in text.split("\n") if len(l.strip()) > 4 and not any(n in l.lower() for n in noise)] text = "\n".join(lines) title_m = re.search(r"]*>([^<]+)", html, re.IGNORECASE) title = title_m.group(1).strip()[:120] if title_m else url return {"url": url, "title": title, "text": text[:max_chars], "ok": True} except Exception as e: return {"url": url, "ok": False, "error": str(e)[:100]} # ─── LLM synthesis (Groq) ───────────────────────────────────────────────────── async def _synthesize(topic: str, sources: list[dict]) -> str: groq_key = os.getenv("GROQ_API_KEY", "") if not groq_key: return "" context = "\n\n".join( f"[{i+1}] {s['title']}\n{s['text'][:800]}" for i, s in enumerate(sources) if s.get("ok") and s.get("text") )[:6000] try: async with httpx.AsyncClient(timeout=30) as c: r = await c.post( "https://api.groq.com/openai/v1/chat/completions", headers={"Authorization": f"Bearer {groq_key}", "Content-Type": "application/json"}, json={ "model": "llama-3.1-8b-instant", "max_tokens": 700, "messages": [ {"role": "system", "content": "Sei un assistente che sintetizza informazioni web. Rispondi sempre in italiano. Sii conciso e preciso."}, {"role": "user", "content": ( f"Argomento: **{topic}**\n\n" f"Fonti trovate:\n{context}\n\n" "Sintetizza le informazioni principali in 3-5 punti chiave, citando le fonti [N]." )}, ], }, ) if r.status_code == 200: _chs = r.json().get("choices") or [] return (_chs[0].get("message", {}).get("content") or "") if _chs else "" except Exception as exc: _logger.debug("synthesis error: %s", exc) return "" # ─── Main endpoint ──────────────────────────────────────────────────────────── @router.post("/research") async def web_research(req: ResearchRequest, request: Request): # S-GAP23: X-Internal-Token guard — protegge consumi Groq API da abusi esterni. _internal_token = os.getenv('INTERNAL_TOKEN', '') if _internal_token and request.headers.get('X-Internal-Token') != _internal_token: raise HTTPException(401, 'Unauthorized') n = min(max(int(req.depth), 1), _MAX_URLS_PER_ROUND) # ── ARL loop ─────────────────────────────────────────────────────────────── _t0 = time.monotonic() visited = set() corpus = "" query = req.topic all_pages: list[dict] = [] rounds = 0 # GF-7: traccia tutte le query tentate per evitare duplicati nei fallback tried_queries: set[str] = {query.lower()} while rounds < _MAX_ROUNDS and (time.monotonic() - _t0) < _TIMEOUT_S: # 1. Search ───────────────────────────────────────────────────────────── if rounds == 0: # GAP-RESEARCH-PARALLEL: round 0 usa 3 query in parallelo per massimizzare # la coverage iniziale senza penalità di latenza (asyncio.gather). alt_queries = _gen_alt_queries(query) if alt_queries: search_batches = await asyncio.gather( _pipeline_search(query, n), *[_pipeline_search(q, n) for q in alt_queries], ) # Dedup preservando ordine: priorità alla query principale _seen_u: set[str] = set() results: list[dict] = [] for batch in search_batches: for r in batch: if r["url"] not in _seen_u: _seen_u.add(r["url"]) results.append(r) _logger.debug( "ARL round 0 parallel: %d queries → %d unique URLs", 1 + len(alt_queries), len(results), ) # GF-7: registra le alt_queries come già tentate for aq in alt_queries: tried_queries.add(aq.lower()) else: results = await _pipeline_search(query, n) else: results = await _pipeline_search(query, n) if not results: results = await _ddg_fallback_search(query, n) if not results: # GF-7: search completamente vuota → prova query alternativa non ancora tentata fb_queries = _gen_fallback_queries(req.topic, tried_queries) if fb_queries: query = fb_queries[0] tried_queries.add(query.lower()) _logger.debug("GF-7: search vuota → fallback query: %r", query) rounds += 1 continue break # 2. Fetch new URLs only ──────────────────────────────────────────────── new_urls = [r["url"] for r in results if r["url"] not in visited][:n] if not new_urls: # GF-7: tutti gli URL già visitati → cambia query invece di arrendersi fb_queries = _gen_fallback_queries(req.topic, tried_queries) if fb_queries: query = fb_queries[0] tried_queries.add(query.lower()) _logger.debug("GF-7: new_urls vuoto → fallback query: %r", query) rounds += 1 continue break for u in new_urls: visited.add(u) pages = await asyncio.gather(*[_fetch_page(u) for u in new_urls]) ok_pages = [p for p in pages if p.get("ok") and p.get("text")] # GF-7: pagine fetch tutte fallite (bloccate/vuote) → cambia query if not ok_pages: fb_queries = _gen_fallback_queries(req.topic, tried_queries) if fb_queries: query = fb_queries[0] tried_queries.add(query.lower()) _logger.debug("GF-7: ok_pages vuoto → fallback query: %r", query) rounds += 1 continue break all_pages.extend(ok_pages) # 3. Build corpus ─────────────────────────────────────────────────────── for p in ok_pages: corpus += f"\n\n[{p['url']}]\n{p['text'][:1500]}" # 4. Coverage check ───────────────────────────────────────────────────── coverage = _goal_coverage(req.topic, corpus) _logger.debug("ARL round %d: %d pages, coverage=%.2f", rounds, len(all_pages), coverage) if coverage >= _MIN_COVERAGE: break # 5. Refine query for next round ───────────────────────────────────────── refined = _refine_query(req.topic, corpus) if not refined or refined.lower() in tried_queries: # GF-7: _refine_query non produce nulla di nuovo → prova fallback fb_queries = _gen_fallback_queries(req.topic, tried_queries) if fb_queries: query = fb_queries[0] tried_queries.add(query.lower()) _logger.debug("GF-7: refine esaurito → fallback query: %r", query) else: break else: query = refined tried_queries.add(query.lower()) rounds += 1 # ── Response ─────────────────────────────────────────────────────────────── if not all_pages: return { "ok": False, "error": "Nessuna pagina leggibile trovata (tutte bloccate o vuote).", } synthesis = "" if req.synthesize: synthesis = await _synthesize(req.topic, all_pages) final_coverage = _goal_coverage(req.topic, corpus) elapsed_ms = round((time.monotonic() - _t0) * 1000) return { "ok": True, "topic": req.topic, "sources": [ {"url": p["url"], "title": p.get("title", ""), "excerpt": p["text"][:500]} for p in all_pages ], "synthesis": synthesis, "count": len(all_pages), # ARL metadata (for debugging / monitoring) "arl": { "rounds": rounds + 1, "rounds_to_converge": rounds + 1, "coverage": round(final_coverage, 3), "elapsed_ms": elapsed_ms, "sources_total": len(all_pages), }, }