feat(agent): scrape linked pages to ground answers in real content

Adds `_fetch_page_content()`, which downloads a URL with a
browser-like User-Agent and uses BeautifulSoup with the lxml
parser to strip scripts, navigation, and other non-content
markup, returning up to 3000 characters of readable text.

The response validator now also compares generated replies
against this scraped text, flagging named entities and claims
that do not appear in the retrieved page content.

Dependencies: adds beautifulsoup4==4.12.3 and pins lxml==5.3.0
for parser compatibility.
This commit is contained in:
2026-10-26 16:48:29 -03:00
parent 9eed02645b
commit 723ca1506b
231 changed files with 20508 additions and 5330 deletions
+95
View File
@@ -264,6 +264,50 @@ def _web_search(query: str, max_results: int = 5) -> List[Dict[str, str]]:
return []
def _fetch_page_content(url: str, max_chars: int = 3000) -> str:
"""Fetch a webpage and extract readable text content.
Uses requests with a browser-like User-Agent and BeautifulSoup to
extract meaningful text while stripping scripts, styles, and markup.
The result is truncated to max_chars to keep prompts manageable.
Args:
url: Target webpage URL.
max_chars: Maximum characters to return.
Returns:
Extracted text content, or an empty string on any failure.
"""
try:
headers = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/120.0.0.0 Safari/537.36"
),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9,pt-BR;q=0.8,pt;q=0.7",
}
response = requests.get(url, headers=headers, timeout=10)
response.raise_for_status()
content_type = response.headers.get("Content-Type", "")
if "text/html" not in content_type.lower() and "application/xhtml" not in content_type.lower():
return ""
from bs4 import BeautifulSoup
soup = BeautifulSoup(response.text, "lxml")
# Remove non-content elements
for tag in soup(["script", "style", "nav", "header", "footer", "aside", "form", "iframe"]):
tag.decompose()
text = soup.get_text(separator="\n", strip=True)
lines = [line.strip() for line in text.splitlines() if line.strip()]
cleaned = "\n".join(lines)
if len(cleaned) > max_chars:
cleaned = cleaned[:max_chars] + "..."
return cleaned
except Exception:
return ""
def set_schedule(days: List[int], start_time: str, end_time: str) -> bool:
"""Validate and persist the proactive message schedule.
@@ -2959,6 +3003,8 @@ def _validate_search_response(
- Does the response contain specific numbers/dates that could indicate
hallucination when they don't appear in the results?
- Does the response claim to have searched when search was not triggered?
- If page content was scraped, does the response mention facts not
present in that content?
This is not a perfect guardrail, but it catches the most common
hallucination patterns seen in production (invented URLs, invented
@@ -3025,6 +3071,45 @@ def _validate_search_response(
"search results. Possible hallucination of statistics."
)
# Check if response contains facts not supported by scraped page content
scraped_contents = [r.get("content", "") for r in web_results if r.get("content")]
if scraped_contents:
combined_content = " ".join(scraped_contents).lower()
# Look for specific claims in the response that might be hallucinated
# Check for named entities, specific facts, or quotes that don't appear in content
import re
# Extract potential named entities (capitalized words/phrases)
entity_pattern = r'\b[A-Z][a-z]+(?:\s+[A-Z][a-z]+)*\b'
response_entities = re.findall(entity_pattern, response)
# Filter out common words that are not named entities
common_words = {
"according", "this", "that", "these", "those", "the", "a", "an",
"according", "based", "accordingly", "however", "therefore",
"moreover", "furthermore", "nevertheless", "nonetheless",
"example", "domain", "search", "results", "found", "encontrei",
"busquei", "busca", "pesquisa", "internet"
}
filtered_entities = [
e for e in response_entities
if e.lower() not in common_words and len(e) > 3
]
# Check if key entities from response appear in scraped content
missing_entities = []
for entity in filtered_entities:
if entity.lower() not in combined_content:
missing_entities.append(entity)
# If more than 50% of filtered named entities are missing from content, flag it
# Only flag if we have enough entities to make a reasonable judgment
if filtered_entities and len(missing_entities) > len(filtered_entities) * 0.5 and len(filtered_entities) >= 2:
result["is_valid"] = False
result["reason"] = (
"Response contains specific named entities or facts that don't "
"appear in the scraped page content. Possible hallucination."
)
return result
@@ -6681,12 +6766,22 @@ async def process_text_message(update: Update, context: ContextTypes.DEFAULT_TYP
break
if search_query:
web_results = _web_search(search_query, max_results=WEB_SEARCH_MAX_RESULTS)
# Fetch page content for each result to provide richer context.
for result in web_results:
page_url = result.get("url", "")
if page_url:
page_text = _fetch_page_content(page_url, max_chars=2500)
if page_text:
result["content"] = page_text
else:
web_results = []
if web_results:
web_context = "\n\n[WEB SEARCH RESULTS - USE THESE TO ANSWER, DO NOT INVENT]\n"
for i, result in enumerate(web_results, 1):
web_context += f"{i}. {result.get('title', '')}\nURL: {result.get('url', '')}\n"
page_content = result.get("content")
if page_content:
web_context += f"Content:\n{page_content}\n"
web_context += "\n[END OF WEB SEARCH RESULTS - Base your answer ONLY on the information above. If the information is not sufficient, say so. NEVER invent facts, URLs, or references.]"
modified = modified + "\n\n" + web_context
else: