Produzione

Limitazione della velocità del modello

Ogni fornitore di modelli applica limiti di velocità. Se li superate, le vostre richieste falliscono con 429 errori. Peggio ancora: in un’implementazione multi-replica, ogni replica martella l’API in modo indipendente, causando un throttling aggregato invisibile ai singoli processi.

Il problema

Scenario: Si distribuiscono 10 repliche del servizio agente. Ogni replica pensa di avere a disposizione 100K token/minuto. Insieme, inviano 1M di token/minuto, 10 volte la quota effettiva. Il provider applica un throttling aggressivo. Le richieste falliscono in modo casuale su tutte le repliche.

**Senza limitazione della velocità

• Le richieste falliscono in modo imprevedibile con 429 secondi
• Nessuna visibilità sulla capacità residua
• I tentativi peggiorano la congestione
• L’esperienza dell’utente peggiora sotto carico

Con la limitazione adattiva della velocità:

• Ogni replica condivide un budget coordinato
• Le richieste vengono messe in coda finché la capacità non è disponibile
• Il backoff si propaga in tutto il cluster
• Degrado graduale invece di guasti

Panoramica

Il pacchetto features/model/middleware fornisce un limitatore di velocità adattativo in stile AIMD sotto il client del modello convalidato. Chiede al provider il conteggio esatto dei token di input, blocca i chiamanti finché non è disponibile quella capacità e regola il budget di token di input al minuto quando il provider applica il throttling. Non stima mai i token e non misura le quote di output.

Strategia AIMD

Il limitatore utilizza una strategia Additive Increase / Multiplicative Decrease (AIMD):

Il numero effettivo di token al minuto (TPM) è limitato da:

• Minimo: 10% del TPM iniziale (soglia minima per prevenire l’inedia)
• Massimo: Il limite massimo configurato maxTPM

Uso di base

Creare un singolo limitatore per processo e avvolgere il client del modello:

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) // storage del runtime fornito dall'host

    // Vertex Gemini fornisce il conteggio esatto richiesto dal limitatore.
    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)
    }
}

Limitazione della velocità consapevole del cluster

Per le distribuzioni multiprocesso, coordinare la limitazione della velocità tra le istanze usando una mappa replicata di 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",  // chiave condivisa per questa famiglia di modelli
        60000,            // initial TPM
        120000,           // max TPM
    )

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

Finché le letture e scritture della mappa replicata riescono:

• Il backoff si propaga a livello globale: Quando un processo riceve ErrRateLimited, tutti i processi riducono il loro budget
• Il probing è coordinato: Le richieste riuscite incrementano il budget condiviso
• Riconciliazione automatica: I processi osservano le modifiche esterne e aggiornano i loro limitatori locali

Una mappa nil o una chiave vuota crea deliberatamente un limitatore locale al processo. Se lo stato condiviso non può essere inizializzato o aggiornato, le chiamate al modello continuano con il budget adattativo locale finché un evento successivo della mappa non lo sincronizza. Monitora Redis quando serve il coordinamento tra processi.

Conteggio esatto dei token

Il limitatore chiama CountTokens sul client avvolto prima di riservare capacità. Richiede Exact=true e non effettua stime.

Vertex Gemini fornisce un’operazione nativa. Bedrock restituisce model.ErrTokenCountingUnsupported quando non può contare esattamente, mentre OpenAI non offre un contatore nativo. Un client senza supporto può essere avvolto, ma la prima chiamata Complete o Stream restituisce quell’errore prima dell’inferenza.

Integrazione con il runtime

Cablare i client a velocità limitata nel runtime di 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)
}

Cosa succede sotto carico

Parametri di regolazione

Esempio: Con initialTPM=60000, maxTPM=120000:

• Piano: 6.000 TPM
• Recupero: +3.000 TPM per lotto riuscito
• Backoff: dimezzare i TPM correnti su 429

Monitoraggio

Misura la latenza delle chiamate al modello e gli errori terminali model.ErrRateLimited nella telemetria del client avvolto. Monitora anche Redis quando usi una mappa replicata.

