Crochets d’agent

Agent Hooks est une fonctionnalité d’Infrastructure d’agent de première classe permettant d’appliquer des contrôles de gouvernance et d’exécution à des points bien définis dans l’exécution d’un agent. Il implémente le contrat AGENT-HOOKS-0.1 neutre pour l’infrastructure, de sorte que les moteurs de stratégie, les passerelles d’approbation, les gardes budgétaires, les filtres de contenu et les contrôles de sortie peuvent cibler une surface de contrôle commune.

Important

Agent Hooks est un plan de contrôle, et non un plan de télémétrie. Chaque intercepteur retourne un verdict. En enforce mode, le framework agit sur ce verdict ; en evaluate_only mode, il enregistre le verdict sans modifier l’exécution. Utilisez l’observabilité pour le suivi passif, les métriques et les journaux.

Les hooks d'agent ne sont pas encore disponibles pour .NET. Utilisez l’intergiciel de l’agent, l’approbation des outils et la sécurité des agents pour ajouter des contrôles d’exécution aux agents .NET.

Agent Hooks est expérimental dans Python. La fabrique émet une ExperimentalWarning fois utilisée pour la première fois, et son API peut changer avant la disponibilité générale.

Quand utiliser des hooks d’agent

Utilisez les hooks d’agent lorsque des contrôles développés indépendamment ont besoin d’un contrat partagé et applicable entre l’entrée de l’agent, les appels de modèle, les appels d’outils et la sortie finale.

Capability Utilisez-le pour
Crochets d’agent Décisions de stratégie standardisées, transformations, approbations, budgets et contrôles de sortie dans le cycle de vie de l’agent.
Intergiciel agent Comportement croisé spécifique à l’application qui n’a pas besoin du contrat Hooks agent ou de ses garanties d’exécution principales.
Sécurité de l’agent avec FIDES Étiquettes de flux d’informations déterministes et stratégies pour le contenu non approuvé ou confidentiel.
Approbation de l’outil Confirmation humaine des appels d’outil de fonction individuels.
Observabilité Traces passives, métriques et journaux qui ne contrôlent pas l’exécution.

Application de l’infrastructure de l’agent

Lorsque vous ajoutez des hooks d’agent à un agent, Agent Framework applique une limite d’application coordonnée entre les exécutions de l’agent, les appels de modèle et les appels d’outil. Le runtime fournit les garanties suivantes :

  • Échec fermé : Un refus bloque l’action protégée. Les contextes non valides, les verdicts non valides, les échecs d’intercepteur et les échecs d’application ne contournent pas silencieusement les contrôles.
  • Transformer l’écriture différée : Une transformation modifie les messages natifs, les arguments de l’outil, les résultats de l’outil ou la réponse finale que l’exécution utilise réellement. Si une transformation ne peut pas être appliquée, l’exécution échoue.
  • Diffusion en mémoire tampon : Aucune mise à jour de réponse n’atteint l’appelant jusqu’à ce que la réponse de modèle complète et la sortie finale passent leurs points d’interception.
  • Persistance contrôlée par le verdict : La persistance attend le verdict qui le couvre. La persistance standard après exécution attend output; la persistance de l’historique des appels par service attend chaque post_model_callfois .
  • Installation complète de l’offre groupée : Les composants agent, conversation et fonction sont installés en tant qu’unité, de sorte qu’une limite d’application incomplète ne peut pas être configurée accidentellement.

Le contrat est coopératif plutôt qu’une limite d’isolation de processus. Les intercepteurs s’exécutent dans le processus hôte et reçoivent le contenu nécessaire pour prendre des décisions. Inscrivez uniquement les intercepteurs que vous approuvez.

Installer les hooks de l’agent

Installez l’option supplémentaire facultative agent-hooks pour le package principal :

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

Si vous utilisez uv:

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

La agent-hooks-sdk dépendance est importée paresseux. agent_framework L’importation ne charge pas le Kit de développement logiciel (SDK), sauf si vous créez un bundle d’intergiciels Agent Hooks.

Note

L’extra agent-hooks n’est intentionnellement pas inclus dans agent-framework-core[all]. Installez-la explicitement lorsque vous souhaitez activer cette surface de contrôle expérimentale.

Ajouter un intercepteur

Un intercepteur reçoit un agent_hooks.AgentContext (mappage de contexte de la spécification, et non l’intergiciel de l’agent agent_framework.AgentContext ) et retourne un verdict. L’intercepteur suivant bloque la sortie finale contenant le mot secret. L’exemple suppose qu’il s’agit client d’un client de conversation Agent Framework déjà configuré.

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

Passez le bundle en tant qu’élément de la liste de l’agent middleware . Installez exactement un bundle Agent Hooks sur chaque agent.

Points d’interception

Agent Framework émet automatiquement les points d’interception applicables :

