feat: encoder structuré via json_schema strict Mistral (C1)
This commit is contained in:
+326
@@ -0,0 +1,326 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
schemas.py — Sliding Pipeline v11 · Chantier C1 (Encoder structuré)
|
||||
===================================================================
|
||||
JSON Schemas stricts par layout pour les structured outputs Mistral
|
||||
(response_format: json_schema, strict: true).
|
||||
|
||||
Principes :
|
||||
- Profil "strict" : additionalProperties=false partout, TOUS les champs
|
||||
en required ; les champs optionnels sont nullables (type: [X, "null"]).
|
||||
Les null sont retirés en post-traitement (encoder_schema.drop_nulls).
|
||||
- Pas de minItems/maxItems dans les schémas (support incertain en mode
|
||||
strict) : les bornes de densité restent des recommandations de prompt,
|
||||
le fitter (chantier C3) gère la dégradation gracieuse.
|
||||
- Les types par champ sont déclarés ici (layouts_v2.yaml ne porte que les
|
||||
noms). verify_against_layouts() contrôle la cohérence au chargement.
|
||||
Cible R13 : migrer ces types dans layouts_v2.yaml (bloc schema:) pour
|
||||
une source unique — ce module deviendra alors un pur générateur.
|
||||
- org_chart : récursion dépliée sur 3 niveaux (root → children →
|
||||
grandchildren), profondeur exacte gérée par render_engine_v2.
|
||||
- matrix_2x2 : axis_x/axis_y/quadrants requis dans le schéma d'encodage
|
||||
(bien qu'optionnels côté moteur) — un libellé vaut mieux qu'un trou.
|
||||
|
||||
Usage :
|
||||
from schemas import get_schema, verify_against_layouts, LAYOUT_SCHEMAS
|
||||
schema = get_schema("kpi_grid") # → dict json_schema strict
|
||||
Python 3.9 compatible. Stdlib uniquement.
|
||||
"""
|
||||
|
||||
import copy
|
||||
|
||||
# ── Helpers de construction ──────────────────────────────────────────────────
|
||||
|
||||
S = {"type": "string"}
|
||||
S_OPT = {"type": ["string", "null"]}
|
||||
I = {"type": "integer"}
|
||||
I_OPT = {"type": ["integer", "null"]}
|
||||
B_OPT = {"type": ["boolean", "null"]}
|
||||
|
||||
|
||||
def arr(item):
|
||||
return {"type": "array", "items": item}
|
||||
|
||||
|
||||
def obj(props):
|
||||
"""Objet strict : toutes les clés en required, aucune clé libre."""
|
||||
return {
|
||||
"type": "object",
|
||||
"properties": props,
|
||||
"required": list(props.keys()),
|
||||
"additionalProperties": False,
|
||||
}
|
||||
|
||||
|
||||
def _bullets():
|
||||
return arr(obj({"texte": S}))
|
||||
|
||||
|
||||
def _org_leaf():
|
||||
return obj({"label": S})
|
||||
|
||||
|
||||
def _org_mid():
|
||||
return obj({"label": S, "children": arr(_org_leaf())})
|
||||
|
||||
|
||||
def _org_root():
|
||||
return obj({"label": S, "children": arr(_org_mid())})
|
||||
|
||||
|
||||
# ── Schémas de contenu par layout (hors clés layout/position, injectées) ────
|
||||
|
||||
LAYOUT_SCHEMAS = {
|
||||
"cover_split": obj({
|
||||
"titre": S,
|
||||
"sous_titre": S_OPT,
|
||||
}),
|
||||
"executive_summary": obj({
|
||||
"titre": S,
|
||||
"situation": S,
|
||||
"complication": S,
|
||||
"resolution": S,
|
||||
}),
|
||||
"section_divider": obj({
|
||||
"titre": S,
|
||||
}),
|
||||
"big_stat": obj({
|
||||
"titre": S,
|
||||
"valeur": S,
|
||||
"description": S_OPT,
|
||||
"source": S_OPT,
|
||||
}),
|
||||
"kpi_grid": obj({
|
||||
"titre": S,
|
||||
"items": arr(obj({"label": S, "valeur": S, "description": S_OPT})),
|
||||
}),
|
||||
"two_cols_text": obj({
|
||||
"titre": S,
|
||||
"left": obj({"titre": S, "bullets": _bullets()}),
|
||||
"right": obj({"titre": S, "bullets": _bullets()}),
|
||||
}),
|
||||
"comparison_table": obj({
|
||||
"titre": S,
|
||||
"headers": arr(S),
|
||||
"rows": arr(obj({"label": S, "values": arr(S)})),
|
||||
}),
|
||||
"key_message": obj({
|
||||
"message": S,
|
||||
"detail": S_OPT,
|
||||
}),
|
||||
"circular_diagram": obj({
|
||||
"titre": S,
|
||||
"segments": arr(obj({"label": S, "description": S_OPT})),
|
||||
}),
|
||||
"default_bullets": obj({
|
||||
"titre": S,
|
||||
"bullets": arr(obj({"texte": S, "niveau": I_OPT})),
|
||||
}),
|
||||
"numbered_steps": obj({
|
||||
"titre": S,
|
||||
"steps": arr(obj({"numero": I_OPT, "titre": S,
|
||||
"description": S_OPT})),
|
||||
}),
|
||||
"process_arrow": obj({
|
||||
"titre": S,
|
||||
"steps": arr(obj({"titre": S, "description": S_OPT})),
|
||||
}),
|
||||
"phases_timeline": obj({
|
||||
"titre": S,
|
||||
"phases": arr(obj({"label": S, "periode": S})),
|
||||
}),
|
||||
"gantt_timeline": obj({
|
||||
"titre": S,
|
||||
"periods": arr(S),
|
||||
"workstreams": arr(obj({
|
||||
"label": S,
|
||||
"tasks": arr(obj({"label": S, "start": I, "end": I})),
|
||||
})),
|
||||
}),
|
||||
"recommendation_card": obj({
|
||||
"numero": I_OPT,
|
||||
"titre": S,
|
||||
"headline": S,
|
||||
"cta": S_OPT,
|
||||
"bullets": _bullets(),
|
||||
}),
|
||||
"end_slide": obj({
|
||||
"message": S,
|
||||
}),
|
||||
"from_to_pairs": obj({
|
||||
"titre": S,
|
||||
"label_from": S_OPT,
|
||||
"label_to": S_OPT,
|
||||
"pairs": arr(obj({"from": S, "to": S})),
|
||||
}),
|
||||
"yearly_timeline": obj({
|
||||
"titre": S,
|
||||
"milestones": arr(obj({"annee": S, "label": S, "actif": B_OPT})),
|
||||
}),
|
||||
"raci_table": obj({
|
||||
"titre": S,
|
||||
"roles": arr(S),
|
||||
"tasks": arr(obj({"label": S, "raci": arr(S)})),
|
||||
}),
|
||||
"org_chart": obj({
|
||||
"titre": S,
|
||||
"root": _org_root(),
|
||||
}),
|
||||
"matrix_2x2": obj({
|
||||
"titre": S,
|
||||
"axis_x": obj({"label": S, "low": S, "high": S}),
|
||||
"axis_y": obj({"label": S, "low": S, "high": S}),
|
||||
"quadrants": obj({"top_left": S, "top_right": S,
|
||||
"bottom_left": S, "bottom_right": S}),
|
||||
"items": arr(obj({"label": S, "x": I, "y": I})),
|
||||
}),
|
||||
"image_split": obj({
|
||||
"titre": S,
|
||||
"image": S,
|
||||
"bullets": _bullets(),
|
||||
"side": {"type": ["string", "null"], "enum": ["left", "right",
|
||||
None]},
|
||||
"legende": S_OPT,
|
||||
}),
|
||||
"image_full": obj({
|
||||
"titre": S,
|
||||
"image": S,
|
||||
"sous_titre": S_OPT,
|
||||
}),
|
||||
"bar_chart": obj({
|
||||
"titre": S,
|
||||
"categories": arr(S),
|
||||
"series": arr(obj({"label": S,
|
||||
"values": arr({"type": "number"})})),
|
||||
"unite": S_OPT,
|
||||
"source": S_OPT,
|
||||
"horizontal": B_OPT,
|
||||
}),
|
||||
"line_chart": obj({
|
||||
"titre": S,
|
||||
"points_x": arr(S),
|
||||
"series": arr(obj({"label": S,
|
||||
"values": arr({"type": "number"})})),
|
||||
"unite": S_OPT,
|
||||
"source": S_OPT,
|
||||
}),
|
||||
"donut_split": obj({
|
||||
"titre": S,
|
||||
"segments": arr(obj({"label": S, "valeur": {"type": "number"}})),
|
||||
"valeur_centrale": S_OPT,
|
||||
"source": S_OPT,
|
||||
}),
|
||||
"waterfall": obj({
|
||||
"titre": S,
|
||||
"depart": obj({"label": S, "valeur": {"type": "number"}}),
|
||||
"marches": arr(obj({"label": S, "delta": {"type": "number"}})),
|
||||
"arrivee": obj({"label": S, "valeur": {"type": "number"}}),
|
||||
"unite": S_OPT,
|
||||
"source": S_OPT,
|
||||
}),
|
||||
"heatmap_table": obj({
|
||||
"titre": S,
|
||||
"headers": arr(S),
|
||||
"rows": arr(obj({"label": S, "scores": arr(I)})),
|
||||
"legende": S_OPT,
|
||||
}),
|
||||
"funnel": obj({
|
||||
"titre": S,
|
||||
"etapes": arr(obj({"label": S, "valeur": {"type": "number"},
|
||||
"description": S_OPT})),
|
||||
"source": S_OPT,
|
||||
}),
|
||||
"agenda": obj({
|
||||
"titre": S,
|
||||
"sections": arr(obj({"numero": I_OPT, "label": S,
|
||||
"duree": S_OPT, "actif": B_OPT})),
|
||||
}),
|
||||
"pyramid": obj({
|
||||
"titre": S,
|
||||
"niveaux": arr(obj({"label": S, "description": S_OPT})),
|
||||
}),
|
||||
}
|
||||
|
||||
# Champ transverse (chantier C3) : notes du présentateur, toujours
|
||||
# optionnel, transcrit par l'Encoder si le plan contient « Notes : ... »
|
||||
# et écrit dans la zone notes PowerPoint par le moteur.
|
||||
for _sch in LAYOUT_SCHEMAS.values():
|
||||
_sch["properties"]["notes"] = {"type": ["string", "null"]}
|
||||
_sch["required"].append("notes")
|
||||
|
||||
|
||||
# ── API publique ─────────────────────────────────────────────────────────────
|
||||
|
||||
def get_schema(layout):
|
||||
"""Schéma JSON strict du contenu d'une slide pour le layout donné.
|
||||
Copie profonde : l'appelant peut annoter sans polluer le référentiel."""
|
||||
if layout not in LAYOUT_SCHEMAS:
|
||||
raise KeyError("Layout sans schéma : %s" % layout)
|
||||
return copy.deepcopy(LAYOUT_SCHEMAS[layout])
|
||||
|
||||
|
||||
def known_layouts():
|
||||
return sorted(LAYOUT_SCHEMAS.keys())
|
||||
|
||||
|
||||
def _nullable(field_schema):
|
||||
t = field_schema.get("type")
|
||||
return isinstance(t, list) and "null" in t
|
||||
|
||||
|
||||
def verify_against_layouts(layouts):
|
||||
"""Croise ce module avec layouts_v2.yaml (dict 'layouts' chargé).
|
||||
Retourne une liste de divergences (vide = cohérent). Non bloquant :
|
||||
l'appelant décide (le facilitator affiche en warning)."""
|
||||
issues = []
|
||||
for name, cfg in layouts.items():
|
||||
if name not in LAYOUT_SCHEMAS:
|
||||
issues.append("layouts_v2 déclare '%s' sans schéma ici" % name)
|
||||
continue
|
||||
declared = set(cfg.get("champs", [])) - {"notes"}
|
||||
here = set(LAYOUT_SCHEMAS[name]["properties"].keys()) - {"notes"}
|
||||
if declared != here:
|
||||
missing = declared - here
|
||||
extra = here - declared
|
||||
if missing:
|
||||
issues.append("%s : champs layouts_v2 absents du schéma : %s"
|
||||
% (name, ", ".join(sorted(missing))))
|
||||
if extra:
|
||||
issues.append("%s : champs du schéma absents de layouts_v2 : %s"
|
||||
% (name, ", ".join(sorted(extra))))
|
||||
# Un champ requis côté layouts_v2 ne doit pas être nullable ici
|
||||
for req in cfg.get("champs_requis", []):
|
||||
prop = LAYOUT_SCHEMAS[name]["properties"].get(req)
|
||||
if prop is not None and _nullable(prop):
|
||||
issues.append("%s : champ requis '%s' nullable dans le schéma"
|
||||
% (name, req))
|
||||
for name in LAYOUT_SCHEMAS:
|
||||
if name not in layouts:
|
||||
issues.append("Schéma '%s' sans entrée layouts_v2" % name)
|
||||
return issues
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import json
|
||||
import sys
|
||||
if len(sys.argv) > 1:
|
||||
print(json.dumps(get_schema(sys.argv[1]), indent=2,
|
||||
ensure_ascii=False))
|
||||
else:
|
||||
print("Layouts couverts (%d) : %s"
|
||||
% (len(LAYOUT_SCHEMAS), ", ".join(known_layouts())))
|
||||
try:
|
||||
import yaml
|
||||
with open("layouts_v2.yaml", encoding="utf-8") as f:
|
||||
layouts = yaml.safe_load(f)["layouts"]
|
||||
issues = verify_against_layouts(layouts)
|
||||
if issues:
|
||||
print("\nDivergences avec layouts_v2.yaml :")
|
||||
for i in issues:
|
||||
print(" ! %s" % i)
|
||||
sys.exit(1)
|
||||
print("Cohérence layouts_v2.yaml : OK (%d layouts)"
|
||||
% len(layouts))
|
||||
except FileNotFoundError:
|
||||
print("layouts_v2.yaml introuvable ici — vérification sautée.")
|
||||
Reference in New Issue
Block a user