"""What the model contributes to an automation exception: a cause when the rules could not name one, and a sentence saying why. PR1 gave every exception a deterministic cause where the error text was unambiguous — ``credential``, ``throttled``, ``target_missing`` and the rest come from :mod:`.taxonomy`'s rule table, which deliberately returns ``None`` rather than guess. That leaves a residue: the errors nobody wrote a rule for, sitting in the workbench as ``unknown`` with the generic "investigate" remedy. Those are exactly the rows an admin has to read raw vendor text to understand, which is the work this whole epic exists to remove. So the model works on **signatures**, not occurrences. That is the whole cost story: a signature seen 4,000 times is one call, not 4,000, and the answer is cached on the exception row (``advice*`` columns, added in PR1 for this). Re-triage happens only when the signature's text changes, which by construction means it is a different problem. Four properties make the output safe to put in front of an admin: **The cause comes from a closed vocabulary.** The model picks from :data:`~.taxonomy.ROOT_CAUSES` or it picks nothing. A free-text cause would break ``REMEDY_BY_CAUSE``'s totality and put a row in the workbench with no next step — the dead end this epic is removing. **It cannot invent a number.** Every integer in the explanation must appear in the fact package, same discipline as ``app/intelligence/contracts.py`` and ``mover_review_ai``. An explanation that overstates how many people are affected is an explanation that gets acted on wrongly. **A rule always beats the model.** When :func:`classify_error` names a cause, that cause stands and the model is not consulted at all. Rules are cheap, deterministic, and were written against real vendor strings; the model is for the residue only. This also means the ledger cost falls as the rule table grows, which is the right direction. **It never closes anything.** Triage writes ``root_cause`` (tagged ``cause_source='ai'``) and an explanation. It does not resolve, does not notify the affected people, and does not touch a human's manual judgement — ``_PROTECTED_CAUSE_SOURCES`` in :mod:`.ingest` already refuses to overwrite ``manual``, and this module re-checks rather than relying on that. Every failure is recorded on the row and surfaced, never swallowed into a blank space that reads like "nothing to worry about". """ from __future__ import annotations import logging import re from dataclasses import dataclass from datetime import UTC, datetime from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.automation_exception.ingest import ( CAUSE_SOURCE_AI, CAUSE_SOURCE_MANUAL, ) from app.domains.automation_exception.models import ( AutomationException, AutomationExceptionOccurrence, ) from app.domains.automation_exception.taxonomy import ( CAUSE_UNKNOWN, ROOT_CAUSES, validate_cause, ) from app.integrations.models import PlatformSetting logger = logging.getLogger(__name__) #: Same platform settings the rest of the AI surface reads. No new #: credential, no new configuration page. _KEY_API = "ai_spending.llm.api_key" _KEY_MODEL = "ai_spending.llm.model" _KEY_REGION = "ai_spending.llm.region" #: Recorded on the row when triage could not run, so the surface can #: say which of these happened instead of rendering an empty cell. FAILURE_NO_KEY = "AI_NO_KEY" FAILURE_MODEL = "AI_MODEL_ERROR" FAILURE_UNGROUNDED = "AI_UNGROUNDED" FAILURE_EMPTY = "AI_EMPTY" _INT_RE = re.compile(r"\d+") _MAX_EXPLANATION_CHARS = 600 #: How much vendor text the model sees. Enough to carry a cause, capped #: because a stack trace can be megabytes and the input guard rejects #: the whole call rather than truncating it. _MAX_ERROR_CHARS = 1200 #: Distinct sample errors shown per signature. The signature already #: collapsed them, so these differ only in the parts that vary per row #: — two is enough to show the shape, more is paying for noise. _MAX_SAMPLES = 2 _SYSTEM = ( "You are an IT platform operations analyst. You are given ONE " "recurring automation failure from an internal IT platform, " "already deduplicated: the same error signature seen many times.\n\n" "Return a single JSON object:\n" ' {"root_cause": "", ' '"confidence": "high"|"medium"|"low", ' '"explanation_zh": "...", "explanation_en": "..."}\n\n' "Rules you must follow:\n" "1. root_cause MUST be exactly one of the allowed values given in " "the user message. If none fits, use \"unknown\".\n" "2. The explanations state WHY this is happening and what an " "administrator should check first. Two sentences at most.\n" "3. You may only cite numbers that appear in the FACTS block. Do " "not compute new numbers, do not estimate, do not mention dates.\n" "4. Do not repeat the raw error text back; the reader can see it.\n" "5. If the evidence does not support a confident cause, say so and " "use \"unknown\" with low confidence. A wrong cause is worse than " "no cause, because it sends someone to fix the wrong thing." ) @dataclass(frozen=True) class TriageFacts: """The only numbers the model is allowed to cite.""" source: str signature: str scope_label: str occurrence_count: int open_occurrence_count: int affected_person_count: int recurrence_count: int samples: tuple[str, ...] integration_label: str | None def citable_numbers(self) -> set[int]: return { self.occurrence_count, self.open_occurrence_count, self.affected_person_count, self.recurrence_count, } async def build_facts( db: AsyncSession, exc: AutomationException, ) -> TriageFacts: """Assemble the fact package for one exception. Sample errors come from the occurrences rather than the exception's own ``sample_error`` so the model sees how the text actually varies across rows — the parts that differ are usually the identifiers, and seeing two makes "this is per-user" versus "this is global" legible. """ rows = ( await db.execute( select(AutomationExceptionOccurrence.error_text) .where( AutomationExceptionOccurrence.exception_id == exc.id, AutomationExceptionOccurrence.error_text.is_not(None), ) .order_by(AutomationExceptionOccurrence.occurred_at.desc()) .limit(20), ) ).scalars().all() seen: list[str] = [] for text in rows: cleaned = (text or "").strip() if not cleaned: continue if cleaned not in seen: seen.append(cleaned[:_MAX_ERROR_CHARS]) if len(seen) >= _MAX_SAMPLES: break if not seen and exc.sample_error: seen.append(exc.sample_error.strip()[:_MAX_ERROR_CHARS]) integration_label: str | None = None if exc.integration_id is not None: from app.integrations.models import Integration # noqa: PLC0415 integration = await db.get(Integration, exc.integration_id) if integration is not None: integration_label = f"{integration.kind}/{integration.provider}" return TriageFacts( source=exc.source, signature=exc.signature, scope_label=exc.scope_key or "(platform-wide)", occurrence_count=exc.occurrence_count, open_occurrence_count=exc.open_occurrence_count, affected_person_count=exc.affected_person_count, recurrence_count=exc.recurrence_count, samples=tuple(seen), integration_label=integration_label, ) def _user_prompt(facts: TriageFacts) -> str: allowed = ", ".join(f'"{c}"' for c in ROOT_CAUSES) lines = [ f"ALLOWED root_cause VALUES: {allowed}", "", "FACTS (the only numbers you may cite):", f"- occurrences_total: {facts.occurrence_count}", f"- occurrences_still_open: {facts.open_occurrence_count}", f"- affected_people: {facts.affected_person_count}", f"- times_this_came_back_after_being_fixed: {facts.recurrence_count}", "", "CONTEXT (not numbers — do not cite):", f"- where it happens: {facts.source}", f"- scope: {facts.scope_label}", ] if facts.integration_label: lines.append(f"- integration: {facts.integration_label}") lines += ["", f"ERROR SIGNATURE: {facts.signature}", "", "SAMPLE ERROR TEXT:"] lines += [f" {s}" for s in facts.samples] or [" (none recorded)"] return "\n".join(lines) def _validate(payload: dict[str, Any], facts: TriageFacts) -> tuple[str, str, str, str]: """Return ``(cause, confidence, explanation_zh, explanation_en)``. Raises ``ValueError`` when the reply is ungrounded or the cause is outside the vocabulary. Rejecting the whole reply rather than keeping the parts that parsed is deliberate: a cause the model reached by inventing a number is not a cause with a small error in it, it is a cause nobody can trust. """ cause = str(payload.get("root_cause") or "").strip().lower() if cause not in ROOT_CAUSES: raise ValueError(f"cause {cause!r} outside the vocabulary") confidence = str(payload.get("confidence") or "").strip().lower() if confidence not in {"high", "medium", "low"}: confidence = "low" zh = str(payload.get("explanation_zh") or "").strip() en = str(payload.get("explanation_en") or "").strip() if not zh and not en: raise ValueError("empty explanation") allowed = facts.citable_numbers() for text in (zh, en): for token in _INT_RE.findall(text): # Four digits and up are years, ids and HTTP-ish tokens, not # claims about this fact package; the prompt forbids dates # and rejecting "2026" would kill correct explanations. if len(token) >= 4: continue if int(token) not in allowed: raise ValueError( f"ungrounded number {token} " f"(published: {sorted(allowed)})", ) return ( cause, confidence, zh[:_MAX_EXPLANATION_CHARS], en[:_MAX_EXPLANATION_CHARS], ) async def _client(db: AsyncSession): from app.domains.bailian.llm_client import BailianLLMClient # noqa: PLC0415 async def _setting(key: str) -> str | None: row = await db.get(PlatformSetting, key) value = row.value if row else None return value if isinstance(value, str) else None api_key = (await _setting(_KEY_API) or "").strip() if not api_key: return None return BailianLLMClient( api_key=api_key, model=await _setting(_KEY_MODEL), region=await _setting(_KEY_REGION), timeout=45.0, purpose="automation_exception.triage", ) def needs_triage(exc: AutomationException) -> bool: """Whether this row would learn anything from a model call. Four reasons to skip, each one a cost the platform should not pay: * a human already judged it — their word stands; * a rule already named the cause — deterministic beats probabilistic and the rule was written against real vendor text; * we already asked about this exact signature — the answer is cached, and the signature changing is what makes it a new question; * the row is resolved — nobody is waiting on an explanation. """ if exc.cause_source == CAUSE_SOURCE_MANUAL: return False if exc.resolved_at is not None: return False if exc.cause_source not in {"none", ""} and exc.root_cause != CAUSE_UNKNOWN: return False return exc.advice_generated_at is None async def triage_exception( db: AsyncSession, exc: AutomationException, ) -> AutomationException: """Ask the model for a cause and an explanation. Never raises. The row is left exactly as usable as it was on every failure path: the deterministic severity, owner and remedy are what the workbench is built on, and this is a layer on top of them. """ if not needs_triage(exc): return exc facts = await build_facts(db, exc) if not facts.samples: # Nothing to reason about. Recording the attempt stops the cron # asking the same empty question every half hour. exc.advice = None exc.advice_model = FAILURE_EMPTY exc.advice_generated_at = datetime.now(UTC) return exc client = await _client(db) if client is None: exc.advice_model = FAILURE_NO_KEY # Deliberately NOT stamping ``advice_generated_at``: the key # being absent is a temporary configuration state, not an # answer. Stamping it would mean these rows are never revisited # once a key is finally pasted in. return exc try: payload = await client.chat_json( system=_SYSTEM, user=_user_prompt(facts), max_tokens=800, temperature=0.1, ) except Exception as call_error: # noqa: BLE001 — recorded, not swallowed logger.warning( "automation-exception triage model call failed (exc=%s): %s", exc.id, call_error, ) exc.advice_model = FAILURE_MODEL return exc try: cause, confidence, zh, en = _validate(payload, facts) except ValueError as bad: logger.warning( "automation-exception triage reply rejected (exc=%s): %s", exc.id, bad, ) exc.advice_model = FAILURE_UNGROUNDED return exc # A human's judgement outranks this even if it landed while the # model was thinking. ``needs_triage`` checked it too; this re-check # closes the window between the two, which is a real window because # a model call takes seconds. if exc.cause_source == CAUSE_SOURCE_MANUAL: return exc if cause != CAUSE_UNKNOWN and confidence in {"high", "medium"}: exc.root_cause = validate_cause(cause) exc.cause_source = CAUSE_SOURCE_AI exc.advice = zh or en exc.advice_refs = [f"confidence:{confidence}", f"cause:{cause}"] exc.advice_model = str(getattr(client, "model", "") or "")[:64] exc.advice_generated_at = datetime.now(UTC) return exc __all__ = [ "FAILURE_EMPTY", "FAILURE_MODEL", "FAILURE_NO_KEY", "FAILURE_UNGROUNDED", "TriageFacts", "build_facts", "needs_triage", "triage_exception", ]