Agent de planification de contenu basé sur A2A et ADK

Le protocole A2A (Agent-to-Agent) est un standard ouvert conçu pour les agents IA. Il vise à permettre l'interopérabilité entre agents de différentes plateformes et stacks techniques, leur permettant de collaborer de manière transparente.

Stack technique

  • Python : version 3.10+
  • UV : gestionnaire de paquets Python
  • Google ADK : Kit de développement d'agents Google
  • SDK A2A : pour la communication agent-à-agent
  • Gemini 2.5 Flash : le modèle de langage sous-jacent
  • Google Search : outil de recherche intégré
  • Uvicorn : serveur ASGI

Prérequis

1. Préparation de l'environnement

Assurez-vous d'avoir installé les éléments suivants :

# Vérifier la version de Python (3.10+ nécessaire)
python --version

# Installer UV si ce n'est pas déjà fait
pip install uv

2. Clés API

Une clé API Google est requise pour utiliser le modèle Gemini et la recherche Google.

Structure du projet

content_planner_agent/
├── point_entree.py            # Script principal
├── executeur_agent.py         # Logique d'exécution de l'agent
├── definition_agent.py        # Configuration de l'agent
├── pyproject.toml             # Configuration du projet
├── environnement.exemple      # Exemple de fichier d'environnement
└── README.md                  # Documentation

Mise en route rapide

Étape 1 : Cloner et naviguer

git clone https://github.com/exemple/agent-planification-contenu.git
cd agent-planification-contenu

Étape 2 : Configurer les variables d'environnement

# Créer le fichier .env à partir de l'exemple
cp environnement.exemple .env
# Ajouter votre clé API dans le fichier .env
echo "CLE_API_GOOGLE=votre_cle_api_ici" >> .env

Étape 3 : Installer et démarrer

# Installer les dépendances et lancer l'agent
uv run .

Par défaut, l'agent sera accessible à l'adresse http://localhost:10001.

Étape 4 : Tester l'agent

Dans un nouveau terminal, utilisez l'outil client pour interagir :

# Se connecter à l'agent
uv run . --agent http://localhost:10001
# Envoyer une requête
"Crée un plan pour un post X concis et motivant sur l'apprentissage de Java"

Analyse du code

1. Point d'entrée (point_entree.py)

@click.command()
@click.option("--hote", default="localhost")
@click.option("--port", default=10001)
def lancer_serveur(hote, port):
    # Métadonnées de l'agent (fiche_agent)
    fiche_agent = MetadataAgent(
        nom='Agent Planificateur de Contenu',
        description=agent_principal.description,
        url=f'http://{hote}:{port}',
        version="1.0.0",
        modes_entree=["text/plain"],
        modes_sortie=["text/plain"],
        capacites=CapacitesAgent(flux=True),
        competences=[
            CompetenceAgent(
                id="planificateur_contenu",
                nom="Création de plans de contenu",
                description="Génère des structures pour un contenu à partir d'une description haute-level.",
                tags=["plan", "structure"],
                exemples=["Crée un plan pour un post X concis et motivant sur l'apprentissage de Java"],
            )
        ],
    )
    # ... lancement du serveur A2A

2. Définiiton de l'agent (definition_agent.py)

from google.adk.agents import Agent
from google.adk.tools import google_search

agent_principal = Agent(
    nom="agent_planificateur",
    modele="gemini-2.5-flash",
    description="Agent qui génère des plans de contenu détaillés et logiques.",
    instruction="Vous êtes un expert en planification de contenu. Créez des plans détaillés à partir de descriptions générales.",
    outils=[google_search],
)

3. Exécuteur d'agent (executeur_agent.py)

class ExecuteurADK(ExecuteurAgent):
    def __init__(self, agent, message_statut="Traitement en cours...", nom_artefact="reponse"):
        self.agent = agent
        self.executeur = Executeur(
            nom_application=agent.nom,
            agent=agent,
            service_artefact=ServiceArtefactMemoire(),
            service_session=ServiceSessionMemoire(),
            service_memoire=ServiceMemoireMemoire(),
        )

    async def executer(self, requete, contexte):
        # Logique d'exécution
        resultat = await self.executeur.run(requete)
        return Artefact(nom=self.nom_artefact, contenu=resultat)

Architecture du système

Phase de requête

Port personnalisé

uv run . --port=8080

Hôte personnalisé

uv run . --hote=0.0.0.0 --port=8080

Variables d'environnement

Fichier .env :

CLE_API_GOOGLE=votre_cle_ici
NIVEAU_LOG=INFO

Dépannage

Problèmes courants

  1. Clé API invalide : Vérifiez la valeur de CLE_API_GOOGLE.
  2. Port occupé : Utilisez l'option --port pour spécifier un autre port.
  3. Échec d'installation : Assurez-vous que uv est installé et essayez uv sync.

Extension et personnalisation

Ajouter des outils

from google.adk.tools import google_search, web_search

agent_principal = Agent(
    # ... autres configurations
    outils=[google_search, web_search],
)

Changer de modèle

agent_principal = Agent(
    modele="gemini-1.5-pro",
    # ... autres configurations
)

Personnaliser les instructions

agent_principal = Agent(
    instruction=(
        "Vous êtes un planificateur spécialisé dans la documentation technique. "
        "Créez des plans incluant des exemples de code et les meilleures pratiques."
    ),
)

Bonnes pratiques

  • Sécurité : Stockez les clés API dans les variables d'environnement et excluez les fichiers .env du versioning.
  • Performance : Choisissez un modèle adapté à vos besoins en termes de coût et de performance.
  • Gestion des erreurs : Implémentez une journalisation robuste et des messages d'erreur clairs.
  • Tests : Écrivez des tests unitaires et d'intégration pour valider les réponses de l'agent.

Étiquettes: google-adk a2a-protocol gemini Python Uvicorn

Publié le 24 juin à 20h20