如何在 HiAPI 上调用 flux-1-kontext-max：curl 与 Python 教程

flux-1-kontext-max 请求长什么样

在写任何代码之前，先了解交互的形态会很有帮助。在 HiAPI 上对 flux-1-kontext-max 发起图像编辑调用，并不是一个在同一 HTTP 调用里返回图片的朴素请求/响应。它是一个两阶段任务：

1. 提交 —— 你 POST 一个 JSON 载荷，描述你的输入图片和想要的编辑。API 接受任务并返回一个任务标识符。
2. 轮询 —— 你用该标识符 GET 任务状态，直到它报告完成，然后从响应中读取输出 URL。

之所以采用这种模式，是因为图像编辑计算密集。在整个渲染过程中保持单个 HTTP 连接打开很脆弱；任务句柄让你可以断开、重试并稍后再回来。

载荷本身在概念上分为三部分：

• 模型选择器 —— 一个指定 flux-1-kontext-max 为要运行模型的字段。
• 输入 —— 通常是一张图片（URL 或 base64 载荷，取决于文档化的 schema）外加一段描述编辑的文本指令。
• 可选控制项 —— 调整生成行为的参数，例如宽高比或种子，前提是模型文档中有说明。

其余一切——认证头、提交端点、状态端点——都和你在每个 HiAPI 任务中使用的管道相同。一旦你内化了这个三步节奏（提交 → 轮询 → 下载），curl 和 Python 版本就可以互换，你可以在两者之间切换而无需重新学习流程。

前置条件与认证

在第一个请求成功之前，你需要准备好几件事。

你需要什么

• 一个 HiAPI 账户，以及从仪表盘创建的 API key。
• HiAPI API 的 基础 URL，如平台文档所示。
• API 能访问到的源图片——最常见的是公开的 HTTPS URL。如果你的图片是私有的，请上传到 API 能获取的地方，或者如果模型 schema 支持 base64 输入，则以内联方式传入。
• 用于 shell 演练的 curl，以及用于代码演练的 Python 3 和 requests 库。

认证

HiAPI 通过在授权头中发送你的 API key 来认证请求。约定是 bearer token：

Authorization: Bearer $HIAPI_API_KEY

并且每个 JSON 请求都应声明其内容类型：

Content-Type: application/json

让密钥远离源代码

把密钥存在环境变量里，而不是粘贴到脚本中：

export HIAPI_API_KEY="your_key_here"

在 Python 中，用 os.environ["HIAPI_API_KEY"] 读取它。在 curl 中，以 $HIAPI_API_KEY 引用它。这一个习惯就能防止凭证泄露最常见的原因——密钥被提交到仓库，然后在压力下被迫轮换。

如果请求返回认证错误，按顺序检查三件事：密钥存在且有效、头部拼写与文档完全一致、环境变量中没有混入多余的空格或换行。

用 curl 发送请求

下面是提交调用。替换为你自己的图片 URL 和编辑指令。

curl -X POST "$HIAPI_BASE_URL/jobs" \
  -H "Authorization: Bearer $HIAPI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "flux-1-kontext-max",
    "input": {
      "image": "https://example.com/input.jpg",
      "prompt": "Replace the sky with a dramatic sunset, keep the buildings unchanged"
    }
  }'

每一部分的作用

• -X POST —— 提交是一个创建操作，所以用 POST。
• 端点 —— jobs 端点接受新任务。基础 URL 来自你的环境，因此命令在不同环境中可移植。
• Authorization —— 你的 bearer token，从环境变量读取而不是内联。
• Content-Type: application/json —— 必需；没有它，服务器可能拒绝或错误解析请求体。
• -d —— JSON 载荷。model 选择 flux-1-kontext-max，input.image 指向你的源图，input.prompt 描述编辑。

读取响应

成功的提交会返回一个 JSON 体，其中包含任务标识符——通常是 id 或 job_id 这样的字段——以及表示任务已排队或处理中的状态。保存该标识符；轮询依赖它。

curl -s -X POST "$HIAPI_BASE_URL/jobs" \
  -H "Authorization: Bearer $HIAPI_API_KEY" \
  -H "Content-Type: application/json" \
  -d @payload.json | python -m json.tool

通过 python -m json.tool 管道输出会美化打印响应，使任务 ID 容易发现。把请求体放进 payload.json，可以让命令随着提示词变长仍然保持可读。