Migliori pratiche

• Un limitatore per modello/provider: Creare limitatori separati per i diversi modelli per isolare i loro budget
• Impostare un TPM iniziale realistico: Iniziare con il limite di velocità documentato dal provider o con una stima prudente
• Utilizzare la limitazione consapevole del cluster in produzione: Coordinarsi tra le repliche per evitare il throttling aggregato
• Monitorare gli eventi di backoff: Registrare o emettere metriche quando si verificano i backoff per rilevare il throttling prolungato
• Impostare il maxTPM al di sopra di quello iniziale: Lasciare spazio per il probing quando il traffico è al di sotto della quota

Osservabilità GenAI con OpenTelemetry

Quando è configurato un tracer, Goa-AI emette span neutrali rispetto al vendor secondo le convenzioni semantiche GenAI di OpenTelemetry per le operazioni agente eseguite dal planner:

• le chiamate al modello usano gen_ai.operation.name="chat" e nomi di span come chat {model}
• le conclusioni delle chiamate agli strumenti usano gen_ai.operation.name="execute_tool" e nomi di span come execute_tool {tool_name}
• la delega agent-as-tool usa gen_ai.operation.name="invoke_agent" e nomi di span come invoke_agent {agent_name}

Questi span includono gen_ai.conversation.id, gen_ai.agent.id, gen_ai.agent.name, gen_ai.request.model, gen_ai.response.model, uso dei token, motivi di terminazione, identificatori degli strumenti e tempo al primo frammento di streaming quando disponibili. Il runtime registra per impostazione predefinita identificatori, conteggi, tempi ed errori; testo dei prompt, cronologia chat, argomenti degli strumenti e risultati degli strumenti restano una scelta dell’applicazione e non vengono allegati automaticamente.

In questo modo la telemetria open source di Goa-AI resta portabile tra backend OpenTelemetry, ma offre ai sistemi di produzione abbastanza struttura per raggruppare una conversazione, confrontare latenza e uso dei token per modello e ispezionare catene di strumenti multi-agente.

Override dei prompt con Mongo Store

In produzione, la gestione dei prompt usa in genere:

• prompt spec baseline registrate in runtime.PromptRegistry, e
• record di override con scope persistiti in Mongo tramite features/prompt/mongo.

Wiring

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),
)

Risoluzione degli override e rollout

La precedenza degli override e deterministica:

1. scope session
2. scope facility
3. scope org
4. scope globale
5. prompt spec baseline (quando non esiste override)

Strategia di rollout consigliata:

• Registra prima le nuove prompt spec baseline.
• Esegui gli override prima su scope ampio (org), poi restringi a facility/session per i canary.
• Traccia le versioni effettive tramite eventi prompt_rendered e model.Request.PromptRefs.
• Esegui rollback scrivendo un override piu recente nello stesso scope (o rimuovendo override specifici per tornare al fallback).

Impostazione temporale

Questa sezione tratta l’impostazione di Temporal per i flussi di lavoro degli agenti durevoli negli ambienti di produzione.

Panoramica

Temporal fornisce un’esecuzione duratura per gli agenti Goa-AI. Le esecuzioni degli agenti diventano flussi di lavoro Temporal con cronologia basata sugli eventi. Le chiamate allo strumento diventano attività con tentativi configurabili. Ogni transizione di stato viene conservata. Un worker riavviato riproduce la cronologia e riprende esattamente da dove si era interrotto.

Come funziona la durata

Esempio concreto: L’agente chiama un LLM, che restituisce 3 chiamate allo strumento. Due strumenti vengono completati. Il servizio del terzo strumento si blocca.

• ❌ Senza Temporale: L’intera esecuzione fallisce. Si esegue nuovamente l’inferenza ($$$) e si rieseguono i due strumenti che hanno avuto successo.
• ✅ Con Temporale: Si riprova solo lo strumento che si è arrestato. Il flusso di lavoro viene riproposto dalla storia: nessuna nuova chiamata a LLM, nessuna riesecuzione di strumenti completati. Costo: un tentativo, non un riavvio completo.

