プロダクション

モデルのレート制限

どのモデルプロバイダーにもレート制限があります。超過すると 429 エラーで失敗します。さらに悪いことに、マルチレプリカ構成では各レプリカが独立に API を叩くため、個々のプロセスからは見えない形で合計のスロットリングが発生します。

問題

シナリオ: エージェントサービスを 10 レプリカでデプロイします。各レプリカは「100K tokens/min 使える」と思っています。合計では 1M tokens/min を送ってしまい、実際のクォータの 10 倍です。プロバイダーは強くスロットリングし、全レプリカでランダムに失敗します。

レート制限なしの場合:

• 429 により予測不能に失敗する
• 残りキャパシティが可視化されない
• リトライが混雑を悪化させる
• 負荷時に UX が劣化する

適応型レート制限ありの場合:

• レプリカ間で協調されたバジェットを共有できる
• キャパシティが空くまでリクエストをキューイングできる
• バックオフがクラスター全体に伝播する
• 失敗ではなく、段階的な劣化にできる

概要

features/model/middleware package は、検証済み model client の下に置く AIMD（Additive Increase / Multiplicative Decrease）スタイルの適応型 rate limiter を提供します。provider に正確な input-token count を問い合わせ、その capacity が利用可能になるまで caller を block し、provider の throttling に応じて input-tokens-per-minute budget を調整します。token を推定せず、output quota は計量しません。

AIMD 戦略

このリミッターは Additive Increase / Multiplicative Decrease (AIMD) を使います。

有効な tokens-per-minute（TPM）は次で制約されます:

• 最小: 初期 TPM の 10%（飢餓を防ぐフロア）
• 最大: 設定した maxTPM の上限

基本的な使い方

プロセスごとに 1 つの limiter を作成し、model client を wrap します:

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) // host が所有する runtime storage

    // Vertex Gemini exposes the exact CountTokens operation required by the limiter.
    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,     // process-local limiter
        "",      // unused for a process-local limiter
        60000,   // initial input tokens per minute
        120000,  // maximum input tokens per minute
    )

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

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

クラスタ対応のレート制限

マルチプロセスのデプロイでは、Pulse の replicated map を使ってインスタンス間でレート制限を協調させます。

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

    limiter := middleware.NewAdaptiveRateLimiter(
        ctx,
        rm,
        "vertex:gemini",  // shared key for this model family
        60000,            // initial TPM
        120000,           // max TPM
    )

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

replicated-map の read／write が成功している間:

• バックオフがグローバルに伝播: どれか 1 つのプロセスが ErrRateLimited を受けると、全プロセスがバジェットを減らします
• プロービングが協調される: 成功リクエストが共有バジェットを増やします
• 自動リコンシリエーション: 外部変更を監視し、ローカルリミッターを更新します

nil map または空 key を渡すと、意図して process-local limiter を作ります。両方を設定したものの shared key がなく、middleware が seed できない場合も process-local operation へ fallback します。startup 後に shared backoff／probe update が失敗しても model call 自体は失敗しません。その process は後続の replicated-map event で reconcile されるまで、local adaptive budget を使い続けます。cluster-wide coordination が必要なら Redis availability を監視してください。

正確な token count

limiter は capacity を予約する前に、wrapped client の CountTokens operation を呼びます。Exact=true を要求し、推定はしません。

Vertex Gemini は native count operation を公開します。Bedrock は対応箇所で Runtime CountTokens を使いますが、structured-output request と、別の AWS Mantle endpoint を必要とする Claude Opus 4.7、Sonnet 5、Mythos 5 などでは model.ErrTokenCountingUnsupported を返します。remote gateway が counting を保つのは gateway.NewCountingRemoteClient で構築した場合だけです。OpenAI には native token counter がありません。

counting 非対応の client も wrap 自体は成功します。最初の Complete または Stream call が inference 前に model.ErrTokenCountingUnsupported を返します。

