"""The model half of the verdict: is this the right answer? The deterministic checks (:mod:`app.assistant.evaluation.checks`) can prove a citation was fabricated and that a required tool never ran. They cannot say whether 「先去 IT 服务台开工单」 was the right thing to tell someone whose VPN is down. Only a reader can, and at the volume this harness runs at the reader is a model. Three properties keep that honest enough to act on: **The judge is blind to the deterministic findings.** It is not told what the checks concluded. A judge shown "this reply cites a URL that was never retrieved" would agree with it, every time, and the harness would then have one opinion wearing two hats. Judged blind, the two halves are independent, their agreement rate is itself a measurement, and a blocking check never needs the judge's permission to fail an attempt. **The judge only sees what the turn actually read.** It is handed the source documents' real text, not a summary and not the model's own account of them. An answer is grounded when the evidence supports it — not when it sounds like something this company would plausibly do. That instruction is repeated in the system prompt because it is exactly the failure mode of an LLM judge: rewarding fluency. **Same-family bias is acknowledged, not pretended away.** The judge runs on the same provider as the assistant (this platform has one LLM connector, and adding a second vendor for the judge alone would be a new credential, a new bill and a new outage surface). A model grading its own family's output is measurably generous. Three things bound the damage: the deterministic half can fail an attempt without the judge, a hard floor on groundedness/accuracy is enforced server-side rather than left to the judge's own ``verdict`` field, and the console shows failures with their full transcript so a human spot-check of the disagreements is a click and not a project. What this harness measures well is *change over time* against a fixed bank; treat one absolute number with more suspicion than a trend. """ from __future__ import annotations import json import logging from dataclasses import dataclass, field from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from app.assistant.evaluation.constants import ( DIMENSION_MAX, FAILURE_TYPES, JUDGE_DIMENSIONS, JUDGE_TIMEOUT_SECONDS, VERDICT_FAIL, VERDICT_PASS, ) logger = logging.getLogger(__name__) #: How much of one source document the judge is shown. Long enough to check a #: policy claim, short enough that ten sources do not blow the input window. _DOC_CHARS = 1800 _MAX_DOCS = 8 _REPLY_CHARS = 4000 _MAX_JUDGE_TOKENS = 1400 _SYSTEM = """You are a strict QA reviewer for an internal IT assistant used by employees of a ~500-person technology company. You are shown one exchange: what the employee asked, what the assistant replied, and THE EXACT EVIDENCE the assistant retrieved while answering. Judge the reply. You are not the assistant and must not answer the question yourself. THE RULE THAT OVERRIDES EVERY OTHER INSTINCT A claim is grounded only if the supplied evidence supports it. It does not matter that a claim is plausible, well-written, or probably true of companies in general. If the assistant stated a policy, a duration, a price, an owner, an approval path, a system name or an eligibility rule that the evidence does not contain, that is ungrounded — score it as such even though the answer reads well. Fluency is not evidence. This is the single most common way a reviewer like you gets it wrong. WHAT A GOOD ANSWER LOOKS LIKE HERE - It is read on a phone, in chat. Short and direct beats complete and long. A correct answer buried in a numbered procedure is a worse answer. - When the platform can DO the thing, offering to do it beats explaining how to do it. A prose walkthrough where an action was available is a failure of actionability, not a stylistic choice. - When there is genuinely no evidence, saying so and offering a support ticket is the CORRECT answer and must score well. Inventing a confident answer in that situation is the worst possible outcome — worse than saying nothing. - Asking one short clarifying question is correct when the request is genuinely ambiguous, and is padding when it is not. - It answers in the employee's own language. SCORE EACH DIMENSION 0-5 (5 is best) - grounded: every factual claim traceable to the supplied evidence. - accurate: what it says is actually correct given that evidence, including numbers, dates, names and statuses. - complete: it answers what was asked, without omitting a condition that changes the answer. - actionable: the employee knows what happens next; an available action was offered rather than described. - escalation: routing to a human/ticket was offered when and only when it was the right move. - tone: chat-appropriate, concise, no lecture, no meta-commentary. VERDICT "pass" only if a real employee would have been correctly served. Any invented fact, any wrong number or name, or a refusal to answer something the evidence plainly covers, is "fail". When you fail it, classify the underlying cause: - kb_doc: the knowledge base has no document covering this. Someone must write one. - retrieval: a document that would have answered it exists in the evidence but the assistant ignored it, or the search plainly returned the wrong things. - tool_field_or_data: a tool ran and returned wrong, stale or missing data. - action: an action/flow should have been offered or executed and was not, or the wrong one was. - out_of_scope: the platform genuinely does not serve this request, and the reply was wrong only in how it said so. Reply in Chinese in the "reasoning" field: two or three sentences naming the specific defect, quoting the offending phrase. "回答质量不高" is useless; 「说 『满 2 年可回购』,但检索到的文档写的是 36 个月」 is what a reviewer needs.""" @dataclass class JudgeVerdict: """What the judge concluded, already validated and clamped.""" available: bool scores: dict[str, int] = field(default_factory=dict) verdict: str | None = None failure_type: str | None = None hallucinations: list[str] = field(default_factory=list) reasoning: str = "" model: str | None = None error: str | None = None def _clamp_score(raw: Any) -> int | None: try: value = int(round(float(raw))) except (TypeError, ValueError): return None return max(0, min(DIMENSION_MAX, value)) def _exchange_block(setup_turns: list[dict[str, str]] | None, question: str) -> str: lines: list[str] = [] for turn in setup_turns or []: role = "员工" if str(turn.get("role")) == "user" else "助手" content = str(turn.get("content") or "").strip() if content: lines.append(f"{role}:{content[:1200]}") lines.append(f"员工(本次待评价的提问):{question[:1200]}") return "\n".join(lines) def _evidence_block(sources: list[dict[str, Any]]) -> str: """The retrieved documents, in full enough to check a claim against. ``body`` is present only when the caller looked the article up; the simulation itself captures titles and links. The judge is told explicitly which it is looking at, because "the evidence had no body" and "the evidence contradicted the reply" are opposite findings and it must not guess between them. """ if not sources: return "(本轮没有检索到任何文档。)" chunks: list[str] = [] for index, doc in enumerate(sources[:_MAX_DOCS], start=1): title = str(doc.get("title") or "(无标题)") body = str(doc.get("body") or "").strip() url = str(doc.get("url") or "") head = f"[{index}] {title}" if url: head += f" <{url}>" if body: chunks.append(f"{head}\n{body[:_DOC_CHARS]}") else: chunks.append(f"{head}\n(正文未随本次评测载入 —— 不要据此判断内容对错。)") if len(sources) > _MAX_DOCS: chunks.append(f"(另有 {len(sources) - _MAX_DOCS} 条未展示。)") return "\n\n".join(chunks) def _expectation_block(expectations: dict[str, Any] | None) -> str: spec = expectations or {} reference = str(spec.get("reference_answer") or "").strip() notes = str(spec.get("notes") or "").strip() if not reference and not notes: return "" parts = ["\n出题人给的参考(不是唯一正确答案,措辞不必一致,判断实质是否等价):"] if reference: parts.append(f"参考答案:{reference[:1500]}") if notes: parts.append(f"这条用例想考的是:{notes[:600]}") return "\n".join(parts) async def judge_attempt( db: AsyncSession, *, question: str, setup_turns: list[dict[str, str]] | None, reply: str, sources: list[dict[str, Any]], tool_calls: list[str], proposed_actions: list[dict[str, Any]], expectations: dict[str, Any] | None, lang: str, category: str, timeout: float = JUDGE_TIMEOUT_SECONDS, ) -> JudgeVerdict: """Score one reply. Never raises — an unavailable judge is a result. Returns ``available=False`` when no LLM backend is configured, which the runner records as ``error`` rather than ``fail``: an unconfigured judge says nothing about the assistant. """ from app.domains.support.copilot import active_llm_identity, one_shot_json _, model = await active_llm_identity(db) action_summary = ( ", ".join(str(a.get("action_key") or "?") for a in proposed_actions) or "(没有提出任何动作)" ) prompt = "\n\n".join( part for part in ( f"员工语言:{lang};这条用例考察的能力:{category}", "对话:\n" + _exchange_block(setup_turns, question), "助手的回答:\n" + (reply or "")[:_REPLY_CHARS], "助手这一轮调用的工具:" + (", ".join(tool_calls) or "(没有调用任何工具)"), "助手这一轮提出的可确认动作:" + action_summary, "助手这一轮实际读到的证据:\n" + _evidence_block(sources), _expectation_block(expectations), ( '请只输出 JSON:{"scores":{' + ",".join(f'"{dim}":0-5' for dim in JUDGE_DIMENSIONS) + '},"verdict":"pass"|"fail",' '"failure_type":"kb_doc"|"retrieval"|"tool_field_or_data"' '|"action"|"out_of_scope"|null,' '"hallucinations":["回答里没有证据支持的具体说法", ...],' '"reasoning":"中文,两三句,点名具体缺陷"}' ), ) if part ) payload = await one_shot_json( db, _SYSTEM, prompt, timeout=timeout, max_tokens=_MAX_JUDGE_TOKENS, ) if payload is None: return JudgeVerdict( available=False, model=model, error="评委模型不可用或未返回可解析的 JSON。", ) raw_scores = payload.get("scores") scores: dict[str, int] = {} if isinstance(raw_scores, dict): for dim in JUDGE_DIMENSIONS: value = _clamp_score(raw_scores.get(dim)) if value is not None: scores[dim] = value if not scores: # A reply with no usable scores is not a verdict, whatever its # ``verdict`` field says. Recording it as a fail would put the judge's # own malfunction on the assistant's trend line. logger.warning("eval judge: reply carried no usable scores: %s", _short(payload)) return JudgeVerdict( available=True, model=model, error="评委没有给出任何可用的维度分。", reasoning=str(payload.get("reasoning") or "")[:2000], ) verdict_raw = str(payload.get("verdict") or "").strip().lower() verdict = verdict_raw if verdict_raw in (VERDICT_PASS, VERDICT_FAIL) else None failure_raw = payload.get("failure_type") failure = str(failure_raw).strip().lower() if failure_raw else None if failure not in FAILURE_TYPES: failure = None hallucinations = [ str(item)[:300] for item in (payload.get("hallucinations") or []) if isinstance(item, (str, int, float)) and str(item).strip() ][:10] return JudgeVerdict( available=True, scores=scores, verdict=verdict, failure_type=failure, hallucinations=hallucinations, reasoning=str(payload.get("reasoning") or "").strip()[:2000], model=model, ) def _short(payload: dict[str, Any]) -> str: try: return json.dumps(payload, ensure_ascii=False)[:400] except (TypeError, ValueError): return str(payload)[:400] __all__ = ["JudgeVerdict", "judge_attempt"]