Production

Limitation du débit du modèle

Chaque fournisseur de modèles applique des limites de débit. Dépassez-les et vos requêtes échouent avec 429 erreurs. Pire encore : dans un déploiement multi-réplicas, chaque réplique martèle indépendamment le API, provoquant une limitation agrégée invisible pour les processus individuels.

Le problème

Scénario : Vous déployez 10 réplicas de votre service d’agent. Chaque réplique pense disposer de 100 000 jetons/minute. Ensemble, ils envoient 1 million de jetons/minute, soit 10 fois votre quota réel. Le fournisseur limite agressivement. Les requêtes échouent de manière aléatoire sur toutes les répliques.

Sans limitation de tarif :

• Les requêtes échouent de manière imprévisible avec les 429
• Aucune visibilité sur la capacité restante
• Les tentatives aggravent la congestion
• L’expérience utilisateur se dégrade sous charge

Avec limitation de débit adaptative :

• Chaque réplique partage un budget coordonné
• Les demandes sont en file d’attente jusqu’à ce que la capacité soit disponible
• Le backoff se propage à travers le cluster
• Une dégradation gracieuse au lieu d’échecs

Aperçu

Le package features/model/middleware fournit un limiteur de débit adaptatif de type AIMD sous le client de modèle validé. Il demande au fournisseur le nombre exact de jetons d’entrée, bloque les appelants jusqu’à ce que cette capacité soit disponible et ajuste son budget de jetons d’entrée par minute en réponse à la limitation du fournisseur. Il n’estime jamais les jetons et ne mesure pas les quotas de sortie.

Stratégie AIMD

Le limiteur utilise une stratégie Augmentation Additive / Diminution Multiplicative (AIMD) :

Le nombre effectif de jetons par minute (TPM) est limité par :

• Minimum : 10 % du TPM initial (plancher pour éviter la famine)
• Maximum : Le plafond maxTPM configuré

Utilisation de base

Créez un seul limiteur par processus et enveloppez votre client modèle :

import (
    "context"

    "goa.design/goa-ai/features/model/middleware"
    "goa.design/goa-ai/runtime/agent/runtime"
)

func main() {
    ctx := context.Background()
    rt := runtime.New(runtimeStore) // stockage du runtime fourni par l'hôte

    // Vertex Gemini fournit le comptage exact exigé par le limiteur.
    modelClient, err := rt.NewVertexGeminiModelClient(ctx, runtime.VertexConfig{
        ProjectID:    "my-gcp-project",
        Location:     "us-central1",
        DefaultModel: "gemini-2.5-flash",
    })
    if err != nil {
        panic(err)
    }

    limiter := middleware.NewAdaptiveRateLimiter(ctx, nil, "", 60000, 120000)
    rateLimitedClient, err := limiter.Middleware()(modelClient)
    if err != nil {
        panic(err)
    }

    if err := rt.RegisterModel("default", rateLimitedClient); err != nil {
        panic(err)
    }
}

Limitation du débit en fonction du cluster

Pour les déploiements multi-processus, coordonnez la limitation du débit entre les instances à l’aide d’une carte répliquée Pulse :

import (
    "context"

    "goa.design/goa-ai/features/model/middleware"
    "goa.design/pulse/rmap"
)

func main() {
    ctx := context.Background()

    // Create a Pulse replicated map backed by Redis
    rm, err := rmap.Join(ctx, "rate-limits", redisClient)
    if err != nil {
        panic(err)
    }
    defer rm.Close()

    // Create cluster-aware limiter
    // All processes sharing this map and key coordinate their budgets
    limiter := middleware.NewAdaptiveRateLimiter(
        ctx,
        rm,
        "vertex:gemini",  // clé partagée pour cette famille de modèles
        60000,            // initial TPM
        120000,           // max TPM
    )

    rateLimitedClient, err := limiter.Middleware()(vertexClient)
    if err != nil {
        panic(err)
    }
}

Tant que les lectures et écritures de la carte répliquée réussissent :

• Le backoff se propage à l’échelle mondiale : lorsqu’un processus reçoit ErrRateLimited, tous les processus réduisent leur budget
• Le sondage est coordonné : les demandes réussies augmentent le budget partagé
• Réconciliation automatique : les processus surveillent les changements externes et mettent à jour leurs limiteurs locaux

Une carte nulle ou une clé vide crée volontairement un limiteur local au processus. Si l’état partagé ne peut pas être initialisé ou mis à jour, les appels de modèle continuent avec le budget adaptatif local jusqu’à ce qu’un événement ultérieur de la carte le resynchronise. Surveillez Redis lorsque la coordination entre processus est requise.

Comptage exact des jetons

Le limiteur appelle CountTokens sur le client enveloppé avant de réserver de la capacité. Il exige Exact=true et n’effectue aucune estimation.

Vertex Gemini fournit une opération native. Bedrock renvoie model.ErrTokenCountingUnsupported lorsqu’il ne peut pas compter exactement, et OpenAI ne fournit aucun compteur natif. Il est possible d’envelopper un client sans support, mais son premier appel Complete ou Stream renvoie cette erreur avant l’inférence.

Intégration avec le runtime

