Anton
052591281d
- Complete agent system with planning and execution - Bridge API with FastAPI server - WhatsApp-style mobile dashboard - Enhanced web search with query improvement - Security guards and policy checks - Comprehensive documentation
325 lines
12 KiB
Python
325 lines
12 KiB
Python
"""CJK + Russian-language + hallucination guard for user-facing chat output.
|
|
|
|
Applied as the final layer before any text is returned to a user.
|
|
Flow:
|
|
raw LLM text
|
|
-> scan_cjk() — detects CJK characters
|
|
-> if CJK: retry once; if still CJK: safe fallback
|
|
-> validate_russian() — Cyrillic dominance + foreign word check
|
|
-> if fails: retry once; if still fails: safe fallback
|
|
-> detect_hallucination() — structured sections with no user context
|
|
-> if hallucination: block, return clarification prompt
|
|
-> save diagnostic artifact on every incident
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
_CJK_RE = re.compile(r"[\u4e00-\u9fff]")
|
|
_LATIN_WORD_RE = re.compile(r"\b([a-zA-Z]{3,})\b")
|
|
_URL_RE = re.compile(r"https?://\S+|www\.\S+")
|
|
_CODE_RE = re.compile(r"`[^`]+`|```[\s\S]+?```")
|
|
_ALNUM_RE = re.compile(r"\b[a-zA-Z]*\d+[a-zA-Z\d]*\b") # model/version names: llama3.2, GPT-4
|
|
|
|
# Hallucination: structured section headers that signal invented project context
|
|
_HALLUCINATION_SECTION_RE = re.compile(
|
|
r"(?m)^\s*(?:Цель|Прогресс|Следующие\s+шаги|Задача|Датасет|Инструменты|"
|
|
r"Статус|Шаги|Данные|Проект)\s*:",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
# Keywords that indicate the user actually provided project/task context
|
|
_PROJECT_CONTEXT_RE = re.compile(
|
|
r"(проект|задач[аи]?|цел[ьи]|данн|датасет|pandas|csv|excel|"
|
|
r"анализ|разработ|таблиц|файл|база\s*данных|прогресс|шаги|инструмент|"
|
|
r"нужно сделать|нам надо|план работ)",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
CJK_SAFE_FALLBACK = (
|
|
"Извините, произошла техническая ошибка. Пожалуйста, повторите запрос."
|
|
)
|
|
|
|
HALLUCINATION_CLARIFY = (
|
|
"Уточните, пожалуйста, задачу или проект."
|
|
)
|
|
|
|
_DIAG_DIR = Path(__file__).resolve().parents[2] / "data" / "diagnostics"
|
|
|
|
RETRY_SYSTEM_PROMPT = (
|
|
"Respond ONLY in Russian. Do not use any other language. "
|
|
"No Chinese, no English, no other script. Russian only."
|
|
)
|
|
|
|
RETRY_RUSSIAN_PROMPT = (
|
|
"Respond ONLY in natural Russian. Do not use English, Spanish, Chinese, "
|
|
"or any other language unless explicitly asked."
|
|
)
|
|
|
|
# Words that clearly indicate non-Russian content when appearing in a Russian reply
|
|
_FOREIGN_WORDS = frozenset({
|
|
# English function words / common vocabulary
|
|
"the", "and", "you", "that", "this", "with", "have", "from", "will",
|
|
"been", "would", "their", "there", "what", "your", "about", "when",
|
|
"which", "than", "then", "into", "also", "could", "just", "like",
|
|
"very", "some", "more", "help", "right", "good", "okay", "well",
|
|
"various", "assist", "different", "ready", "thank", "thanks", "yes",
|
|
"all", "aider",
|
|
# English verbs that slip into Russian responses
|
|
"learns", "improves", "uses", "allows", "provides", "creates", "helps",
|
|
"includes", "requires", "supports", "assisting", "helping",
|
|
"assistir", "working", "making", "being", "having",
|
|
# English nouns/adjectives seen in mixed llama3.2:3b output
|
|
"human", "clear", "mainly", "audience", "deadlines", "breaks", "whom",
|
|
"experience", "experiences", "important", "different", "possible",
|
|
"available", "specific", "current", "following", "including",
|
|
# Spanish
|
|
"gracias", "hola", "bien", "como", "pero", "para", "porque",
|
|
})
|
|
|
|
|
|
def scan_cjk(text: str) -> list[dict[str, Any]]:
|
|
"""Return list of CJK findings: pos, char, codepoint, context."""
|
|
return [
|
|
{
|
|
"pos": m.start(),
|
|
"char": m.group(),
|
|
"codepoint": f"U+{ord(m.group()):04X}",
|
|
"context": text[max(0, m.start() - 20) : m.end() + 20],
|
|
}
|
|
for m in _CJK_RE.finditer(text)
|
|
]
|
|
|
|
|
|
def has_cjk(text: str) -> bool:
|
|
return bool(_CJK_RE.search(text))
|
|
|
|
|
|
def validate_russian(text: str) -> dict[str, Any] | None:
|
|
"""Check that text is predominantly Russian.
|
|
|
|
Returns None if text passes (is sufficiently Russian).
|
|
Returns {"reason": ..., "details": ...} if validation fails.
|
|
|
|
Exceptions (not flagged):
|
|
- URLs and code blocks
|
|
- All-caps acronyms (API, HTTP, SQL)
|
|
- Alphanumeric model/version identifiers (llama3.2, GPT-4)
|
|
"""
|
|
if not text or len(text.strip()) < 15:
|
|
return None
|
|
|
|
clean = _URL_RE.sub(" ", text)
|
|
clean = _CODE_RE.sub(" ", clean)
|
|
clean = _ALNUM_RE.sub(" ", clean)
|
|
|
|
cyrillic = sum(1 for c in clean if "\u0400" <= c <= "\u04ff")
|
|
latin_chars = sum(1 for c in clean if "a" <= c.lower() <= "z")
|
|
total_alpha = cyrillic + latin_chars
|
|
|
|
if total_alpha < 15:
|
|
return None
|
|
|
|
cyrillic_ratio = cyrillic / total_alpha
|
|
|
|
if cyrillic_ratio < 0.6 and total_alpha >= 25:
|
|
return {
|
|
"reason": "low_cyrillic_ratio",
|
|
"details": f"ratio={cyrillic_ratio:.2f} ({cyrillic} cyrillic / {total_alpha} alpha)",
|
|
}
|
|
|
|
suspicious: list[str] = []
|
|
for m in _LATIN_WORD_RE.finditer(clean):
|
|
word = m.group()
|
|
if word.isupper():
|
|
continue # all-caps acronym (API, HTTP, SQL)
|
|
if word.lower() in _FOREIGN_WORDS:
|
|
suspicious.append(word.lower())
|
|
|
|
if suspicious:
|
|
return {
|
|
"reason": "foreign_words_detected",
|
|
"details": f"words={suspicious[:5]}",
|
|
}
|
|
|
|
return None
|
|
|
|
|
|
def detect_hallucination(text: str, user_message: str) -> dict[str, Any] | None:
|
|
"""Detect invented project/task context in a reply the user never asked for.
|
|
|
|
Fires when:
|
|
1. The reply contains structured section headers (Цель:, Прогресс:, etc.)
|
|
2. The user message contains no project/task context keywords
|
|
|
|
Returns None if no hallucination detected.
|
|
Returns {"reason": ..., "sections": [...]} if hallucination detected.
|
|
"""
|
|
sections = _HALLUCINATION_SECTION_RE.findall(text)
|
|
if not sections:
|
|
return None # no structured sections → cannot be this kind of hallucination
|
|
|
|
if _PROJECT_CONTEXT_RE.search(user_message):
|
|
return None # user explicitly mentioned project/task context → allowed
|
|
|
|
return {
|
|
"reason": "hallucinated_project_context",
|
|
"sections": [s.strip() for s in sections],
|
|
}
|
|
|
|
|
|
def _save_incident(trace_id: str, kind: str, artifact: dict[str, Any]) -> None:
|
|
try:
|
|
_DIAG_DIR.mkdir(parents=True, exist_ok=True)
|
|
path = _DIAG_DIR / f"chat_{kind}_{trace_id}.json"
|
|
path.write_text(
|
|
json.dumps(artifact, indent=2, ensure_ascii=False), encoding="utf-8"
|
|
)
|
|
except Exception as exc:
|
|
print(f"[GUARD] diag save failed: {exc}")
|
|
|
|
|
|
def guard_reply(
|
|
raw: str,
|
|
prompt: str,
|
|
trace_id: str,
|
|
retry_fn, # callable(prompt: str) -> str
|
|
user_message: str = "",
|
|
) -> tuple[str, bool, bool]:
|
|
"""Apply CJK + Russian-language + hallucination guard to a raw LLM reply.
|
|
|
|
Args:
|
|
raw: raw text from LLM, before any processing
|
|
prompt: the original user-facing prompt that produced raw
|
|
trace_id: for logging and artifact naming
|
|
retry_fn: callable that re-runs generation with a given prompt
|
|
user_message: original user text (used for hallucination context check)
|
|
|
|
Returns:
|
|
(final_text, fallback_used, lang_incident)
|
|
lang_incident: True if any guard fired
|
|
"""
|
|
text = raw
|
|
lang_incident = False
|
|
|
|
# ── Empty response guard ──────────────────────────────────────────────────
|
|
if not text.strip():
|
|
print(f"[GUARD] EMPTY RESPONSE trace_id={trace_id} — retrying")
|
|
text = retry_fn(prompt)
|
|
lang_incident = True
|
|
if not text.strip():
|
|
print(f"[GUARD] EMPTY STILL after retry — using safe fallback trace_id={trace_id}")
|
|
return CJK_SAFE_FALLBACK, True, True
|
|
|
|
# ── CJK guard ────────────────────────────────────────────────────────────
|
|
if has_cjk(text):
|
|
lang_incident = True
|
|
findings = scan_cjk(text)
|
|
print(
|
|
f"[GUARD] CJK DETECTED trace_id={trace_id} "
|
|
f"chars={len(findings)} fragment={findings[0]['context']!r}"
|
|
)
|
|
|
|
retry_prompt = f"{RETRY_SYSTEM_PROMPT}\n\n{prompt}"
|
|
retried = retry_fn(retry_prompt)
|
|
|
|
if has_cjk(retried):
|
|
print(f"[GUARD] CJK STILL PRESENT after retry — using safe fallback trace_id={trace_id}")
|
|
_save_incident(trace_id, "cjk", {
|
|
"trace_id": trace_id,
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"guard": "cjk",
|
|
"original_prompt": prompt,
|
|
"raw_response": raw,
|
|
"cjk_findings": findings,
|
|
"retried_response": retried,
|
|
"final_sent": CJK_SAFE_FALLBACK,
|
|
"fallback_used": True,
|
|
})
|
|
return CJK_SAFE_FALLBACK, True, True
|
|
|
|
print(f"[GUARD] CJK retry clean trace_id={trace_id}")
|
|
_save_incident(trace_id, "cjk", {
|
|
"trace_id": trace_id,
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"guard": "cjk",
|
|
"original_prompt": prompt,
|
|
"raw_response": raw,
|
|
"cjk_findings": findings,
|
|
"retried_response": retried,
|
|
"final_sent": retried,
|
|
"fallback_used": False,
|
|
})
|
|
text = retried
|
|
|
|
# ── Russian-language guard ────────────────────────────────────────────────
|
|
validation = validate_russian(text)
|
|
if validation:
|
|
lang_incident = True
|
|
print(
|
|
f"[GUARD] NON-RUSSIAN trace_id={trace_id} "
|
|
f"reason={validation['reason']} {validation['details']}"
|
|
)
|
|
|
|
retry_prompt = f"{RETRY_RUSSIAN_PROMPT}\n\n{prompt}"
|
|
retried = retry_fn(retry_prompt)
|
|
|
|
retry_validation = validate_russian(retried)
|
|
if retry_validation:
|
|
print(f"[GUARD] NON-RUSSIAN STILL after retry — using safe fallback trace_id={trace_id}")
|
|
_save_incident(trace_id, "lang", {
|
|
"trace_id": trace_id,
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"guard": "russian_validation",
|
|
"original_prompt": prompt,
|
|
"raw_response": raw,
|
|
"validation_failure": validation,
|
|
"retried_response": retried,
|
|
"retry_validation_failure": retry_validation,
|
|
"final_sent": CJK_SAFE_FALLBACK,
|
|
"fallback_used": True,
|
|
})
|
|
return CJK_SAFE_FALLBACK, True, True
|
|
|
|
print(f"[GUARD] Russian retry clean trace_id={trace_id}")
|
|
_save_incident(trace_id, "lang", {
|
|
"trace_id": trace_id,
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"guard": "russian_validation",
|
|
"original_prompt": prompt,
|
|
"raw_response": raw,
|
|
"validation_failure": validation,
|
|
"retried_response": retried,
|
|
"retry_validation_failure": None,
|
|
"final_sent": retried,
|
|
"fallback_used": False,
|
|
})
|
|
text = retried
|
|
|
|
# ── Hallucination guard ───────────────────────────────────────────────────
|
|
hallucination = detect_hallucination(text, user_message)
|
|
if hallucination:
|
|
lang_incident = True
|
|
print(
|
|
f"[GUARD] HALLUCINATION trace_id={trace_id} "
|
|
f"sections={hallucination['sections']}"
|
|
)
|
|
_save_incident(trace_id, "hallucination", {
|
|
"trace_id": trace_id,
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"guard": "hallucination",
|
|
"user_message": user_message,
|
|
"original_prompt": prompt,
|
|
"raw_response": raw,
|
|
"hallucination_details": hallucination,
|
|
"final_sent": HALLUCINATION_CLARIFY,
|
|
"fallback_used": False,
|
|
})
|
|
return HALLUCINATION_CLARIFY, False, True
|
|
|
|
return text, False, lang_incident
|