如何在5分钟内为您的AI Agent添加Kill Switch
看起来您只提供了来源链接,而没有贴出需要翻译的正文内容。请把您想要翻译的文本(包括任何 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:8080 的 Gateway。
将你的 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 scoring 和 policy enforcement。
MCP 安全(可选)
如果您使用 模型上下文协议 (MCP),MCP 安全扫描器会审计您的 MCP 服务器配置,且 MCP 策略网关会对 MCP 工具调用强制执行策略。
许可证与来源
完整平台在 Apache 2.0 许可证下开源。
AIR Blackbox 是 AI 代理的飞行记录仪——记录每一次决策,重放每一次事件,执行每一项政策。如果你的代理在生产环境中做决策,它们需要一个黑匣子。