アンビエントエージェント

アンビエントエージェントの実行

トリガーハンドラーが必要です。サポートされているパスは with_invoker で、adk_core::AgentInvoker を実装する任意のものを受け取ります — Runner は次のようにします。

use adk_agent::ambient::{AmbientAgent, RunnerTriggerConfig};
use std::sync::Arc;

let mut ambient = AmbientAgent::new(agent, source)
    .with_invoker(runner, RunnerTriggerConfig::new("system"))
    .with_max_concurrent_triggers(4);

let mut outputs = ambient.take_output(64);
ambient.start().await?;

RunnerTriggerConfig は3つのことを制御します。

PerTrigger がデフォルトなのは、毎分実行されるスケジュールを 1 つの共有セッションに送り続けると、そのセッションの履歴、ひいては後続の各実行にかかるトークンコストが際限なく増加するためです。 Runner は、同じ共有セッションを対象とする外部から呼び出されたターンを、各イベントストリームが完了するまで直列化します。一方、異なるセッション ID は引き続き同時実行できます。

AgentInvoker::invoke は、セッションが存在しない場合に作成します。Runner::run は作成せず、既存の セッションを解決し、それ以外の場合はストリームを通じて session.not_found を返します。外部からトリガーされた実行には、事前に登録する機会がありません。

呼び出し元が実行可能ルートを公開している場合（Runner など）、with_invoker はそのエージェントを環境ログと診断に使用します。これにより、テレメトリ上では 1 つのエージェントが示されているのに、実際には別のエージェントがトリガーを処理しているという事態を防げます。

ハンドラーを直接指定する

with_trigger_handler は、Runner 以外のものを駆動する呼び出し元向けに引き続き利用できます。これはイベントとエージェントを受け取り、イベントストリームを返す必要があります。その場合、セッションの作成はハンドラーの責任となります。

use adk_agent::ambient::{AmbientAgent, TriggerHandler};
use std::sync::Arc;

let handler: TriggerHandler = Arc::new(move |event, agent| {
    let backend = backend.clone();
    Box::pin(async move { backend.dispatch(event, agent).await })
});

let mut ambient = AmbientAgent::new(agent, source).with_trigger_handler(handler);

start は、ハンドラーなしでは失敗します。

AmbientAgent has no trigger handler, so starting it would log trigger events without ever
invoking the agent. Call `with_trigger_handler` with a closure that drives the agent through a
Runner.

重要: 以前はハンドラーなしで開始でき、その後各トリガーがログに記録されていたため、AmbientAgent::new(..).start() は実行されていないエージェントを実行しているように見えていました。

出力と並行性

生成されたイベントは以前はデバッグレベルでログに記録された後に破棄されていたため、呼び出し元は実行内容や失敗したかどうかを確認できませんでした。また、トリガーも厳密に一度に1つずつ処理されていました。つまり、ループはソースを再度ポーリングする前にハンドラーのイベントストリーム全体を排出していたため、1つの遅いトリガーが後続のすべてのトリガーをブロックしていました。

注: この上限が制御するのは並行性であり、並列性ではありません。ハンドラーは共有のアンビエントタスクを使用するため、スレッドをブロックするハンドラーがあると、依然としてループ全体が停止します。そのような場合は tokio::task::spawn_blocking を使用してください。永続的なトリガーオフセット、デッドレター処理、再試行については、引き続き呼び出し元の責任となります。

アンビエントエージェントは、ユーザーのターンではなくイベントに反応します。EventSource は TriggerEvents を生成し、エージェントはそれぞれに対して実行されます。

ambient 機能が必要です。

[dependencies]
adk-agent = { version = "2.1.0", features = ["ambient"] }

イベントソース

TriggerEvent::principal により、ハンドラーはすべてのイベントを同じように信頼するのではなく、認証済みのトリガーと匿名のトリガーを区別できます。

取りこぼしたティック

CronTrigger::subscribe は、呼び出された時点を基準に次のティックを計算します。そのため、停止時間後に再起動するトリガーや、ホストがサスペンドする環境で実行されるトリガーは、次に来る未来のティックから再開し、その間に発生したすべてのティックは破棄されます。

MissedTickPolicy は、その期間に何が起こるかを決定します。

ポリシーだけでは、1つのサブスクリプション内の空白しか対象にできません。プロセスの再起動をまたぐ空白を検出するには、スケジュールがどこまで進んだかを記録する TickWatermark が必要です。

