Atelier

Lección 2 — Construire pas à pas

L'agent et ses journaux structurés

5 min de lectura

Este contenido aún no está traducido. Se muestra la versión original.

L'agent et ses journaux structurés

Le dossier est l'agent

adk deploy cloud_run prend un dossier et cherche dedans un fichier agent.py qui définit root_agent. Ici, ce dossier est directement files/ :

flowchart TB
  D["files/ (le dossier de l'agent)"] --> A["agent.py : root_agent + outils"]
  D --> R["requirements.txt : dépendances installées dans l'image"]
  D --> I[".gcloudignore : ce qui n'entre PAS dans l'image"]
  D --> C["client.py : appel du service déployé"]
  D --> T["test_agent.py : tests sans clé"]
  D --> S["deployer.sh / nettoyer.sh"]

Les outils

Un outil ADK est une fonction Python ordinaire : ADK lit sa signature et sa docstring pour décrire l'outil au modèle. Les politiques sont fictives et tiennent dans un dictionnaire ; dans un vrai projet, ce serait un intranet ou une base.

files/agent.py
def chercher_politique(sujet: str) -> dict:
    """Cherche la politique interne sur un sujet (congés, télétravail, notes de frais, VPN, mot de passe).

    Args:
        sujet: le thème de la question, en quelques mots.
    """
    cle = _normaliser(sujet)          # minuscules, sans accents : « Télétravail » -> « teletravail »
    for nom, texte in POLITIQUES.items():
        if nom in cle or cle in nom:
            journal("politique_trouvee", sujet=nom)
            return {"trouve": True, "sujet": nom, "politique": texte}
    journal("politique_absente", sujet=cle)
    return {"trouve": False, "sujets_connus": sorted(POLITIQUES)}

Quand rien n'est trouvé, l'outil renvoie la liste des sujets connus plutôt qu'une erreur : le modèle peut alors reformuler ou dire honnêtement qu'il ne sait pas.

Pourquoi chercher_politique renvoie-t-il sujets_connus quand il ne trouve rien, plutôt que lever une exception ?

Des journaux que Cloud Logging comprend

Sur Cloud Run, tout ce que le conteneur écrit sur la sortie standard part dans Cloud Logging. Si la ligne est un objet JSON, elle devient une entrée structurée : le champ severity fixe la sévérité, message le texte affiché, et chaque autre champ devient filtrable (jsonPayload.evenement="ticket_ouvert").

files/agent.py
def journal(evenement: str, **champs) -> None:
    print(json.dumps({"severity": "INFO", "message": evenement, "evenement": evenement, **champs},
                     ensure_ascii=False), file=sys.stdout, flush=True)
Écrire…Ce que vous obtenez dans Cloud Logging
print("ticket ouvert TCK-0001")une ligne de texte (textPayload), recherche plein texte seulement
print(json.dumps({...}))jsonPayload avec des champs filtrables et une vraie sévérité
logging.info(...) sans configurationdu texte, souvent avec la sévérité par défaut

Le conteneur écrit la ligne {"severity": "INFO", "message": "ticket_ouvert", "ticket_id": "TCK-0001"} sur stdout. Quel filtre Cloud Logging la retrouve ?

L'agent

Le modèle vient d'une variable d'environnement (MODEL, défaut gemini-2.5-flash, validé sur Vertex AI en europe-west1 le 2026-10-15). Les réessais sur 429/503 sont confiés au client Gemini d'ADK : le quota Vertex AI est partagé par tout le projet.

files/agent.py
root_agent = Agent(
    name="assistant_interne",
    model=Gemini(model=MODEL, retry_options=types.HttpRetryOptions(attempts=5, initial_delay=2)),
    description="Assistant interne qui répond sur les politiques RH/IT et ouvre des tickets.",
    instruction=(
        "Tu es l'assistant interne d'une entreprise fictive. Tu réponds en français, en 3 phrases maximum. "
        "Pour toute question sur une règle interne, appelle d'abord chercher_politique et cite la règle. "
        "Si la personne signale un problème qu'elle ne peut pas résoudre seule, appelle ouvrir_ticket ..."
    ),
    tools=[chercher_politique, ouvrir_ticket],
)

Ce qui entre dans l'image

adk deploy cloud_run copie le dossier dans l'image en respectant .gcloudignore. On exclut le .env, les tests et les scripts :

files/.gcloudignore
.env
test_*.py
*.sh
__pycache__/

Tester sans clé

Les tests vérifient les outils (fonctions pures), le format du journal, et le client HTTP grâce à httpx.MockTransport, qui simule le serveur :

terminal
cd files
python -m venv .venv && . .venv/bin/activate
pip install -r requirements.txt pytest
pytest -q

Résultat obtenu

pytest (2026-10-15)
......                                                                   [100%]
6 passed, 1 warning in 0.47s

Exercice

Ajoutez un champ duree_ms au journal de chercher_politique (durée de la recherche), puis écrivez le filtre Cloud Logging qui retrouve les recherches sans résultat de l'assistant.

Voir la solution
files/agent.py (extrait modifié)
import time

def chercher_politique(sujet: str) -> dict:
    debut = time.perf_counter()
    cle = _normaliser(sujet)
    for nom, texte in POLITIQUES.items():
        if nom in cle or cle in nom:
            journal("politique_trouvee", sujet=nom, duree_ms=round((time.perf_counter() - debut) * 1000, 2))
            return {"trouve": True, "sujet": nom, "politique": texte}
    journal("politique_absente", sujet=cle, duree_ms=round((time.perf_counter() - debut) * 1000, 2))
    return {"trouve": False, "sujets_connus": sorted(POLITIQUES)}

Filtre (langage de requête de Cloud Logging) :

resource.type="cloud_run_revision"
resource.labels.service_name="lab-agent-cloud-run-iam"
jsonPayload.evenement="politique_absente"

Ces sujets sans réponse sont votre liste de politiques à rédiger en priorité.

Je sais faire

  • Organiser un dossier d'agent prêt pour adk deploy cloud_run
  • Écrire un journal structuré JSON filtrable dans Cloud Logging
  • Exclure .env et tests de l'image avec .gcloudignore
  • Tester outils et client sans clé ni projet
Siguiente