Agenten-Hooks

Agent Hooks ist eine Kernfunktion des Agent-Frameworks, mit der sich Governance- und Laufzeitkontrollen an klar definierten Punkten während der Ausführung eines Agenten anwenden lassen. Es implementiert die framework-neutrale AGENT-HOOKS-0.1-Spezifikation, sodass Policy-Engines, Genehmigungs-Gateways, Budgetkontrollen, Inhaltsfilter und Egress-Kontrollen eine gemeinsame Kontrollschnittstelle ansprechen können.

Important

Agent Hooks ist eine Steuerebene, keine Telemetrieebene. Jeder Interceptor gibt eine Bewertung zurück. Im enforce Modus fungiert das Framework für diese Bewertung. Im evaluate_only Modus zeichnet es die Bewertung auf, ohne die Ausführung zu ändern. Verwenden Sie Observability für passives Tracing, Metriken und Protokolle.

Agent Hooks sind für .NET noch nicht verfügbar. Verwenden Sie Agent-Middleware, Toolgenehmigung und Agentsicherheit, um Laufzeitsteuerelemente zu .NET Agents hinzuzufügen.

Agent Hooks ist experimentell in Python. Bei der ersten Verwendung gibt die Factory ein ExperimentalWarning aus, und ihre API kann sich vor der allgemeinen Verfügbarkeit noch ändern.

Wann Agent-Hooks verwendet werden sollen

Verwenden Sie Agent-Hooks, wenn unabhängig entwickelte Steuerelemente einen freigegebenen, erzwingbaren Vertrag über die Agenteingabe, Modellanrufe, Toolanrufe und die endgültige Ausgabe benötigen.

Fähigkeit Verwenden Sie es für
Agent-Hooks Standardisierte Richtlinienentscheidungen, Transformationen, Genehmigungen, Budgets und Egress-Kontrollen über den gesamten Agent-Lebenszyklus hinweg.
Middleware für Agenten Anwendungsspezifisches Querschneidverhalten, das den Agent Hooks-Vertrag oder seine Kernlaufzeitgarantien nicht benötigt.
Agent-Sicherheit mit FIDES Deterministische Informationsflussbezeichnungen und Richtlinien für nicht vertrauenswürdige oder vertrauliche Inhalte.
Toolgenehmigung Menschliche Bestätigung einzelner Funktionstoolaufrufe.
Beobachtbarkeit Passive Ablaufverfolgungen, Metriken und Protokolle, die die Ausführung nicht steuern.

Was das Agent-Framework erzwingt

Wenn Sie Agent-Hooks zu einem Agent hinzufügen, wendet Agent Framework eine koordinierte Erzwingungsgrenze für Agentausführungen, Modellanrufe und Toolaufrufe an. Die Laufzeit bietet die folgenden Garantien:

  • Fehler geschlossen: Eine Ablehnung blockiert die geschützte Aktion. Ungültige Kontexte, ungültige Entscheidungen, Fehler bei Interzeptoren und Erzwingungsfehler umgehen Kontrollen nicht stillschweigend.
  • Transform-Rückschreiben: Eine Transformation ändert die nativen Nachrichten, Tool-Argumente, Tool-Ergebnisse oder die endgültige Antwort, die bei der Ausführung tatsächlich verwendet werden. Wenn keine Transformation angewendet werden kann, schlägt die Ausführung fehl.
  • Gepuffertes Streaming: Keine Aktualisierung der Antwort kommt beim Aufrufer an, bis die vollständige Modellantwort und die endgültige Ausgabe ihre Abfangpunkte durchlaufen haben.
  • Urteilsabhängige Persistenz: Die Persistenz wartet auf das Urteil, das sie betrifft. Die Standard-Persistenz nach der Ausführung wartet auf output; die Persistenz des Verlaufs pro Dienstaufruf wartet auf jeden post_model_call.
  • Vollständige Bundleinstallation: Die Agent-, Chat- und Funktionsteile werden als eine Einheit installiert, sodass eine unvollständige Erzwingungsgrenze nicht versehentlich konfiguriert werden kann.