use std::sync::Arc;
use adk_agent::ambient::{CronTrigger, FileTickWatermark, MissedTickPolicy};

let trigger = CronTrigger::new("0 */5 * * * *")?
    .with_missed_tick_policy(MissedTickPolicy::CoalesceOne)
    .with_watermark(Arc::new(FileTickWatermark::new("/var/lib/my-agent/sweep.tick")));

FileTickWatermark は1つの RFC 3339 カーソルを保存します。固有の同階層一時ファイルに書き込み、同期したうえで、Unix と Windows の両方で保存先をアトミックに置き換えます。その他のバックエンドストアには TickWatermark を実装してください。

再生の上限

頻繁なスケジュールでの All は、長時間の停止後に数千件のティックを未処理のまま残す可能性があります。with_max_catch_up は、1回のパスで再生する件数に上限を設けます（デフォルトは64件）。上限に達すると、ギャップの残りは破棄され、永続カーソルはその先へ進み、トリガーは次の未来のティックから再開します。その際、破棄されたティックの件数がログに記録されます。次の通常のティックが発生する前に再起動しても、破棄された残りは復元されません。

配信契約

ウォーターマークは、コンシューマーが処理を完了したときではなく、トリガーがティックを発行したときに進みます。発行後、完了前にクラッシュすると、その実行は繰り返されずに失われます。これは少なくとも1回ではなく、最大1回の配信です。これにより、ポーリングを停止したコンシューマーが再起動のたびに同じギャップを再生するのを防ぎます。実行中のクラッシュ後も処理を維持する必要があるコンシューマーは、独自の完了状態を記録してください。設定されたウォーターマークを永続化できない場合、cron ストリームは、影響を受けるイベントを発行する前に停止します。この保証が暗黙に弱められることはありません。

Webhook トリガー

到達可能な webhook はアプリケーションロジックへのリモートエントリーポイントであるため、WebhookTrigger はデフォルトでループバックを使用し、検証機能なしに、より広いアドレスでサービスを提供することを拒否します。

ローカル開発

use adk_agent::ambient::WebhookTrigger;

// Binds 127.0.0.1 — reachable only from this host.
let trigger = WebhookTrigger::new(8080, "/webhook");

外部公開リスナーには検証機能が必要

use adk_agent::ambient::{WebhookRequest, WebhookTrigger, WebhookVerifier};
use std::net::SocketAddr;
use std::sync::Arc;

#[derive(Debug)]
struct SharedToken(String);

impl WebhookVerifier for SharedToken {
    fn verify(&self, request: &WebhookRequest<'_>) -> Result<String, String> {
        match request.header("x-webhook-token") {
            Some(value) if value == self.0 => Ok("ci-system".to_string()),
            Some(_) => Err("token mismatch".to_string()),
            None => Err("missing x-webhook-token".to_string()),
        }
    }
}

let address: SocketAddr = "0.0.0.0:8080".parse().unwrap();
let trigger = WebhookTrigger::new(8080, "/webhook")
    .with_bind_address(address)
    .with_verifier(Arc::new(SharedToken(std::env::var("WEBHOOK_TOKEN").unwrap_or_default())));

検証器なしでループバック以外のアドレスにサブスクライブすると、agent.ambient.webhook_unauthenticated で失敗します。このチェックはサブスクライブ時に行われるため、誤りの修正コストがまだ低い段階で検出できます。

重要: 署名方式は受信した正確なバイト列に対して計算されるため、検証器はWebhookRequest::body を介して生のボディを受け取ります。リクエストを解析した形式を信頼する前に検証してください。

リクエスト処理

401 には、認証情報のどの部分が失敗したかについての詳細は含まれません。代わりに理由が記録されるため、エンドポイントを使用して認証情報を探ることはできません。

accept_non_json はデフォルトで無効になっています。文字列としてラップされた不正な本文が、意図的なイベントと区別できないトリガーイベントを生成するためです。

ライフタイム

HTTP リスナーは、subscribe が返すストリームに属します。ストリームを破棄すると、サーバーが正常にシャットダウンされてポートが解放されるため、同じポートを再バインドできます。

let stream = trigger.subscribe().await?;
// ... consume events ...
drop(stream); // the listener stops and the port is free

注: 以前は、リスナーの存続期間がそのコンシューマーより長くなっていました。ストリームを破棄してもサーバーはバインドされたままになり、配信できないリクエストを受け付け続け、同じポートでの再起動に失敗していました。