"""The model-free half of the verdict. Everything here is a fact about the transcript that can be established without asking a model anything, and it runs **before** the judge. The split is not an optimisation, it is the thing that keeps the harness honest: * A judge is a model reading a model. Ask it "is this answer grounded?" and it will reason about plausibility, because plausibility is what it can see. It cannot know whether the URL in the reply was in any tool result — but we can, exactly, by comparing strings. * Conversely a judge is the only thing that can say whether an answer is *the right answer*. Neither half is sufficient; running the deterministic half first means the judge is never the only thing standing between a fabricated citation and a green check. A :data:`~app.assistant.evaluation.constants.BLOCKING_CHECKS` finding fails the attempt on its own. The judge still runs — its reasoning is what makes the failure diagnosable — but it cannot vote the attempt back up. """ from __future__ import annotations import re from dataclasses import dataclass from typing import Any from app.assistant.evaluation.constants import ( BLOCKING_CHECKS, CHECK_EMPTY_REPLY, CHECK_FORBIDDEN_PHRASE, CHECK_MISSING_ACTION, CHECK_MISSING_PHRASE, CHECK_MISSING_TOOL, CHECK_POLICY_CONFLICT, CHECK_TURN_FAILED, CHECK_UNAVAILABLE, CHECK_UNEXPECTED_ACTION, CHECK_UNGROUNDED_CITATION, CHECK_UNRESOLVED_PLACEHOLDER, CHECK_WRONG_LANGUAGE, ERROR_CHECKS, ) from app.assistant.reply_hygiene import PORTAL, PORTAL_REQUESTS SEVERITY_BLOCKING = "blocking" SEVERITY_ERROR = "error" SEVERITY_INFO = "info" @dataclass(frozen=True) class Finding: code: str severity: str detail: str def as_dict(self) -> dict[str, str]: return {"code": self.code, "severity": self.severity, "detail": self.detail} def _severity(code: str) -> str: if code in ERROR_CHECKS: return SEVERITY_ERROR if code in BLOCKING_CHECKS: return SEVERITY_BLOCKING return SEVERITY_INFO def _finding(code: str, detail: str) -> Finding: return Finding(code=code, severity=_severity(code), detail=detail[:500]) # --------------------------------------------------------------------------- # Citation grounding # --------------------------------------------------------------------------- _URL_RE = re.compile(r"https?://[^\s)\]<>\"'，。、；]+", re.IGNORECASE) _UUID_RE = re.compile( r"\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\b", re.IGNORECASE, ) #: Trailing punctuation a model routinely welds onto a URL. Stripped before #: comparison so ``…/doc)`` and ``…/doc`` are not two different citations. #: Punctuation a model puts AFTER a URL, stripped before comparison. ``*`` #: is in here because ``**https://…/my/requests**`` is ordinary model output and #: the emphasis markers are not part of the address — left in, the portal link #: the prompt requires failed the first-party check and was scored as a #: fabricated citation, which is the defect this whole allowlist exists to fix. _URL_TRAILING = ".,;:!?)]}>*\"'。，、；：！？）】》" def _normalise_url(url: str) -> str: return url.rstrip(_URL_TRAILING).rstrip("/").casefold() def _grounded_in(needle: str, evidence: list[str]) -> bool: folded = needle.casefold() return any(folded in blob.casefold() for blob in evidence) #: Destinations the assistant is REQUIRED to emit, and which therefore cannot #: be fabrications. #: #: The system prompt ends with an unconditional instruction — name Helix and #: give the link, ``/my/requests`` for what they submitted and the portal root #: for their own devices — and ``reply_hygiene.add_missing_destination`` appends #: the same link deterministically when the model dodges it. No tool result #: contains either string, so every correctly-behaved answer that pointed the #: employee somewhere was scored as an ungrounded citation: a BLOCKING check, #: which fails the attempt outright, past the judge, filed under the wrong #: cause. That is a systematic depression of the pass rate produced by the #: assistant doing exactly what it was told. #: #: Deliberately an exact-URL set rather than a host allowlist: a made-up PATH on #: the right host ("helix.pingcap.net/assets/12345") is still a fabricated link #: and still has to be caught. _FIRST_PARTY_DESTINATIONS: frozenset[str] = frozenset( _normalise_url(url) for url in (PORTAL, PORTAL_REQUESTS) ) def _is_first_party(url: str) -> bool: """Is this one of the destinations the prompt requires, query aside? The query and fragment are dropped before the comparison, and only those: ``…/my/requests?tab=all`` is the same page the prompt told the model to point at, and calling it a fabricated citation would put the fix back where it started for a very common shape. The PATH still has to match exactly — ``…/assets/12345`` on the same host is still an invented destination, and an employee cannot tell that from a real one. """ bare = url.split("?", 1)[0].split("#", 1)[0].rstrip("/") return bare in _FIRST_PARTY_DESTINATIONS def citation_findings(reply: str, evidence: list[str]) -> list[Finding]: """Citations in the reply that were in nothing the turn actually read. ``evidence`` is every tool result verbatim, plus the serialised source documents the server pre-fetched. A URL or a UUID that appears in the reply and in none of them was not retrieved — it was produced. That is the single most damaging failure this assistant has, because a fabricated link is indistinguishable from a real one to the employee reading it on a phone, and it is invisible to a judge model that has no way to check. Deliberately narrow. Prose claims are the judge's job; this only asserts things that are exactly comparable. A reply that cites nothing raises nothing here — "unsupported" is a different finding from "fabricated", and the judge owns it. """ if not reply: return [] findings: list[Finding] = [] normalised_evidence = [blob for blob in evidence if blob] seen_urls: set[str] = set() for raw in _URL_RE.findall(reply): url = _normalise_url(raw) if not url or url in seen_urls: continue seen_urls.add(url) if _is_first_party(url): continue if not _grounded_in(url, normalised_evidence): findings.append( _finding( CHECK_UNGROUNDED_CITATION, f"回答里的链接未出现在本轮任何工具结果中:{raw[:200]}", ) ) seen_ids: set[str] = set() for raw in _UUID_RE.findall(reply): ident = raw.casefold() if ident in seen_ids: continue seen_ids.add(ident) if not _grounded_in(ident, normalised_evidence): findings.append( _finding( CHECK_UNGROUNDED_CITATION, f"回答里的 id 未出现在本轮任何工具结果中:{raw}", ) ) return findings # --------------------------------------------------------------------------- # Language # --------------------------------------------------------------------------- # Written as escapes rather than literal characters: these ranges are the whole # discriminator between three of the languages this platform serves, and a # literal 「゠」 in source is one careless re-encoding away from silently # becoming a different range. _KANA_RE = re.compile("[\\u3040-\\u309f\\u30a0-\\u30ff]") _CJK_RE = re.compile("[\\u4e00-\\u9fff\\u3400-\\u4dbf]") _HANGUL_RE = re.compile("[\\uac00-\\ud7af]") #: An English reply may legitimately carry a Chinese product or team name. This #: is the share of CJK characters above which it is no longer an English reply #: that mentions something, but a Chinese reply. _EN_CJK_TOLERANCE = 0.08 #: Below this many characters, script ratios are noise — 「好的。」 is a valid #: reply and proves nothing about language handling. _LANG_MIN_CHARS = 12 def language_findings(reply: str, expected_lang: str) -> list[Finding]: """Whether the reply came back in the language the employee asked in. The employee's language is resolved from their own profile and threaded into the prompt, so answering in the wrong one is a real defect and a common one — this platform serves zh-CN, en-US and ja-JP, and the Japanese surface has historically ended up with Chinese mixed in. Kana is the discriminator that makes this checkable at all: Japanese and Chinese share the han characters, so counting CJK cannot separate them, but natural Japanese prose of any length contains kana and Chinese contains none. """ text = (reply or "").strip() stripped = re.sub(r"[\s\d\W_]+", "", text, flags=re.UNICODE) if len(stripped) < _LANG_MIN_CHARS: return [] has_kana = bool(_KANA_RE.search(text)) cjk_count = len(_CJK_RE.findall(text)) hangul = bool(_HANGUL_RE.search(text)) total = max(len(stripped), 1) if expected_lang == "ja": if not has_kana: return [ _finding( CHECK_WRONG_LANGUAGE, "期望日文,但回答里没有任何假名 —— 这不是日文。", ) ] return [] if expected_lang == "zh": if has_kana: return [ _finding( CHECK_WRONG_LANGUAGE, "期望中文,但回答里出现了假名。", ) ] if cjk_count == 0: return [ _finding( CHECK_WRONG_LANGUAGE, "期望中文,但回答里没有任何汉字。", ) ] return [] if expected_lang == "en": if has_kana or hangul: return [ _finding( CHECK_WRONG_LANGUAGE, "期望英文,但回答里出现了假名/谚文。", ) ] if cjk_count / total > _EN_CJK_TOLERANCE: return [ _finding( CHECK_WRONG_LANGUAGE, f"期望英文,但汉字占比 {cjk_count / total:.0%} —— 这是中文回答。", ) ] return [] # --------------------------------------------------------------------------- # Placeholders # --------------------------------------------------------------------------- #: A template that never got rendered, a Python ``None`` that reached the #: employee, a JS object stringified by accident. Each of these has actually #: shipped somewhere in this industry; none of them is ever a correct reply. _PLACEHOLDER_PATTERNS = ( (re.compile(r"\{\{\s*[\w.]+\s*\}\}"), "未渲染的模板占位符 {{…}}"), (re.compile(r"(?", re.IGNORECASE), "占位标记"), ) def placeholder_findings(reply: str) -> list[Finding]: text = reply or "" findings: list[Finding] = [] for pattern, label in _PLACEHOLDER_PATTERNS: match = pattern.search(text) if match: findings.append( _finding( CHECK_UNRESOLVED_PLACEHOLDER, f"{label}:{match.group(0)[:80]}", ) ) return findings # --------------------------------------------------------------------------- # Expectations # --------------------------------------------------------------------------- def _as_alternatives(entry: Any) -> list[str]: """One expectation entry, as the list of strings that satisfy it. A plain string is one alternative; a list is "any of these". Real specifications need the second form far more often than they look like they will — 「36 个月」 and 「三年」 are the same correct answer, and a harness that cannot say so trains its authors to write expectations so loose they stop catching anything. """ if isinstance(entry, str): return [entry] if entry.strip() else [] if isinstance(entry, (list, tuple)): return [str(item) for item in entry if str(item).strip()] return [] def _normalise_for_match(text: str) -> str: """Fold whitespace and case so a wrapped reply still matches its phrase.""" return re.sub(r"\s+", "", (text or "")).casefold() def expectation_findings( *, reply: str, tool_calls: list[str], proposed_actions: list[dict[str, Any]], expectations: dict[str, Any] | None, escalated: bool, ) -> list[Finding]: """The author's own spec for this case, checked literally.""" spec = expectations or {} findings: list[Finding] = [] folded_reply = _normalise_for_match(reply) for entry in spec.get("must_mention") or []: alternatives = _as_alternatives(entry) if not alternatives: continue if not any(_normalise_for_match(alt) in folded_reply for alt in alternatives): findings.append( _finding( CHECK_MISSING_PHRASE, "回答里缺少必需内容:" + " / ".join(alternatives)[:200], ) ) for entry in spec.get("must_not_mention") or []: for alt in _as_alternatives(entry): if _normalise_for_match(alt) and _normalise_for_match(alt) in folded_reply: findings.append( _finding( CHECK_FORBIDDEN_PHRASE, f"回答里出现了不该出现的内容:{alt[:120]}", ) ) expected_tools = [str(t) for t in (spec.get("expect_tools") or []) if str(t).strip()] if expected_tools: called = set(tool_calls or []) missing = [tool for tool in expected_tools if tool not in called] if missing: findings.append( _finding( CHECK_MISSING_TOOL, "这一轮没有调用应当调用的工具:" + ", ".join(missing)[:200], ) ) expect_action = spec.get("expect_action") if isinstance(expect_action, bool): offered = bool(proposed_actions) if expect_action and not offered: findings.append( _finding( CHECK_MISSING_ACTION, "应当给出可确认的动作卡片,但这一轮只给了文字。", ) ) elif not expect_action and offered: keys = [str(a.get("action_key") or "?") for a in proposed_actions] findings.append( _finding( CHECK_UNEXPECTED_ACTION, "不应当提出动作,但这一轮提出了:" + ", ".join(keys)[:200], ) ) expect_escalate = spec.get("expect_escalate") if isinstance(expect_escalate, bool): if expect_escalate and not escalated: findings.append( _finding( CHECK_MISSING_PHRASE, "应当引导到人工/工单,但回答里没有这层意思。", ) ) elif not expect_escalate and escalated: findings.append( _finding( CHECK_FORBIDDEN_PHRASE, "不应当转工单 —— 这个问题助手自己就该答得了。", ) ) return findings # --------------------------------------------------------------------------- # Escalation detection # --------------------------------------------------------------------------- #: Phrases the assistant uses when it routes to a human. Matched on the folded #: reply, so wrapping and spacing do not matter. Kept as data rather than asked #: of the judge because ``expect_escalate`` is a deterministic expectation and #: must not depend on a second model's mood. _ESCALATION_MARKERS = ( "转人工", "开工单", "提工单", "提交工单", "创建工单", "联系it", "找it", "it支持", "it服务台", "人工支持", "支持工单", "supportticket", "openaticket", "fileaticket", "raiseaticket", "contactit", "ithelpdesk", "問い合わせ", "サポートチケット", ) def looks_escalated(reply: str) -> bool: folded = _normalise_for_match(reply) return any(marker in folded for marker in _ESCALATION_MARKERS) # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- def run_checks( *, reply: str, tool_calls: list[str], sources: list[dict[str, Any]], proposed_actions: list[dict[str, Any]], policy_conflicts: list[dict[str, Any]], evidence: list[str], expectations: dict[str, Any] | None, expected_lang: str, turn_failed: bool, available: bool, canned_replies: tuple[str, ...] = (), ) -> list[Finding]: """Every model-free finding for one attempt, in reporting order. ``canned_replies`` are the channel's own fallback strings. A turn that returned one of them did not answer — counting it as an answer with a low score would put a provider outage on the quality trend line, which is the specific way a harness stops being trusted. """ findings: list[Finding] = [] if not available: findings.append(_finding(CHECK_UNAVAILABLE, "助手未启用或未配置模型后端。")) return findings if turn_failed: findings.append(_finding(CHECK_TURN_FAILED, "这一轮调用模型失败。")) return findings text = (reply or "").strip() if not text: findings.append(_finding(CHECK_EMPTY_REPLY, "回答为空。")) return findings folded = _normalise_for_match(text) for canned in canned_replies: if canned and _normalise_for_match(canned) == folded: findings.append( _finding(CHECK_TURN_FAILED, "返回的是渠道兜底话术,不是一次真正的回答。") ) return findings findings.extend(citation_findings(text, evidence)) findings.extend(language_findings(text, expected_lang)) findings.extend(placeholder_findings(text)) findings.extend( expectation_findings( reply=text, tool_calls=tool_calls or [], proposed_actions=proposed_actions or [], expectations=expectations, escalated=looks_escalated(text), ) ) if policy_conflicts: findings.append( _finding( CHECK_POLICY_CONFLICT, f"这一轮检索到 {len(policy_conflicts)} 条与当前制度冲突的文档。", ) ) return findings def worst_severity(findings: list[Finding]) -> str | None: """``error`` beats ``blocking`` beats ``info``; None when there is nothing.""" severities = {f.severity for f in findings} for level in (SEVERITY_ERROR, SEVERITY_BLOCKING, SEVERITY_INFO): if level in severities: return level return None __all__ = [ "Finding", "SEVERITY_BLOCKING", "SEVERITY_ERROR", "SEVERITY_INFO", "citation_findings", "expectation_findings", "language_findings", "looks_escalated", "placeholder_findings", "run_checks", "worst_severity", ]