Câblez des clients à débit limité dans le runtime Goa-AI :

vertexLimiter := middleware.NewAdaptiveRateLimiter(ctx, nil, "", 60000, 120000)
limitedVertex, err := vertexLimiter.Middleware()(vertexClient)
if err != nil {
    panic(err)
}

rt := runtime.New(runtimeStore, runtime.WithEngine(temporalEng))
if err := rt.RegisterModel("gemini", limitedVertex); err != nil {
    panic(err)
}

Que se passe-t-il sous charge

Paramètres de réglage

Exemple : Avec initialTPM=60000, maxTPM=120000 :

• Plancher : 6 000 TPM
• Récupération : +3 000 TPM par lot réussi
• Backoff : réduire de moitié le TPM actuel sur 429

Surveillance

Mesurez la latence des appels de modèle et les erreurs terminales model.ErrRateLimited dans la télémétrie du client enveloppé. Surveillez aussi Redis lorsque vous utilisez une carte répliquée.

Meilleures pratiques

• Un limiteur par modèle/fournisseur : créez des limiteurs distincts pour différents modèles afin d’isoler leurs budgets
• Définissez un TPM initial réaliste : commencez par la limite de débit documentée par votre fournisseur ou une estimation prudente
• Utilisez la limitation adaptée aux clusters en production : coordonnez les réplicas pour éviter la limitation globale
• Surveillez les événements de backoff : enregistrez ou émettez des métriques lorsque des backoffs se produisent pour détecter une limitation soutenue.
• Définissez maxTPM au-dessus de la valeur initiale : laissez une marge pour sonder lorsque le trafic est inférieur au quota

Observabilité GenAI avec OpenTelemetry

Lorsqu’un traceur est configuré, Goa-AI émet des spans indépendants du fournisseur qui suivent les conventions sémantiques GenAI d’OpenTelemetry pour les opérations d’agent exécutées par le planner :

• les appels de modèle utilisent gen_ai.operation.name="chat" et des noms de span comme chat {model}
• les fins d’exécution d’outils utilisent gen_ai.operation.name="execute_tool" et des noms de span comme execute_tool {tool_name}
• la délégation agent-comme-outil utilise gen_ai.operation.name="invoke_agent" et des noms de span comme invoke_agent {agent_name}

Ces spans incluent gen_ai.conversation.id, gen_ai.agent.id, gen_ai.agent.name, gen_ai.request.model, gen_ai.response.model, l’utilisation des jetons, les raisons d’arrêt, les identifiants d’outil et le temps jusqu’au premier fragment de streaming lorsque ces valeurs sont disponibles. Le runtime enregistre par défaut les identifiants, les compteurs, les durées et les erreurs ; le texte des prompts, l’historique de conversation, les arguments d’outil et les résultats d’outil restent une politique de l’application et ne sont pas attachés automatiquement.

Cette approche garde la télémétrie open source de Goa-AI portable entre backends OpenTelemetry tout en donnant aux systèmes de production assez de structure pour regrouper une conversation, comparer la latence et l’utilisation des jetons par modèle, et inspecter les chaînes d’outils multi-agents.

Remplacements d’invite avec le magasin Mongo

La gestion des invites de production utilise généralement :

• spécifications d’invite de base enregistrées dans runtime.PromptRegistry, et
• les enregistrements de remplacement limités ont persisté dans Mongo via features/prompt/mongo.

Câblage

import (
    promptmongo "goa.design/goa-ai/features/prompt/mongo"
    clientmongo "goa.design/goa-ai/features/prompt/mongo/clients/mongo"
    "goa.design/goa-ai/runtime/agent/runtime"
)

promptClient, err := clientmongo.New(clientmongo.Options{
    Client:     mongoClient,
    Database:   "assistant",
    Collection: "prompt_overrides", // optional (default is prompt_overrides)
})
if err != nil {
    panic(err)
}

promptStore, err := promptmongo.NewStore(promptClient)
if err != nil {
    panic(err)
}

rt := runtime.New(
    runtimeStore,
    runtime.WithEngine(temporalEng),
    runtime.WithPromptStore(promptStore),
)

Remplacer la résolution et le déploiement

La priorité de remplacement est déterministe :

1. Portée session
2. Portée facility
3. Portée org
4. portée mondiale
5. spécification de base (quand aucun remplacement n’existe)

Stratégie de déploiement recommandée :

• Enregistrez d’abord les nouvelles spécifications de base.
• Déployez les remplacements à grande échelle (org), puis limitez-les à facility/session pour les Canaries.
• Suivez les versions efficaces via les événements prompt_rendered et model.Request.PromptRefs.
• Revenez en arrière en écrivant un remplacement plus récent dans la même portée (ou en supprimant les remplacements spécifiques à la portée pour revenir en arrière).

Configuration Temporal

Cette section couvre la configuration de Temporal pour des flux de travail d’agent durables dans les environnements de production.

Aperçu

Temporal offre une exécution durable pour vos agents Goa-AI. Les exécutions d’agents deviennent des workflows Temporal avec un historique basé sur les événements. Les appels d’outils deviennent des activités avec des tentatives configurables. Chaque transition d’état est persistante. Un travailleur redémarré relit l’historique et reprend exactement là où il s’était arrêté.