Cosa sopravvive ai guasti

Installazione

Opzione 1: Docker (sviluppo)

Una sola riga per lo sviluppo locale:

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

Opzione 2: Temporalite (Sviluppo)

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

Opzione 3: Nuvola Temporale (Produzione)

Registrarsi su temporal.io e configurare il client con le credenziali del cloud.

Opzione 4: Self-Hosted (Produzione)

Distribuire Temporal utilizzando Docker Compose o Kubernetes. Vedere la Documentazione di Temporal per le guide alla distribuzione.

Configurazione del runtime

Goa-AI astrae il backend di esecuzione dietro l’interfaccia Engine. È possibile cambiare i motori senza modificare il codice dell’agente:

In-Memory Engine (sviluppo):

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

Motore temporale (produzione):

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))

Proprietà dell’archiviazione del runtime

runtime.New richiede un solo storage.Store. In produzione, un unico servizio dell’applicazione deve possedere il database che contiene stato delle sessioni, metadati delle esecuzioni, checkpoint di continuazione e record immutabili. I worker degli agenti chiamano quel proprietario tramite un’API tipizzata; non aprono connessioni separate alle sue raccolte.

I dati del prodotto restano di proprietà del servizio del prodotto. Per esempio, un servizio chat conserva trascrizioni, valutazioni e campi di ricerca anche quando un altro servizio possiede l’archivio del runtime Goa-AI.

Memoria e sessioni definisce il contratto completo dello storage: i cambiamenti del ciclo di vita e i record vengono salvati insieme, i retry identici restituiscono il risultato accettato, i nuovi figli richiedono un padre attivo, il JSON degli eventi salvati è rigoroso e l’origine dell’annullamento viene preservata. Runtime descrive come gli host convalidano e consegnano gli eventi finali dopo la chiusura della cronologia del motore. Mantieni queste regole nello storage del runtime che ne è responsabile invece di ripeterle in ogni worker.

Il passaggio da session.Store e runlog.Store è una modifica coordinata dello storage. Prima che il nuovo runtime scriva, metadati, checkpoint e record esistenti devono rispettare il contratto integrato di storage.Store. Distribuisci insieme il proprietario dello storage e tutti i worker che lo chiamano. I vecchi writer degli storage separati e i nuovi writer dello storage integrato non devono sovrapporsi.

Eseguire la conversione senza traffico con un programma di migrazione monouso in un job o pod temporaneo del database. L’applicazione host possiede quel programma; non fa parte del release normale e va eliminato dopo la verifica.

1. Creare e verificare un backup e confermare che non vi siano writer.
2. Eseguire la migrazione in modalità verifica e correggere ogni record rifiutato.
3. Applicare la conversione e verificare schema, indici, sessioni, metadati, checkpoint v7 e record immutabili.
4. Distribuire insieme proprietario e worker, quindi eliminare il programma.

Dopo l’avvio della conversione, il rollback ripristina il backup completo. Non eseguire writer precedenti su un database convertito solo in parte.

Timing e retry delle attività

Usa il DSL per budget semantici di run: quanto può durare l’intero run, quanto può durare un tentativo del planner e quanto può durare un tentativo del tool.

Goa-AI avvia ogni workflow dell’agente una sola volta. Non riavvia l’intero workflow dopo un errore perché un tentativo precedente può avere già chiamato tool o salvato un record finale. La durabilità deriva dal replay della cronologia e dai retry delle singole activity di planner, tool, hook e storage.

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

L’adapter Temporal possiede le meccaniche del workflow engine, come queue-wait e timeout di liveness. Configurale sull’engine, non nel 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,
        },
    },
})

Le attività generate di plan/resume, execute-tool e pubblicazione hook usano politiche di retry sicure solo quando i retry sono logicamente idempotenti. Gli eventi hook hanno chiavi evento stabili, e le esecuzioni dei tool devono persistire o riprodurre i risultati canonici per ToolCallID invece di ripetere effetti collaterali irreversibili.

