Búsqueda web con DuckDuckGo, sin API key
Primera tool del agente. No depende del modelo, así que se puede usar y
verificar de punta a punta antes de que exista el modelo propio.
Sobre el endpoint: la API oficial de DuckDuckGo ("Instant Answer") no sirve
para esto — responde definiciones y fichas de entidades, y para una consulta
normal devuelve 200 con el cuerpo vacío. Los resultados web reales solo están
en el endpoint lite, que no es una API con contrato: el HTML puede cambiar y
hay límite de tasa. Por eso el backend está detrás de una interfaz y queda
implementado también SearxNG autoalojado, que es un cambio de una línea de
config cuando haga falta.
Decisiones:
- Presupuesto de contexto explícito: con seq_len 2048 los resultados compiten
con la memoria recuperada y el turno del usuario, así que se devuelven cinco
recortados en vez de diez completos.
- Las redirecciones /l/?uddg= se desenvuelven al destino real: guardarlas
ensuciaría la memoria episódica, donde dos búsquedas a la misma página
parecerían páginas distintas.
- Caché con TTL, que es de comportamiento y no de rendimiento: en una casa las
mismas preguntas se repiten muchas veces por día.
- Sin resultados devuelve "Sin resultados." en vez de vacío, para que el modelo
pueda decir que no encontró nada en lugar de inventar.
Se quitan Brave y Tavily de .env.example: ya no hay credenciales que gestionar
ni un tercero al que informarle qué busca la familia.
23 tests nuevos (71 en total), contra una respuesta real guardada como fixture:
el parser es lo que se rompe cuando cambia el markup, y tiene que fallar en la
suite y no en producción. Ningún test sale a internet.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,292 @@
|
||||
"""Búsqueda web. Sin API key y sin cuenta en ningún servicio.
|
||||
|
||||
Se usa DuckDuckGo por su endpoint ligero, que devuelve resultados web sin
|
||||
credenciales. Dos aclaraciones sobre por qué está hecho así:
|
||||
|
||||
- La API oficial de DuckDuckGo (api.duckduckgo.com, "Instant Answer") no
|
||||
sirve acá: responde definiciones y fichas de entidades, y para una consulta
|
||||
normal devuelve 200 con el cuerpo vacío. Los resultados web reales solo
|
||||
están en el endpoint lite/html.
|
||||
- Ese endpoint no es una API con contrato: el HTML puede cambiar y hay límite
|
||||
de tasa. Por eso el backend está detrás de una interfaz y hay una
|
||||
implementación alternativa contra SearxNG autoalojado, que sí es estable y
|
||||
no depende de terceros. Cambiar de una a otra es una línea de config.
|
||||
|
||||
El resultado se normaliza y se recorta con un presupuesto de caracteres: con un
|
||||
modelo chico y 2048 tokens de contexto, volcarle diez resultados completos es
|
||||
peor que darle tres buenos.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
import json
|
||||
import re
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from dataclasses import dataclass
|
||||
from typing import Protocol
|
||||
|
||||
# Un navegador real: el endpoint lite rechaza clientes sin User-Agent.
|
||||
_USER_AGENT = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125 Safari/537.36"
|
||||
|
||||
_RE_LINK = re.compile(
|
||||
r"<a[^>]*?href=[\"'](?P<url>[^\"']+)[\"'][^>]*?class=['\"]result-link['\"][^>]*?>"
|
||||
r"(?P<title>.*?)</a>",
|
||||
re.DOTALL | re.IGNORECASE,
|
||||
)
|
||||
_RE_SNIPPET = re.compile(
|
||||
r"class=['\"]result-snippet['\"][^>]*>(?P<snippet>.*?)</td>", re.DOTALL | re.IGNORECASE
|
||||
)
|
||||
_RE_TAGS = re.compile(r"<[^>]+>")
|
||||
_RE_SPACES = re.compile(r"\s+")
|
||||
|
||||
|
||||
class SearchError(RuntimeError):
|
||||
"""La búsqueda no se pudo completar. El agente responde que no pudo buscar."""
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SearchResult:
|
||||
title: str
|
||||
url: str
|
||||
snippet: str
|
||||
|
||||
def render(self, max_chars: int) -> str:
|
||||
"""Una línea compacta para inyectar en el contexto del modelo."""
|
||||
cuerpo = self.snippet[:max_chars].rstrip()
|
||||
return f"{self.title} — {cuerpo} ({self.url})"
|
||||
|
||||
|
||||
def _clean(raw: str) -> str:
|
||||
"""Quita etiquetas, resuelve entidades y colapsa espacios."""
|
||||
return _RE_SPACES.sub(" ", html.unescape(_RE_TAGS.sub("", raw))).strip()
|
||||
|
||||
|
||||
def _unwrap_redirect(url: str) -> str:
|
||||
"""DuckDuckGo a veces envuelve el destino en /l/?uddg=<url>.
|
||||
|
||||
Guardar la redirección en vez del destino ensucia la memoria episódica y
|
||||
hace imposible reconocer que dos búsquedas llegaron a la misma página.
|
||||
"""
|
||||
if "uddg=" not in url:
|
||||
return url if url.startswith("http") else f"https:{url}" if url.startswith("//") else url
|
||||
query = urllib.parse.urlparse(url if url.startswith("http") else f"https:{url}").query
|
||||
destino = urllib.parse.parse_qs(query).get("uddg", [])
|
||||
return destino[0] if destino else url
|
||||
|
||||
|
||||
def parse_duckduckgo_html(body: str, max_results: int) -> list[SearchResult]:
|
||||
"""Extrae los resultados del HTML del endpoint lite.
|
||||
|
||||
Está separado de la red a propósito: es la parte que se rompe cuando
|
||||
DuckDuckGo cambia el markup, y se testea contra una respuesta real guardada
|
||||
en tests/fixtures/ sin salir a internet.
|
||||
"""
|
||||
enlaces = list(_RE_LINK.finditer(body))
|
||||
fragmentos = [m.group("snippet") for m in _RE_SNIPPET.finditer(body)]
|
||||
|
||||
resultados: list[SearchResult] = []
|
||||
for i, enlace in enumerate(enlaces):
|
||||
titulo = _clean(enlace.group("title"))
|
||||
url = _unwrap_redirect(html.unescape(enlace.group("url")))
|
||||
if not titulo or not url.startswith("http"):
|
||||
continue
|
||||
# Los snippets vienen en el mismo orden que los enlaces, pero puede
|
||||
# faltar alguno: se empareja por posición y se tolera la ausencia.
|
||||
snippet = _clean(fragmentos[i]) if i < len(fragmentos) else ""
|
||||
resultados.append(SearchResult(title=titulo, url=url, snippet=snippet))
|
||||
if len(resultados) >= max_results:
|
||||
break
|
||||
return resultados
|
||||
|
||||
|
||||
class SearchBackend(Protocol):
|
||||
def search(self, query: str, max_results: int) -> list[SearchResult]: ...
|
||||
|
||||
|
||||
class DuckDuckGoBackend:
|
||||
"""Endpoint lite de DuckDuckGo. Sin clave, sin cuenta, sin terceros."""
|
||||
|
||||
name = "duckduckgo"
|
||||
ENDPOINT = "https://lite.duckduckgo.com/lite/"
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
region: str = "es-es",
|
||||
timeout: float = 10.0,
|
||||
retries: int = 2,
|
||||
backoff: float = 1.5,
|
||||
) -> None:
|
||||
self.region = region
|
||||
self.timeout = timeout
|
||||
self.retries = retries
|
||||
self.backoff = backoff
|
||||
|
||||
def search(self, query: str, max_results: int) -> list[SearchResult]:
|
||||
datos = urllib.parse.urlencode({"q": query, "kl": self.region}).encode()
|
||||
request = urllib.request.Request(
|
||||
self.ENDPOINT, data=datos, headers={"User-Agent": _USER_AGENT}
|
||||
)
|
||||
|
||||
ultimo: Exception | None = None
|
||||
for intento in range(self.retries + 1):
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=self.timeout) as respuesta:
|
||||
body = respuesta.read().decode("utf-8", errors="replace")
|
||||
return parse_duckduckgo_html(body, max_results)
|
||||
except (urllib.error.URLError, TimeoutError, OSError) as exc:
|
||||
ultimo = exc
|
||||
if intento < self.retries:
|
||||
# Espera creciente: el endpoint limita por tasa y reintentar
|
||||
# de inmediato solo empeora el bloqueo.
|
||||
time.sleep(self.backoff * (intento + 1))
|
||||
|
||||
raise SearchError(f"DuckDuckGo no respondió tras {self.retries + 1} intentos: {ultimo}")
|
||||
|
||||
|
||||
class SearxNGBackend:
|
||||
"""SearxNG autoalojado: la opción estable y sin depender de terceros.
|
||||
|
||||
Es el destino natural cuando el endpoint lite empiece a fallar o cuando el
|
||||
volumen de consultas de la casa lo justifique. SearxNG puede agregar
|
||||
DuckDuckGo entre sus fuentes, así que el resultado es equivalente.
|
||||
"""
|
||||
|
||||
name = "searxng"
|
||||
|
||||
def __init__(self, base_url: str, timeout: float = 10.0, language: str = "es") -> None:
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.timeout = timeout
|
||||
self.language = language
|
||||
|
||||
def search(self, query: str, max_results: int) -> list[SearchResult]:
|
||||
url = f"{self.base_url}/search?" + urllib.parse.urlencode(
|
||||
{"q": query, "format": "json", "language": self.language}
|
||||
)
|
||||
request = urllib.request.Request(url, headers={"User-Agent": _USER_AGENT})
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=self.timeout) as respuesta:
|
||||
payload = json.loads(respuesta.read())
|
||||
except (urllib.error.URLError, TimeoutError, OSError, json.JSONDecodeError) as exc:
|
||||
raise SearchError(f"SearxNG en {self.base_url} no respondió: {exc}") from None
|
||||
|
||||
return [
|
||||
SearchResult(
|
||||
title=_clean(item.get("title", "")),
|
||||
url=item.get("url", ""),
|
||||
snippet=_clean(item.get("content", "")),
|
||||
)
|
||||
for item in payload.get("results", [])[:max_results]
|
||||
]
|
||||
|
||||
|
||||
class WebSearch:
|
||||
"""Tool de búsqueda: backend + presupuesto de contexto + caché.
|
||||
|
||||
La caché no es una optimización de rendimiento sino de comportamiento: en
|
||||
una casa las mismas preguntas se repiten muchas veces por día, y repetir la
|
||||
consulta externa no aporta nada. La versión persistente vive después en la
|
||||
memoria episódica; esta es en memoria y por proceso.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
backend: SearchBackend,
|
||||
max_results: int = 5,
|
||||
snippet_chars: int = 240,
|
||||
cache_ttl: float = 300.0,
|
||||
) -> None:
|
||||
self.backend = backend
|
||||
self.max_results = max_results
|
||||
self.snippet_chars = snippet_chars
|
||||
self.cache_ttl = cache_ttl
|
||||
self._cache: dict[str, tuple[float, list[SearchResult]]] = {}
|
||||
|
||||
def search(self, query: str) -> list[SearchResult]:
|
||||
query = query.strip()
|
||||
if not query:
|
||||
raise SearchError("la consulta está vacía")
|
||||
|
||||
ahora = time.monotonic()
|
||||
entrada = self._cache.get(query)
|
||||
if entrada is not None and ahora - entrada[0] < self.cache_ttl:
|
||||
return entrada[1]
|
||||
|
||||
resultados = self.backend.search(query, self.max_results)
|
||||
self._cache[query] = (ahora, resultados)
|
||||
return resultados
|
||||
|
||||
def render(self, query: str) -> str:
|
||||
"""Bloque de texto listo para inyectar como resultado de la tool.
|
||||
|
||||
Se numera y se recorta: el modelo tiene que poder citar "el segundo
|
||||
resultado" y el bloque entero tiene que entrar en el presupuesto de
|
||||
contexto sin desplazar la memoria ni el turno del usuario.
|
||||
"""
|
||||
resultados = self.search(query)
|
||||
if not resultados:
|
||||
return "Sin resultados."
|
||||
return "\n".join(
|
||||
f"{i}. {r.render(self.snippet_chars)}" for i, r in enumerate(resultados, 1)
|
||||
)
|
||||
|
||||
|
||||
def build_backend(config) -> SearchBackend:
|
||||
"""Construye el backend declarado en la config del agente."""
|
||||
if config.backend == "duckduckgo":
|
||||
return DuckDuckGoBackend(
|
||||
region=config.region,
|
||||
timeout=config.timeout,
|
||||
retries=config.retries,
|
||||
)
|
||||
if config.backend == "searxng":
|
||||
if not config.searxng_url:
|
||||
raise SearchError(
|
||||
"backend searxng sin searxng_url: definí ENLACE_SEARXNG_URL en .env"
|
||||
)
|
||||
return SearxNGBackend(
|
||||
base_url=config.searxng_url, timeout=config.timeout, language=config.language
|
||||
)
|
||||
raise SearchError(f"backend de búsqueda desconocido: {config.backend}")
|
||||
|
||||
|
||||
def build_search(config) -> WebSearch:
|
||||
"""Tool de búsqueda completa a partir de la config del agente."""
|
||||
return WebSearch(
|
||||
backend=build_backend(config),
|
||||
max_results=config.max_results,
|
||||
snippet_chars=config.snippet_chars,
|
||||
cache_ttl=config.cache_ttl,
|
||||
)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""`python -m enlace.agent.tools.search "<consulta>"`
|
||||
|
||||
Sirve para verificar la búsqueda sin el modelo: es la única parte del agente
|
||||
que se puede probar de punta a punta antes de que exista el modelo propio.
|
||||
"""
|
||||
import sys
|
||||
|
||||
from enlace.config.load import ConfigError, load_agent_config
|
||||
|
||||
consulta = " ".join(sys.argv[1:]).strip()
|
||||
if not consulta:
|
||||
print('uso: python -m enlace.agent.tools.search "<consulta>"', file=sys.stderr)
|
||||
return 2
|
||||
try:
|
||||
cfg = load_agent_config()
|
||||
tool = build_search(cfg.search)
|
||||
print(f"[{tool.backend.name}] {consulta}\n")
|
||||
print(tool.render(consulta))
|
||||
except (SearchError, ConfigError) as exc:
|
||||
print(f"[enlace] {exc}", file=sys.stderr)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user