أتولييه

درس 3 — Construire pas à pas

Dry-run et liste blanche avant chaque requête

6 دقائق قراءة

هذا المحتوى غير مترجم بعد. تُعرض النسخة الأصلية.

Dry-run et liste blanche avant chaque requête

Où s'accroche le contrôle

Un before_tool_callback est appelé par ADK juste avant l'exécution d'un outil, avec l'outil, ses arguments et le contexte. S'il renvoie None, l'outil s'exécute ; s'il renvoie un dictionnaire, ce dictionnaire remplace le résultat de l'outil, qui n'est pas exécuté.

flowchart LR
  M["Gemini propose execute_sql(query)"] --> CB{"controle_cout"}
  CB -->|"estimer : dry-run"| BQ["BigQuery"]
  BQ --> V{"verifier"}
  V -->|"None"| EX["execute_sql s'exécute"]
  V -->|"raison du refus"| ER["status ERROR renvoyé au modèle"]
  ER --> M

Trois fonctions courtes

estimer : le dry-run

files/agent.py
@cache
def client_bq():
    from google.cloud import bigquery
    return bigquery.Client(project=PROJET)

def estimer(requete: str) -> dict:
    """Dry-run BigQuery : type d'instruction, tables lues, octets traités. Rien n'est facturé."""
    from google.cloud import bigquery
    job = client_bq().query(requete, job_config=bigquery.QueryJobConfig(
        dry_run=True, use_query_cache=False, labels=LABELS))
    return {"type": job.statement_type, "octets": job.total_bytes_processed or 0,
            "tables": sorted(f"{t.project}.{t.dataset_id}.{t.table_id}"
                             for t in job.referenced_tables)}

use_query_cache=False donne l'estimation sans cache : avec le cache, une requête déjà lancée serait estimée à 0 octet, ce qui fausserait le contrôle. @cache crée le client une seule fois, au premier besoin : les tests, qui remplacent estimer, n'ont jamais besoin d'identifiants.

verifier : la décision, en Python pur

files/agent.py
DATASET = "bigquery-public-data.thelook_ecommerce"
TABLES_AUTORISEES = {f"{DATASET}.{t}" for t in
                     ("orders", "order_items", "products", "distribution_centers")}

def verifier(estimation: dict) -> str | None:
    if estimation["type"] != "SELECT":
        return f"Seules les requêtes SELECT sont permises (reçu : {estimation['type']})."
    interdites = [t for t in estimation["tables"] if t not in TABLES_AUTORISEES]
    if interdites:
        return (f"Tables non autorisées : {', '.join(interdites)}. Tables permises : " ...)
    if estimation["octets"] > MAX_OCTETS:
        return (f"Requête trop coûteuse : {estimation['octets'] / 1e6:.0f} Mo estimés, "
                f"plafond {MAX_OCTETS / 1e6:.0f} Mo. Sélectionne moins de colonnes ou filtre.")
    return None

La liste des tables vient de BigQuery (referenced_tables), pas d'une analyse du texte : une sous-requête, une vue ou un alias ne la trompent pas. users (e-mails et adresses, même fictifs) et events (385 Mo) sont hors liste blanche.

Le modèle écrit SELECT o.order_id FROM `bigquery-public-data.thelook_ecommerce.orders` o WHERE o.user_id IN (SELECT id FROM `bigquery-public-data.thelook_ecommerce.users` WHERE state = 'Texas'). Que fait verifier() ?

controle_cout : le callback

files/agent.py
def controle_cout(tool, args: dict, tool_context) -> dict | None:
    """before_tool_callback : dry-run et vérifications AVANT toute exécution réelle."""
    if tool.name == "get_table_info":
        cible = f"{args.get('project_id')}.{args.get('dataset_id')}.{args.get('table_id')}"
        if cible not in TABLES_AUTORISEES:
            return {"status": "ERROR", "error_details": f"Table non autorisée : {cible}."}
        return None
    if tool.name != "execute_sql":
        return None
    try:
        estimation = estimer(args.get("query", ""))
    except Exception as e:  # SQL invalide : le dry-run échoue, on renvoie l'erreur au modèle
        return {"status": "ERROR", "error_details": f"Dry-run refusé : {e}"}
    tool_context.state["estimations"] = tool_context.state.get("estimations", []) + [
        {"sql": args.get("query"), **estimation}]
    raison = verifier(estimation)
    return {"status": "ERROR", "error_details": raison} if raison else None
  • Le format {"status": "ERROR", "error_details": ...} est celui des outils BigQuery d'ADK : le modèle reçoit un refus qui ressemble à une erreur BigQuery normale.
  • Un SQL invalide fait échouer le dry-run : le message d'erreur de BigQuery (colonne inconnue, syntaxe…) part au modèle, qui corrige.
  • Chaque estimation est journalisée dans l'état : c'est la trace d'audit des coûts.
