Lab 12: DS-STAR Workshop - Analyse Multi-Fichiers

Navigation : Lab 11 << | Index | >> Lab 13

Objectifs d’apprentissage

À la fin de ce laboratoire, vous saurez : 1. Combiner FileAnalyzer + Planner-Coder-Verifier en pipeline complet 2. Analyser plusieurs fichiers de données de manière autonome 3. Générer un rapport d’analyse structuré automatiquement 4. Gérer les erreurs et itérations dans un workflow complexe

Prérequis

  • Lab 10 et Lab 11 complétés
  • Compréhension de l’architecture DS-STAR
  • Configuration multi-provider active

Durée estimée : 50-60 minutes

Repère bibliographique. Ce workshop assemble le pipeline agent de bout en bout de DS-STAR (FileAnalyzer → Planner → Coder → Executor → Verifier) — l’architecture multi-agent complète décrite par Nam et al., DS-STAR: Data Science Agent for Solving Diverse Tasks across Heterogeneous Formats and Open-Ended Queries, arXiv:2509.21825, 2025. Le principe général d’un agent LLM orchestrant une chaîne de modules spécialisés (perception, planification, action, vérification) est synthétisé par Z. Xi et al., The Rise and Potential of Large Language Model Based Agents: A Survey, arXiv:2309.07864, 2023.

1. Configuration

Ce workshop assemble le pipeline FileAnalyzer → Planner ADK → Coder ADK → Executor → Verifier ADK et l’exerce sur un dataset de ventes. Les rôles de raisonnement utilisent le runtime partagé utils.adk_runtime, qui construit de vrais agents Google ADK sur le provider actif de la série.

Le sys.path.insert expose les modules config et utils communs à Track2. LLMClient reste importé uniquement pour les exercices de compatibilité en fin de notebook ; le pipeline démontré n’appelle pas directement LiteLLM et ne contient aucun endpoint ni secret en dur.

import sys
import warnings
from pathlib import Path

sys.path.insert(0, str(Path().resolve().parent))
warnings.filterwarnings(
    "ignore",
    message=r"\[EXPERIMENTAL\] feature FeatureName.JSON_SCHEMA_FOR_FUNC_DECL is enabled\.",
    category=UserWarning,
    module=r"google\.adk\.models\.llm_request",
)

import os
import json
import re
import pandas as pd
import numpy as np
from typing import Optional, Dict, List, Tuple
from dataclasses import dataclass, asdict
from enum import Enum

from config import get_settings
from utils import LLMClient
from utils.adk_runtime import build_agent, run_agent_turn

print("Imports OK : pandas, numpy, LLMClient et runtime Google ADK réel")
Imports OK : pandas, numpy, LLMClient et runtime Google ADK réel

get_settings() charge le provider actif sans afficher de clé ni d’endpoint sensible. Ce diagnostic précède les appels multi-agents : il permet d’attribuer les sorties au provider réellement configuré, tandis que le modèle ADK est construit par le runtime partagé.

settings = get_settings()
print(f'Provider: {settings.active_provider}')
Provider: openai

2. Imports des Modules DS-STAR

On définit ici les structures de données qui circulent entre les agents — le contrat d’interface du pipeline. Comparé au Lab 11 (Planner-Coder isolé), ce workshop enrichit FileMetadata d’un champ sample_data : le FileAnalyzer collectera désormais un échantillon des 3 premières lignes pour nourrir le contexte du Planner.

  • Plan : étapes + raisonnement produits par le Planner.
  • ExecutionResult : sortie de l’Executor (succès, stdout, erreur, code).
  • FileMetadata : métadonnées du fichier + sample_data (nouveauté du workshop).
  • VerificationStatus : verdict (SUCCESS / NEEDS_REFINEMENT / FAILED) qui pilote la boucle.
# Data classes reutilisees
@dataclass
class Plan:
    steps: List[str]
    reasoning: str

@dataclass
class ExecutionResult:
    success: bool
    output: str
    error: Optional[str] = None
    code: Optional[str] = None

@dataclass
class FileMetadata:
    filename: str
    format: str
    size_bytes: int
    num_rows: int = None
    num_columns: int = None
    columns: list = None
    sample_data: list = None

class VerificationStatus(Enum):
    SUCCESS = 'success'
    NEEDS_REFINEMENT = 'needs_refinement'
    FAILED = 'failed'

print("Data classes et enums definis.")
Data classes et enums definis.

3. FileAnalyzer Module

Le FileAnalyzer est l’entrée du pipeline : il lit le fichier, en extrait un contexte riche et le transmet aux agents suivants. Sa richesse conditionne la qualité de toute la chaîne. Cette version est plus complète que celle du Lab 10 : pour chaque colonne numérique, elle calcule des statistiques (min/max/mean) et un taux de valeurs manquantes (missing_pct), et elle stocke un échantillon de 3 lignes (sample_data).

Le format .xlsx (Excel) est désormais supporté en plus du CSV et JSON — d’où l’attribut SUPPORTED. La méthode generate_context synthétise ces métadonnées en un texte court (nom, format, taille en KB, 5 premières colonnes) que le Planner et le Coder « verront » du dataset.

