Spaces:
Configuration error
Configuration error
Pulka
fix(research): reconstruct split elif re.match line — orphaned continuation removed
da73758 verified | """ | |
| 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'<a[^>]+class="result__url"[^>]*href="([^"]+)"[^>]*>([^<]*)</a>', re.DOTALL) | |
| title_pattern = re.compile(r'<a[^>]+class="result__a"[^>]*href="[^"]+"[^>]*>([^<]+)</a>', 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"<title[^>]*>([^<]+)</title>", 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 ──────────────────────────────────────────────────────────── | |
| 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'<a[^>]+class="result__url"[^>]*href="([^"]+)"[^>]*>([^<]*)</a>', re.DOTALL) | |
| title_pattern = re.compile(r'<a[^>]+class="result__a"[^>]*href="[^"]+"[^>]*>([^<]+)</a>', 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"<title[^>]*>([^<]+)</title>", 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 ──────────────────────────────────────────────────────────── | |
| 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), | |
| }, | |
| } | |