笙亿网络策划

调完 FLUX 3 Max 一直拿不到结果——该看任务状态还是回调地址？

约 11 分钟读完 5 阅
文章导读
请求已经发出去，却迟迟拿不到成品，先别急着改回调地址。判断的分界点是：手上有没有一个提交成功返回的任务标识。没有标识，问题多半停在提交那一步；有标识，就把它拿去查一次任务状态；只有查询明确显示任务已经完成、而本地始终没有收到投递，才轮到回调地址和接收端。这个顺序反过来，就会在一个可能根本没提交成功的请求上反复检查回调配置。
📋 目录
  1. A 先确认提交那一步有没有拿到标识
  2. B 拿标识去查一次任务状态
  3. C 回调收不到时先在本地起一个能打日志的接收端
  4. D 把超时、重试和去重放在一起定
A A

请求已经发出去，却迟迟拿不到成品，先别急着改回调地址。判断的分界点是：手上有没有一个提交成功返回的任务标识。没有标识，问题多半停在提交那一步；有标识，就把它拿去查一次任务状态；只有查询明确显示任务已经完成、而本地始终没有收到投递，才轮到回调地址和接收端。这个顺序反过来，就会在一个可能根本没提交成功的请求上反复检查回调配置。

先用提交返回的任务标识把「提交失败 / 任务未完成 / 回调没送达」三类情况分开：记录提交响应的状态码与完整响应体，用标识查一次状态，再在本地起一个能打印请求的接收端，看请求到底有没有到达。适用场景是请求已发出、日志里只有一条发送记录、界面上没有成品。验证方式是接收端日志与状态字段是否进入终态。字段名和状态枚举通常随平台版本变化，需要以实际返回和后台页面确认，本页只给排查顺序与字段清单。

先确认提交那一步有没有拿到标识

调用生成接口后，本地日志里至少要能回答一个问题：这次请求有没有被受理。受理和生成是两件事，接口返回 200 也可能只是「任务已入队」，此时没有成品是正常的，需要拿标识继续查。相反，如果提交阶段就返回 4xx 或连接超时，后面所有关于回调的判断都没有意义。

提交成功时建议把下列字段一起落到日志里，不要只取其中一两个：

  • HTTP 状态码，以及响应头中的 request-id / trace-id 一类追踪字段
  • 任务标识，常见字段名是 task_id、job_id、request_id，以实际返回为准
  • 初始状态字段，例如 status 或 state 的第一个值
  • 提交时刻的本地时间戳，用于和任务创建时间对齐、排除时区差
  • 完整的原始响应体，不要只解析一个字段就丢掉，字段名对不上时靠它回头核对
  • 本次请求的参数摘要：模型名与版本、尺寸或步数一类关键参数，便于复现

提交失败时，判断顺序建议是：先看 HTTP 状态码属于客户端错误、服务端错误还是网络超时；再看错误体里的 code 字段；然后看 message 和它指向的具体参数；最后才回头检查自己的请求体是否漏字段、类型不对或超出限制。先读状态码再读 message，可以避免被一句笼统的提示带去改无关配置。如果错误是连接超时或读取超时，那就属于传输层问题，和回调地址无关。

拿标识去查一次任务状态

有了标识，查一次状态就能把「还在跑」和「已经失败」分开。这是一个只读动作，可以随时执行，不会产生额外任务。下面是一段通用骨架，URL、字段名和状态枚举都需要按你的平台实际返回替换，不要直接照抄：

import os, json, requests

API_KEY = os.environ['FLUX_API_KEY']
TASK_ID = 'task_xxx'  # 换成提交返回的标识
URL = 'https://api.example.com/v1/tasks/{task_id}'.format(task_id=TASK_ID)

resp = requests.get(
    URL,
    headers={
        'Authorization': 'Bearer ' + API_KEY,
        'Accept': 'application/json',
    },
    timeout=15,
)
print('http_status:', resp.status_code)
print('request_id:', resp.headers.get('x-request-id'))

body = resp.json()
print('raw:', json.dumps(body, ensure_ascii=False))

state = body.get('status') or body.get('state')  # 字段名以实际返回为准
print('task_state:', state)

if state in ('queued', 'pending', 'submitted'):
    print('排队中，等待，不要重复提交')
elif state in ('running', 'processing', 'in_progress'):
    print('处理中，记录时间点后再查')
elif state in ('succeeded', 'success', 'completed'):
    print('已完成，结果在响应体里或由回调投递，转去查回调')
elif state in ('failed', 'error', 'canceled'):
    print('已失败，读取错误字段:', body.get('error') or body.get('fail_reason'))
