Navigation : Index | << 09-Building-CLR
A la fin de ce notebook, vous saurez : 1. Concevoir une architecture multi-agents avec Semantic Kernel 2. Implementer des plugins specialises pour chaque rôle (Admin, Coder, Reviewer) 3. Créer des stratégies de sélection et terminaison personnalisees 4. Orchestrer une conversation AgentGroupChat jusqu’a validation 5. Gerer l’etat d’un notebook via une machine a etats (NotebookState)
@kernel_function et FunctionChoiceBehavior.Auto()| Section | Contenu | Concepts cles |
|---|---|---|
| 1 | Configuration projet | ipywidgets, UI interactive |
| 2 | Configuration LLM | dotenv, endpoints OpenAI |
| 3 | NotebookState | Machine a etats, Papermill |
| 4 | Plugins | Coder, Reviewer, Admin |
| 5 | Stratégies | Sélection, Termination |
| 6 | Agents | ChatCompletionAgent x3 |
| 7 | Orchestration | AgentGroupChat, conversation |
Le graphe Mermaid de la cellule suivante montre la séquence Admin → Coder → Reviewer et la manière dont chaque agent fait progresser l’état partagé du notebook.
| Agent | Rôle | Plugin | Actions principales |
|---|---|---|---|
| AdminAgent | Chef de projet | AdminNotebookPlugin | Editer markdown, approuver/rejeter |
| CoderAgent | Developpeur | CoderNotebookPlugin | Modifier code, finish_implementation |
| ReviewerAgent | Validateur | ReviewerNotebookPlugin | Executer, validate_notebook |
Dans cette section, plusieurs modes vous sont proposés pour définir la tâche à réaliser :
.ipynb.Cliquez sur Valider pour confirmer votre choix. Les cellules suivantes prendront automatiquement en compte cette configuration.
# Cellule Code : Installation des dépendances pour l'interface utilisateur
# ----------------------------------------------------------------------
# Exécutez cette cellule EN PREMIER, puis redémarrez le noyau si nécessaire
# avant d'exécuter la cellule suivante.
print("Dépendances UI installées. Redémarrez le noyau si vous rencontrez des problèmes d'affichage des widgets.")Dépendances UI installées. Redémarrez le noyau si vous rencontrez des problèmes d'affichage des widgets.# %% Configuration du projet - UI
import time
import random
import ipywidgets as widgets
from dotenv import load_dotenv as _load_dotenv
from IPython.display import display
from jupyter_ui_poll import ui_events
# ---- Tâches prédéfinies proposées ----
POSSIBLE_TASKS = [
"Créer un notebook Python qui génère aléatoirement un DataFrame de ventes mensuelles (12 mois), affiche des graphiques d'évolution et exporte un rapport PDF.",
"Créer un notebook Python qui crée un dossier local avec quelques fichiers, puis compresse ce dossier en ZIP, et vérifie la taille et l'intégrité après décompression.",
"Créer un notebook Python qui implémente un mini jeu console (Snake ou Pong) en mode 'demo' et se termine après un certain nombre de 'ticks'.",
"Créer un notebook Python utilisant openpyxl (ou xlsxwriter) pour générer deux fichiers Excel puis les fusionner avec un résumé global.",
"Créer un notebook Python qui télécharge un flux RSS public (p.ex. CNN), stocke les titres dans un CSV, puis génère un nuage de mots (WordCloud).",
"Créer un notebook Python qui requête DBpedia (SPARQL) et affiche un graphique final (Plotly).",
"Créer un notebook Python qui charge le dataset Titanic depuis une URL, effectue une analyse basique et un court modèle de classification.",
"Construire un notebook scikit-learn sur le dataset IRIS et réaliser un court modèle de classification."
]
# ---- Widgets de configuration ----
task_selector = widgets.Dropdown(
options=POSSIBLE_TASKS,
description='Tâche :',
style={'description_width': 'initial'}
)
custom_task = widgets.Textarea(
placeholder="Décrivez votre projet en détail...",
layout={'width': '90%', 'height': '120px'}
)
uploader = widgets.FileUpload(accept='.ipynb', multiple=False)
submit_btn = widgets.Button(description="Valider", button_style='success', icon='rocket')
tabs = widgets.Tab()
tabs.children = [
widgets.VBox([widgets.HTML("<i>Une tâche aléatoire sera générée</i>")]),
widgets.VBox([widgets.Label("Choisissez une tâche type :"), task_selector]),
widgets.VBox([widgets.Label("Écrivez vos instructions :"), custom_task]),
widgets.VBox([widgets.Label("Uploader votre notebook :"), uploader])
]
tabs.set_title(0, '🎲 Aléatoire')
tabs.set_title(1, '📚 Bibliothèque')
tabs.set_title(2, '✨ Personnalisé')
tabs.set_title(3, '📤 Importer')
# ---- Stockage de la configuration ----
config = {
'mode': None,
'task_description': None,
'uploaded_file': None
}
config_ready = False
def on_submit(_):
"""Callback déclenché au clic du bouton."""
global config_ready
try:
config['mode'] = tabs.selected_index
if config['mode'] == 3:
if uploader.value:
config['uploaded_file'] = uploader.value[0]
elif config['mode'] == 0:
config['task_description'] = random.choice(POSSIBLE_TASKS)
elif config['mode'] == 1:
config['task_description'] = task_selector.value
elif config['mode'] == 2:
config['task_description'] = custom_task.value
submit_btn.disabled = True
print("Configuration validée !")
except Exception as e:
print(f"Erreur pendant la configuration : {e}")
config_ready = True
submit_btn.on_click(on_submit)
# ---- Chemin headless (réexécution automatique) ----
# Charger le fichier partagé avant de décider si l'exécution est interactive.
# L'API OpenAI officielle laisse OPENAI_BASE_URL vide : la présence de la clé
# est donc le signal fiable, commun à la cellule de configuration LLM suivante.
import os as _os
_load_dotenv("../.env")
if _os.environ.get("OPENAI_API_KEY"):
_tache = _os.environ.get("NOTEBOOKMAKER_TASK", "")
config['mode'] = 2 if _tache else 0
config['task_description'] = _tache or random.choice(POSSIBLE_TASKS)
config_ready = True
print(
f"[headless] Tâche fixée sans UI (mode {config['mode']}) : "
f"{config['task_description'][:70]}"
)
# ---- Affichage ----
display(widgets.HTML("<h3>🔧 Configuration du Projet</h3>"))
display(tabs)
display(submit_btn)
# ---- Boucle bloquante synchrone ----
print("En attente du clic sur Valider...")
with ui_events() as poll:
while not config_ready:
poll(10)
time.sleep(0.1)
print("✅ Config terminée, vous pouvez exécuter les cellules suivantes !")[headless] Tâche fixée sans UI (mode 0) : Créer un notebook Python utilisant openpyxl (ou xlsxwriter) pour générEn attente du clic sur Valider...
✅ Config terminée, vous pouvez exécuter les cellules suivantes !Dans cette section, nous allons :
.env est présent (et déjà configuré) ou non.OPENAI_API_KEY),OPENAI_BASE_URL),OPENAI_CHAT_MODEL_ID)..env une fois la configuration validée.Les cellules ultérieures se baseront sur ces informations pour orchestrer les agents.
import os
import time
import requests
from dotenv import load_dotenv
import ipywidgets as widgets
from IPython.display import display
from jupyter_ui_poll import ui_events
# -------------------------------------------------------------------------
# Fonctions utilitaires
# -------------------------------------------------------------------------
def list_models(api_base, api_key):
"""Retourne un dict avec la liste des modèles ou un champ 'error'."""
url = f"{api_base}/models"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
try:
resp = requests.get(url, headers=headers, timeout=20)
if resp.status_code == 200:
return resp.json() # dict, ex: {"data":[...], "object":"list"}
else:
return {"error": f"status={resp.status_code}", "text": resp.text}
except Exception as e:
return {"error": str(e)}
# -------------------------------------------------------------------------
# Lecture .env
# -------------------------------------------------------------------------
load_dotenv("../.env")
sd_fake = "sk-proj-1234567890"
openai_api_key = os.getenv("OPENAI_API_KEY", sd_fake).strip()
openai_base_url = os.getenv("OPENAI_BASE_URL", "").strip()
openai_chat_model_id = os.getenv("OPENAI_CHAT_MODEL_ID", "gpt-5.6-luna").strip()
already_configured = (
openai_api_key != sd_fake
or (openai_base_url not in ["", "https://api.openai.com/v1"])
)
# Flag indiquant quand la config est OK
env_config_ready = False
# -------------------------------------------------------------------------
# Widgets
# -------------------------------------------------------------------------
message_output = widgets.Output()
api_key_input = widgets.Password(
value="",
placeholder=f"Ex: {sd_fake}",
description="Clé API :",
layout={'width': '80%'}
)
base_url_input = widgets.Text(
value=openai_base_url if already_configured else "https://api.openai.com/v1",
placeholder="ex: https://api.my-llm.com/v1",
description="Endpoint :",
layout={'width': '80%'}
)
model_dropdown = widgets.Dropdown(
options=[], # Vide initialement
description="Modèle :",
layout={'width': '80%', 'display': 'none'} # masqué tant qu'on n'a pas listé
)
list_models_btn = widgets.Button(
description="Lister modèles",
button_style='info',
icon='search'
)
validate_llm_btn = widgets.Button(
description="Enregistrer .env",
button_style='success',
icon='save'
)
ui_box = widgets.VBox([
api_key_input,
base_url_input,
model_dropdown,
widgets.HBox([list_models_btn, validate_llm_btn])
])
# -------------------------------------------------------------------------
# Callbacks
# -------------------------------------------------------------------------
def on_list_models_click(_):
"""Appelé au clic sur 'Lister modèles'."""
new_base_url = base_url_input.value.strip()
new_api_key = api_key_input.value.strip() or openai_api_key
with message_output:
message_output.clear_output()
if not new_base_url or not new_api_key:
print("⚠️ Veuillez saisir un Endpoint et une clé API avant de lister les modèles.")
return
info = list_models(new_base_url, new_api_key)
if "error" in info:
print(f"Erreur /models: {info['error']} - {info.get('text','')}")
else:
data_list = info.get("data", [])
if not data_list:
print("Aucun modèle n'a été retourné (data=[]).")
else:
model_ids = [m.get("id", "(inconnu)") for m in data_list]
model_dropdown.options = model_ids
# Si le .env mentionne déjà un modèle existant, on le sélectionne
if openai_chat_model_id in model_ids:
model_dropdown.value = openai_chat_model_id
else:
model_dropdown.value = model_ids[0]
model_dropdown.layout.display = 'block'
print(f"✅ {len(model_ids)} modèle(s) trouvé(s).")
def on_validate_llm_click(_):
"""Appelé au clic sur 'Enregistrer .env'."""
global env_config_ready
new_api_key = api_key_input.value.strip() or openai_api_key
new_base_url = base_url_input.value.strip()
chosen_model = "gpt-5.6-luna"
if model_dropdown.options and (model_dropdown.layout.display != 'none'):
chosen_model = model_dropdown.value.strip()
with message_output:
message_output.clear_output()
try:
with open('.env', 'w', encoding='utf-8') as f:
f.write(f"OPENAI_API_KEY={new_api_key}\n")
f.write(f"OPENAI_BASE_URL={new_base_url}\n")
f.write(f"OPENAI_CHAT_MODEL_ID={chosen_model}\n")
print("✅ Fichier .env créé/mis à jour avec :")
print(f" - OPENAI_API_KEY = {'configuree' if new_api_key and new_api_key != 'sk-fake' else '(non configuree)'}")
print(f" - OPENAI_BASE_URL = {new_base_url or '(API OpenAI officiel)'}")
print(f" - OPENAI_CHAT_MODEL_ID = {chosen_model}")
api_key_input.value = ""
api_key_input.value = ""
env_config_ready = True
except Exception as e:
print(f"❌ Erreur lors de l'écriture du fichier .env: {str(e)}")
# -------------------------------------------------------------------------
# Suppression des anciens callbacks (si la cellule est rejouée)
# -------------------------------------------------------------------------
list_models_btn._click_handlers.callbacks = []
validate_llm_btn._click_handlers.callbacks = []
list_models_btn.on_click(on_list_models_click)
validate_llm_btn.on_click(on_validate_llm_click)
# -------------------------------------------------------------------------
# Affichage
# -------------------------------------------------------------------------
# Imprimer un message d'intro
if already_configured:
print("✅ Configuration LLM détectée dans .env :")
print(f" - OPENAI_API_KEY = {'configuree' if openai_api_key and openai_api_key != sd_fake else '(non configuree)'}")
print(f" - OPENAI_BASE_URL = {openai_base_url or '(API officielle)'}")
print(f" - OPENAI_CHAT_MODEL_ID = {openai_chat_model_id}")
print("Aucune saisie supplémentaire n'est requise.")
env_config_ready = True
else:
print("Veuillez :\n1) Saisir votre Endpoint et clé API")
print("2) Cliquer sur [Lister modèles] (pour un endpoint custom)")
print("3) Cliquer sur [Enregistrer .env] pour finaliser la configuration")
display(ui_box)
display(message_output)
# -------------------------------------------------------------------------
# Boucle bloquante: attend le clic sur "Enregistrer .env"
# -------------------------------------------------------------------------
with ui_events() as poll:
while not env_config_ready:
poll(10)
time.sleep(0.1)
# Sortie de la boucle => On peut masquer ui_box (facultatif)
ui_box.layout.display = 'none'
print("✅ Configuration LLM terminée, vous pouvez exécuter la suite !")✅ Configuration LLM détectée dans .env :
- OPENAI_API_KEY = configuree
- OPENAI_BASE_URL = (API officielle)
- OPENAI_CHAT_MODEL_ID = gpt-5.6-terra
Aucune saisie supplémentaire n'est requise.La configuration étant terminée, les étapes suivantes vont se lancer :
NotebookState pour piloter le cycle de vie du notebook.specified, implemented, tested, validated).AdminAgent, CoderAgent, ReviewerAgent).Objectif final : Obtenir un notebook entièrement fonctionnel et validé.
Nous installons ici les bibliothèques nécessaires :
Installation terminée. Si nécessaire, redémarrez le kernel pour activer les nouveaux packages.Nous importons :
semantic-kernel pour la gestion de nos agents conversationnels.# Cellule Code : Imports et configuration du logger
# -------------------------------------------------
import os
import json
import hashlib
import logging
import nbformat
import papermill as pm
import random
from datetime import datetime
# Imports liés à Semantic Kernel
from semantic_kernel import Kernel
from semantic_kernel.agents import ChatCompletionAgent, AgentGroupChat
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
from semantic_kernel.functions import kernel_function
from semantic_kernel.agents.strategies.selection.selection_strategy import SelectionStrategy
from semantic_kernel.agents.strategies.termination.termination_strategy import TerminationStrategy
from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior
from semantic_kernel.functions.kernel_arguments import KernelArguments
class ColorFormatter(logging.Formatter):
"""
Un formatter coloré pour rendre les logs plus lisibles.
"""
colors = {
'DEBUG': '\033[94m',
'INFO': '\033[92m',
'WARNING': '\033[93m',
'ERROR': '\033[91m',
'CRITICAL': '\033[91m\033[1m'
}
reset = '\033[0m'
def format(self, record: logging.LogRecord) -> str:
msg = super().format(record)
return f"{self.colors.get(record.levelname, '')}{msg}{self.reset}"
logger = logging.getLogger("Orchestration")
logger.setLevel(logging.INFO)
if not logger.handlers:
handler = logging.StreamHandler()
handler.setLevel(logging.INFO)
formatter = ColorFormatter(
fmt="%(asctime)s [%(levelname)s] %(name)s - %(message)s",
datefmt="%H:%M:%S"
)
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.info("Configuration initiale terminée (niveau INFO).")NotebookState : gestion de l’état du notebookCette classe regroupe :
.ipynb,specified, implemented, tested, validated,Elle centralise toutes les opérations afin que chaque agent puisse y accéder.
# --- Hygiene helpers (issue #6443, Stop & Repair type C) ---
# Module-level redaction of PII-lite machine paths from any text before it reaches
# the log pipeline. Patterns target the well-known Python/AppData and home-dir
# layouts seen when a subprocess emits a path on Windows or Unix.
import re as _re_redact
_REDACT_PATH_PATTERNS = [
# Windows: prefix 'C:\\Users\\' then user-name then '\\...' remainder
_re_redact.compile(r'(C:\\Users\\)([^\\/\s"\']+)(\\[^\s"\']*)'),
# Windows JSON-echappe : le dump log_notebook_state (json.dumps) double les antislashes
_re_redact.compile(r'(C:\\\\Users\\\\)([^\\/\s"\']+)(\\\\[^\s"\']*)'),
# Unix: /home/<name>/... and /Users/<name>/...
_re_redact.compile(r'(/home/)([^/\s"\']+)(/[^\s"\']*)'),
_re_redact.compile(r'(/Users/)([^/\s"\']+)(/[^\s"\']*)'),
]
def _redact_paths(text):
"""Remplace les chemins machine (Windows ``C:\\Users\\<name>\\...`` et Unix
``/home/<name>/...`` / ``/Users/<name>/...``) par le placeholder ``<USER>`` avant
journalisation. Représentation volontairement conservative et content-based
(cf secrets-hygiene règle 6, Stop & Repair type C : on corrige la source,
jamais la sortie)."""
if not text:
return text
redacted = text
for pat in _REDACT_PATH_PATTERNS:
redacted = pat.sub(lambda m: f"{m.group(1)}<USER>{m.group(3)}", redacted)
return redacted
# #6443 — hygiène des sorties pip (chemin machine / build bdist_wheel).
# `wordcloud` et autres paquets à binaire émettent via setup.py des avertissements
# qui contournent `pip install --quiet` et fuient un chemin <user>/AppData/.../Scripts.
import re as _re_pip_noise
_PIP_BUILD_NOISE = _re_pip_noise.compile(
r"(is installed in|Scripts'|DEPRECATION: Building|bdist_wheel|"
r"Collecting |Downloading |Successfully installed|building \w+|"
r"running \w+|creating \.|writing |copying )",
_re_pip_noise.IGNORECASE,
)
class _PipBuildNoiseFilter(logging.Filter):
"""Filtre de log : supprime les enregistrements Papermill porteurs de bruit build-pip."""
def filter(self, record: logging.LogRecord) -> bool:
msg = record.getMessage()
for line in msg.split("\n"):
if line and not _PIP_BUILD_NOISE.search(line):
return True # au moins une ligne légitime : on garde l'enregistrement
return False # que du bruit build-pip : on drop
_PIP_BUILD_NOISE_FILTER = _PipBuildNoiseFilter()
class NotebookState:
"""
Gère le statut et le contenu d'un notebook.
Les états possibles sont :
- 'specified'
- 'implemented'
- 'tested'
- 'validated'
Attributes:
notebook_path (str): Chemin du fichier notebook (.ipynb).
_cached_notebook (nbformat.NotebookNode): Le contenu du notebook stocké en mémoire.
_status (str): L'état courant du notebook.
_previous_states (list[str]): Historique simple des états antérieurs.
"""
VALID_STATES = ["specified", "implemented", "tested", "validated"]
def __init__(self, notebook_path: str) -> None:
self.notebook_path = notebook_path
self._cached_notebook = None
self._status = "specified"
self._previous_states = []
self._load_notebook()
def _load_notebook(self) -> None:
"""Charge le notebook depuis le chemin spécifié, en utilisant nbformat."""
if not os.path.exists(self.notebook_path):
raise FileNotFoundError(f"Notebook introuvable: {self.notebook_path}")
with open(self.notebook_path, "r", encoding="utf-8") as f:
self._cached_notebook = nbformat.read(f, as_version=4)
logger.debug(
f"[NotebookState] Chargé '{self.notebook_path}' avec {len(self._cached_notebook.cells)} cellules."
)
def save_notebook(self, path: str = "") -> None:
"""
Enregistre le notebook sur disque (par défaut au même chemin).
"""
if not path:
path = self.notebook_path
try:
with open(path, "w", encoding="utf-8") as f:
nbformat.write(self._cached_notebook, f)
logger.info(f"[NotebookState] Notebook sauvegardé sous {path}")
except Exception as e:
logger.error(f"[save_notebook] Erreur de sauvegarde: {e}")
raise
def get_notebook_json(self) -> str:
"""Retourne une représentation JSON (str) du notebook actuellement en mémoire."""
return json.dumps(self._cached_notebook, indent=2, ensure_ascii=False)
def log_notebook_state(self, max_length: int = 10000) -> None:
"""Affiche dans les logs une partie du JSON pour diagnostic (tronquée si trop longue).
Hygiène (issue #6443) : tout chemin machine (Windows ``C:\\Users\\<name>\\...``
ou Unix ``/home/<name>/...``, ``/Users/<name>/...``) présent dans le snippet est
remplacé par ``<USER>`` *avant* journalisation via ``_redact_paths`` pour éviter
la fuite PII-lite (cf secrets-hygiene règle 6, Stop & Repair type C)."""
notebook_json = self.get_notebook_json()
snippet = notebook_json[:max_length]
if len(notebook_json) > max_length:
snippet += ">>> TRUNCATED <<<"
sanitized = _redact_paths(snippet)
logger.debug(f"[NotebookState] Current notebook state (truncated):\n{sanitized}")
def path_leak_report(self) -> str:
"""Detection mecanique de fuite de chemin machine (issue #6443).
Retourne une ligne par cellule dont un output contient un chemin que
`_redact_paths` reecrit (Windows ou Unix) — vide si aucun. C'est le
signal que validate_notebook consulte avant d'approuver : la garde
ne repose pas seulement sur le prompt du Reviewer (cécité #6443).
"""
lines = []
for i, c in enumerate(self._cached_notebook.cells):
flag = False
for out in (c.get("outputs") or []):
text = out.get("text", "")
if isinstance(text, list):
text = "".join(text)
if text and _redact_paths(text) != text:
flag = True
break
if flag:
lines.append(f"cellule {i:02d} : LEAK-CHEMIN")
return "\n".join(lines)
def get_status(self) -> str:
"""Renvoie l'état courant du notebook."""
return self._status
def set_status(self, new_status: str) -> None:
"""
Met à jour l'état du notebook et log la transition.
Ne fait rien si new_status est invalide.
"""
if new_status not in self.VALID_STATES:
logger.error(f"[NotebookState] État invalide: {new_status}")
return
logger.info(f"[NotebookState] Passage de l'état {self._status} → {new_status}")
self._previous_states.append(self._status)
self._status = new_status
def reset_outputs(self) -> None:
"""
Efface les sorties de toutes les cellules (execution_count, outputs).
Utile avant ré-exécution si on veut repartir à zéro.
"""
for cell in self._cached_notebook["cells"]:
if "outputs" in cell:
cell["outputs"] = []
if "execution_count" in cell:
cell["execution_count"] = None
logger.debug("[NotebookState] Sorties réinitialisées dans le notebook.")
def execute_notebook(self) -> bool:
"""
Exécute le notebook via Papermill et retourne True si tout se passe bien,
ou False si une exception survient.
#6443 — hygiène des sorties : un `pip install` (ex. `wordcloud`, paquet à
binaire) émet des avertissements de *build* (DEPRECATION bdist_wheel,
"script ... is installed in <chemin>/PythonXXX/Scripts") qui passent au
travers de `--quiet` (ils proviennent de setup.py, pas de pip). Avec
`log_output=True`, Papermill les renvoie vers son logger puis vers stderr,
faisant fuiter un chemin machine (PII-lite) dans la sortie orchestrée.
On attache donc un filtre sanitisant au logger Papermill pendant l'exécution.
"""
logger.info(f"[NotebookState] Exécution Papermill sur {self.notebook_path}.")
self.save_notebook() # Sauvegarde avant exécution
# #6443 : attacher le filtre anti-bruit build-pip au logger Papermill pour cette exécution
papermill_logger = logging.getLogger("papermill")
papermill_logger.addFilter(_PIP_BUILD_NOISE_FILTER)
success = True
try:
pm.execute_notebook(
input_path=self.notebook_path,
output_path=self.notebook_path,
kernel_name="python3",
progress_bar=False,
log_output=True
)
except Exception as e:
logger.error(f"[execute_notebook] Erreur lors de l'exécution: {e}")
success = False
finally:
papermill_logger.removeFilter(_PIP_BUILD_NOISE_FILTER)
self._load_notebook() # Recharger, car Papermill a peut-être modifié le contenu
self.sanitize_install_outputs() # #6443 : nettoyer les sorties résiduelles
self.save_notebook()
logger.info("[NotebookState] Notebook mis à jour après exécution (avec sorties).")
return success
def sanitize_install_outputs(self) -> None:
"""
#6443 — retire des sorties du notebook exécuté les avertissements de build
de paquets pip qui fuient un chemin machine (chemin Scripts, DEPRECATION
bdist_wheel, Collecting/Downloading). Profondeur de défense : même si le
filtre logger (execute_notebook) capte le bruit propagé via `log_output`,
les sorties du notebook généré peuvent encore contenir ces lignes.
Ne supprime jamais une sortie légitime (n'agit que sur le bruit build-pip).
"""
cleaned = 0
for cell in self._cached_notebook.cells:
if "outputs" not in cell:
continue
for out in cell["outputs"]:
if out.get("output_type") != "stream":
continue
text = out.get("text", "")
if isinstance(text, list):
text = "".join(text)
if not text:
continue
lines = text.split("\n")
kept = [ln for ln in lines if not _PIP_BUILD_NOISE.search(ln)]
if len(kept) != len(lines):
out["text"] = "\n".join(kept)
cleaned += len(lines) - len(kept)
if cleaned:
logger.info(f"[NotebookState] #6443 : {cleaned} ligne(s) de bruit build-pip retirée(s).")
def update_cell(self, cell_index: int, new_source: str) -> None:
"""
Met à jour la source d’une cellule (index) et sauvegarde le notebook.
"""
old_src = self._cached_notebook.cells[cell_index].source
self._cached_notebook.cells[cell_index].source = new_source
logger.info(f"[NotebookState] Mise à jour de la cellule {cell_index}")
logger.debug(f"[NotebookState] Ancien contenu:\n{old_src}")
logger.debug(f"[NotebookState] Nouveau contenu:\n{new_source}")
self.save_notebook()
def find_cell_indices_by_content(self, content_pattern: str) -> list:
"""Retourne la liste des indices de cellules contenant `content_pattern`."""
indices = []
for i, c in enumerate(self._cached_notebook.cells):
if content_pattern in c.source:
indices.append(i)
return indices
def is_approved(self) -> bool:
"""Renvoie True si l'état du notebook est 'validated'."""
return self._status == "validated"La classe NotebookState offre des méthodes pour manipuler un notebook, mais pas pour analyser sa complexite structurelle.
Objectif : Implementer une fonction compute_complexity_metrics() qui evalue la complexite d’un notebook.
Indices : - # Étape 1 : Compter les cellules de code vs markdown pour obtenir le ratio - # Étape 2 : Calculer la longueur moyenne des cellules de code (en caractères) - # Étape 3 : Detecter les cellules contenant des mots-cles complexes (class, def, import, for, while) - # Indice : Un ratio code/markdown > 2 ou une longueur moyenne > 500 caractères indique un notebook complexe
def compute_complexity_metrics(notebook_state: NotebookState) -> dict:
"""
TODO etudiant : Evaluer la complexite structurelle d'un notebook.
Returns:
dict avec les cles: code_count, markdown_count, ratio_code_markdown,
avg_code_length, complex_keyword_count, complexity_level (simple/medium/complex)
"""
# TODO etudiant : analyser notebook_state._cached_notebook.cells
return None
# Test :
# metrics = compute_complexity_metrics(notebook_state)
# print(f"Complexite: {metrics['complexity_level']} (ratio={metrics['ratio_code_markdown']:.1f})")
print("Exercice a completer")Exercice a completerVoici les étapes automatisées :
Notebook-Template.ipynb, ou bien le fichier .ipynb téléversé,Si tout se passe bien, la phase d’orchestration multi-agents peut commencer.
# Cellule Code : Initialisation du notebook ciblé et injection de la tâche
# -----------------------------------------------------------------------
import shutil
import os
import requests
TEMPLATE_URL = "https://raw.githubusercontent.com/jsboige/CoursIA/refs/heads/main/MyIA.AI.Notebooks/GenAI/SemanticKernel/Notebook-Template.ipynb"
TEMPLATE_FILE = "Notebook-Template.ipynb"
def apply_task_description(notebook_state: NotebookState, task_description: str) -> bool:
"""
Remplace le placeholder {{TASK_DESCRIPTION}} dans la première cellule Markdown appropriée.
Retourne True si le placeholder a été trouvé et remplacé, False sinon.
"""
for idx, cell in enumerate(notebook_state._cached_notebook.cells):
if "{{TASK_DESCRIPTION}}" in cell.source:
new_source = cell.source.replace("{{TASK_DESCRIPTION}}", task_description)
notebook_state.update_cell(idx, new_source)
return True
return False
DEST_NOTEBOOK = "Notebook-Generated.ipynb"
# Gestion différenciée selon le mode sélectionné
if config['mode'] == 3: # Mode upload
if config['uploaded_file']:
DEST_NOTEBOOK = config['uploaded_file'].name
with open(DEST_NOTEBOOK, 'wb') as f:
f.write(config['uploaded_file'].content)
logger.info(f"Notebook uploadé : {DEST_NOTEBOOK}")
else: # Modes template (aléatoire, bibliothèque, personnalisé)
if not os.path.exists(TEMPLATE_FILE):
print(f"{TEMPLATE_FILE} introuvable, téléchargement depuis {TEMPLATE_URL}")
try:
response = requests.get(TEMPLATE_URL, timeout=10)
response.raise_for_status() # Gère les erreurs HTTP
with open(TEMPLATE_FILE, "wb") as f:
f.write(response.content)
print(f"Téléchargement terminé, fichier {TEMPLATE_FILE} créé.")
except Exception as e:
print(f"Échec du téléchargement : {e}")
else:
print(f"Le fichier {TEMPLATE_FILE} existe déjà, aucune action nécessaire.")
shutil.copy2(TEMPLATE_FILE, DEST_NOTEBOOK)
logger.info(f"Création depuis le template : {TEMPLATE_FILE}")
# Instanciation du notebook state
notebook_state = NotebookState(DEST_NOTEBOOK)
notebook_state.log_notebook_state()
# Injection de la tâche pour les modes template
if config['mode'] != 3:
changed = apply_task_description(notebook_state, config['task_description'])
if changed:
logger.info("Placeholder {{TASK_DESCRIPTION}} remplacé avec succès.")
logger.info("Exécution du notebook pour validation initiale...")
notebook_state.execute_notebook()
logger.info("Notebook ré-exécuté après injection de la tâche.")
notebook_state.log_notebook_state()
else:
logger.warning("Aucun placeholder détecté. Vérifiez la cellule Markdown contenant {{TASK_DESCRIPTION}}.")
else:
logger.info("Mode upload - Aucune injection de tâche nécessaire")Le fichier Notebook-Template.ipynb existe déjà, aucune action nécessaire.Nous définissons des plugins pour manipuler NotebookState :
get_notebook_content()).Chaque plugin est un ensemble de fonctions décorées (@kernel_function), utilisables par les agents via Semantic Kernel.
# ================================
# Plugins actualisés (avec logs)
# ================================
class BaseNotebookPlugin:
"""
Plugin de base pour manipuler NotebookState.
Fournit la méthode get_notebook_content()
pour récupérer le notebook en JSON.
"""
def __init__(self, state: NotebookState) -> None:
self.state = state
self._get_content_counter = 0 # Pour logguer tous les 5 appels
@kernel_function(
name="get_notebook_content",
description="Renvoie le notebook (format JSON) révisé actuellement."
)
def get_notebook_content(self) -> str:
"""
Retourne la représentation JSON du notebook.
Loggue un extrait toutes les 5 demandes pour éviter la surcharge.
"""
self._get_content_counter += 1
content = self.state.get_notebook_json()
if (self._get_content_counter % 5) == 0:
logger.info(f"[BaseNotebookPlugin] get_notebook_content() (appel n°{self._get_content_counter}) - log complet")
self.state.log_notebook_state(max_length=10000)
else:
snippet = content[:200]
snippet += "..." if len(content) > 200 else ""
logger.info(f"[BaseNotebookPlugin] get_notebook_content() (appel n°{self._get_content_counter}) -> {snippet}")
return content
class NotebookEditingMixin:
"""
Mixin fournissant la fonction d'édition de cellule (update_cell_anyway).
Peut être hérité par Coder ou Admin, qui ont tous deux besoin d'éditer.
"""
@staticmethod
def _modernize_generated_source(source: str) -> str:
"""Remplace les appels dépréciés avant d'écrire le code généré."""
source = source.replace(
"datetime.datetime.utcnow().isoformat() + 'Z'",
"datetime.datetime.now(datetime.UTC).isoformat().replace('+00:00', 'Z')",
)
source = source.replace(
'datetime.datetime.utcnow().isoformat() + "Z"',
'datetime.datetime.now(datetime.UTC).isoformat().replace("+00:00", "Z")',
)
source = source.replace(
"datetime.datetime.utcnow()",
"datetime.datetime.now(datetime.UTC)",
)
# Seaborn 0.13 déprécie `palette` sans `hue`. Pour les barplots
# générés avec des arguments simples x/y, rendre le mapping explicite.
barplot_pattern = _re_redact.compile(
r"sns\.barplot\(x=(?P<x>[^,\n]+),\s*y=(?P<y>[^,\n]+),\s*"
r"palette=(?P<palette>[^,\)\n]+)\)"
)
return barplot_pattern.sub(
lambda match: (
f"sns.barplot(x={match.group('x')}, y={match.group('y')}, "
f"hue={match.group('x')}, palette={match.group('palette')}, "
"legend=False)"
),
source,
)
def update_cell_anyway(self, pattern: str, new_source: str, cell_type: str = None) -> str:
"""
Recherche la première cellule contenant 'pattern' (dans le code ou markdown),
puis remplace son contenu par 'new_source', si l'état le permet.
"""
indices = []
for idx, cell in enumerate(self.state._cached_notebook.cells):
content_match = (pattern in cell.source)
type_match = (cell_type is None) or (cell.cell_type == cell_type)
if content_match and type_match:
indices.append(idx)
if not indices:
return_message = f"Aucune cellule ({cell_type or 'tout type'}) ne contient '{pattern}'."
elif len(indices) > 1:
return_message = f"Plusieurs cellules ({cell_type or 'tout type'}) contiennent '{pattern}'."
else:
old_status = self.state.get_status()
if old_status == "validated":
return_message = "Édition impossible (notebook déjà validé)."
else:
source_to_write = new_source
if cell_type == "code":
source_to_write = self._modernize_generated_source(new_source)
self.state.update_cell(indices[0], source_to_write)
return_message = f"Cellule {cell_type} contenant '{pattern}' mise à jour."
logger.info(f"[NotebookEditingMixin] update_cell_anyway -> {return_message}")
return return_message
class CoderNotebookPlugin(BaseNotebookPlugin, NotebookEditingMixin):
"""
Plugin pour l'agent 'Coder'.
Hérite de BaseNotebookPlugin (lecture) et NotebookEditingMixin (édition).
"""
@kernel_function(
name="update_cell_by_content",
description="Modifie la première cellule de Code contenant 'content_pattern'."
)
def update_cell_by_content(self, content_pattern: str, new_source: str) -> str:
logger.info("[CoderNotebookPlugin] update_cell_by_content()")
status = self.state.get_status()
if status not in ["specified", "implemented"]:
msg = f"Erreur: état '{status}' => modifications bloquées."
else:
msg = self.update_cell_anyway(
pattern=content_pattern,
new_source=new_source,
cell_type="code"
)
logger.info(f"[CoderNotebookPlugin] update_cell_by_content -> {msg}")
return msg
@kernel_function(
name="finish_implementation",
description="Déclare le notebook 'implemented' lorsque le code est prêt."
)
def finish_implementation(self) -> str:
logger.info("[CoderNotebookPlugin] finish_implementation()")
status = self.state.get_status()
if status == "specified":
self.state.set_status("implemented")
msg = "Le notebook passe à l'état 'implemented'."
elif status == "implemented":
msg = "Le notebook est déjà en état 'implemented'."
else:
msg = f"Impossible de passer en 'implemented' depuis '{status}'."
logger.info(f"[CoderNotebookPlugin] finish_implementation -> {msg}")
return msg
class ReviewerNotebookPlugin(BaseNotebookPlugin):
"""
Plugin pour l'agent 'Reviewer'.
Il ne peut pas éditer le notebook, mais peut l'exécuter et approuver (ou non).
"""
@kernel_function(
name="validate_notebook",
description="Exécute le notebook et approuve ou non (approve=True/False)."
)
def validate_notebook(self, approve: bool = True) -> str:
logger.info(f"[ReviewerNotebookPlugin] validate_notebook(approve={approve})")
status = self.state.get_status()
if status != "implemented":
msg = f"Le reviewer ne peut pas valider, état actuel = '{status}'."
else:
success = self.state.execute_notebook()
if not success:
self.state.set_status("specified")
msg = ("Erreur d'exécution dans le notebook (voir logs). "
"Retour à l'état 'specified' pour corrections.")
else:
if approve:
leak = self.state.path_leak_report()
if leak:
self.state.set_status("specified")
msg = ("Refus LEAK-CHEMIN : output contenant un chemin "
"machine détecté. Retour à 'specified' pour "
"nettoyage. Détail : " + leak)
else:
self.state.set_status("tested")
msg = "Le reviewer approuve => état 'tested'."
else:
self.state.set_status("specified")
msg = "Le reviewer refuse => retour à 'specified'."
logger.info(f"[ReviewerNotebookPlugin] validate_notebook -> {msg}")
return msg
class AdminNotebookPlugin(BaseNotebookPlugin, NotebookEditingMixin):
"""
Plugin pour l'agent 'Admin'.
Peut lire, éditer et valider ou invalider le notebook.
"""
@kernel_function(
name="update_markdown_cell",
description="Modifie la première cellule MARKDOWN contenant 'content_pattern'."
)
def admin_edit_markdown_cell(self, content_pattern: str, new_source: str) -> str:
logger.info("[AdminNotebookPlugin] admin_edit_markdown_cell()")
msg = self.update_cell_anyway(
pattern=content_pattern,
new_source=new_source,
cell_type="markdown"
)
# Après édition, on repasse l'état à 'specified'.
self.state.set_status("specified")
logger.info(f"[AdminNotebookPlugin] admin_edit_markdown_cell -> {msg}")
return msg + " -> Revert à 'specified'."
@kernel_function(
name="approve_notebook",
description="Validation finale: admin_ok=True => 'validated', sinon 'specified'."
)
def approve_notebook(self, admin_ok: bool = True) -> str:
logger.info(f"[AdminNotebookPlugin] approve_notebook(admin_ok={admin_ok})")
status = self.state.get_status()
if status != "tested":
msg = f"Impossible d'approuver: l'état est '{status}' (attendu: 'tested')."
else:
if admin_ok:
self.state.set_status("validated")
msg = "Notebook validé => état 'validated'."
else:
self.state.set_status("specified")
msg = "Admin refuse => retour à 'specified'."
logger.info(f"[AdminNotebookPlugin] approve_notebook -> {msg}")
return msgDans un notebook collaboratif, certaines cellules (headers, imports) ne doivent pas etre modifiees par les agents.
Objectif : Créer une classe CellLockMixin qui permet de proteger des cellules contre toute modification.
Indices : - # Étape 1 : Maintenir un ensemble _locked_indices: set[int] des indices de cellules verrouillees - # Étape 2 : Implementer lock_cell(index) et unlock_cell(index) - # Étape 3 : Surcharger update_cell pour verifier le verrouillage avant modification - # Indice : Lever une exception ou retourner un message d’erreur si une cellule verrouillee est modifiée
class CellLockMixin:
"""
TODO etudiant : Mixin pour proteger des cellules contre la modification.
"""
def __init__(self):
# TODO etudiant : initialiser _locked_indices
pass
def lock_cell(self, cell_index: int) -> str:
"""TODO etudiant : Verrouiller une cellule."""
return None
def unlock_cell(self, cell_index: int) -> str:
"""TODO etudiant : Deverrouiller une cellule."""
return None
def is_locked(self, cell_index: int) -> bool:
"""TODO etudiant : Verifier si une cellule est verrouillee."""
return None
print("Exercice a completer")Exercice a completerDeux stratégies principales :
specified ⇒ CoderAgentimplemented ⇒ ReviewerAgenttested ⇒ AdminAgentvalidated ⇒ plus d’agent (arrêt de la conversation)from pydantic import PrivateAttr
class ApprovedBasedTerminationStrategy(TerminationStrategy):
"""Met fin à la conversation après validation ou épuisement du budget."""
_state: NotebookState = PrivateAttr()
_max_steps: int = PrivateAttr(default=24)
def __init__(self, state: NotebookState, max_steps: int = 24):
super().__init__()
self._state = state
self._max_steps = max_steps
self._current_step = 0
async def should_agent_terminate(self, agent, history) -> bool:
self._current_step += 1
is_approved = self._state.is_approved()
logger.debug(
f"[TerminationStrategy] Step={self._current_step}/{self._max_steps}, IsApproved={is_approved}"
)
if is_approved:
logger.info("[TerminationStrategy] Notebook approuvé => arrêt.")
return True
if self._current_step >= self._max_steps:
logger.warning(f"[TerminationStrategy] max_steps={self._max_steps} atteint => arrêt.")
return True
return False
class NotebookAwareSelectionStrategy(SelectionStrategy):
"""Sélectionne l'agent à partir de l'état réel du notebook."""
def __init__(self, state: NotebookState):
super().__init__()
self._state = state
self._has_first_agent_run = False
def reset(self) -> None:
self._has_first_agent_run = False
def _has_code_placeholders(self) -> bool:
"""Détecte les cellules code encore réduites au marqueur du template."""
for cell in self._state._cached_notebook.cells:
source = cell.source.strip()
if (
cell.cell_type == "code"
and source.startswith("# Cellule ")
and "\n" not in source
):
return True
return False
async def select_agent(self, agents, history):
current_status = self._state.get_status()
logger.debug(
f"[SelectionStrategy] nb_agents={len(agents)}, statut={current_status}, "
f"first_run={not self._has_first_agent_run}"
)
if current_status == "validated":
logger.info("[SelectionStrategy] Notebook déjà validé => fin.")
return None
coder = next((a for a in agents if a.name == "CoderAgent"), None)
reviewer = next((a for a in agents if a.name == "ReviewerAgent"), None)
admin = next((a for a in agents if a.name == "AdminAgent"), None)
if not self._has_first_agent_run and current_status == "specified" and admin:
self._has_first_agent_run = True
logger.info("Première intervention : AdminAgent (révision du Markdown).")
return admin
if (
current_status == "specified"
and self._has_first_agent_run
and "implemented" not in self._state._previous_states
and not self._has_code_placeholders()
):
self._state.set_status("implemented")
current_status = "implemented"
logger.info(
"Tous les placeholders code ont été remplacés : "
"transition déterministe vers implemented."
)
if current_status == "specified" and coder:
selected = coder
elif current_status == "implemented" and reviewer:
selected = reviewer
elif current_status == "tested" and admin:
selected = admin
else:
logger.warning(
f"[SelectionStrategy] Aucun agent trouvé pour état='{current_status}' => stop."
)
selected = None
if selected and not self._has_first_agent_run:
self._has_first_agent_run = True
if selected:
logger.info(f"[SelectionStrategy] Agent sélectionné : {selected.name}")
return selectedkernel indépendant, relié à un plugin dédié,.env,# Cellule Code : Création des 3 agents et configuration avec prise en compte du .env
# ---------------------------------------------------------------------------------
from openai import AsyncOpenAI
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
import os
from dotenv import load_dotenv
load_dotenv("../.env")
openai_api_key = os.getenv("OPENAI_API_KEY")
openai_base_url_from_env = os.getenv("OPENAI_BASE_URL", "").strip()
openai_chat_model_id = os.getenv("OPENAI_CHAT_MODEL_ID", "gpt-5.6-luna")
def create_chat_completion_service(service_id: str = "default"):
"""
Crée une instance de ChatCompletion (OpenAI) en fonction
des variables d'environnement. Gère explicitement l'URL par défaut pour éviter
les erreurs de protocole non supporté.
Variables .env utilisées :
- OPENAI_API_KEY : Clé API (obligatoire).
- OPENAI_CHAT_MODEL_ID : Nom du modèle (ex: "gpt-5.6-terra", "gpt-5.6-luna").
- OPENAI_BASE_URL : URL personnalisée (optionnelle). Si vide, utilise l'API OpenAI officielle.
"""
if not openai_api_key:
logger.error("La variable d'environnement OPENAI_API_KEY n'est pas définie !")
raise ValueError("OPENAI_API_KEY is not set in the environment.")
if openai_base_url_from_env:
final_base_url = openai_base_url_from_env
logger.info(f"Utilisation d'un endpoint compatible OpenAI : {final_base_url}")
else:
final_base_url = "https://api.openai.com/v1"
logger.info(f"Utilisation du service OpenAI officiel (URL explicite: {final_base_url}).")
try:
openai_async_client = AsyncOpenAI(
api_key=openai_api_key,
base_url=final_base_url
)
except Exception as client_error:
logger.error(f"Erreur lors de la création du client AsyncOpenAI pour base_url='{final_base_url}': {client_error}")
raise
try:
sk_service = OpenAIChatCompletion(
service_id=service_id,
ai_model_id=openai_chat_model_id,
async_client=openai_async_client
)
logger.debug(f"Service Semantic Kernel '{service_id}' créé pour le modèle '{openai_chat_model_id}' pointant vers '{final_base_url}'")
return sk_service
except Exception as sk_service_error:
logger.error(f"Erreur lors de la création du service OpenAIChatCompletion de Semantic Kernel: {sk_service_error}")
raise
def create_kernel_for_agent(agent_id: str, plugin_instance) -> Kernel:
"""Instancie un Kernel Semantic Kernel et lui ajoute le plugin de l'agent."""
k = Kernel()
chat_service = create_chat_completion_service(service_id="default")
k.add_service(chat_service)
k.add_plugin(plugin_instance, plugin_name=f"{agent_id}_plugin")
return k
coder_plugin = CoderNotebookPlugin(notebook_state)
reviewer_plugin = ReviewerNotebookPlugin(notebook_state)
admin_plugin = AdminNotebookPlugin(notebook_state)
coder_kernel = create_kernel_for_agent("coder_kernel", coder_plugin)
reviewer_kernel = create_kernel_for_agent("reviewer_kernel", reviewer_plugin)
admin_kernel = create_kernel_for_agent("admin_kernel", admin_plugin)
coder_settings = coder_kernel.get_prompt_execution_settings_from_service_id("default")
coder_settings.function_choice_behavior = FunctionChoiceBehavior.Auto()
# Famille gpt-5.6 : le reasoning par defaut est incompatible avec les function tools
# sur /v1/chat/completions -- l API exige une desactivation explicite (reasoning_effort=none).
coder_settings.extra_body = {"reasoning_effort": "none"}
reviewer_settings = reviewer_kernel.get_prompt_execution_settings_from_service_id("default")
reviewer_settings.function_choice_behavior = FunctionChoiceBehavior.Auto()
# Famille gpt-5.6 : le reasoning par defaut est incompatible avec les function tools
# sur /v1/chat/completions -- l API exige une desactivation explicite (reasoning_effort=none).
reviewer_settings.extra_body = {"reasoning_effort": "none"}
admin_settings = admin_kernel.get_prompt_execution_settings_from_service_id("default")
admin_settings.function_choice_behavior = FunctionChoiceBehavior.Auto()
# Famille gpt-5.6 : le reasoning par defaut est incompatible avec les function tools
# sur /v1/chat/completions -- l API exige une desactivation explicite (reasoning_effort=none).
admin_settings.extra_body = {"reasoning_effort": "none"}
coder_agent = ChatCompletionAgent(
kernel=coder_kernel,
name="CoderAgent",
instructions=(
"Vous êtes le **Coder**. Votre rôle : Implémenter avec le plus grand soin dans les cellules de code (pas la markdown) les instructions établies dans les cellules de markdown du notebook\n"
"1) Visualisez systématiquement le JSON du notebook via get_notebook_content().\n"
"2) Analysez et modifiez les cellules de code avec update_cell_by_content(), sans reprendre le markdown.\n"
"2bis) N'ajoutez JAMAIS `%pip install`, `!pip install` ni d'installation par subprocess. Les dépendances sont préparées hors notebook.\n"
"2ter) Hygiène des sorties : n'affichez jamais de chemin absolu. Utilisez des chemins relatifs (`outputs/...`) et affichez seulement leur forme relative ou leur basename. Neutralisez à la source les FutureWarning prévisibles avec les API modernes ou un filtre ciblé ; aucun warning ne doit imprimer un chemin de kernel temporaire.\n"
"3) Préservez les marqueurs importants des cellules de code.\n"
"4) Quand c'est prêt, appelez finish_implementation().\n"
"5) Si besoin, faites de nouvelles itérations.\n"
),
arguments=KernelArguments(settings=coder_settings)
)
reviewer_agent = ChatCompletionAgent(
kernel=reviewer_kernel,
name="ReviewerAgent",
instructions=(
"Vous êtes le **Reviewer**. Vérifiez le travail du codeur et validez-le seulement quand il est exécutable et propre.\n"
"1) Consultez toujours le notebook complet après chaque mise à jour.\n"
"2) Appelez validate_notebook(approve=True/False) selon la qualité.\n"
"3) Vérifiez les erreurs et l'alignement code/Markdown.\n"
"3bis) Refusez toute installation inline (`%pip`, `!pip`, subprocess). Refusez aussi tout output contenant un chemin machine absolu (répertoire utilisateur Windows ou Unix, lettre de lecteur) : validate_notebook refuse automatiquement (marqueur LEAK-CHEMIN) tant qu'un chemin machine apparaît dans un output.\n"
),
arguments=KernelArguments(settings=reviewer_settings)
)
admin_agent = ChatCompletionAgent(
kernel=admin_kernel,
name="AdminAgent",
instructions=(
"Vous êtes l'**Admin**. Vous spécifiez le notebook et exigez un rendu pédagogique de haute qualité.\n"
"1) Intervenez au début pour détailler sans ambiguïté les cellules Markdown.\n"
"1bis) Ne prescrivez JAMAIS d'installation inline. Décrivez les dépendances comme prérequis hors notebook.\n"
"1ter) Exigez des chemins de sortie relatifs (`outputs/...`) : aucun chemin absolu ni chemin temporaire de kernel ne doit apparaître dans les sorties. Demandez un traitement ciblé des warnings prévisibles à la source.\n"
"Corrigez et détaillez les explications Markdown via admin_edit_markdown_cell.\n"
"2) Après chaque édition, le Coder implémente les cellules de code.\n"
"3) Après validation du Reviewer, éditez, validez ou invalidez via approve_notebook(admin_ok=True/False).\n"
"4) Une fois satisfait, validez définitivement avec admin_ok=True.\n"
),
arguments=KernelArguments(settings=admin_settings)
)
termination_strategy = ApprovedBasedTerminationStrategy(notebook_state)
selection_strategy = NotebookAwareSelectionStrategy(notebook_state)
group_chat = AgentGroupChat(
agents=[coder_agent, reviewer_agent, admin_agent],
selection_strategy=selection_strategy,
termination_strategy=termination_strategy
)
logger.info("Agents créés et group_chat initialisé avec instructions mises à jour.")Nous lançons enfin la conversation multi-agents :
specified, implemented, tested, validated).update_cell_by_content, validate_notebook, approve_notebook, etc.).Le statut final du notebook (approuvé ou non) est alors visible dans les logs et dans son contenu.
import asyncio
class StepBufferHandler(logging.Handler):
"""
Tamponne les records du logger et les émet en un seul bloc par step.
Avec le StreamHandler par défaut, chaque record de log devient un objet
de sortie distinct : une conversation multi-agents en produit plus de
180, ce qui fait régresser le ratchet output-flood à chaque
re-exécution. Ce handler conserve chaque ligne, dans l'ordre, mais
les regroupe en un bloc par step (les appels d'outils d'un agent
arrivent avant son marqueur [STEP n] et restent donc dans son bloc).
"""
def __init__(self, formatter: logging.Formatter):
super().__init__()
self.setLevel(logging.DEBUG)
self.setFormatter(formatter)
self._buffer = []
def emit(self, record: logging.LogRecord) -> None:
self._buffer.append(self.format(record))
def flush_block(self) -> None:
if self._buffer:
print("\n".join(self._buffer))
self._buffer = []
async def run_conversation():
"""Lance la conversation multi-agents jusqu'à validation du notebook."""
stream_handler = logger.handlers[0]
step_handler = StepBufferHandler(stream_handler.formatter)
logger.removeHandler(stream_handler)
logger.addHandler(step_handler)
try:
logger.info("Version initiale du notebook :")
notebook_state.log_notebook_state()
initial_content = notebook_state.get_notebook_json()
group_chat.history.add_system_message(f"NOTEBOOK CONTENT:\n{initial_content}")
group_chat.history.add_user_message(
"Finalisez ce notebook de manière entièrement autonome. Remplissez "
"toutes les cellules code encore réduites à un placeholder '# Cellule N'. "
"Après les éditions, appelez finish_implementation(). Si le Reviewer refuse "
"ou si Papermill trouve une erreur, corrigez la cellule signalée puis "
"rappelez finish_implementation() sans demander de confirmation humaine. "
"Continuez jusqu'à la validation finale par l'Admin."
)
logger.info("=== Début de la conversation entre agents ===")
step_handler.flush_block() # bloc d'introduction (version initiale + début)
iteration = 0
async for message in group_chat.invoke():
iteration += 1
# Les appels de plugins journalisent déjà chaque action utile. Ne pas recopier
# ici le contenu LLM : il peut contenir du code généré, des chemins ou des
# messages d'erreur hypothétiques qui seraient alors confondus avec le résultat
# réel de l'exécution. On conserve une trace structurelle vérifiable.
content_length = len(str(message.content))
logger.info(
f"[STEP {iteration} - {message.name}] réponse reçue "
f"({content_length} caractères)."
)
step_handler.flush_block() # un bloc par step : outils de l'agent + marqueur
if notebook_state.is_approved():
logger.info("Notebook approuvé => fin de la conversation.")
break
logger.info("Version finale du notebook, après la conversation :")
notebook_state.log_notebook_state(max_length=20000)
except Exception as e:
logger.error(f"Erreur inattendue: {str(e)}")
finally:
logger.info(f"Statut final - Approuvé: {notebook_state.is_approved()}")
logger.info("=== Fin de la conversation ===")
step_handler.flush_block() # bloc de clôture (version finale + statut)
logger.removeHandler(step_handler)
logger.addHandler(stream_handler)
await run_conversation()09:57:46 [INFO] Orchestration - Version initiale du notebook : 09:57:46 [INFO] Orchestration - === Début de la conversation entre agents === 09:57:46 [INFO] Orchestration - Première intervention : AdminAgent (révision du Markdown). 09:57:49 [INFO] Orchestration - [BaseNotebookPlugin] get_notebook_content() (appel n°1) -> { "cells": [ { "cell_type": "markdown", "id": "516d2854", "metadata": { "papermill": { "duration": 0.001661, "end_time": "2026-09-09T07:57:45.571591... 09:58:01 [INFO] Orchestration - [AdminNotebookPlugin] admin_edit_markdown_cell() 09:58:01 [INFO] Orchestration - [NotebookState] Mise à jour de la cellule 0 09:58:01 [INFO] Orchestration - [NotebookState] Notebook sauvegardé sous Notebook-Generated.ipynb 09:58:01 [INFO] Orchestration - [NotebookEditingMixin] update_cell_anyway -> Cellule markdown contenant '# Notebook de travail' mise à jour. 09:58:01 [INFO] Orchestration - [NotebookState] Passage de l'état specified → specified 09:58:01 [INFO] Orchestration - [AdminNotebookPlugin] admin_edit_markdown_cell -> Cellule markdown contenant '# Notebook de travail' mise à jour. 09:58:01 [INFO] Orchestration - [AdminNotebookPlugin] admin_edit_markdown_cell() 09:58:01 [INFO] Orchestration - [NotebookState] Mise à jour de la cellule 2 09:58:01 [INFO] Orchestration - [NotebookState] Notebook sauvegardé sous Notebook-Generated.ipynb 09:58:01 [INFO] Orchestration - [NotebookEditingMixin] update_cell_anyway -> Cellule markdown contenant '## 1. Préparation' mise à jour. 09:58:01 [INFO] Orchestration - [NotebookState] Passage de l'état specified → specified 09:58:01 [INFO] Orchestration - [AdminNotebookPlugin] admin_edit_markdown_cell -> Cellule markdown contenant '## 1. Préparation' mise à jour. 09:58:01 [INFO] Orchestration - [AdminNotebookPlugin] admin_edit_markdown_cell() 09:58:01 [INFO] Orchestration - [NotebookState] Mise à jour de la cellule 4 09:58:01 [INFO] Orchestration - [NotebookState] Notebook sauvegardé sous Notebook-Generated.ipynb 09:58:01 [INFO] Orchestration - [NotebookEditingMixin] update_cell_anyway -> Cellule markdown contenant '## 2. Initialisation' mise à jour. 09:58:01 [INFO] Orchestration - [NotebookState] Passage de l'état specified → specified 09:58:01 [INFO] Orchestration - [AdminNotebookPlugin] admin_edit_markdown_cell -> Cellule markdown contenant '## 2. Initialisation' mise à jour. 09:58:01 [INFO] Orchestration - [AdminNotebookPlugin] admin_edit_markdown_cell() 09:58:01 [INFO] Orchestration - [NotebookState] Mise à jour de la cellule 6 09:58:01 [INFO] Orchestration - [NotebookState] Notebook sauvegardé sous Notebook-Generated.ipynb 09:58:01 [INFO] Orchestration - [NotebookEditingMixin] update_cell_anyway -> Cellule markdown contenant '## 3. Traitement' mise à jour. 09:58:01 [INFO] Orchestration - [NotebookState] Passage de l'état specified → specified 09:58:01 [INFO] Orchestration - [AdminNotebookPlugin] admin_edit_markdown_cell -> Cellule markdown contenant '## 3. Traitement' mise à jour. 09:58:01 [INFO] Orchestration - [AdminNotebookPlugin] admin_edit_markdown_cell() 09:58:01 [INFO] Orchestration - [NotebookState] Mise à jour de la cellule 8 09:58:01 [INFO] Orchestration - [NotebookState] Notebook sauvegardé sous Notebook-Generated.ipynb 09:58:01 [INFO] Orchestration - [NotebookEditingMixin] update_cell_anyway -> Cellule markdown contenant '## 4. Analyse' mise à jour. 09:58:01 [INFO] Orchestration - [NotebookState] Passage de l'état specified → specified 09:58:01 [INFO] Orchestration - [AdminNotebookPlugin] admin_edit_markdown_cell -> Cellule markdown contenant '## 4. Analyse' mise à jour. 09:58:01 [INFO] Orchestration - [AdminNotebookPlugin] admin_edit_markdown_cell() 09:58:01 [INFO] Orchestration - [NotebookState] Mise à jour de la cellule 10 09:58:01 [INFO] Orchestration - [NotebookState] Notebook sauvegardé sous Notebook-Generated.ipynb 09:58:01 [INFO] Orchestration - [NotebookEditingMixin] update_cell_anyway -> Cellule markdown contenant '## 5. Conclusion' mise à jour. 09:58:01 [INFO] Orchestration - [NotebookState] Passage de l'état specified → specified 09:58:01 [INFO] Orchestration - [AdminNotebookPlugin] admin_edit_markdown_cell -> Cellule markdown contenant '## 5. Conclusion' mise à jour. 09:58:02 [INFO] Orchestration - [AdminNotebookPlugin] approve_notebook(admin_ok=False) 09:58:02 [INFO] Orchestration - [AdminNotebookPlugin] approve_notebook -> Impossible d'approuver: l'état est 'specified' (attendu: 'tested'). 09:58:04 [INFO] Orchestration - [STEP 1 - AdminAgent] réponse reçue (173 caractères). 09:58:04 [INFO] Orchestration - [SelectionStrategy] Agent sélectionné : CoderAgent 09:58:06 [INFO] Orchestration - [BaseNotebookPlugin] get_notebook_content() (appel n°1) -> { "cells": [ { "cell_type": "markdown", "id": "516d2854", "metadata": { "papermill": { "duration": 0.001661, "end_time": "2026-09-09T07:57:45.571591... 09:58:30 [INFO] Orchestration - [CoderNotebookPlugin] update_cell_by_content() 09:58:30 [INFO] Orchestration - [NotebookState] Mise à jour de la cellule 1 09:58:30 [INFO] Orchestration - [NotebookState] Notebook sauvegardé sous Notebook-Generated.ipynb 09:58:30 [INFO] Orchestration - [NotebookEditingMixin] update_cell_anyway -> Cellule code contenant '# Cellule 0' mise à jour. 09:58:30 [INFO] Orchestration - [CoderNotebookPlugin] update_cell_by_content -> Cellule code contenant '# Cellule 0' mise à jour. 09:58:30 [INFO] Orchestration - [CoderNotebookPlugin] update_cell_by_content() 09:58:30 [INFO] Orchestration - [NotebookState] Mise à jour de la cellule 3 09:58:30 [INFO] Orchestration - [NotebookState] Notebook sauvegardé sous Notebook-Generated.ipynb 09:58:30 [INFO] Orchestration - [NotebookEditingMixin] update_cell_anyway -> Cellule code contenant '# Cellule 1' mise à jour. 09:58:30 [INFO] Orchestration - [CoderNotebookPlugin] update_cell_by_content -> Cellule code contenant '# Cellule 1' mise à jour. 09:58:30 [INFO] Orchestration - [CoderNotebookPlugin] update_cell_by_content() 09:58:30 [INFO] Orchestration - [NotebookState] Mise à jour de la cellule 5 09:58:30 [INFO] Orchestration - [NotebookState] Notebook sauvegardé sous Notebook-Generated.ipynb 09:58:30 [INFO] Orchestration - [NotebookEditingMixin] update_cell_anyway -> Cellule code contenant '# Cellule 2' mise à jour. 09:58:30 [INFO] Orchestration - [CoderNotebookPlugin] update_cell_by_content -> Cellule code contenant '# Cellule 2' mise à jour. 09:58:30 [INFO] Orchestration - [CoderNotebookPlugin] update_cell_by_content() 09:58:30 [INFO] Orchestration - [NotebookState] Mise à jour de la cellule 7 09:58:30 [INFO] Orchestration - [NotebookState] Notebook sauvegardé sous Notebook-Generated.ipynb 09:58:30 [INFO] Orchestration - [NotebookEditingMixin] update_cell_anyway -> Cellule code contenant '# Cellule 3' mise à jour. 09:58:30 [INFO] Orchestration - [CoderNotebookPlugin] update_cell_by_content -> Cellule code contenant '# Cellule 3' mise à jour. 09:58:30 [INFO] Orchestration - [CoderNotebookPlugin] update_cell_by_content() 09:58:30 [INFO] Orchestration - [NotebookState] Mise à jour de la cellule 9 09:58:30 [INFO] Orchestration - [NotebookState] Notebook sauvegardé sous Notebook-Generated.ipynb 09:58:30 [INFO] Orchestration - [NotebookEditingMixin] update_cell_anyway -> Cellule code contenant '# Cellule 4' mise à jour. 09:58:30 [INFO] Orchestration - [CoderNotebookPlugin] update_cell_by_content -> Cellule code contenant '# Cellule 4' mise à jour. 09:58:30 [INFO] Orchestration - [CoderNotebookPlugin] update_cell_by_content() 09:58:30 [INFO] Orchestration - [NotebookState] Mise à jour de la cellule 11 09:58:30 [INFO] Orchestration - [NotebookState] Notebook sauvegardé sous Notebook-Generated.ipynb 09:58:30 [INFO] Orchestration - [NotebookEditingMixin] update_cell_anyway -> Cellule code contenant '# Cellule 5' mise à jour. 09:58:30 [INFO] Orchestration - [CoderNotebookPlugin] update_cell_by_content -> Cellule code contenant '# Cellule 5' mise à jour. 09:58:32 [INFO] Orchestration - [CoderNotebookPlugin] finish_implementation() 09:58:32 [INFO] Orchestration - [NotebookState] Passage de l'état specified → implemented 09:58:32 [INFO] Orchestration - [CoderNotebookPlugin] finish_implementation -> Le notebook passe à l'état 'implemented'. 09:58:34 [INFO] Orchestration - [STEP 2 - CoderAgent] réponse reçue (130 caractères). 09:58:34 [INFO] Orchestration - [SelectionStrategy] Agent sélectionné : ReviewerAgent 09:58:35 [INFO] Orchestration - [BaseNotebookPlugin] get_notebook_content() (appel n°1) -> { "cells": [ { "cell_type": "markdown", "id": "516d2854", "metadata": { "papermill": { "duration": 0.001661, "end_time": "2026-09-09T07:57:45.571591... 09:58:37 [INFO] Orchestration - [ReviewerNotebookPlugin] validate_notebook(approve=True) 09:58:37 [INFO] Orchestration - [NotebookState] Exécution Papermill sur Notebook-Generated.ipynb. 09:58:37 [INFO] Orchestration - [NotebookState] Notebook sauvegardé sous Notebook-Generated.ipynb 09:58:39 [INFO] Orchestration - [NotebookState] Notebook sauvegardé sous Notebook-Generated.ipynb 09:58:39 [INFO] Orchestration - [NotebookState] Notebook mis à jour après exécution (avec sorties). 09:58:39 [INFO] Orchestration - [NotebookState] Passage de l'état implemented → tested 09:58:39 [INFO] Orchestration - [ReviewerNotebookPlugin] validate_notebook -> Le reviewer approuve => état 'tested'. 09:58:41 [INFO] Orchestration - [STEP 3 - ReviewerAgent] réponse reçue (177 caractères). 09:58:41 [INFO] Orchestration - [SelectionStrategy] Agent sélectionné : AdminAgent 09:58:43 [INFO] Orchestration - [BaseNotebookPlugin] get_notebook_content() (appel n°2) -> { "cells": [ { "cell_type": "markdown", "id": "516d2854", "metadata": { "papermill": { "duration": 0.002828, "end_time": "2026-09-09T07:58:38.562886... 09:58:44 [INFO] Orchestration - [AdminNotebookPlugin] approve_notebook(admin_ok=True) 09:58:44 [INFO] Orchestration - [NotebookState] Passage de l'état tested → validated 09:58:44 [INFO] Orchestration - [AdminNotebookPlugin] approve_notebook -> Notebook validé => état 'validated'. 09:58:46 [INFO] Orchestration - [TerminationStrategy] Notebook approuvé => arrêt. 09:58:46 [INFO] Orchestration - [STEP 4 - AdminAgent] réponse reçue (297 caractères). 09:58:46 [INFO] Orchestration - Notebook approuvé => fin de la conversation. 09:58:46 [INFO] Orchestration - Version finale du notebook, après la conversation : 09:58:46 [INFO] Orchestration - Statut final - Approuvé: True 09:58:46 [INFO] Orchestration - === Fin de la conversation ===
Après une session NotebookMaker, il est utile de generer un rapport recapitulatif pour analyser les performances.
Objectif : Implementer une fonction generate_execution_report() qui compile les metriques d’une session complete.
Indices : - # Étape 1 : Collecter le statut final (notebook_state.is_approved()), le nombre d’etats précédents (_previous_states) - # Étape 2 : Calculer le nombre de transitions “specified -> implemented” (cycles de correction) - # Étape 3 : Produire un rapport textuel avec les metriques cles - # Indice : Le nombre de cycles de correction indique la qualite initiale des specifications
def generate_execution_report(notebook_state: NotebookState, iteration_count: int) -> str:
"""
TODO etudiant : Generer un rapport d'execution de la session NotebookMaker.
Args:
notebook_state: L'etat final du notebook
iteration_count: Nombre d'iterations total de la conversation
Returns:
Rapport textuel avec metriques (statut, iterations, cycles de correction)
"""
# TODO etudiant : analyser notebook_state._previous_states et _status
return None
# Test :
# report = generate_execution_report(notebook_state, iteration=10)
# print(report)
print("Exercice a completer")Exercice a completer
Comment demarrer
Schema : conversation d’agents et etat du notebook
Trois agents collaborent en sequence (Admin -> Coder -> Reviewer), chacun faisant progresser l’etat partage du notebook (specified -> implemented -> tested -> validated).