过去一年，模型相关的故障大多是“跑飞了”：某个后台智能体在循环里反复调用工具，某个批处理任务误用了生产密钥，某个实验脚本忘了设上限。出问题时，团队往往只能眼看账单上涨，或者花很久才定位到哪个进程在打请求。
这篇教程要搭的是一套“控制室”：先登记每个工作负载，再追踪它们的运行状态，最后装上一个能在分钟级真正生效的急停开关，并用演练验证它确实能停。做完之后，你应该能拿到四个具体成果：
1. 一份工作负载登记表，明确每个调用方的负责人、预算和急停开关名；
2. 一套带标签的运行指标，能在出事的几分钟内看出是哪个负载、哪个环节在消耗额度；
3. 三级熔断策略：限流、降级、停止，按严重程度自动或手动触发；
4. 一次完整的急停演练记录，包含从按下开关到请求归零的实际耗时。
关于平台界面的说明：Anthropic 控制台里工作区、API 密钥、用量与成本、速率限制这些能力的位置和命名可能随版本调整，部分账号里会有专门的控制面板入口。下文用“控制台”统称这些管理界面，涉及具体菜单位置请以官方文档当前版本为准。真正决定急停是否有效的，是下面这套工程做法，而不是某个按钮。
前置条件清单
动手前先确认这些条件，能省掉后面一半的返工。
- 权限：控制台里能创建工作区、签发与吊销 API 密钥、查看用量与成本的角色。
- 凭据管理：有地方存密钥（Vault、云厂商 Secrets Manager、K8s Secret 均可），不能散落在代码和聊天记录里。
- 标签规范：约定好工作负载 ID 的格式，例如
<团队>-<系统>-<用途>，如growth-agent-daily-report。 - 一个配置中心：Redis、etcd、配置服务、甚至对象存储上的一个 JSON 文件都能当“开关总线”，关键是所有进程都能读到。
- 可观测性底座：能打日志、能上报自定义指标的地方（Prometheus、云监控、日志平台任选）。
- 值班机制：每个工作负载至少有一个负责人在册，出事时知道该找谁。
先别急着上工具。下面第一步是纯人工的，但它的价值比后面所有代码都高。
步骤一：先盘清“谁在跑模型”——建立工作负载登记表
控制室的前提是“有名单”。没有名单，你看到用量暴涨也不知道该关谁。
把登记表当成代码来管，放在仓库里，走 PR 评审。建议的最小字段集：
```yaml
workloads/registry.yaml
version: 1
workloads:
- id: growth-agent-daily-report
display_name: 每日增长分析智能体
owner: team-growth
oncall: "@zhangsan" # 值班人，必须能被打醒
env: prod # prod / staging / dev / experiment
purpose: 每天拉取渠道数据并生成分析摘要
data_class: internal # public / internal / confidential
workspace: ws-prod-growth # 控制台工作区，决定账单归属
credential_ref: secret/prod/growth-agent-key # 只存引用，不存明文
kill_switch_key: ks:workload:growth-agent-daily-report
budgets:
daily_usd: 40
monthly_usd: 600
limits:
max_steps: 60 # 智能体单次任务最大步数
max_wall_clock_min: 25 # 单次任务最长挂钟时间
max_tokens_per_task: 400000
runbook: docs/runbooks/growth-agent.md
enabled: true
```
几个字段值得单独解释：
credential_ref而不是密钥本身：登记表会进仓库，明文密钥进去就等于泄露。kill_switch_key：提前约定好开关名，急停时不用临时想名字，也不会关错对象。max_steps/max_wall_clock_min：长跑智能体失控最常见的原因就是“没有步数上限”，模型自己停不下来。oncall：没有负责人的工作负载，出事时没人能判断“该不该停”。
登记表要有校验。加一个 CI 检查，字段缺失或 ID 重复就挂掉：
```python
scripts/validate_registry.py
import sys, yaml, re
REQUIRED = ["id", "owner", "oncall", "env", "workspace",
"credential_ref", "kill_switch_key", "budgets", "limits", "runbook"]
def main(path="workloads/registry.yaml"):
data = yaml.safe_load(open(path, encoding="utf-8"))
seen, errors = set(), []
for w in data.get("workloads", []):
for f in REQUIRED:
if not w.get(f):
errors.append(f"{w.get('id','?')}: 缺少字段 {f}")
if w["id"] in seen:
errors.append(f"{w['id']}: ID 重复")
seen.add(w["id"])
if not re.fullmatch(r"[a-z0-9]+(-[a-z0-9]+){2,}", w["id"]):
errors.append(f"{w['id']}: ID 需为 三段小写短横线格式")
if w["env"] == "prod" and w["budgets"].get("daily_usd", 0) <= 0:
errors.append(f"{w['id']}: 生产负载必须设置日预算")
if errors:
print("\n".join(errors)); sys.exit(1)
print(f"OK: {len(seen)} 个工作负载")
if __name__ == "__main__":
main()
```
跑一次：
```bash
python scripts/validate_registry.py
```
清单建立之后，写一条硬规则：新负载不登记，不给密钥。这条规则执行三周，登记表就会自然成为可信来源。
步骤二：一个负载一份凭据，全部打上标签
急停之所以经常失效，是因为十个负载共用一个 API 密钥：吊销这把密钥，全公司一起挂；不吊销，失控的负载继续跑。
做法是每个工作负载独立签发密钥，按环境分工作区。命名带上工作负载 ID，方便在控制台里反查：
```bash
密钥只写进密钥管理系统，不落到终端历史
export ANTHROPIC_API_KEY="$(vault kv get -field=key secret/prod/growth-agent-key)"
固定请求头，方便网关层做归因
curl https://api.anthropic.com/v1/messages \
-H "x-api-key: $ANTHROPIC_API_KEY" \
-H "anthropic-version: 2023-06-01" \
-H "content-type: application/json" \
-H "x-workload-id: growth-agent-daily-report" \
-d '{
"model": "'"$MODEL"'",
"max_tokens": 1024,
"messages": [{"role": "user", "content": "ping"}]
}'
```
x-workload-id 是自定义头。如果请求走自建网关，网关把它写进指标标签；如果不走网关，就在 SDK 客户端里统一注入。模型名用环境变量传入，具体可用型号以官方文档当前版本为准，不要硬编码在某处。
同时给每个工作负载设两个“软闸”：
- 速率限制：在控制台的用量/限额页面按工作区设定，先设一个明显够用但不离谱的值，运行一周后按 P95 上调。
- 预算告警：日预算 70% 提醒负责人，100% 触发自动降级（下一步实现）。
步骤三：在调用路径上装一个能读开关的包装器
这一步是整个方案的核心。所有模型调用必须经过同一个函数，函数在做任何事之前先做三件事：查开关、查预算、查步数。绕过它直接调 SDK 的代码，等于给自己留了后门。
```python
llm_gate.py
import os, time, json, redis, requests
REDIS = redis.Redis.from_url(os.environ["SWITCH_REDIS_URL"])
API_KEY = os.environ["ANTHROPIC_API_KEY"]
WORKLOAD_ID = os.environ["WORKLOAD_ID"] # 从登记表注入
BASE = "https://api.anthropic.com/v1/messages"
class Stopped(Exception):
"""急停触发，调用方必须向上冒泡，禁止重试"""
def read_switch() -> dict:
"""返回 {'level': 0|1|2|3, 'reason': str}，读不到时按保守值处理"""
raw = REDIS.get(f"ks:workload:{WORKLOAD_ID}")
if raw is None:
读不到开关：生产环境按 L1 限流，避免误放行失控负载
return {"level": 1, "reason": "switch-unreachable"}
return json.loads(raw)
def call_model(prompt: str, *, task: dict, max_tokens: int = 1024) -> str:
sw = read_switch()
if sw["level"] >= 3:
raise Stopped(f"L3 急停生效: {sw['reason']}")
步数与挂钟时间护栏
if task["steps"] >= int(os.environ["MAX_STEPS"]):
raise Stopped("超出最大步数")
if time.time() - task["started_at"] > int(os.environ["MAX_WALL_CLOCK_SEC"]):
raise Stopped("超出最大挂钟时间")
L2：降级——缩短输出、关闭工具、换小模型
payload = {"model": os.environ["MODEL"], "max_tokens": max_tokens,
"messages": [{"role": "user", "content": prompt}]}
if sw["level"] == 2:
payload["max_tokens"] = min(max_tokens, 256)
payload["model"] = os.environ["MODEL_FALLBACK"]
for attempt in range(2):
r = requests.post(BASE, timeout=60, json=payload, headers={
"x-api-key": API_KEY, "anthropic-version": "2023-06-01",
"content-type": "application/json", "x-workload-id": WORKLOAD_ID,
})
if r.status_code == 429 and sw["level"] == 1 and attempt == 0:
time.sleep(2 ** attempt) # L1 只做有限重试，且不无限退避
continue
r.raise_for_status()
task["steps"] += 1
return r.json()["content"][0]["text"]
raise RuntimeError("调用失败")
```
配套的环境变量在启动时从登记表注入，容器启动脚本里做一次即可：
```bash
export WORKLOAD_ID=growth-agent-daily-report
export MAX_STEPS=60
export MAX_WALL_CLOCK_SEC=1500
export MODEL="$(cat /etc/llm/model.txt)" # 具体型号以官方文档为准
export MODEL_FALLBACK="$(cat /etc/llm/model_fallback.txt)"
```
写开关的辅助函数，让“按开关”变成一个命令：
```python
ks.py
import json, sys, redis, os
r = redis.Redis.from_url(os.environ["SWITCH_REDIS_URL"])
wid, level, reason = sys.argv[1], int(sys.argv[2]), sys.argv[3]
r.set(f"ks:workload:{wid}", json.dumps({"level": level, "reason": reason}))
print(f"{wid} -> L{level} ({reason})")
```
```bash
python ks.py growth-agent-daily-report 3 "发票数据异常，人工介入"
```
注意：Stopped 异常不允许被重试逻辑吞掉。很多事故的真相是“急停生效了，但队列的重试把任务又跑了一遍”。在任务队列层也要加判断，收到 Stopped 就把任务标记为 aborted 而不是 failed，避免自动重投。
步骤四：让运行态看得见——四条必看指标
登记和开关解决“能不能停”，可见性解决“该不该停”。每条调用产生一条结构化日志：
```json
{
"ts": "2025-01-01T10:00:00Z",
"workload_id": "growth-agent-daily-report",
"task_id": "t-8f21",
"step": 17,
"model": "…",
"input_tokens": 3120,
"output_tokens": 480,
"latency_ms": 2380,
"tool_calls": ["fetch_channel_data", "fetch_channel_data"],
"switch_level": 0,
"outcome": "ok"
}
```
从日志里聚出四类指标，做成一块看板，出问题时只看这一块：
1. 花费速率：每分钟消耗金额，按 workload_id 拆线。看的是斜率，不是总额。
2. 步数分布：每个任务的实际步数直方图。某个负载的 P95 步数突然翻倍，就是循环的前兆。
3. 工具调用重复度：同名工具在单任务内被调用超过 N 次，几乎可以判定为原地打转。
4. 开关状态：所有负载的当前等级，一眼看到谁被限流、谁被停。
一个能直接跑的重复度检测，接在任务循环里：
```python
from collections import Counter
def check_loop(task: dict, tool_name: str, threshold: int = 5):
task.setdefault("tool_seq", []).append(tool_name)
c = Counter(task["tool_seq"][-20:])
if c[tool_name] >= threshold:
raise Stopped(f"工具 {tool_name} 在最近 20 步内重复 {c[tool_name]} 次")
```
这类基于“形状”的检测，比单纯看花费更早发现问题——账单涨起来的时候，通常已经晚了二十分钟。
步骤五：分级熔断——先限流，再降级，最后才停
一上来就全停，业务方第二天就会要求拆掉这套东西。三级策略更容易长期存活：
| 等级 | 触发场景 | 动作 | 谁按 |
|---|---|---|---|
| L1 限流 | 花费速率超日预算的 1.5 倍、错误率升高 | 并发降到 1、退避重试、排队 | 自动 |
| L2 降级 | 日预算用尽、连续 3 次步数超限 | max_tokens 压到 256、关工具、换备用模型 | 自动 |
| L3 急停 | 数据泄露迹象、明确跑飞、人工判断 | 拒绝一切新请求，任务标记 aborted | 人（或强规则自动） |
自动升级规则用一个定时任务实现，每 60 秒评估一次：
```python
autobrake.py —— 每分钟跑一次
import os, json, redis, datetime
r = redis.Redis.from_url(os.environ["SWITCH_REDIS_URL"])
for wid in os.environ["MANAGED_WORKLOADS"].split(","):
spent = float(r.get(f"cost:today:{wid}") or 0)
daily = float(r.get(f"budget:daily:{wid}") or 0)
if daily and spent > daily * 1.5:
r.set(f"ks:workload:{wid}", json.dumps({"level": 3, "reason": f"日预算 {daily} 已用 {spent:.0f}，超 1.5 倍"}))
elif daily and spent > daily:
r.set(f"ks:workload:{wid}", json.dumps({"level": 2, "reason": "日预算用尽，降级运行"}))
```
用 SET 覆盖而不是追加，保证任意时刻只有一个明确的等级；评估函数是幂等的，重复执行不会出错。
步骤六：一键急停要“一键”，还要真的“停”
急停按钮有三个层次，按见效速度从快到慢排：
第一层：翻开关（秒级，可逆）
```bash
python ks.py growth-agent-daily-report 3 "跑飞，人工介入"
```
包装器下一次调用就会抛 Stopped。前提是包装器每次都读开关，而不是启动时读一次缓存十分钟。
第二层：吊销密钥（分钟级，影响面精确到该工作负载）
在控制台里禁用该工作负载对应的 API 密钥。这一步拦住的是所有新请求，包括绕过包装器的野代码。注意它拦不住已经发出的在途请求，也拦不住“重启后重新读到新密钥”的容器——所以先翻开关，再吊销。
第三层：下游断供（准实时）
如果工作负载会调用自建工具网关，在网关层加同样的开关判断，直接返回 503。这一层往往比模型侧更快生效，因为工具调用通常是失控循环里最贵、最有副作用的部分。
三件事按顺序做，写成 runbook，放在登记表链接的路径里：
```markdown
急停 Runbook：growth-agent-daily-report
1. 翻开关：python ks.py <id> 3 "<原因>" # 目标 < 60s
2. 确认指标：花费速率曲线在 2 分钟内归零
3. 若未归零：控制台禁用该工作负载密钥，排查是否有人绕过包装器
4. 若仍有请求：网关层封禁该 workload_id
5. 在 #incident 频道 @oncall，附上开关命令与时间戳
6. 恢复前必须记录根因，且把 MAX_STEPS 或预算调整一并提交 PR
```
步骤七：演练，否则你不知道它到底能不能停
没演练过的急停，和没有急停差不多。每月一次，选一个非关键负载，做一次完整的“假事故”：
```bash
1. 起一个持续跑的任务（在测试环境）
python run_agent.py --workload growth-agent-daily-report-test --loop 200 &
2. 记录时间，翻开关
date +%s > /tmp/t0
python ks.py growth-agent-daily-report-test 3 "月度演练"
3. 观察日志里最后一次成功调用与第一次 Stopped
grep -E "switch_level|Stopped" /var/log/agent.jsonl | tail -20
4. 计算 MTTS（mean time to stop）
echo $(( $(date +%s) - $(cat /tmp/t0) ))
```
演练要记录三个数字：按下开关到最后一个请求完成的秒数（MTTS）、是否有任务被队列重投、恢复后是否反复触发。MTTS 超过 5 分钟，就说明某处缓存了开关或者存在绕过路径，回去改包装器。
演练产出一份 5 行记录，归档即可：
```yaml
- date: 2025-01-15
workload: growth-agent-daily-report-test
mtts_seconds: 12
side_effects: 无
issues: 库存同步任务因重试被重投 1 次，已在队列层加 aborted 处理
```
常见坑与排错
急停按了，请求还在跑。 九成是开关被进程内缓存，TTL 又设得很长。把开关读取放在每次调用前，缓存 TTL 不超过 5 秒；或者用 Redis pub/sub 主动推送失效通知。
按了开关，但队列把任务重跑了。 任务框架把 Stopped 当普通异常重试。在消费者里显式区分：aborted 不重投，failed 才重投。
吊销密钥后，容器重启又活了。 密钥来自环境变量或挂载文件，新容器读到了另一份未吊销的凭据。急停时要同时检查密钥来源，把所有同源凭据一起处理。
日预算设了，但没拦住。 预算只统计到天，而失控负载可能在 20 分钟内烧完全天额度。必须同时有“花费速率”告警（每分钟金额），速率比总量更早报警。
看板上一堆流量不知道是谁的。 有负载没打 x-workload-id。用网关做强制校验：没有该头的请求直接 400，逼迫团队接入。
告警发到没人看的频道。 把告警直接发给登记表里的 oncall，写到人，不写到群。
演练时一切正常，真实故障时失效。 演练用的是测试负载，而生产负载有额外的缓存层或批量合并逻辑。至少每季度拿一个生产负载做“只能停 30 秒”的轻量演练。
日志里带上了敏感数据。 结构日志只记元数据（token 数、耗时、工具名），不记 prompt 正文；需要排查时按 task_id 去受控存储里取。
下一步建议
控制室建起来之后，按这个顺序继续加固，投入产出比更高：
1. 把登记表接进 CI/CD：部署流水线启动时校验 WORKLOAD_ID 是否已登记、预算是否已设，未登记直接拒绝部署。
2. 把开关做成有审批的操作：L3 可以一键按，但恢复要两个人确认，避免“按了又马上恢复”的来回抖动。
3. 按任务类型分预算：交互式请求和批量分析的成本结构完全不同，混在一个预算里会互相挤占。
4. 给工具调用加副作用隔离：写操作类工具（发邮件、改库存、调支付）走一个带幂等键和二次确认的网关，即使模型跑飞也不会造成不可逆后果。
5. 季度复盘一次“差一点出事”：把花费速率异常但没触发熔断的案例拿出来看，调整阈值。
最后提醒一句：控制台的具体页面路径、可用模型、限额规则都在持续变化，配置时以官方文档当前版本为准；但这套“登记 → 可见 → 分级 → 演练”的结构不依赖任何具体版本，换平台也能照搬。真正的急停按钮不是某个页面上的开关，而是你的调用路径上那个每次都会执行的检查。