Point d’interception Quand elle est émise Transformer la cible
agent_startup Avant la première entrée dans une session Agent Hooks Non transformable
input Lorsqu’une demande externe entre l’agent Contenu et rôle d’entrée
pre_model_call Avant chaque demande de modèle Messages envoyés au modèle
post_model_call Après chaque réponse complète du modèle Contenu de réponse, appels d’outils exécutés par l’infrastructure et raison de fin
pre_tool_call Avant chaque appel d’outil exécuté par l’infrastructure Arguments de l’outil
post_tool_call Une fois qu’un outil réussit ou échoue Résultat de l’outil
output Avant que la réponse finale atteigne l’appelant Contenu de la réponse finale
agent_shutdown Une fois la session Agent Hooks terminée, échoue ou est annulée Non transformable

Une exécution qui appelle un outil émet généralement :

agent_startup input → → pre_model_callpost_model_callpre_tool_callpost_tool_callpre_model_callpost_model_call → → outputagent_shutdown

Verdicts

Le contrat a trois décisions : allow, denyet transform. Le sdk Python fournit également des assistances pour les avertissements et les refuses pouvant être levées.

Result API Python Comportement
Permettre ALLOW ou Verdict(decision=Decision.ALLOW) Continuez avec la cible inchangée.
Autoriser avec avertissement Verdict.warn(...) Poursuivez et incluez l’avertissement dans l’enregistrement d’interception.
Deny Verdict.deny(...) Bloquez l’action protégée.
Refuser l’approbation en attente Verdict.escalate(...) Bloquer, sauf si le programme de résolution d’approbation configuré retourne un verdict d’autorisation.
Transform Verdict(decision=Decision.TRANSFORM, transform=Transform(...)) Réécrire une valeur sous $target, puis continuer avec la valeur réécrite.

Le niveau d’exécution et le niveau du modèle refuse l’augmentation InterceptionBlocked et empêche le résultat protégé d’atteindre l’appelant ou la phase suivante. Sur une couture d’outil, un refus de stratégie empêche l’action de l’outil ou ignore son résultat et retourne une erreur de contrôle contenant la raison de la stratégie, sans la charge utile cible refusée, au modèle. Cela permet à la boucle de l’agent de continuer. Un hôte ou un échec d’application interrompt l’exécution.

Appliquer une transformation

Un chemin de transformation doit commencer à $target. Par exemple, un intercepteur peut remplacer le contenu de la réponse finale :

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

Les transformations sont appliquées aux valeurs agent Framework Content , en préservant le contenu enrichi pris en charge plutôt que de réduire chaque valeur en texte brut. Un chemin d’accès incorrect ou un remplacement incompatible échoue au lieu de poursuivre avec la valeur d’origine.

Transformations d’approbation et d’argument des outils

L’approbation de l’outil Agent Framework et le seam d’approbation d’Agent Hooks sont des mécanismes distincts. Pour un outil de fonction avec approval_mode="always_require", Agent Framework crée la demande d’approbation humaine avant l’exécution du middleware de fonction. Une pre_tool_call transformation peut donc modifier les arguments après que l’utilisateur a approuvé les valeurs d’origine.

Avertissement

Ne transformez pas d’arguments pour pre_tool_call les outils qui utilisent approval_mode="always_require". Transformez l’appel de l’outil pour post_model_call que la demande d’approbation de l’infrastructure contienne les valeurs transformées, ou retournez Verdict.escalate(...) et pre_tool_call résolvez l’approbation par le biais des hooks resolverde l’agent.

Diffusion en continu et persistance

Les hooks d’agent conservent l’API de diffusion en continu, mais utilisent la sémantique de sortie mise en mémoire tampon. Agent Framework assemble la réponse complète du modèle, émet post_model_call, assemble la réponse finale de l’agent et émet output avant de publier les mises à jour. Si l’un ou l’autre point refuse la réponse, l’appelant ne reçoit aucune mise à jour partielle.

Ce comportement échange la latence de jeton par jeton pour l’application de sortie non fermée. Une transformation de sortie est également reflétée dans les mises à jour finalement publiées sur l’appelant.

La persistance est contrôlée par le point d’interception qui couvre l’opération de persistance :

  • Par défaut, l’historique et d’autres fournisseurs après exécution attendent le output verdict. Une sortie refusée n’est pas conservée et une transformation de sortie est conservée après la transformation.
  • Lorsque vous définissez require_per_service_call_history_persistence=True sur le Agent constructeur ou client.as_agent(...), chaque échange de modèle est conservé après son post_model_call verdict l’autorise. Un refus ultérieur output ne restaure pas l’historique déjà autorisé.
  • Pour la persistance après exécution par défaut, les nouvelles tentatives restent derrière la décision finale output . Le mode d’appel par service conserve à la place chaque réponse de modèle qui passe post_model_call.

Important

Si le contenu du modèle ne doit pas devenir durable, appliquez cette stratégie au moment require_per_service_call_history_persistence=Truepost_model_call . Une stratégie de sortie uniquement protège ce qui atteint l’appelant, mais elle ne supprime pas rétroactivement les échanges de modèles déjà autorisés et conservés à post_model_call.

