Agent d'IA durable avec Gemini et Temporal

Ce tutoriel vous explique comment créer une boucle agentive de style ReAct qui utilise l' API Gemini pour le raisonnement et Temporal pour la durabilité. Le code source complet de ce tutoriel est disponible sur GitHub.

L'agent peut appeler des outils, par exemple pour rechercher des alertes météo ou géolocaliser une adresse IP, et il effectue une boucle jusqu'à ce qu'il dispose de suffisamment d'informations pour répondre.

La différence avec une démonstration d'agent classique réside dans la durabilité. Chaque appel de LLM, chaque appel d'outil et chaque étape de la boucle agentive sont conservés par Temporal. En cas de plantage du processus, de perte de réseau ou d'expiration d'une API, Temporal effectue automatiquement une nouvelle tentative et reprend à partir de la dernière étape terminée. L'historique des conversations n'est pas perdu et aucun appel d'outil n'est répété par erreur.

Architecture

L'architecture se compose de trois parties :

• Workflow : boucle agentive qui orchestre la logique d'exécution.
• Activités : unités de travail individuelles (appels de LLM, appels d'outils) que Temporal rend durables.
• Nœud de calcul : processus qui exécute les workflows et les activités.

Dans cet exemple, vous allez placer ces trois éléments dans un seul fichier (durable_agent_worker.py). Dans une implémentation réelle, vous les séparerez pour bénéficier de divers avantages en termes de déploiement et d'évolutivité. Vous placerez le code qui fournit un prompt à l'agent dans un deuxième fichier (start_workflow.py).

Prérequis

Pour suivre ce guide, vous aurez besoin des éléments suivants :

• Une clé API Gemini. Vous pouvez en créer une sans frais dans Google AI Studio.
• Python version 3.10 ou ultérieure.
• La CLI Temporal pour exécuter un serveur de développement local.

Configuration

Avant de commencer, assurez-vous qu'un serveur de développement Temporal est en cours d'exécution localement :

temporal server start-dev

Ensuite, installez les dépendances requises :

pip install temporalio google-genai httpx pydantic python-dotenv

Créez un fichier .env dans le répertoire de votre projet avec votre clé API Gemini. Vous pouvez obtenir une clé API depuis Google AI Studio.

echo "GOOGLE_API_KEY=your-api-key-here" > .env

Implémentation

Le reste de ce tutoriel décrit le fichier durable_agent_worker.py de haut en bas, en créant l'agent pièce par pièce. Créez le fichier et suivez les instructions.

Importations et configuration du bac à sable

Commencez par les importations qui doivent être définies à l'avance. Le bloc workflow.unsafe.imports_passed_through() indique au bac à sable de workflow de Temporal de laisser passer certains modules sans restriction. Cela est nécessaire, car plusieurs bibliothèques (notamment httpx, qui sous-classe urllib.request.Request) utilisent des modèles que le bac à sable bloquerait autrement.

from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    import pydantic_core  # noqa: F401
    import annotated_types  # noqa: F401

    import httpx
    from pydantic import BaseModel, Field
    from google import genai
    from google.genai import types

Instructions système

Définissez ensuite la personnalité de l'agent. Les instructions système indiquent au modèle comment se comporter. Cet agent est invité à répondre en haïkus lorsqu'aucun outil n'est nécessaire.

SYSTEM_INSTRUCTIONS = """
You are a helpful agent that can use tools to help the user.
You will be given an input from the user and a list of tools to use.
You may or may not need to use the tools to satisfy the user ask.
If no tools are needed, respond in haikus.
"""

Définitions d'outils

Définissez maintenant les outils que l'agent peut utiliser. Chaque outil est une fonction asynchrone avec une chaîne de documentation descriptive. Les outils qui acceptent des paramètres utilisent un modèle Pydantic comme argument unique. Il s'agit d'une bonne pratique Temporal qui permet de maintenir la stabilité des signatures d'activité lorsque vous ajoutez des champs facultatifs au fil du temps.

import json

NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

class GetWeatherAlertsRequest(BaseModel):
    """Request model for getting weather alerts."""

    state: str = Field(description="Two-letter US state code (e.g. CA, NY)")

async def get_weather_alerts(request: GetWeatherAlertsRequest) -> str:
    """Get weather alerts for a US state.

    Args:
        request: The request object containing:
            - state: Two-letter US state code (e.g. CA, NY)
    """
    headers = {"User-Agent": USER_AGENT, "Accept": "application/geo+json"}
    url = f"