此时你已提交了任务，但尚未收到图片。图片要在轮询之后才会返回。

Python 中的同一请求

Python 版本承载相同的逻辑，对于任何你计划长期保留的东西，有着更舒适的人体工学。

import os
import requests

BASE_URL = os.environ["HIAPI_BASE_URL"]
API_KEY = os.environ["HIAPI_API_KEY"]

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

payload = {
    "model": "flux-1-kontext-max",
    "input": {
        "image": "https://example.com/input.jpg",
        "prompt": "Replace the sky with a dramatic sunset, keep the buildings unchanged",
    },
}

response = requests.post(
    f"{BASE_URL}/jobs",
    headers=headers,
    json=payload,
    timeout=30,
)
response.raise_for_status()
job = response.json()
print("job id:", job["id"])

为什么这与 curl 调用镜像一致

json=payload 参数会序列化字典并为你设置内容类型——发送的字节与 curl 相同。timeout=30 限制提交请求，使挂起的连接大声失败，而不是拖住你的进程。raise_for_status() 把 4xx 或 5xx 变成你可以捕获的异常，而不是静默返回一个错误文档。

重要的小差异

• 字典代替字符串 —— 你把载荷构建为 Python 对象，这使得计算提示词或注入变量变得容易。
• 异常处理 —— 把调用包在 try/except requests.HTTPError 中，以记录响应体；错误载荷通常会解释输入哪里有问题。
• 无需 shell 转义 —— 带引号的长提示词不再需要转义，这消除了一整类错误。

这里返回的任务 ID 直接进入下面的轮询步骤。任务不会因为是从 Python 而不是 curl 提交的而有任何变化。

处理异步任务并轮询结果

提交很快返回；渲染则不会。你通过轮询任务 ID 的状态端点来等待。

轮询循环

import time

def wait_for_job(job_id, headers, interval=3, max_attempts=100):
    url = f"{BASE_URL}/jobs/{job_id}"
    for attempt in range(max_attempts):
        r = requests.get(url, headers=headers, timeout=30)
        r.raise_for_status()
        data = r.json()
        status = data.get("status")
        if status == "succeeded" or status == "completed":
            return data
        if status == "failed" or status == "error":
            raise RuntimeError(f"Job failed: {data}")
        time.sleep(interval)
    raise TimeoutError(f"Job {job_id} did not finish in time")

解读任务状态

预期一个小词汇表：工作开始前的 queued/pending 状态，模型渲染期间的 processing/running 状态，以及一个 终止 状态——成功或失败。有些 API 使用略有不同的拼写；检查你实际得到的响应并映射这些字符串，不要假设。关键规则：只有当状态是终止成功状态且存在输出 URL 时，才把任务视为完成。 状态端点返回 200 并不意味着图片已就绪。

轮询礼仪

• 间隔你的请求。 图像任务每次轮询之间隔几秒就足够了。每 100 毫秒轮询一次会浪费速率限制，毫无收益。
• 限制等待时间。 max_attempts 限制循环，使卡住的任务让你的代码失败，而不是永远挂起。
• 对 429 退避。 如果触发了速率限制，休眠更久再重试，而不是猛打。
• 如果提供 webhook，优先使用。 如果平台文档中记录了回调机制，它完全消除轮询，并且在规模上更友好。

对于 shell 测试，一个快速循环以相同方式工作：

JOB_ID="your_job_id"
while true; do
  BODY=$(curl -s "$HIAPI_BASE_URL/jobs/$JOB_ID" -H "Authorization: Bearer $HIAPI_API_KEY")
  echo "$BODY"
  echo "$BODY" | grep -q '"succeeded"' && break
  sleep 3
done

一旦你跳出循环，终止响应中持有输出引用。那是下一步。

下载并保存输出

完成的任务响应指向渲染后的图片，通常是一个 URL（有时是一个列表）。用第二个请求获取它并将字节写入磁盘。

import pathlib

def download_output(job_data, path="edit.png"):
    out = job_data.get("output")
    if isinstance(out, list):
        out = out[0]
    url = out["url"] if isinstance(out, dict) else out

    r = requests.get(url, timeout=60)
    r.raise_for_status()
    pathlib.Path(path).write_bytes(r.content)
    return path

需要注意的要点

