msaldain
4048936067
ruff estaba configurado en pyproject.toml desde el primer commit y jamás se había corrido. Tenía 15 hallazgos. Dos importan más allá del estilo: - zip() sin strict= trunca en silencio al más corto. En las comparaciones de lotes eso significa que un test podía pasar sin haber comparado todo. Donde los largos deben coincidir ahora es strict=True; donde difieren a propósito (pares consecutivos) queda strict=False, que documenta la intención. - Un import sin usar delataba algo peor: BraveBackend se había escrito sin una sola prueba. Se agregan ocho, contra una respuesta con la forma que devuelve la API, incluidas la limpieza de etiquetas, el caso de límite de tasa —que tiene que distinguirse de 'no respondió'— y que la credencial viaje en la cabecera y nunca en la URL. El respaldo tiene que funcionar justo cuando el primario ya falló; merecía la misma cobertura. El resto es orden de imports, collections.abc y líneas largas.
419 lines
15 KiB
Python
419 lines
15 KiB
Python
"""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 BraveBackend:
|
|
"""Brave Search API. Una API con contrato, a diferencia del endpoint lite.
|
|
|
|
Existe como red de seguridad del primario: cuando DuckDuckGo limite por tasa
|
|
o cambie el HTML, esto responde. La contrapartida es que cada consulta pasa
|
|
por un tercero asociado a una cuenta, así que no se usa como primario salvo
|
|
que se elija explícitamente.
|
|
"""
|
|
|
|
name = "brave"
|
|
ENDPOINT = "https://api.search.brave.com/res/v1/web/search"
|
|
|
|
def __init__(
|
|
self, api_key: str, timeout: float = 10.0, country: str = "uy", language: str = "es"
|
|
) -> None:
|
|
self.api_key = api_key
|
|
self.timeout = timeout
|
|
self.country = country
|
|
self.language = language
|
|
|
|
def search(self, query: str, max_results: int) -> list[SearchResult]:
|
|
url = (
|
|
self.ENDPOINT
|
|
+ "?"
|
|
+ urllib.parse.urlencode(
|
|
{
|
|
"q": query,
|
|
"count": max_results,
|
|
"country": self.country,
|
|
"search_lang": self.language,
|
|
}
|
|
)
|
|
)
|
|
request = urllib.request.Request(
|
|
url,
|
|
headers={
|
|
"Accept": "application/json",
|
|
"X-Subscription-Token": self.api_key,
|
|
"User-Agent": _USER_AGENT,
|
|
},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=self.timeout) as respuesta:
|
|
payload = json.loads(respuesta.read())
|
|
except urllib.error.HTTPError as exc:
|
|
# 429 es el caso interesante: si el fallback también está limitado,
|
|
# el mensaje tiene que decirlo en vez de parecer un fallo genérico.
|
|
detalle = "límite de tasa alcanzado" if exc.code == 429 else f"HTTP {exc.code}"
|
|
raise SearchError(f"Brave rechazó la consulta: {detalle}") from None
|
|
except (urllib.error.URLError, TimeoutError, OSError, json.JSONDecodeError) as exc:
|
|
raise SearchError(f"Brave no respondió: {exc}") from None
|
|
|
|
resultados = payload.get("web", {}).get("results", [])
|
|
return [
|
|
SearchResult(
|
|
title=_clean(item.get("title", "")),
|
|
url=item.get("url", ""),
|
|
snippet=_clean(item.get("description", "")),
|
|
)
|
|
for item in resultados[:max_results]
|
|
if item.get("url", "").startswith("http")
|
|
]
|
|
|
|
|
|
class SearxNGBackend:
|
|
"""SearxNG autoalojado: la opción estable y sin depender de terceros.
|
|
|
|
Es el destino natural cuando el volumen de consultas de la casa lo
|
|
justifique. SearxNG puede agregar DuckDuckGo entre sus fuentes, así que el
|
|
resultado es equivalente sin exponer las consultas a un tercero.
|
|
"""
|
|
|
|
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 CadenaDeBackends:
|
|
"""Prueba los backends en orden y devuelve el primero que responda.
|
|
|
|
La razón de existir: el endpoint lite de DuckDuckGo no es una API con
|
|
contrato — puede limitar por tasa o cambiar el HTML sin aviso. Con esto un
|
|
fallo deja de ser una respuesta vacía para la familia y pasa a ser un
|
|
reintento contra otro proveedor.
|
|
|
|
Dos matices que importan:
|
|
|
|
- **Cero resultados no es un fallo.** Si el primario responde bien y no
|
|
encontró nada, esa es la respuesta correcta; encadenar al siguiente
|
|
proveedor por eso gastaría cuota y devolvería resultados peores.
|
|
Solo se avanza ante un error real.
|
|
- **Se recuerda quién respondió** (`ultimo_backend`), porque al depurar
|
|
una respuesta rara lo primero que hay que saber es de dónde salió.
|
|
"""
|
|
|
|
def __init__(self, backends: list[SearchBackend]) -> None:
|
|
if not backends:
|
|
raise SearchError("la cadena de búsqueda no tiene ningún backend")
|
|
self.backends = backends
|
|
self.ultimo_backend: str | None = None
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return " -> ".join(getattr(b, "name", type(b).__name__) for b in self.backends)
|
|
|
|
def search(self, query: str, max_results: int) -> list[SearchResult]:
|
|
fallos: list[str] = []
|
|
for backend in self.backends:
|
|
nombre = getattr(backend, "name", type(backend).__name__)
|
|
try:
|
|
resultados = backend.search(query, max_results)
|
|
except SearchError as exc:
|
|
fallos.append(f"{nombre}: {exc}")
|
|
continue
|
|
self.ultimo_backend = nombre
|
|
return resultados
|
|
|
|
raise SearchError("ningún backend de búsqueda respondió — " + " | ".join(fallos))
|
|
|
|
|
|
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 _construir_uno(nombre: str, config) -> SearchBackend:
|
|
if nombre == "duckduckgo":
|
|
return DuckDuckGoBackend(
|
|
region=config.region, timeout=config.timeout, retries=config.retries
|
|
)
|
|
if nombre == "brave":
|
|
if not config.brave_api_key:
|
|
raise SearchError(
|
|
"backend brave sin brave_api_key: definí ENLACE_BRAVE_API_KEY en .env"
|
|
)
|
|
return BraveBackend(
|
|
api_key=config.brave_api_key,
|
|
timeout=config.timeout,
|
|
country=config.country,
|
|
language=config.language,
|
|
)
|
|
if nombre == "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: {nombre}")
|
|
|
|
|
|
def build_backend(config) -> SearchBackend:
|
|
"""Construye la cadena declarada en la config: primario + respaldos.
|
|
|
|
Con un solo backend devuelve ese backend pelado, para no envolver en una
|
|
cadena algo que no la necesita.
|
|
"""
|
|
cadena = [_construir_uno(nombre, config) for nombre in config.cadena]
|
|
return cadena[0] if len(cadena) == 1 else CadenaDeBackends(cadena)
|
|
|
|
|
|
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())
|