Der Vertrag ist kooperativ und nicht eine Prozessisolationsgrenze. Interceptors werden im Hostprozess ausgeführt und erhalten die Inhalte, die erforderlich sind, um Entscheidungen zu treffen. Registrieren Sie nur Interceptors, die Sie als vertrauenswürdig einstufen.

Agent-Hooks installieren

Installieren Sie den optionalen agent-hooks Zusatz für das Basispaket:

pip install "agent-framework-core[agent-hooks]"

Wenn Sie uv verwenden:

uv add "agent-framework-core[agent-hooks]"

Die agent-hooks-sdk Abhängigkeit wird faul importiert. Beim Importieren agent_framework wird das SDK nicht geladen, es sei denn, Sie erstellen ein Agent Hooks Middleware-Bundle.

Note

Der agent-hooks Zusatz ist absichtlich nicht in agent-framework-core[all] enthalten. Installieren Sie sie explizit, wenn Sie diese experimentelle Steueroberfläche aktivieren möchten.

Hinzufügen eines Interceptors

Ein Interceptor empfängt ein agent_hooks.AgentContext (die Kontextzuordnung der Spezifikation, nicht das agent_framework.AgentContext, das von der Agent-Middleware verwendet wird) und gibt ein Urteil zurück. Der folgende Interceptor blockiert die endgültige Ausgabe, die das Wort secret enthält. Es wird angenommen, dass es sich client um einen bereits konfigurierten Agent Framework-Chatclient handelt.

from agent_framework import Agent, create_agent_hooks_middleware
from agent_hooks import ALLOW, AgentContext, InterceptionBlocked, Verdict


class SecretEgressGuard:
    def intercept(self, context: AgentContext) -> Verdict:
        if (
            context["interception_point"] == "output"
            and "secret" in str(context["target"]).lower()
        ):
            return Verdict.deny(
                reason="secret_in_output",
                message="The final response contains restricted content.",
            )
        return ALLOW


hooks = create_agent_hooks_middleware(
    {"secret-egress": SecretEgressGuard()},
)

agent = Agent(
    client=client,
    instructions="You are a helpful assistant.",
    middleware=[hooks],
)

try:
    response = await agent.run("Summarize the account details.")
except InterceptionBlocked as exc:
    print(f"Blocked: {exc.result.verdict.reason}")

Übergeben Sie das Paket als ein Element der middleware Liste des Agenten. Installieren Sie genau ein Agent Hooks Bundle auf jedem Agent.

Abfangpunkte

Das Agent-Framework gibt automatisch die entsprechenden Abfangenpunkte aus:

Abfangenpunkt Wenn sie ausgegeben wird Transformationsziel
agent_startup Vor der ersten Eingabe in einer Agent Hooks-Sitzung Nicht transformierbar
input Wenn eine externe Anfrage beim Agenten eingeht Eingabeinhalt und -rolle
pre_model_call Vor jeder Modellanforderung Nachrichten, die an das Modell gesendet werden
post_model_call Nach jeder vollständigen Modellantwort Antwortinhalte, vom Framework ausgeführte Toolaufrufe und Endgrund
pre_tool_call Vor jedem vom Framework ausgeführten Toolaufruf Toolargumente
post_tool_call Nachdem ein Tool erfolgreich war oder fehlschlägt Ergebnis des Tools
output Bevor die endgültige Antwort den Anrufer erreicht Endgültiger Antwortinhalt
agent_shutdown Wenn die Agent-Hooks-Sitzung abgeschlossen, fehlgeschlagen oder abgebrochen wird Nicht transformierbar

Eine Ausführung, die ein Tool aufruft, gibt in der Regel Folgendes aus:

agent_startup input → → pre_model_callpost_model_callpre_tool_callpost_tool_callpre_model_callpost_model_call → → outputagent_shutdown

Urteile

Der Vertrag hat drei Entscheidungen: allow, , denyund transform. Das Python-SDK bietet außerdem Hilfsfunktionen für Warnungen und aufhebbare Sperren.