limiter は成功した unary call または正常な stream end の後に上向き probe を行い、terminal model.ErrRateLimited の後に backoff します。terminal outcome に到達せず stream を開閉した場合は capacity を変更しません。

ランタイムとの統合

レート制限したクライアントを Goa-AI runtime に配線します。

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

負荷時に何が起きるか

チューニングパラメータ

例: initialTPM=60000, maxTPM=120000 の場合:

• Floor: 6,000 TPM
• Recovery: 成功バッチごとに +3,000 TPM
• Backoff: 429 のたびに現在 TPM を半減

監視

wrapped client 周辺の telemetry で model-call latency と terminal model.ErrRateLimited error を測定します。replicated map を使う場合は Redis も監視します。shared-state initialization や update が失敗すると、middleware は意図して process-local budget で model call を続行するためです。

ベストプラクティス

• モデル/プロバイダーごとに 1 リミッター: 異なるモデルのバジェットを分離するために、リミッターを分けます
• 現実的な初期 TPM: プロバイダーが提示するレート制限（または保守的な見積もり）から始めます
• 本番ではクラスタ対応を使う: レプリカの合算スロットリングを避けるため、インスタンス間で協調させます
• バックオフを監視: バックオフ発生時のログやメトリクスを出して、持続的なスロットリングを検出します
• maxTPM は初期より上に: クォータ未満のときにプロービングできる余白を残します

OpenTelemetry GenAI オブザーバビリティ

tracer を設定すると、Goa-AI は planner が実行するエージェント操作について、ベンダー非依存の OpenTelemetry GenAI semantic convention に沿った span を出力します。

• モデル呼び出しは gen_ai.operation.name="chat" を使い、chat {model} のような span 名になります
• ツール完了は gen_ai.operation.name="execute_tool" を使い、execute_tool {tool_name} のような span 名になります
• agent-as-tool の委譲は gen_ai.operation.name="invoke_agent" を使い、invoke_agent {agent_name} のような span 名になります

これらの span には、利用可能な場合に gen_ai.conversation.id、gen_ai.agent.id、gen_ai.agent.name、gen_ai.request.model、gen_ai.response.model、トークン使用量、終了理由、ツール識別子、streaming の time-to-first-chunk が含まれます。runtime は識別子、件数、時間、エラーを既定で記録します。prompt 本文、chat 履歴、ツール引数、ツール結果を添付するかどうかはアプリケーション側のポリシーであり、自動では添付されません。

これにより、Goa-AI の open source telemetry は OpenTelemetry backend 間で移植可能なまま、production システムは conversation のグルーピング、モデルごとの latency と token 使用量の比較、multi-agent tool chain の調査に必要な構造を得られます。

Mongo Store を使った Prompt Override

本番での Prompt 管理は通常、次の組み合わせで行います。

• runtime.PromptRegistry に登録したベースライン prompt spec
• features/prompt/mongo で Mongo に永続化したスコープ付き override レコード

配線

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

Override 解決順序とロールアウト

override の優先順位は決定的です。

1. session スコープ
2. facility スコープ
3. org スコープ
4. グローバルスコープ
5. ベースライン spec（override が存在しない場合）

推奨ロールアウト手順:

• まず新しいベースライン prompt spec を登録する
• override は広いスコープ（org）から開始し、facility/session へ絞ってカナリア展開する
• prompt_rendered イベントと model.Request.PromptRefs で実効バージョンを追跡する
• 同一スコープにより新しい override を書く（またはスコープ固有 override を削除する）ことでロールバックする

Temporal セットアップ

このセクションは、本番環境における耐久性のあるエージェントワークフローのために Temporal をセットアップする方法を扱います。

概要

Temporal は Goa-AI エージェントに耐久実行を提供します。エージェントの run は Temporal workflow になり、イベントソースな履歴を持ちます。ツール呼び出しは、リトライを設定可能な activity になります。すべての状態遷移が永続化され、ワーカーの再起動後も履歴をリプレイしてまったく同じ地点から再開します。

