Upload brain.py with huggingface_hub

brain.py

@@ -0,0 +1,212 @@
+import asyncio
+import os
+from knowledge_base import KnowledgeBase
+from scraper import search_and_extract
+from errors import get_logger, GenerAIError, ErrorCode, fmt_exc
+
+log = get_logger("brain")
+
+LOCAL_THRESHOLD = 0.40
+
+# Modello HuggingFace da usare (fine-tunato o pubblico)
+# Imposta HF_MODEL=tuo_username/generai-model nel file .env o come variabile d'ambiente
+HF_MODEL = os.environ.get("HF_MODEL", "")
+HF_TOKEN = os.environ.get("HF_TOKEN", "")
+
+
+def _rerank(results: list, query: str) -> list:
+    words = set(query.lower().split())
+    for r in results:
+        stored_q = r["metadata"].get("query", "")
+        overlap = len(words & set(stored_q.lower().split()))
+        r["score"] = r["distance"] - overlap * 0.12
+    ranked = sorted(results, key=lambda x: x["score"])
+    if ranked:
+        log.debug("Rerank: best score=%.3f (dist=%.3f)", ranked[0]["score"], ranked[0]["distance"])
+    return ranked
+
+
+def _extract_sentences(text: str, query: str, max_chars: int = 700) -> str:
+    words = set(query.lower().split())
+    sentences = [s.strip() for s in text.replace("\n", ". ").split(".") if len(s.strip()) > 20]
+    scored = [(sum(1 for w in words if w in s.lower()), s) for s in sentences]
+    scored.sort(key=lambda x: -x[0])
+    result = ""
+    for _, s in scored:
+        if len(result) + len(s) + 2 > max_chars:
+            break
+        result += s + ". "
+    return result.strip() or text[:max_chars]
+
+
+# ── HuggingFace LLM client (opzionale) ────────────────────────────────────────
+
+class _HFClient:
+    """Wrapper leggero per HuggingFace Inference API."""
+
+    SYSTEM = (
+        "Sei GenerAI, un assistente AI specializzato in lingua italiana. "
+        "Rispondi in modo chiaro, preciso e sempre in italiano. "
+        "Se ti viene fornito del contesto, basati su quello per rispondere."
+    )
+
+    def __init__(self, model: str, token: str):
+        try:
+            from huggingface_hub import InferenceClient
+            self._client = InferenceClient(model=model, token=token or None)
+            self._model = model
+            log.info("HuggingFace LLM pronto: %s", model)
+        except ImportError:
+            raise GenerAIError(
+                ErrorCode.BRAIN_ASK_FAILED,
+                "huggingface_hub non installato. Esegui: pip install huggingface-hub",
+            )
+        except Exception as e:
+            raise GenerAIError(
+                ErrorCode.BRAIN_ASK_FAILED,
+                f"Impossibile connettersi a HuggingFace ({model}): {fmt_exc(e)}",
+                cause=e,
+            )
+
+    def generate(self, question: str, context: str = "") -> str:
+        user_msg = question
+        if context:
+            user_msg = f"Contesto:\n{context}\n\nDomanda: {question}"
+
+        messages = [
+            {"role": "system",  "content": self.SYSTEM},
+            {"role": "user",    "content": user_msg},
+        ]
+        log.debug("HF generate — modello=%s, context_len=%d", self._model, len(context))
+        try:
+            response = self._client.chat_completion(
+                messages=messages,
+                max_tokens=512,
+                temperature=0.3,
+            )
+            return response.choices[0].message.content.strip()
+        except Exception as e:
+            raise GenerAIError(
+                ErrorCode.BRAIN_ASK_FAILED,
+                f"Generazione HF fallita: {fmt_exc(e)}",
+                cause=e,
+            )
+
+
+# ── Brain ──────────────────────────────────────────────────────────────────────
+
+class Brain:
+    def __init__(self):
+        log.info("Avvio Brain...")
+        try:
+            self.kb = KnowledgeBase()
+        except GenerAIError as e:
+            e.log(log)
+            raise
+
+        self._last_doc_id: str | None = None
+
+        # Carica client HF se configurato
+        self._hf: _HFClient | None = None
+        if HF_MODEL:
+            try:
+                self._hf = _HFClient(HF_MODEL, HF_TOKEN)
+            except GenerAIError as e:
+                e.log(log)
+                log.warning("Fallback alla ricerca semantica (HF non disponibile).")
+        else:
+            log.info("HF_MODEL non impostato — uso ricerca semantica locale.")
+
+        log.info("Brain pronto. Modalità: %s", "LLM+KB" if self._hf else "KB+Web")
+
+    async def ask(self, question: str) -> tuple[str, str]:
+        """
+        Returns (answer, status).
+        status: "local" | "searched" | "archived" | "unknown" | "error" | "llm"
+        """
+        self._last_doc_id = None
+        log.info("Domanda: %r", question)
+
+        # ── 1. Ricerca KB locale ───────────────────────────────────────────────
+        try:
+            results = _rerank(self.kb.search(question, n_results=12), question)
+        except Exception as e:
+            log.warning("KB search fallita: %s", fmt_exc(e))
+            results = []
+
+        best_context = ""
+        best_source = ""
+
+        if results and results[0]["score"] < LOCAL_THRESHOLD:
+            best = results[0]
+            best_context = best["answer"]
+            best_source  = best["metadata"].get("source", "")
+            log.info("Match KB locale (score=%.3f)", results[0]["score"])
+
+            # Se c'è un LLM, usa il contesto KB per generare una risposta migliore
+            if self._hf:
+                return await self._llm_answer(question, best_context, best_source, status="llm")
+
+            answer = _extract_sentences(best_context, question)
+            if best_source and best_source != "grammatica_italiana":
+                answer += f"\n\n*Fonte: {best_source}*"
+            return answer, "local"
+
+        # ── 2. Ricerca web ─────────────────────────────────────────────────────
+        log.info("Nessun match locale — ricerca web...")
+        try:
+            web_results = await asyncio.to_thread(search_and_extract, question)
+        except Exception as e:
+            err = GenerAIError(ErrorCode.BRAIN_ASK_FAILED, f"Ricerca web fallita: {fmt_exc(e)}", cause=e)
+            err.log(log)
+            return err.user_message(), "error"
+
+        if not web_results:
+            log.warning("Nessun risultato web per: %r", question)
+            return (
+                "Non ho trovato informazioni su questo argomento nel mio database né sul web. "
+                "Prova a riformulare la domanda.",
+                "unknown",
+            )
+
+        combined = "\n\n---\n\n".join(f"[{r['title']}]\n{r['text']}" for r in web_results)
+        sources   = ", ".join(r["url"] for r in web_results if r.get("url"))
+
+        # ── 3. Genera risposta (LLM o estrazione testuale) ─────────────────────
+        if self._hf:
+            answer, status = await self._llm_answer(question, combined, sources, status="llm")
+        else:
+            answer = _extract_sentences(combined, question)
+            if sources:
+                answer += f"\n\n*Fonte: {sources}*"
+            status = "searched"
+
+        return answer, status
+
+    async def _llm_answer(
+        self, question: str, context: str, source: str, status: str
+    ) -> tuple[str, str]:
+        try:
+            answer = await asyncio.to_thread(self._hf.generate, question, context)
+            if source and source != "grammatica_italiana":
+                answer += f"\n\n*Fonte: {source}*"
+            log.info("Risposta generata da LLM (%d chars)", len(answer))
+            return answer, status
+        except GenerAIError as e:
+            e.log(log)
+            # Fallback all'estrazione testuale
+            log.warning("Fallback a estrazione testuale.")
+            answer = _extract_sentences(context, question)
+            if source and source != "grammatica_italiana":
+                answer += f"\n\n*Fonte: {source}*"
+            return answer, "searched"
+
+    def give_feedback(self, positive: bool):
+        if self._last_doc_id:
+            self.kb.reinforce(self._last_doc_id, positive)
+        else:
+            log.debug("give_feedback chiamato senza _last_doc_id.")
+
+    @property
+    def kb_size(self) -> int:
+        return self.kb.count()