class FileAnalyzer:
    """Analyse déterministe de fichiers de données avant les tours ADK."""

    SUPPORTED = {'.csv': 'csv', '.json': 'json', '.xlsx': 'excel'}

    def analyze_csv(self, path: str) -> FileMetadata:
        df = pd.read_csv(path)
        columns = []
        for name in df.columns:
            info = {
                'name': name,
                'dtype': str(df[name].dtype),
                'missing_pct': round(df[name].isna().mean() * 100, 2),
            }
            if pd.api.types.is_numeric_dtype(df[name]):
                info['stats'] = {
                    'min': float(df[name].min()),
                    'max': float(df[name].max()),
                    'mean': float(df[name].mean()),
                }
            columns.append(info)
        return FileMetadata(
            filename=Path(path).name,
            format='csv',
            size_bytes=os.path.getsize(path),
            num_rows=len(df),
            num_columns=len(df.columns),
            columns=columns,
            sample_data=df.head(3).to_dict('records'),
        )

    def generate_context(self, meta: FileMetadata) -> str:
        lines = [
            f"Fichier: {meta.filename}",
            f"Format: {meta.format}",
            f"Taille: {meta.size_bytes / 1024:.1f} KB",
            f"Lignes: {meta.num_rows}, Colonnes: {meta.num_columns}",
        ]
        for column in (meta.columns or [])[:5]:
            lines.append(f"  - {column['name']} ({column['dtype']})")
        return "\n".join(lines)

print("FileAnalyzer prêt : profil CSV déterministe sans appel LLM caché")
FileAnalyzer prêt : profil CSV déterministe sans appel LLM caché

4. Planner-Coder-Verifier sur Google ADK

Les trois rôles de raisonnement sont désormais de vrais google.adk.agents.Agent, exécutés dans des sessions distinctes par un Runner :

  • Planner ADK : appelle obligatoirement l’outil déterministe get_file_schema, puis produit un plan structuré ;
  • Coder ADK : transforme ce plan en code Pandas visible et exécutable ;
  • Verifier ADK : confronte la sortie à la question et rend un verdict actionnable.

L’Executor reste local : ADK porte les rôles LLM et leurs événements, tandis que le code généré demeure inspectable. Les anciennes classes synchrones basées sur LLMClient ne sont plus utilisées par le pipeline du workshop ; LLMClient reste seulement dans les exercices qui demandent d’étendre un composant historique.

print(
    "Planner ADK : construit dans DSStarPipeline et exécuté par "
    "run_agent_turn avec l'outil get_file_schema"
)
Planner ADK : construit dans DSStarPipeline et exécuté par run_agent_turn avec l'outil get_file_schema

Le Planner n’est plus un wrapper synchrone autour de LLMClient. DSStarPipeline construit un véritable Agent Google ADK et lui impose un appel à get_file_schema avant la rédaction du plan. La trace d’événements permet ensuite de vérifier le round-trip de cet outil.

Le Coder est lui aussi un agent ADK distinct ; son prompt reçoit le plan et le schéma exact du DataFrame déjà chargé.

print(
    "Coder ADK : construit dans DSStarPipeline et contraint à produire "
    "un bloc Python exécutable sur df"
)
Coder ADK : construit dans DSStarPipeline et contraint à produire un bloc Python exécutable sur df

Le Coder ADK transforme le plan en code Python visible. Le prompt précise que le DataFrame df est déjà chargé par l’Executor : lire à nouveau un chemin de fichier serait une violation de l’interface. Le pipeline extrait le bloc python de la réponse finale, puis transmet ce code à l’Executor local.

L’Executor garde ainsi l’action déterministe et inspectable, tandis que le raisonnement reste porté par le vrai LLM.

class Executor:
    def __init__(self, df: pd.DataFrame):
        self.df = df
        self.namespace = {'df': df, 'pd': pd, 'np': np, 'print': print}

    def execute(self, code: str) -> ExecutionResult:
        from io import StringIO
        import sys
        old_stdout = sys.stdout
        sys.stdout = StringIO()
        try:
            exec(code, self.namespace)
            return ExecutionResult(success=True, output=sys.stdout.getvalue(), code=code)
        except Exception as e:
            return ExecutionResult(success=False, output=sys.stdout.getvalue(), error=str(e), code=code)
        finally:
            sys.stdout = old_stdout

print("Executor pret.")
Executor pret.

L’Executor exécute le code généré dans un namespace isolé (df, pd, np, print). La redirection de sys.stdout vers un StringIO capture la sortie standard pour que le Verifier puisse l’inspecter. Toute exception est attrapée et encapsulée dans ExecutionResult.error — le pipeline ne plante jamais, il signale (pattern fail-soft indispensable à un agent autonome).

Le Verifier examine ensuite ce résultat.

class Verifier:
    """Base locale conservée uniquement pour l'exercice BusinessVerifier."""

    def __init__(self, llm: LLMClient):
        self.llm = llm

    def verify(self, question: str, result: ExecutionResult) -> Tuple[VerificationStatus, str]:
        if not result.success:
            return VerificationStatus.FAILED, f"Erreur: {result.error}"
        return VerificationStatus.NEEDS_REFINEMENT, "À évaluer par l'agent ADK"

print("Compatibilité pédagogique Verifier conservée ; le pipeline utilise Verifier ADK")
Compatibilité pédagogique Verifier conservée ; le pipeline utilise Verifier ADK

5. Orchestrateur DS-STAR complet

DSStarPipeline assemble FileAnalyzer, les trois rôles Google ADK et l’Executor local dans une boucle Analyze → Plan → Code → Execute → Verify. C’est le cœur du workshop : là où les labs précédents isolaient un composant (Lab 10 FileAnalyzer, Lab 11 Planner-Coder), celui-ci montre la composition et surtout la boucle d’itération, l’implémentation du pattern « plan-execute-verify-refine » de la littérature DS-STAR.

À chaque itération, la trace conserve les identifiants de session, le nombre d’événements, ainsi que les appels et réponses d’outil. Un échec de code ou un verdict NEEDS_REFINEMENT enrichit le contexte avant la tentative suivante : l’erreur exacte de l’Executor et le verdict du Verifier sont ajoutés au contexte (Tentative N : erreur=..., verdict=..., corrige le code.) avant que Planner et Coder ne rejouent avec ce supplément d’information. La boucle est asynchrone parce que chaque rôle traverse réellement Runner.run_async ; elle ne se contente plus d’appeler un client LLM direct sous un nom d’agent.