耐久性が機能する仕組み

具体例: エージェントが LLM を呼び出し、3 つのツール呼び出しが返ります。2 つは完了し、3 つ目のツール実行中にサービスがクラッシュしました。

• ❌ Temporal なし: run 全体が失敗します。推論（$$$）を再実行し、成功済み 2 ツールも再実行します。
• ✅ Temporal あり: クラッシュしたツールだけがリトライされます。workflow は履歴からリプレイされ、新たな LLM 呼び出しも完了済みツールの再実行もありません。コストは「1 回のリトライ」で済みます。

何が失敗に耐えるか

インストール

オプション 1: Docker（開発）

ローカル開発向けのワンライナー:

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

オプション 2: Temporalite（開発）

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

オプション 3: Temporal Cloud（本番）

temporal.io でサインアップし、クラウド認証情報でクライアントを設定します。

オプション 4: セルフホスト（本番）

Docker Compose または Kubernetes で Temporal をデプロイします。デプロイガイドは Temporal documentation を参照してください。

ランタイム設定

Goa-AI は実行バックエンドを Engine インタフェースの背後に抽象化しています。エージェントコードを変更せずにエンジンを差し替えられます。

インメモリエンジン（開発）:

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

Temporal エンジン（本番）:

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

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

temporalEng, err := runtimeTemporal.NewWorker(runtimeTemporal.Options{
    ClientOptions: &temporalclient.Options{
        HostPort:  "127.0.0.1:7233",
        Namespace: "default",
    },
    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))

ランタイムストレージの所有者

runtime.New には一つの storage.Store が必要です。プロダクションでは、一つのホストサービスがセッション状態、ランのメタデータ、継続用チェックポイント、変更不可の記録を格納するデータベースを所有します。エージェントワーカーは型付き API を通してそのサービスを呼び、自分で同じ collection に接続しません。

プロダクトデータはプロダクトサービスが引き続き所有します。たとえば、別のサービスが Goa-AI のランタイムストアを所有していても、チャットサービスはトランスクリプト、評価、検索用フィールドを保持します。

Memory & Sessions は storage contract 全体を定義しています。lifecycle change と record はまとめて 保存し、完全に同じ retry には受理済みの結果を返します。新しい child には running の parent が必要です。保存済み event の JSON は厳密に検証し、キャンセル理由の 由来も正確に保ちます。Runtime は engine history が閉じた後に host が最終 run event を検証して配信する方法を 定義します。各 worker で同じ規則を作り直さず、規則を所有する runtime store に 実装してください。

session.Store と runlog.Store からの変更は、storage 全体を協調して 切り替える必要があります。新 runtime が書き込む前に、既存の run metadata、 checkpoint、record が統合された storage.Store contract を満たしていなければ なりません。storage owner と、それを呼ぶすべての worker をまとめて deploy します。旧 split-store writer と新 integrated-store writer を同時に動かしては いけません。

traffic のない状態で、temporary database job または pod から disposable migration program を実行します。この program は host application が所有し、通常 release には 含めず、cutover の検証後に削除します。

1. runtime database を backup し、writer がないことを確認します。
2. verification mode で migration を実行し、拒否された record をすべて修正します。
3. conversion を適用し、schema、index、session、run metadata、現在の形式の checkpoint、 immutable record を検証します。
4. storage owner と全 worker をまとめて deploy し、migration program を削除します。

この物理 storage の変換は suspension 形式を変換しません。保存済み work は別途、 現在の continuation contract を満たす必要があります。完了済み history と保存済み tool result は保持してください。

conversion 開始後の rollback は database 全体の restore です。partial conversion の database に旧 writer を接続してはいけません。

ClientOptions.DataConverter を設定しないでください。Temporal engine は custom converter を拒否し、すべての worker と client が同じ永続 contract を使うよう Goa-AI の bounded converter を自ら install します。

Temporal payload contract

workflow または activity の各 argument list には、encode 後の aggregate limit engine.MaxPayloadBytes（1 MiB）があります。converter は encode 前に、depth が 64 level を超える value graph または visited value が 100,000 を超える graph も拒否します。oversized data を truncate しません。

planner.ToolResult は in-process value で、Temporal boundary を越えられません。workflow は代わりに正規 JSON byte を持つ api.ToolEvent を運びます。有効な tool result が 1 MiB を超え得る場合、tool executor は application-owned storage に保存し、型付き reference を返さなければなりません。runtime が黙って result を置き換えることはありません。

Timing と Activity Retry

Goa-AI は各 agent workflow を一度だけ開始します。以前の実行がすでに tool を 呼んだり最終 record を保存したりしている可能性があるため、失敗後に workflow 全体を最初から実行し直しません。耐久性は workflow history のリプレイと、 planner、tool、hook、storage の各 activity の再試行によって実現します。

DSL は semantic run budget、つまり run 全体にどれだけ時間を使えるか、planner attempt と tool attempt がどれだけ実行できるかを表します。

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

Temporal adapter は queue-wait や liveness timeout などの workflow-engine mechanics を所有します。DSL ではなく engine 側で設定してください:

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

生成される plan/resume、execute-tool、hook-publishing activity の retry policy は、retry が論理的に idempotent な場合にだけ安全です。hook event は安定した event key を持ち、tool execution は不可逆な副作用を繰り返すのではなく、ToolCallID 単位で canonical result を永続化または replay するべきです。

ワーカーのセットアップ

ワーカーは task queue をポーリングし、workflow/activity を実行します。登録された各エージェントに対してワーカーは自動的に開始されるため、ほとんどのケースで手動設定は不要です。

透過的なロールアウト

Temporal の durability と透過的な release は別の保証です。Temporal は workflow history を保存します。deployment は、その history が使われている間、互換性のある worker code と必要な downstream service を利用可能に保たなければなりません。

上の設定は worker を Temporal Worker Deployment Versioning に参加させます。releaseBuildID は 1 つの immutable binary または container image を識別しなければなりません。異なる workflow code に build ID を再利用したり、latest のような mutable tag を使ったりしないでください。

worker version は次の順序で release します。

1. 保持中の全 worker version と並べて新 worker を起動する。
2. 新 process が readiness を通過し、Temporal への registration に成功するまで待つ。
3. 新しい Worker Deployment Version を current にする。Temporal は新規 workflow をそこへ割り当て、既存 workflow は開始時の version に pinned されたままにする。
4. worker process が API も提供する場合、通常の API traffic は current ready build だけへ route する。古い pod は Temporal 用に生かすが、新規 API request は送らない。API deployment を分ける設計も有効だが必須ではない。
5. Temporal が drained と報告した後だけ古い version を削除する。pod が停止したことは、その code を必要とする workflow がない証明にはならない。

受理された各 user input は top-level Goa-AI workflow を 1 つ開始します。Goa-AI は human または external input を要求すると workflow を終了し、completed run ID の下に非公開 checkpoint を保存します。受理された answer は current worker version で新しい workflow を開始します。そのため透過的 release では、新 version が保存済み checkpoint version、生成 result codec、必要な tool 名との互換性を保つ必要があります。Worker Versioning は互換性のない保存値を変換できません。

現在の runtime が受理するのは goa-ai.run-suspension.v8 だけで、それ以前の形式は 拒否します。互換性のない upgrade では、worker を重ねて動かすこの手順ではなく、 以下の「生成 contract の変更」に従ってください。

application の残りの部分も、同じ overlap 中に availability を保つ必要があります。

• downstream Service には常に 1 つ以上の ready endpoint が必要です。readiness-gated rolling replacement を使います。Recreate rollout は gap を作ります。
• downstream API は retained worker と current worker の両方からの call を受け付けなければなりません。
• database migration は古い version が drained になるまで両 release を支える必要があります。古い code が schema を使わなくなる前に置き換えず、expand-then-contract sequence を使います。
• 1 process が API traffic と Temporal work の両方を提供する場合、traffic selector は Temporal が retained worker へ到達する仕組みとは別に current build を識別しなければなりません。

