在免费服务器上通过故障注入测试 LLM 管道的韧性,学会在 rate limit、timeout、5xx 等真实故障下的恢复路径。
免费的基础设施是不可预测的。这不是 bug —— 这是一种测试机会。混沌工程通常被描述为拥有专职 SRE 和生产集群的团队的奢侈品。事实恰恰相反。最需要故障注入的系统正是那些运行在免费服务器和免费模型访问上的系统,因为它们的故障模式最多样、最最缺乏记录。
披露:本文作为 MonkeyCode 产品推广的一部分撰写。
付费 API 给你一份合同。免费层给你一个概率分布。速率限制毫无预警地出现。超时超过你的耐心。当批处理任务进行到第三小时时,偶尔的 5xx 就出现了。你无法修复这些故障——它们是免费访问的代价。但你可以准确地了解你的代码在它们发生时如何表现。
这种学习正是混沌测试的目的。你在受控的方式下故意破坏东西,这样当它们真正坏掉时,恢复路径已经是肌肉记忆。诀窍是你不需要专用的测试集群。你已有的免费服务器正是故意破坏东西的完美场所。
注入故障最干净的方法不是修改你的应用代码,而是在你的代码和模型 API 之间放置一个小型代理,由它来决定每个请求是否失败以及如何失败。这保持你的生产代码不变,同时让你完全控制混沌。
# chaos_proxy.py — inject failures into any OpenAI-compatible API call
# usage: python chaos_proxy.py --port 8765 --upstream https://api.example.com/v1
# then point your client at http://localhost:8765/v1
import argparse
import json
import random
import time
import urllib.request
from http.server import BaseHTTPRequestHandler, HTTPServer
class ChaosHandler(BaseHTTPRequestHandler):
rate_limit_prob = 0.0
timeout_prob = 0.0
error_prob = 0.0
upstream = ""
def do_POST(self):
r = random.random()
if r < self.rate_limit_prob:
self.send_response(429)
self.end_headers()
self.wfile.write(b'{"error": {"message": "rate limit (injected)"}}')
return
if r < self.rate_limit_prob + self.timeout_prob:
time.sleep(15) # longer than your client's timeout
return
if r < self.rate_limit_prob + self.timeout_prob + self.error_prob:
self.send_response(500)
self.end_headers()
self.wfile.write(b'{"error": {"message": "internal error (injected)"}}')
return
length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length)
req = urllib.request.Request(
self.upstream + self.path,
data=body,
headers=dict(self.headers),
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=30) as resp:
data = resp.read()
self.send_response(resp.status)
for k, v in resp.headers.items():
self.send_header(k, v)
self.end_headers()
self.wfile.write(data)
except Exception as e:
self.send_response(502)
self.end_headers()
self.wfile.write(str(e).encode())
def log_message(self, fmt, *args):
print(f"[chaos] {self.address_string()} {fmt % args}")
def main():
p = argparse.ArgumentParser()
p.add_argument("--port", type=int, default=8765)
p.add_argument("--upstream", required=True)
p.add_argument("--rate-limit-prob", type=float, default=0.05)
p.add_argument("--timeout-prob", type=float, default=0.03)
p.add_argument("--error-prob", type=float, default=0.02)
args = p.parse_args()
ChaosHandler.rate_limit_prob = args.rate_limit_prob
ChaosHandler.timeout_prob = args.timeout_prob
ChaosHandler.error_prob = args.error_prob
ChaosHandler.upstream = args.upstream
server = HTTPServer(("localhost", args.port), ChaosHandler)
print(f"chaos proxy on :{args.port} -> {args.upstream}")
print(f"probabilities: rate_limit={args.rate_limit_prob} timeout={args.timeout_prob} error={args.error_prob}")
server.serve_forever()
if __name__ == "__main__":
main()
用 --rate-limit-prob 0.2 运行它,然后观察你的 agent 循环如何跌倒。那次跌倒正是你需要的信息。
从速率限制实验开始。将概率设置为 0.2 并运行你的正常工作负载。观察会发生什么。大多数客户端立即重试,这使 429 更糟。正确的响应是指数退避加抖动——每次重试时增长的随机延迟。如果你的代码不这样做,代理会准确地向你展示重试风暴建立的速度。
超时实验更微妙。15 秒的延迟并不总是触发你的客户端超时。一些 HTTP 库默认 30 秒,一些默认 60 秒,一些默认无穷大。代理使这种不匹配变得可见。你会了解到你真正的超时预算,而不是你假设的那个。
500 实验测试你的检查点。当调用在运行中途失败时,你的作业是从最后一个完成的步骤恢复还是从零重新开始?在免费服务器上,这种区别是 10 分钟恢复和 3 小时重做的区别。
用代理运行一个小时,对代表性的工作负载记录三个数字:失败请求的百分比、每次失败的恢复时间,以及重试浪费的 token。最后一个数字是没有人测量的。一个重试会重新发送完整的对话历史,所以一个单一的 429 可能让你花费原始调用十倍的 token。
这就是免费访问的隐藏经济。配额不是被工作本身消耗的,而是被故障消耗的。5% 的失败率与朴素重试可以将你的 token 消耗膨胀 20-30%。混沌测试将这种不可见的成本变成一个你可以看到和修复的数字。
如果你的工作负载是单个没有重试逻辑的同步调用,代理什么也添加不了——你已经知道失败是什么样的。如果你正在构建一个用户面向的助手,其中 15 秒的延迟本身就是产品失败,混沌测试只会确认你已经知道的事情:你需要具有延迟保证的付费访问。而且如果你不能容忍你的管道中的任何人为故障,请在单独的环境中运行实验,而不是对你的实时作业运行。
混沌测试不是为了贬低免费基础设施,而是为了使你的代码诚实于它能生存的东西。免费服务器加上免费模型访问给你提供了一个地方来运行这种诚实检查,而无需花费一分钱。在真正的速率限制到来之前,故意破坏东西,测量恢复,并修复重试逻辑。
如果你想要一个运行这些实验的地方,MonkeyCode 的免费服务器和免费模型访问是一个实用的起点。将代理指向他们的 API,设置概率,看看你的代码在压力下实际做什么。