Pourquoi Claude + Temporal ?

Dans le monde des agents IA, la gestion d'état est une arme à double tranchant. Les puissantes capacités de raisonnement de Claude brillent dans les tâches complexes à étapes multiples, mais maintenir l'état à travers les tentatives de reprise, les échecs ou le scaling introduit de la fragilité. Entrez Temporal : une plateforme open-source pour orchestrer des workflows durables qui garantit l'exécution même dans des environnements distribués et sujets aux pannes.

Ce guide compare les implémentations naïves de Claude (sans état, sujettes aux erreurs) aux workflows orchestrés par Temporal (fiables, scalables). Nous construirons de vrais exemples en utilisant Python, l'API Claude et le SDK Temporal — parfait pour les développeurs créant des agents IA de qualité production.

Le Problème : Agents IA avec état sans orchestration

Construire des agents IA avec état avec Claude implique typiquement :

• Stockage d'état personnalisé (par ex., Redis, PostgreSQL)
• Logique de retry manuelle
• Suivi en mémoire qui plante lors des redémarrages

Exemple naïf : Chercheur multi-étapes fragile

# Pseudo-code: Brittle stateful agent
import anthropic
client = anthropic.Anthropic()
state = {"research_topic": "AI ethics", "steps": [], "data": {}}

try:
    # Step 1: Initial research
    msg = client.messages.create(model="claude-3-5-sonnet-20241022", max_tokens=1000,
                                 messages=[{"role": "user", "content": f"Research {state['research_topic']}"}])
    state["steps"].append(msg.content[0].text)
    
    # Step 2: Analyze (assumes state persists)
    analysis = client.messages.create(model="claude-3-5-sonnet-20241022", max_tokens=500,
                                     messages=[{"role": "user", "content": f"Analyze: {state['steps'][0]}"}])
    state["data"]["analysis"] = analysis.content[0].text
    
    # Fails here? Lose everything.
except Exception:
    print("Workflow died—state lost.")

Points douloureux :

• Pas de retries automatiques
• L'état disparaît en cas de crash
• Scaling vers plusieurs workers ? Chaos
• Tâches longues (heures/jours) timeout

Temporal résout cela avec une exécution durable : les workflows persistent sous forme d'événements dans une base de données, rejouables en cas d'échec.

Qu'est-ce que Temporal ?

Temporal est un moteur de workflow pour microservices et applications IA :

• Durable : Survives crashes, auto-retries activities
• Avec état : Event sourcing intégré — pas de DB externe nécessaire
• Scalable : Gère des millions de workflows
• Langages : Python, Go, Java, TypeScript

Claude s'intègre parfaitement : Traitez les appels API Claude comme des activités (unités idempotentes), les workflows comme coordinateurs d'agents.

Tableau de comparaison : Naïf vs Temporal

Configuration rapide

1. Installer les dépendances

pip install temporalio anthropic

1. Démarrer le serveur Temporal (dev local) :

# Docker Compose (recommandé)
docker run --rm -p 7233:7233 temporalio/server:latest
# Ou full stack: https://docs.temporal.io/dev-guide/go#local-development

1. Clé API Anthropic

export ANTHROPIC_API_KEY=your_key_here

Construire votre premier workflow : Agent chercheur Claude

Orchestrons un chercheur durable : Étape 1) Brainstorming de sujets, Étape 2) Plongée approfondie, Étape 3) Résumé du rapport.

Activités (fonctions pures, appels Claude) :

from temporalio import activity
import anthropic
from typing import Dict

client = anthropic.Anthropic()

@activity.defn
def brainstorm_topics(topic: str) -> list[str]:
    msg = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=500,
        messages=[{"role": "user", "content": f"Brainstorm 3 subtopics for research on '{topic}'. Output as JSON list."}]
    )
    # Parse response (in prod, use structured outputs)
    return eval(msg.content[0].text)  # Simplified

@activity.defn
def research_subtopic(subtopic: str) -> str:
    msg = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=2000,
        messages=[{"role": "user", "content": f"Conduct in-depth research on: {subtopic}. Be factual, cite sources."}]
    )
    return msg.content[0].text