else:
    print('状态值不认识，先按原始响应体判断，不要猜')

鉴权头通常写成 Authorization: Bearer <KEY>，密钥从环境变量读取，不要写进代码仓库。状态字段的读取要先打印完整响应体再分支，因为不同平台可能把状态放在 data.status 或 result.state 里。查询结果如果长时间停留在排队或处理中，属于任务侧问题，此时改回调地址没有任何作用；如果状态是失败，错误原因通常已经写在响应体里，也不需要等回调。

调完 FLUX 3 Max 一直拿不到结果——该看任务状态还是回调地址？

回调收不到时先在本地起一个能打日志的接收端

只有一种情况值得怀疑回调地址：查询显示任务已完成，但接收端、日志和本地文件里都没有对应的投递记录。这时先不要动上游配置，先在本地起一个最小接收端，把请求方法、请求头和请求体都打出来，确认链路本身通不通。

from http.server import BaseHTTPRequestHandler, HTTPServer

class Hook(BaseHTTPRequestHandler):
    def do_POST(self):
        length = int(self.headers.get('Content-Length', 0))
        raw = self.rfile.read(length)
        print('path:', self.path)
        print('method:', self.command)
        for k, v in self.headers.items():
            print('header:', k, v)
        print('body:', raw.decode('utf-8', errors='replace'))
        self.send_response(200)
        self.send_header('Content-Type', 'application/json')
        self.end_headers()
        self.wfile.write(b'{"ok": true}')

    def log_message(self, fmt, *args):
        print('access:', fmt % args)

HTTPServer(('0.0.0.0', 8080), Hook).serve_forever()

启动后先用本机 curl 打一次这个地址，确认接收端自己会打日志、会回 200，否则后面看到的「没有请求」可能只是接收端没起来或被防火墙挡住。之后把上游回调地址指向一个公网可达的地址和路径，再触发一次任务。

判断「请求根本没来」的方法比较直接：接收端连 access 日志都没有，说明连接层就没到达；有 access 日志但方法或路径不对，说明地址配到了别的路由；有请求体但解析报错，说明是格式或字段问题，属于接收端要改。同时检查这几项：地址是否要求 HTTPS、端口是否放行、路径是否与配置完全一致、上游是否会在投递前做可达性校验、接收端返回码是否在 2xx 区间。接收端返回非 2xx 时，很多平台会判定投递失败并重投，日志里会看到重复请求，这属于重试而不是重复提交。

把超时、重试和去重放在一起定

拿不到结果时，人容易直接加大重试次数，结果同一个业务请求提交了多份，最后收到多份成品或重复扣费。重试要和去重一起设计：重试次数、退避间隔、去重键三个参数要在同一处定义，并且写进日志。

  • 重试次数建议控制在少量几次，覆盖瞬时超时和 5xx 即可，不要无限循环
  • 退避间隔用递增方式，例如从 2 秒开始按倍数放大，避免在服务端排队时反复冲击
  • 去重键用业务侧已有的唯一号，比如订单号加参数摘要，比每次生成随机值更可靠
  • 只对超时、连接错误、5xx 和限流类错误重试，参数错误和鉴权错误重试也不会成功
import time, hashlib, json, requests

DEDUP_KEY = hashlib.sha256(
    (biz_id + json.dumps(payload, sort_keys=True)).encode('utf-8')
).hexdigest()

headers = {
    'Authorization': 'Bearer ' + API_KEY,
    'Content-Type': 'application/json',
    'Idempotency-Key': DEDUP_KEY,
}

result = None
max_retries = 3
base_delay = 2
for attempt in range(1, max_retries + 1):
    try:
        r = requests.post(SUBMIT_URL, json=payload, headers=headers, timeout=30)
        print('attempt:', attempt, 'status:', r.status_code, 'key:', DEDUP_KEY)
        if r.status_code < 400:
            result = r.json()
            break
        if r.status_code in (400, 401, 403, 404, 422):
            print('客户端错误，不重试:', r.text)
            break
    except (requests.Timeout, requests.ConnectionError) as e:
        print('attempt:', attempt, '网络异常:', repr(e))

    if attempt < max_retries:
        time.sleep(base_delay * (2 ** (attempt - 1)))

print('result:', result)

这段骨架的用途是让每次重试都带着同一个去重键，服务端可以据此判断是否已经受理过同一请求。是否真的支持幂等键、字段名叫什么，需要按平台文档或实际行为确认；如果不支持，就要在本地记录已提交的标识，重试前先查一次状态，避免重复提交。最后提醒一点：状态查询、回调投递和本地接收端的时间要以同一时区对齐，时间戳错位时很容易把「还没到」误判成「没送到」。