调用 FLUX API 报错时,先别急着改代码。第一步是拿到完整的请求和响应记录,然后按 HTTP 状态码把故障范围分成客户端和服务端两类,再逐项检查密钥、请求体和超时重试逻辑。这个过程能让你在几分钟内定位到大概方向。
适用场景是 FLUX API 调用返回非 2xx 状态码且报错信息不明确。操作动作:先记录请求上下文,按状态码分层,再检查密钥、请求体和重试。验证方式:用 curl 重放、diff 请求体、模拟超时重试。风险边界:不能只凭状态码下结论,需结合日志和响应体判断。
记录完整的请求与响应上下文
任何排查都建立在可复现的信息上。建议在调用 FLUX API 的代码里加上结构化日志,记录请求 URL、请求头(不含密钥)、请求体和响应状态码。保留一个 request_id 字段,方便把重试、错误日志串联起来。
import requests, logging, os, uuid
logging.basicConfig(filename='flux_api.log', level=logging.INFO)
request_id = uuid.uuid4().hex
headers = {'Authorization': 'Bearer ' + os.environ['FLUX_API_KEY']}
body = {'prompt': 'a cat', 'width': 512, 'height': 512}
resp = requests.post('https://api.flux.example/v1/generate', headers=headers, json=body)
logging.info(f"request_id={request_id} url={resp.url} headers={ {k:v for k,v in headers.items() if k.lower()!='authorization'} } body={body} status={resp.status_code}")
logging.info(f"response={resp.text[:500]}")
把 request_id 同时放进日志行和请求头(如 X-Request-ID),这样服务端日志也可以关联。建议把这段日志直接写到文件,不要只打 stdout。
按HTTP状态码区分故障层
状态码首先帮我们区分故障方向。通常 4xx 指向客户端,比如鉴权失败、请求参数不对;5xx 指向服务端,比如模型处理超时或内部错误。不要混在一起排查,应该对每个状态码单独复现,并保留同一次请求的完整上下文。
| 状态码范围 | 故障层 | 需要确认的信息 |
|---|---|---|
| 4xx | 客户端 | 密钥、请求体格式、参数类型、URL路径 |
| 5xx | 服务端 | 请求是否被重试、服务端响应体、耗时 |
下面是日志定位流程图,可以按这个顺序走:
收到非2xx响应 → 查日志确认request_id和状态码
├─ 4xx → 检查密钥 → 检查请求体 → 用curl重放
└─ 5xx → 检查超时时间 → 做间隔重试 → 记录最终错误
检查密钥与权限配置
401/403 最常见的原因是密钥没传、传错或权限不足。先用代码确认环境变量里的 key 是否真的读到了,再打印请求头里 key 是否生效(不要打印完整 key,只确认非空和前缀)。
import os
from dotenv import load_dotenv
load_dotenv() # 重新读取 .env 文件
print("key exists:", bool(os.environ.get('FLUX_API_KEY')))
print("key prefix:", os.environ.get('FLUX_API_KEY', '')[:5])
然后用 curl 重放一次相同请求,对比结果。如果 curl 成功而代码失败,说明问题在代码里的请求头拼接;如果都失败,问题在密钥本身或服务端限制。
curl -X POST https://api.flux.example/v1/generate \
-H "Authorization: Bearer $FLUX_API_KEY" \
-H "Content-Type: application/json" \
-d '{"prompt":"test","width":512,"height":512}'
检查请求体与模型参数合法性
遇到 400 错误时,先把当前请求体和最近一次成功的请求体做 diff,看哪些字段变了。常见问题集中在 prompt 字段长度、图像尺寸类型(比如传了字符串而不是整数)。
import diff
old_body = {'prompt': 'cat', 'width': 512, 'height': 512}
new_body = {'prompt': 'a very long prompt...', 'width': '512', 'height': 512}
for k in set(old_body) | set(new_body):
if old_body.get(k) != new_body.get(k):
print(k, old_body.get(k), '->', new_body.get(k))
也可以直接用 JSON Schema 校验,在发送前拦截不合法参数:
import jsonschema
schema = {
"type": "object",
"properties": {
"prompt": {"type": "string", "maxLength": 1000},
"width": {"type": "integer", "minimum": 64, "maximum": 2048},
"height": {"type": "integer", "minimum": 64, "maximum": 2048}
},
"required": ["prompt"]
}
jsonschema.validate(new_body, schema) # 抛出异常说明字段不合法
模拟重试与后端超时场景
5xx 可能是临时故障,也可能是持续故障。写一个带超时和重试的脚本,记录每次超时的时间点,并确认最终是否成功。重试间隔建议用指数退避,避免雪崩。
import time, requests
max_retries = 3
for attempt in range(max_retries):
try:
resp = requests.post(url, headers=headers, json=body, timeout=30)
if resp.status_code >= 500:
raise RuntimeError(f"server error: {resp.status_code}")
print("成功", resp.status_code)
break
except Exception as e:
logging.error(f"attempt {attempt+1} failed: {e} at {time.time()}")
time.sleep(2 ** attempt)
else:
print(f"最终错误: {e}")
如果在多次重试后仍然失败,把最终错误和最近的日志一起打印给用户。注意:重试只适合临时故障,如果日志里连续几个小时都超时,应该先暂停调用,而不是无限重试。