Sessions et enregistrements d’audit

Par défaut, chaque agent exécute une session Agent Hooks. agent_startup et agent_shutdown crocheter l’exécution, et les enregistrements reçoivent un ID de session avec une séquence monotoniquement croissante.

Permet record_sink de recevoir chaque InterceptionRecord:

records = []

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

Les enregistrements d’interception capturent la décision, la raison, le résumé de l’intercepteur, le mode, l’identité et la séquence sans copier la charge utile interceptée dans l’enregistrement d’audit. L’intercepteur lui-même reçoit toujours le contexte complet.

Étendre plusieurs exécutions avec une session

Utilisez create_agent_hooks_middleware_from_emitter() quand l’application possède une session de hooks d’agent de longue durée, telle qu’une conversation avec un registre d’approbation :

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

Dans ce formulaire, l’application configure l’émetteur et possède le démarrage, l’arrêt et le nettoyage des erreurs. L’intergiciel émet les points par exécution à partir de inputoutput.

Configurer l’application

create_agent_hooks_middleware() accepte les contrôles suivants :

Paramètre Purpose
interceptors Séquence d’intercepteurs ou mappage de nom à intercepteur. Un élément minimum est nécessaire.
resolver Résout les refuser par le biais d’un canal d’approbation. Sans programme de résolution, le refus reste en vigueur.
mode "enforce" applique les verdicts. "evaluate_only" enregistre ce qui se passerait, mais autorise chaque action.
composition Sélectionne la façon dont plusieurs verdicts d’intercepteur sont combinés.
identity_provider Produit des identités de contexte liées au contenu. La valeur par défaut est "jcs-sha256".
timeout Délai d’expiration par intercepteur et programme de résolution pour les appels attendus. La valeur par défaut est de cinq secondes. Un intercepteur synchrone ou un programme de résolution qui bloque la boucle d’événement ne peut pas être préempté par ce délai d’expiration.
record_sink Reçoit chaque enregistrement d’interception sans charge utile.

La composition par défaut est séquentielle first_deny avec l’approbation configurée pour arrêter le pliage. L’ordre d’intercepteur importe donc : placez les contrôles qui doivent toujours s’exécuter avant les contrôles pouvant demander l’approbation. Consultez la liste de contrôle de production Agent Hooks avant de sélectionner un autre profil de composition.

Déploiement avec le mode d’évaluation uniquement

Permet evaluate_only de mesurer le comportement de stratégie avant l’application :

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

Dans ce mode, les intercepteurs s’exécutent et enregistrent leurs verdicts, mais aucune action n’est bloquée ou transformée. Ne décrivez pas un evaluate_only déploiement comme gouvernance appliquée.

Règles de composition

Placez d’abord l’offre groupée dans la liste des intergiciels de l’agent afin qu’elle forme la limite d’application la plus externe :

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

Suivez ces règles :

  • Installez exactement un ensemble de hooks d’agent par agent. Les bundles empilés sont rejetés.
  • Conservez l’offre groupée intacte. Son agent, sa conversation et son intergiciel de fonction ne peuvent pas être installés séparément.
  • Installez l’offre groupée sur Agent, pas directement sur un client de conversation ou via un fournisseur de contexte.
  • Intergiciel placé avant que le bundle ne soit en dehors de la limite d’application. Traitez la position externe comme une confiance externe.
  • Donnez à chaque agent imbriqué son propre bundle lorsque son modèle interne et son activité d’outil ont également besoin d’interception.

Limitations actuelles

  • Python uniquement : Les hooks d'agent ne sont pas encore implémentés dans les kits sdk .NET ou Go.
  • API expérimentale : Les signatures et le comportement de fabrique peuvent changer avant la disponibilité générale.
  • Diffusion en mémoire tampon : Les mises à jour ne sont pas publiées par jeton, car la sortie doit être terminée avant un verdict fermé en échec.
  • Outils hébergés : Les outils exécutés par un fournisseur de modèles ne passent pas par le biais du seam d’appel de fonction d’Agent Framework. Leurs appels et sorties sont exposés, post_model_callmais pre_tool_callpost_tool_call ne peuvent pas bloquer l’exécution côté serveur du fournisseur.
  • Limite coopérative : Agent Hooks ne protège pas les intercepteurs de bac à sable ni contre un hôte hostile. Les chemins de code qui contournent le pipeline de l’agent protégé ne sont pas couverts.
  • La disponibilité de l’intercepteur affecte la disponibilité de l’agent : En mode d’application, un échec ou un délai d’expiration d’intercepteur bloque l’action protégée par conception.

Pour connaître le déploiement de production, les raisons de l’échec et les instructions d’alerte, consultez le runbook des opérations De hooks agent.

Les hooks d’agent ne sont pas encore disponibles pour Go. Utilisez l’intergiciel de l’agent, l’approbation des outils et la sécurité des agents pour ajouter des contrôles d’exécution aux agents Go.

Étapes suivantes