Files
sliding-automation/archive/v1_pipeline/prompt_injection.py
T

627 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
prompt_injection.py — Sliding Pipeline · Pernod Ricard
=======================================================
Génère les blocs de contraintes injectés dans les prompts des agents.
Ce script lit les 3 YAML (theme, components, layouts) et produit :
1. agent_constraints.md — fichier de référence complet (debug/doc)
2. prompt_the_designer_injected.md — prompt Designer avec blocs remplis
3. prompt_the_encoder_injected.md — prompt Encoder avec schémas remplis
Usage :
python prompt_injection.py
python prompt_injection.py --theme theme.yaml --components components.yaml
--layouts layouts.yaml
--designer prompt_the_designer.md
--encoder prompt_the_encoder.md
Les fichiers injectés sont prêts à être copiés dans Mistral Studio.
"""
import argparse
import os
import sys
from pathlib import Path
import yaml
# ─────────────────────────────────────────────────────────────────────────────
# DONNÉES ÉDITORIALES (non générables depuis les YAML — savoirs métier)
# ─────────────────────────────────────────────────────────────────────────────
SEQUENCING_RULES = """### Règles de fluidité
- Maximum 2 layouts texte consécutifs : `default_bullets`, `two_cols_text`
- Maximum 2 tableaux consécutifs : `comparison_table`, `benchmark`, `raci_table`, `boxes_grid`
- Après 3 slides denses → intercaler une respiration : `big_stat`, `key_message`, `section_divider`
- Jamais 2 `section_divider` consécutifs
- Le dernier slide de contenu avant `end_slide` doit être narratif (pas un tableau, pas un gantt)
### Règles de choix de layout
- 1 seul chiffre décisif → `big_stat` (jamais `kpi_grid` avec 1 item)
- 2 à 6 indicateurs chiffrés → `kpi_grid`
- Transformation conceptuelle (avant/après) → `from_to` (pas `two_cols_text`)
- Plus de 4 étapes avec timing → `process_arrow` ou `phases_timeline` (pas `numbered_steps`)
- Recommandation unique et précise → `recommendation_card`
- Données à comparer sur plusieurs critères avec plusieurs acteurs → `benchmark`
- Données à comparer en tableau structuré → `comparison_table`
- Si le contenu ne rentre dans aucun layout spécialisé → `default_bullets`
- Citation ou message à marteler seul → `key_message`
- Organigramme de gouvernance → `org_chart`
- Responsabilités par rôle → `raci_table`
### Règles de quantité
- Présentation 20 min → 10 à 15 slides max
- Présentation 10 min → 6 à 10 slides max
- Fusionner si > 15 slides : regrouper les slides proches thématiquement
- Un `###` du Markdown avec un seul chiffre fort → envisager `big_stat` séparé
- Un `##` du Markdown = 1 `section_divider` (sauf présentation < 6 slides)"""
NARRATIVE_PATTERNS = """### Pattern "Problem → Solution → Proof"
Adapté aux présentations de recommandation stratégique.
```
cover_split
executive_summary (synthèse SCR dès le début)
section_divider ("Le problème")
big_stat (chiffre choc)
default_bullets ou comparison_table
section_divider ("Notre réponse")
from_to ou numbered_steps
kpi_grid ou chart_callout (preuve que ça marche)
recommendation_card (ce qu'on demande)
end_slide
```
### Pattern "Roadmap Deck"
Adapté aux présentations de planification / lancement de projet.
```
cover_split
executive_summary (où on va et pourquoi)
kpi_grid (état des lieux chiffré)
phases_timeline ou gantt_timeline
numbered_steps (comment on s'organise)
org_chart ou raci_table (qui fait quoi)
recommendation_card (décisions à prendre)
end_slide
```
### Pattern "Data Storytelling"
Adapté aux présentations de revue de performance ou data governance.
```
cover_split
big_stat (chiffre choc d'entrée)
default_bullets (contexte et enjeux)
kpi_grid (panorama des indicateurs)
chart_callout (analyse d'un graphique clé)
from_to (implication / transformation attendue)
yearly_timeline (historique ou prospective)
end_slide
```
### Pattern "Executive Briefing"
Adapté aux présentations courtes (< 10 slides) pour un CODIR.
```
cover_split
executive_summary
key_message (le So What en 1 slide)
kpi_grid ou big_stat
recommendation_card
end_slide
```
### Règles d'assemblage des patterns
- Les patterns sont des points de départ, pas des contraintes rigides
- Hybrider 2 patterns est possible si le contenu le justifie
- Toujours préserver : cover_split en premier, end_slide en dernier
- Les section_dividers sont optionnels pour les patterns courts (< 8 slides)"""
# ─────────────────────────────────────────────────────────────────────────────
# GÉNÉRATEURS DE BLOCS
# ─────────────────────────────────────────────────────────────────────────────
def generate_layouts_catalogue(layouts: dict) -> str:
"""
Génère le bloc LAYOUTS_CATALOGUE pour le prompt du Designer.
Format compact, optimisé pour la lecture LLM.
"""
lines = []
current_famille = None
for layout_name, cfg in layouts.items():
famille = cfg.get("famille", "")
if famille != current_famille:
lines.append(f"\n#### {famille}")
current_famille = famille
layout_id = cfg.get("id", "")
roles = " / ".join(cfg.get("roles_narratifs", []))
hint = cfg.get("agent_hint", "").strip().replace("\n", " ").replace(" ", " ")
# Tronquer le hint à 120 caractères pour rester compact
if len(hint) > 120:
hint = hint[:117] + "..."
lines.append(f"\n`{layout_name}` ({layout_id}) — Rôles : {roles}")
lines.append(f"{hint}")
return "\n".join(lines)
def generate_yaml_schemas(layouts: dict) -> str:
"""
Génère le bloc YAML_SCHEMAS pour le prompt de l'Encoder.
Pour chaque layout : champs requis / optionnels avec types et contraintes.
"""
lines = []
for layout_name, cfg in layouts.items():
layout_id = cfg.get("id", "")
schema = cfg.get("json_schema", {})
constraints = cfg.get("constraints", {})
if not schema:
continue
lines.append(f"\n### `{layout_name}` ({layout_id})")
# Champs requis
required = schema.get("required", [])
if required:
lines.append(f"**Requis :** {', '.join(required)}")
# Champs optionnels
optional = schema.get("optional", [])
if optional:
lines.append(f"**Optionnels :** {', '.join(optional)}")
# Contraintes importantes
constraint_lines = []
for key, val in constraints.items():
if isinstance(val, dict):
# Schéma imbriqué — on extrait les infos clés
sub_req = val.get("required", [])
sub_opt = val.get("optional", [])
sub_min = val.get("min_items") or val.get(f"{key}_min")
sub_max = val.get("max_items") or val.get(f"{key}_max")
if sub_req or sub_opt:
parts = []
if sub_req:
parts.append(f"requis: {', '.join(str(x) for x in sub_req)}")
if sub_opt:
parts.append(f"optionnels: {', '.join(str(x) for x in sub_opt)}")
constraint_lines.append(f" {key} : {' | '.join(parts)}")
elif key.endswith("_min") or key.endswith("_max"):
field = key.rsplit("_", 1)[0]
bound = key.rsplit("_", 1)[1]
constraint_lines.append(f" {field} : {bound} = {val}")
elif key.endswith("_max_chars"):
field = key.replace("_max_chars", "")
constraint_lines.append(f" {field} : max {val} caractères")
elif key == "items_min":
constraint_lines.append(f" items : min {val}")
elif key == "items_max":
constraint_lines.append(f" items : max {val}")
if constraint_lines:
lines.append("**Contraintes :**")
lines.extend(constraint_lines)
# Exemple YAML minimal
lines.append("**Exemple minimal :**")
lines.append("```yaml")
lines.append(f"layout: {layout_name}")
lines.append(f'titre: "Votre titre affirmatif"')
# Génère quelques champs d'exemple selon le layout
example_fields = _generate_example_fields(layout_name, required, constraints)
lines.extend(example_fields)
lines.append("```")
return "\n".join(lines)
def _generate_example_fields(layout_name: str, required: list, constraints: dict) -> list:
"""Génère des champs d'exemple YAML pour un layout donné."""
examples = {
"cover_split": ['sous_titre: "Présentation au CODIR — juin 2026"'],
"section_divider": ["numero_section: 1"],
"agenda": [
"items:",
' - numero: 1',
' titre: "Contexte et enjeux"',
' - numero: 2',
' titre: "Notre proposition"',
],
"content_marker": ["current_index: 2"],
"end_slide": [
'message: "Merci pour votre attention"',
"next_steps:",
' - texte: "Valider le modèle — juillet"',
' niveau: 1',
],
"default_bullets": [
"bullets:",
' - texte: "Premier argument clé"',
" niveau: 1",
' - texte: "Détail ou preuve"',
" niveau: 2",
],
"two_cols_text": [
"left:",
' titre: "Titre colonne gauche"',
' contenu: "Texte de la colonne gauche..."',
"right:",
' titre: "Titre colonne droite"',
' contenu: "Texte de la colonne droite..."',
],
"key_message": ['message: "Le message clé en une phrase forte."'],
"executive_summary": [
'situation: "État des lieux factuel..."',
'complication: "Le problème ou la tension..."',
'resolution: "La réponse proposée..."',
],
"kpi_grid": [
"items:",
' - titre: "Indicateur 1"',
' valeur: "85%"',
' sous_titre: "Contexte de la valeur"',
' - titre: "Indicateur 2"',
' valeur: "+25%"',
],
"big_stat": [
'valeur: "2 400"',
'label: "jours/homme de réconciliation par an"',
'source: "Estimation interne 2026"',
],
"comparison_table": [
"headers:",
' - "Critère"',
' - "Option A"',
' - "Option B"',
"rows:",
' - ["Coût", "Élevé", "Moyen"]',
' - ["Délai", "3 mois", "6 mois"]',
],
"chart_callout": [
"chart_type: bar",
"data:",
' - label: "T1"',
" valeur: 40",
' - label: "T2"',
" valeur: 65",
'insight: "La croissance s\'accélère au T2 grâce au pilote."',
],
"benchmark": [
"criteria:",
' - "Coût"',
' - "Délai"',
"actors:",
' - "PR"',
' - "Concurrent A"',
"scores:",
" - [80, 60]",
" - [70, 85]",
],
"matrix_2x2": [
"axis_x:",
' label: "Effort"',
"axis_y:",
' label: "Impact"',
"items:",
' - label: "Initiative A"',
" x: 20",
" y: 80",
" taille: 3",
],
"pyramid": [
"levels:",
' - label: "Vision"',
' description: "Callout explicatif optionnel"',
' - label: "Stratégie"',
' - label: "Opérations"',
],
"circular_diagram": [
"segments:",
' - label: "Segment 1"',
' description: "Description courte"',
' - label: "Segment 2"',
' description: "Description courte"',
' - label: "Segment 3"',
' description: "Description courte"',
],
"from_to": [
"pairs:",
' - from: "Situation actuelle"',
' to: "Situation cible"',
' - from: "Processus manuel"',
' to: "Processus automatisé"',
],
"boxes_grid": [
"columns:",
' - "Colonne 1"',
' - "Colonne 2"',
"rows:",
" - label: \"Ligne A\"",
' kpi: "xx%"',
' contents: ["Contenu 1", "Contenu 2"]',
],
"numbered_steps": [
"steps:",
" - numero: 1",
' titre: "Première étape"',
' description: "Ce que ça implique concrètement"',
" - numero: 2",
' titre: "Deuxième étape"',
],
"process_arrow": [
"phases:",
' - label: "Phase 1"',
' duree: "Juin"',
" actif: false",
' bullets: ["Livrable A", "Livrable B"]',
' - label: "Phase 2"',
' duree: "Juil-Sept"',
" actif: true",
],
"gantt_timeline": [
"period:",
' start: "2026-06"',
' end: "2026-12"',
"workstreams:",
' - label: "Workstream 1"',
" tasks:",
' - start: "2026-06"',
' end: "2026-08"',
],
"yearly_timeline": [
"milestones:",
' - annee: "2024"',
' label: "Lancement du projet"',
' - annee: "2025"',
' label: "Pilote Suède"',
" actif: true",
' - annee: "2026"',
' label: "Déploiement nordique"',
],
"phases_timeline": [
"phases:",
' - label: "PREP"',
' periode: "Juin"',
' items: ["Brief équipe", "Setup outil"]',
' - label: "PROD"',
' periode: "Juil-Oct"',
' items: ["Développement", "Tests"]',
],
"org_chart": [
"root:",
' label: "Data Gov Leader"',
" children:",
' - label: "Data Owner Finance"',
' children:',
' - label: "Data Steward"',
' - label: "Data Owner Supply"',
],
"raci_table": [
"roles:",
' - "Data Owner"',
' - "Data Steward"',
' - "IT"',
"tasks:",
' - label: "Définir les règles qualité"',
' raci: ["A", "R", "C"]',
' - label: "Exécuter les contrôles"',
' raci: ["A", "R", "I"]',
],
"decision_tree": [
'question: "Faut-il déployer le pilote en Suède ?"',
"branches:",
" yes:",
' label: "Engagement DG confirmé"',
' options: ["Démarrer en juin", "Allouer 0.5 ETP"]',
" no:",
' label: "Engagement DG manquant"',
' options: ["Reporter à septembre", "Choisir une autre filiale"]',
],
"recommendation_card": [
"numero: 1",
' headline: "TROIS DÉCISIONS AVANT FIN JUIN"',
"bullets:",
' - texte: "Valider le modèle avec les DG locaux"',
" niveau: 1",
' - texte: "Nommer les Data Owners"',
" niveau: 1",
' - texte: "Allouer 0.5 ETP par filiale"',
" niveau: 1",
'cta: "Décider en réunion du 30 juin"',
],
}
return examples.get(layout_name, [])
# ─────────────────────────────────────────────────────────────────────────────
# GÉNÉRATION DES FICHIERS
# ─────────────────────────────────────────────────────────────────────────────
def generate_agent_constraints(layouts: dict, output_path: str):
"""Génère agent_constraints.md — fichier de référence complet."""
lines = [
"# agent_constraints.md",
"# Généré automatiquement par prompt_injection.py",
"# Ne pas modifier manuellement.\n",
"=" * 70,
"## SECTION A — CATALOGUE DES LAYOUTS (pour The Designer)",
"=" * 70,
generate_layouts_catalogue(layouts),
"\n" + "=" * 70,
"## SECTION B — RÈGLES DE SÉQUENÇAGE (pour The Designer)",
"=" * 70,
SEQUENCING_RULES,
"\n" + "=" * 70,
"## SECTION C — PATTERNS NARRATIFS (pour The Designer)",
"=" * 70,
NARRATIVE_PATTERNS,
"\n" + "=" * 70,
"## SECTION D — SCHÉMAS YAML (pour The Encoder)",
"=" * 70,
generate_yaml_schemas(layouts),
]
with open(output_path, "w", encoding="utf-8") as f:
f.write("\n".join(lines))
print(f"✓ agent_constraints.md généré : {output_path}")
def inject_prompt(template_path: str, layouts: dict, output_path: str):
"""
Lit un template de prompt, remplace les placeholders {{...}}
et écrit le prompt injecté.
"""
with open(template_path, encoding="utf-8") as f:
template = f.read()
replacements = {
"{{LAYOUTS_CATALOGUE}}": generate_layouts_catalogue(layouts),
"{{SEQUENCING_RULES}}": SEQUENCING_RULES,
"{{NARRATIVE_PATTERNS}}": NARRATIVE_PATTERNS,
"{{YAML_SCHEMAS}}": generate_yaml_schemas(layouts),
}
injected = template
for placeholder, content in replacements.items():
injected = injected.replace(placeholder, content)
with open(output_path, "w", encoding="utf-8") as f:
f.write(injected)
# Vérifie qu'il ne reste pas de placeholders non résolus
remaining = [p for p in replacements if p in injected]
if remaining:
print(f" ⚠ Placeholders non résolus dans {output_path} : {remaining}")
else:
print(f"✓ Prompt injecté : {output_path}")
# ─────────────────────────────────────────────────────────────────────────────
# RAPPORT DE COHÉRENCE
# ─────────────────────────────────────────────────────────────────────────────
def check_coherence(layouts: dict, components: dict):
"""Vérifie la cohérence entre layouts et components."""
print("\n── Rapport de cohérence ──────────────────────────────────")
layout_to_components = components.get("layout_to_components", {})
all_comp_ids = {v["id"] for v in components.get("components", {}).values()}
errors = 0
for layout_name in layouts:
if layout_name not in layout_to_components:
print(f" ⚠ Layout '{layout_name}' absent de layout_to_components")
errors += 1
else:
comp_ids = layout_to_components[layout_name]
for cid in comp_ids:
if cid not in all_comp_ids:
print(f" ⚠ Composant '{cid}' (layout {layout_name}) introuvable")
errors += 1
if errors == 0:
print(f"{len(layouts)} layouts × {len(all_comp_ids)} composants — aucune erreur")
else:
print(f"{errors} erreur(s) détectée(s)")
print()
# ─────────────────────────────────────────────────────────────────────────────
# CLI
# ─────────────────────────────────────────────────────────────────────────────
def main():
parser = argparse.ArgumentParser(
description="Sliding prompt_injection — Génère les prompts enrichis des agents")
parser.add_argument("--theme",
default="theme.yaml")
parser.add_argument("--components",
default="components.yaml")
parser.add_argument("--layouts",
default="layouts.yaml")
parser.add_argument("--designer",
default="prompt_the_designer.md")
parser.add_argument("--encoder",
default="prompt_the_encoder.md")
parser.add_argument("--output-dir",
default=".",
help="Dossier de sortie des fichiers générés")
args = parser.parse_args()
# Vérification des fichiers sources
for path in [args.theme, args.components, args.layouts,
args.designer, args.encoder]:
if not os.path.exists(path):
print(f"✗ Fichier introuvable : {path}")
sys.exit(1)
out = Path(args.output_dir)
out.mkdir(parents=True, exist_ok=True)
# Chargement des YAML
with open(args.theme, encoding="utf-8") as f:
theme = yaml.safe_load(f)
with open(args.components, encoding="utf-8") as f:
components_data = yaml.safe_load(f)
with open(args.layouts, encoding="utf-8") as f:
layouts_data = yaml.safe_load(f)
layouts = layouts_data["layouts"]
components = components_data
print(f"\n── Sliding prompt_injection ──────────────────────────────")
print(f" Layouts : {len(layouts)}")
print(f" Composants: {len(components.get('components', {}))}")
print()
# Rapport de cohérence
check_coherence(layouts, components)
# Génération des fichiers
generate_agent_constraints(
layouts,
str(out / "agent_constraints.md")
)
inject_prompt(
args.designer,
layouts,
str(out / "prompt_the_designer_injected.md")
)
inject_prompt(
args.encoder,
layouts,
str(out / "prompt_the_encoder_injected.md")
)
print()
print("── Fichiers générés ──────────────────────────────────────")
for f in ["agent_constraints.md",
"prompt_the_designer_injected.md",
"prompt_the_encoder_injected.md"]:
path = out / f
if path.exists():
size = path.stat().st_size
print(f" {f} ({size:,} octets)")
print()
print("✓ Injection terminée. Copiez les prompts _injected.md")
print(" dans le champ 'Instructions' de chaque agent Mistral Studio.")
if __name__ == "__main__":
main()