Spaces:
Running
Running
| from __future__ import annotations | |
| import unicodedata | |
| from ddgs import DDGS | |
| from config import CONFIG | |
| # Topic stems for a turn that needs current, public facts. Prefix-matched, so | |
| # inflections hit (resultado/resultados). Excludes chitchat (hoy, ahora, cuando) | |
| # to avoid false positives that add a slow search mid-conversation. | |
| _SEARCH_STEMS = ( | |
| "clima", "tiempo", "temperatura", "lluvia", "pronostic", | |
| "notici", "actualidad", "precio", "cuesta", "dolar", | |
| "partido", "marcador", "result", "horario", "campeonat", "torneo", | |
| "eleccion", "elecc", "candidat", "ganador", "president", | |
| "voto", "vota", "balota", "congres", "alcald", "gobern", | |
| ) | |
| # Cues that mark a follow-up riding on the previous turn's topic. | |
| _FOLLOWUP_CUES = { | |
| "mas", "ese", "esa", "eso", "esos", "esas", "quien", "quienes", | |
| "cual", "cuales", "detalle", "detalles", "cuenta", "cuentame", "dime", | |
| } | |
| def _norm(text: str) -> str: | |
| """Lowercase and strip accents so 'mañana' matches 'manana'.""" | |
| decomposed = unicodedata.normalize("NFD", text.lower()) | |
| return "".join(c for c in decomposed if unicodedata.category(c) != "Mn") | |
| def _tokens(text: str) -> list[str]: | |
| """Normalize text and split it into words.""" | |
| return _norm(text).replace("¿", " ").replace("?", " ").split() | |
| def _has_topic(text: str) -> bool: | |
| """True when a word starts with one of the current-events stems.""" | |
| return any(w.startswith(stem) for w in _tokens(text) for stem in _SEARCH_STEMS) | |
| def needs_search(text: str, history: list[dict] | None = None) -> bool: | |
| """True when the turn looks like it needs current public facts. | |
| Fires on a topic word, or on a follow-up question inside a thread whose | |
| previous turn was already about a current-events topic -- so "cuéntame más | |
| sobre los candidatos" keeps searching even with no topic word of its own. | |
| """ | |
| if not CONFIG.search_enabled: | |
| return False | |
| if _has_topic(text): | |
| return True | |
| if history: | |
| prev_user = next( | |
| (m["content"] for m in reversed(history) | |
| if m.get("role") == "user" and m.get("content")), "", | |
| ) | |
| if prev_user and _has_topic(prev_user): | |
| if "?" in text or (set(_tokens(text)) & _FOLLOWUP_CUES): | |
| return True | |
| return False | |
| def search_web(query: str) -> str: | |
| """Return a short, LLM-friendly digest of top results.""" | |
| if not query: | |
| return "Sin consulta." | |
| try: | |
| with DDGS() as ddgs: | |
| results = list( | |
| ddgs.text(query, max_results=CONFIG.search_max_results) | |
| ) | |
| except Exception as exc: # noqa: BLE001 - degrade gracefully for the elder | |
| return f"No pude buscar en internet ahora mismo ({exc})." | |
| if not results: | |
| return "No encontre resultados." | |
| lines = [] | |
| for r in results: | |
| title = r.get("title", "") | |
| body = r.get("body", "") | |
| lines.append(f"- {title}: {body}") | |
| return "\n".join(lines) | |