Comment fonctionne la durabilité

Exemple concret : Votre agent appelle un LLM, qui renvoie 3 appels d’outil. Deux outils complets. Le service du troisième outil plante.

• ❌ Sans Temporal : L’exécution entière échoue. Vous réexécutez l’inférence ($$$) et réexécutez les deux outils réussis.
• ✅ Avec Temporal : Seul l’outil en panne réessaye. Le flux de travail est relu à partir de l’historique : pas de nouvel appel LLM, pas de réexécution des outils terminés. Coût : une nouvelle tentative, pas un redémarrage complet.

Ce qui survit aux échecs

Installation

Option 1 : Docker (développement)

Un mot d’ordre pour le développement local :

docker run --rm -d --name temporal-dev -p 7233:7233 temporalio/auto-setup:latest

Option 2 : Temporalite (Développement)

go install go.temporal.io/server/cmd/temporalite@latest
temporalite start

Option 3 : Cloud Temporal (Production)

Inscrivez-vous sur temporal.io et configurez votre client avec des informations d’identification cloud.

Option 4 : auto-hébergé (production)

Déployez Temporal à l’aide de Docker Compose ou Kubernetes. Consultez la documentation Temporal pour les guides de déploiement.

Configuration d’exécution

Goa-AI résume le backend d’exécution derrière l’interface Engine. Échangez les moteurs sans changer le code de l’agent :

Moteur en mémoire (développement) :

// Default: no external dependencies
rt := runtime.New(storageinmem.New())

Moteur Temporal (production) :

import (
    runtimeTemporal "goa.design/goa-ai/runtime/agent/engine/temporal"
    temporalclient "go.temporal.io/sdk/client"
    "go.temporal.io/sdk/worker"
    "go.temporal.io/sdk/workflow"

    // Your generated tool specs aggregate.
    // The generated package exposes: func Spec(tools.Ident) (*tools.ToolSpec, bool)
    specs "<module>/gen/<service>/agents/<agent>/specs"
)

const releaseBuildID = "git-sha-or-image-digest"

temporalEng, err := runtimeTemporal.NewWorker(runtimeTemporal.Options{
    ClientOptions: &temporalclient.Options{
        HostPort:  "127.0.0.1:7233",
        Namespace: "default",
        // Required: enforce goa-ai's workflow boundary contract.
        // Tool results and server-data cross workflow boundaries as canonical JSON bytes
        // (for example api.ToolEvent payloads), not decoded planner.ToolResult values.
        DataConverter: runtimeTemporal.NewAgentDataConverter(specs.Spec),
    },
    WorkerOptions: runtimeTemporal.WorkerOptions{
        TaskQueue: "orchestrator.chat",
        Options: worker.Options{
            DeploymentOptions: worker.DeploymentOptions{
                UseVersioning: true,
                Version: worker.WorkerDeploymentVersion{
                    DeploymentName: "assistant",
                    BuildID:        releaseBuildID,
                },
                DefaultVersioningBehavior: workflow.VersioningBehaviorPinned,
            },
        },
    },
})
if err != nil {
    panic(err)
}
defer temporalEng.Close()

rt := runtime.New(runtimeStore, runtime.WithEngine(temporalEng))

Propriété du stockage du runtime

runtime.New exige un seul storage.Store. En production, un service unique de l’application doit posséder la base de données qui contient l’état des sessions, les métadonnées des exécutions, les points de reprise et les enregistrements immuables. Les workers d’agents appellent ce propriétaire par une API typée ; ils n’ouvrent pas de connexions distinctes vers ses collections.

Les données du produit restent la propriété du service produit. Par exemple, un service de chat conserve ses transcriptions, évaluations et champs de recherche même si un autre service possède le stockage du runtime Goa-AI.

Mémoire et sessions définit le contrat complet du stockage : les changements de cycle de vie et leurs enregistrements sont sauvegardés ensemble, les nouvelles tentatives identiques renvoient le résultat accepté, tout nouvel enfant exige un parent actif, le JSON des événements enregistrés est strict et l’origine de l’annulation est préservée. Runtime explique comment les hôtes valident et livrent les événements finaux après la fermeture de l’historique du moteur. Conservez ces règles dans le stockage du runtime qui en est responsable au lieu de les reproduire dans chaque worker.

Le remplacement de session.Store et runlog.Store est une modification coordonnée du stockage. Avant que le nouveau runtime écrive, les métadonnées, points de reprise et enregistrements existants doivent respecter le contrat intégré de storage.Store. Déployez ensemble le propriétaire du stockage et tous les workers qui l’appellent. Les anciens writers des stockages séparés et les nouveaux writers du stockage intégré ne doivent pas se chevaucher.

Effectuez la conversion hors trafic avec un programme de migration jetable dans un job ou pod temporaire de la base de données. L’application hôte possède ce programme ; il ne fait pas partie de la version normale et doit être supprimé après vérification.

1. Créez et vérifiez une sauvegarde, puis confirmez l’absence de writers.
2. Exécutez la migration en mode vérification et corrigez chaque rejet.
3. Appliquez la conversion et vérifiez schéma, index, sessions, métadonnées, checkpoints v7 et enregistrements immuables.
4. Déployez ensemble le propriétaire et tous les workers, puis supprimez le programme.

