Agent Hooks

Agent Hooks 是 Agent Framework 原生支援的一項能力,可在代理程式執行流程中明確定義的節點套用治理機制與執行階段控制。 它實作了框架中立的 AGENT-HOOKS-0.1 合約,因此政策引擎、核准閘道、預算守護、內容過濾器及出口控制都能針對一個共同的控制面。

Important

Agent Hooks 是控制平面,不是遙測平面。 每個攔截器都會返回判定結果。 在 enforce 模式下,框架會根據該判決採取行動;在 evaluate_only 模式下,它記錄判決而不改變執行。 使用 可觀察 性來進行被動追蹤、指標和日誌。

Agent Hooks 尚未在 .NET 上架。 利用代理中介軟體工具審核代理安全,為 .NET 代理加入執行時控制。

Agent Hooks 在 Python 中是實驗性的。 工廠在首次使用時會發出一個 ExperimentalWarning,且其 API 可能會在正式推出前發生變更。

何時使用 Agent Hooks

當獨立開發的控制項需要一個共享且可執行的合約,涵蓋代理輸入、模型呼叫、工具呼叫及最終輸出時,請使用代理鉤子。

能力 用它來做
特工胡克斯 標準化政策決策、轉換、核准、預算及退出控制,涵蓋代理生命週期。
代理中介軟體 應用程式專屬的跨領域行為,不需要 Agent Hooks 合約或其核心執行時保證。
FIDES 的特務安全 針對不可信或機密內容的確定性資訊流標籤與政策。
工具核准 人工確認個別函式工具呼叫。
可觀察性 不控制執行的被動追蹤資料、指標和日誌。

代理框架所強制執行的內容

當你為代理加入代理鉤子時,代理框架會在代理執行、模型呼叫和工具呼叫間套用協調執行邊界。 執行時提供以下保證:

  • 失敗已結: 拒絕會阻擋防守動作。 無效情境、無效判決、攔截器失效及執行失敗,並不會悄然繞過控制。
  • 轉換寫回: 轉換會改變執行實際使用的原生訊息、工具參數、工具結果或最終回應。 如果無法套用轉換,該執行程序會以封閉模式失敗。
  • 緩衝串流: 直到完整的模型回應與最終輸出通過攔截點前,呼叫者不會收到回應更新。
  • 受裁定控制的持久化: 持久化會等待涵蓋它的裁定。 標準的執行後持續儲存會等待 output;每次服務呼叫的歷史記錄持續儲存則會等待每個 post_model_call
  • 完整套裝安裝: 代理、聊天和功能部分是安裝成一個整體,因此不完整的執行邊界不會被誤設定。

合約是合作性質,而非流程隔離界線。 攔截器在主機程序中運行,接收決策所需的內容。 只註冊你信任的攔截器。

安裝代理掛鉤

安裝核心套件的選配 agent-hooks 附加功能:

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

如果您使用 uv

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

agent-hooks-sdk 相依性會以延遲方式匯入。 匯入 agent_framework 時不會載入 SDK,除非你建立 Agent Hooks 中介軟體套件。

Note

agent-hooks 額外項目特意未納入 agent-framework-core[all]。 當你想啟用這個實驗控制表面時,請明確安裝它。

加裝攔截機

攔截器會接收 agent_hooks.AgentContext(即規格中的上下文對應,而非代理中介軟體所使用的 agent_framework.AgentContext),並回傳判定結果。 以下攔截器會阻擋包含字詞 secret 的最終輸出。 這個範例假設 client 是已設定好的 Agent Framework 聊天用戶端。

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

將該套件作為代理程式 middleware 清單中的其中一個元素傳遞。 在每個代理程式上恰好安裝一個 Agent Hooks 套件組。

攔截點

代理框架會自動發出適用的攔截點:

攔截點 當它被發射時 轉換目標
agent_startup 在 Agent Hooks 會話的首次輸入之前 不可變形
input 當外部請求進入代理程式時 輸入內容與角色
pre_model_call 在每次模型請求之前 發送給模型的訊息
post_model_call 每次模型完成回應後 回應內容、框架執行的工具呼叫與完成原因
pre_tool_call 在每個框架執行的工具調用前 工具引數
post_tool_call 工具成功或失敗後 工具結果
output 在最終回應到達來電者之前 最終回應內容
agent_shutdown 當 Agent Hooks 會話完成、失敗或被取消時 不可變形

一次呼叫工具的執行通常會產生:

agent_startuppost_tool_callpre_tool_callpre_model_callpost_model_calloutputpost_model_callpre_model_callinputagent_shutdown

評決

合約有三個決定:allowdenytransform。 Python SDK 也提供用於警告和可解除的拒絕項目的輔助函式。

Result Python API Behavior
允許 ALLOWVerdict(decision=Decision.ALLOW) 在目標維持不變的情況下繼續。
允許並警告 Verdict.warn(...) 繼續執行,並將警告包含在攔截紀錄中。
拒絕 Verdict.deny(...) 封鎖受保護的動作。
待審核後拒絕 Verdict.escalate(...) 除非已設定的核准解析器回傳許可判決,否則就封鎖。
轉換 Verdict(decision=Decision.TRANSFORM, transform=Transform(...)) $target 下覆寫一個值,然後使用覆寫後的值繼續。

執行層級和模型層級的拒絕動作會引發 InterceptionBlocked,並防止受保護的結果傳遞到呼叫端或下一個階段。 在工具銜接點,政策拒絕會阻止工具動作或捨棄其結果,並將包含政策原因、但不含遭拒目標有效負載的控制錯誤回傳給模型。 這讓代理迴圈得以繼續。 主機或強制執行失敗會使執行停止。

套用變換

轉換路徑必須從 $target開始。 例如,攔截器可以替換最終反應內容:

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

轉換會套用到 Agent Framework Content 的值,保留支援的豐富內容,而不是將所有值簡化為純文字。 格式錯誤的路徑或不相容的替換值會以封閉方式失敗,而不是繼續使用原始值。

工具認可與參數轉換

Agent Framework 工具核准與 Agent Hooks 核准介面是兩個獨立的機制。 對於具有 approval_mode="always_require" 的函式工具,Agent Framework 會在函式中介軟體執行之前建立人工核准要求。 pre_tool_call因此,變換可以在使用者批准原始值後更改參數。

Warning

對於使用 approval_mode="always_require" 的工具,不要在 pre_tool_call 轉換參數。 將位於 post_model_call 的工具呼叫進行轉換,使框架核准請求包含轉換後的值;或者在 pre_tool_call 回傳 Verdict.escalate(...),並透過 Agent Hooks resolver 完成核准流程。

串流與持久性

Agent Hooks 保留了串流 API,但使用緩衝輸出語意。 代理框架會組裝完整的模型回應、輸出 post_model_call、組裝代理程式的最終回應,並在發布任何更新之前輸出 output。 若任一點拒絕回應,呼叫者將不會收到部分更新。

此行為以代幣間延遲換取失敗封閉輸出強制執行。 輸出轉換也會反映在最終提供給呼叫端的更新內容中。

持久性由涵蓋持久性操作的攔截點所限制:

  • 預設情況下,歷史記錄和其他執行後的提供者工作會等待 output 判定結果。 被拒絕的輸出不會被持久化,而輸出轉換則在轉換後被持續存在。
  • 當你在 Agent 建構函式或 client.as_agent(...) 上設定 require_per_service_call_history_persistence=True 時,每次模型交換都會在其 post_model_call 判定允許後保存。 後來 output 的拒絕並不會推翻已經被允許的歷史。
  • 對於預設的執行後保留機制,重試嘗試仍取決於最終的 output 決定。 每次服務呼叫模式則會持久保存每個通過 post_model_call 的模型回應。

Important

如果模型內容不得持久保存,則在 post_model_callrequire_per_service_call_history_persistence=True 時強制執行該政策。 僅限輸出的出口政策可保護傳送給呼叫端的內容,但無法追溯移除已獲允許並儲存在 post_model_call 的模型互動內容。

會議與審計紀錄