registry-backed tool provider

Goa-AI tool provider も readiness-gated rolling replacement に対応します。同じ生成 schema と admission revision を持つ provider replica は同じ registry admission に参加し、overlap できます。どちらかが変わる場合、old admission が authority を持つ間、replacement provider は生存したまま registration を retry します。old provider は新しい call の claim を止め、受理済み work を確定し、lease を release してから new admission が実行可能になります。そのため、異なる 2 つの tool contract が同じ toolset を同時に提供することはありません。

active toolset に healthy provider がないとき、有効な CallTool request は既存 execution deadline 内で待ちます。request publication は、call を append する同じ Redis operation で選択 provider を検証します。health check 後に old provider が draining を始めた場合、未 publish の call は deadline を延長せず replacement を選び直します。provider assignment が永久に固定されるのは publication 成功時だけです。caller cancellation が終了させるのはその transport attempt だけで、完全に同じ retry は未 publish call を続行できます。deadline expiry は通常の durable call_not_admitted decision を記録します。

registry が所有するのは provider health であり deployment intent ではありません。rollout handoff と別の provider outage を区別できないため、同じ bounded wait が両方に適用されます。pod 名や version string を調べず、model に retry を求めません。この contract により、互換性のない provider generation の overlap を許さず、consumer は provider change に 1 つの rolling release policy を使えます。

release 中も registry client、server、provider は互換 wire protocol を使う必要があります。envelope を互換性なく変更する必要がある場合、まず両形式を受理する code を release してください。registry は rolling overlap 中に protocol version を negotiate しません。

Worker Deployment Versioning は workflow replay を保護します。dependency が利用不能、API が非互換、checkpoint が非互換な場合から workflow を保護するものではありません。

生成 contract の変更

生成 agent、completion package、永続 runtime payload を互換性なく変更する場合、上記の mixed-version 手順を適用しません。全 agent と completion を再生成し、影響を受ける work を drain または停止して、runtime、worker、caller を 1 回の coordinated release で deploy します。Goa-AI は生成 runtime contract の dual-read mode を提供しません。

runtime が受理するのは、正確な goa-ai.run-suspension.v8 schema だけです。question、clarification、external tool を待つ planner は provider の ModelToolCallID を保持し、workflow は停止データを保存する前に別の runtime ToolCallID を割り当てます。ほかの suspension schema は resume しません。version eight は、受理済み recovery plan が input を待つとき、提示した選択肢も保持します。古い保存済み work を所有する worker の置き換え前に、External Input and Workflow Continuations で説明した、host による保存方法と再開可能性の判断を行ってください。dual reader を追加したり field を推測したりしてはいけません。

新規 work を受け付ける前に、すべての workflow queue と activity queue で協調する worker、生成済み package、caller を更新します。新 worker が v8 suspension を保存した 後は、旧 worker ではそれを再開できません。image を戻すだけでは再開可能性は戻りません。 完了済み history と保存済み tool result は保持します。物理 storage の変換は、別途 host が所有する操作です。

release の検証

次の check がすべて通るまで release を透過的とみなさないでください。

• promotion 前に開始した workflow が元の build で完了する。
• 新しい workflow が current build で開始して完了する。
• promotion 前に作られた external-input request が、promotion 後に新しい workflow として正常に続行する。
• API traffic が current ready build だけに到達する。
• Temporal が drained と報告するまで old worker が ready のままである。
• replacement 中、各 downstream Service に ready endpoint が残る。
• observation window に新しい workflow failure、container restart、readiness gap がない。

1 turn につき 1 workflow と cross-workflow event identity の contract は External Input and Workflow Continuations を参照してください。

ベストプラクティス

