diff --git a/.env.example b/.env.example index fc46cef..92d991e 100644 --- a/.env.example +++ b/.env.example @@ -1,7 +1,20 @@ # Google AI API Key — ottenila da https://aistudio.google.com/apikey GOOGLE_API_KEY=your-api-key-here -# Flask +# ── Search backends (cascading: Degoog → SearXNG → DuckDuckGo) ── + +# Degoog (primary) — self-hosted metasearch, default port 4444 +# Avvia con: docker run -d --name degoog -p 4444:4444 ghcr.io/degoog-org/degoog:latest +DEGOOG_URL=http://localhost:4444 + +# SearXNG (fallback 1) — self-hosted metasearch, default port 8080 +# Avvia con: docker run -d -p 8080:8080 searxng/searxng +SEARXNG_URL=http://localhost:8080 + +# DuckDuckGo (fallback 2) — nessun server richiesto +SIBILLA_DDG_FALLBACK=1 + +# ── Flask ── SECRET_KEY=change-me-to-a-random-string SIBILLA_HOST=127.0.0.1 SIBILLA_PORT=5000 diff --git a/README.md b/README.md index 1d2e08b..9a1fabf 100644 --- a/README.md +++ b/README.md @@ -5,33 +5,46 @@ **Sibilla** è un motore di ricerca AI self-hosted che combina: -- 🧠 **Google Gemini API** — modelli linguistici avanzati con grounding su web search -- 🔧 **MCP (Model Context Protocol)** — tool use per estendere le capacità dell'AI -- 🔍 **Web search integrata** — ricerca e sintesi delle fonti in tempo reale +- 🧠 **Google Gemini** — LLM per sintesi e query expansion +- 🔍 **Degoog** — metasearch engine self-hosted (primary) con MCP nativo +- 🔄 **SearXNG + DuckDuckGo** — fallback a cascata se Degoog non risponde +- 🔧 **MCP (Model Context Protocol)** — server stdio per agent AI esterni - 🖥️ **Flask web app** — interfaccia pulita e reattiva ``` -Tu chiedi → Sibilla cerca nel web → Sintetizza le fonti → Ti risponde con citazioni +Tu chiedi → Query expansion → Degoog search → Fetch pagine → Gemini sintesi → Risposta con fonti ``` -## Architettura (pianificata) +## Architettura ``` sibilla/ ├── src/sibilla/ -│ ├── app.py # Flask app factory -│ ├── cli.py # Entry point -│ ├── search.py # Web search + scraping -│ ├── ai.py # Google Gemini API client -│ ├── mcp.py # MCP tool integration -│ └── prompts.py # Prompt templates +│ ├── app.py # Flask app factory + pipeline /api/ask +│ ├── cli.py # Entry point server web +│ ├── search.py # DegoogClient, SearXNGClient, DuckDuckGoClient, PageFetcher +│ ├── ai.py # GeminiClient (sintesi + query rewriting) +│ ├── mcp.py # MCP server su stdio (per agent AI esterni) +│ └── prompts.py # System prompt stile Perplexity ├── templates/ │ └── index.html # Interfaccia di ricerca ├── static/ -│ ├── style.css # Stili +│ ├── style.css # Stili dark theme │ └── app.js # Logica frontend └── tests/ - └── ... +``` + +## Pipeline di ricerca + +``` +POST /api/ask {query: "..."} + │ + ├─ 1. Gemini riscrive la query in 3 varianti (query expansion) + ├─ 2. Multi-query search: Degoog → SearXNG → DuckDuckGo (cascading) + ├─ 3. PageFetcher scarica il contenuto completo delle top pagine + ├─ 4. Prompt assembly con fonti numerate + ├─ 5. Gemini sintesi con citazioni [1], [2], ... + └─ 6. JSON {answer, sources[], model, search_engine, tokens_used} ``` ## Setup @@ -43,30 +56,44 @@ source .venv/bin/activate pip install -e ".[dev]" ``` -Crea un file `.env` con la tua API key Google: +### Prerequisiti +1. **Google API Key** — [https://aistudio.google.com/apikey](https://aistudio.google.com/apikey) +2. **Degoog** (consigliato) — motore di ricerca primary: + ```bash + docker run -d --name degoog -p 4444:4444 ghcr.io/degoog-org/degoog:latest + ``` +3. **SearXNG** (opzionale, fallback) — se vuoi un secondo motore di backup: + ```bash + docker run -d -p 8080:8080 searxng/searxng + ``` + +Crea un file `.env`: + +```env +GOOGLE_API_KEY=la-tua-api-key +DEGOOG_URL=http://localhost:4444 ``` -GOOGLE_API_KEY=your-api-key-here -``` + +Senza Degoog, Sibilla userà automaticamente SearXNG e DuckDuckGo come fallback. ## Uso ```bash -# Avvia il server +# Avvia la web app sibilla +# → http://127.0.0.1:5000 -# Oppure -python -m sibilla +# Avvia il server MCP (per agent AI esterni) +sibilla-mcp ``` -Poi apri http://127.0.0.1:5000 nel browser. - ## Sviluppo ```bash -pip install -e ".[dev]" -pytest +pip install -e ".[dev,mcp]" ruff check src tests +pytest ``` ## Licenza diff --git a/pyproject.toml b/pyproject.toml index 5f9b079..ff9988a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,12 +1,12 @@ [project] name = "sibilla" version = "0.1.0" -description = "sibilla - Self-hosted Perplexity-like AI search engine with Google Gemini + MCP tool use, built with Flask" +description = "sibilla - Self-hosted Perplexity-like AI search engine with Google Gemini + MCP tool use, powered by Degoog/SearXNG, built with Flask" readme = "README.md" requires-python = ">=3.10" license = { text = "MIT" } authors = [{ name = "Author" }] -keywords = ["ai", "search", "gemini", "perplexity", "mcp", "llm", "rag", "flask"] +keywords = ["ai", "search", "gemini", "perplexity", "mcp", "llm", "rag", "flask", "degoog", "searxng"] classifiers = [ "Programming Language :: Python :: 3", "Programming Language :: Python :: 3 :: Only", @@ -30,9 +30,13 @@ dev = [ "pytest>=7", "ruff>=0.5", ] +mcp = [ + "mcp>=1.0", +] [project.scripts] sibilla = "sibilla.cli:main" +sibilla-mcp = "sibilla.mcp:main" [build-system] requires = ["setuptools>=61.0", "wheel"] diff --git a/src/sibilla/ai.py b/src/sibilla/ai.py new file mode 100644 index 0000000..dd921ff --- /dev/null +++ b/src/sibilla/ai.py @@ -0,0 +1,169 @@ +"""sibilla AI — Google Gemini LLM client (without native search grounding). + +Uses Gemini purely as an LLM for synthesis. Search results are fed +into the prompt context manually — giving us full control over sources. +""" + +import os +from dataclasses import dataclass +from typing import Optional + +from google import genai +from google.genai import types + + +@dataclass +class SynthesisResult: + """Result of an AI synthesis with structured citations.""" + + answer: str + model: str = "" + tokens_used: int = 0 + error: Optional[str] = None + + +class GeminiClient: + """Google Gemini LLM client for Perplexity-style synthesis. + + Args: + api_key: Google AI API key. If None, reads from GOOGLE_API_KEY env var. + model: Model name (default: gemini-2.5-flash). + """ + + def __init__( + self, + api_key: Optional[str] = None, + model: str = "gemini-2.5-flash", + ) -> None: + self.api_key = api_key or os.environ.get("GOOGLE_API_KEY", "") + self.model = model + self._client: Optional[genai.Client] = None + + @property + def client(self) -> genai.Client: + """Lazy-initialize the Gemini client.""" + if self._client is None: + if not self.api_key: + raise ValueError( + "GOOGLE_API_KEY is not set. " + "Export it or pass api_key to GeminiClient()." + ) + self._client = genai.Client(api_key=self.api_key) + return self._client + + def synthesize( + self, + system_prompt: str, + user_prompt: str, + temperature: float = 0.4, + max_tokens: int = 4096, + ) -> SynthesisResult: + """Send a prompt to Gemini and get a synthesized response. + + Args: + system_prompt: System instructions for the model. + user_prompt: The user query with source context. + temperature: Sampling temperature (0.0–1.0). + max_tokens: Maximum tokens in the response. + + Returns: + SynthesisResult with the model's answer. + """ + if not self.api_key: + return SynthesisResult( + answer="", + error=( + "GOOGLE_API_KEY non configurata. " + "Crea un file .env con GOOGLE_API_KEY=your-key" + ), + ) + + try: + response = self.client.models.generate_content( + model=self.model, + contents=[user_prompt], + config=types.GenerateContentConfig( + system_instruction=system_prompt, + temperature=temperature, + max_output_tokens=max_tokens, + ), + ) + + text = "" + if response.candidates: + candidate = response.candidates[0] + if candidate.content and candidate.content.parts: + text = "".join( + part.text + for part in candidate.content.parts + if part.text + ) + + usage = response.usage_metadata + tokens = ( + usage.total_token_count + if usage + else 0 + ) + + return SynthesisResult( + answer=text.strip(), + model=self.model, + tokens_used=tokens, + ) + except Exception as e: + return SynthesisResult( + answer="", + model=self.model, + error=str(e), + ) + + def rewrite_query( + self, original_query: str, num_variants: int = 3 + ) -> list[str]: + """Generate alternative search queries to improve recall. + + This implements a simplified version of Perplexity's query + rewriting / expansion step. + + Args: + original_query: The user's original query. + num_variants: Number of alternative queries to generate. + + Returns: + List of query strings (original + variants). + """ + if not self.api_key: + return [original_query] + + prompt = f"""Riscrivi la seguente domanda in {num_variants} varianti +per migliorare la ricerca web. Ogni variante deve essere su una riga. +Non aggiungere altro testo, solo le query, una per riga. + +Domanda originale: {original_query}""" + + try: + response = self.client.models.generate_content( + model=self.model, + contents=[prompt], + config=types.GenerateContentConfig( + temperature=0.3, + max_output_tokens=300, + ), + ) + text = "" + if response.candidates: + candidate = response.candidates[0] + if candidate.content and candidate.content.parts: + text = "".join( + part.text for part in candidate.content.parts if part.text + ) + + variants = [ + line.strip("-• *").strip() + for line in text.strip().splitlines() + if line.strip() + ] + return [original_query] + variants[:num_variants] + except Exception: + return [original_query] diff --git a/src/sibilla/app.py b/src/sibilla/app.py index c912c68..b14d7b4 100644 --- a/src/sibilla/app.py +++ b/src/sibilla/app.py @@ -1,7 +1,21 @@ -"""sibilla web application — Flask app factory.""" +"""sibilla web application — Flask app factory with Perplexity-style pipeline.""" import os -from flask import Flask, render_template +from datetime import datetime +from typing import Optional + +from flask import Flask, render_template, request, jsonify + +from sibilla.search import DegoogClient, SearXNGClient, DuckDuckGoClient, PageFetcher +from sibilla.ai import GeminiClient +from sibilla.prompts import ( + SYNTHESIS_SYSTEM_PROMPT, + FALLBACK_PROMPT, + build_synthesis_prompt, +) + +# Project root: go up from src/sibilla/app.py → src/sibilla → src → project root +_PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(__file__))) def create_app(testing: bool = False) -> Flask: @@ -10,14 +24,65 @@ def create_app(testing: bool = False) -> Flask: Args: testing: If True, configure the app for testing (no .env loading). """ - app = Flask(__name__) + app = Flask( + __name__, + template_folder=os.path.join(_PROJECT_ROOT, "templates"), + static_folder=os.path.join(_PROJECT_ROOT, "static"), + ) + + if not testing: + from dotenv import load_dotenv + + load_dotenv() app.config.update( SECRET_KEY=os.environ.get("SECRET_KEY", "dev-secret-key-change-me"), GOOGLE_API_KEY=os.environ.get("GOOGLE_API_KEY", ""), + DEGOOG_URL=os.environ.get("DEGOOG_URL", "http://localhost:4444"), + SEARXNG_URL=os.environ.get("SEARXNG_URL", "http://localhost:8080"), + SIBILLA_DDG_FALLBACK=os.environ.get("SIBILLA_DDG_FALLBACK", "1") == "1", ) - # ── Routes ────────────────────────────────────────────────────── + # ── Lazy-init services ──────────────────────────────────────── + + _degoog: Optional[DegoogClient] = None + _searxng: Optional[SearXNGClient] = None + _ddg: Optional[DuckDuckGoClient] = None + _fetcher: Optional[PageFetcher] = None + _gemini: Optional[GeminiClient] = None + + def get_degoog() -> DegoogClient: + nonlocal _degoog + if _degoog is None: + _degoog = DegoogClient(base_url=app.config["DEGOOG_URL"]) + return _degoog + + def get_searxng() -> SearXNGClient: + nonlocal _searxng + if _searxng is None: + _searxng = SearXNGClient(base_url=app.config["SEARXNG_URL"]) + return _searxng + + def get_ddg() -> DuckDuckGoClient: + nonlocal _ddg + if _ddg is None: + _ddg = DuckDuckGoClient() + return _ddg + + def get_fetcher() -> PageFetcher: + nonlocal _fetcher + if _fetcher is None: + _fetcher = PageFetcher() + return _fetcher + + def get_gemini() -> GeminiClient: + nonlocal _gemini + if _gemini is None: + _gemini = GeminiClient(api_key=app.config["GOOGLE_API_KEY"]) + return _gemini + + # ── Routes ──────────────────────────────────────────────────── + @app.route("/") def index(): """Home page — the search interface.""" @@ -26,6 +91,157 @@ def create_app(testing: bool = False) -> Flask: @app.route("/api/health") def health(): """Health check endpoint.""" - return {"status": "ok", "version": __import__("sibilla").__version__} + gemini_ok = bool(app.config["GOOGLE_API_KEY"]) + return { + "status": "ok", + "version": __import__("sibilla").__version__, + "gemini_configured": gemini_ok, + "degoog_url": app.config["DEGOOG_URL"], + "searxng_url": app.config["SEARXNG_URL"], + } + + @app.route("/api/ask", methods=["POST"]) + def ask(): + """Perplexity-style search + synthesis endpoint. + + Pipeline: + 1. Receive user query + 2. Query expansion (generate alternative queries) + 3. Multi-query web search via SearXNG (or DDG fallback) + 4. Fetch full content from top results + 5. Build synthesis prompt with sources + 6. Gemini LLM synthesis with citations + 7. Return JSON {answer, sources, metadata} + """ + data = request.get_json(silent=True) or {} + query = data.get("query", "").strip() + num_sources = min(data.get("num_sources", 5), 10) + + if not query: + return jsonify({"error": "La query non può essere vuota."}), 400 + + gemini = get_gemini() + degoog = get_degoog() + searxng = get_searxng() + ddg = get_ddg() + fetcher = get_fetcher() + + # ── Step 1: Query expansion (improves search recall) ────── + expanded_queries = gemini.rewrite_query(query, num_variants=3) + + # ── Step 2: Multi-query search (Degoog → SearXNG → DDG) ─── + all_results: list = [] + seen_urls: set[str] = set() + search_error: Optional[str] = None + primary_engine = "degoog" + + for eq in expanded_queries[:2]: # original + 1 variant + resp = degoog.search(eq, num_results=num_sources) + if resp.results: + for r in resp.results: + if r.url not in seen_urls: + seen_urls.add(r.url) + all_results.append(r) + if resp.error: + search_error = resp.error + + # Fallback to SearXNG if Degoog returned nothing + if not all_results: + for eq in expanded_queries[:2]: + resp = searxng.search(eq, num_results=num_sources) + if resp.results: + primary_engine = "searxng" + for r in resp.results: + if r.url not in seen_urls: + seen_urls.add(r.url) + all_results.append(r) + if resp.error and not search_error: + search_error = resp.error + + # Fallback to DuckDuckGo if both Degoog and SearXNG failed + if not all_results and app.config["SIBILLA_DDG_FALLBACK"]: + ddg_resp = ddg.search(query, num_results=num_sources) + if ddg_resp.results: + primary_engine = "duckduckgo" + for r in ddg_resp.results: + if r.url not in seen_urls: + seen_urls.add(r.url) + all_results.append(r) + if ddg_resp.error and not search_error: + search_error = ddg_resp.error + + # ── Handle no results ───────────────────────────────────── + if not all_results: + if search_error: + return jsonify({ + "error": f"Ricerca web fallita: {search_error}", + "answer": "", + "sources": [], + }), 502 + + # No error but no results — fallback to pure LLM + result = gemini.synthesize( + system_prompt=FALLBACK_PROMPT, + user_prompt=f"Query: {query}", + temperature=0.5, + ) + return jsonify({ + "answer": result.answer if not result.error else ( + f"Nessun risultato trovato per: {query}" + ), + "sources": [], + "model": gemini.model, + "search_engine": "none", + "tokens_used": result.tokens_used, + "error": result.error if result.error else None, + }) + + # ── Step 3: Fetch page content ──────────────────────────── + enriched = fetcher.fetch_results(all_results, max_results=num_sources) + + # ── Step 4: Build synthesis prompt ──────────────────────── + today = datetime.now().strftime("%d %B %Y") + sources_for_prompt = [ + { + "index": i, + "title": r.title, + "url": r.url, + "content": r.full_content or r.snippet, + } + for i, r in enumerate(enriched[:num_sources], 1) + ] + prompt = build_synthesis_prompt( + query, sources_for_prompt, f"Data di oggi: {today}" + ) + + # ── Step 5: AI Synthesis ────────────────────────────────── + synthesis = gemini.synthesize( + system_prompt=SYNTHESIS_SYSTEM_PROMPT, + user_prompt=prompt, + temperature=0.4, + max_tokens=4096, + ) + + # ── Step 6: Build response ──────────────────────────────── + sources_out = [ + { + "index": i, + "title": r.title, + "url": r.url, + "snippet": r.snippet[:200] if r.snippet else "", + } + for i, r in enumerate(enriched[:num_sources], 1) + ] + + return jsonify({ + "answer": synthesis.answer if not synthesis.error else ( + f"⚠️ Errore nella sintesi AI: {synthesis.error}" + ), + "sources": sources_out, + "model": gemini.model, + "search_engine": primary_engine, + "tokens_used": synthesis.tokens_used, + "error": synthesis.error if synthesis.error else None, + }) return app diff --git a/src/sibilla/cli.py b/src/sibilla/cli.py index d4fb82f..df9c690 100644 --- a/src/sibilla/cli.py +++ b/src/sibilla/cli.py @@ -2,11 +2,15 @@ import os +from dotenv import load_dotenv + from sibilla.app import create_app def main() -> None: """Run the sibilla web server.""" + load_dotenv() + host = os.environ.get("SIBILLA_HOST", "127.0.0.1") port = int(os.environ.get("SIBILLA_PORT", "5000")) debug = os.environ.get("SIBILLA_DEBUG", "1") == "1" diff --git a/src/sibilla/mcp.py b/src/sibilla/mcp.py new file mode 100644 index 0000000..3879aeb --- /dev/null +++ b/src/sibilla/mcp.py @@ -0,0 +1,240 @@ +"""sibilla MCP server — Model Context Protocol tools for web search. + +Exposes search and fetch tools via MCP stdio transport so that +Gemini (or any MCP-compatible client) can use them as tools. + +Architecture: + AI Agent ←→ MCP Client ←─ stdio ─→ MCP Server (this module) + ├── web_search(query) → Degoog → SearXNG → DDG + └── fetch_page(url) → PageFetcher + +Usage (standalone): + python -m sibilla.mcp + +The server reads/writes JSON-RPC on stdin/stdout following the MCP protocol. +""" + +from __future__ import annotations + +import json +import sys +from typing import Any + +from sibilla.search import DegoogClient, SearXNGClient, PageFetcher, DuckDuckGoClient + + +# ── Tool registry ────────────────────────────────────────────────── + +class MCPServer: + """Minimal MCP-compatible server using stdio transport. + + Implements the Model Context Protocol (MCP) specification + for tool discovery and invocation over stdin/stdout JSON-RPC. + + Supported methods: + - initialize + - tools/list + - tools/call + """ + + def __init__( + self, + degoog_url: str = "http://localhost:4444", + searxng_url: str = "http://localhost:8080", + fallback_to_ddg: bool = True, + ) -> None: + self._degoog = DegoogClient(base_url=degoog_url) + self._searxng = SearXNGClient(base_url=searxng_url) + self._ddg = DuckDuckGoClient() if fallback_to_ddg else None + self._fetcher = PageFetcher() + self._tools = self._build_tools() + self._initialized = False + + def _build_tools(self) -> list[dict[str, Any]]: + """Define the available tools following MCP schema.""" + return [ + { + "name": "web_search", + "description": ( + "Cerca nel web usando un metasearch engine " + "(Degoog, con fallback a SearXNG e DuckDuckGo). " + "Restituisce titolo, URL e snippet per ogni risultato. " + "Usa questo tool per trovare informazioni aggiornate su " + "qualsiasi argomento." + ), + "inputSchema": { + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "La query di ricerca.", + }, + "num_results": { + "type": "integer", + "description": "Numero massimo di risultati (default: 8).", + "default": 8, + }, + }, + "required": ["query"], + }, + }, + { + "name": "fetch_page", + "description": ( + "Scarica ed estrae il contenuto testuale di una pagina web. " + "Restituisce il testo pulito (max 8000 caratteri). " + "Usalo per leggere il contenuto completo di un risultato di ricerca." + ), + "inputSchema": { + "type": "object", + "properties": { + "url": { + "type": "string", + "description": "L'URL della pagina da scaricare.", + }, + }, + "required": ["url"], + }, + }, + ] + + # ── Tool implementations ─────────────────────────────────── + + def _web_search(self, query: str, num_results: int = 8) -> str: + """Execute a web search with cascading fallbacks: Degoog → SearXNG → DDG.""" + response = self._degoog.search(query, num_results=num_results) + + # Fallback to SearXNG + if not response.results: + response = self._searxng.search(query, num_results=num_results) + + # Fallback to DuckDuckGo + if not response.results and self._ddg: + response = self._ddg.search(query, num_results=num_results) + + if response.error and not response.results: + return f"Errore nella ricerca: {response.error}" + + if not response.results: + return "Nessun risultato trovato." + + lines = [f"Risultati per: {query} (via {response.engine})\n"] + for i, r in enumerate(response.results, 1): + lines.append(f"[{i}] {r.title}") + lines.append(f" URL: {r.url}") + lines.append(f" {r.snippet}\n") + + return "\n".join(lines) + + def _fetch_page(self, url: str) -> str: + """Fetch and extract text from a web page.""" + content = self._fetcher.fetch(url) + if not content: + return f"Impossibile scaricare il contenuto da: {url}" + return content + + # ── JSON-RPC handling ────────────────────────────────────── + + def _handle_request(self, request: dict) -> dict | None: + """Process a single JSON-RPC request. Returns response or None for notifications.""" + method = request.get("method", "") + req_id = request.get("id") + params = request.get("params", {}) + + if method == "initialize": + self._initialized = True + return self._response(req_id, { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": { + "name": "sibilla-mcp", + "version": "0.1.0", + }, + }) + + if method == "notifications/initialized": + return None # No response for notifications + + if method == "tools/list": + return self._response(req_id, {"tools": self._tools}) + + if method == "tools/call": + tool_name = params.get("name", "") + arguments = params.get("arguments", {}) + + if tool_name == "web_search": + result = self._web_search( + query=arguments.get("query", ""), + num_results=arguments.get("num_results", 8), + ) + return self._response(req_id, { + "content": [{"type": "text", "text": result}], + }) + + if tool_name == "fetch_page": + result = self._fetch_page( + url=arguments.get("url", ""), + ) + return self._response(req_id, { + "content": [{"type": "text", "text": result}], + }) + + return self._error(req_id, -32601, f"Tool '{tool_name}' not found") + + # Default: method not found + return self._error(req_id, -32601, f"Method '{method}' not found") + + @staticmethod + def _response(req_id: Any, result: Any) -> dict: + return {"jsonrpc": "2.0", "id": req_id, "result": result} + + @staticmethod + def _error(req_id: Any, code: int, message: str) -> dict: + return { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": code, "message": message}, + } + + # ── Main loop ────────────────────────────────────────────── + + def run(self) -> None: + """Run the MCP server on stdin/stdout.""" + for line in sys.stdin: + line = line.strip() + if not line: + continue + + try: + request = json.loads(line) + response = self._handle_request(request) + if response is not None: + sys.stdout.write(json.dumps(response) + "\n") + sys.stdout.flush() + except json.JSONDecodeError: + # MCP protocol: send parse error + err = self._error(None, -32700, "Parse error") + sys.stdout.write(json.dumps(err) + "\n") + sys.stdout.flush() + + +# ── CLI ──────────────────────────────────────────────────────────── + +def main() -> None: + """Entry point for running the MCP server standalone.""" + import os + + degoog_url = os.environ.get("DEGOOG_URL", "http://localhost:4444") + searxng_url = os.environ.get("SEARXNG_URL", "http://localhost:8080") + fallback = os.environ.get("SIBILLA_DDG_FALLBACK", "1") == "1" + + server = MCPServer( + degoog_url=degoog_url, + searxng_url=searxng_url, + fallback_to_ddg=fallback, + ) + server.run() + + +if __name__ == "__main__": + main() diff --git a/src/sibilla/prompts.py b/src/sibilla/prompts.py new file mode 100644 index 0000000..05b8dd7 --- /dev/null +++ b/src/sibilla/prompts.py @@ -0,0 +1,117 @@ +"""sibilla prompt templates — Perplexity-style system prompts. + +These prompts guide the Gemini model to act like Perplexity: +- Answer directly based on provided sources +- Include inline citations like [1], [2], etc. +- Structure answers clearly with headers, lists, and paragraphs +""" + +# ═══════════════════════════════════════════════════════════════════ +# Main synthesis system prompt +# ═══════════════════════════════════════════════════════════════════ + +SYNTHESIS_SYSTEM_PROMPT = """Sei Sibilla, un oracolo AI che risponde alle domande cercando informazioni +nel web e sintetizzandole in risposte chiare, accurate e ben strutturate. + +## Il tuo comportamento + +1. **Rispondi direttamente alla domanda** usando SOLO le fonti fornite. +2. **Cita le fonti** con il formato [1], [2], ecc. nel corpo della risposta. +3. **Se le fonti non contengono la risposta**, dillo onestamente. +4. **Struttura la risposta** con paragrafi chiari, liste puntate e titoli + (Markdown) quando utile. +5. **Sii concisa ma completa** — non aggiungere informazioni che non sono + nelle fonti. +6. **Rispondi in italiano**, a meno che la domanda non sia in un'altra lingua. +7. **NON inventare URL o riferimenti** — cita solo le fonti effettivamente + fornite qui sotto. +8. Se la domanda riguarda eventi recenti e le fonti non sono aggiornate, + segnalalo. + +## Formato della risposta + +Usa il formato Markdown. Alla fine della risposta, includi SEMPRE una sezione +"## Fonti" con l'elenco numerato delle fonti usate, completo di URL. + +Esempio: +``` +## Fonti +[1] Titolo della fonte — https://example.com/articolo +[2] Altro titolo — https://example.com/altro +``` +""" + + +def build_synthesis_prompt( + query: str, + sources: list[dict], + date_context: str = "", +) -> str: + """Build the full synthesis prompt with query + source context. + + Args: + query: The user's question. + sources: List of source dicts with 'index', 'title', 'url', 'content'. + date_context: Optional date context (e.g. "Data di oggi: 16 Luglio 2026"). + + Returns: + Complete prompt string ready to send to the LLM. + """ + parts = [] + + if date_context: + parts.append(f"### Contesto temporale\n{date_context}\n") + + parts.append(f"### Domanda dell'utente\n{query}\n") + + parts.append("### Fonti da utilizzare\n") + for src in sources: + idx = src.get("index", "?") + title = src.get("title", "Senza titolo") + url = src.get("url", "") + content = src.get("content", src.get("snippet", "")) + parts.append( + f"--- Fonte [{idx}] ---\n" + f"Titolo: {title}\n" + f"URL: {url}\n" + f"Contenuto:\n{content}\n" + ) + + parts.append( + "### Istruzioni\n" + "Sintetizza una risposta completa alla domanda usando ESCLUSIVAMENTE " + "le fonti sopra. Cita le fonti con [1], [2], ecc. Includi la sezione " + "## Fonti alla fine con titolo e URL di ogni fonte usata." + ) + + return "\n".join(parts) + + +# ═══════════════════════════════════════════════════════════════════ +# Query rewriting prompt +# ═══════════════════════════════════════════════════════════════════ + +QUERY_REWRITE_PROMPT = """Sei un assistente specializzato nel migliorare le query di ricerca web. +Riscrivi la domanda dell'utente in {num_variants} varianti ottimizzate per +motori di ricerca, ognuna su una riga. + +Regole: +- Ogni variante deve essere una query di ricerca efficace (poche parole chiave) +- Varia il phrasing e la prospettiva +- Includi parole chiave rilevanti in italiano e inglese se utile +- NON aggiungere numeri, proiettili o altro testo +- SOLO le query, una per riga""" + + +# ═══════════════════════════════════════════════════════════════════ +# Fallback prompt (when no search results) +# ═══════════════════════════════════════════════════════════════════ + +FALLBACK_PROMPT = """Sei Sibilla, un assistente AI. L'utente ha fatto una domanda ma la ricerca +web non ha prodotto risultati utili. + +Rispondi onestamente: spiega che la ricerca web non ha trovato risultati +rilevanti e offri suggerimenti su come l'utente potrebbe riformulare la +domanda. + +NON inventare informazioni. Sii utile e trasparente.""" diff --git a/src/sibilla/search.py b/src/sibilla/search.py new file mode 100644 index 0000000..96e1500 --- /dev/null +++ b/src/sibilla/search.py @@ -0,0 +1,414 @@ +"""sibilla web search — SearXNG integration and page scraping.""" + +import re +import time +from dataclasses import dataclass, field +from typing import Optional +from urllib.parse import parse_qs, unquote, urlparse + +import requests +from bs4 import BeautifulSoup + + +# ── Module-level helpers ─────────────────────────────────────────── + +def _clean_html(text: str) -> str: + """Strip HTML tags from text.""" + return re.sub(r"<[^>]+>", "", text).strip() + + +@dataclass +class SearchResult: + """A single search result with metadata and optional full content.""" + + title: str + url: str + snippet: str = "" + full_content: str = "" + fetched: bool = False + + +@dataclass +class SearchResponse: + """Complete search response.""" + + query: str + results: list[SearchResult] = field(default_factory=list) + total_results: int = 0 + engine: str = "searxng" + error: Optional[str] = None + + +class DegoogClient: + """Client for the Degoog metasearch engine. + + Degoog is a modular, self-hosted search aggregator — a lighter, + more hackable alternative to SearXNG. It exposes a clean JSON API + at GET /api/search. + + Args: + base_url: URL of the Degoog instance (default: http://localhost:4444). + timeout: Request timeout in seconds. + """ + + def __init__( + self, + base_url: str = "http://localhost:4444", + timeout: int = 15, + ) -> None: + self.base_url = base_url.rstrip("/") + self.timeout = timeout + self._session = requests.Session() + self._session.headers.update( + { + "User-Agent": ( + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" + ), + "Accept": "application/json", + } + ) + + def search( + self, + query: str, + num_results: int = 8, + language: str = "it", + ) -> SearchResponse: + """Perform a web search via Degoog. + + Args: + query: The search query. + num_results: Maximum number of results to return. + language: Language code for results. + + Returns: + SearchResponse with results and metadata. + """ + params: dict = { + "q": query, + "lang": language, + "type": "web", + } + + try: + resp = self._session.get( + f"{self.base_url}/api/search", + params=params, + timeout=self.timeout, + ) + resp.raise_for_status() + data = resp.json() + return self._parse_response(data, query, num_results) + except requests.RequestException as e: + return SearchResponse( + query=query, error=str(e), engine="degoog" + ) + + def _parse_response( + self, data: dict, query: str, num_results: int + ) -> SearchResponse: + """Parse Degoog JSON response into SearchResponse.""" + results: list[SearchResult] = [] + raw_results = data.get("results", [])[:num_results] + + for item in raw_results: + results.append( + SearchResult( + title=_clean_html(item.get("title", "")), + url=item.get("url", ""), + # Degoog uses 'snippet'; also check 'content' + # for OpenWebUI compatibility + snippet=_clean_html( + item.get("content") or item.get("snippet", "") + ), + ) + ) + + return SearchResponse( + query=query, + results=results, + total_results=len(raw_results), + engine="degoog", + ) + + +class SearXNGClient: + """Client for SearXNG metasearch engine. + + SearXNG is a privacy-respecting metasearch engine that aggregates + results from multiple search engines (Google, Bing, DuckDuckGo, etc.) + and can be self-hosted. + + Args: + base_url: URL of the SearXNG instance (default: http://localhost:8080). + timeout: Request timeout in seconds. + """ + + def __init__( + self, + base_url: str = "http://localhost:8080", + timeout: int = 15, + ) -> None: + self.base_url = base_url.rstrip("/") + self.timeout = timeout + self._session = requests.Session() + self._session.headers.update( + { + "User-Agent": ( + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" + ), + "Accept": "application/json, text/html", + } + ) + + def search( + self, + query: str, + num_results: int = 8, + engines: Optional[str] = None, + language: str = "it", + ) -> SearchResponse: + """Perform a web search via SearXNG. + + Args: + query: The search query. + num_results: Maximum number of results to return. + engines: Comma-separated list of search engines to use + (e.g. "google,bing,duckduckgo"). If None, uses + the SearXNG instance defaults. + language: Language code for results. + + Returns: + SearchResponse with results and metadata. + """ + params: dict = { + "q": query, + "format": "json", + "language": language, + "safesearch": "0", + "categories": "general", + } + if engines: + params["engines"] = engines + + try: + resp = self._session.get( + f"{self.base_url}/search", + params=params, + timeout=self.timeout, + ) + resp.raise_for_status() + data = resp.json() + return self._parse_response(data, query, num_results) + except requests.RequestException as e: + return SearchResponse(query=query, error=str(e)) + + def _parse_response(self, data: dict, query: str, num_results: int) -> SearchResponse: + """Parse SearXNG JSON response into SearchResponse.""" + results: list[SearchResult] = [] + raw_results = data.get("results", [])[:num_results] + + for item in raw_results: + results.append( + SearchResult( + title=_clean_html(item.get("title", "")), + url=item.get("url", ""), + snippet=_clean_html(item.get("content", "")), + ) + ) + + return SearchResponse( + query=query, + results=results, + total_results=data.get("number_of_results", 0), + ) + + + +class PageFetcher: + """Fetch and extract readable content from web pages. + + Uses BeautifulSoup to extract the main text content, stripping + navigation, ads, scripts, and other non-content elements. + """ + + def __init__(self, timeout: int = 10, max_chars: int = 8000) -> None: + self.timeout = timeout + self.max_chars = max_chars + self._session = requests.Session() + self._session.headers.update( + { + "User-Agent": ( + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" + ), + "Accept": "text/html,application/xhtml+xml", + "Accept-Language": "it,en;q=0.9", + } + ) + + def fetch(self, url: str) -> str: + """Fetch and extract readable text from a URL. + + Args: + url: The URL to fetch. + + Returns: + Extracted text content, truncated to max_chars. + """ + try: + resp = self._session.get(url, timeout=self.timeout) + resp.raise_for_status() + + # Check content type + content_type = resp.headers.get("Content-Type", "") + if "text/html" not in content_type: + return "" + + soup = BeautifulSoup(resp.text, "html.parser") + return self._extract_text(soup) + except requests.RequestException: + return "" + + def _extract_text(self, soup: BeautifulSoup) -> str: + """Extract readable text from a BeautifulSoup document. + + Removes scripts, styles, navigation, headers, footers, and + other non-content elements. + """ + # Remove non-content elements + for tag in soup.find_all( + ["script", "style", "nav", "footer", "header", "aside", "noscript"] + ): + tag.decompose() + + # Remove common non-content classes/ids + for selector in [ + ".sidebar", ".nav", ".menu", ".advertisement", ".ads", + ".cookie", ".banner", ".popup", "#sidebar", "#nav", + '[role="navigation"]', '[role="banner"]', + ]: + for tag in soup.select(selector): + tag.decompose() + + # Get body text + body = soup.find("body") + if not body: + body = soup + + text = body.get_text(separator="\n", strip=True) + + # Clean up whitespace + lines = [line.strip() for line in text.splitlines() if line.strip()] + text = "\n".join(lines) + + # Truncate + if len(text) > self.max_chars: + text = text[: self.max_chars] + "..." + + return text + + def fetch_results( + self, + results: list[SearchResult], + max_results: int = 5, + ) -> list[SearchResult]: + """Fetch full content for a list of search results. + + Args: + results: List of SearchResult objects to enrich. + max_results: Maximum number of pages to fetch (sequential, rate-limited). + + Returns: + The same list with full_content populated where possible. + """ + # Sequential for simplicity; could be async in the future + for i, result in enumerate(results[:max_results]): + if result.fetched: + continue + content = self.fetch(result.url) + if content: + result.full_content = content + result.fetched = True + time.sleep(0.3) # Be polite to servers + + return results + + +# ── DuckDuckGo fallback (no external server needed) ──────────────── + +class DuckDuckGoClient: + """Lightweight DuckDuckGo search as fallback when SearXNG is unavailable. + + Uses DuckDuckGo's HTML interface (no API key required). + """ + + DDG_URL = "https://html.duckduckgo.com/html/" + + def __init__(self, timeout: int = 10) -> None: + self.timeout = timeout + self._session = requests.Session() + self._session.headers.update( + { + "User-Agent": ( + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" + ), + } + ) + + def search(self, query: str, num_results: int = 8) -> SearchResponse: + """Search DuckDuckGo and return results.""" + try: + resp = self._session.post( + self.DDG_URL, + data={"q": query, "b": "", "kl": "it-it"}, + timeout=self.timeout, + ) + resp.raise_for_status() + + soup = BeautifulSoup(resp.text, "html.parser") + results = [] + + for item in soup.select(".result")[:num_results]: + title_el = item.select_one(".result__title a") + snippet_el = item.select_one(".result__snippet") + + if not title_el: + continue + + title = title_el.get_text(strip=True) + + # Extract real URL from DuckDuckGo redirect + href = title_el.get("href", "") + url = self._extract_redirect(href) + + snippet = ( + snippet_el.get_text(strip=True) if snippet_el else "" + ) + + results.append( + SearchResult(title=title, url=url, snippet=snippet) + ) + + return SearchResponse( + query=query, + results=results, + total_results=len(results), + engine="duckduckgo", + ) + except requests.RequestException as e: + return SearchResponse(query=query, error=str(e)) + + @staticmethod + def _extract_redirect(href: str) -> str: + """Extract the real URL from DuckDuckGo's redirect URL.""" + if "uddg=" in href: + parsed = urlparse(href) + params = parse_qs(parsed.query) + uddg = params.get("uddg", [""])[0] + if uddg: + return unquote(uddg) + return href diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_ai.py b/tests/test_ai.py new file mode 100644 index 0000000..6493e43 --- /dev/null +++ b/tests/test_ai.py @@ -0,0 +1,135 @@ +"""Tests for sibilla.ai — GeminiClient with mocked API.""" + +import pytest +from unittest import mock +from sibilla.ai import GeminiClient, SynthesisResult + + +class TestSynthesisResult: + def test_defaults(self): + sr = SynthesisResult(answer="") + assert sr.answer == "" + assert sr.model == "" + assert sr.tokens_used == 0 + assert sr.error is None + + +class TestGeminiClient: + def test_no_api_key_returns_error(self): + client = GeminiClient(api_key="") + result = client.synthesize("system", "user") + assert result.answer == "" + assert "GOOGLE_API_KEY" in result.error + + def test_lazy_client_raises_without_key(self): + client = GeminiClient(api_key="") + with pytest.raises(ValueError, match="GOOGLE_API_KEY"): + _ = client.client + + def test_synthesize_success(self): + client = GeminiClient(api_key="fake-key") + mock_genai = mock.MagicMock() + client._client = mock_genai + + mock_part = mock.MagicMock() + mock_part.text = "This is the answer." + mock_content = mock.MagicMock() + mock_content.parts = [mock_part] + mock_candidate = mock.MagicMock() + mock_candidate.content = mock_content + mock_usage = mock.MagicMock() + mock_usage.total_token_count = 42 + mock_response = mock.MagicMock() + mock_response.candidates = [mock_candidate] + mock_response.usage_metadata = mock_usage + + mock_genai.models.generate_content.return_value = mock_response + + result = client.synthesize("system prompt", "user query") + + assert result.answer == "This is the answer." + assert result.tokens_used == 42 + assert result.model == "gemini-2.5-flash" + assert result.error is None + + def test_synthesize_no_candidates(self): + client = GeminiClient(api_key="fake-key") + mock_genai = mock.MagicMock() + client._client = mock_genai + + mock_response = mock.MagicMock() + mock_response.candidates = [] + mock_response.usage_metadata = None + mock_genai.models.generate_content.return_value = mock_response + + result = client.synthesize("s", "u") + assert result.answer == "" + assert result.error is None + + def test_synthesize_api_exception(self): + client = GeminiClient(api_key="fake-key") + mock_genai = mock.MagicMock() + client._client = mock_genai + + mock_genai.models.generate_content.side_effect = RuntimeError("API down") + + result = client.synthesize("s", "u") + assert result.error == "API down" + assert result.answer == "" + + def test_rewrite_query_no_api_key(self): + client = GeminiClient(api_key="") + variants = client.rewrite_query("test query", num_variants=3) + assert variants == ["test query"] + + def test_rewrite_query_success(self): + client = GeminiClient(api_key="fake-key") + mock_genai = mock.MagicMock() + client._client = mock_genai + + mock_part = mock.MagicMock() + mock_part.text = "rewritten query 1\nrewritten query 2\nrewritten query 3" + mock_content = mock.MagicMock() + mock_content.parts = [mock_part] + mock_candidate = mock.MagicMock() + mock_candidate.content = mock_content + mock_response = mock.MagicMock() + mock_response.candidates = [mock_candidate] + mock_genai.models.generate_content.return_value = mock_response + + variants = client.rewrite_query("original", num_variants=3) + + assert len(variants) == 4 + assert variants[0] == "original" + assert "rewritten query 1" in variants + + def test_rewrite_query_strips_bullets(self): + client = GeminiClient(api_key="fake-key") + mock_genai = mock.MagicMock() + client._client = mock_genai + + mock_part = mock.MagicMock() + mock_part.text = "- bullet query\n• dot query" + mock_content = mock.MagicMock() + mock_content.parts = [mock_part] + mock_candidate = mock.MagicMock() + mock_candidate.content = mock_content + mock_response = mock.MagicMock() + mock_response.candidates = [mock_candidate] + mock_genai.models.generate_content.return_value = mock_response + + variants = client.rewrite_query("original", num_variants=2) + + assert "bullet query" in variants + assert "dot query" in variants + assert not any(v.startswith("-") for v in variants) + + def test_rewrite_query_exception_fallback(self): + client = GeminiClient(api_key="fake-key") + mock_genai = mock.MagicMock() + client._client = mock_genai + + mock_genai.models.generate_content.side_effect = RuntimeError("boom") + + variants = client.rewrite_query("original", num_variants=2) + assert variants == ["original"] diff --git a/tests/test_app.py b/tests/test_app.py new file mode 100644 index 0000000..2a096c1 --- /dev/null +++ b/tests/test_app.py @@ -0,0 +1,252 @@ +"""Tests for sibilla.app — Flask routes and search pipeline.""" + +import os +import pytest +from unittest import mock +from sibilla.app import create_app +from sibilla.search import SearchResult, SearchResponse +from sibilla.ai import SynthesisResult + + +@pytest.fixture +def app(): + """Create a Flask app in testing mode with a fake Gemini API key.""" + app = create_app(testing=True) + # Point templates to the project-level templates/ directory + app.template_folder = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "templates" + ) + app.config.update({ + "TESTING": True, + "GOOGLE_API_KEY": "fake-test-key", + "DEGOOG_URL": "http://degoog.test:4444", + "SEARXNG_URL": "http://searxng.test:8080", + "SIBILLA_DDG_FALLBACK": True, + }) + return app + + +@pytest.fixture +def client(app): + return app.test_client() + + +class TestHealthEndpoint: + def test_health_returns_ok(self, client): + resp = client.get("/api/health") + assert resp.status_code == 200 + data = resp.get_json() + assert data["status"] == "ok" + assert "version" in data + assert data["gemini_configured"] is True + assert "degoog" in data["degoog_url"] + + def test_health_gemini_not_configured(self, app): + app.config["GOOGLE_API_KEY"] = "" + client = app.test_client() + resp = client.get("/api/health") + data = resp.get_json() + assert data["gemini_configured"] is False + + +class TestIndexRoute: + def test_index_returns_html(self, client): + resp = client.get("/") + assert resp.status_code == 200 + assert "text/html" in resp.content_type + html = resp.data.decode("utf-8") + assert "Sibilla" in html + assert 'id="search-form"' in html + assert 'id="query"' in html + assert 'id="response-area"' in html + + +class TestAskEndpoint: + def test_empty_query_returns_400(self, client): + resp = client.post("/api/ask", json={"query": ""}) + assert resp.status_code == 400 + assert "non può essere vuota" in resp.get_json()["error"] + + def test_ask_with_degoog_success(self, app): + """Full pipeline test with mocked Degoog search and Gemini synthesis.""" + client = app.test_client() + + # Mock DegoogClient.search to return results + def mock_degoog_search(self, query, num_results=8, language="it"): + return SearchResponse( + query=query, + results=[ + SearchResult(title="R1", url="https://a.com", snippet="Content A"), + SearchResult(title="R2", url="https://b.com", snippet="Content B"), + ], + engine="degoog", + ) + + # Mock Gemini client to return a synthesized answer + def mock_synthesize(self, system_prompt, user_prompt, temperature=0.4, max_tokens=4096): + return SynthesisResult( + answer="Ecco la risposta sintetizzata [1][2].", + model="gemini-2.5-flash", + tokens_used=100, + ) + + def mock_rewrite_query(self, original_query, num_variants=3): + return [original_query, "rewritten variant"] + + with mock.patch( + "sibilla.app.DegoogClient.search", mock_degoog_search + ), mock.patch( + "sibilla.app.GeminiClient.synthesize", mock_synthesize + ), mock.patch( + "sibilla.app.GeminiClient.rewrite_query", mock_rewrite_query + ), mock.patch( + "sibilla.app.PageFetcher.fetch_results", + lambda self, results, max_results=5: results, # passthrough + ): + resp = client.post("/api/ask", json={"query": "test query"}) + + assert resp.status_code == 200 + data = resp.get_json() + assert "Ecco la risposta sintetizzata" in data["answer"] + assert data["search_engine"] == "degoog" + assert len(data["sources"]) == 2 + assert data["sources"][0]["url"] == "https://a.com" + assert data["model"] == "gemini-2.5-flash" + assert data["tokens_used"] == 100 + + def test_ask_fallback_to_searxng(self, app): + """When Degoog returns no results, fall back to SearXNG.""" + client = app.test_client() + + def mock_degoog_search_empty(self, query, num_results=8, language="it"): + return SearchResponse(query=query, results=[], engine="degoog") + + def mock_searxng_search(self, query, num_results=8, engines=None, language="it"): + return SearchResponse( + query=query, + results=[ + SearchResult(title="SR1", url="https://s.com", snippet="S content"), + ], + engine="searxng", + ) + + def mock_synthesize(self, system_prompt, user_prompt, temperature=0.4, max_tokens=4096): + return SynthesisResult(answer="Answer from SearXNG sources [1].", model="g", tokens_used=50) + + def mock_rewrite_query(self, original_query, num_variants=3): + return [original_query] + + with mock.patch( + "sibilla.app.DegoogClient.search", mock_degoog_search_empty + ), mock.patch( + "sibilla.app.SearXNGClient.search", mock_searxng_search + ), mock.patch( + "sibilla.app.GeminiClient.synthesize", mock_synthesize + ), mock.patch( + "sibilla.app.GeminiClient.rewrite_query", mock_rewrite_query + ), mock.patch( + "sibilla.app.PageFetcher.fetch_results", + lambda self, results, max_results=5: results, + ): + resp = client.post("/api/ask", json={"query": "fallback test"}) + + assert resp.status_code == 200 + data = resp.get_json() + assert data["search_engine"] == "searxng" + + def test_ask_no_results_fallback_to_llm(self, app): + """When all search backends fail, fall back to pure LLM.""" + client = app.test_client() + + def mock_empty_search(self, *args, **kwargs): + return SearchResponse(query="q", results=[], engine="none") + + def mock_synthesize(self, system_prompt, user_prompt, temperature=0.4, max_tokens=4096): + return SynthesisResult(answer="Nessun risultato, ma posso aiutarti...", model="g", tokens_used=10) + + def mock_rewrite_query(self, original_query, num_variants=3): + return [original_query] + + with mock.patch( + "sibilla.app.DegoogClient.search", mock_empty_search + ), mock.patch( + "sibilla.app.SearXNGClient.search", mock_empty_search + ), mock.patch( + "sibilla.app.DuckDuckGoClient.search", mock_empty_search + ), mock.patch( + "sibilla.app.GeminiClient.synthesize", mock_synthesize + ), mock.patch( + "sibilla.app.GeminiClient.rewrite_query", mock_rewrite_query + ): + resp = client.post("/api/ask", json={"query": "impossible query"}) + + assert resp.status_code == 200 + data = resp.get_json() + assert data["search_engine"] == "none" + assert data["sources"] == [] + + def test_ask_all_backends_error(self, app): + """When all backends return errors, return 502.""" + client = app.test_client() + + def mock_error_search(self, *args, **kwargs): + return SearchResponse(query="q", results=[], error="Connection refused", engine="degoog") + + def mock_rewrite_query(self, original_query, num_variants=3): + return [original_query] + + with mock.patch( + "sibilla.app.DegoogClient.search", mock_error_search + ), mock.patch( + "sibilla.app.SearXNGClient.search", mock_error_search + ), mock.patch( + "sibilla.app.DuckDuckGoClient.search", mock_error_search + ), mock.patch( + "sibilla.app.GeminiClient.rewrite_query", mock_rewrite_query + ): + resp = client.post("/api/ask", json={"query": "doomed query"}) + + assert resp.status_code == 502 + + def test_ask_respects_num_sources(self, app): + client = app.test_client() + + call_count = 0 + + def mock_degoog_search(self, query, num_results=8, language="it"): + nonlocal call_count + call_count += 1 + return SearchResponse( + query=query, + results=[ + SearchResult(title=f"R{i}", url=f"https://{i}.com", snippet=f"S{i}") + for i in range(10) + ], + engine="degoog", + ) + + def mock_synthesize(self, system_prompt, user_prompt, temperature=0.4, max_tokens=4096): + return SynthesisResult(answer="Ok.", model="g", tokens_used=1) + + def mock_rewrite_query(self, original_query, num_variants=3): + return [original_query] + + with mock.patch( + "sibilla.app.DegoogClient.search", mock_degoog_search + ), mock.patch( + "sibilla.app.GeminiClient.synthesize", mock_synthesize + ), mock.patch( + "sibilla.app.GeminiClient.rewrite_query", mock_rewrite_query + ), mock.patch( + "sibilla.app.PageFetcher.fetch_results", + lambda self, results, max_results=5: results, + ): + resp = client.post("/api/ask", json={"query": "test", "num_sources": 3}) + + assert resp.status_code == 200 + data = resp.get_json() + assert len(data["sources"]) <= 3 + + def test_ask_without_json_body(self, client): + resp = client.post("/api/ask", data="not json") + assert resp.status_code == 400 diff --git a/tests/test_prompts.py b/tests/test_prompts.py new file mode 100644 index 0000000..c5edf6e --- /dev/null +++ b/tests/test_prompts.py @@ -0,0 +1,79 @@ +"""Tests for sibilla.prompts — prompt templates and builders.""" + +from sibilla.prompts import ( + SYNTHESIS_SYSTEM_PROMPT, + FALLBACK_PROMPT, + QUERY_REWRITE_PROMPT, + build_synthesis_prompt, +) + + +class TestPromptTemplates: + """Verify all prompt templates are non-empty and well-formed.""" + + def test_synthesis_system_prompt_not_empty(self): + assert len(SYNTHESIS_SYSTEM_PROMPT) > 50 + assert "Sibilla" in SYNTHESIS_SYSTEM_PROMPT + assert "[1]" in SYNTHESIS_SYSTEM_PROMPT + + def test_fallback_prompt_not_empty(self): + assert len(FALLBACK_PROMPT) > 20 + assert "Sibilla" in FALLBACK_PROMPT + + def test_query_rewrite_prompt_has_placeholder(self): + assert "{num_variants}" in QUERY_REWRITE_PROMPT + + +class TestBuildSynthesisPrompt: + """Test build_synthesis_prompt with various inputs.""" + + def test_minimal_prompt(self): + sources = [ + {"index": 1, "title": "Test", "url": "https://example.com", "content": "Hello world"} + ] + prompt = build_synthesis_prompt("what is this?", sources) + + assert "what is this?" in prompt + assert "Test" in prompt + assert "https://example.com" in prompt + assert "Hello world" in prompt + assert "[1]" in prompt + + def test_multiple_sources(self): + sources = [ + {"index": 1, "title": "A", "url": "https://a.com", "content": "Content A"}, + {"index": 2, "title": "B", "url": "https://b.com", "content": "Content B"}, + {"index": 3, "title": "C", "url": "https://c.com", "content": "Content C"}, + ] + prompt = build_synthesis_prompt("query", sources) + + assert "--- Fonte [1] ---" in prompt + assert "--- Fonte [2] ---" in prompt + assert "--- Fonte [3] ---" in prompt + assert "Content A" in prompt + assert "Content B" in prompt + assert "Content C" in prompt + + def test_with_date_context(self): + sources = [{"index": 1, "title": "X", "url": "https://x.com", "content": "Y"}] + prompt = build_synthesis_prompt("q", sources, date_context="Data: 16 Luglio 2026") + + assert "16 Luglio 2026" in prompt + assert "Contesto temporale" in prompt + + def test_empty_sources(self): + prompt = build_synthesis_prompt("query", []) + assert "query" in prompt + assert "### Fonti da utilizzare" in prompt + + def test_source_missing_fields(self): + sources = [{"index": 1}] # missing title, url, content + prompt = build_synthesis_prompt("q", sources) + assert "Senza titolo" in prompt # fallback title + assert "[1]" in prompt + + def test_source_with_snippet_fallback(self): + """When 'content' is missing, fall back to 'snippet'.""" + sources = [{"index": 1, "title": "T", "url": "https://t.com", "snippet": "snippet text"}] + prompt = build_synthesis_prompt("q", sources) + assert "snippet text" in prompt diff --git a/tests/test_search.py b/tests/test_search.py new file mode 100644 index 0000000..1d56303 --- /dev/null +++ b/tests/test_search.py @@ -0,0 +1,398 @@ +"""Tests for sibilla.search — search clients and page fetcher.""" + +from unittest import mock +from sibilla.search import ( + DegoogClient, + SearXNGClient, + DuckDuckGoClient, + PageFetcher, + SearchResult, + SearchResponse, + _clean_html, +) + + +class TestCleanHtml: + def test_removes_tags(self): + assert _clean_html("<p>Hello <b>world</b></p>") == "Hello world" + + def test_plain_text_unchanged(self): + assert _clean_html("Plain text") == "Plain text" + + def test_empty_string(self): + assert _clean_html("") == "" + + def test_nested_tags(self): + assert _clean_html('<div><span class="x">Text</span></div>') == "Text" + + def test_strips_html_entities(self): + # HTML entities stay as-is (not entity-decoded), just tags stripped + assert _clean_html("<b>Bold & Brave</b>") == "Bold & Brave" + + +class TestSearchDataclasses: + def test_search_result_defaults(self): + sr = SearchResult(title="T", url="https://u.com") + assert sr.snippet == "" + assert sr.full_content == "" + assert sr.fetched is False + + def test_search_response_defaults(self): + sr = SearchResponse(query="q") + assert sr.results == [] + assert sr.total_results == 0 + assert sr.engine == "searxng" + assert sr.error is None + + +# ── DegoogClient tests ──────────────────────────────────────────── + +class TestDegoogClient: + def test_search_success(self): + client = DegoogClient(base_url="http://degoog.test:4444") + + mock_data = { + "results": [ + {"title": "R1", "url": "https://a.com", "snippet": "S1"}, + {"title": "R2", "url": "https://b.com", "snippet": "S2"}, + ] + } + + with mock.patch.object(client._session, "get") as mock_get: + mock_get.return_value.json.return_value = mock_data + mock_get.return_value.raise_for_status.return_value = None + + resp = client.search("test query", num_results=5) + + assert len(resp.results) == 2 + assert resp.results[0].title == "R1" + assert resp.results[0].url == "https://a.com" + assert resp.results[0].snippet == "S1" + assert resp.engine == "degoog" + assert resp.error is None + + def test_search_respects_num_results(self): + client = DegoogClient(base_url="http://degoog.test:4444") + + mock_data = { + "results": [ + {"title": f"R{i}", "url": f"https://{i}.com", "snippet": f"S{i}"} + for i in range(10) + ] + } + + with mock.patch.object(client._session, "get") as mock_get: + mock_get.return_value.json.return_value = mock_data + mock_get.return_value.raise_for_status.return_value = None + + resp = client.search("q", num_results=3) + + assert len(resp.results) == 3 + + def test_search_empty_results(self): + client = DegoogClient(base_url="http://degoog.test:4444") + + with mock.patch.object(client._session, "get") as mock_get: + mock_get.return_value.json.return_value = {"results": []} + mock_get.return_value.raise_for_status.return_value = None + + resp = client.search("no results query") + + assert resp.results == [] + assert resp.error is None + + def test_search_handles_request_exception(self): + client = DegoogClient(base_url="http://degoog.test:4444") + + with mock.patch.object(client._session, "get") as mock_get: + import requests + mock_get.side_effect = requests.RequestException("Connection refused") + + resp = client.search("q") + + assert resp.error is not None + assert resp.results == [] + + def test_search_cleans_html_in_results(self): + """Verify that _clean_html is applied to title and snippet.""" + client = DegoogClient(base_url="http://degoog.test:4444") + + mock_data = { + "results": [ + { + "title": "<b>Bold Title</b>", + "url": "https://x.com", + "snippet": "<p>Some <em>text</em></p>", + } + ] + } + + with mock.patch.object(client._session, "get") as mock_get: + mock_get.return_value.json.return_value = mock_data + mock_get.return_value.raise_for_status.return_value = None + + resp = client.search("q") + + assert resp.results[0].title == "Bold Title" + assert resp.results[0].snippet == "Some text" + + def test_search_uses_content_fallback_for_snippet(self): + """Degoog may return 'content' instead of 'snippet' (OpenWebUI compat).""" + client = DegoogClient(base_url="http://degoog.test:4444") + + mock_data = { + "results": [ + {"title": "R", "url": "https://x.com", "content": "Content field"} + ] + } + + with mock.patch.object(client._session, "get") as mock_get: + mock_get.return_value.json.return_value = mock_data + mock_get.return_value.raise_for_status.return_value = None + + resp = client.search("q") + + assert resp.results[0].snippet == "Content field" + + def test_search_query_params(self): + client = DegoogClient(base_url="http://degoog.test:4444") + + with mock.patch.object(client._session, "get") as mock_get: + mock_get.return_value.json.return_value = {"results": []} + mock_get.return_value.raise_for_status.return_value = None + + client.search("test query", language="en") + + call_args = mock_get.call_args + assert call_args[1]["params"]["q"] == "test query" + assert call_args[1]["params"]["lang"] == "en" + + +# ── SearXNGClient tests ─────────────────────────────────────────── + +class TestSearXNGClient: + def test_search_success(self): + client = SearXNGClient(base_url="http://searxng.test:8080") + + mock_data = { + "results": [ + {"title": "R1", "url": "https://a.com", "content": "S1 content"}, + ], + "number_of_results": 1, + } + + with mock.patch.object(client._session, "get") as mock_get: + mock_get.return_value.json.return_value = mock_data + mock_get.return_value.raise_for_status.return_value = None + + resp = client.search("test", num_results=5) + + assert len(resp.results) == 1 + assert resp.results[0].snippet == "S1 content" + assert resp.total_results == 1 + + def test_search_handles_error(self): + client = SearXNGClient(base_url="http://searxng.test:8080") + + with mock.patch.object(client._session, "get") as mock_get: + import requests + mock_get.side_effect = requests.RequestException("Timeout") + + resp = client.search("q") + + assert resp.error is not None + assert "Timeout" in resp.error + + def test_search_with_engines_param(self): + client = SearXNGClient(base_url="http://searxng.test:8080") + + with mock.patch.object(client._session, "get") as mock_get: + mock_get.return_value.json.return_value = {"results": []} + mock_get.return_value.raise_for_status.return_value = None + + client.search("q", engines="google,bing") + + call_args = mock_get.call_args + assert call_args[1]["params"]["engines"] == "google,bing" + + +# ── DuckDuckGoClient tests ──────────────────────────────────────── + +DUCKDUCKGO_HTML = """ +<!DOCTYPE html> +<html> +<body> +<div class="result"> + <div class="result__title"> + <a href="https://duckduckgo.com/l/?uddg=https%3A%2F%2Fexample.com&rut=abc">Example Page</a> + </div> + <div class="result__snippet">This is the snippet text for the first result.</div> +</div> +<div class="result"> + <div class="result__title"> + <a href="https://another-site.org/page">Another Site</a> + </div> + <div class="result__snippet">Second snippet over here.</div> +</div> +<div class="result"> + <div class="result__title"> + <a href="/l/?uddg=https%3A%2F%2Fthird.com">Third Result</a> + </div> + <!-- No snippet --> +</div> +</body> +</html> +""" + + +class TestDuckDuckGoClient: + def test_extract_redirect(self): + client = DuckDuckGoClient() + url = "https://duckduckgo.com/l/?uddg=https%3A%2F%2Fexample.com&rut=abc" + result = client._extract_redirect(url) + assert result == "https://example.com" + + def test_extract_redirect_no_uddg(self): + client = DuckDuckGoClient() + result = client._extract_redirect("https://example.com/page") + assert result == "https://example.com/page" + + def test_search_parses_html_results(self): + """Test that DuckDuckGo HTML results are parsed correctly.""" + client = DuckDuckGoClient() + + mock_resp = mock.MagicMock() + mock_resp.text = DUCKDUCKGO_HTML + mock_resp.raise_for_status.return_value = None + + with mock.patch.object(client._session, "post") as mock_post: + mock_post.return_value = mock_resp + + resp = client.search("test query", num_results=5) + + assert len(resp.results) == 3 + assert resp.results[0].title == "Example Page" + assert resp.results[0].url == "https://example.com" # redirect extracted + assert resp.results[0].snippet == "This is the snippet text for the first result." + assert resp.results[1].url == "https://another-site.org/page" + assert resp.results[2].snippet == "" # no snippet element + assert resp.engine == "duckduckgo" + assert resp.error is None + + def test_search_handles_error(self): + client = DuckDuckGoClient() + + with mock.patch.object(client._session, "post") as mock_post: + import requests + mock_post.side_effect = requests.RequestException("Network error") + + resp = client.search("q") + + assert resp.error is not None + assert "Network error" in resp.error + + +# ── PageFetcher tests ───────────────────────────────────────────── + +class TestPageFetcher: + def test_fetch_extracts_text(self): + fetcher = PageFetcher() + + html = """<html><body> + <script>alert('xss')</script> + <nav>Menu</nav> + <p>Hello world</p> + <footer>Footer</footer> + </body></html>""" + + with mock.patch.object(fetcher._session, "get") as mock_get: + mock_get.return_value.text = html + mock_get.return_value.headers = {"Content-Type": "text/html; charset=utf-8"} + mock_get.return_value.raise_for_status.return_value = None + + content = fetcher.fetch("https://example.com") + + assert "Hello world" in content + assert "alert" not in content # script removed + assert "Footer" not in content # footer removed + + def test_fetch_non_html_returns_empty(self): + fetcher = PageFetcher() + + with mock.patch.object(fetcher._session, "get") as mock_get: + mock_get.return_value.headers = {"Content-Type": "application/pdf"} + mock_get.return_value.raise_for_status.return_value = None + + content = fetcher.fetch("https://example.com/doc.pdf") + + assert content == "" + + def test_fetch_handles_error(self): + fetcher = PageFetcher() + + with mock.patch.object(fetcher._session, "get") as mock_get: + import requests + mock_get.side_effect = requests.RequestException("Connection refused") + + content = fetcher.fetch("https://down.example.com") + + assert content == "" + + def test_fetch_truncates_to_max_chars(self): + fetcher = PageFetcher(max_chars=50) + + long_text = "A" * 200 + html = f"<html><body><p>{long_text}</p></body></html>" + + with mock.patch.object(fetcher._session, "get") as mock_get: + mock_get.return_value.text = html + mock_get.return_value.headers = {"Content-Type": "text/html"} + mock_get.return_value.raise_for_status.return_value = None + + content = fetcher.fetch("https://example.com") + + assert len(content) <= 53 # 50 + "..." truncation marker + + def test_fetch_results_enriches_objects(self): + fetcher = PageFetcher() + results = [ + SearchResult(title="A", url="https://a.com"), + SearchResult(title="B", url="https://b.com"), + ] + + with mock.patch.object(fetcher, "fetch") as mock_fetch: + mock_fetch.side_effect = ["Content A", "Content B"] + + enriched = fetcher.fetch_results(results, max_results=2) + + assert enriched[0].full_content == "Content A" + assert enriched[0].fetched is True + assert enriched[1].full_content == "Content B" + assert enriched[1].fetched is True + + def test_fetch_results_respects_max(self): + fetcher = PageFetcher() + results = [SearchResult(title=f"R{i}", url=f"https://{i}.com") for i in range(5)] + + with mock.patch.object(fetcher, "fetch") as mock_fetch: + mock_fetch.return_value = "Content" + + fetcher.fetch_results(results, max_results=2) + + assert mock_fetch.call_count == 2 + + def test_fetch_results_skips_already_fetched(self): + fetcher = PageFetcher() + results = [ + SearchResult(title="A", url="https://a.com", fetched=True, full_content="Old"), + SearchResult(title="B", url="https://b.com"), + ] + + with mock.patch.object(fetcher, "fetch") as mock_fetch: + mock_fetch.return_value = "New" + + fetcher.fetch_results(results, max_results=2) + + # Only the non-fetched result should be fetched + assert mock_fetch.call_count == 1 + assert results[0].full_content == "Old" # unchanged