Result Python-API Behavior
Allow ALLOW oder Verdict(decision=Decision.ALLOW) Fahren Sie mit dem Ziel unverändert fort.
Zulassen mit Warnung Verdict.warn(...) Fahren Sie fort, und schließen Sie die Warnung in den Abfangendatensatz ein.
Verweigern Verdict.deny(...) Blockieren Sie die geschützte Aktion.
Ausstehende Genehmigung verweigern Verdict.escalate(...) Blockieren, sofern der konfigurierte Genehmigungs-Resolver keine Permit-Entscheidung zurückgibt.
Umwandeln Verdict(decision=Decision.TRANSFORM, transform=Transform(...)) Überschreiben Sie einen Wert unter $target, und fahren Sie dann mit dem überschriebenen Wert fort.

Ausführungsebene und Modellebene verweigern das InterceptionBlocked Auslösen und Verhindern, dass das geschützte Ergebnis den Anrufer oder die nächste Stufe erreicht. An einer Toolschnittstelle verhindert ein Richtlinienverbot die Tool-Aktion oder verwirft deren Ergebnis und gibt einen Steuerungsfehler, der den Grund für die Verweigerung durch die Richtlinie enthält, jedoch ohne die Nutzlast des verweigerten Ziels, an das Modell zurück. Dadurch kann die Agenten-Schleife fortgesetzt werden. Ein Hostfehler oder ein Fehler bei der Durchsetzung stoppt die Ausführung.

Anwenden einer Transformation

Ein Transformationspfad muss bei $target beginnen. Ein Interceptor kann z. B. den endgültigen Antwortinhalt ersetzen:

from agent_hooks import ALLOW, AgentContext, Decision, Transform, Verdict


class OutputRedactor:
    def intercept(self, context: AgentContext) -> Verdict:
        if context["interception_point"] != "output":
            return ALLOW

        return Verdict(
            decision=Decision.TRANSFORM,
            reason="redacted_output",
            transform=Transform(
                path="$target.content",
                value="[Response removed by policy]",
            ),
        )

Transformationen werden auf Agent Framework-Werte Content angewendet, wobei unterstützte Rich-Inhalte beibehalten werden, anstatt jeden Wert auf Nur-Text zu reduzieren. Ein falsch formatierter Pfad oder ein inkompatibler Ersatz schlägt fehl, anstatt mit dem ursprünglichen Wert fortzufahren.

Toolfreigabe und Argumentumwandlungen

Die Freigabe für das Agent-Framework-Tool und die Freigabestelle für Agent Hooks sind zwei getrennte Mechanismen. Für ein Funktionstool mit approval_mode="always_require" erstellt Agent Framework die Anfrage zur menschlichen Genehmigung, bevor die Funktionsmiddleware ausgeführt wird. Eine pre_tool_call Transformation kann daher Argumente ändern, nachdem der Benutzer die ursprünglichen Werte genehmigt hat.

Warning

Transformieren Sie keine Argumente bei pre_tool_call für Tools, die approval_mode="always_require" verwenden. Transformieren Sie den Werkzeugaufruf bei post_model_call so, dass die Genehmigungsanforderung des Frameworks die transformierten Werte enthält, oder geben Sie resolver bei Verdict.escalate(...) zurück und wickeln Sie die Genehmigung über die Agent Hooks pre_tool_call ab.

Streaming und Persistenz

Agent Hooks behält die Streaming-API bei, verwendet jedoch die Semantik gepufferter Ausgabe. Das Agent-Framework setzt die vollständige Modellantwort zusammen, gibt post_model_call aus, setzt die endgültige Agentenantwort zusammen und gibt output aus, bevor Aktualisierungen freigegeben werden. Wenn eines der Punkte die Antwort verweigert, empfängt der Anrufer keine partiellen Aktualisierungen.

Dieses Verhalten tauscht die Token-für-Token-Latenz gegen die Erzwingung einer Fail-Closed-Ausgabe. Eine Ausgabetransformation wirkt sich auch auf die Updates aus, die schließlich an den Aufrufer zurückgegeben werden.