Après le début de la conversion, le rollback restaure la sauvegarde complète. N’exécutez pas d’anciens writers sur une base partiellement convertie.

Contrat des charges utiles Temporal

La taille encodée cumulée des arguments d’un workflow ou d’une activité est limitée à engine.MaxPayloadBytes (1 Mio). Avant l’encodage, le convertisseur refuse également un graphe de valeurs de plus de 64 niveaux ou de plus de 100 000 valeurs visitées. Il ne tronque jamais les données trop grandes.

planner.ToolResult est une valeur interne au processus et ne peut pas traverser une frontière Temporal. Les workflows transportent à la place un api.ToolEvent contenant les octets JSON canoniques. Si un résultat d’outil valide peut dépasser 1 Mio, son exécuteur doit l’enregistrer dans le stockage de l’application et renvoyer une référence typée ; le runtime ne remplace pas silencieusement le résultat.

Synchronisation et tentatives d’activité

Utilisez le DSL pour les budgets d’exécution sémantiques : combien de temps l’exécution entière peut prendre, combien de temps un la tentative du planificateur peut s’exécuter et la durée pendant laquelle une tentative d’outil peut s’exécuter.

Goa-AI démarre chaque workflow d’agent une seule fois. Il ne redémarre pas le workflow complet après un échec, car une tentative précédente peut déjà avoir appelé des outils ou enregistré un résultat final. La durabilité vient de la relecture de l’historique et des nouvelles tentatives de chaque activité de planification, d’outil, de hook et de stockage.

Agent("operator", "Production operations agent", func() {
    RunPolicy(func() {
        DefaultCaps(MaxToolCalls(20), MaxRecoveryTurns(3))
        Timing(func() {
            Budget("5m")
            Plan("45s")
            Tools("90s")
        })
    })
})

L’adaptateur Temporal possède des mécanismes de moteur de flux de travail tels que l’attente en file d’attente et délais d’attente de vivacité. Configurez-les sur le moteur, pas dans le DSL :

temporalEng, err := runtimeTemporal.NewWorker(runtimeTemporal.Options{
    ClientOptions: &client.Options{
        HostPort:  "127.0.0.1:7233",
        Namespace: "default",
    },
    WorkerOptions: runtimeTemporal.WorkerOptions{
        TaskQueue: "orchestrator.chat",
    },
    ActivityDefaults: runtimeTemporal.ActivityDefaults{
        Planner: runtimeTemporal.ActivityTimeoutDefaults{
            QueueWaitTimeout: 30 * time.Second,
            LivenessTimeout:  20 * time.Second,
        },
        Tool: runtimeTemporal.ActivityTimeoutDefaults{
            QueueWaitTimeout: 2 * time.Minute,
            LivenessTimeout:  20 * time.Second,
        },
    },
})

Les activités de plan/CV généré, d’exécution d’outil et de publication de hook utilisent une nouvelle tentative des politiques qui ne sont sûres que lorsque les tentatives sont logiquement idempotentes. Événements de crochet transporter des clés d’événement stables et les exécutions d’outils doivent persister ou rejouer canoniquement résultats par ToolCallID plutôt que de répéter des effets secondaires irréversibles.

Configuration du travailleur

Les travailleurs interrogent les files d’attente de tâches et exécutent des flux de travail/activités. Les travailleurs sont automatiquement démarrés pour chaque agent enregistré : aucune configuration manuelle n’est nécessaire dans la plupart des cas.

Déploiements transparents

La durabilité Temporal et les versions transparentes sont deux garanties distinctes. Temporal conserve l’historique du workflow ; le déploiement doit maintenir le code de worker compatible et chaque service requis tant que cet historique reste actif.

La configuration ci-dessus inscrit le worker dans le versionnement des Worker Deployments de Temporal. releaseBuildID doit désigner un binaire ou une image immuable. Ne réutilisez jamais un ID de build pour un code différent et n’utilisez pas un tag mutable tel que latest.

Publiez une version de worker dans cet ordre :

1. Démarrez les nouveaux workers à côté de toutes les versions conservées.
2. Attendez que le nouveau processus soit prêt et correctement enregistré dans Temporal.
3. Rendez la nouvelle Worker Deployment Version courante. Temporal lui affecte les nouveaux workflows, tandis que les workflows existants restent liés à leur version d’origine.
4. Si le même processus sert aussi une API, n’envoyez le trafic normal qu’au build courant prêt. Gardez les anciens pods disponibles pour Temporal sans leur envoyer de nouvelles requêtes API.
5. Ne retirez une ancienne version qu’après que Temporal l’a déclarée drainée. L’arrêt d’un pod ne prouve pas qu’aucun workflow n’a encore besoin de son code.

Chaque entrée utilisateur acceptée démarre un workflow Goa-AI de premier niveau. Lorsqu’une entrée humaine ou externe est requise, Goa-AI termine ce workflow et enregistre un point de reprise privé sous l’ID de l’exécution achevée. La réponse acceptée démarre un nouveau workflow sur la version courante. Cette version doit donc rester compatible avec la version du point de reprise, les codecs générés et les noms d’outils requis. Le versionnement des workers ne traduit pas des valeurs enregistrées incompatibles.

