Terminal / api /research.py
Pulka
fix(research): reconstruct split elif re.match line — orphaned continuation removed
da73758 verified
Raw
History Blame
36.1 kB
"""
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 ────────────────────────────────────────────────────────────
@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'<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 ────────────────────────────────────────────────────────────
@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),
},
}