"""No-LLM stub mode. OFF by default, and it must stay that way. WHY THIS EXISTS --------------- Azure OpenAI access is pending, and so is `imh`. Without a model there is no classifier and no prose, so `POST /ask` cannot return anything at all - which means the entire chain either side of the model is unproven: the browser, Caddy, Authelia, the API, entity resolution, Cube, retrieval, the contracts, the banners, the error paths. That is a lot of untested surface to leave until the day the key arrives, and it is all testable now. So this module substitutes the two steps that need a model: * classification -> the class is supplied by the caller (a dropdown in the UI). Not guessed by keywords: a crude keyword classifier would produce a *plausible* label, and a plausible wrong label is the failure this system is built to avoid. Choosing explicitly is honest, and it lets each branch be driven deliberately. * prose -> a fixed placeholder string per class. Deterministic, and written to satisfy the contract for its class. WHAT IT PROVES -------------- The transport chain end to end. Entity resolution. Cube queries and their numbers. Retrieval plumbing and citation assembly. The fixture banner. The scope banners. Contract validation - including its failures, because a stub payload goes through `enforce_contract()` exactly like a generated one and a 422 is a real result, not a bug in the stub. WHAT IT DOES NOT PROVE ---------------------- Anything about the model. Whether the classifier would label the question correctly - the whole point is that a human labelled it. Whether retrieval finds the RIGHT chunk: stub mode retrieves lexically because embedding the question needs the model, and lexical hits are not vector hits. And no prose quality, because there is no prose. It also cannot produce `prerequisites_verbatim` for a Procedural answer. Pulling the prerequisites out of a retrieved chunk is an extraction task, and doing it with a regex would be exactly the "synthesised from fragments" failure the Procedural contract exists to prevent. So the list comes back empty and the answer says so. THE SAFETY RULES STILL RUN -------------------------- `apply_safety_rules()` still runs over the forced classification. Contracts are still validated after assembly and before returning. Retrieval still filters superseded revisions. Nothing here is a bypass - it is the same pipeline with the model-shaped holes filled by constants. TURNING IT OFF AGAIN -------------------- `NO_LLM_STUB=false` in ~/ai/api.env, and restart ai-api. Every answer produced in this mode carries `stub_mode: true` and a banner in the same place, and for the same reason, as the fixture banner: an answer nobody generated is otherwise indistinguishable from one that was. """ from __future__ import annotations import logging from typing import Any from classifier import Classification, apply_safety_rules from config import settings from contracts import QuestionClass log = logging.getLogger("stub") BANNER = ( "NO-LLM STUB MODE. No language model was called. The class was chosen by " "hand, retrieval was lexical rather than semantic, and this text is a fixed " "placeholder - not a generated answer. The figures, citations and document " "identities below are real query results." ) def enabled() -> bool: return settings().no_llm_stub def classify(question: str, forced_class: str | None) -> Classification: """Take the class from the caller and put it through the real safety rules. Confidence is 1.0 because a person chose it, not because anything was inferred. The rules still run: a forced HISTORICAL with no time window still becomes UNCLEAR, which is worth seeing. """ try: klass = QuestionClass(forced_class) if forced_class else QuestionClass.UNCLEAR except ValueError: log.warning("unknown forced class %r - falling back to unclear", forced_class) klass = QuestionClass.UNCLEAR raw = Classification( question_class=klass, confidence=1.0, alternatives={}, entities={}, missing_context=[] if forced_class else ["what you are asking about"], ) result = apply_safety_rules(raw, settings().classifier_confidence_threshold) result.downgraded_reason = ( (result.downgraded_reason + "; ") if result.downgraded_reason else "" ) + "class supplied by the caller - NO_LLM_STUB is on, nothing was classified" return result # --------------------------------------------------------------------------- # One payload builder per class. Each returns what the generate() step would # have returned, minus the prose. The factual fields are attached from evidence # by the caller in agent.py, exactly as they are for a generated answer. # --------------------------------------------------------------------------- def _historical(question: str, evidence: dict[str, Any]) -> dict[str, Any]: row_count = evidence.get("row_count", 0) window = evidence.get("time_window", {}).get("description", "the window queried") if row_count == 0: # Must say so, and must not state a quantity. The contract checks both. answer = ( f"{BANNER} The query returned no records for {window}." ) else: # Row counts and figures live in the structured fields, which the UI # renders. Keeping them out of the prose keeps this string honest - # it is not a summary of anything. answer = ( f"{BANNER} The query ran and returned rows for {window}. " "The figures are in the rows and query fields, not in this text." ) return {"answer": answer} def _reference(question: str, evidence: dict[str, Any]) -> dict[str, Any]: if not evidence.get("citations"): answer = ( f"{BANNER} No documents in the active set matched this question." ) else: answer = ( f"{BANNER} The documents below matched lexically. Whether they " "actually answer the question is not established - that is the " "judgement a language model would have made." ) return {"answer": answer} def _procedural(question: str, evidence: dict[str, Any]) -> dict[str, Any]: """Identity only. No prerequisites, and emphatically no steps. The procedure identity is assembled from the retrieved chunk's own metadata - document number, revision, effective date - which is database content, not generated text. That is the part of a Procedural answer that matters most, and it is the part that needs no model at all. """ chunks = evidence.get("chunks") or [] if not chunks: return { "answer": ( f"{BANNER} No controlled procedure was found in the active " "document set for this question." ), "procedure": None, "prerequisites_verbatim": [], } top = chunks[0] procedure = { "doc_number": top.get("doc_number") or top.get("source_file"), "title": top.get("section_title") or top.get("source_file"), "revision": top.get("revision") or "unknown", "effective_date": top.get("effective_date"), "authorising_role": None, "controlled_copy_location": ( "Controlled copy location is not held in the document index - " "obtain the controlled copy through document control." ), } answer = ( f"{BANNER} A candidate controlled procedure was located and is " "identified below. Prerequisites were not extracted: quoting them " "verbatim is an extraction step this mode does not perform, and " "reconstructing them from the retrieved text is precisely what this " "class forbids. Work from the controlled copy." ) return {"answer": answer, "procedure": procedure, "prerequisites_verbatim": []} def _advisory(question: str, evidence: dict[str, Any]) -> dict[str, Any]: """Evidence rows come from Cube. The deferral is the whole point. Note what is NOT here: no number is selected, ranked or described as suitable. The stub could not do that even if asked - it has no way to form an opinion, which makes this the one class where a stub behaves almost exactly as the real thing should. """ rows = evidence.get("rows") or [] window = evidence.get("time_window", {}).get("description", "the window queried") def _interesting(row: dict[str, Any]) -> list[str]: # Ordering only. It decides which column is shown, never what it says - # the value is whatever Cube returned. keys = [ k for k, v in row.items() if isinstance(v, (int, float)) and not isinstance(v, bool) ] preferred = [ k for k in keys if any(w in k.lower() for w in ("rate", "discharge", "level", "count")) ] return preferred + [k for k in keys if k not in preferred] items: list[dict[str, Any]] = [] for row in rows[:6]: for key in _interesting(row): value = row[key] items.append( { "description": f"{key} over {window}", "metric": key, "value": float(value), "unit": "", "sample_size": len(rows), } ) break if items: answer = ( f"{BANNER} What the history shows over {window} is set out in the " "evidence below. No operating parameter has been selected from it." ) else: answer = ( f"{BANNER} There are no records of station operations for {window}." ) return { "answer": answer, "evidence": items, "documented_limits": [], "deferral": ( "This system does not recommend setpoints or operating parameters. " "That decision needs a competent person with sight of current " "equipment condition and concurrent operations. In this mode it " "could not make a recommendation even if it were permitted to - no " "model was called." ), } BUILDERS = { QuestionClass.HISTORICAL: _historical, QuestionClass.REFERENCE: _reference, QuestionClass.PROCEDURAL: _procedural, QuestionClass.ADVISORY: _advisory, } def generated_fields( klass: QuestionClass, question: str, evidence: dict[str, Any] ) -> dict[str, Any]: """Stand in for the model's half of the payload. Deterministic.""" builder = BUILDERS.get(klass) if builder is None: raise ValueError(f"no stub builder for {klass}") return builder(question, evidence)