Le reste de l’application doit rester disponible pendant le chevauchement :

• chaque service en aval conserve au moins un endpoint prêt ; utilisez un remplacement progressif soumis à la readiness, pas Recreate ;
• les API en aval acceptent les appels des workers conservés et courants ;
• les migrations de base de données prennent en charge les deux versions jusqu’au drainage de l’ancienne ;
• si un processus sert l’API et Temporal, le sélecteur de trafic distingue le build courant de l’accès de Temporal aux anciens workers.

Fournisseurs d’outils adossés au registre

Les fournisseurs d’outils Goa-AI prennent également en charge un remplacement progressif soumis à la readiness. Des réplicas qui partagent le même schéma généré et la même révision d’admission peuvent se chevaucher. Si l’une de ces valeurs change, le nouveau fournisseur reste vivant et retente son enregistrement pendant que l’ancienne admission reste autoritaire. L’ancien cesse d’accepter des appels, termine le travail admis et libère son bail avant que le nouveau puisse exécuter ; deux contrats incompatibles ne servent donc jamais le même ensemble d’outils simultanément.

Un CallTool valide qui trouve un ensemble actif sans fournisseur sain attend dans son délai d’exécution existant. La publication de la requête vérifie le fournisseur sélectionné dans la même opération Redis que l’ajout de l’appel. Si l’ancien fournisseur commence à se drainer, l’appel non publié sélectionne son remplaçant et réessaie sans prolonger son délai. L’affectation ne devient définitive qu’après la publication. L’annulation de l’appelant ne termine que cette tentative de transport ; une nouvelle tentative identique peut reprendre l’appel non publié. L’expiration enregistre la décision durable normale call_not_admitted.

Les clients, serveurs et fournisseurs du registre doivent conserver un protocole wire compatible pendant la version. Pour une modification incompatible de cette enveloppe, publiez d’abord un code qui accepte les deux formes ; le registre ne négocie pas les versions du protocole.

Changements de contrats générés

Lorsque les agents, packages de complétion ou charges utiles persistées changent de manière incompatible, n’appliquez pas la procédure de versions mixtes. Régénérez tous les agents et complétions, drainez ou arrêtez le travail concerné, puis déployez ensemble le runtime, les workers et les appelants. Goa-AI n’offre aucun mode de double lecture pour ses contrats générés.

Le runtime accepte uniquement le schéma exact goa-ai.run-suspension.v7. Les planificateurs qui attendent des questions, une clarification ou des outils externes conservent le ModelToolCallID du fournisseur ; le workflow attribue un ToolCallID d’exécution distinct avant d’enregistrer la suspension. Aucun autre schéma de suspension ne peut reprendre. Un changement futur doit inventorier et retirer le travail enregistré incompatible avant la version coordonnée ; n’ajoutez ni double lecteur ni déduction de champs.

Vérification d’une version

Une version n’est transparente que si :

• un workflow démarré avant la promotion se termine sur son build d’origine ;
• un nouveau workflow démarre et se termine sur le build courant ;
• une demande d’entrée externe créée avant la promotion continue correctement dans un nouveau workflow après la promotion ;
• le trafic API atteint seulement le build courant prêt ;
• les anciens workers restent prêts jusqu’à ce que Temporal les déclare drainés ;
• chaque service en aval conserve un endpoint prêt pendant le remplacement ;
• aucune nouvelle panne de workflow, aucun redémarrage de conteneur et aucune interruption de readiness n’apparaît pendant l’observation.

Le contrat d’un workflow par tour et d’identité des événements entre workflows est décrit dans Entrées externes et continuations de workflow.

Meilleures pratiques

• Utilisez des espaces de noms distincts pour différents environnements (dev, staging, prod)
• Configurer les politiques de nouvelle tentative par ensemble d’outils en fonction des caractéristiques de fiabilité
• Surveiller l’exécution du flux de travail à l’aide du UI et des outils d’observabilité de Temporal
• Définissez des délais d’attente appropriés pour les activités : équilibrez la fiabilité et la détection des blocages
• Utilisez Temporal Cloud pour la production afin d’éviter la charge opérationnelle

Streaming UI

Cette section montre comment diffuser les événements d’agent sur UIs en temps réel à l’aide de l’infrastructure de streaming de Goa-AI.

Aperçu

Goa-AI publie des flux appartenant à la session d’événements typés qui peuvent être transmis à UIs via :

• Événements envoyés par le serveur (SSE)
• WebSockets
• Bus de messages (Pulse, Redis Streams, etc.)

Tous les événements visibles dans le flux pour une session sont ajoutés à un seul flux : session/<session_id>. Chaque événement comporte à la fois run_id et session_id afin que UIs puisse regrouper les événements en couloirs/cartes par course. Les exécutions d’agents imbriquées sont liées via des événements child_run_linked. UIs ferme SSE/WebSocket de manière déterministe lorsqu’il observe run_stream_end pour l’exécution active.

Interface du récepteur de flux

Implémentez l’interface stream.Sink :

