siop2/apps/ai/src/siop_ai/assistant.py

"""L'assistant (D2 — « sourcé ou silencieux ») et la suggestion de bilan.

- `repondre` : recherche sémantique → extraits au-dessus du seuil de
  pertinence, ou refus HONNÊTE qui dit ce qui a été cherché (écran 2 des
  maquettes). La rédaction est déléguée au `Generateur` (opt-in ADR-004 §3) ;
  sans lui, le mode extractif est la réponse.
- `suggerer_bilan` : similarité sémantique entre la description libre et les
  libellés ACTIFS des référentiels (l'IA ne peut suggérer que des codes
  existants) + comptage des bilans similaires du parc. Sans LLM : rapide,
  déterministe, explicable.
"""

from dataclasses import dataclass

import asyncpg

from .embeddings import Embeddeur
from .generation import Generateur
from .recherche import ExtraitTrouve, chercher

# Répliques des libellés français de @siop/shared (BILAN_FIELD_LABELS) —
# utilisés pour contextualiser les embeddings des codes.
CHAMPS_BILAN = {
    "DOOR_STATE": "état des portes",
    "CABIN_POSITION": "position cabine",
    "ANOMALY": "anomalie constatée",
    "EXTERNAL_CAUSE": "cause extérieure",
    "ACTION_TAKEN": "action réalisée",
    "COMPONENT_CONCERNED": "élément concerné",
}

SEUIL_PERTINENCE = 0.30  # en dessous : le corpus ne porte pas la réponse
SEUIL_SUGGESTION = 0.35
SEUIL_CONFIANCE_FORTE = 0.55


@dataclass
class ReponseAssistant:
    mode: str  # « extractif » | « genere » | « refus »
    answer: str | None
    extraits: list[ExtraitTrouve]
    documents_corpus: int
    bilans_corpus: int


async def _taille_corpus(cnx: asyncpg.Connection) -> tuple[int, int]:
    ligne = await cnx.fetchrow(
        '''
        SELECT
          (SELECT count(DISTINCT "documentId") FROM "RagChunk"
            WHERE "sourceType" = 'DOCUMENT') AS documents,
          (SELECT count(*) FROM "RagChunk" WHERE "sourceType" = 'WORK_ORDER') AS bilans
        '''
    )
    return ligne["documents"], ligne["bilans"]


async def repondre(
    cnx: asyncpg.Connection,
    embeddeur: Embeddeur,
    generateur: Generateur,
    question: str,
    limite: int = 5,
) -> ReponseAssistant:
    documents, bilans = await _taille_corpus(cnx)
    extraits = await chercher(cnx, embeddeur, question, limite)
    pertinents = [e for e in extraits if e.score >= SEUIL_PERTINENCE]

    if not pertinents:
        # D2 : refus explicite — on dit ce qu'on a cherché, on n'invente rien.
        return ReponseAssistant(
            mode="refus",
            answer=None,
            extraits=[],
            documents_corpus=documents,
            bilans_corpus=bilans,
        )

    redige = generateur.rediger(question, pertinents)
    return ReponseAssistant(
        mode="genere" if redige else "extractif",
        answer=redige,
        extraits=pertinents,
        documents_corpus=documents,
        bilans_corpus=bilans,
    )


@dataclass
class SuggestionBilan:
    field: str
    value_id: str
    label: str
    confidence: str  # « FORTE » | « MOYENNE »
    similar_reports: int  # bilans du parc portant déjà ce code (« 9 bilans similaires »)
    score: float


def _cosinus(a: list[float], b: list[float]) -> float:
    return sum(x * y for x, y in zip(a, b))  # vecteurs déjà normés


async def suggerer_bilan(
    cnx: asyncpg.Connection,
    embeddeur: Embeddeur,
    description: str,
) -> list[SuggestionBilan]:
    valeurs = await cnx.fetch(
        'SELECT id, field, label FROM "ReferenceValue" WHERE "isActive" ORDER BY field, label'
    )
    if not valeurs:
        return []

    textes = [description] + [
        f"{CHAMPS_BILAN.get(v['field'], v['field'])} : {v['label']}" for v in valeurs
    ]
    vecteurs = embeddeur.encoder(textes)
    v_description, v_valeurs = vecteurs[0], vecteurs[1:]

    # Le meilleur code par champ, au-dessus du seuil — jamais plus d'une
    # suggestion par champ, jamais un code inventé.
    meilleurs: dict[str, tuple[asyncpg.Record, float]] = {}
    for valeur, vecteur in zip(valeurs, v_valeurs):
        score = _cosinus(v_description, vecteur)
        if score < SEUIL_SUGGESTION:
            continue
        champ = valeur["field"]
        if champ not in meilleurs or score > meilleurs[champ][1]:
            meilleurs[champ] = (valeur, score)

    suggestions: list[SuggestionBilan] = []
    for valeur, score in meilleurs.values():
        # « 9 bilans similaires sur ce parc » : les bilans clôturés portant ce code
        similaires = await cnx.fetchval(
            'SELECT count(*) FROM "RagChunk" WHERE "sourceType" = \'WORK_ORDER\' AND content ILIKE $1',
            f"%{valeur['label']}%",
        )
        suggestions.append(
            SuggestionBilan(
                field=valeur["field"],
                value_id=str(valeur["id"]),
                label=valeur["label"],
                confidence="FORTE" if score >= SEUIL_CONFIANCE_FORTE else "MOYENNE",
                similar_reports=similaires,
                score=round(score, 4),
            )
        )
    suggestions.sort(key=lambda s: s.score, reverse=True)
    return suggestions