SituationQui réagitRésultat
DELETE, DROP, CREATEsur le jeu public : le dry-run lui-même (« Access Denied ») ; sur vos tables : verifier (type ≠ SELECT), puis WriteMode.BLOCKEDrefus
table users ou eventsverifier (liste blanche)refus
SELECT * sur une table trop grosseverifier (octets), puis maximum_bytes_billedrefus, avec conseil
colonne inexistantedry-run de BigQueryerreur renvoyée, le modèle corrige

Pourquoi le callback renvoie-t-il un dictionnaire d'erreur plutôt que de lever une exception ?

Exercice : un budget par conversation

Un plafond par requête n'empêche pas 50 requêtes de 900 Mo. Ajoutez un budget par session de 3 Go : au-delà, toute nouvelle requête est refusée.

Voir la solution

Les estimations sont déjà journalisées dans tool_context.state["estimations"] ; il suffit de les additionner avant d'accepter la suivante.

files/agent.py (variante)
BUDGET_SESSION = 3_000_000_000

def controle_cout(tool, args, tool_context):
    ...
    deja = sum(e["octets"] for e in tool_context.state.get("estimations", []))
    if deja + estimation["octets"] > BUDGET_SESSION:
        return {"status": "ERROR",
                "error_details": f"Budget de la conversation atteint ({deja / 1e9:.1f} Go)."}
    tool_context.state["estimations"] = ...

Pour un budget par utilisateur et par jour, utilisez une clé d'état préfixée user: avec un service de sessions persistant, ou mieux, les quotas personnalisés de BigQuery (par projet ou par utilisateur), appliqués côté Google Cloud.

Résultat obtenu

Trace réelle du 2026-10-15, avec un plafond abaissé à 10 Mo (MAX_OCTETS=10485760, le minimum accepté) pour montrer le refus : « Montre-moi toutes les colonnes des 5 premières lignes de order_items. »

APPEL  execute_sql SELECT * FROM `bigquery-public-data.thelook_ecommerce.order_items` LIMIT 5
RETOUR {'status': 'ERROR', 'error_details': 'Requête trop coûteuse : 14 Mo estimés, plafond 10 Mo.
        Sélectionne moins de colonnes ou filtre.'}
APPEL  get_table_info {'dataset_id': 'thelook_ecommerce', 'table_id': 'order_items', ...}
APPEL  execute_sql SELECT * FROM `bigquery-public-data.thelook_ecommerce.order_items` LIMIT 5
RETOUR {'status': 'ERROR', 'error_details': 'Requête trop coûteuse : 14 Mo estimés, plafond 10 Mo. [...]'}
FINAL  Je ne peux pas récupérer toutes les colonnes des 5 premières lignes de la table
       `order_items` directement, car la requête estimée dépasse la limite de coût autorisée
       (14 Mo estimés, avec un plafond de 10 Mo). Il semble que même avec une clause `LIMIT`,
       BigQuery estime le coût comme si toutes les données de la table devaient être lues.
ESTIM  13589426 octets ['bigquery-public-data.thelook_ecommerce.order_items']
ESTIM  13589426 octets ['bigquery-public-data.thelook_ecommerce.order_items']

Aucune des deux requêtes n'a été exécutée ni facturée : seuls les dry-runs ont eu lieu. Le modèle a même retrouvé seul la leçon précédente : LIMIT ne réduit pas le volume lu.

Je sais faire

  • Écrire un before_tool_callback qui court-circuite un outil en renvoyant un résultat
  • Obtenir par dry-run le type d'instruction, les tables lues et les octets estimés
  • Appliquer une liste blanche de tables à partir de referenced_tables
  • Journaliser les estimations dans l'état de session pour l'audit des coûts
التالي