type Sink interface {
    Send(ctx context.Context, event stream.Event) error
    Close(ctx context.Context) error
}

Types d’événements

Le package stream définit des types d’événements concrets qui implémentent stream.Event. Les plus courants pour UIs sont :

Les transports utilisent généralement un commutateur de type sur stream.Event pour la sécurité au moment de la compilation :

switch e := evt.(type) {
case stream.AssistantReply:
    // e.Data.Text
case stream.PlannerThought:
    // e.Data.Note or structured thinking fields
case stream.ToolStart:
    // e.Data.ToolCallID, e.Data.ToolName, e.Data.Payload
case stream.ToolEnd:
    // e.Data.Result, e.Data.Error, e.Data.ResultPreview
case stream.ChildRunLinked:
    // e.Data.ToolName, e.Data.ToolCallID, e.Data.ChildRunID, e.Data.ChildAgentID
case stream.RunStreamEnd:
    // run has no more stream-visible events
}

Exemple : Évier SSE

type SSESink struct {
    w http.ResponseWriter
}

func (s *SSESink) Send(ctx context.Context, event stream.Event) error {
    switch e := event.(type) {
    case stream.AssistantReply:
        fmt.Fprintf(s.w, "data: assistant: %s\n\n", e.Data.Text)
    case stream.PlannerThought:
        if e.Data.Note != "" {
            fmt.Fprintf(s.w, "data: thinking: %s\n\n", e.Data.Note)
        }
    case stream.ToolStart:
        fmt.Fprintf(s.w, "data: tool_start: %s\n\n", e.Data.ToolName)
    case stream.ToolEnd:
        fmt.Fprintf(s.w, "data: tool_end: %s status=%v\n\n",
            e.Data.ToolName, e.Data.Error == nil)
    case stream.ChildRunLinked:
        fmt.Fprintf(s.w, "data: child_run_linked: %s child=%s\n\n",
            e.Data.ToolName, e.Data.ChildRunID)
    case stream.RunStreamEnd:
        fmt.Fprintf(s.w, "data: run_stream_end: %s\n\n", e.RunID())
    }
    s.w.(http.Flusher).Flush()
    return nil
}

func (s *SSESink) Close(ctx context.Context) error {
    return nil
}

Abonnement au flux de session (Pulse)

En production, UIs consomme le flux de session (session/<session_id>) à partir d’un bus partagé (Pulse / Redis Streams) et filtre par run_id. Fermez SSE/WebSocket lorsque vous observez run_stream_end pour l’analyse active.

Puits de flux global

Pour diffuser toutes les exécutions via un récepteur global (par exemple, Pulse), configurez le runtime avec un récepteur de flux :

rt := runtime.New(
    runtimeStore,
    runtime.WithStream(pulseSink), // or your custom sink
)

Le runtime installe un stream.Subscriber par défaut qui :

• mappe les événements de hook aux valeurs stream.Event
• utilise le StreamProfile par défaut, qui émet les réponses de l’assistant, les réflexions du planificateur, le démarrage/la mise à jour/la fin de l’outil, les attentes, l’utilisation, le flux de travail, les liens child_run_linked et le marqueur du terminal run_stream_end.

Profils de flux

Tous les consommateurs n’ont pas besoin de chaque événement. Les profils de diffusion filtrent les événements pour différents publics, réduisant ainsi le bruit et la bande passante pour des cas d’utilisation spécifiques.

Utilisation des profils intégrés :

// User-facing chat: replies, tool status, completion
profile := stream.UserChatProfile()

// Debug view: everything including planner thoughts
profile := stream.AgentDebugProfile()

// Metrics pipeline: just usage and workflow events
profile := stream.MetricsProfile()

sub, _ := stream.NewSubscriberWithProfile(sink, profile)

Profils personnalisés :

// Fine-grained control over which events to emit
profile := stream.StreamProfile{
    Assistant:  true,
    Thoughts:   false,  // Skip planner thinking
    ToolStart:  true,
    ToolUpdate: true,
    ToolEnd:    true,
    Usage:      false,  // Skip usage events
    Workflow:   true,
    ChildRuns:  true,   // Include parent tool → child run links
}

sub, _ := stream.NewSubscriberWithProfile(sink, profile)

Les profils personnalisés sont utiles lorsque :

• Vous avez besoin d’événements spécifiques pour un consommateur spécialisé (par exemple, suivi des progrès)
• Vous souhaitez réduire la taille de la charge utile pour les clients mobiles
• Vous créez des pipelines d’analyse qui n’ont besoin que de certains événements

Avancé : Pulse et ponts de flux

Pour les configurations de production, vous souhaitez souvent :

• publier des événements sur un bus partagé (par exemple, Pulse)
• utiliser un flux appartenant à la session sur ce bus (session/<session_id>)

Goa-AI fournit :

• features/stream/pulse – une implémentation stream.Sink soutenue par Pulse
• runtime/agent/stream/bridge – aides pour câbler le bus à crochet à n’importe quel évier

Câblage typique :