Die Persistenz wird durch den Abfangpunkt gesteuert, der den Persistenzvorgang abdeckt:

  • Standardmäßig warten der Verlauf und andere Provider-Aufgaben nach der Ausführung auf das output Ergebnis. Eine abgelehnte Ausgabe wird nicht gespeichert, und eine Ausgabetransformation wird nach der Transformation gespeichert.
  • Wenn Sie Agent beim client.as_agent(...)-Konstruktor oder bei require_per_service_call_history_persistence=True festlegen, wird jeder Modellaustausch gespeichert, nachdem seine post_model_call-Entscheidung dies erlaubt hat. Eine spätere output Verweigerung macht die bereits zugelassene Historie nicht rückgängig.
  • Bei standardmäßiger Nachlaufpersistenz verbleiben Wiederholungsversuche hinter der endgültigen output Entscheidung. Im Modus für Dienstaufrufe werden stattdessen die einzelnen Modellantworten beibehalten, die übergeben post_model_callwerden.

Important

Wenn Modellinhalte nicht dauerhaft gespeichert werden dürfen, erzwingen Sie diese Richtlinie unter post_model_call, wenn require_per_service_call_history_persistence=True. Eine reine Ausgangsrichtlinie schützt das, was den Aufrufer erreicht, entfernt jedoch nicht rückwirkend Austausche mit dem Modell, die bereits zugelassen und unter post_model_call gespeichert wurden.

Sitzungen und Auditprotokolle

Standardmäßig erstellt jeder Ausgeführte Agent eine Agent Hooks-Sitzung. agent_startup und agent_shutdown markieren den Beginn und das Ende der Ausführung, und Datensätze erhalten eine Sitzungs-ID mit monoton ansteigender Sequenznummer.

Verwenden Sie record_sink, um jede InterceptionRecord zu empfangen:

records = []

hooks = create_agent_hooks_middleware(
    {"secret-egress": SecretEgressGuard()},
    record_sink=records.append,
)

Interzeptionsdatensätze erfassen die Entscheidung, den Grund, die Interzeptorzusammenfassung, den Modus, die Identität und die Sequenz, ohne die abgefangene Nutzlast in den Audit-Datensatz zu kopieren. Der Interceptor selbst erhält weiterhin den vollständigen Kontext.

Mehrere Durchläufe in einer Sitzung

Verwenden Sie create_agent_hooks_middleware_from_emitter(), wenn die Anwendung eine langlebigere Agent-Hooks-Sitzung unterhält, z. B. eine Unterhaltung mit einem Approval-Ledger:

from agent_framework import Agent, create_agent_hooks_middleware_from_emitter
from agent_hooks import AgentContextBuilder, InterceptionEmitter


emitter = InterceptionEmitter().register(SecretEgressGuard())
builder = AgentContextBuilder(
    agent_id="support-agent",
    framework="agent-framework",
    session_id="conversation-42",
)

hooks = create_agent_hooks_middleware_from_emitter(emitter, builder)
agent = Agent(client=client, middleware=[hooks])

await emitter.emit(builder.agent_startup(tools_registered=[]))
await agent.run("First turn")
await agent.run("Second turn")
await emitter.emit(builder.agent_shutdown(reason="completed"))

In dieser Form konfiguriert die Anwendung den Emitter und übernimmt den Startvorgang, das Herunterfahren und die Fehlerbehandlung. Die Middleware gibt die Punkte pro Ausführung von input bis output aus.

Konfigurieren der Erzwingung

create_agent_hooks_middleware() akzeptiert die folgenden Steuerelemente:

Parameter Purpose
interceptors Eine Abfolge von Interzeptoren oder eine Zuordnung von Namen zu Interzeptoren. Mindestens eines ist erforderlich.
resolver Löst aufhebbare Ablehnungen über einen Genehmigungsworkflow auf. Ohne einen Resolver bleibt die Ablehnung wirksam.
mode "enforce" wendet Urteile an. "evaluate_only" zeichnet auf, was passiert, aber jede Aktion zulässt.
composition Wählt aus, wie mehrere Interceptor-Entscheidungen kombiniert werden.
identity_provider Erzeugt inhaltsgebundene Kontextidentitäten. Der Standardwert lautet "jcs-sha256".
timeout Timeout pro Interceptor und Resolver für wartende Anrufe. Der Standardwert ist fünf Sekunden. Ein synchroner Interceptor oder Resolver, der die Event-Loop blockiert, kann durch dieses Timeout nicht unterbrochen werden.
record_sink Empfängt jeden nutzlastfreien Abfangen-Datensatz.

