Add Brave LLM Context endpoint as opt-in web search mode
Brave's /res/v1/llm/context returns pre-extracted text chunks optimized for LLM consumption instead of URLs + short snippets. Enable with BRAVE_LLM_CONTEXT=1 env var; same API key and pricing. - Add _search_llm_context() and _normalize_llm_context() to brave_search.py - Wire opt-in flag through _search_web() in last30days.py - Update module docstring (free tier eliminated Feb 2026) - Add 23 tests covering normalization, filtering, date parsing
This commit is contained in:
Jeffrey Sperling
@@ -632,8 +632,10 @@ def _search_web(
|
||||
topic, from_date, to_date, config["PARALLEL_API_KEY"], depth=depth,
|
||||
)
|
||||
elif backend == "brave":
|
||||
use_llm_ctx = os.environ.get("BRAVE_LLM_CONTEXT", "").strip() == "1"
|
||||
raw_results = brave_search.search_web(
|
||||
topic, from_date, to_date, config["BRAVE_API_KEY"], depth=depth,
|
||||
topic, from_date, to_date, config["BRAVE_API_KEY"],
|
||||
depth=depth, use_llm_context=use_llm_ctx,
|
||||
)
|
||||
elif backend == "openrouter":
|
||||
raw_results = openrouter_search.search_web(
|
||||
|
||||
+118
-2
@@ -1,7 +1,12 @@
|
||||
"""Brave Search web search for last30days skill.
|
||||
|
||||
Uses the Brave Search API as a fallback web search backend.
|
||||
Simple, cheap (free tier: 2,000 queries/month), widely available.
|
||||
Uses the Brave Search API as a web search backend.
|
||||
Requires a paid Brave Search subscription (free tier eliminated Feb 2026).
|
||||
|
||||
Two modes:
|
||||
- Standard: /res/v1/web/search — returns URLs + snippets (default)
|
||||
- LLM Context: /res/v1/llm/context — returns pre-extracted text chunks
|
||||
optimized for LLM consumption. Enable with BRAVE_LLM_CONTEXT=1 env var.
|
||||
|
||||
API docs: https://api-dashboard.search.brave.com/app/documentation/web-search/get-started
|
||||
"""
|
||||
@@ -16,6 +21,7 @@ from urllib.parse import urlencode, urlparse
|
||||
from . import http
|
||||
|
||||
ENDPOINT = "https://api.search.brave.com/res/v1/web/search"
|
||||
LLM_CONTEXT_ENDPOINT = "https://api.search.brave.com/res/v1/llm/context"
|
||||
|
||||
# Freshness codes: pd=24h, pw=7d, pm=31d
|
||||
FRESHNESS_MAP = {1: "pd", 7: "pw", 31: "pm"}
|
||||
@@ -33,6 +39,7 @@ def search_web(
|
||||
to_date: str,
|
||||
api_key: str,
|
||||
depth: str = "default",
|
||||
use_llm_context: bool = False,
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""Search the web via Brave Search API.
|
||||
|
||||
@@ -42,6 +49,7 @@ def search_web(
|
||||
to_date: End date (YYYY-MM-DD)
|
||||
api_key: Brave Search API key
|
||||
depth: 'quick', 'default', or 'deep'
|
||||
use_llm_context: Use LLM Context endpoint for pre-extracted content
|
||||
|
||||
Returns:
|
||||
List of result dicts with keys: url, title, snippet, source_domain, date, relevance
|
||||
@@ -49,6 +57,9 @@ def search_web(
|
||||
Raises:
|
||||
http.HTTPError: On API errors
|
||||
"""
|
||||
if use_llm_context:
|
||||
return _search_llm_context(topic, from_date, to_date, api_key, depth)
|
||||
|
||||
count = {"quick": 8, "default": 15, "deep": 25}.get(depth, 15)
|
||||
|
||||
# Calculate days for freshness filter
|
||||
@@ -81,6 +92,48 @@ def search_web(
|
||||
return _normalize_results(response, from_date, to_date)
|
||||
|
||||
|
||||
def _search_llm_context(
|
||||
topic: str,
|
||||
from_date: str,
|
||||
to_date: str,
|
||||
api_key: str,
|
||||
depth: str = "default",
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""Search via Brave LLM Context endpoint for pre-extracted web content.
|
||||
|
||||
Returns results in the same schema as search_web() for downstream compatibility.
|
||||
Snippets contain actual page content instead of short descriptions.
|
||||
"""
|
||||
count = {"quick": 5, "default": 20, "deep": 50}.get(depth, 20)
|
||||
max_tokens = {"quick": 2048, "default": 8192, "deep": 16384}.get(depth, 8192)
|
||||
|
||||
days = _days_between(from_date, to_date)
|
||||
freshness = _brave_freshness(days)
|
||||
|
||||
params = {
|
||||
"q": topic,
|
||||
"count": count,
|
||||
"maximum_number_of_tokens": max_tokens,
|
||||
"context_threshold_mode": "balanced",
|
||||
}
|
||||
if freshness:
|
||||
params["freshness"] = freshness
|
||||
|
||||
url = f"{LLM_CONTEXT_ENDPOINT}?{urlencode(params)}"
|
||||
|
||||
sys.stderr.write(f"[Web] Searching Brave LLM Context for: {topic}\n")
|
||||
sys.stderr.flush()
|
||||
|
||||
response = http.request(
|
||||
"GET",
|
||||
url,
|
||||
headers={"X-Subscription-Token": api_key},
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
return _normalize_llm_context(response)
|
||||
|
||||
|
||||
def _days_between(from_date: str, to_date: str) -> int:
|
||||
"""Calculate days between two YYYY-MM-DD dates."""
|
||||
try:
|
||||
@@ -169,6 +222,69 @@ def _normalize_results(
|
||||
return items
|
||||
|
||||
|
||||
def _normalize_llm_context(response: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||
"""Convert Brave LLM Context response to websearch item schema.
|
||||
|
||||
LLM Context returns grounding.generic[] with url, title, snippets[].
|
||||
Sources metadata provides hostname and age for each URL.
|
||||
"""
|
||||
items = []
|
||||
grounding = response.get("grounding", {})
|
||||
sources = response.get("sources", {})
|
||||
|
||||
for i, result in enumerate(grounding.get("generic", [])):
|
||||
if not isinstance(result, dict):
|
||||
continue
|
||||
|
||||
url = result.get("url", "")
|
||||
if not url:
|
||||
continue
|
||||
|
||||
# Skip excluded domains
|
||||
try:
|
||||
domain = urlparse(url).netloc.lower()
|
||||
if domain in EXCLUDED_DOMAINS:
|
||||
continue
|
||||
if domain.startswith("www."):
|
||||
domain = domain[4:]
|
||||
except Exception:
|
||||
domain = ""
|
||||
|
||||
title = str(result.get("title", "")).strip()
|
||||
snippets = result.get("snippets", [])
|
||||
snippet = "\n".join(str(s).strip() for s in snippets if s)
|
||||
|
||||
if not title and not snippet:
|
||||
continue
|
||||
|
||||
# Parse date from sources metadata
|
||||
source_meta = sources.get(url, {})
|
||||
age_list = source_meta.get("age") or []
|
||||
date = None
|
||||
for age_str in age_list:
|
||||
date = _parse_brave_date(age_str, None)
|
||||
if date:
|
||||
break
|
||||
date_confidence = "med" if date else "low"
|
||||
|
||||
items.append({
|
||||
"id": f"W{i+1}",
|
||||
"title": title[:200],
|
||||
"url": url,
|
||||
"source_domain": source_meta.get("hostname", domain),
|
||||
"snippet": snippet[:1500], # LLM Context returns richer content
|
||||
"date": date,
|
||||
"date_confidence": date_confidence,
|
||||
"relevance": 0.7, # LLM Context pre-filters for relevance
|
||||
"why_relevant": "",
|
||||
})
|
||||
|
||||
sys.stderr.write(f"[Web] Brave LLM Context: {len(items)} results\n")
|
||||
sys.stderr.flush()
|
||||
|
||||
return items
|
||||
|
||||
|
||||
def _clean_html(text: str) -> str:
|
||||
"""Remove HTML tags and decode entities."""
|
||||
text = re.sub(r"<[^>]*>", "", text)
|
||||
|
||||
@@ -0,0 +1,197 @@
|
||||
"""Tests for Brave Search module, including LLM Context endpoint."""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import unittest
|
||||
|
||||
# Ensure scripts/ is on path
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'scripts'))
|
||||
|
||||
from lib.brave_search import (
|
||||
_normalize_results,
|
||||
_normalize_llm_context,
|
||||
_days_between,
|
||||
_brave_freshness,
|
||||
_parse_brave_date,
|
||||
EXCLUDED_DOMAINS,
|
||||
)
|
||||
|
||||
|
||||
class TestDaysBetween(unittest.TestCase):
|
||||
def test_same_day(self):
|
||||
self.assertEqual(_days_between("2026-03-01", "2026-03-01"), 1)
|
||||
|
||||
def test_one_week(self):
|
||||
self.assertEqual(_days_between("2026-03-01", "2026-03-08"), 7)
|
||||
|
||||
def test_invalid_dates(self):
|
||||
self.assertEqual(_days_between("bad", "dates"), 30)
|
||||
|
||||
|
||||
class TestBraveFreshness(unittest.TestCase):
|
||||
def test_one_day(self):
|
||||
self.assertEqual(_brave_freshness(1), "pd")
|
||||
|
||||
def test_one_week(self):
|
||||
self.assertEqual(_brave_freshness(7), "pw")
|
||||
|
||||
def test_one_month(self):
|
||||
self.assertEqual(_brave_freshness(31), "pm")
|
||||
|
||||
def test_longer_returns_range(self):
|
||||
result = _brave_freshness(60)
|
||||
self.assertIn("to", result)
|
||||
|
||||
def test_none(self):
|
||||
self.assertIsNone(_brave_freshness(None))
|
||||
|
||||
|
||||
class TestParseBraveDate(unittest.TestCase):
|
||||
def test_hours_ago(self):
|
||||
result = _parse_brave_date("3 hours ago", None)
|
||||
self.assertIsNotNone(result)
|
||||
self.assertRegex(result, r"\d{4}-\d{2}-\d{2}")
|
||||
|
||||
def test_days_ago(self):
|
||||
result = _parse_brave_date("5 days ago", None)
|
||||
self.assertIsNotNone(result)
|
||||
|
||||
def test_weeks_ago(self):
|
||||
result = _parse_brave_date("2 weeks ago", None)
|
||||
self.assertIsNotNone(result)
|
||||
|
||||
def test_iso_date(self):
|
||||
self.assertEqual(_parse_brave_date("2026-03-10T12:00:00", None), "2026-03-10")
|
||||
|
||||
def test_none(self):
|
||||
self.assertIsNone(_parse_brave_date(None, None))
|
||||
|
||||
|
||||
class TestNormalizeResults(unittest.TestCase):
|
||||
def test_merges_news_and_web(self):
|
||||
response = {
|
||||
"news": {"results": [
|
||||
{"url": "https://news.example.com/a", "title": "News A", "description": "News desc"},
|
||||
]},
|
||||
"web": {"results": [
|
||||
{"url": "https://blog.example.com/b", "title": "Blog B", "description": "Blog desc"},
|
||||
]},
|
||||
}
|
||||
items = _normalize_results(response, "2026-03-01", "2026-03-10")
|
||||
self.assertEqual(len(items), 2)
|
||||
self.assertEqual(items[0]["title"], "News A")
|
||||
self.assertEqual(items[1]["title"], "Blog B")
|
||||
|
||||
def test_excludes_reddit_and_x(self):
|
||||
response = {
|
||||
"web": {"results": [
|
||||
{"url": "https://www.reddit.com/r/test/123", "title": "Reddit", "description": "text"},
|
||||
{"url": "https://x.com/user/status/1", "title": "X post", "description": "text"},
|
||||
{"url": "https://example.com/ok", "title": "OK", "description": "text"},
|
||||
]},
|
||||
}
|
||||
items = _normalize_results(response, "2026-03-01", "2026-03-10")
|
||||
self.assertEqual(len(items), 1)
|
||||
self.assertEqual(items[0]["title"], "OK")
|
||||
|
||||
def test_default_relevance(self):
|
||||
response = {"web": {"results": [
|
||||
{"url": "https://a.com", "title": "A", "description": "desc"},
|
||||
]}}
|
||||
items = _normalize_results(response, "2026-03-01", "2026-03-10")
|
||||
self.assertEqual(items[0]["relevance"], 0.6)
|
||||
|
||||
|
||||
class TestNormalizeLlmContext(unittest.TestCase):
|
||||
def _make_response(self, generic=None, sources=None):
|
||||
return {
|
||||
"grounding": {"generic": generic or []},
|
||||
"sources": sources or {},
|
||||
}
|
||||
|
||||
def test_basic_result(self):
|
||||
resp = self._make_response(
|
||||
generic=[{
|
||||
"url": "https://docs.example.com/page",
|
||||
"title": "Example Page",
|
||||
"snippets": ["First chunk of text.", "Second chunk of text."],
|
||||
}],
|
||||
sources={
|
||||
"https://docs.example.com/page": {
|
||||
"title": "Example Page",
|
||||
"hostname": "docs.example.com",
|
||||
"age": ["2026-03-05", "5 days ago"],
|
||||
}
|
||||
},
|
||||
)
|
||||
items = _normalize_llm_context(resp)
|
||||
self.assertEqual(len(items), 1)
|
||||
item = items[0]
|
||||
self.assertEqual(item["title"], "Example Page")
|
||||
self.assertEqual(item["url"], "https://docs.example.com/page")
|
||||
self.assertIn("First chunk", item["snippet"])
|
||||
self.assertIn("Second chunk", item["snippet"])
|
||||
self.assertEqual(item["date"], "2026-03-05")
|
||||
self.assertEqual(item["date_confidence"], "med")
|
||||
self.assertEqual(item["relevance"], 0.7)
|
||||
self.assertEqual(item["source_domain"], "docs.example.com")
|
||||
|
||||
def test_excludes_reddit(self):
|
||||
resp = self._make_response(
|
||||
generic=[
|
||||
{"url": "https://www.reddit.com/r/test", "title": "Reddit", "snippets": ["text"]},
|
||||
{"url": "https://example.com", "title": "OK", "snippets": ["text"]},
|
||||
],
|
||||
)
|
||||
items = _normalize_llm_context(resp)
|
||||
self.assertEqual(len(items), 1)
|
||||
self.assertEqual(items[0]["title"], "OK")
|
||||
|
||||
def test_empty_grounding(self):
|
||||
resp = self._make_response()
|
||||
items = _normalize_llm_context(resp)
|
||||
self.assertEqual(items, [])
|
||||
|
||||
def test_snippet_truncation(self):
|
||||
long_snippet = "x" * 2000
|
||||
resp = self._make_response(
|
||||
generic=[{"url": "https://a.com", "title": "A", "snippets": [long_snippet]}],
|
||||
)
|
||||
items = _normalize_llm_context(resp)
|
||||
self.assertLessEqual(len(items[0]["snippet"]), 1500)
|
||||
|
||||
def test_no_date_gives_low_confidence(self):
|
||||
resp = self._make_response(
|
||||
generic=[{"url": "https://a.com", "title": "A", "snippets": ["text"]}],
|
||||
sources={"https://a.com": {"hostname": "a.com", "age": None}},
|
||||
)
|
||||
items = _normalize_llm_context(resp)
|
||||
self.assertIsNone(items[0]["date"])
|
||||
self.assertEqual(items[0]["date_confidence"], "low")
|
||||
|
||||
def test_multiple_age_entries_picks_first_valid(self):
|
||||
resp = self._make_response(
|
||||
generic=[{"url": "https://a.com", "title": "A", "snippets": ["text"]}],
|
||||
sources={"https://a.com": {
|
||||
"hostname": "a.com",
|
||||
"age": ["Monday, March 10, 2026", "2026-03-10", "1 day ago"],
|
||||
}},
|
||||
)
|
||||
items = _normalize_llm_context(resp)
|
||||
self.assertEqual(items[0]["date"], "2026-03-10")
|
||||
|
||||
def test_ids_are_sequential(self):
|
||||
resp = self._make_response(
|
||||
generic=[
|
||||
{"url": "https://a.com", "title": "A", "snippets": ["a"]},
|
||||
{"url": "https://b.com", "title": "B", "snippets": ["b"]},
|
||||
{"url": "https://c.com", "title": "C", "snippets": ["c"]},
|
||||
],
|
||||
)
|
||||
items = _normalize_llm_context(resp)
|
||||
ids = [item["id"] for item in items]
|
||||
self.assertEqual(ids, ["W1", "W2", "W3"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user