Configurazione del lavoratore

I worker eseguono il polling delle code di attività e l’esecuzione di flussi di lavoro/attività. I worker vengono avviati automaticamente per ogni agente registrato; nella maggior parte dei casi non è necessaria una configurazione manuale dei worker.

Rollout trasparenti

Temporal riproduce la cronologia di un workflow con il codice del worker che lo esegue. In produzione usare la Worker Deployment Versioning di Temporal con build ID immutabili e mantenere ogni vecchia versione finché Temporal non segnala che è stata completamente drenata. Avviare un worker nuovo non rende sicuro riprodurre un workflow esistente con codice diverso.

Le modifiche compatibili possono essere distribuite in modalità rolling solo quando preservano i nomi richiesti degli strumenti, i codec generati, i payload persistiti e la semantica delle attività. La disponibilità delle dipendenze e la compatibilità delle API restano responsabilità dell’applicazione: il versionamento dei worker protegge il replay, non un servizio a valle assente o incompatibile.

Modifiche ai contratti generati

Quando agenti generati, package di completion o payload runtime persistiti cambiano in modo incompatibile, rigenerare tutti gli agenti e le completion, drenare o arrestare il lavoro interessato e distribuire runtime, worker e chiamanti come un’unica release coordinata. Goa-AI non offre una modalità di lettura doppia per i contratti runtime generati.

Il runtime accetta esclusivamente lo schema esatto goa-ai.run-suspension.v7. I planner che attendono domande, chiarimenti o strumenti esterni conservano il ModelToolCallID del provider; il workflow assegna il distinto ToolCallID del runtime prima di salvare la sospensione. Gli altri schemi di sospensione non vengono ripresi. Un futuro cambio di schema deve censire e ritirare il lavoro salvato incompatibile prima della release coordinata; non aggiungere un lettore doppio e non inferire campi.

Verifica della release

Prima di considerare trasparente una release, verificare che un workflow iniziato prima della promozione termini sulla build originale, che un nuovo workflow termini sulla build corrente e che una richiesta di input esterno creata prima della promozione continui come nuovo workflow. I vecchi worker devono restare pronti finché Temporal non li segnala drenati e la finestra di osservazione non deve mostrare nuovi errori, riavvii o intervalli senza endpoint pronti.

Migliori pratiche

• Usare spazi dei nomi separati per i diversi ambienti (dev, staging, prod)
• Configurare i criteri di riprova per ogni set di strumenti in base alle caratteristiche di affidabilità
• Monitorare l’esecuzione del flusso di lavoro utilizzando l’interfaccia utente e gli strumenti di osservabilità di Temporal
• Impostare timeout appropriati per le attività - bilanciare l’affidabilità rispetto al rilevamento delle sospensioni
• Usare Temporal Cloud per la produzione per evitare oneri operativi

Interfaccia utente di streaming

Questa sezione mostra come trasmettere gli eventi dell’agente alle interfacce utente in tempo reale, utilizzando l’infrastruttura di streaming di Goa-AI.

Panoramica

Goa-AI pubblica uno stream di proprietà della sessione di eventi tipizzati che può essere fornito alle interfacce utente tramite:

• Eventi inviati dal server (SSE)
• WebSocket
• Bus di messaggi (Pulse, Redis Streams, ecc.)

Tutti gli eventi visibili per una sessione vengono aggiunti a un unico stream: session/<session_id>. Ogni evento include run_id e session_id e il runtime emette child_run_linked per collegare una chiamata allo strumento padre a un’esecuzione figlia, oltre a run_stream_end come marcatore esplicito per chiudere SSE/WebSocket senza timer.

Interfaccia Stream Sink

Implementare l’interfaccia stream.Sink:

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

Tipi di evento

Il pacchetto stream definisce tipi di eventi concreti che implementano stream.Event. Quelli più comuni per le interfacce utente sono:

I trasporti di solito commutano di tipo su stream.Event per la sicurezza in fase di compilazione:

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
}

Esempio: Lavello 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
}

Abbonamento allo stream di sessione (Pulse)

In produzione, le UI consumano lo stream di sessione (session/<session_id>) da un bus condiviso e filtrano per run_id. Chiudi SSE/WebSocket quando osservi run_stream_end per la run attiva.

Flusso globale

Per eseguire lo streaming di tutte le corse attraverso un sink globale (ad esempio, Pulse), configurare il runtime con uno stream sink:

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

Il runtime installa un stream.Subscriber predefinito che:

• mappa gli eventi hook a valori stream.Event
• utilizza il default StreamProfile, che emette le risposte dell’assistente, i pensieri del pianificatore, l’avvio/aggiornamento/fine dello strumento, le attese, l’utilizzo, il flusso di lavoro, i collegamenti child_run_linked e il marcatore terminale run_stream_end

Profili dei flussi

Non tutti i consumatori hanno bisogno di tutti gli eventi. I profili di flusso filtrano gli eventi per pubblici diversi, riducendo il rumore e la larghezza di banda per casi d’uso specifici.

Utilizzo di profili integrati:

// 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)

Profili personalizzati:

// 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)

I profili personalizzati sono utili quando:

• Si ha bisogno di eventi specifici per un utente specializzato (ad esempio, il monitoraggio dei progressi)
• Si desidera ridurre la dimensione del payload per i client mobili
• Si stanno costruendo pipeline di analisi che necessitano solo di determinati eventi

Avanzato: Ponti di impulsi e flussi

Per le configurazioni di produzione, spesso si desidera:

• pubblicare eventi su un bus condiviso (ad esempio, Pulse)
• utilizzare uno stream di proprietà della sessione su quel bus (session/<session_id>)

Goa-AI fornisce:

• features/stream/pulse - un’implementazione stream.Sink supportata da Pulse
• runtime/agent/stream/bridge - helper per collegare il bus hook a qualsiasi sink

Cablaggio tipico:

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),
)

Promemoria del sistema

I modelli vanno alla deriva. Dimenticano le istruzioni. Ignorano un contesto che era chiaro 10 turni fa. Quando il vostro agente esegue compiti di lunga durata, avete bisogno di un modo per iniettare guida dinamica e contestuale senza inquinare la conversazione con l’utente.

Il problema

Scenario: Il vostro agente gestisce un elenco di cose da fare. Dopo 20 turni, l’utente chiede “cosa c’è dopo?”, ma il modello si è allontanato e non ricorda che c’è una cosa in sospeso in corso. È necessario dare una spinta senza che l’utente veda un messaggio imbarazzante del tipo “Promemoria: hai un impegno in corso”.

**Senza promemoria di sistema

• Si gonfia il prompt di sistema con ogni possibile scenario
• La guida si perde in lunghe conversazioni
• Non c’è modo di inserire un contesto basato sui risultati dello strumento
• Gli utenti vedono l’impalcatura interna dell’agente

**Con i promemoria di sistema

• Iniettare la guida dinamicamente in base allo stato del runtime
• Limitazione della frequenza dei suggerimenti ripetitivi per evitare l’ingrossamento dei prompt
• I livelli di priorità assicurano che la guida alla sicurezza non venga mai soppressa
• Invisibile agli utenti: viene iniettato come blocco <system-reminder>

Panoramica

Il pacchetto runtime/agent/reminder fornisce:

• Ricordi strutturati con livelli di priorità, punti di aggancio e politiche di limitazione del tasso di esecuzione
• Magazzino con scansione delle esecuzioni che si pulisce automaticamente al termine di ogni esecuzione
• Iniezione automatica nelle trascrizioni dei modelli come blocchi <system-reminder>
• aPI PlannerContext per registrare e rimuovere i promemoria dai pianificatori e dagli strumenti

Concetti fondamentali

Struttura del promemoria

Un reminder.Reminder ha:

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)
}

Tierini di priorità

