msaldain
59b7c3acb6
El CI, apenas empezó a ejecutarse de verdad, hizo fallar dos de sus tres trabajos. La causa es un defecto de diseño, no del CI: la config del agente declara a Brave como respaldo, y la validación exigía la credencial para poder *leer* el archivo. Como la config vive en el repositorio y la credencial no, un clon limpio quedaba sin poder cargar su propia configuración. Mis pruebas en Gigastar no lo veían porque acá el archivo .env existe. Ahora se distingue por rol, que es lo que corresponde: - El primario sin credencial sigue siendo error fatal. Sin él no queda ninguna búsqueda en pie, y descubrirlo en la primera consulta real es tarde. - Un respaldo sin credencial se cae de la cadena y el primario sigue andando. Degradarse es la respuesta correcta: tener red de seguridad es mejor que no tenerla, pero no tenerla es mejor que no arrancar. Omitir no es esconder. La propiedad respaldos_omitidos deja el motivo a la vista, y build_backend emite un aviso al construir la cadena, que es el punto por el que pasa cualquier entrypoint que use búsqueda. Verificado escondiendo .env y corriendo la suite y el validador como lo haría un clon limpio: 124 tests en verde con credencial y sin ella. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
431 lines
16 KiB
Python
431 lines
16 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
|
|
import warnings
|
|
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.
|
|
|
|
Los respaldos sin credencial ya vienen filtrados de la config, pero se avisa
|
|
acá: quedarse sin red de seguridad no puede pasar en silencio, y este es el
|
|
punto por el que pasa cualquier entrypoint que use búsqueda.
|
|
"""
|
|
for nombre, variable in config.respaldos_omitidos:
|
|
warnings.warn(
|
|
f"búsqueda: el respaldo '{nombre}' queda deshabilitado porque falta "
|
|
f"{variable}. Si el primario '{config.backend}' falla, no hay red.",
|
|
RuntimeWarning,
|
|
stacklevel=2,
|
|
)
|
|
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())
|