pulseClient := pulse.NewClient(redisClient)
s, err := pulseSink.NewSink(pulseSink.Options{
    Client: pulseClient,
    // Optional: override stream naming (defaults to `session/<SessionID>`).
    StreamID: func(ev stream.Event) (string, error) {
        if ev.SessionID() == "" {
            return "", errors.New("missing session id")
        }
        return fmt.Sprintf("session/%s", ev.SessionID()), nil
    },
})
if err != nil { log.Fatal(err) }

rt := runtime.New(
    runtimeStore,
    runtime.WithEngine(eng),
    runtime.WithStream(s),
)

Rappels système

Les modèles dérivent. Ils oublient les instructions. Ils ignorent le contexte qui était clair il y a 10 tours. Lorsque votre agent exécute des tâches de longue durée, vous avez besoin d’un moyen d’injecter des guides dynamiques et contextuels sans polluer la conversation de l’utilisateur.

Le problème

Scénario : Votre agent gère une liste de tâches. Après 20 tours, l’utilisateur demande « quelle est la prochaine étape ? » mais le modèle a dérivé : il ne se souvient pas qu’il y a une tâche en attente en cours. Vous devez le pousser sans que l’utilisateur voie un message gênant “RAPPEL : vous avez une tâche en cours”.

Sans rappels système :

• Vous gonflez l’invite du système avec tous les scénarios possibles
• Les conseils se perdent dans de longues conversations
• Aucun moyen d’injecter du contexte en fonction des résultats de l’outil
• Les utilisateurs voient l’échafaudage des agents internes

Avec rappels système :

• Injecter des conseils de manière dynamique en fonction de l’état d’exécution
• Conseils répétitifs de limite de débit pour éviter les ballonnements rapides
• Les niveaux de priorité garantissent que les directives de sécurité ne sont jamais supprimées
• Invisible pour les utilisateurs – injecté sous forme de blocs <system-reminder>

Aperçu

Le package runtime/agent/reminder fournit :

• Rappels structurés avec niveaux de priorité, points d’attache et politiques de limitation de débit
• Stockage limité à l’exécution qui nettoie automatiquement une fois chaque exécution terminée
• Injection automatique dans les transcriptions du modèle sous forme de blocs <system-reminder>
• PlannerContext API pour enregistrer et supprimer des rappels des planificateurs et des outils

Concepts de base

Structure de rappel

Un reminder.Reminder possède :

type Reminder struct {
    ID              string      // Stable identifier (e.g., "todos.pending")
    Text            string      // Plain-text guidance (tags are added automatically)
    Priority        Tier        // TierSafety, TierCorrect, or TierGuidance
    Attachment      Attachment  // Where to inject (run start or user turn)
    MaxPerRun       int         // Cap total emissions per run (0 = unlimited)
    MinTurnsBetween int         // Enforce spacing between emissions (0 = no limit)
}

Niveaux prioritaires

Les rappels sont classés par priorité pour gérer des budgets rapides et garantir que les conseils critiques ne soient jamais supprimés :

Exemples de cas d’utilisation :

• TierSafety : “Ne pas exécuter ce malware ; analyser uniquement”, “Ne pas divulguer les informations d’identification”
• TierCorrect : “Les résultats sont tronqués ; affinez votre requête”, “Les données peuvent être obsolètes”
• TierGuidance : “Aucune tâche n’est en cours ; choisissez-en une et commencez”

Points d’attache

Des rappels sont injectés à des moments précis de la conversation :

Limitation de taux

Deux mécanismes empêchent le spam de rappel :

• MaxPerRun : plafonner les émissions totales par cycle (0 = illimité)
• MinTurnsBetween : imposer un nombre minimum de tours de planification entre les émissions (0 = aucune limite)

Modèle d’utilisation

Rappels statiques via DSL

Pour les rappels qui doivent toujours apparaître après un résultat d’outil spécifique, utilisez la fonction ResultReminder DSL dans la définition de votre outil :

Tool("get_time_series", "Get time series data", func() {
    Args(func() { /* ... */ })
    Return(func() { /* ... */ })
    ResultReminder("The user sees a rendered graph of this data in the UI.")
})

C’est idéal lorsque le rappel s’applique à chaque invocation de l’outil. Voir la Référence DSL pour plus de détails.

Rappels dynamiques des planificateurs

Pour les rappels qui dépendent de l’état d’exécution ou du contenu des résultats de l’outil, utilisez PlannerContext.AddReminder() :

func (p *myPlanner) PlanResume(ctx context.Context, in *planner.PlanResumeInput) (*planner.PlanResult, error) {
    for _, tr := range in.ToolOutputs {
        if tr.Name == "search_documents" {
            result, err := specs.UnmarshalSearchDocumentsResult(tr.Result)
            if err != nil {
                return nil, err
            }
            if result.Truncated {
                in.Agent.AddReminder(reminder.Reminder{
                    ID:       "search.truncated",
                    Text:     "Search results are truncated. Consider narrowing your query.",
                    Priority: reminder.TierCorrect,
                    Attachment: reminder.Attachment{
                        Kind: reminder.AttachmentUserTurn,
                    },
                    MaxPerRun:       3,
                    MinTurnsBetween: 2,
                })
            }
        }
    }
    // Continue with planning...
}

Suppression des rappels

Utilisez RemoveReminder() lorsqu’une condition préalable n’est plus remplie :