預設會在每次代理程式執行時建立一個 Agent Hooks 工作階段。 agent_startupagent_shutdown 會界定此次執行的前後範圍,而每筆記錄都會收到一個工作階段 ID 和一個單調遞增的序號。

使用 record_sink 來接收每個 InterceptionRecord

records = []

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

攔截紀錄可捕捉決策、理由、攔截者摘要、模式、身份及序列,且不會將攔截的有效載荷複製到稽核記錄中。 攔截機本身仍能接收完整的上下文。

以單一工作階段涵蓋多次執行

當應用程式維持較長生命週期的 Agent Hooks 工作階段時(例如與單一核准記錄帳本進行對話時),請使用 create_agent_hooks_middleware_from_emitter()

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

在此形式中,應用程式負責設定發射器,並負責啟動、關機及錯誤清理。 中介軟體會輸出從 inputoutput 的每次執行點。

設定強制執行

create_agent_hooks_middleware() 接受以下控制:

Parameter Purpose
interceptors 一連串攔截器或名稱到攔截器映射。 至少需要一個。
resolver 透過核准流程處理可解除的拒絕項目。 沒有解決器,拒絕權仍然有效。
mode "enforce" 套用裁決。 "evaluate_only" 記錄會發生什麼,但允許所有行動。
composition 選擇如何合併多個攔截器的判定結果。
identity_provider 產生與內容綁定的情境識別。 預設值為 "jcs-sha256"
timeout 每個攔截器和解析器的可等待呼叫逾時設定。 預設是五秒鐘。 阻擋事件迴圈的同步攔截器或解析器無法被此逾時取代。
record_sink 接收每筆不含承載資料的攔截記錄。

預設的組合是順序的 first_deny ,並設定了批准以停止摺疊。 因此,攔截器的順序很重要:必須一律先執行的控制項,應放在可以請求核准的控制項之前。 在選擇其他組合設定檔之前,請先參閱 Agent Hooks 的正式環境檢查清單

以純評估模式部署

用於 evaluate_only 在執行前衡量政策行為:

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

在此模式下,攔截器會執行,記錄中也會包含其判定結果,但不會封鎖或轉換任何動作。 不要把部署 evaluate_only 描述為強制治理。

組合規則

將此套件放在代理程式的中介軟體清單最前面,使其形成最外層的強制執行邊界:

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

請遵循以下規則:

  • 每個代理只安裝一個 Agent Hooks 套件。 堆疊的組合會被拒絕。
  • 保持包裹完整。 它的代理程式、聊天軟體和功能中介軟體無法分開安裝。
  • 請將套件組安裝在 Agent 上,而不要直接安裝在聊天用戶端上,也不要透過內容提供者安裝。
  • 放在 bundle 前的中介軟體則不在執行邊界之外。 將外部位置視為外部信任。
  • 當每個巢狀代理的內部模型和工具活動也需要攔截時,給每個巢狀代理自己的套件。

目前的限制

  • 僅限 Python:Agent Hooks 尚未在 .NET 或 Go SDK 中實作。
  • 實驗性 API: 工廠簽名和行為在正式上市前可能會改變。
  • 緩衝串流: 更新不會逐詞元釋出,因為必須先完成輸出,才能做出失敗即封閉的判定。
  • 代管工具: 由模型供應商執行的工具不會經過 Agent Framework 的函式呼叫介面。 他們的呼叫和輸出顯示在 post_model_call,但pre_tool_callpost_tool_call無法阻擋提供者的伺服器端執行。
  • 協作邊界:Agent Hooks 不會將攔截器隔離於沙箱中,也不會防範惡意主機。 繞過受保護代理管線的程式碼路徑則不被涵蓋。
  • 攔截器的可用性會影響代理程式的可用性: 在強制模式下,攔截器發生故障或逾時時,依設計會阻止受保護的操作。

關於生產部署、故障原因及警示指引,請參閱 Agent Hooks 作業手冊

Agent Hooks 目前尚未支援 Go。 利用 代理中介軟體工具審核代理安全 ,為 Go 代理加入執行時控制。

下一步