max_iterations borne le nombre de tentatives — donc le coût — et empêche une correction sans fin. Le défaut 2 laisse une seconde chance après un premier échec, et c’est la valeur utilisée dans les sections 7 et 8. Si toutes les itérations échouent, le pipeline rend un résultat explicite — erreur, verdict, code généré et preuve collectée — plutôt qu’un échec silencieux : l’appelant peut distinguer un succès métier d’une simple réponse textuelle plausible.

class DSStarPipeline:
    """Pipeline DS-STAR : analyse locale et rôles LLM exécutés par Google ADK."""

    def __init__(self, max_iterations: int = 2):
        self.file_analyzer = FileAnalyzer()
        self.max_iterations = max_iterations
        self.run_count = 0

    async def analyze_file(self, file_path: str, question: str) -> Dict:
        self.run_count += 1
        run_id = f"lab12-run-{self.run_count}"
        print(f"[FILE ANALYZER] Analyse de {Path(file_path).name} ({run_id})...")
        meta = self.file_analyzer.analyze_csv(file_path)
        context = self.file_analyzer.generate_context(meta)
        df_local = pd.read_csv(file_path)
        executor = Executor(df_local)

        def get_file_schema() -> Dict:
            """Retourne au Planner ADK le schéma réellement analysé."""
            return {
                "filename": meta.filename,
                "rows": int(meta.num_rows),
                "columns": {name: str(dtype) for name, dtype in df_local.dtypes.items()},
            }

        planner = build_agent(
            "lab12_planner",
            "Planifie une analyse de fichier à partir du schéma observé.",
            (
                "Appelle obligatoirement get_file_schema. Réponds ensuite avec "
                "REASONING: sur une ligne, puis STEPS: et trois étapes numérotées."
            ),
            tools=(get_file_schema,),
        )
        coder = build_agent(
            "lab12_coder",
            "Transforme un plan en code Pandas exécutable sur le DataFrame df.",
            (
                "Réponds uniquement avec un bloc ```python```. Le DataFrame df est "
                "déjà chargé : ne lis aucun fichier. Respecte exactement le schéma et "
                "la métrique demandée. Utilise print() pour le résultat final. Affiche "
                "les montants avec deux décimales, sans séparateur de milliers. Pour "
                "une agrégation mensuelle avec pd.Grouper, utilise freq='ME', jamais 'M'."
            ),
        )
        verifier = build_agent(
            "lab12_verifier",
            "Vérifie qu'une sortie d'analyse répond à la question initiale.",
            (
                "Commence impérativement par SUCCESS, NEEDS_REFINEMENT ou FAILED, "
                "puis justifie en une phrase. Refuse une sortie vide ou hors schéma."
            ),
        )

        evidence = {}
        last_result = ExecutionResult(False, "", "Aucune exécution")
        last_verdict = "FAILED"
        code = ""

        for iteration in range(self.max_iterations):
            print(f"\n=== ITERATION {iteration + 1}/{self.max_iterations} ===")

            planner_session = f"{run_id}-planner-{iteration}"
            plan_turn = await run_agent_turn(
                planner,
                f"QUESTION: {question}\nCONTEXTE: {context}",
                session_id=planner_session,
            )
            evidence[f"planner-{iteration}"] = {
                "session_id": planner_session,
                "events": plan_turn.event_count,
                "tool_calls": ", ".join(plan_turn.tool_calls),
                "tool_responses": ", ".join(plan_turn.tool_responses),
            }
            print(f"[PLANNER ADK] {plan_turn.response_text[:160]}...")

            coder_session = f"{run_id}-coder-{iteration}"
            code_turn = await run_agent_turn(
                coder,
                (
                    f"QUESTION: {question}\nPLAN:\n{plan_turn.response_text}\n"
                    f"SCHÉMA: {json.dumps(get_file_schema(), ensure_ascii=False)}\n"
                    f"CONTEXTE: {context}"
                ),
                session_id=coder_session,
            )
            evidence[f"coder-{iteration}"] = {
                "session_id": coder_session,
                "events": code_turn.event_count,
                "tool_calls": ", ".join(code_turn.tool_calls),
                "tool_responses": ", ".join(code_turn.tool_responses),
            }
            match = re.search(
                r"```(?:python)?\s*(.*?)\s*```",
                code_turn.response_text,
                re.DOTALL,
            )
            code = match.group(1).strip() if match else code_turn.response_text.strip()
            last_result = executor.execute(code)
            print(f"[EXECUTOR] success={last_result.success}")
            if last_result.error:
                print(f"[EXECUTOR ERROR] {last_result.error}")

            verifier_session = f"{run_id}-verifier-{iteration}"
            verify_turn = await run_agent_turn(
                verifier,
                (
                    f"QUESTION: {question}\n"
                    f"SCHÉMA: {json.dumps(get_file_schema(), ensure_ascii=False)}\n"
                    f"SUCCÈS EXÉCUTION: {last_result.success}\n"
                    f"ERREUR: {last_result.error}\nSORTIE:\n{last_result.output[:1200]}"
                ),
                session_id=verifier_session,
            )
            evidence[f"verifier-{iteration}"] = {
                "session_id": verifier_session,
                "events": verify_turn.event_count,
                "tool_calls": ", ".join(verify_turn.tool_calls),
                "tool_responses": ", ".join(verify_turn.tool_responses),
            }
            last_verdict = verify_turn.response_text.strip()
            print(f"[VERIFIER ADK] {last_verdict}")
            verdict_match = re.match(
                r"^(SUCCESS|NEEDS_REFINEMENT|FAILED)\b",
                last_verdict.upper(),
            )
            verdict_token = verdict_match.group(1) if verdict_match else "FAILED"
            if last_result.success and verdict_token == "SUCCESS":
                return {
                    "success": True,
                    "output": last_result.output,
                    "code": code,
                    "iterations": iteration + 1,
                    "verdict": last_verdict,
                    "planner_tool_round_trip": plan_turn.tool_was_invoked,
                    "evidence": evidence,
                }

            context += (
                f"\nTentative {iteration + 1}: erreur={last_result.error}; "
                f"verdict={last_verdict}; corrige le code."
            )

        return {
            "success": False,
            "output": last_result.output,
            "error": last_result.error or "Max iterations atteint",
            "code": code,
            "iterations": self.max_iterations,
            "verdict": last_verdict,
            "planner_tool_round_trip": any(
                item["tool_calls"] and item["tool_responses"]
                for role, item in evidence.items()
                if role.startswith("planner-")
            ),
            "evidence": evidence,
        }

