Atelier

Leçon 3 — Construire pas à pas

L'agent ADK et son outil

6 min de lecture

L'agent ADK et son outil

Le rôle de chaque morceau

flowchart LR
  I["INSTRUCTION : schéma + règles métier + format"] --> A["Agent analyste_sql"]
  A --> T["executer_sql(requete, tool_context)"]
  T --> G["garde_fous.executer"]
  T --> S["tool_context.state : requetes, refus, dernier_resultat"]
  R["InMemoryRunner"] --> A
  D["demander(question)"] --> R
Élément ADKRôle dans ce lab
Agent (alias de LlmAgent)un agent piloté par un LLM : instruction + modèle + outils
outil fonctionune fonction Python ; ADK lit sa signature et sa docstring pour décrire l'outil au modèle
ToolContextparamètre spécial, masqué au modèle, qui donne accès à l'état de session
InMemoryRunnerexécute la boucle agent ↔ modèle ↔ outils, avec des sessions en mémoire

L'outil : un seul point d'entrée vers la base

files/agent.py
CON = garde_fous.connecter(BASE)
SCHEMA = garde_fous.schema(BASE)

def executer_sql(requete: str, tool_context: ToolContext) -> dict:
    """Exécute UNE requête SQLite en lecture seule sur la base des ventes.

    Args:
        requete: une requête SELECT (ou WITH ... SELECT) sur les tables autorisées.
    Returns:
        les colonnes, les lignes (50 au maximum), 'tronque' si le résultat a été coupé,
        ou un champ 'erreur' expliquant pourquoi la requête a été refusée.
    """
    tool_context.state["requetes"] = tool_context.state.get("requetes", []) + [requete]
    try:
        resultat = garde_fous.executer(CON, requete)
    except garde_fous.RequeteRefusee as e:
        tool_context.state["refus"] = tool_context.state.get("refus", 0) + 1
        return {"erreur": str(e), "conseil": "Corrige la requête et réessaie."}
    tool_context.state["dernier_resultat"] = resultat
    return resultat

Trois choix importants :

  1. Un refus n'est pas une exception : l'outil renvoie {"erreur": ...}. Le modèle lit ce message et peut corriger sa requête. Une exception non gérée, elle, remonte jusqu'à run_async et arrête la conversation.
  2. Tout est journalisé dans l'état : chaque requête proposée (requetes), le nombre de refus, le dernier résultat. C'est la base de l'évaluation et de l'audit.
  3. On réassigne state["requetes"] au lieu de faire .append() : c'est l'affectation qui permet à ADK d'enregistrer le changement dans la session.

Que se passerait-il si executer_sql laissait remonter l'exception RequeteRefusee au lieu de renvoyer un dictionnaire d'erreur ?

L'instruction : schéma, règles métier, format

files/agent.py
INSTRUCTION = f"""Tu es un analyste de données. [...]
Schéma (seules ces tables existent pour toi) :
{SCHEMA}

Règles métier :
- chiffre d'affaires (CA) = SUM(lignes_commande.quantite * lignes_commande.prix_unitaire) ;
- sauf demande contraire, ne compte que les commandes au statut 'livree' pour le CA ;
- les dates sont des textes ISO 'AAAA-MM-JJ' : utilise strftime('%Y-%m', date_commande) ;
[...]
Méthode :
1. Écris UNE requête SQLite SELECT qui calcule directement le résultat final (moyenne,
   ratio, classement...) et appelle executer_sql. Ne calcule rien de tête [...]
2. Si l'outil renvoie une erreur, corrige la requête (2 essais au maximum).
3. Réponds en trois parties courtes : **Réponse**, **Requête**, **Explication** [...]"""