@activity.defn
def generate_report(researches: Dict[str, str]) -> str:
    content = "\
".join([f"{k}: {v}" for k,v in researches.items()])
    msg = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1000,
        messages=[{"role": "user", "content": f"Summarize this research into a concise report:\
{content}"}]
    )
    return msg.content[0].text

Workflow (orchestrateur) :

from temporalio import workflow
from temporalio.client import Client
from typing import list

@workflow.defn
class ResearcherWorkflow:
    async def run(self, topic: str) -> str:
        # Step 1
        subtopics = await workflow.execute_activity(
            brainstorm_topics,
            topic,
            start_to_close_timeout=timedelta(minutes=5),
            retry_policy=workflow.ActivityRetryPolicy(
                maximum_attempts=3,
                backoff_coefficient=2.0,
                maximum_interval=timedelta(minutes=1)
            )
        )
        
        # Step 2: Parallel research
        research_futures = [
            workflow.execute_activity(
                research_subtopic,
                subtopic,
                start_to_close_timeout=timedelta(minutes=10),
                retry_policy=workflow.ActivityRetryPolicy(maximum_attempts=5)
            ) for subtopic in subtopics
        ]
        researches = await workflow.wait_for_all(research_futures)
        
        # Step 3
        report = await workflow.execute_activity(
            generate_report,
            dict(zip(subtopics, researches)),
            start_to_close_timeout=timedelta(minutes=2)
        )
        return report

Lancer

import asyncio
from temporalio.client import Client
from datetime import timedelta

async def main():
    client = await Client.connect("localhost:7233")
    handle = await client.start_workflow(
        ResearcherWorkflow.run,
        "Quantum Computing",
        id="researcher-1",
        task_queue="claude-queue"
    )
    result = await handle.result()
    print(result)

asyncio.run(main())

Victoires clés :

• Tuez le worker ? Temporal rejoue depuis l'historique.
• Claude rate-limited ? Retries intelligents.
• Étapes parallèles scalent à travers les workers.

Avancé : Human-in-the-Loop & Logique conditionnelle

Étendez pour de vrais agents :

@workflow.defn
class AdvancedAgentWorkflow:
    async def run(self, query: str) -> str:
        initial_plan = await workflow.execute_activity(plan_query, query)
        
        # Human approval signal
        await workflow.wait_condition(lambda: self.human_approved)
        
        if self.needs_revision:
            revised = await workflow.execute_activity(revise_with_claude, initial_plan)
        
        return await workflow.execute_activity(finalize, revised or initial_plan)

Utilisez Signals/Queries pour un contrôle externe (par ex., approbation via API).

Gestion d'erreurs :

• Non-retryable : activity.heartbeat(details=error)
• Timeouts : Intégrés

Scaling en production

• Déployer les Workers :

# worker.py
from temporalio.worker import Worker

async def main():
    client = await Client.connect("temporal-host:7233")
    worker = Worker(client, task_queue="claude-queue", workflows=[ResearcherWorkflow], activities=[brainstorm_topics, ...])
    await worker.run()

Kubernetes/Helm charts disponibles.

• Optimisations Claude :

  • Utilisez claude-3-haiku-20240307 pour activités rapides
  • Tool use : Pass MCP servers as tools
  • Coût : Activités granulaires, cachez les étapes non-Claude

• Monitoring : UI Temporal (localhost:8080), metrics Prometheus

Comparaison de performance (Benchmark sur AWS) :

Bonnes pratiques pour Claude + Temporal

• Idempotence : Les activités doivent être pures ; utilisez les IDs de workflow pour déduplication
• Heartbeats : Pour appels Claude longs (>60s)
• Timeouts : Adaptez aux limites Claude (par ex., 60s pour Sonnet)
• Batching : Groupez les appels API dans les activités
• Secrets : Vars d'environnement ou attributs de recherche personnalisés Temporal
• Tests : temporalio.testing.WorkflowEnvironment

Conclusion

Temporal transforme Claude d'un raisonner puissant mais éphémère en une bête de production pour agents avec état. Commencez avec l'exemple du chercheur, scalez vers des workflows enterprise. Consultez les docs Temporal (temporal.io) et la référence API Claude pour des plongées plus profondes.

Prochaines étapes : Forkez le repo GitHub (lien en commentaires), déployez sur Temporal Cloud, construisez votre agent !

Nombre de mots : ~1450