Ambient-Agenten

Ausführen eines Ambient-Agenten

Ein Trigger-Handler ist erforderlich. Der unterstützte Pfad ist with_invoker, der alles akzeptiert, was adk_core::AgentInvoker implementiert — Runner tut dies:

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 steuert drei Dinge:

PerTrigger ist die Standardeinstellung, da ein Zeitplan, der jede Minute in eine gemeinsame Sitzung auslöst, den Verlauf dieser Sitzung – und die Token-Kosten jedes späteren Laufs – unbegrenzt anwachsen lässt.
Runner serialisiert extern aufgerufene Durchläufe, die auf dieselbe gemeinsame Sitzung zielen, bis jeder Ereignisstrom abgeschlossen ist, während Sitzungen mit unterschiedlichen IDs weiterhin gleichzeitig ausgeführt werden können.

AgentInvoker::invoke erstellt die Sitzung, wenn sie nicht existiert. Runner::run tut dies nicht: Es löst eine bestehende Sitzung auf und liefert andernfalls session.not_found über den Stream, wofür ein extern ausgelöster Lauf keine Möglichkeit zur Vorabregistrierung hat.

Wenn der Aufrufer sein ausführbares Stammverzeichnis bereitstellt, wie es Runner tut, verwendet with_invoker diesen Agenten für Umgebungsprotokollierung und Diagnose. Dadurch wird verhindert, dass die Telemetrie einen Agenten nennt, während tatsächlich ein anderer den Auslöser verarbeitet.

Einen Handler direkt bereitstellen

with_trigger_handler bleibt für Aufrufer verfügbar, die etwas anderes als einen Runner steuern. Es erhält das Ereignis und den Agenten und muss den Ereignisstrom zurückgeben; das Erstellen der Sitzung liegt dann in der Verantwortung des Handlers.

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 schlägt ohne einen Handler fehl:

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.

Wichtig: Das Starten ohne Handler war zuvor erfolgreich und protokollierte anschließend jeden Auslöser, sodass AmbientAgent::new(..).start() den Eindruck erweckte, einen Agenten auszuführen, der nie ausgeführt wurde.

Ausgabe und Nebenläufigkeit

Erzeugte Ereignisse wurden zuvor auf Debug-Ebene protokolliert und verworfen, sodass ein Aufrufer nicht erkennen konnte, was ein Lauf getan hatte oder ob er fehlgeschlagen war. Trigger wurden außerdem strikt nacheinander verarbeitet – die Schleife leerte den gesamten Ereignisstrom eines Handlers, bevor sie die Quelle erneut abfragte –, sodass ein langsamer Trigger alle späteren blockierte.

Hinweis: Die Begrenzung steuert die Nebenläufigkeit, nicht die Parallelität. Handler teilen sich die umgebende Task, sodass ein Handler, der den Thread blockiert, weiterhin die Schleife anhält. Verwenden Sie dafür tokio::task::spawn_blocking. Persistente Trigger-Offsets, die Behandlung nicht zustellbarer Nachrichten und Wiederholungsversuche bleiben in der Verantwortung des Aufrufers.

Umgebende Agenten reagieren auf Ereignisse statt auf eine Benutzereingabe. Ein EventSource erzeugt TriggerEvents, und der Agent wird für jedes davon ausgeführt.

Erfordert das Feature ambient:

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

Ereignisquellen

TriggerEvent::principal ermöglicht es einem Handler, einen autorisierten Trigger von einem anonymen Trigger zu unterscheiden, anstatt jedes Ereignis als gleichermaßen vertrauenswürdig zu behandeln.

Verpasste Ticks

CronTrigger::subscribe berechnet den nächsten Tick ab dem Zeitpunkt, zu dem es aufgerufen wird. Ein Trigger, der nach einer Ausfallzeit neu gestartet wird oder auf einem Host ausgeführt wird, der in den Ruhezustand wechselt, setzt daher beim nächsten zukünftigen Tick fort, und jeder dazwischen fällig gewordene Tick wird verworfen.

MissedTickPolicy legt fest, was mit dieser Zeitspanne geschieht:

Eine Richtlinie allein deckt nur Lücken innerhalb eines Abonnements ab. Um eine Lücke zu erkennen, die sich über einen Prozessneustart erstreckt, wird ein TickWatermark benötigt, das festhält, wo der Zeitplan aufgehört hat:

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 speichert einen RFC-3339-Cursor. Es schreibt über eine eindeutige temporäre Geschwisterdatei, synchronisiert diese und ersetzt das Ziel unter Unix und Windows atomar. Implementieren Sie TickWatermark für andere Speicher.

Begrenzung einer Wiederholung

All bei einem häufigen Zeitplan kann nach einem langen Ausfall Tausende ausstehende Ticks hinterlassen. with_max_catch_up begrenzt, wie viele Ticks ein Durchlauf wiederholt (standardmäßig 64); sobald das Limit erreicht ist, wird der verbleibende Teil der Lücke verworfen, der dauerhafte Cursor darüber hinaus weitergesetzt und der Trigger beim nächsten zukünftigen Tick fortgesetzt, wobei protokolliert wird, wie viele Ticks verworfen wurden. Ein Neustart vor dem nächsten regulären Tick stellt den verworfenen Rest nicht wieder her.

Zustellungsvertrag

Der Wasserstand wird weitergesetzt, wenn der Trigger einen Tick ausgibt, nicht wenn der Verbraucher die Verarbeitung abgeschlossen hat. Ein Absturz zwischen Ausgabe und Abschluss verwirft diese Ausführung, anstatt sie zu wiederholen — höchstens einmal, nicht mindestens einmal. Dadurch wird verhindert, dass ein Verbraucher, der das Abrufen beendet, bei jedem Neustart dieselbe Lücke erneut abspielt. Verbraucher, deren Arbeit einen Absturz während der Verarbeitung überstehen muss, sollten ihren eigenen Abschlussstatus speichern. Wenn ein konfigurierter Wasserstand nicht dauerhaft gespeichert werden kann, hält der Cron-Stream vor der Ausgabe des betroffenen Ereignisses an, anstatt diese Garantie stillschweigend abzuschwächen.

Webhook-Trigger

Ein erreichbarer Webhook ist ein entfernter Einstiegspunkt in die Anwendungslogik, daher verwendet WebhookTrigger standardmäßig die Loopback-Adresse und verweigert es, eine größere Adresse ohne Verifizierer bereitzustellen.

Lokale Entwicklung

use adk_agent::ambient::WebhookTrigger;

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

Nach außen geöffnete Listener erfordern einen Verifizierer

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

Das Abonnieren einer Nicht-Loopback-Adresse ohne Verifizierer schlägt mit agent.ambient.webhook_unauthenticated fehl. Die Prüfung erfolgt zum Zeitpunkt des Abonnierens, wenn der Fehler noch kostengünstig behoben werden kann.

Wichtig: Ein Verifizierer erhält den unformatierten Body über WebhookRequest::body, da Signaturschemata über die exakt empfangenen Bytes berechnet werden. Verifizieren Sie die Anfrage, bevor Sie einer geparsten Form der Anfrage vertrauen.

Anfrageverarbeitung

Ein 401 enthält keine Details darüber, welcher Teil einer Zugangsdaten fehlerhaft war; stattdessen wird der Grund protokolliert, sodass der Endpunkt nicht zum Ausspähen von Zugangsdaten verwendet werden kann.

accept_non_json ist standardmäßig deaktiviert, da ein fehlerhafter, als String verpackter Body ein Trigger-Ereignis erzeugt, das von einem absichtlich ausgelösten Ereignis nicht zu unterscheiden ist.

Lebensdauer

Der HTTP-Listener gehört zum von subscribe zurückgegebenen Stream. Wird der Stream verworfen, fährt der Server ordnungsgemäß herunter und gibt den Port frei, sodass derselbe Port erneut gebunden werden kann:

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

Hinweis: Zuvor bestand der Listener über seinen Consumer hinaus. Das Verwerfen des Streams ließ den Server gebunden und weiterhin Anfragen annehmen, die er nicht zustellen konnte, und ein Neustart am selben Port schlug fehl.