print("DS-STAR prêt : Planner, Coder et Verifier utilisent de vrais Agent/Runner")
DS-STAR prêt : Planner, Coder et Verifier utilisent de vrais Agent/Runner

6. Création d’un dataset de test

Le dataset synthétique contient 150 ventes réparties entre trois produits et quatre régions. np.random.seed(42) rend les valeurs reproductibles : les oracles Pandas des deux sections suivantes calculent donc des résultats stables.

Le problème reste lisible, mais il exerce plusieurs capacités distinctes du pipeline : inspection de schéma par outil, agrégation catégorielle, conversion temporelle, génération de code et vérification métier indépendante.

# Dataset de ventes multi-produits
import tempfile
test_dir = tempfile.mkdtemp()

np.random.seed(42)
df = pd.DataFrame({
    'date': pd.date_range('2024-01-01', periods=150, freq='D'),
    'product': np.random.choice(['Widget A', 'Widget B', 'Gadget X'], 150),
    'region': np.random.choice(['Nord', 'Sud', 'Est', 'Ouest'], 150),
    'revenue': np.random.uniform(100, 2000, 150).round(2),
    'units': np.random.randint(1, 50, 150)
})

csv_path = os.path.join(test_dir, 'sales.csv')
df.to_csv(csv_path, index=False)
print(f'Dataset cree: {os.path.basename(test_dir)}{os.sep}{os.path.basename(csv_path)}')
print(f'{len(df)} lignes, colonnes: {list(df.columns)}')
Dataset cree: tmpyulx5vy9\sales.csv
150 lignes, colonnes: ['date', 'product', 'region', 'revenue', 'units']

7. Première analyse réelle : revenu total par produit

Le premier scénario traverse toute la chaîne FileAnalyzer → Planner ADK → Coder ADK → Executor → Verifier ADK. Le Planner doit consulter le schéma par outil ; le Coder opère ensuite sur le DataFrame df déjà injecté.

La cellule calcule aussi, directement avec Pandas, le produit et le revenu attendus. Ces valeurs indépendantes servent d’oracle métier : un verdict LLM positif ne suffit pas si la sortie exécutée ne contient pas le bon résultat.

question_product = (
    "Quel produit génère le plus de revenu total ? "
    "Affiche son nom et le revenu total correspondant."
)
pipeline = DSStarPipeline(max_iterations=2)
product_result = await pipeline.analyze_file(csv_path, question_product)

product_totals = df.groupby('product')['revenue'].sum()
expected_product = product_totals.idxmax()
expected_revenue = float(product_totals.max())
observed_numbers = [
    float(value.replace(',', '.'))
    for value in re.findall(r"\d+(?:[.,]\d+)?", product_result['output'])
]

assert product_result['planner_tool_round_trip'], (
    "Le Planner ADK n'a pas effectué le round-trip get_file_schema"
)
assert product_result['success'], product_result.get('error', product_result['verdict'])
assert expected_product in product_result['output'], (
    "La sortie exécutée ne nomme pas le produit calculé indépendamment"
)
assert any(abs(value - expected_revenue) < 0.02 for value in observed_numbers), (
    "La sortie exécutée ne contient pas le revenu total attendu"
)

print(f"Produit attendu (Pandas): {expected_product}")
print(f"Revenu attendu (Pandas): {expected_revenue:.2f}")
print(f"Succès pipeline: {product_result['success']}")
print(f"Itérations: {product_result['iterations']}")
print(f"Planner tool round-trip: {product_result['planner_tool_round_trip']}")
print(f"Verdict ADK: {product_result['verdict']}")
print(f"Sortie Executor:\n{product_result['output'].strip()}")
print("\nPreuve ADK:")
for role, item in product_result['evidence'].items():
    print(
        f"- {role}: session={item['session_id']}, events={item['events']}, "
        f"tool_calls={item['tool_calls'] or 'aucun'}, "
        f"tool_responses={item['tool_responses'] or 'aucune'}"
    )
[FILE ANALYZER] Analyse de sales.csv (lab12-run-1)...

=== ITERATION 1/2 ===
[PLANNER ADK] REASONING: Pour déterminer quel produit génère le plus de revenu total, nous allons analyser les données du fichier sales.csv en faisant une somme des revenus p...
[EXECUTOR] success=True
[VERIFIER ADK] SUCCESS, la sortie répond correctement à la question initiale en indiquant le produit qui génère le plus de revenu total avec son nom et le montant correspondant.
Produit attendu (Pandas): Gadget X
Revenu attendu (Pandas): 59242.44
Succès pipeline: True
Itérations: 1
Planner tool round-trip: True
Verdict ADK: SUCCESS, la sortie répond correctement à la question initiale en indiquant le produit qui génère le plus de revenu total avec son nom et le montant correspondant.
Sortie Executor:
Gadget X: 59242.44