I promemoria sono ordinati per priorità, in modo da gestire i budget tempestivi e garantire che le indicazioni critiche non vengano mai soppresse:

Casi d’uso esemplificativi:

• TierSafety: “Non eseguire questo malware; solo analizzare”, “Non divulgare le credenziali”
• TierCorrect: “I risultati sono troncati; restringi la ricerca”, “I dati potrebbero essere obsoleti”
• TierGuidance: “Non ci sono attività in corso; sceglietene una e iniziate”

Punti allegati

I promemoria vengono iniettati in punti specifici della conversazione:

Limitazione della velocità

Due meccanismi impediscono lo spam di promemoria:

• MaxPerRun: Limitare le emissioni totali per ogni corsa (0 = illimitato)
• MinTurnsBetween: Imporre un numero minimo di giri del pianificatore tra un’emissione e l’altra (0 = nessun limite)

Schema di utilizzo

Ricordi statici via DSL

Per i promemoria che devono sempre apparire dopo un risultato specifico dello strumento, utilizzare la funzione DSL ResultReminder nella definizione dello strumento:

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.")
})

Questa funzione è ideale quando il promemoria si applica a ogni invocazione dello strumento. Per maggiori dettagli, vedere il Riferimento DSL.

Ricordi dinamici dai pianificatori

Per i promemoria che dipendono dallo stato di esecuzione o dal contenuto dei risultati dello strumento, utilizzare 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...
}

Rimuovere i promemoria

Usare RemoveReminder() quando una precondizione non è più valida:

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

Preservare i contatori dei limiti di frequenza

AddReminder() preserva i contatori delle emissioni quando si aggiorna un promemoria esistente per ID. Se è necessario modificare il contenuto del promemoria ma mantenere i limiti di velocità:

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

Antipattern: Non chiamate RemoveReminder() seguito da AddReminder() per lo stesso ID: questo azzera i contatori e bypassa MinTurnsBetween.

Iniezione e formattazione

Etichettatura automatica

Il runtime avvolge automaticamente il testo del promemoria in tag <system-reminder> quando lo inietta nelle trascrizioni:

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

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

Spiegare i promemoria ai modelli

Includere reminder.DefaultExplanation nel prompt del sistema in modo che i modelli sappiano come interpretare i blocchi <system-reminder>:

const systemPrompt = `
You are a helpful assistant.

` + reminder.DefaultExplanation + `

Follow all instructions carefully.
`

Esempio completo

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)
}

Principi di progettazione

Minimale e autorevole: Il sottosistema dei promemoria fornisce una struttura sufficiente per gli schemi comuni, senza eccedere nell’ingegnerizzazione.

Limitare il tasso prima di tutto: Lo spam dei promemoria degrada le prestazioni del modello. Il motore impone l’uso di maiuscole e spaziature in modo dichiarativo.

Agnostico rispetto ai provider: I promemoria funzionano con qualsiasi backend del modello (Bedrock, OpenAI, ecc.).

Predisposizione alla telemetria: Gli ID strutturati e le priorità rendono i promemoria osservabili.

Modelli avanzati

Promemoria di sicurezza

Usare TierSafety per una guida che non deve mai essere soppressa:

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
})

Rimandi tra agenti

I promemoria sono a livello di run. Se un agente-as-tool emette un promemoria di sicurezza, questo ha effetto solo su quell’esecuzione figlia. Per propagare i promemoria attraverso i confini degli agenti, il pianificatore genitore deve registrarli nuovamente in modo esplicito in base ai risultati del figlio o utilizzare lo stato di sessione condiviso.

Quando usare i promemoria

Come appaiono i promemoria nella trascrizione

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?

Il modello vede il promemoria; l’utente vede solo il suo messaggio e la risposta. I promemoria sono iniettati in modo trasparente dal runtime.

Passi successivi

• Conoscere Memoria e sessioni per la persistenza delle trascrizioni
• Esplorare Agent Composition per i modelli di agente come strumento
• Leggere Toolsets per i modelli di esecuzione degli strumenti