Pipeline Sliding : archive v1 + pipeline v2 opérationnel (21 layouts)

This commit is contained in:
2026-06-15 22:09:49 +02:00
commit a95f51dba3
29 changed files with 22355 additions and 0 deletions
+620
View File
@@ -0,0 +1,620 @@
#!/usr/bin/env python3
"""
facilitator.py — Sliding Pipeline v3 · Pernod Ricard
=====================================================
Orchestre les 3 agents du nouveau pipeline :
The Narrator (Agent 1) → Markdown narratif
↓ approbation Bastien
The Designer (Agent 2) → Plan de présentation texte
↓ approbation optionnelle
The Encoder (Agent 3) → YAML valide
render_engine.py → PPTX
Variables .env requises :
MISTRAL_API_KEY
NARRATOR_AGENT_ID
DESIGNER_AGENT_ID
ENCODER_AGENT_ID
OUTPUT_DIR (dossier de sortie des YAML/PPTX, défaut : ./output)
THEME_PATH (défaut : theme.yaml)
COMPONENTS_PATH (défaut : components.yaml)
LAYOUTS_PATH (défaut : layouts.yaml)
"""
import json
import os
import re
import sys
import textwrap
from datetime import datetime
from pathlib import Path
import requests
import yaml
from dotenv import load_dotenv
from typing import Optional, Tuple, Dict, Union, List
load_dotenv()
# ─────────────────────────────────────────────────────────────────────────────
# CONFIGURATION
# ─────────────────────────────────────────────────────────────────────────────
API_KEY = os.getenv("MISTRAL_API_KEY")
NARRATOR_ID = os.getenv("NARRATOR_AGENT_ID")
DESIGNER_ID = os.getenv("DESIGNER_AGENT_ID")
ENCODER_ID = os.getenv("ENCODER_AGENT_ID")
OUTPUT_DIR = Path(os.getenv("OUTPUT_DIR", "./output"))
THEME_PATH = os.getenv("THEME_PATH", "theme.yaml")
COMPONENTS_PATH = os.getenv("COMPONENTS_PATH", "components.yaml")
LAYOUTS_PATH = os.getenv("LAYOUTS_PATH", "layouts.yaml")
HEADERS = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json",
}
VERSION = "3.0"
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
# ─────────────────────────────────────────────────────────────────────────────
# UTILITAIRES D'AFFICHAGE
# ─────────────────────────────────────────────────────────────────────────────
def banner():
print("\n" + "" * 62)
print(f" SLIDING PIPELINE v{VERSION} — Pernod Ricard")
print(f" {datetime.now().strftime('%d/%m/%Y %H:%M')}")
print("" * 62 + "\n")
def section(title: str):
print(f"\n{'' * 62}")
print(f" {title}")
print(f"{'' * 62}\n")
def info(msg: str):
print(f" {msg}")
def ok(msg: str):
print(f"{msg}")
def warn(msg: str):
print(f"{msg}")
def ask(prompt: str) -> str:
try:
return input(f"\n {prompt} ").strip()
except (EOFError, KeyboardInterrupt):
print("\n Session interrompue.")
sys.exit(0)
def display_markdown(text: str, max_lines: int = 60):
"""Affiche le Markdown avec un max de lignes."""
lines = text.splitlines()
if len(lines) <= max_lines:
print(textwrap.indent(text, " "))
else:
preview = "\n".join(lines[:max_lines])
print(textwrap.indent(preview, " "))
print(f"\n ... [{len(lines) - max_lines} lignes supplémentaires — "
f"choisissez [4] pour tout voir]")
def display_plan(text: str, max_lines: int = 80):
"""Affiche le plan du Designer — extrait les lignes SLIDE pour un résumé rapide."""
slides = [l for l in text.splitlines() if l.strip().startswith("SLIDE")]
if slides:
print(" Résumé du plan :\n")
for s in slides:
print(f" {s.strip()}")
print(f"\n ({len(slides)} slides au total)")
else:
display_markdown(text, max_lines)
# ─────────────────────────────────────────────────────────────────────────────
# APPEL AGENT MISTRAL (avec gestion PAUSE/continue)
# ─────────────────────────────────────────────────────────────────────────────
def call_agent(agent_id: str, messages: list) -> str:
"""
Appelle un agent Mistral et gère la pagination PAUSE/continue.
Retourne la réponse complète concaténée.
"""
full_response = ""
current_messages = messages.copy()
while True:
resp = requests.post(
"https://api.mistral.ai/v1/agents/completions",
headers=HEADERS,
json={"agent_id": agent_id, "messages": current_messages},
timeout=120,
)
resp.raise_for_status()
content = resp.json()["choices"][0]["message"]["content"]
full_response += content
# Détection pagination
if "PAUSE" in content and "FIN —" not in content:
info("Pagination détectée → continuation automatique...")
current_messages.append({"role": "assistant", "content": content})
current_messages.append({"role": "user", "content": "continue"})
else:
break
return full_response
def call_with_feedback(agent_id: str, initial_message: str,
feedback: str, previous_output: str) -> str:
"""Relance un agent avec le feedback utilisateur et la sortie précédente."""
messages = [
{"role": "user", "content": initial_message},
{"role": "assistant", "content": previous_output},
{"role": "user", "content": feedback},
]
return call_agent(agent_id, messages)
# ─────────────────────────────────────────────────────────────────────────────
# VALIDATION YAML (The Encoder)
# ─────────────────────────────────────────────────────────────────────────────
def extract_yaml(raw: str) -> str:
"""
Extrait le bloc YAML d'une réponse (enlève les ```yaml ... ``` si présents).
"""
# Bloc markdown yaml
match = re.search(r"```ya?ml\s*(.*?)```", raw, re.DOTALL | re.IGNORECASE)
if match:
return match.group(1).strip()
# Pas de bloc → retourne tel quel
return raw.strip()
def validate_yaml(raw: str, layouts: dict) -> tuple[bool, str, Optional[dict]]:
"""
Valide la syntaxe YAML et vérifie les contraintes de base.
Retourne (is_valid, message_erreur, data_parsée).
"""
try:
content = extract_yaml(raw)
data = yaml.safe_load(content)
except yaml.YAMLError as e:
return False, f"Erreur de syntaxe YAML : {e}", None
if not isinstance(data, dict):
return False, "Le YAML doit être un dictionnaire à la racine.", None
slides = data.get("slides")
if not slides:
return False, "Clé 'slides' manquante ou vide.", None
errors = []
valid_layouts = set(layouts.keys())
for i, slide in enumerate(slides):
pos = slide.get("position", i + 1)
layout = slide.get("layout", "")
if not layout:
errors.append(f"Slide {pos} : clé 'layout' manquante")
continue
if layout not in valid_layouts:
errors.append(f"Slide {pos} : layout '{layout}' inconnu "
f"(valides : {', '.join(sorted(valid_layouts)[:5])}...)")
continue
# Vérification des champs requis
schema = layouts[layout].get("json_schema", {})
required_fields = schema.get("required", [])
for field in required_fields:
if field not in slide:
errors.append(f"Slide {pos} ({layout}) : champ requis '{field}' manquant")
if errors:
return False, "\n ".join(errors), data
return True, f"{len(slides)} slides valides.", data
# ─────────────────────────────────────────────────────────────────────────────
# SAUVEGARDE
# ─────────────────────────────────────────────────────────────────────────────
def save_output(content: str, suffix: str, ext: str) -> Path:
"""Sauvegarde un fichier de sortie avec timestamp."""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = OUTPUT_DIR / f"sliding_{timestamp}_{suffix}.{ext}"
filename.write_text(content, encoding="utf-8")
ok(f"Sauvegardé : {filename}")
return filename
# ─────────────────────────────────────────────────────────────────────────────
# CHARGEMENT DES LAYOUTS (pour validation)
# ─────────────────────────────────────────────────────────────────────────────
def load_layouts() -> dict:
if not os.path.exists(LAYOUTS_PATH):
warn(f"layouts.yaml introuvable ({LAYOUTS_PATH}) — validation désactivée")
return {}
with open(LAYOUTS_PATH, encoding="utf-8") as f:
data = yaml.safe_load(f)
return data.get("layouts", {})
# ─────────────────────────────────────────────────────────────────────────────
# ÉTAPE 1 — THE NARRATOR
# ─────────────────────────────────────────────────────────────────────────────
def run_narrator(brief: str) -> str:
"""
Lance The Narrator sur le brief.
Boucle d'approbation : valider / feedback / régénérer / afficher.
Retourne le Markdown approuvé.
"""
section("ÉTAPE 1 — THE NARRATOR")
info("Génération du Markdown narratif...")
messages = [{"role": "user", "content": brief}]
output = call_agent(NARRATOR_ID, messages)
ok(f"Markdown généré ({len(output)} caractères)")
while True:
print()
display_markdown(output)
print("\n [1] Valider → passer au Designer")
print(" [2] Donner du feedback au Narrator")
print(" [3] Régénérer complètement")
print(" [4] Afficher le Markdown complet")
print(" [5] Sauvegarder le Markdown")
choix = ask("Votre choix :")
if choix == "1":
ok("Markdown approuvé. Passage au Designer...")
save_output(output, "narrator", "md")
return output
elif choix == "2":
feedback = ask("Votre feedback :")
if not feedback:
continue
info("Envoi du feedback au Narrator...")
output = call_with_feedback(NARRATOR_ID, brief, feedback, output)
ok(f"Nouveau Markdown ({len(output)} caractères)")
elif choix == "3":
info("Régénération depuis le brief...")
messages = [{"role": "user", "content": brief}]
output = call_agent(NARRATOR_ID, messages)
ok(f"Nouveau Markdown ({len(output)} caractères)")
elif choix == "4":
print()
print(textwrap.indent(output, " "))
elif choix == "5":
save_output(output, "narrator_draft", "md")
else:
warn("Choix invalide. Entrez 1 à 5.")
# ─────────────────────────────────────────────────────────────────────────────
# ÉTAPE 2 — THE DESIGNER
# ─────────────────────────────────────────────────────────────────────────────
def run_designer(markdown: str) -> str:
"""
Lance The Designer sur le Markdown approuvé.
Boucle d'approbation légère : valider / feedback / afficher.
Retourne le plan texte structuré.
"""
section("ÉTAPE 2 — THE DESIGNER")
info("Génération du plan de présentation...")
messages = [{"role": "user", "content": markdown}]
output = call_agent(DESIGNER_ID, messages)
ok(f"Plan généré ({len(output)} caractères)")
while True:
print()
display_plan(output)
print("\n [1] Valider → passer à l'Encoder")
print(" [2] Donner du feedback au Designer")
print(" [3] Afficher le plan complet")
print(" [4] Sauvegarder le plan")
print(" [0] Revenir au Narrator (modifier le Markdown)")
choix = ask("Votre choix :")
if choix == "1":
ok("Plan approuvé. Passage à l'Encoder...")
save_output(output, "designer", "txt")
return output
elif choix == "2":
feedback = ask("Votre feedback :")
if not feedback:
continue
info("Envoi du feedback au Designer...")
output = call_with_feedback(DESIGNER_ID, markdown, feedback, output)
ok(f"Nouveau plan ({len(output)} caractères)")
elif choix == "3":
print()
print(textwrap.indent(output, " "))
elif choix == "4":
save_output(output, "designer_draft", "txt")
elif choix == "0":
return None # Signal : retour au Narrator
else:
warn("Choix invalide. Entrez 0 à 4.")
# ─────────────────────────────────────────────────────────────────────────────
# ÉTAPE 3 — THE ENCODER
# ─────────────────────────────────────────────────────────────────────────────
def run_encoder(plan: str, layouts: dict) -> tuple[str, dict]:
"""
Lance The Encoder sur le plan du Designer.
Valide le YAML produit. Boucle jusqu'à YAML valide ou abandon.
Retourne (yaml_string, yaml_dict).
"""
section("ÉTAPE 3 — THE ENCODER")
info("Encodage du plan en YAML...")
messages = [{"role": "user", "content": plan}]
output = call_agent(ENCODER_ID, messages)
ok(f"YAML brut reçu ({len(output)} caractères)")
attempts = 0
max_attempts = 3
while attempts < max_attempts:
is_valid, message, data = validate_yaml(output, layouts)
if is_valid:
ok(f"YAML valide : {message}")
yaml_clean = extract_yaml(output)
save_output(yaml_clean, "encoder", "yaml")
return yaml_clean, data
else:
attempts += 1
warn(f"YAML invalide (tentative {attempts}/{max_attempts}) :")
print(f" {message}\n")
if attempts >= max_attempts:
warn("Nombre maximum de tentatives atteint.")
break
info("Correction automatique...")
correction_msg = (
f"Le YAML que tu as généré contient des erreurs :\n\n"
f"{message}\n\n"
f"Corrige ces erreurs et renvoie uniquement le YAML corrigé "
f"sans aucun texte avant ou après."
)
output = call_with_feedback(ENCODER_ID, plan, correction_msg, output)
ok(f"YAML corrigé reçu ({len(output)} caractères)")
# Après max_attempts : choix utilisateur
print()
print(" [1] Accepter le YAML tel quel (avec erreurs)")
print(" [2] Donner un feedback manuel à l'Encoder")
print(" [0] Abandonner et revenir au Designer")
choix = ask("Votre choix :")
if choix == "1":
yaml_clean = extract_yaml(output)
save_output(yaml_clean, "encoder_unvalidated", "yaml")
try:
data = yaml.safe_load(yaml_clean) or {}
except Exception:
data = {}
return yaml_clean, data
elif choix == "2":
feedback = ask("Votre feedback :")
output = call_with_feedback(ENCODER_ID, plan, feedback, output)
yaml_clean = extract_yaml(output)
save_output(yaml_clean, "encoder_manual", "yaml")
try:
data = yaml.safe_load(yaml_clean) or {}
except Exception:
data = {}
return yaml_clean, data
else:
return None, None # Signal : retour au Designer
# ─────────────────────────────────────────────────────────────────────────────
# ÉTAPE 4 — RENDER ENGINE
# ─────────────────────────────────────────────────────────────────────────────
def run_render(yaml_data: dict) -> Optional[Path]:
"""
Appelle render_engine.py avec le YAML validé.
Retourne le chemin du PPTX généré, ou None si erreur.
"""
section("ÉTAPE 4 — RENDER ENGINE")
# Vérifie que render_engine.py est accessible
render_path = Path("render_engine.py")
if not render_path.exists():
warn("render_engine.py introuvable dans le dossier courant.")
warn("Vérifiez que render_engine.py est dans le même dossier que facilitator.py")
return None
# Vérifie les YAML de config
for path in [THEME_PATH, COMPONENTS_PATH, LAYOUTS_PATH]:
if not os.path.exists(path):
warn(f"Fichier YAML manquant : {path}")
return None
# Sauvegarde temporaire du YAML pour render_engine
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
yaml_tmp = OUTPUT_DIR / f"sliding_{timestamp}_input.yaml"
pptx_out = OUTPUT_DIR / f"sliding_{timestamp}.pptx"
if not yaml_data:
warn("yaml_data est vide — impossible de générer le PPTX.")
return None
yaml_content = yaml.dump(yaml_data, allow_unicode=True, default_flow_style=False)
yaml_tmp.write_text(yaml_content, encoding="utf-8")
# Vérification que le fichier n'est pas vide
if yaml_tmp.stat().st_size == 0:
warn(f"Fichier YAML temporaire vide : {yaml_tmp}")
return None
info(f"Lancement de render_engine.py...")
info(f"Sortie : {pptx_out}")
import subprocess
result = subprocess.run(
[
sys.executable, str(render_path),
str(yaml_tmp), str(pptx_out),
"--theme", THEME_PATH,
"--components", COMPONENTS_PATH,
"--layouts", LAYOUTS_PATH,
],
capture_output=True,
text=True,
)
if result.returncode == 0:
ok(result.stdout.strip() or f"PPTX généré : {pptx_out}")
return pptx_out
else:
warn("Erreur dans render_engine.py :")
print(textwrap.indent(result.stderr or result.stdout, " "))
return None
# ─────────────────────────────────────────────────────────────────────────────
# BOUCLE PRINCIPALE
# ─────────────────────────────────────────────────────────────────────────────
def run():
banner()
# Vérification config minimale
missing = [k for k, v in {
"MISTRAL_API_KEY": API_KEY,
"NARRATOR_AGENT_ID": NARRATOR_ID,
"DESIGNER_AGENT_ID": DESIGNER_ID,
"ENCODER_AGENT_ID": ENCODER_ID,
}.items() if not v]
if missing:
for m in missing:
warn(f"Variable .env manquante : {m}")
sys.exit(1)
# Chargement layouts pour validation
layouts = load_layouts()
if layouts:
ok(f"layouts.yaml chargé — {len(layouts)} layouts")
print(" Commandes : brief libre | quit")
print(" Tapez votre brief pour démarrer une nouvelle présentation.\n")
# ── Boucle de session ─────────────────────────────────────
while True:
try:
brief = input(" Vous : ").strip()
except (EOFError, KeyboardInterrupt):
print("\n Session terminée.")
break
if brief.lower() in ("quit", "exit", "q"):
print(" Au revoir.")
break
if not brief:
continue
# ── Pipeline complet ──────────────────────────────────
# ÉTAPE 1 — Narrator
markdown = run_narrator(brief)
if not markdown:
continue
# ÉTAPE 2 — Designer (avec possibilité de revenir au Narrator)
while True:
plan = run_designer(markdown)
if plan is None:
# Retour au Narrator
section("RETOUR AU NARRATOR")
info("Donnez un feedback pour modifier le Markdown :")
feedback = ask("Feedback :")
if feedback:
markdown = call_with_feedback(
NARRATOR_ID, brief, feedback, markdown)
ok(f"Markdown mis à jour ({len(markdown)} caractères)")
continue
# ÉTAPE 3 — Encoder (avec possibilité de revenir au Designer)
yaml_str, yaml_data = run_encoder(plan, layouts)
if yaml_str is None:
# Retour au Designer
info("Retour au Designer...")
continue
break # YAML OK → on sort de la boucle Designer/Encoder
# ÉTAPE 4 — Render
if yaml_data:
pptx_path = run_render(yaml_data)
if pptx_path and pptx_path.exists():
section("PRÉSENTATION GÉNÉRÉE")
ok(f"Fichier PPTX : {pptx_path}")
ok(f"Taille : {pptx_path.stat().st_size / 1024:.1f} Ko")
else:
warn("Le PPTX n'a pas pu être généré.")
info("Le YAML est disponible dans le dossier output/.")
# ── Prochaine présentation ? ──────────────────────────
print()
suite = ask("Générer une nouvelle présentation ? (o/n) :").lower()
if suite not in ("o", "oui", "y", "yes"):
print(" Au revoir.")
break
# ─────────────────────────────────────────────────────────────────────────────
# POINT D'ENTRÉE
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
run()