Preuve ADK:
- planner-0: session=lab12-run-1-planner-0, events=3, tool_calls=get_file_schema, tool_responses=get_file_schema
- coder-0: session=lab12-run-1-coder-0, events=1, tool_calls=aucun, tool_responses=aucune
- verifier-0: session=lab12-run-1-verifier-0, events=1, tool_calls=aucun, tool_responses=aucune

Lecture du résultat : agrégation par produit

La sortie précédente est une preuve de bout en bout, pas un simple texte de modèle. Elle montre trois sessions ADK distinctes et le round-trip get_file_schema du Planner. Le code du Coder a réellement été exécuté sur df, puis son résultat a été confronté à l’agrégation Pandas calculée indépendamment dans la même cellule.

Analyse de la sortie :

Étape Rôle Preuve observable dans la sortie
Analyse locale FileAnalyzer Profil déterministe du CSV (sales.csv, 150 lignes, schéma réel), sans appel LLM caché
Planification Planner ADK Session dédiée, 3 événements, appel et réponse d’outil get_file_schema — le plan se fonde sur les colonnes réellement disponibles
Génération Coder ADK Session distincte, bloc python extrait de la réponse, contenant le groupby demandé
Exécution Executor success=True, sortie Gadget X: 59242.44 imprimée sur le DataFrame déjà chargé
Vérification Verifier ADK Verdict préfixé SUCCESS suivi de sa justification
Oracle métier Pandas Produit et revenu recalculés hors pipeline, comparés à la sortie à 0,02 près

Dans le bloc « Preuve ADK » de la sortie, chaque rôle apparaît avec son identifiant de session, son nombre d’événements et ses échanges d’outils : le Planner y montre le couple appel/réponse get_file_schema, tandis que Coder et Verifier terminent en un seul événement de réponse finale — ils n’ont pas d’outil à invoquer. Cette trace observable est ce qui distingue une exécution d’agent d’un simple appel de complétion déguisé.

Les assertions portent donc sur trois niveaux complémentaires : protocole ADK (le round-trip d’outil a bien eu lieu), verdict du Verifier et exactitude métier du nom et du montant (tolérance de 0,02 sur le revenu total). Toute divergence fait échouer le notebook au lieu d’être reformulée après coup dans cette prose.

Le runtime multi-provider reste configurable : changer de modèle peut modifier le plan ou la forme du code, mais pas l’oracle attendu. C’est ce qui rend l’expérience reproductible et falsifiable d’un provider à l’autre.

Leçon pédagogique : un verdict LLM positif n’est pas une preuve. Le Verifier ADK peut se montrer indulgent ou valider une réponse bien formulée mais fausse ; l’oracle Pandas, lui, est déterministe. La sortie exécutée doit donc contenir le montant attendu à la tolérance près, quoi que déclare l’agent. Un pipeline agentique crédible sépare toujours le jugement génératif de la vérification falsifiable.

# Exercice : posez une nouvelle question au pipeline Google ADK.
# L'exemple guidé précédent a déjà prouvé l'exécution réelle du runtime.

# TODO étudiant : décommentez puis adaptez la question ou max_iterations.
# extension_pipeline = DSStarPipeline(max_iterations=2)
# extension_question = "Quelle région a vendu le plus d'unités ?"
# extension_result = await extension_pipeline.analyze_file(csv_path, extension_question)
# print(f"Succès: {extension_result['success']}")
# print(f"Itérations: {extension_result['iterations']}")
# print(f"Sortie: {extension_result['output'][:300]}")

print("Exercice à compléter : formuler et vérifier une nouvelle analyse ADK")
Exercice à compléter : formuler et vérifier une nouvelle analyse ADK

Exercice d’extension du pipeline

La cellule précédente reste volontairement un exercice : elle invite à déclencher une autre analyse après configuration du provider, sans fournir la solution. Le workshop a déjà exécuté le pipeline réel dans l’exemple guidé ci-dessus ; ce squelette commenté sert à modifier la question, le nombre d’itérations ou le provider sans confondre démonstration résolue et travail étudiant. Il est conforme à la convention des notebooks : ni appel débranché ni erreur volontaire, uniquement un TODO étudiant à compléter.

Pistes d’extension :

  • Changer la question métier — par exemple la région qui totalise le plus d’unités vendues, le produit au revenu médian ou le mois le plus faible — et écrire l’oracle Pandas correspondant avant de lancer le pipeline : colonne de groupement, mesure, opération et format de sortie attendus.
  • Ajuster max_iterations : le baisser à 1 pour observer un échec sans retente, le monter pour une question plus difficile, et lire la différence dans le champ iterations du résultat comme dans le coût en tours. L’objectif n’est pas d’obtenir toujours un succès, mais de savoir attribuer un échec à l’interface, au code généré ou au critère métier.
  • Explorer le dictionnaire renvoyé : success, iterations, verdict, code et surtout evidence, qui conserve pour chaque rôle l’identifiant de session, le nombre d’événements et les échanges d’outils.

Observez ensuite les trois niveaux de diagnostic :

  1. le Planner a-t-il appelé get_file_schema et utilisé les bons noms de colonnes ?
  2. le code du Coder travaille-t-il sur df sans dépendre d’un chemin local ?
  3. le Verifier distingue-t-il une sortie exacte d’une réponse plausible mais incomplète ?