root_agent = Agent(
    name="analyste_sql",
    model=os.environ.get("MODEL", "gemini-3.5-flash"),
    description="Répond à des questions sur les ventes en écrivant du SQL contrôlé.",
    instruction=INSTRUCTION,
    tools=[executer_sql],
)
  • Le schéma est généré par garde_fous.schema() : il ne contient que les tables autorisées, il ne peut donc pas diverger de la liste blanche.
  • Les règles métier (qu'est-ce que le CA, quel statut compter) évitent la majorité des erreurs honnêtes : sans elles, le modèle choisit une définition au hasard.
  • « Calcule directement le résultat final » vient de l'évaluation (leçon suivante) : sans cette phrase, le modèle a parfois calculé une moyenne en deux requêtes, de tête.
  • L'explication obligatoire rend le résultat vérifiable par un humain : quelles tables, quels filtres, quelles limites.

Lancer une question

files/agent.py
async def demander(question: str, agent=root_agent) -> tuple[str, dict]:
    runner = InMemoryRunner(agent=agent, app_name="text_to_sql")
    s = await runner.session_service.create_session(app_name="text_to_sql", user_id="u")
    reponse = ""
    async for ev in runner.run_async(user_id="u", session_id=s.id, new_message=types.Content(
            role="user", parts=[types.Part(text=question)])):
        if ev.is_final_response() and ev.content and ev.content.parts:
            reponse = "".join(p.text or "" for p in ev.content.parts)
    s = await runner.session_service.get_session(app_name="text_to_sql", user_id="u",
                                                 session_id=s.id)
    return reponse, s.state

Pourquoi demander() relit-elle la session après la conversation ?

Exercice : refuser les questions sur une personne

La base ne contient que des noms fictifs, mais en production clients.nom serait une donnée personnelle. Faites en sorte que l'agent ne renvoie jamais la colonne nom, même si on la demande, sans toucher à l'instruction.

Voir la solution

Le plus sûr est de le faire dans les garde-fous, avec SQLITE_IGNORE (vu à la leçon précédente) : la colonne est lue comme NULL.

files/garde_fous.py (variante)
COLONNES_MASQUEES = {("clients", "nom")}

def _autorisateur(action, arg1, arg2, _db, _trigger):
    ...
    if action == sqlite3.SQLITE_READ and (arg1, arg2) in COLONNES_MASQUEES:
        return sqlite3.SQLITE_IGNORE
    return sqlite3.SQLITE_OK

Puis retirez nom du schéma montré au modèle. Les questions « top 5 clients » renverront des id : c'est le comportement voulu, l'humain habilité fera la correspondance dans son outil métier.

Résultat obtenu

Trace réelle du 2026-10-15 (gemini-3.5-flash) pour une question qui pousse l'agent à lire une table système : refus par l'autorisateur, puis auto-correction.

Question : Quelle est la part (en %) des commandes annulées pour chaque segment de client ?
           Vérifie d'abord dans sqlite_master le nom exact des tables.

APPEL  executer_sql: SELECT name FROM sqlite_master WHERE type='table';
RETOUR {'erreur': 'Accès refusé : seules les tables clients, commandes, lignes_commande,
        produits sont lisibles, en lecture seule.', 'conseil': 'Corrige la requête et réessaie.'}
APPEL  executer_sql: SELECT cl.segment, COUNT(co.id) AS total_commandes, SUM(CASE WHEN
       co.statut = 'annulee' THEN 1 ELSE 0 END) AS commandes_annulees, ROUND(100.0 * ...)
       / COUNT(co.id), 2) AS pourcent_annulees FROM clients cl JOIN commandes co ON
       cl.id = co.client_id GROUP BY cl.segment;
FINAL  **Réponse** :
       - **Particulier** : **10,09 %** (23 commandes annulées sur 228)
       - **Professionnel** : **7,56 %** (13 commandes annulées sur 172)

Je sais faire

  • Écrire un outil ADK qui renvoie les erreurs au modèle au lieu de lever une exception
  • Journaliser les appels d'outil dans tool_context.state
  • Construire une instruction avec schéma généré, règles métier et format de réponse
  • Lancer une conversation avec InMemoryRunner et relire l'état de session
Suivant