如何在5分钟内为您的AI Agent添加Kill Switch

发布: (2026年2月22日 GMT+8 06:42)
6 分钟阅读
原文: Dev.to

看起来您只提供了来源链接,而没有贴出需要翻译的正文内容。请把您想要翻译的文本(包括任何 Markdown 格式)粘贴在这里,我就可以按照要求为您翻译成简体中文。

介绍

您的 AI 代理正在生产环境中运行,调用 API、做出决策并消耗资金。如果它出现异常——陷入循环、产生幻觉式的工具调用,或耗尽您的 API 预算——唯一的办法就是手动终止进程。本教程在五分钟内为您添加真正的紧急停止开关,且无需更改代理的核心代码。

一个反向代理位于您的代理与 LLM 提供商之间。所有请求都通过它传递,策略以 YAML 定义,当策略触发时,请求将在到达模型之前被阻止。

前置条件

  • 已安装 Docker 和 Docker Compose
  • OpenAI API 密钥(或任何兼容 OpenAI 的提供商)
  • 使用 OpenAI API 格式的 AI 代理

设置

git clone https://github.com/airblackbox/air-platform.git
cd air-platform
cp .env.example .env

编辑 .env 并添加你的 API 密钥:

OPENAI_API_KEY=sk-your-key-here

启动平台:

make up

六个服务大约在 8 秒内启动。重要的服务是运行在 http://localhost:8080Gateway

将你的 Agent 指向 Gateway

Python (OpenAI SDK)

from openai import OpenAI

# 之前 — 直接调用 OpenAI
# client = OpenAI()

# 之后 — 通过 AIR Blackbox Gateway 调用
client = OpenAI(base_url="http://localhost:8080/v1")

LangChain

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    openai_api_base="http://localhost:8080/v1",
    model="gpt-4o"
)

CrewAI

import os
os.environ["OPENAI_API_BASE"] = "http://localhost:8080/v1"
# CrewAI 会自动读取该环境变量

cURL

curl http://localhost:8080/v1/chat/completions \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "gpt-4o",
    "messages": [{"role": "user", "content": "Hello"}]
  }'

你的 Agent 的工作方式完全相同,只是现在所有调用都经过 Gateway。

定义 Kill‑Switch 策略

编辑 config/policies.yaml。下面是一个涵盖常见故障模式的入门策略:

policies:
  # Kill switch: stop runaway loops
  - name: loop-detector
    description: "Kill agent if it makes more than 50 requests in 60 seconds"
    trigger:
      type: rate-limit
      max_requests: 50
      window_seconds: 60
    action: block
    alert: true

  # Kill switch: budget cap
  - name: budget-cap
    description: "Kill agent if it exceeds $5 in token spend"
    trigger:
      type: cost-limit
      max_cost_usd: 5.00
    action: block
    alert: true

  # Kill switch: restrict dangerous tools
  - name: tool-restriction
    description: "Block agent from executing shell commands"
    trigger:
      type: tool-call
      blocked_tools:
        - "execute_command"
        - "run_shell"
        - "delete_file"
    action: block
    alert: true

  # Risk tiers: require human approval for high‑risk actions
  - name: high-risk-gate
    description: "Flag requests that involve payments or external APIs"
    trigger:
      type: content-match
      patterns:
        - "payment"
        - "transfer"
        - "external_api"
    action: flag
    risk_tier: critical

保存文件。策略引擎会自动检测更改——无需重启。

测试循环检测器

from openai import OpenAI
import time

client = OpenAI(base_url="http://localhost:8080/v1")

# Simulate a runaway agent — rapid repeated calls
for i in range(60):
    try:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": f"Request {i}"}]
        )
        print(f"Request {i}: OK")
    except Exception as e:
        print(f"Request {i}: BLOCKED — {e}")
        break
    time.sleep(0.5)

您应该会看到正常的响应,直到触发速率限制,之后请求将被阻止,代理将停止。

可观测性

  • Jaeger trace viewer – 每个请求的完整跟踪。
  • Prometheus metrics – 成本和请求统计。
  • Episode Store API – 在上下文、延迟和成本下重放完整序列。

您在 5 分钟内获得的收益

  • 循环检测 – 自动终止失控的代理。
  • 预算上限 – 不会出现意外的 API 费用。
  • 工具限制 – 危险功能被阻止。
  • 风险层级 – 高风险操作会被标记为需要人工审查。
  • 完整审计日志 – 每个决策都会被记录并可重放。

全部无需修改代理核心逻辑的任何一行;终止开关位于基础设施层。

自定义策略

策略引擎支持任意基于 YAML 的规则。您可以:

  • 阻止特定模型。
  • 限制每个请求的 token 数量。
  • 对特定工具调用要求人工批准。
  • 定义您需要的任何模式。

框架插件

对于更深入的集成,信任插件可用于:

  • CrewAI
  • LangChain
  • AutoGen
  • OpenAI Agents SDK

这些在框架层面添加了 trust scoringpolicy enforcement

MCP 安全(可选)

如果您使用 模型上下文协议 (MCP),MCP 安全扫描器会审计您的 MCP 服务器配置,且 MCP 策略网关会对 MCP 工具调用强制执行策略。

许可证与来源

完整平台在 Apache 2.0 许可证下开源。

AIR Blackbox 是 AI 代理的飞行记录仪——记录每一次决策,重放每一次事件,执行每一项政策。如果你的代理在生产环境中做决策,它们需要一个黑匣子。

0 浏览
Back to Blog

相关文章

阅读更多 »

Subnetting 详解

什么是 Subnetting?可以把它想象成把一栋大型公寓楼拆分成不同的楼层。每层 subnet 拥有自己的编号主机(hosts),以及建筑……