La démarche attendue est celle des sections 7 et 8 : formuler l’oracle avant de lire la sortie de l’agent, pour que la comparaison reste une vérification indépendante et non une justification a posteriori. Un bon réflexe d’agentique : ce qu’un LLM affirme se vérifie, ce qu’un calcul déterministe produit se constate.

8. Deuxième analyse réelle : évolution mensuelle

Le second scénario exige une transformation temporelle : convertir date, agréger revenue par mois, puis imprimer chaque période et son total. Il réutilise la même architecture ADK, mais avec de nouvelles sessions et un nouveau code généré.

Un oracle Pandas indépendant vérifie que chaque total mensuel attendu apparaît dans la sortie. Cette comparaison attrape notamment une colonne inventée (revenu au lieu de revenue) ou une agrégation sur la mauvaise métrique, même si un LLM tentait de déclarer SUCCESS.

question_month = (
    "Montre l'évolution du revenu total par mois. Convertis date en datetime, "
    "agrège revenue par période mensuelle et affiche chaque mois avec son total."
)
monthly_result = await pipeline.analyze_file(csv_path, question_month)

expected_monthly = (
    df.assign(date=pd.to_datetime(df['date']))
    .groupby(pd.Grouper(key='date', freq='ME'))['revenue']
    .sum()
)
observed_monthly_numbers = [
    float(value.replace(',', '.'))
    for value in re.findall(r"\d+(?:[.,]\d+)?", monthly_result['output'])
]

assert monthly_result['planner_tool_round_trip'], (
    "Le second Planner ADK n'a pas appelé get_file_schema"
)
assert monthly_result['success'], monthly_result.get('error', monthly_result['verdict'])
for total in expected_monthly:
    assert any(abs(value - float(total)) < 0.02 for value in observed_monthly_numbers), (
        f"Total mensuel attendu absent de la sortie: {total:.2f}"
    )

print("Totaux mensuels attendus (Pandas):")
for month, total in expected_monthly.items():
    print(f"- {month:%Y-%m}: {total:.2f}")
print(f"Succès pipeline: {monthly_result['success']}")
print(f"Itérations: {monthly_result['iterations']}")
print(f"Planner tool round-trip: {monthly_result['planner_tool_round_trip']}")
print(f"Verdict ADK: {monthly_result['verdict']}")
print(f"Sortie Executor:\n{monthly_result['output'].strip()}")
print("\nPreuve ADK:")
for role, item in monthly_result['evidence'].items():
    print(
        f"- {role}: session={item['session_id']}, events={item['events']}, "
        f"tool_calls={item['tool_calls'] or 'aucun'}, "
        f"tool_responses={item['tool_responses'] or 'aucune'}"
    )
[FILE ANALYZER] Analyse de sales.csv (lab12-run-2)...

=== ITERATION 1/2 ===
[PLANNER ADK] REASONING: Le fichier contient des données de ventes avec des colonnes pertinentes pour l'analyse des revenus par mois.

STEPS:
1. Convertir la colonne `date` e...
[EXECUTOR] success=True
[VERIFIER ADK] SUCCESS, la sortie d'analyse montre correctement l'évolution du revenu total par mois en agrégant les revenus par période mensuelle et en affichant chaque mois avec son total.
Totaux mensuels attendus (Pandas):
- 2024-01: 33626.92
- 2024-02: 30399.24
- 2024-03: 38575.22
- 2024-04: 32180.79
- 2024-05: 30531.25
Succès pipeline: True
Itérations: 1
Planner tool round-trip: True
Verdict ADK: SUCCESS, la sortie d'analyse montre correctement l'évolution du revenu total par mois en agrégant les revenus par période mensuelle et en affichant chaque mois avec son total.
Sortie Executor:
date
2024-01-31    33626.92
2024-02-29    30399.24
2024-03-31    38575.22
2024-04-30    32180.79
2024-05-31    30531.25
Freq: ME, Name: revenue, dtype: object

Preuve ADK:
- planner-0: session=lab12-run-2-planner-0, events=3, tool_calls=get_file_schema, tool_responses=get_file_schema
- coder-0: session=lab12-run-2-coder-0, events=1, tool_calls=aucun, tool_responses=aucune
- verifier-0: session=lab12-run-2-verifier-0, events=1, tool_calls=aucun, tool_responses=aucune

Lecture du résultat : agrégation temporelle

La cellule précédente expose de nouveau les événements des trois rôles, puis affiche côte à côte les totaux Pandas attendus et la sortie réellement produite par le code du Coder. Le schéma transmis contient explicitement date et revenue, ce qui réduit la dérive de vocabulaire tout en laissant au LLM la responsabilité de construire l’analyse.

Points clés :

  1. Question plus exigeante : convertir date en datetime, agréger par période, afficher chaque mois — trois étapes où le code peut dévier, contre une seule pour l’agrégation par produit. Le code généré reste libre dans sa forme, mais il doit préserver les cinq périodes et leurs montants.
  2. Pièges attrapés par l’oracle : une colonne inventée (revenu au lieu de revenue), une agrégation sur la mauvaise métrique ou un mois manquant font échouer l’assertion, même si le Verifier ADK déclarait SUCCESS.
  3. Détail d’API réel : l’instruction du Coder impose freq='ME' ; l’alias historique 'M' est déprécié par Pandas pour les périodes mensuelles, et il reste fréquent dans les exemples plus anciens — le prompt ancre donc le code généré sur ce qu’attend la version installée.

Le test ne se contente pas de rechercher le mot SUCCESS : chaque total mensuel calculé indépendamment doit apparaître dans la sortie capturée par l’Executor. L’agent reste génératif, mais son résultat est évalué par un invariant métier déterministe.