if allTodosCompleted {
    in.Agent.RemoveReminder("todos.no_active")
}

Préservation des compteurs de limite de débit

AddReminder() préserve les compteurs d’émissions lors de la mise à jour d’un rappel existant par ID. Si vous devez modifier le contenu du rappel tout en conservant les limites de débit :

in.Agent.AddReminder(reminder.Reminder{
    ID:              "todos.pending",
    Text:            buildUpdatedText(snap),
    Priority:        reminder.TierGuidance,
    Attachment:      reminder.Attachment{Kind: reminder.AttachmentUserTurn},
    MinTurnsBetween: 3,
})

Anti-modèle : n’appelez pas RemoveReminder() suivi de AddReminder() pour le même ID : cela réinitialise les compteurs et contourne MinTurnsBetween.

Injection et formatage

Marquage automatique

Le moteur d’exécution encapsule automatiquement le texte de rappel dans les balises <system-reminder> lors de l’injection dans les transcriptions :

// You provide plain text:
Text: "Results are truncated. Narrow your query."

// Runtime injects:
<system-reminder>Results are truncated. Narrow your query.</system-reminder>

Expliquer les rappels aux modèles

Incluez reminder.DefaultExplanation dans l’invite de votre système afin que les modèles sachent comment interpréter les blocs <system-reminder> :

const systemPrompt = `
You are a helpful assistant.

` + reminder.DefaultExplanation + `

Follow all instructions carefully.
`

Exemple complet

func (p *myPlanner) PlanResume(ctx context.Context, in *planner.PlanResumeInput) (*planner.PlanResult, error) {
    for _, tr := range in.ToolOutputs {
        if tr.Name == "todos.update_todos" {
            snap, err := specs.UnmarshalUpdateTodosResult(tr.Result)
            if err != nil {
                return nil, err
            }
            
            var rem *reminder.Reminder
            if len(snap.Items) == 0 {
                in.Agent.RemoveReminder("todos.no_active")
                in.Agent.RemoveReminder("todos.all_completed")
            } else if hasCompletedAll(snap) {
                rem = &reminder.Reminder{
                    ID:       "todos.all_completed",
                    Text:     "All todos are completed. Provide your final response now.",
                    Priority: reminder.TierGuidance,
                    Attachment: reminder.Attachment{
                        Kind: reminder.AttachmentUserTurn,
                    },
                    MaxPerRun: 1,
                }
            } else if hasPendingNoActive(snap) {
                rem = &reminder.Reminder{
                    ID:       "todos.no_active",
                    Text:     buildTodosNudge(snap),
                    Priority: reminder.TierGuidance,
                    Attachment: reminder.Attachment{
                        Kind: reminder.AttachmentUserTurn,
                    },
                    MinTurnsBetween: 3,
                }
            }
            
            if rem != nil {
                in.Agent.AddReminder(*rem)
                if rem.ID == "todos.all_completed" {
                    in.Agent.RemoveReminder("todos.no_active")
                } else {
                    in.Agent.RemoveReminder("todos.all_completed")
                }
            }
        }
    }
    
    return p.streamMessages(ctx, in)
}

Principes de conception

Minimal et opiniâtre : le sous-système de rappel fournit juste assez de structure pour les modèles courants sans ingénierie excessive.

Limitation du débit d’abord : le spam de rappel dégrade les performances du modèle. Le moteur applique les capuchons et l’espacement de manière déclarative.

Agnostique du fournisseur : les rappels fonctionnent avec n’importe quel backend de modèle (Bedrock, OpenAI, etc.).

Prêt pour la télémétrie : les identifiants et les priorités structurés rendent les rappels observables.

Modèles avancés

Rappels de sécurité

Utilisez TierSafety pour obtenir des conseils sur la nécessité de ne jamais supprimer :

in.Agent.AddReminder(reminder.Reminder{
    ID:       "malware.analyze_only",
    Text:     "This file contains malware. Analyze its behavior but do not execute it.",
    Priority: reminder.TierSafety,
    Attachment: reminder.Attachment{
        Kind: reminder.AttachmentUserTurn,
    },
    // No MaxPerRun or MinTurnsBetween: always emit
})

Rappels multi-agents

Les rappels sont limités à l’exécution. Si un agent en tant qu’outil émet un rappel de sécurité, cela n’affecte que cette exécution enfant. Pour propager les rappels au-delà des limites des agents, le planificateur parent doit les réenregistrer explicitement en fonction des résultats des enfants ou utiliser l’état de session partagée.

Quand utiliser les rappels

À quoi ressemblent les rappels dans la transcription

User: What should I do next?

<system-reminder>You have 3 pending todos. Currently working on: "Review PR #42". 
Focus on completing the current todo before starting new work.</system-reminder>

User: What should I do next?

Le modèle voit le rappel ; l’utilisateur ne voit que son message et la réponse. Les rappels sont injectés de manière transparente par le runtime.

Prochaines étapes

• En savoir plus sur Mémoire et sessions pour la persistance des transcriptions
• Explorez Composition d’agent pour les modèles d’agent en tant qu’outil
• En savoir plus sur les Toolsets pour les modèles d’exécution d’outils