Die Standardkomposition ist sequenziell first_deny , wobei die Genehmigung so konfiguriert ist, dass der Fold gestoppt wird. Die Reihenfolge der Interceptoren ist daher wichtig: Ordnen Sie Kontrollmechanismen, die immer ausgeführt werden müssen, vor solchen an, die eine Genehmigung anfordern können. Sehen Sie sich die Prüfliste für die Agent Hooks-Produktion an, bevor Sie ein anderes Kompositionsprofil auswählen.

Führen Sie das Rollout im Modus „Nur auswerten“ durch.

Verwenden Sie evaluate_only, um das Richtlinienverhalten vor der Erzwingung zu messen:

hooks = create_agent_hooks_middleware(
    {"secret-egress": SecretEgressGuard()},
    mode="evaluate_only",
    record_sink=records.append,
)

In diesem Modus werden Interzeptoren ausgeführt, und Datensätze enthalten deren Bewertungen, aber keine Aktion wird blockiert oder umgewandelt. Beschreiben Sie eine evaluate_only-Bereitstellung nicht als verbindliche Governance.

Kompositionsregeln

Platzieren Sie das Bündel zuerst in der Middlewareliste des Agents, sodass es die äußerste Erzwingungsgrenze bildet:

agent = Agent(
    client=client,
    middleware=[
        create_agent_hooks_middleware([SecretEgressGuard()]),
        application_middleware,
    ],
)

Folgen Sie diesen Regeln:

  • Installieren Sie genau ein Agent Hooks Bundle pro Agent. Gestapelte Bündel werden abgelehnt.
  • Halten Sie das Bündel intakt. Die Middleware für Agent, Chat und Funktionen kann nicht separat installiert werden.
  • Installieren Sie das Bundle auf Agent, nicht direkt auf einem Chatclient oder über einen Kontextanbieter.
  • Middleware, die vor dem Bundle platziert ist, befindet sich außerhalb der Durchsetzungsgrenze. Behandeln Sie die äußere Position als äußeres Vertrauen.
  • Weisen Sie jedem geschachtelten Agenten ein eigenes Bündel zu, wenn auch die Aktivität seines internen Modells und seiner Tools abgefangen werden muss.

Aktuelle Einschränkungen

  • Python nur: Agent-Hooks sind noch nicht in den .NET oder Go SDKs implementiert.
  • Experimentelle API: Factory-Signaturen und Verhalten können sich vor der allgemeinen Verfügbarkeit ändern.
  • Gepuffertes Streaming: Aktualisierungen werden nicht Token für Token ausgegeben, da die Ausgabe vollständig vorliegen muss, bevor eine Fail-Closed-Entscheidung getroffen wird.
  • Gehostete Tools: Tools, die von einem Modellanbieter ausgeführt werden, laufen nicht über die Schnittstelle für Funktionsaufrufe des Agent Frameworks. Ihre Aufrufe und Ausgaben werden in post_model_call angezeigt, aber pre_tool_call und post_tool_call können die serverseitige Ausführung des Anbieters nicht blockieren.
  • Kooperative Abgrenzung: Agent Hooks isoliert Interceptors nicht in einer Sandbox und schützt nicht vor einem feindlichen Host. Codepfade, die die geschützte Agentpipeline umgehen, sind nicht abgedeckt.
  • Die Verfügbarkeit von Interceptor wirkt sich auf die Verfügbarkeit des Agents aus: Im Erzwingungsmodus blockiert ein Interceptor-Fehler oder Timeout die überwachte Aktion standardmäßig.

Informationen zum Produktionsrollout, zu Fehlerursachen und zur Alarmierung finden Sie im Operations-Runbook für Agent Hooks.

Agent Hooks ist für Go noch nicht verfügbar. Verwenden Sie Agent-Middleware, Tool-Freigabe und Agentensicherheit, um Go-Agenten Laufzeitkontrollen hinzuzufügen.

Nächste Schritte