|
""" |
|
agent.completion_guard — architectural guard against hallucinated completion. |
|
|
|
When an assistant produces a final text response without tool_calls, verify |
|
that any completion claims in the text are supported by successful tool |
|
calls in the recent history. If not, emit a HallucinationReport and let |
|
the agent loop inject a corrective message and retry. |
|
|
|
Design principle: model-agnostic, opt-in, bounded retries, non-fatal. |
|
|
|
See design document for rationale. |
|
""" |
|
|
|
from __future__ import annotations |
|
|
|
import json |
|
import logging |
|
import re |
|
from dataclasses import dataclass |
|
from typing import Any, Dict, List, Literal, Optional, Tuple |
|
|
|
logger = logging.getLogger(__name__) |
|
|
|
ClaimType = Literal["action", "gathering"] |
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────── |
|
# Data classes |
|
# ──────────────────────────────────────────────────────────────────────── |
|
|
|
|
|
@dataclass |
|
class Claim: |
|
claim_type: ClaimType |
|
claim_snippet: str # the matched span (+ some surrounding context) |
|
matched_pattern: str # debug: which regex id matched |
|
start: int # char offset in original text |
|
end: int |
|
|
|
|
|
@dataclass |
|
class HallucinationReport: |
|
claim_type: ClaimType |
|
claim_snippet: str |
|
missing_evidence: str |
|
matched_pattern: str |
|
|
|
|
|
@dataclass |
|
class ToolCallRecord: |
|
tool_name: str |
|
args: Dict[str, Any] |
|
result: Any # parsed dict or raw string |
|
success: bool |
|
classification: str # "action" | "gathering" | "both" | "unknown" |
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────── |
|
# Claim extraction — regex patterns |
|
# ──────────────────────────────────────────────────────────────────────── |
|
|
|
# Action completion patterns. Each is (regex, matched_pattern_id). |
|
# We use non-capturing groups to keep .group() clean. |
|
ACTION_PATTERNS: List[Tuple[str, str]] = [ |
|
# FR — "c'est fait" standalone |
|
(r"\bc'?est\s+fait\b", "action.fr.cest_fait"), |
|
# FR — "j'ai + verbe d'action au passé" |
|
( |
|
r"\bj'?ai\s+(?:envoy[eé]|install[eé]|cr[eé]{1,2}|supprim[eé]|lanc[eé]|ex[eé]cut[eé]|d[eé]ploy[eé]|corrig[eé]|modifi[eé]|patch[eé]|publi[eé]|pouss[eé]|merg[eé]|appliqu[eé]|redémarr[eé]|red[eé]marr[eé]|install[eé]|configur[eé])", |
|
"action.fr.jai_verb", |
|
), |
|
# FR — "X a/ont été [verbe passé]" |
|
( |
|
r"\b(?:le|la|les|un|une|mon|ma|mes|ce|cette|ces)\s+\w+(?:\s+\w+){0,3}?\s+(?:a|ont)\s+[eé]t[eé]\s+(?:envoy[eé]|install[eé]|cr[eé]{1,2}|supprim[eé]|lanc[eé]|ex[eé]cut[eé]|d[eé]ploy[eé]|corrig[eé]|modifi[eé]|patch[eé]|publi[eé]|appliqu[eé]|redémarr[eé]|red[eé]marr[eé])", |
|
"action.fr.passive", |
|
), |
|
# FR — "X est/sont lancé/envoyé/..." (present state implying completion) |
|
( |
|
r"\b(?:le|la|les)\s+(?:cron|mail|email|fichier|patch|script|service|gateway|module)\s+(?:est|sont)\s+(?:lanc[eé]|envoy[eé]|cr[eé]{1,2}|install[eé]|d[eé]ploy[eé]|appliqu[eé]|redémarr[eé]|red[eé]marr[eé]|actif|active|actifs|actives)", |
|
"action.fr.entity_state", |
|
), |
|
# FR — "Point 1 terminé / Point 2 fait" |
|
(r"\bpoint\s+\d+\s+(?:termin[eé]|fait|ex[eé]cut[eé]|r[eé]alis[eé]|appliqu[eé])", "action.fr.point_done"), |
|
# EN — "done" standalone |
|
(r"\b(?:it's|all)\s+done\b", "action.en.done"), |
|
(r"\bdone\.\s*$", "action.en.done_end"), |
|
# EN — "I [verbed]" |
|
( |
|
r"\bi(?:\s+have|'ve)?\s+(?:sent|installed|created|deleted|ran|fixed|deployed|patched|launched|published|pushed|merged|applied|restarted)", |
|
"action.en.i_verb", |
|
), |
|
# EN — "X has been/was [verbed]" |
|
( |
|
r"\b(?:the|a|an|my|this)\s+\w+(?:\s+\w+){0,3}?\s+(?:has\s+been|was|is\s+now)\s+(?:sent|installed|created|deleted|run|fixed|deployed|launched|published|applied|restarted|active)", |
|
"action.en.passive", |
|
), |
|
] |
|
|
|
# Gathering completion patterns. |
|
GATHERING_PATTERNS: List[Tuple[str, str]] = [ |
|
# FR — "voici [le/la/les] [rapport/audit/etc.]" |
|
( |
|
r"\bvoici\s+(?:le|la|les|l'|mon|ma|mes)\s*(?:rapport|audit|diagnostic|liste|analyse|[eé]tat|r[eé]sultat|r[eé]sum[eé]|synth[eè]se|d[eé]tail|premiers?\s+constats?)", |
|
"gather.fr.voici", |
|
), |
|
# FR — "j'ai [vérifié/audité/lu/analysé]" |
|
( |
|
r"\bj'?ai\s+(?:v[eé]rifi[eé]|inspect[eé]|audit[eé]|consult[eé]|analys[eé]|interrog[eé]|examin[eé]|parcouru|passé\s+en\s+revue)", |
|
"gather.fr.jai_check", |
|
), |
|
# FR — "audit/analyse/diagnostic terminé/complet" |
|
( |
|
r"\b(?:audit|analyse|diagnostic|v[eé]rification|inspection)\s+(?:termin[eé]e?|complet|complète|faite?|r[eé]el|r[eé]elle)", |
|
"gather.fr.audit_done", |
|
), |
|
# FR — "[analyse|audit|diagnostic|propositions] pour/sur l'instance X" |
|
( |
|
r"\b(?:analyse|audit|diagnostic|[eé]tat|propositions|recommandations)\b[^.!\n]{0,60}\bpour\s+l'?\s*(?:instance|service|site|code|base|serveur|syst[eè]me|module)", |
|
"gather.fr.for_instance", |
|
), |
|
# FR — "Analyse et propositions" |
|
( |
|
r"\b(?:analyse|audit|diagnostic|r[eé]sum[eé])\s+et\s+(?:propositions?|recommandations?|d[eé]tails?|r[eé]sultats?)", |
|
"gather.fr.analyse_et_props", |
|
), |
|
# FR — "Il/Elle + verbe de communication/état" (third-party report) |
|
( |
|
r"\b(?:il|elle|ils|elles)\s+(?:se\s+plaint|dit|raconte|mentionne|signale|indique|affirme|tourne\s+en\s+rond|est\s+dans\s+le\s+rouge|attribue|explique|rapporte|confirme|demande)", |
|
"gather.fr.third_party_pronoun", |
|
), |
|
# FR — "ProperNoun + verbe d'état spécifique" (X tourne en rond / X se plaint) |
|
( |
|
r"\b[A-Z][a-zA-Z]{2,}\s+(?:se\s+plaint|tourne\s+en\s+rond|est\s+dans\s+le\s+rouge|confirme|rapporte|se\s+trouve|mentionne|demande|signale)", |
|
"gather.fr.entity_state_verb", |
|
), |
|
# EN |
|
( |
|
r"\bhere\s+(?:is|are)\s+(?:the|my)\s+(?:report|audit|list|analysis|summary|state|result|diagnos|findings)", |
|
"gather.en.here_is", |
|
), |
|
( |
|
r"\bi(?:\s+have)?\s+(?:checked|verified|inspected|audited|analyzed|reviewed|examined|looked\s+at)", |
|
"gather.en.i_checked", |
|
), |
|
(r"\b(?:audit|analysis|diagnostic|check|inspection)\s+(?:complete|done|finished|completed)", "gather.en.audit_done"), |
|
] |
|
|
|
# Compiled patterns |
|
_ACTION_COMPILED = [(re.compile(p, re.IGNORECASE), mid) for p, mid in ACTION_PATTERNS] |
|
_GATHERING_COMPILED = [(re.compile(p, re.IGNORECASE), mid) for p, mid in GATHERING_PATTERNS] |
|
|
|
# Negation guard — skip claims preceded by a negation within ~30 chars |
|
_NEGATION_BEFORE_RE = re.compile( |
|
r"\b(?:pas|jamais|rien|aucun|impossible|n'?ai\s+pas|n'?a\s+pas|not|never|nothing|no|cannot|can't|couldn't|failed\s+to)\b", |
|
re.IGNORECASE, |
|
) |
|
|
|
# Question guard — skip if the containing sentence ends with "?" |
|
_SENTENCE_END_RE = re.compile(r"[.!?\n]") |
|
|
|
|
|
def _has_negation_nearby(text: str, claim_start: int, claim_end: int, window: int = 40) -> bool: |
|
"""Return True if a negation particle appears within `window` chars |
|
BEFORE the claim start. Tuned to catch 'Je n'ai pas envoyé le mail'.""" |
|
lo = max(0, claim_start - window) |
|
snippet = text[lo:claim_end] |
|
return bool(_NEGATION_BEFORE_RE.search(snippet)) |
|
|
|
|
|
def _is_in_question(text: str, claim_start: int, claim_end: int) -> bool: |
|
"""Return True if the claim is inside a sentence that ends with a question mark.""" |
|
# Find end of the sentence containing the claim |
|
end_match = _SENTENCE_END_RE.search(text, pos=claim_end) |
|
if end_match and text[end_match.start()] == "?": |
|
return True |
|
# Or: a '?' appears anywhere in the window around |
|
window_end = min(len(text), claim_end + 60) |
|
window_start = max(0, claim_start - 20) |
|
# Only consider '?' within the same sentence |
|
segment = text[window_start:window_end] |
|
if "?" in segment and not any(s in segment for s in [".\n", ". ", "!\n", "! "]): |
|
return True |
|
return False |
|
|
|
|
|
def extract_claims(text: str) -> List[Claim]: |
|
"""Extract completion claims from a text. |
|
|
|
Ignores negated claims and questions. |
|
""" |
|
if not text or not isinstance(text, str): |
|
return [] |
|
claims: List[Claim] = [] |
|
for compiled, mid in _ACTION_COMPILED: |
|
for m in compiled.finditer(text): |
|
if _has_negation_nearby(text, m.start(), m.end()): |
|
continue |
|
if _is_in_question(text, m.start(), m.end()): |
|
continue |
|
claims.append( |
|
Claim( |
|
claim_type="action", |
|
claim_snippet=_extract_context(text, m.start(), m.end()), |
|
matched_pattern=mid, |
|
start=m.start(), |
|
end=m.end(), |
|
) |
|
) |
|
for compiled, mid in _GATHERING_COMPILED: |
|
for m in compiled.finditer(text): |
|
if _has_negation_nearby(text, m.start(), m.end()): |
|
continue |
|
if _is_in_question(text, m.start(), m.end()): |
|
continue |
|
claims.append( |
|
Claim( |
|
claim_type="gathering", |
|
claim_snippet=_extract_context(text, m.start(), m.end()), |
|
matched_pattern=mid, |
|
start=m.start(), |
|
end=m.end(), |
|
) |
|
) |
|
return claims |
|
|
|
|
|
def _extract_context(text: str, start: int, end: int, context: int = 60) -> str: |
|
"""Extract a snippet including a bit of surrounding context, up to |
|
sentence boundaries.""" |
|
# Walk backward to find sentence start |
|
lo = max(0, start - context) |
|
# Walk forward to find sentence end |
|
hi = min(len(text), end + context) |
|
snippet = text[lo:hi].strip() |
|
# Collapse whitespace |
|
snippet = re.sub(r"\s+", " ", snippet) |
|
# Trim to nearest sentence boundary if possible |
|
return snippet[:200] |
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────── |
|
# Tool result success evaluation |
|
# ──────────────────────────────────────────────────────────────────────── |
|
|
|
_TRACEBACK_RE = re.compile(r"Traceback \(most recent call last\)", re.IGNORECASE) |
|
_GENERIC_ERROR_RE = re.compile( |
|
r"(?i)\b(?:error|exception|failed|failure|cannot\s+find|not\s+found|denied|refused|forbidden)\b" |
|
) |
|
_BENIGN_ERROR_CONTEXT_RE = re.compile( |
|
r"(?i)\b(?:no\s+error|error:\s*null|error:\s*none|error:\s*false|\"error\"\s*:\s*null|\"error\"\s*:\s*false)\b" |
|
) |
|
|
|
|
|
def tool_result_is_success(content: Any) -> bool: |
|
"""Heuristic: is a tool result a successful execution? |
|
|
|
Rules: |
|
- dict with success=False → fail |
|
- dict with error (non-null) → fail |
|
- dict with exit_code != 0 → fail |
|
- dict with output containing Traceback → fail |
|
- string containing Traceback → fail |
|
- string matching error keywords → fail (unless there's a 'no error' nearby) |
|
- otherwise → success |
|
""" |
|
if content is None: |
|
return False |
|
|
|
# Parse JSON string if needed |
|
if isinstance(content, str): |
|
stripped = content.strip() |
|
if stripped.startswith("{") or stripped.startswith("["): |
|
try: |
|
parsed = json.loads(stripped) |
|
return tool_result_is_success(parsed) |
|
except json.JSONDecodeError: |
|
pass # fall through to plain-string handling |
|
# Plain string — check for error markers |
|
if _TRACEBACK_RE.search(content): |
|
return False |
|
if _BENIGN_ERROR_CONTEXT_RE.search(content): |
|
return True |
|
if _GENERIC_ERROR_RE.search(content): |
|
return False |
|
return True |
|
|
|
if isinstance(content, dict): |
|
if content.get("success") is False: |
|
return False |
|
err = content.get("error") |
|
if err not in (None, "", False): |
|
return False |
|
exit_code = content.get("exit_code") |
|
if isinstance(exit_code, int) and exit_code != 0: |
|
return False |
|
# Check output for traceback |
|
output = content.get("output") or content.get("stdout") or "" |
|
if isinstance(output, str) and _TRACEBACK_RE.search(output): |
|
return False |
|
return True |
|
|
|
if isinstance(content, list): |
|
# Multi-part content — consider success if no part signals error |
|
return all(tool_result_is_success(part) for part in content) |
|
|
|
return True # opaque type — assume success |
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────── |
|
# Tool call extraction from message history |
|
# ──────────────────────────────────────────────────────────────────────── |
|
|
|
|
|
def _classify_terminal_command(command: str) -> str: |
|
"""Classify a terminal command as 'action' | 'gathering' | 'both'.""" |
|
if not command: |
|
return "both" |
|
cmd = command.lower().strip() |
|
# Strip SSH prefix if any — the real command is after the host |
|
m = re.match(r"ssh\s+(?:-\S+\s+)*\S+\s+['\"]?(.*?)['\"]?$", cmd, re.DOTALL) |
|
if m: |
|
cmd = m.group(1).strip() |
|
|
|
read_prefixes = ( |
|
"ls", "cat", "head", "tail", "grep", "find", "wc", "du", "df", "ps", |
|
"top", "date", "pwd", "which", "file ", "stat", "echo", "printf", |
|
"env", "whoami", "id", "hostname", "uname", "uptime", "free", "lscpu", |
|
"awk", "sed -n", "jq", "xxd", "od ", "man ", "curl -s ", "curl -x", |
|
"git log", "git status", "git show", "git diff", "git branch", "git blame", |
|
"journalctl", "systemctl status", "launchctl list", "launchctl print", |
|
"sudo -n /usr/bin/systemctl status", "sudo journalctl", |
|
"python3 -c \"print", "python -c \"print", |
|
) |
|
for p in read_prefixes: |
|
if cmd.startswith(p): |
|
return "gathering" |
|
|
|
action_prefixes = ( |
|
"rm ", "rm\t", "mv ", "cp ", "mkdir", "touch", "chmod", "chown", |
|
"systemctl start", "systemctl stop", "systemctl restart", "systemctl enable", |
|
"launchctl start", "launchctl stop", "launchctl kickstart", |
|
"pkill", "kill ", |
|
"git push", "git commit", "git merge", "git pull", "git fetch", "git reset --hard", "git checkout", |
|
"npm install", "pip install", "brew install", "apt install", "apt-get install", |
|
"curl -d", "curl -X POST", "curl --data", |
|
"ssh-copy-id", "scp ", |
|
"docker run", "docker start", "docker stop", "docker rm", |
|
) |
|
for p in action_prefixes: |
|
if cmd.startswith(p): |
|
return "action" |
|
|
|
# 'open URL' — opens in browser, not an action that modifies state |
|
if cmd.startswith("open "): |
|
return "gathering" |
|
|
|
# python script — ambiguous, could be either |
|
if cmd.startswith("python") or cmd.startswith("./"): |
|
return "both" |
|
|
|
# default — be permissive |
|
return "both" |
|
|
|
|
|
def _classify_tool_call(tool_name: str, args: Dict[str, Any]) -> str: |
|
"""Classify a tool call as 'action' | 'gathering' | 'both' | 'unknown'.""" |
|
name = (tool_name or "").lower() |
|
|
|
# Pure gathering |
|
gathering_tools = { |
|
"read_file", "search_files", "skill_view", "skills_list", |
|
"session_search", "session_get", "web_search", "web_extract", |
|
"web_read", "vision_analyze", "list_directory", "grep", |
|
} |
|
if name in gathering_tools: |
|
return "gathering" |
|
|
|
# Pure action |
|
action_tools = { |
|
"write_file", "patch", "edit", "delete_file", "move_file", |
|
"delegate_task", "spawn_subagent", |
|
"typefully_publish", "post_to_social", |
|
"gmail_send", "himalaya_send", |
|
} |
|
if name in action_tools: |
|
return "action" |
|
|
|
# Memory / skill management — action (modify state) |
|
if name in ("memory", "skill_manage", "todo"): |
|
# Only action if args suggest a write; heuristic: |
|
action_hint = args.get("action", "") |
|
if isinstance(action_hint, str) and action_hint.lower() in ( |
|
"add", "remove", "update", "patch", "create", "delete", "replace", "edit" |
|
): |
|
return "action" |
|
return "gathering" |
|
|
|
# Cronjob — depends on action arg |
|
if name == "cronjob": |
|
action = args.get("action", "") |
|
if isinstance(action, str): |
|
action = action.lower() |
|
if action in ("run", "execute", "trigger", "start"): |
|
return "action" |
|
if action in ("list", "get", "show", "view"): |
|
return "gathering" |
|
return "both" |
|
|
|
# Terminal — look at the command |
|
if name == "terminal": |
|
cmd = args.get("command", "") or "" |
|
return _classify_terminal_command(str(cmd)) |
|
|
|
# execute_code — usually an action |
|
if name == "execute_code": |
|
return "both" # could be either |
|
|
|
# Unknown tool — permissive |
|
return "unknown" |
|
|
|
|
|
def get_recent_tool_calls( |
|
messages: List[Dict[str, Any]], lookback_turns: int = 5 |
|
) -> List[ToolCallRecord]: |
|
"""Extract a flattened list of (tool_call, result) pairs from the last |
|
`lookback_turns` user/assistant turns. |
|
""" |
|
if not messages: |
|
return [] |
|
|
|
# Walk backward to find the last `lookback_turns` user/assistant turns. |
|
# A "turn" = one user or assistant message; tool results attached to an |
|
# assistant message are considered part of the same turn. |
|
turn_count = 0 |
|
cutoff_idx = 0 |
|
for i in range(len(messages) - 1, -1, -1): |
|
role = messages[i].get("role") |
|
if role in ("user", "assistant"): |
|
turn_count += 1 |
|
if turn_count >= lookback_turns: |
|
cutoff_idx = i |
|
break |
|
relevant = messages[cutoff_idx:] |
|
|
|
# Build a map from tool_call_id → result content |
|
result_map: Dict[str, Any] = {} |
|
for msg in relevant: |
|
if msg.get("role") == "tool": |
|
tid = msg.get("tool_call_id", "") |
|
content = msg.get("content") |
|
result_map[tid] = content |
|
|
|
records: List[ToolCallRecord] = [] |
|
for msg in relevant: |
|
if msg.get("role") != "assistant": |
|
continue |
|
tool_calls = msg.get("tool_calls") or [] |
|
for tc in tool_calls: |
|
if not isinstance(tc, dict): |
|
continue |
|
fn = tc.get("function") or {} |
|
name = fn.get("name", "") |
|
raw_args = fn.get("arguments", "{}") |
|
if isinstance(raw_args, str): |
|
try: |
|
args = json.loads(raw_args) if raw_args.strip() else {} |
|
except json.JSONDecodeError: |
|
args = {"__raw": raw_args} |
|
elif isinstance(raw_args, dict): |
|
args = raw_args |
|
else: |
|
args = {} |
|
call_id = tc.get("id", "") |
|
result = result_map.get(call_id) |
|
# If no matching result, treat as failed (safer) |
|
success = tool_result_is_success(result) if result is not None else False |
|
classification = _classify_tool_call(name, args) |
|
records.append( |
|
ToolCallRecord( |
|
tool_name=name, |
|
args=args, |
|
result=result, |
|
success=success, |
|
classification=classification, |
|
) |
|
) |
|
return records |
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────── |
|
# Evidence matching |
|
# ──────────────────────────────────────────────────────────────────────── |
|
|
|
|
|
_META_GATHERING_TOOLS = frozenset({ |
|
"skill_view", "skills_list", "session_search", "session_get", |
|
}) |
|
|
|
# French common sentence-starter words we should NOT treat as entities even |
|
# if capitalized. Deliberately short — we only need to avoid obvious noise. |
|
_COMMON_STARTERS_FR = frozenset({ |
|
"monsieur", "madame", "mademoiselle", |
|
"voici", "voilà", "voila", |
|
"analyse", "audit", "diagnostic", "point", "test", "stratégie", "strategie", |
|
"compris", "note", "noté", "note:", |
|
"non", "oui", "merci", "rien", "tout", "bien", "mal", |
|
"premier", "première", "deuxième", "troisième", |
|
"je", "tu", "il", "elle", "on", "nous", "vous", "ils", "elles", |
|
"le", "la", "les", "un", "une", "des", "ce", "cette", "ces", |
|
"mon", "ma", "mes", "ton", "ta", "tes", "son", "sa", "ses", |
|
"mais", "or", "et", "ou", "puis", "alors", "donc", "aussi", "ainsi", |
|
"par", "pour", "avec", "sans", "sur", "sous", "dans", "vers", "chez", |
|
"après", "avant", |
|
"rectification", "correction", "erreur", "fait", "conclusion", |
|
"état", "etat", "résultat", "resultat", "fenêtre", "fenetre", |
|
}) |
|
|
|
_COMMON_STARTERS_EN = frozenset({ |
|
"here", "now", "well", "ok", "okay", "yes", "no", "thanks", "done", |
|
"first", "second", "third", "finally", "indeed", "right", "good", |
|
"the", "a", "an", "i", "we", "you", "they", "he", "she", "it", |
|
"my", "your", "his", "her", "their", "its", "our", |
|
"but", "and", "or", "so", "then", "also", "thus", |
|
"for", "with", "without", "on", "under", "in", "at", "by", |
|
"result", "conclusion", "error", |
|
}) |
|
|
|
_ALL_COMMON_STARTERS = _COMMON_STARTERS_FR | _COMMON_STARTERS_EN |
|
|
|
|
|
def _extract_target_entities(text: str) -> List[str]: |
|
"""Extract entities that the claim is 'about' (instance/service X, proper nouns). |
|
|
|
Returns lowercased entity strings. Used to verify that at least one |
|
gathering tool actually touched the entity (via argument substring match). |
|
""" |
|
entities = set() |
|
|
|
# Pattern 1: "instance/service/serveur X" → X |
|
inst_re = re.compile( |
|
r"\b(?:instance|service|serveur|server|site|application|app|module|compte|account|bot|vps)\s+" |
|
r"(?:de\s+|of\s+|d'\s*|du\s+|des\s+)?" |
|
r"([A-Z][a-zA-Z0-9_\-\.]{1,})", |
|
re.IGNORECASE, |
|
) |
|
for m in inst_re.finditer(text): |
|
entities.add(m.group(1).lower()) |
|
|
|
# Pattern 2: proper nouns (capital-initial words) that aren't common starters |
|
proper_re = re.compile(r"\b([A-Z][a-zA-Z]{2,})\b") |
|
for m in proper_re.finditer(text): |
|
word = m.group(1) |
|
if word.lower() not in _ALL_COMMON_STARTERS: |
|
entities.add(word.lower()) |
|
|
|
return sorted(entities) |
|
|
|
|
|
def _args_contain_entity(args: Dict[str, Any], entity: str) -> bool: |
|
"""True if the entity substring appears (case-insensitive) in the tool args JSON.""" |
|
try: |
|
args_str = json.dumps(args).lower() |
|
except (TypeError, ValueError): |
|
args_str = str(args).lower() |
|
return entity in args_str |
|
|
|
|
|
def _result_contains_entity(result: Any, entity: str) -> bool: |
|
"""True if the tool result content (string or JSON) contains the entity. |
|
|
|
Used as a fallback when the args don't mention the entity but the tool |
|
output might — e.g., `read_file(CLIENTS.md)` could return text containing |
|
'Matthieu' somewhere. |
|
""" |
|
if result is None: |
|
return False |
|
try: |
|
if isinstance(result, (dict, list)): |
|
result_str = json.dumps(result).lower() |
|
else: |
|
result_str = str(result).lower() |
|
except (TypeError, ValueError): |
|
return False |
|
return entity in result_str |
|
|
|
|
|
def _has_successful_action_tool(tool_calls: List[ToolCallRecord]) -> bool: |
|
"""True if at least one tool call was a successful action.""" |
|
for tc in tool_calls: |
|
if not tc.success: |
|
continue |
|
if tc.classification in ("action", "both"): |
|
return True |
|
return False |
|
|
|
|
|
def _has_successful_gathering_tool( |
|
tool_calls: List[ToolCallRecord], |
|
required_entities: Optional[List[str]] = None, |
|
) -> bool: |
|
"""True if at least one SUBSTANTIVE gathering tool succeeded. |
|
|
|
If `required_entities` is non-empty, the tool must ALSO have touched at |
|
least one of those entities (checked via args first, then result). |
|
""" |
|
for tc in tool_calls: |
|
if not tc.success: |
|
continue |
|
if tc.classification not in ("gathering", "both"): |
|
continue |
|
if tc.tool_name in _META_GATHERING_TOOLS: |
|
continue # meta introspection doesn't count as substantive gathering |
|
if required_entities: |
|
matched = any( |
|
_args_contain_entity(tc.args, e) or _result_contains_entity(tc.result, e) |
|
for e in required_entities |
|
) |
|
if not matched: |
|
continue |
|
return True |
|
return False |
|
|
|
|
|
def claim_has_supporting_evidence( |
|
claim: Claim, tool_calls: List[ToolCallRecord] |
|
) -> bool: |
|
if claim.claim_type == "action": |
|
return _has_successful_action_tool(tool_calls) |
|
if claim.claim_type == "gathering": |
|
entities = _extract_target_entities(claim.claim_snippet) |
|
return _has_successful_gathering_tool(tool_calls, required_entities=entities) |
|
return True # unknown type — permissive |
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────── |
|
# Main entry point |
|
# ──────────────────────────────────────────────────────────────────────── |
|
|
|
|
|
def detect_hallucinated_completion( |
|
final_text: str, |
|
recent_messages: List[Dict[str, Any]], |
|
lookback_turns: int = 5, |
|
) -> Optional[HallucinationReport]: |
|
"""Primary detection function. |
|
|
|
Returns None if the text is grounded or contains no claims. |
|
Returns HallucinationReport when a claim is detected without supporting |
|
evidence in the recent tool history. |
|
""" |
|
if not final_text or not isinstance(final_text, str): |
|
return None |
|
|
|
claims = extract_claims(final_text) |
|
if not claims: |
|
return None |
|
|
|
tool_calls = get_recent_tool_calls(recent_messages, lookback_turns) |
|
|
|
# Check each claim; return the first mismatch |
|
for claim in claims: |
|
if claim_has_supporting_evidence(claim, tool_calls): |
|
continue |
|
return HallucinationReport( |
|
claim_type=claim.claim_type, |
|
claim_snippet=claim.claim_snippet, |
|
missing_evidence=_describe_missing_evidence(claim, tool_calls), |
|
matched_pattern=claim.matched_pattern, |
|
) |
|
|
|
return None |
|
|
|
|
|
def _describe_missing_evidence( |
|
claim: Claim, tool_calls: List[ToolCallRecord] |
|
) -> str: |
|
"""Build a human-readable explanation of what's missing.""" |
|
if not tool_calls: |
|
if claim.claim_type == "action": |
|
return ( |
|
"No tool call was made in the recent turns. " |
|
"An action claim requires a successful action tool." |
|
) |
|
return ( |
|
"No tool call was made in the recent turns. " |
|
"A gathering claim requires at least one successful read/search tool." |
|
) |
|
|
|
# Describe what we saw |
|
lines = [] |
|
for tc in tool_calls[-6:]: |
|
status = "ok" if tc.success else "FAILED" |
|
name = tc.tool_name |
|
cls = tc.classification |
|
lines.append(f" - {name} [{cls}, {status}]") |
|
|
|
header_by_type = { |
|
"action": ( |
|
"No successful action tool call (category 'action' or 'both') " |
|
"found in the recent turns that could back this completion claim." |
|
), |
|
"gathering": ( |
|
"No substantive gathering tool call (read_file, search_files, " |
|
"terminal with read commands, web_search, etc.) found in the recent turns. " |
|
"Meta-introspection tools (skill_view, session_search) do not count as " |
|
"gathering evidence for a claim about a specific instance/service/file." |
|
), |
|
} |
|
header = header_by_type.get(claim.claim_type, "No supporting tool call found.") |
|
return header + "\nRecent tool calls:\n" + "\n".join(lines) |
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────── |
|
# Correction message builder |
|
# ──────────────────────────────────────────────────────────────────────── |
|
|
|
|
|
def build_correction_message(report: HallucinationReport) -> str: |
|
return ( |
|
f"[System correction — completion_guard] Your previous response claims:\n\n" |
|
f" \"{report.claim_snippet}\"\n\n" |
|
f"But the tool history does NOT support this claim.\n\n" |
|
f"{report.missing_evidence}\n\n" |
|
f"Either:\n" |
|
f"(a) Execute the required tool(s) NOW to actually perform the action / " |
|
f"gather the data, then produce a response grounded in the actual tool results.\n" |
|
f"(b) Revise your response to accurately reflect what was actually done " |
|
f"(or not done), including any failures or limitations.\n\n" |
|
f"Do NOT repeat the unsupported claim. Either act or correct — " |
|
f"do not apologize at length." |
|
) |
