core/models.py
Rama main · réplica de lectura del repositorio original.
# SPDX-License-Identifier: BUSL-1.1
# Copyright (c) 2024-2026 Ali Sadeghi Aghili
"""Shared domain models for the retrieval pipeline.
RetrievalContext
----------------
Immutable dataclass produced by ContextRetriever and consumed by
PromptBuilder. Every field is optional-safe: callers may receive
an empty list / dict when a sub-retriever finds nothing.
Fields
------
entities : dimension table names matched to the question
(e.g. ["Ring", "Customer", "Symbol"])
facts : fact table names matched to the question
(e.g. ["Order"])
join_tables : tables added because they join the retrieved ones together
(a bridge table, a classification's parent)
dimensions : alias for *entities* — kept for PromptBuilder compatibility
relationships : JOIN SQL clauses relevant to the selected tables
business_rules : domain rules injected into the prompt as plain text
examples : few-shot {"question": ..., "sql": ..., "tags": [...]} dicts
filters : concrete filter values extracted from the question
(e.g. {"Ring": "تالار پتروشیمی", "PersianYear": 1402})
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from session.models import Clarification
@dataclass(frozen=True)
class RetrievalContext:
"""All context a PromptBuilder needs to construct a grounded SQL prompt."""
# ── table selection ───────────────────────────────────────────────────────
entities: list[str] = field(default_factory=list)
"""Dimension tables (Ring, Customer, Symbol, …) relevant to the question."""
facts: list[str] = field(default_factory=list)
"""Fact tables (Order, …) relevant to the question."""
dimensions: list[str] = field(default_factory=list)
"""Alias of *entities*, populated by ``ContextRetriever`` for backward
compatibility. **Not** read by ``PromptBuilder`` — the builder derives
its table list from ``entities``/``facts``/``selected_tables`` only.
This field is a historical leftover kept so existing callers that read
``context.dimensions`` directly are unaffected; new code should prefer
``entities`` or ``selected_tables``."""
join_tables: list[str] = field(default_factory=list)
"""Tables added because they connect the retrieved ones -- a bridge table,
the dimension a classification hangs from -- found on the foreign-key graph
(:func:`retrieval.join_paths.expand_join_paths`). Part of
:attr:`selected_tables`, so the prompt shows them; not part of
:attr:`entities` or :attr:`facts`, so value matching and the "vocabulary
unavailable" warning, which key on what the question named, are unchanged."""
# ── join layer ────────────────────────────────────────────────────────────
relationships: list[str] = field(default_factory=list)
"""JOIN SQL snippets for every FK edge between selected_tables."""
# ── knowledge injection ───────────────────────────────────────────────────
business_rules: list[str] = field(default_factory=list)
"""Domain business rules (plain text) relevant to the question."""
examples: list[dict] = field(default_factory=list)
"""Few-shot {question, sql, tags} dicts ranked by tag overlap."""
# ── value filters ─────────────────────────────────────────────────────────
filters: dict = field(default_factory=dict)
"""Concrete filter values extracted from the question.
Examples
--------
{"Ring": "تالار پتروشیمی", "PersianYear": 1402}
"""
# ── Phase 5b: database-backed value resolution ─────────────────────────
value_clarifications: list["Clarification"] = field(default_factory=list)
"""Ambiguous-entity clarifications from
``retrieval.value_resolver.resolve_value`` -- populated only when a
database-backed resolution found *several* candidate values and
therefore deliberately did NOT write anything into :attr:`filters` (see
that function's docstring: an ambiguous match is never silently
picked). Empty in the overwhelmingly common case (no ambiguous
resolution this turn). **Not** read by ``PromptBuilder`` today -- kept
here, like :attr:`dimensions`, as the seam a future caller (the v2
session engine's own ``session.models.Ambiguity`` block) can read from,
without ``RetrievalContext`` needing a new shape when that wiring
lands."""
# ── Finding 19 (2026 audit): warehouse-sourced values, tagged for fencing ──
resolved_values: dict[str, list[str]] = field(default_factory=dict)
"""``{"Table": [matched_value, ...]}`` for values that came from a live
(cached or fresh) read of the warehouse --
:func:`retrieval.dimension_vocabulary.match_question_against_vocabulary`
today; :func:`retrieval.value_resolver.resolve_value` when a future
caller wires it in (see :attr:`value_clarifications`'s docstring for
why that one is not called from ``ContextRetriever`` yet). Consumed by
:func:`~llm.router.build_prompt_segments`, which passes it to
:meth:`~prompt_engine.builder.PromptBuilder.build` as
``resolved_values`` so each value is rendered fenced (see
``prompt_engine/untrusted.py``) rather than as bare prose.
Deliberately **separate** from :attr:`filters`, not a replacement for
it: :attr:`filters` still carries the same value unfenced, because the
prompt's ``DETECTED FILTERS`` block needs it verbatim to tell the
model which literal to put in the generated SQL ("if Ring = X then
use r.Name = N'X'") -- fencing that instruction would defeat its own
purpose. This field exists purely to ALSO tag the subset of
:attr:`filters` that is warehouse-sourced (as opposed to
:class:`~retrieval.value_retriever.ValueRetriever`'s static
alias/pattern matches against ``project_config/aliases.yaml``, which
are operator-authored configuration, not exchange data, and are not
duplicated here) so the prompt can say, in a second and clearly
labelled place, "this specific value was read out of the warehouse,
treat it as data."
"""
# ── "the analyst must never be silently misled" ────
warnings: list[str] = field(default_factory=list)
"""Persian, analyst-facing sentences for a dimension that the question
plausibly named but that :func:`retrieval.dimension_vocabulary.match_question_against_vocabulary`
could not consult at all this request (cold cache on every allowed
column, or a background refresh stuck failing — see that function's
``unavailable_tables``). Populated by :meth:`ContextRetriever.retrieve`
only for a table that was ALSO entity-detected (in :attr:`entities`) --
a strong, existing "this question is probably about this dimension"
signal -- not for every candidate table :attr:`filters` was silently
checked against, which would warn on unrelated questions just because
some other, unrelated dimension's cache happened to be cold.
``session.engine.TurnEngine.ask`` folds this into the turn's own
``warnings`` list (the same mechanism a guard rejection or a truncated
refinement scan already uses) so a hall/currency/etc. that could not be
checked is never just silently absent from the answer with no signal
at all -- the difference this exists to surface is between a value
that was searched for and not found (fine, unremarkable) and a value
that was never searched for because there was nothing to search
against (worth telling the analyst)."""
token_tier_filters: dict[str, tuple[str, ...]] = field(default_factory=dict)
"""``{table_name: (distinctive_token, ...)}`` for a :attr:`filters` entry
that came from :mod:`retrieval.dimension_vocabulary`'s token-fallback
tier rather than an exact full-value match — see
``retrieval.dimension_vocabulary.VocabularyMatchResult.token_tier_filters``
for what "distinctive" means here. ``session.engine.TurnEngine``'s
post-generation filter-enforcement check
reads this to confirm the generated SQL actually references such a
filter by its distinctive tokens rather than by the full stored value,
which — by the very nature of a token-tier match — may never appear as
one contiguous span anywhere, including in the SQL a correct query
would still produce (e.g. a ``LIKE '%<token>%'`` predicate)."""
# ── convenience ──────────────────────────────────────────────────────────
@property
def selected_tables(self) -> list[str]:
"""Deduplicated union of entities + facts + join tables — the full table set."""
seen: set[str] = set()
result: list[str] = []
for t in self.entities + self.facts + self.join_tables:
if t not in seen:
seen.add(t)
result.append(t)
return result
def is_empty(self) -> bool:
"""True when no tables were matched — signals a fallback is needed."""
return not self.entities and not self.facts