Les identifiants lab12-run-2-{planner,coder,verifier}-0 prouvent aussi que le deuxième scénario ne réutilise pas silencieusement les sessions du premier : le Planner effectue un nouveau round-trip de schéma et le Coder produit un nouveau programme ; seule la configuration partagée du pipeline est réutilisée.

Si une tentative échoue, lisez les preuves dans l’ordre : erreur de l’Executor, premier token du verdict, puis trace d’outil du Planner. Cette séquence permet de distinguer un code invalide, une sortie métier incomplète et un plan fondé sur un schéma erroné, sans attribuer automatiquement le défaut au modèle.

Leçon pédagogique : plus la transformation demandée est riche (typage temporel, agrégation, formatage), plus l’oracle doit être précis. Comparer chaque total mensuel un à un attrape des erreurs qu’aucun verdict LLM ne détecterait.

9. Nettoyage

Suppression du répertoire temporaire contenant le dataset de test. Bonne pratique pour les notebooks qui créent des fichiers éphémères : on évite d’encombrer le système de fichiers du runner et on garantit qu’une exécution suivante repart d’un état propre.

import shutil

shutil.rmtree(test_dir)
print(f"Répertoire temporaire supprimé: {Path(test_dir).name}")
Répertoire temporaire supprimé: tmpyulx5vy9

10. Résumé du workshop

Architecture exécutée

Le pipeline sépare le raisonnement LLM, porté par trois vrais agents Google ADK, de l’action Pandas locale et inspectable. Le Planner consulte le schéma par outil avant que le Coder produise du code ; le Verifier rend ensuite un verdict sur la sortie de l’Executor.

flowchart TD
    F[Fichier CSV] --> A[FileAnalyzer déterministe]
    A --> P[Planner Agent + get_file_schema]
    P --> C[Coder Agent]
    C --> E[Executor Pandas local]
    E --> V[Verifier Agent]
    V -->|SUCCESS| R[Résultat vérifié]
    V -->|NEEDS_REFINEMENT ou FAILED| P

Points clés

  1. Runtime réel : chaque rôle LLM traverse une session et un Runner Google ADK.
  2. Preuve observable : événements, appels d’outil et réponses d’outil sont conservés.
  3. Code inspectable : le Coder génère le code, mais seul l’Executor local l’exécute sur df.
  4. Validation falsifiable : les exemples comparent la sortie à des calculs Pandas indépendants.
  5. Extensions : les exercices proposent le multi-fichiers, des critères métier et un rapport Markdown sans livrer leur solution.

Prochaine étape

Le Lab 13 étend cette architecture vers la recherche de modèles SOTA.

Exercice : Pipeline Multi-Fichiers

Etendez le pipeline DS-STAR pour analyser plusieurs fichiers en sequence et generer un rapport consolide.

Objectifs

  1. Créer 3 fichiers CSV avec des données complementaires
  2. Modifier le pipeline pour traiter plusieurs fichiers
  3. Generer un rapport final synthetique

Instructions

# Exercice: Creez 3 fichiers CSV lies par une cle commune
# Exemple : clients.csv, commandes.csv, produits.csv
import tempfile
test_dir = tempfile.mkdtemp()

# Exercice: Creez les datasets
clients = pd.DataFrame({
    'client_id': [...],
    'nom': [...],
    'region': [...]
})
clients.to_csv(f"{test_dir}/clients.csv", index=False)

# Exercice: Implementez une fonction analyse_multi_fichiers
def analyse_multi_fichiers(file_paths: list, questions: list) -> dict:
    """
    Analyse plusieurs fichiers avec le pipeline DS-STAR.
    
    Args:
        file_paths: Liste de chemins vers les fichiers CSV
        questions: Liste de questions pour chaque fichier
    
    Returns:
        Dictionnaire avec resultats par fichier et synthese
    """
    pipeline = DSStarPipeline(max_iterations=2)
    resultats = {}
    
    # Exercice: Parcourir chaque fichier et analyser
    # Exercice: Consolidation des resultats
    
    return resultats

# Exercice: Testez avec vos fichiers et questions
print("Exercice a completer")
Exercice a completer

Exercice : Amelioration du Verifier avec Critères Metier

Le Verifier actuel est generique (SUCCESS / NEEDS_REFINEMENT). L’objectif est de créer un Verifier specialise qui evalue la qualite des résultats selon des critères metier précis (exhaustivite, precision numérique, format de sortie).

Objectifs

  1. Définir 3 critères de qualite pour les analyses de données
  2. Implementer un BusinessVerifier avec scoring
  3. Comparer les verdicts du Verifier generique vs. le BusinessVerifier

Indice : - Critères possibles : presence de chiffres dans le résultat, longueur minimale, mots-cles attendus - Attribuez un score 0-1 par critere et un seuil global de validation - Testez sur les mêmes questions que le pipeline original

# Exercice : BusinessVerifier avec criteres metier
# Objectif : Ameliorer la verification avec des criteres quantitatifs

import re

