mi222eh
461 lines
18 KiB
Python
461 lines
18 KiB
Python
"""Research tools for the coding agent: web search and URL fetching.
|
||
|
||
Search backend priority (web_search):
|
||
1. SearXNG — self-hosted at SEARXNG_URL (default: https://searxng.meeks.freeddns.org)
|
||
2. DDGS — duckduckgo-search library, with exponential-backoff retry
|
||
3. httpx — raw DuckDuckGo HTML scrape (zero-dep last resort)
|
||
|
||
Extraction pipeline (fetch_url):
|
||
1. trafilatura — state-of-the-art boilerplate removal, Markdown output
|
||
2. readability-lxml + markdownify — Mozilla Readability port, fallback
|
||
3. markdownify on full HTML — last resort if readability fails
|
||
4. regex strip — zero-dep absolute last resort
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import time
|
||
from typing import Any
|
||
|
||
import httpx
|
||
from duckduckgo_search import DDGS # type: ignore[import-untyped]
|
||
from duckduckgo_search.exceptions import ( # type: ignore[import-untyped]
|
||
DuckDuckGoSearchException,
|
||
RatelimitException,
|
||
)
|
||
|
||
from gitea.config import SEARXNG_URL, SEARXNG_USERNAME, SEARXNG_PASSWORD
|
||
|
||
logger: logging.Logger = logging.getLogger("research-tools")
|
||
|
||
_DEFAULT_TIMEOUT: int = 20
|
||
_MAX_CONTENT_CHARS: int = 20_000
|
||
_USER_AGENT: str = (
|
||
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
|
||
"(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
|
||
)
|
||
# SearXNG instance
|
||
_SEARXNG_URL: str = SEARXNG_URL
|
||
_SEARXNG_USERNAME: str = SEARXNG_USERNAME
|
||
_SEARXNG_PASSWORD: str = SEARXNG_PASSWORD
|
||
|
||
|
||
|
||
|
||
class ResearchTools:
|
||
"""Web search and URL fetching tools for the coding agent."""
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# Internal helpers
|
||
# ------------------------------------------------------------------ #
|
||
|
||
def _smart_truncate(
|
||
self,
|
||
text: str,
|
||
max_chars: int = _MAX_CONTENT_CHARS,
|
||
char_offset: int = 0,
|
||
) -> str:
|
||
"""Slice [char_offset : char_offset+max_chars] and truncate at a paragraph boundary.
|
||
|
||
Prefers cutting at a blank-line paragraph boundary rather than mid-sentence
|
||
so the LLM receives a coherent chunk.
|
||
"""
|
||
total: int = len(text)
|
||
chunk: str = text[char_offset : char_offset + max_chars]
|
||
if char_offset == 0 and len(chunk) <= max_chars and total <= max_chars:
|
||
return text # Common fast-path: content fits entirely
|
||
if len(chunk) >= max_chars:
|
||
last_para: int = chunk.rfind("\n\n")
|
||
if last_para > int(max_chars * 0.7):
|
||
chunk = chunk[:last_para]
|
||
next_offset: int = char_offset + len(chunk)
|
||
if next_offset < total:
|
||
tail = (
|
||
f"\n\n[Content truncated — {total} chars total. "
|
||
f"Showing chars {char_offset}–{next_offset}. "
|
||
f"Re-call fetch_url with char_offset={next_offset} to read more.]"
|
||
)
|
||
return chunk + tail
|
||
return chunk
|
||
|
||
def _html_to_markdown(self, html: str) -> str:
|
||
"""Convert HTML to clean Markdown.
|
||
|
||
Fallback chain:
|
||
1. trafilatura (best content extractor — removes nav/ads/sidebars)
|
||
2. readability-lxml + markdownify (Mozilla Readability port)
|
||
3. markdownify on full HTML
|
||
4. regex strip (zero-dep last resort)
|
||
"""
|
||
# --- 1. trafilatura ---
|
||
try:
|
||
import trafilatura # type: ignore[import-untyped]
|
||
|
||
extracted: str | None = trafilatura.extract(
|
||
html,
|
||
output_format="markdown",
|
||
include_tables=True,
|
||
include_comments=False,
|
||
favor_precision=True,
|
||
deduplicate=True,
|
||
)
|
||
if extracted and len(extracted) > 200:
|
||
return re.sub(r"\n{3,}", "\n\n", extracted).strip()
|
||
except ImportError:
|
||
logger.debug("trafilatura not installed; falling back to readability")
|
||
except Exception as exc:
|
||
logger.debug(f"trafilatura extraction failed: {exc}")
|
||
|
||
# --- 2. readability-lxml + markdownify ---
|
||
try:
|
||
from readability import Document # type: ignore[import-untyped]
|
||
from markdownify import markdownify as md # type: ignore[import-untyped]
|
||
|
||
doc = Document(html)
|
||
clean_html: str = doc.summary()
|
||
text: str = md(clean_html, strip=["script", "style"])
|
||
text = re.sub(r"\n{3,}", "\n\n", text)
|
||
if text.strip() and len(text.strip()) > 100:
|
||
return text.strip()
|
||
except ImportError:
|
||
logger.debug("readability-lxml or markdownify not installed")
|
||
except Exception as exc:
|
||
logger.debug(f"readability+markdownify extraction failed: {exc}")
|
||
|
||
# --- 3. markdownify on full HTML ---
|
||
try:
|
||
from markdownify import markdownify as md # type: ignore[import-untyped]
|
||
|
||
text = md(html, strip=["script", "style", "nav", "footer", "aside"])
|
||
text = re.sub(r"\n{3,}", "\n\n", text)
|
||
if text.strip():
|
||
return text.strip()
|
||
except ImportError:
|
||
logger.debug("markdownify not installed; using regex fallback")
|
||
except Exception as exc:
|
||
logger.debug(f"markdownify failed: {exc}")
|
||
|
||
# --- 4. Regex strip (zero-dep fallback) ---
|
||
html = re.sub(
|
||
r"<(script|style)[^>]*?>.*?</\1>", "", html,
|
||
flags=re.DOTALL | re.IGNORECASE,
|
||
)
|
||
text = re.sub(r"<[^>]+>", " ", html)
|
||
text = re.sub(r"[ \t]+", " ", text)
|
||
text = re.sub(r"\n{3,}", "\n\n", text)
|
||
entities: dict[str, str] = {
|
||
"&": "&", "<": "<", ">": ">",
|
||
""": '"', "'": "'", " ": " ",
|
||
"—": "—", "–": "–", "…": "…",
|
||
}
|
||
for entity, char in entities.items():
|
||
text = text.replace(entity, char)
|
||
return text.strip()
|
||
|
||
def _format_results(
|
||
self, results: list[dict[str, str]], query: str
|
||
) -> str:
|
||
"""Format search results as a numbered list for LLM consumption."""
|
||
lines: list[str] = [f"Search results for: '{query}'\n"]
|
||
for i, r in enumerate(results, 1):
|
||
title: str = r.get("title", "No title")
|
||
url: str = r.get("href") or r.get("url", "")
|
||
snippet: str = (
|
||
r.get("body") or r.get("snippet") or r.get("content", "")
|
||
)[:250]
|
||
date: str = r.get("published_date", "")
|
||
date_str: str = f"\n Date: {date}" if date else ""
|
||
lines.append(
|
||
f"[{i}] {title}\n"
|
||
f" URL: {url}\n"
|
||
f" {snippet}{date_str}\n"
|
||
)
|
||
return "\n".join(lines)
|
||
|
||
def _search_searxng(
|
||
self,
|
||
query: str,
|
||
num_results: int,
|
||
time_range: str = "",
|
||
categories: str = "general",
|
||
language: str = "en",
|
||
) -> str | None:
|
||
"""Search via self-hosted SearXNG JSON API.
|
||
|
||
Returns formatted results string on success, or None if the instance
|
||
is unreachable so the caller can fall through to the next backend.
|
||
"""
|
||
if not _SEARXNG_URL:
|
||
logger.info("SearXNG URL is not configured; skipping SearXNG search.")
|
||
return None
|
||
params: dict[str, Any] = {
|
||
"q": query,
|
||
"format": "json",
|
||
"language": language,
|
||
"categories": categories,
|
||
}
|
||
if time_range:
|
||
params["time_range"] = time_range
|
||
|
||
auth = None
|
||
if _SEARXNG_USERNAME and _SEARXNG_PASSWORD:
|
||
auth = (_SEARXNG_USERNAME, _SEARXNG_PASSWORD)
|
||
|
||
try:
|
||
with httpx.Client(
|
||
timeout=_DEFAULT_TIMEOUT, follow_redirects=True, auth=auth
|
||
) as client:
|
||
response = client.get(
|
||
f"{_SEARXNG_URL}/search",
|
||
params=params,
|
||
headers={"User-Agent": _USER_AGENT},
|
||
)
|
||
response.raise_for_status()
|
||
data: dict[str, Any] = response.json()
|
||
except Exception as exc:
|
||
logger.warning(f"SearXNG unavailable ({_SEARXNG_URL}): {exc}")
|
||
return None
|
||
|
||
raw_results: list[dict[str, Any]] = data.get("results", [])
|
||
if not raw_results:
|
||
return None # Let caller fall through to next backend
|
||
|
||
# Normalise SearXNG fields to our standard format
|
||
normalised: list[dict[str, str]] = [
|
||
{
|
||
"title": r.get("title", "No title"),
|
||
"href": r.get("url", ""),
|
||
"body": r.get("content", "")[:250],
|
||
"published_date": r.get("publishedDate", ""),
|
||
}
|
||
for r in raw_results[:num_results]
|
||
]
|
||
return self._format_results(normalised, query)
|
||
|
||
def _web_search_fallback(self, query: str, num_results: int) -> str:
|
||
"""DuckDuckGo HTML scraping fallback when duckduckgo-search is not installed."""
|
||
try:
|
||
headers: dict[str, str] = {
|
||
"User-Agent": _USER_AGENT,
|
||
"Accept-Language": "en-US,en;q=0.9",
|
||
}
|
||
with httpx.Client(timeout=_DEFAULT_TIMEOUT, follow_redirects=True) as client:
|
||
response = client.get(
|
||
"https://html.duckduckgo.com/html/",
|
||
params={"q": query, "kl": "us-en"},
|
||
headers=headers,
|
||
)
|
||
response.raise_for_status()
|
||
|
||
blocks = re.findall(
|
||
r'<a[^>]+class="result__a"[^>]+href="([^"]+)"[^>]*>(.*?)</a>'
|
||
r'.*?<a[^>]+class="result__snippet"[^>]*>(.*?)</a>',
|
||
response.text,
|
||
flags=re.DOTALL,
|
||
)
|
||
if not blocks:
|
||
return (
|
||
f"No results found for '{query}'. "
|
||
"Try rephrasing or use fetch_url with a known documentation URL."
|
||
)
|
||
lines: list[str] = [f"Search results for: '{query}'\n"]
|
||
for i, (url, title, snippet) in enumerate(blocks[:num_results], 1):
|
||
clean_title = re.sub(r"<[^>]+>", "", title).strip()
|
||
clean_snippet = re.sub(r"<[^>]+>", "", snippet).strip()
|
||
lines.append(f"[{i}] {clean_title}\n URL: {url}\n {clean_snippet}\n")
|
||
return "\n".join(lines)
|
||
except Exception as exc:
|
||
return f"Search error: {exc}"
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# Public tools
|
||
# ------------------------------------------------------------------ #
|
||
|
||
def web_search(
|
||
self,
|
||
query: str,
|
||
num_results: int = 8,
|
||
time_range: str = "",
|
||
categories: str = "general",
|
||
language: str = "en",
|
||
) -> str:
|
||
"""Search the web and return structured, numbered results.
|
||
|
||
Uses a self-hosted SearXNG instance as primary backend (private,
|
||
no rate limits, aggregates Google/Bing/Wikipedia/etc.), falling back
|
||
to DuckDuckGo (DDGS) if SearXNG is unreachable.
|
||
|
||
Use this tool when you need to:
|
||
- Find documentation for a library, framework, or API
|
||
- Look up error messages, stack traces, or known bugs
|
||
- Discover best practices, community conventions, or coding patterns
|
||
- Find package release notes, changelogs, or migration guides
|
||
- Research a technology, tool, or concept you are unfamiliar with
|
||
|
||
Do NOT use this tool if you already have the exact URL — use fetch_url instead.
|
||
Result URLs can be passed directly to fetch_url for full page content.
|
||
|
||
Args:
|
||
query: Specific natural-language or technical search query.
|
||
Good: "Python httpx async retry on timeout 2024"
|
||
Bad: "httpx"
|
||
num_results: Results to return (1–20, default 8). 5–10 is optimal.
|
||
time_range: Optional recency filter — "day", "month", or "year".
|
||
categories: SearXNG category — "general" (default), "it", "news",
|
||
"science", "files", "videos", "music".
|
||
language: ISO language code, e.g. "en" (default), "sv", "de".
|
||
|
||
Returns:
|
||
Numbered list [1], [2], ... each with title, URL, snippet, and date.
|
||
On failure, returns an actionable error string — use fetch_url as fallback.
|
||
"""
|
||
num_results = min(max(1, num_results), 20)
|
||
|
||
# --- Tier 1: SearXNG (self-hosted, preferred) ---
|
||
searxng_result = self._search_searxng(
|
||
query, num_results, time_range=time_range,
|
||
categories=categories, language=language,
|
||
)
|
||
if searxng_result is not None:
|
||
return searxng_result
|
||
logger.info("SearXNG unavailable; falling back to DuckDuckGo")
|
||
|
||
# --- Tier 2: DDGS library (handles sessions, cookies, rate limits) ---
|
||
last_exc: Exception | None = None
|
||
for attempt in range(3):
|
||
try:
|
||
with DDGS(timeout=_DEFAULT_TIMEOUT) as ddgs:
|
||
results: list[dict[str, str]] = list(
|
||
ddgs.text(query, max_results=num_results, region="us-en")
|
||
)
|
||
if not results:
|
||
return (
|
||
f"No results found for '{query}'. "
|
||
"Try rephrasing, or use fetch_url with a known documentation URL."
|
||
)
|
||
return self._format_results(results, query)
|
||
|
||
except RatelimitException as exc:
|
||
last_exc = exc
|
||
wait: int = 2 ** attempt # 1 s → 2 s → 4 s
|
||
logger.warning(
|
||
f"DuckDuckGo rate limit (attempt {attempt + 1}/3), "
|
||
f"retrying in {wait}s…"
|
||
)
|
||
time.sleep(wait)
|
||
|
||
except DuckDuckGoSearchException as exc:
|
||
return (
|
||
f"Search unavailable: {exc}. "
|
||
"Try fetch_url with a direct documentation URL instead."
|
||
)
|
||
|
||
except Exception as exc:
|
||
logger.warning(f"web_search error: {exc}")
|
||
return f"Search error: {exc}"
|
||
|
||
return (
|
||
f"DuckDuckGo rate limit exceeded after 3 retries ({last_exc}). "
|
||
"Wait a moment and retry, or use fetch_url with a known URL."
|
||
)
|
||
|
||
def fetch_url(
|
||
self,
|
||
url: str,
|
||
extract_text: bool = True,
|
||
max_chars: int = _MAX_CONTENT_CHARS,
|
||
char_offset: int = 0,
|
||
) -> str:
|
||
"""Fetch the content of a URL and return it as clean, readable Markdown.
|
||
|
||
Use this tool when you need to:
|
||
- Read the full content of a documentation page, API reference, or README
|
||
- Follow up on a URL returned by web_search to get the complete text
|
||
- Read a GitHub issue, Stack Overflow answer, or blog post in full
|
||
- Access a package's changelog, migration guide, or specification
|
||
- Read a JSON API response, config schema, or data format at a known URL
|
||
|
||
Do NOT use this tool for binary files (images, PDFs, executables).
|
||
Note: pages that require JavaScript to render may return incomplete content.
|
||
For JS-heavy pages, prefer web_search first to find a cached/static mirror.
|
||
|
||
Args:
|
||
url: The full URL to fetch (must start with http:// or https://).
|
||
extract_text: If True (default), extract and clean the main content
|
||
as Markdown, stripping navigation, ads, and boilerplate.
|
||
Set to False to get raw HTML/JSON (useful for schemas).
|
||
max_chars: Maximum characters to return per call (default 20,000).
|
||
The tool cuts at a paragraph boundary when truncating.
|
||
char_offset: Character offset into the extracted content to start
|
||
reading from (default 0). Increment by max_chars to
|
||
page through content larger than max_chars.
|
||
|
||
Returns:
|
||
Clean Markdown text of the main content (HTML pages), pretty-printed
|
||
JSON (JSON responses), or plain text (text/plain, .md, .txt).
|
||
Returns an error string on HTTP errors, timeouts, or invalid URLs.
|
||
"""
|
||
if not url.startswith(("http://", "https://")):
|
||
return f"Invalid URL '{url}': must start with http:// or https://"
|
||
|
||
try:
|
||
headers: dict[str, str] = {
|
||
"User-Agent": _USER_AGENT,
|
||
"Accept": (
|
||
"text/html,application/xhtml+xml,application/xml;"
|
||
"q=0.9,application/json,*/*;q=0.8"
|
||
),
|
||
"Accept-Language": "en-US,en;q=0.9",
|
||
}
|
||
with httpx.Client(
|
||
timeout=_DEFAULT_TIMEOUT, follow_redirects=True
|
||
) as client:
|
||
response = client.get(url, headers=headers)
|
||
response.raise_for_status()
|
||
|
||
content_type: str = response.headers.get("content-type", "")
|
||
raw: str = response.text
|
||
text: str
|
||
|
||
# JSON → pretty print
|
||
if "application/json" in content_type:
|
||
try:
|
||
data: Any = response.json()
|
||
text = json.dumps(data, indent=2)
|
||
except Exception:
|
||
text = raw
|
||
|
||
# Plain text / Markdown / reStructuredText → return as-is
|
||
elif "text/plain" in content_type or url.endswith(
|
||
(".md", ".txt", ".rst")
|
||
):
|
||
text = raw
|
||
|
||
# HTML → extract main content as Markdown
|
||
elif extract_text and (
|
||
"text/html" in content_type
|
||
or raw.lstrip().startswith(("<html", "<!DOCTYPE", "<!doctype"))
|
||
):
|
||
text = self._html_to_markdown(raw)
|
||
|
||
else:
|
||
text = raw
|
||
|
||
return self._smart_truncate(text, max_chars, char_offset)
|
||
|
||
except httpx.HTTPStatusError as exc:
|
||
return (
|
||
f"Failed to fetch '{url}' "
|
||
f"(HTTP {exc.response.status_code}): {exc}"
|
||
)
|
||
except httpx.TimeoutException:
|
||
return (
|
||
f"Request to '{url}' timed out after {_DEFAULT_TIMEOUT}s. "
|
||
"Try a different URL or break the page into smaller fetches."
|
||
)
|
||
except Exception as exc:
|
||
logger.warning(f"fetch_url error for '{url}': {exc}")
|
||
return f"Error fetching '{url}': {exc}"
|