• 不要硬编码形状。 输出字段因模型而异——有时是字符串 URL，有时是带 url 键的对象，有时是列表。检查真实响应并在下载前归一化。
• 输出 URL 可能过期。 把 URL 视为短生命周期：立即下载，或马上复制到你自己的存储。不要把 URL 作为图片的永久记录存储。
• 验证字节。 检查 Content-Type 以 image/ 开头且文件大小合理；一个保存为 .png 的错误页面是静默失败。
• 重试瞬态下载错误。 用小规模退避重试包裹 GET——签名 URL 偶尔因时钟漂移返回 403，网络也会抖动。

对于快速手动检查，curl 就够用：

curl -L -o edit.png "$OUTPUT_URL"

-L 标志跟随重定向，CDN 托管的输出经常使用重定向。文件落盘后，打开它并确认编辑确实符合提示词——这是你的端到端验证，在图片必须在失败时重新生成的管道中，值得自动化一个视觉或哈希检查。

常见失败点与修复

大多数首次集成失败都以少数几种方式之一出错。

认证与头部

• 401/403 —— 密钥错误、过期或有多余空格。重新导出环境变量并确认头部形式与文档匹配。
• 缺少 Content-Type —— 发送 JSON 而不带该头部可能产生令人困惑的 4xx 错误。包含它，或在 Python 中使用 json=。

输入问题

• 图片无法访问 —— API 在服务端获取你的图片。localhost URL 或需要认证的链接会失败。使用公开 HTTPS URL，或在支持时内联 base64。
• 提示词被忽略或拒绝 —— 模糊的指令产生模糊的结果。描述要改变什么、保留什么，如日落示例所示。非常长的提示词可能被截断。
• 不支持的参数 —— 传入模型不接受的字段可能导致验证错误。从最小载荷开始，然后逐个添加文档化的控制项。

异步处理错误

• 从提交响应中读取图片 —— 提交调用返回任务句柄，不是像素。先轮询。
• 把任何 200 当作完成 —— 状态端点对于仍在运行的任务也返回 200。检查状态字段和输出 URL 是否存在。
• 轮询太快或永远 —— 添加间隔和上限，否则你会耗尽速率限制并挂起你的 worker。

下载问题

• 把 HTML 保存为图片 —— 始终检查内容类型和大小。
• 输出 URL 过期 —— 完成后立即下载；不要把 URL 排队留到以后。

结构性错误

• 任务 ID 不匹配 —— 如果你提交多个任务，用你自己的请求 ID 标记每个 ID，这样轮询结果不会串线。
• 没有超时 —— 每个请求都应有一个。没有它，一个挂起的调用可能拖住一个批次。

把演练变成可复用的辅助函数

三个步骤——提交、轮询、下载——属于一个函数，这样调用者就不必重复编排。

def edit_image(image_url, prompt, model="flux-1-kontext-max",
               headers=None, poll_interval=3, max_attempts=100):
    payload = {"model": model,
               "input": {"image": image_url, "prompt": prompt}}
    r = requests.post(f"{BASE_URL}/jobs", headers=headers,
                      json=payload, timeout=30)
    r.raise_for_status()
    job_id = r.json()["id"]

    data = wait_for_job(job_id, headers,
                        interval=poll_interval,
                        max_attempts=max_attempts)
    return download_output(data, path=f"{job_id}.png")

是什么让它达到生产就绪

• 单一入口点。 调用者传入图片和提示词，得到文件路径或异常。工作流细节保持隐藏。
• 显式失败。 每个请求在 HTTP 错误时抛出，轮询器在任务失败或超时时抛出，这样你的队列或调用者可以重试。
• 可配置的时序。 poll_interval 和 max_attempts 是参数，因此你可以对更重的编辑放宽它们而无需触碰逻辑。
• 可预测的输出命名。 使用任务 ID 作为文件名，使文件与日志容易关联。

下一步往哪里走

用应用层关注点包裹调用：记录任务 ID 和延迟，用抖动重试失败的提交，并把下载的字节直接推送到对象存储而不是本地磁盘。如果你处理大量编辑，先提交一批任务，然后一起轮询——提交很便宜，并发存在于等待中，而不是上传中。

HiAPI 文档中的 flux-1-kontext-max 页面是接受字段和当前参数的参考。保持辅助函数精简，让文档驱动 schema，同一个函数就能同时覆盖探索性脚本和真实管道。