"""Model-phrased result summaries, grounded in the result and never trusted blindly. The database result is the source of truth; `results.summary` turns it into `ResultFacts` (every figure the summary may state, with any comparison arithmetic already done in Python) and a deterministic sentence. This module optionally asks the project's existing LLM client to *phrase* those same facts more naturally, then accepts the phrasing only if it is grounded: * every number in the model's text must be a figure the facts contain (a raw result value, a computed total/difference/percentage, a row count, a limit, or a number that appears in a label or in the question), allowing only display rounding; * it may not claim an increase or decrease the comparison arithmetic does not support; * it must be short, plain prose that never mentions SQL, databases, prompts or models. Anything else - an unavailable client, a timeout, malformed output, an ungrounded figure - falls back to the deterministic sentence. A query never fails because of its summary: `generate_grounded_summary` cannot raise. """ from __future__ import annotations import asyncio import json import math import re from typing import Any from app.core.config import Settings from app.core.logging import get_logger from app.nlp.client import LLMClient from app.results.money import RUPEE from app.planning.reconciliation.text import humanize_identifier from app.results.summary import ( COMPARISON, GROUPED_COMPARISON, ResultFacts, analyze_result, compose_summary, ) logger = get_logger(__name__) SYSTEM_PROMPT = """You are a data assistant explaining the result of a finished report query to a business user. You will be given the user's question, what the query computed, the returned result, facts already computed from that result, and a plain draft summary. Write one or two natural, conversational sentences that directly answer the user's question using ONLY the supplied result and facts. Rules: - Use only values present in the supplied result or facts. Never invent, estimate, round differently, or change a value. - Do not invent causes, explanations, trends, or context that is not in the result. - Do not claim an increase or decrease unless the facts state one. - Preserve the meaning of aggregate operations (a total is a total, an average is an average, a count is a count). - If several periods are present, mention each period with its value. If grouped values are present, mention the important ones (highest, lowest, total) as given in the facts. - Keep it concise. Return plain text only - no markdown, no lists, no headings. - Never mention SQL, databases, tables, queries, prompts, models, or how the answer was produced. Respond with a JSON object of the form {"summary": ""}.""" _FORBIDDEN_TERMS = re.compile( r"\b(?:sql|select\s+\w+|database|db\b|table|query|queries|prompt|model|llm|json|schema|row\s+count|" r"as an ai|language model)\b", re.IGNORECASE, ) # A snake_case or dotted schema identifier ("place_of_supply", "invoices.total"). _IDENTIFIER_IN_TEXT_RE = re.compile(r"\b[a-z]+(?:_[a-z0-9]+)+\b|\b[a-z_]+\.[a-z_]+\b") _INCREASE_WORDS = re.compile(r"\b(?:increase[sd]?|increasing|grew|grow[ns]?|growth|rose|rise[sn]?|rising|up|higher|gain(?:ed|s)?|improv\w*|climb\w*)\b", re.IGNORECASE) _DECREASE_WORDS = re.compile(r"\b(?:decrease[sd]?|decreasing|fell|fall[sn]?|falling|drop\w*|declin\w*|down|lower|loss|lost|shrank|shrink\w*|reduc\w*)\b", re.IGNORECASE) _NUMBER_IN_TEXT_RE = re.compile( r"(? str: """The plain-text summary for a finished result: the model's phrasing when it is available and grounded, else the deterministic sentence. Never raises.""" facts = analyze_result( question=question, data=data, column_kinds=column_kinds, plan_context=plan_context, currency_symbol=currency_symbol, ) deterministic = compose_summary(facts) if not settings.summary_llm_enabled or llm_client is None: return deterministic if getattr(llm_client, "is_external", False) and not settings.summary_llm_allow_external: # Result rows never leave the premises unless explicitly allowed. return deterministic try: candidate = await asyncio.wait_for( llm_client.generate_text( SYSTEM_PROMPT, build_user_prompt(question, data, facts, deterministic, plan_context, settings.summary_llm_max_rows), timeout_seconds=settings.summary_llm_timeout_seconds, ), timeout=float(settings.summary_llm_timeout_seconds) + 1.0, ) except Exception as exc: # noqa: BLE001 - any failure means "use the deterministic summary" logger.info("summary_llm_unavailable", extra={"error_type": exc.__class__.__name__}) return deterministic accepted = _clean(candidate) reason = grounding_violation(accepted, facts, currency_symbol) if reason is not None: logger.info("summary_llm_rejected", extra={"reason": reason}) return deterministic return accepted def build_user_prompt( question: str, data: Any, facts: ResultFacts, deterministic: str, plan_context: dict[str, Any] | None, max_rows: int, ) -> str: """The structured brief the model phrases from. Large row lists are described through the computed facts rather than row by row.""" context = plan_context or {} intent = _describe_intent(facts, context) if isinstance(data, list): if len(data) > max_rows: result_text = ( f"{len(data)} rows (too many to list; use the computed facts below). First {max_rows} rows: " + json.dumps(data[:max_rows], default=str, ensure_ascii=False) ) else: result_text = json.dumps(data, default=str, ensure_ascii=False) else: result_text = json.dumps({"result": data}, default=str, ensure_ascii=False) computed = _describe_facts(facts) return ( f"USER QUESTION:\n{question}\n\n" f"QUERY INTENT:\n{intent}\n\n" f"QUERY RESULT:\n{result_text}\n\n" f"COMPUTED FACTS (already calculated, use these figures as written):\n{computed}\n\n" f"DRAFT SUMMARY (correct but plain; you may rephrase it naturally without changing any value):\n{deterministic}\n\n" "INSTRUCTIONS: Explain the result naturally and answer the question directly, using only the values above." ) def _describe_intent(facts: ResultFacts, context: dict[str, Any]) -> str: parts = [] op = context.get("operation") or "LIST" parts.append(f"operation: {op}") if facts.entity_plural: parts.append(f"entity: {facts.entity_plural}") if facts.metric_label: parts.append(f"metric: {facts.metric_label}") if context.get("group_by"): # Words, never identifiers - the model must not echo "place_of_supply". parts.append(f"grouped by: {', '.join(humanize_identifier(key) for key in context['group_by'])}") if context.get("sort"): parts.append(f"ordered by {humanize_identifier(str(context['sort'].get('field')))} {context['sort'].get('direction')}") if context.get("limit") is not None: parts.append(f"limited to {context['limit']} row(s)") if context.get("periods"): parts.append(f"periods: {', '.join(context['periods'])}") if context.get("comparison"): c = context["comparison"] parts.append(f"comparison: {c.get('current_label')} vs {c.get('previous_label')}") if context.get("classification"): cls = context["classification"] parts.append(f"classification: {' vs '.join(cls.get('labels', []))} ({cls.get('side')})") if context.get("share_of_total"): parts.append("share of the total") if context.get("growth"): parts.append(f"growth per {context['growth']}") if context.get("seasonality"): parts.append(f"seasonality across years {', '.join(str(y) for y in context['seasonality'].get('years', []))}") if facts.period_phrase: parts.append(f"period from the question:{facts.period_phrase}") return "; ".join(parts) def _describe_facts(facts: ResultFacts) -> str: lines = [f"result shape: {facts.shape}", f"rows returned: {facts.row_count}"] if facts.scalar_display is not None: lines.append(f"value: {facts.scalar_display}") for label, display, _raw in facts.items[:20]: lines.append(f"{label}: {display}") if facts.total_display: lines.append(f"total: {facts.total_display}") if facts.highest: lines.append(f"highest: {facts.highest[0]} at {facts.highest[1]}") if facts.lowest: lines.append(f"lowest: {facts.lowest[0]} at {facts.lowest[1]}") if facts.comparison: c = facts.comparison lines.append(f"{c['current_label']}: {c['current_display']}; {c['previous_label']}: {c['previous_display']}") if c.get("change") is None: lines.append("change: not computable") elif c["change"] == 0: lines.append("change: none (unchanged)") else: word = "increase" if c["change"] > 0 else "decrease" pct = f" ({abs(c['change_pct']):,.2f}%)" if c.get("change_pct") is not None else "" lines.append(f"change: {word} of {c['change_display']}{pct}") if facts.tied: lines.append("note: the top values are tied") for key in ("share_pct", "biggest_rise", "biggest_fall", "first", "last"): if facts.extra.get(key) is not None: lines.append(f"{key.replace('_', ' ')}: {facts.extra[key]}") for key, value in facts.extra.items(): if key.endswith(" share_pct") and value is not None: lines.append(f"{key}: {value}") return "\n".join(lines) # --------------------------------------------------------------------------- # Grounding # --------------------------------------------------------------------------- def _clean(text: str) -> str: text = text.strip() # A model that ignored the JSON instruction and wrapped the sentence. if text.startswith("{"): try: parsed = json.loads(text) if isinstance(parsed, dict) and isinstance(parsed.get("summary"), str): text = parsed["summary"].strip() except ValueError: pass text = re.sub(r"[*_`#>]+", "", text) return re.sub(r"\s+", " ", text).strip() _CURRENCY_SIGNS = "$€£₹¥" def grounding_violation(text: str, facts: ResultFacts, currency_symbol: str | None = None) -> str | None: """None when `text` is an acceptable phrasing of `facts`; otherwise a short reason it is not.""" if not text: return "empty" if currency_symbol and any(sign in text for sign in _CURRENCY_SIGNS if sign != currency_symbol): return "wrong currency" # the organisation's money, never re-labelled if len(text) > MAX_SUMMARY_CHARS: return "too long" if _FORBIDDEN_TERMS.search(text): return "mentions internals" if _IDENTIFIER_IN_TEXT_RE.search(text): return "mentions an identifier" # "place_of_supply" is not a word a reader should see allowed = _allowed_numbers(facts) for number in _numbers_in(text): if not _matches_any(number, allowed): return f"ungrounded number {number}" if facts.shape in (COMPARISON, GROUPED_COMPARISON) and facts.comparison: change = facts.comparison.get("change") says_up = _INCREASE_WORDS.search(text) is not None says_down = _DECREASE_WORDS.search(text) is not None if change is None and (says_up or says_down): return "direction claimed without a computable change" if change is not None: if change > 0 and says_down and not says_up: return "claims a decrease against an increase" if change < 0 and says_up and not says_down: return "claims an increase against a decrease" if change == 0 and (says_up or says_down): return "claims a change where there is none" return None def _allowed_numbers(facts: ResultFacts) -> set[float]: allowed: set[float] = set() for value in facts.numbers: allowed.add(abs(value)) for scale in (1e3, 1e5, 1e6, 1e7, 1e9): allowed.add(abs(value) / scale) return allowed def _numbers_in(text: str) -> list[float]: numbers: list[float] = [] for match in _NUMBER_IN_TEXT_RE.finditer(text): raw, unit = match.group(1), (match.group(2) or "").lower() try: value = abs(float(raw.replace(",", ""))) except ValueError: continue numbers.append(value) if unit and unit != "%": numbers.append(value * _SCALES.get(unit, 1)) return numbers def _matches_any(number: float, allowed: set[float]) -> bool: for candidate in allowed: if math.isclose(number, candidate, rel_tol=0.011, abs_tol=0.0051): return True # A value stated to fewer decimals than the display ("₹1.2 lakh" # for 1.23 lakh) - accept only when it is the truncated/rounded form. if candidate and number and math.isclose(round(candidate, 1), number, abs_tol=1e-9): return True if candidate and number and math.isclose(round(candidate), number, abs_tol=1e-9): return True return False