• 環境ごとに namespace を分ける（dev / staging / prod）
• ツールセットごとにリトライポリシーを調整する（信頼性特性に合わせる）
• Temporal UI と観測性ツールで実行状況を監視する
• activity のタイムアウトを適切に設定する（信頼性とハング検知のバランス）
• 本番は Temporal Cloud を推奨（運用負荷を下げる）

UI ストリーミング

このセクションでは、Goa-AI のストリーミング基盤を使って、エージェントのイベントを UI にリアルタイム配信する方法を説明します。

概要

Goa-AI は run ごとに型付きイベントストリームを公開しており、以下の手段で UI に届けられます:

• Server-Sent Events（SSE）
• WebSockets
• メッセージバス（Pulse、Redis Streams など）

セッションに属するストリーミング可視イベントは、単一のストリーム session/<session_id> に追記されます。各イベントは run_id と session_id を持ち、child_run_linked で親ツールコールと子 run をリンクします。UI はアクティブ run の run_stream_end を観測したら SSE/WebSocket を終了できます（タイマー不要）。

Stream Sink インタフェース

stream.Sink インタフェースを実装します。

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

イベント型

stream パッケージは stream.Event を実装する具体的なイベント型を定義します。UI でよく使うものは次です。

トランスポート層は通常、コンパイル時の安全性のために stream.Event に対して type switch します。

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
}

例: SSE Sink

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
}

セッションストリームの購読（Pulse）

プロダクションでは UI はセッションストリーム（session/<session_id>）を購読し、run_id でフィルタして描画します。アクティブ run の run_stream_end を観測したら SSE/WebSocket を終了します。

グローバルな Stream Sink

すべての run をグローバル sink（たとえば Pulse）に流したい場合、runtime に stream sink を設定します。

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

runtime はデフォルトの stream.Subscriber をインストールします。これは:

• hook イベントを stream.Event にマップし
• **デフォルト StreamProfile**を使用して、アシスタント返信、プランナー思考、ツール start/update/end、await、usage、workflow、child_run_linked、および終端マーカー run_stream_end を出力します

Stream Profile

すべてのコンシューマがすべてのイベントを必要とするわけではありません。Stream profile は用途に応じてイベントをフィルタし、ノイズや帯域を削減します。

組み込みプロファイルの使用:

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

カスタムプロファイル:

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

カスタムプロファイルは次のような場合に有用です:

• 特定コンシューマ向けのイベントだけが必要（例: 進捗トラッキング）
• モバイル向けにペイロードを削減したい
• 一部イベントだけの分析パイプラインを作りたい

高度: Pulse と Stream ブリッジ

本番ではよく次が必要になります:

• イベントを共有バス（例: Pulse）へ publish する
• そのバス上で session-owned stream (session/<session_id>) を使う

Goa-AI は次を提供します:

• features/stream/pulse – Pulse backed な stream.Sink
• runtime/agent/stream/bridge – hook bus を任意の sink に配線するためのヘルパ

典型的な配線は次の通りです。

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

システムリマインダー

モデルはドリフトします。指示を忘れます。10 ターン前には明らかだった文脈を無視します。長時間タスクを実行するエージェントでは、ユーザー会話を汚さずに 動的で文脈的なガイダンスを注入する仕組みが必要です。

問題

シナリオ: エージェントが todo リストを管理しています。20 ターン後にユーザーが「次は何？」と尋ねましたが、モデルがドリフトしていて「進行中の todo がある」ことを覚えていません。ユーザーに「REMINDER: …」のような不自然なメッセージを見せずに、モデルだけに促しを入れたいです。

システムリマインダーなしの場合:

• system prompt があらゆるケースで肥大化する
• 会話が長くなるほどガイダンスが埋もれる
• ツール結果に基づいて文脈を注入できない
• ユーザーに内部の足場が見えてしまう

システムリマインダーありの場合:

• ランタイム状態に応じて動的にガイダンスを注入できる
• 反復ヒントをレート制限してプロンプト肥大を防げる
• 優先度により、安全上重要なガイダンスは抑制されない
• ユーザーには見えない（<system-reminder> ブロックとして注入される）

概要

runtime/agent/reminder パッケージは次を提供します:

• 優先度 tiers、アタッチポイント、レート制限ポリシーを持つ 構造化リマインダー
• run スコープのストレージ（run 完了後に自動クリーンアップ）
• モデルトランスクリプトへの <system-reminder> 自動注入
• planner や tool からリマインダーを登録・削除するための PlannerContext API

コア概念

リマインダー構造

reminder.Reminder は次を持ちます:

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

優先度 tiers

リマインダーは優先度順に並べられ、プロンプト予算を管理しつつ重要なガイダンスが抑制されないようにします。

例:

• TierSafety: 「このマルウェアは実行せず、分析のみ行う」「認証情報を漏らさない」
• TierCorrect: 「結果が切り捨てられているのでクエリを絞る」「データが古い可能性がある」
• TierGuidance: 「進行中の todo がないので 1 つ選んで開始する」

アタッチポイント

リマインダーは会話の特定ポイントに注入されます。

レート制限

2 つの仕組みでリマインダーのスパムを防ぎます:

• MaxPerRun: run あたりの総出力回数（0 = 無制限）
• MinTurnsBetween: 出力の間隔として必要な planner turn 数（0 = 無制限）

使用パターン

DSL による静的リマインダー

特定ツール結果の後に常に出したいリマインダーは、ツール定義で ResultReminder DSL 関数を使います。

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

これは、ツール呼び出しのたびに適用される場合に向いています。詳細は DSL Reference を参照してください。

プランナーからの動的リマインダー

実行時状態やツール結果内容に依存する場合は 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...
}

リマインダーの削除

前提条件が成立しなくなったら RemoveReminder() を使います。

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

レート制限カウンターの保持

AddReminder() は、同じ ID の既存リマインダーを更新する場合に出力カウンターを保持します。内容を更新しつつレート制限も保ちたいなら:

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

アンチパターン: 同じ ID に対して RemoveReminder() の後に AddReminder() を呼ぶのは避けてください。カウンターがリセットされ、MinTurnsBetween をバイパスします。

注入とフォーマット

自動タグ付け

ランタイムは、トランスクリプトに注入する際に自動で <system-reminder> タグを付けます。

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

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

モデルへの説明

モデルが <system-reminder> ブロックをどう解釈すべきか理解できるように、system prompt に reminder.DefaultExplanation を含めます。

const systemPrompt = `
You are a helpful assistant.

` + reminder.DefaultExplanation + `

Follow all instructions carefully.
`

完全な例

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

設計原則

最小限で意見的: よくあるパターンに十分な構造だけを提供し、過剰に作り込みません。

レート制限ファースト: リマインダーのスパムはモデル性能を下げます。エンジンが上限と間隔を宣言的に強制します。

プロバイダー非依存: Bedrock、OpenAI など、任意のバックエンドで動きます。

テレメトリ容易: 構造化された ID と優先度により観測しやすくなります。

高度なパターン

安全リマインダー

絶対に抑制してはいけないガイダンスには TierSafety を使います。

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

クロスエージェントのリマインダー

リマインダーは run スコープです。agent-as-tool が安全リマインダーを出しても、その子 run にしか影響しません。エージェント境界を越えて伝播させたい場合、親プランナーが子結果に基づいて明示的に再登録するか、共有セッション状態を使う必要があります。

いつリマインダーを使うか

トランスクリプト上での見え方

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?

モデルはリマインダーを見ますが、ユーザーには自分のメッセージと応答しか見えません。リマインダーはランタイムが透過的に注入します。

次のステップ

• トランスクリプト永続化のために Memory & Sessions を読む
• agent-as-tool パターンとして Agent Composition を読む
• ツール実行モデルとして Toolsets を読む