feat(search): validate responses against web results
Adds a post-generation validation step that checks whether a search-grounded reply is actually supported by the retrieved results: it extracts cited URLs from the response and flags replies that claim to have searched without citing any of the provided sources, as well as replies containing specific numbers or dates absent from the results. Also switches the default model to xiaomi/mimo-v2.6-flash and documents the anti-hallucination behavior in the README.
This commit is contained in:
@@ -2925,6 +2925,144 @@ def ensure_min_lines(response: str, user_message: str, profile: Dict[str, Any])
|
||||
return response
|
||||
|
||||
|
||||
# ===================== VALIDATION =====================
|
||||
|
||||
def _extract_urls(text: str) -> List[str]:
|
||||
"""Extract URLs from text using simple regex patterns.
|
||||
|
||||
Args:
|
||||
text: Text to search for URLs.
|
||||
|
||||
Returns:
|
||||
List of URL strings found in the text.
|
||||
"""
|
||||
import re
|
||||
patterns = [
|
||||
r'https?://[^\s)>\]"\']+',
|
||||
r'www\.[^\s)>\]"\']+',
|
||||
]
|
||||
urls = []
|
||||
for pattern in patterns:
|
||||
urls.extend(re.findall(pattern, text))
|
||||
return urls
|
||||
|
||||
|
||||
def _validate_search_response(
|
||||
response: str,
|
||||
web_results: List[Dict[str, str]],
|
||||
user_message: str,
|
||||
) -> Dict[str, Any]:
|
||||
"""Validate whether a search-based response appears grounded in results.
|
||||
|
||||
Performs lightweight heuristic checks:
|
||||
- If web results exist, does the response cite at least one result URL?
|
||||
- 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?
|
||||
|
||||
This is not a perfect guardrail, but it catches the most common
|
||||
hallucination patterns seen in production (invented URLs, invented
|
||||
statistics, false claims of having searched).
|
||||
|
||||
Args:
|
||||
response: Assistant reply text.
|
||||
web_results: Search results that were injected into the prompt.
|
||||
user_message: Original user message.
|
||||
|
||||
Returns:
|
||||
Dict with keys:
|
||||
- "is_valid": bool, whether the response passed validation
|
||||
- "reason": str, explanation when invalid
|
||||
- "cited_urls": list of URLs found in the response
|
||||
- "available_urls": list of URLs from web_results
|
||||
"""
|
||||
result = {
|
||||
"is_valid": True,
|
||||
"reason": "",
|
||||
"cited_urls": [],
|
||||
"available_urls": [],
|
||||
}
|
||||
|
||||
# Build available URLs from search results
|
||||
available_urls = [r.get("url", "") for r in web_results if r.get("url")]
|
||||
result["available_urls"] = available_urls
|
||||
|
||||
# Extract URLs cited in the response
|
||||
cited_urls = _extract_urls(response)
|
||||
result["cited_urls"] = cited_urls
|
||||
|
||||
# If search results were provided, the response should cite at least one
|
||||
if available_urls:
|
||||
has_citation = any(url in cited_urls for url in available_urls)
|
||||
if not has_citation:
|
||||
# Check if response mentions search was performed
|
||||
search_indicators = [
|
||||
"search", "busquei", "busca", "found", "encontrei",
|
||||
"resultados", "results", "pesquisa", "internet"
|
||||
]
|
||||
claims_search = any(ind in response.lower() for ind in search_indicators)
|
||||
if claims_search:
|
||||
result["is_valid"] = False
|
||||
result["reason"] = (
|
||||
"Response claims to have searched but does not cite any "
|
||||
"of the provided search result URLs. Possible hallucination."
|
||||
)
|
||||
|
||||
# Check for suspicious patterns: specific numbers without context
|
||||
if web_results:
|
||||
# If response contains very specific numbers (percentages, exact counts)
|
||||
# but search results don't contain similar numbers, flag it
|
||||
import re
|
||||
number_pattern = r'\b\d+[\.,]?\d*\s*(%|milhões|mil|bilhões|bilhão)?\b'
|
||||
response_numbers = re.findall(number_pattern, response)
|
||||
result_text = " ".join(r.get("title", "") + " " + r.get("url", "") for r in web_results)
|
||||
result_numbers = re.findall(number_pattern, result_text)
|
||||
|
||||
if response_numbers and not result_numbers:
|
||||
result["is_valid"] = False
|
||||
result["reason"] = (
|
||||
"Response contains specific numbers that don't appear in "
|
||||
"search results. Possible hallucination of statistics."
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def _sanitize_response(
|
||||
response: str,
|
||||
web_results: List[Dict[str, str]],
|
||||
user_message: str,
|
||||
) -> str:
|
||||
"""Post-process a response to reduce hallucination risk.
|
||||
|
||||
If the response appears to contain hallucinated content based on
|
||||
search results, append a disclaimer or sanitize the response.
|
||||
|
||||
Args:
|
||||
response: Assistant reply text.
|
||||
web_results: Search results that were injected into the prompt.
|
||||
user_message: Original user message.
|
||||
|
||||
Returns:
|
||||
Sanitized response text, or original if no issues detected.
|
||||
"""
|
||||
if not web_results:
|
||||
return response
|
||||
|
||||
validation = _validate_search_response(response, web_results, user_message)
|
||||
if not validation["is_valid"]:
|
||||
# Append a disclaimer instead of sending potentially false info
|
||||
disclaimer = (
|
||||
"\n\n[Note: I cannot fully verify the specific details above. "
|
||||
"Please check the official sources for accurate information.]"
|
||||
)
|
||||
# Only add disclaimer if not already present
|
||||
if "cannot fully verify" not in response.lower():
|
||||
response = response + disclaimer
|
||||
|
||||
return response
|
||||
|
||||
|
||||
AUTO_ENRICH_EVERY = int(os.getenv("AUTO_ENRICH_EVERY", "2"))
|
||||
_auto_enrich_counter = 0
|
||||
|
||||
@@ -6543,15 +6681,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)
|
||||
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"
|
||||
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:
|
||||
web_results = []
|
||||
|
||||
response = await get_psagentai_response(modified)
|
||||
if response:
|
||||
# Post-validation: if search results were provided, ensure the
|
||||
# response is grounded in them; otherwise append a disclaimer.
|
||||
response = _sanitize_response(response, web_results, user_message)
|
||||
add_to_context("assistant", response)
|
||||
await update.message.reply_text(response, parse_mode=ParseMode.HTML)
|
||||
# Fire-and-forget background tasks: profile enrichment + self-improvement.
|
||||
|
||||
Reference in New Issue
Block a user