class BusinessVerifier(Verifier):
    """Verifier specialise avec criteres metier quantifiables."""
    
    def __init__(self, llm: LLMClient, min_length: int = 50, score_threshold: float = 0.6):
        super().__init__(llm)
        self.min_length = min_length
        self.score_threshold = score_threshold
    
    def score_result(self, question: str, result_output: str) -> dict:
        """
        Score le resultat selon 3 criteres metier.
        
        Returns:
            Dictionnaire avec score par critere (0.0-1.0) et score global
        """
        scores = {}
        
        # Critere 1: Exhaustivite - le resultat contient-il des chiffres ?
        has_numbers = bool(re.search(r'\d+\.?\d*', result_output))
        scores['exhaustivite_numerique'] = 1.0 if has_numbers else 0.0
        
        # Critere 2: Longueur suffisante
        scores['longueur_minimale'] = 1.0 if len(result_output) >= self.min_length else len(result_output) / self.min_length
        
        # Critere 3: Pertinence - le resultat mentionne-t-il des mots de la question ?
        # TODO etudiant : extrayez les mots-cles de la question et verifiez leur presence
        question_words = set(question.lower().split()) - {'le', 'la', 'les', 'de', 'du', 'des', 'un', 'une', 'et', 'est', 'quel', 'quelle', 'montre', 'quel'}
        output_lower = result_output.lower()
        matched = sum(1 for w in question_words if w in output_lower)
        scores['pertinence_mots_cles'] = matched / max(len(question_words), 1)
        
        # Score global
        scores['global'] = sum(scores.values()) / len(scores)
        
        return scores
    
    def verify_business(self, question: str, result) -> Tuple[VerificationStatus, str]:
        """Verification avec scoring metier."""
        if not result.success:
            return VerificationStatus.FAILED, f"Erreur: {result.error}"
        
        scores = self.score_result(question, result.output)
        
        if scores['global'] >= self.score_threshold:
            return VerificationStatus.SUCCESS, f"Score: {scores['global']:.2f} - {scores}"
        return VerificationStatus.NEEDS_REFINEMENT, f"Score insuffisant: {scores['global']:.2f} - {scores}"

# TODO: Testez le BusinessVerifier
# b_verifier = BusinessVerifier(LLMClient(), min_length=30, score_threshold=0.5)
# 
# # Simulez un resultat
# fake_result = ExecutionResult(success=True, output="Le revenu total par region: Nord=12500, Sud=18300, Est=9800")
# status, msg = b_verifier.verify_business("Quel est le revenu par region?", fake_result)
# print(f"Status: {status.value}")
# print(f"Message: {msg}")

print("Exercice a completer : BusinessVerifier avec criteres metier")
Exercice a completer : BusinessVerifier avec criteres metier

Exercice : Generation de Rapport Automatique

Créez un générateur de rapport Markdown qui consolide les résultats de plusieurs analyses DS-STAR en un document structure. L’objectif est de produire un rapport avec table des matieres, sections par question et tableau recapitulatif.

Objectifs

  1. Implementer un ReportGenerator qui collecte les résultats du pipeline
  2. Generer un rapport Markdown avec sections formatees
  3. Inclure un tableau recapitulatif des questions, statuts et itérations necessaires

Indice : - Utilisez des f-strings pour construire le Markdown dynamiquement - Structurez en sections : titre, context, résultats par question, synthese - Incluez un tableau avec | Question | Statut | Itérations |

# Exercice : Generation de rapport Markdown automatique
# Objectif : Consolidider les resultats DS-STAR en un rapport structure

class SimpleReportGenerator:
    """Generateur de rapport Markdown pour les analyses DS-STAR."""
    
    def __init__(self, title: str = "Rapport d'Analyse DS-STAR"):
        self.title = title
        self.results = []
    
    def add_result(self, question: str, success: bool, output: str, iterations: int):
        """Ajoute un resultat d'analyse au rapport."""
        # TODO etudiant : stockez le resultat dans self.results
        pass
    
    def generate(self) -> str:
        """
        Genere le rapport Markdown complet.
        
        Returns:
            String contenant le rapport en Markdown
        """
        # Etape 1: En-tete du rapport
        report = f"# {self.title}\n\n"
        
        # Etape 2: Resume executif
        # TODO etudiant : ajoutez le nombre total de questions, le taux de succes
        
        # Etape 3: Tableau recapitulatif
        # TODO etudiant : creez un tableau | Question | Statut | Iterations |
        # report += "| Question | Statut | Iterations |\n"
        # report += "|----------|--------|------------|\n"
        
        # Etape 4: Details par question
        # TODO etudiant : pour chaque resultat, ajoutez une section avec la question et l'output
        
        # Etape 5: Synthese et recommandations
        # TODO etudiant : resumez les points cles et les limites observees
        
        return report

# TODO: Testez le generateur de rapport
# reporter = SimpleReportGenerator("Rapport d'Analyse des Ventes")
# reporter.add_result("Revenu par region", True, "Nord: 12K, Sud: 18K", 1)
# reporter.add_result("Top produits", True, "Widget A: 5000 euros", 2)
# reporter.add_result("Tendance mensuelle", False, "Max iterations atteint", 3)
# 
# rapport = reporter.generate()
# print(rapport)

print("Exercice a completer : generation de rapport Markdown automatique")
Exercice a completer : generation de rapport Markdown automatique

Extension (bonus)

  • Ajoutez une étape de jointure automatique entre fichiers
  • Implementez un cache pour eviter de re-analyser les mêmes fichiers
  • Generez un rapport Markdown final avec les insights cles

Références

  1. Nam et al., DS-STAR: Data Science Agent for Solving Diverse Tasks across Heterogeneous Formats and Open-Ended Queries, arXiv:2509.21825, 2025. Agent DS-STAR complet (architecture orchestrée FileAnalyzer + Planner-Coder-Verifier) — synthèse des Labs 10-12.
  2. Z. Xi et al., The Rise and Potential of Large Language Model Based Agents: A Survey, arXiv:2309.07864, 2023. Cadre conceptuel de l’agent LLM modulaire de bout en bout (perception-planification-action-vérification).
  3. S. Yao et al., ReAct: Synergizing Reasoning and Acting in Language Models, arXiv:2210.03629, ICLR 2023. Paradigme de la boucle raisonnement-action (suite Lab 11).
  4. N. Shinn et al., Reflexion: Language Agents with Verbal Reinforcement Learning, arXiv:2303.11366, NeurIPS 2023. Auto-raffinement après vérification (suite Lab 11).
Retour au sommet