Muse Image 接入 Python 项目生成图片的代码实现

Muse Image 的接口形式会随部署环境而变，但接入 Python 项目的思路是固定的：把访问凭证、HTTP 连接、请求参数和返回解析封装成独立模块，再暴露给业务代码调用。这里给出一套基于 requests 的通用接入骨架，适用于大多数提供 HTTP API 的图像生成服务。你需要先确认实际接口的鉴权头、请求体和返回字段，再替换代码中的占位配置。

Muse Image 接入 Python 项目，核心是封装一个可复用的图像生成客户端：用环境变量保存 URL 和密钥，用 requests.Session 复用连接，把参数传递和状态轮询收敛到 generate_image 方法里。适用场景是已有 HTTP API 的图像服务；操作方向是替换配置和字段映射；验证方式是用真实凭证调用一次并检查返回图片；风险边界是不同服务的鉴权方式和异步策略需要各自适配。

确认访问凭证与接口入口

首先确认服务方提供给你们的调用地址（endpoint）和访问凭证。凭证通常是 API Key 或 Token，一般放在请求头（如 Authorization: Bearer ）里。不要把它们硬编码到代码中，建议通过环境变量传递。在项目根目录的 .env 文件或启动脚本中设置：

MUSE_IMAGE_API_URL=https://your-api.example.com/v1/generate
MUSE_IMAGE_API_KEY=your_secret_token

然后在 Python 代码中读取：

import os

API_URL = os.getenv("MUSE_IMAGE_API_URL")
API_KEY = os.getenv("MUSE_IMAGE_API_KEY")

if not API_URL or not API_KEY:
    raise RuntimeError("MUSE_IMAGE_API_URL and MUSE_IMAGE_API_KEY must be set")

这样配置和代码分离，也避免密钥被提交到版本库。

建立HTTP客户端基础连接

使用 requests 库创建一个带超时和默认头部的 Session。Session 会复用底层 TCP 连接，减少重复握手开销。超时配置要覆盖连接和读取两个阶段，避免线程被长期挂起。

import requests

SESSION = requests.Session()
SESSION.headers.update({
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
})
SESSION.timeout = (10, 60)  # (connect, read)

如果接口要求的鉴权方式不是 Bearer Token，把这个头改成服务方规定的格式即可。之后所有请求都通过这个 SESSION 发送。

封装图片生成方法

将业务逻辑收敛到 generate_image(prompt, size, steps) 方法中。这个方法负责组装请求参数、发送 POST 请求、处理响应。不同服务的返回有两种常见形式：直接返回图片二进制流；或返回一个 JSON，里面包含图片 URL 或下载链接。下面的实现同时处理这两种情况，返回图片二进制数据，由调用方决定如何保存。

class GenImage:
    def __init__(self, session):
        self.session = session

    def generate_image(self, prompt, size="1024x1024", steps=25):
        payload = {
            "prompt": prompt,
            "size": size,
            "steps": steps
        }
        resp = self.session.post(API_URL, json=payload)
        resp.raise_for_status()

        # 如果响应是图片流，直接返回二进制
        if resp.headers.get("content-type", "").startswith("image/"):
            return resp.content

        # 如果响应是 JSON，尝试从常见字段中取图片地址
        data = resp.json()
        image_url = data.get("image_url") or data.get("url") or data.get("data")
        if isinstance(image_url, str) and image_url.startswith("http"):
            img_resp = self.session.get(image_url)
            img_resp.raise_for_status()
            return img_resp.content

        raise ValueError(f"Unexpected response format: {data}")

这里先按同步阻塞模型处理。如果接口是异步的，需要在收到任务 ID 后继续轮询，见下一节。

处理异步生成的状态轮询

部分图像服务提交任务后直接返回一个 task_id，图片生成需要几秒到几十秒。这时需要主动查询任务状态。轮询间隔不建议小于 1 秒，避免给服务端造成压力。设置一个最大等待时间防止死循环。

import time

MAX_WAIT_SECONDS = 120
POLL_INTERVAL = 2

def wait_for_image(session, task_id):
    started = time.time()
    task_url = f"{API_URL}/tasks/{task_id}"

    while time.time() - started < MAX_WAIT_SECONDS:
        resp = session.get(task_url)
        resp.raise_for_status()
        data = resp.json()
        status = data.get("status")

        if status == "succeeded":
            return data.get("image_url") or data.get("output")
        if status in {"failed", "error"}:
            raise RuntimeError(f"Image generation failed: {data}")

        time.sleep(POLL_INTERVAL)

    raise TimeoutError(f"Task {task_id} did not finish within {MAX_WAIT_SECONDS}s")

如果接口的状态字段名不同，比如用 state 或 phase，需要做对应替换。轮询结束后要么返回图片地址，要么抛出明确的异常。

集成到业务代码中的调用示例

在项目中使用时，把异常捕获、图片保存和自定义错误放在一起，保证业务代码能看到清晰的失败原因。下面是一个完整的调用示例：

from pathlib import Path

class MuseImageError(Exception):
    pass

def save_image_with_error_handling():
    client = GenImage(session=SESSION)
    try:
        image_binary = client.generate_image(
            prompt="A mountain lake at sunset",
            size="768x768",
            steps=30
        )
        output_path = Path("output.png")
        output_path.write_bytes(image_binary)
        print(f"Image saved to {output_path}")
        return output_path

    except requests.exceptions.Timeout as exc:
        raise MuseImageError("Request timed out") from exc
    except requests.exceptions.HTTPError as exc:
        raise MuseImageError(f"API returned HTTP {exc.response.status_code}") from exc
    except ValueError as exc:
        raise MuseImageError(f"Invalid response: {exc}") from exc

调用 save_image_with_error_handling() 后，生成的图片会保存到 output.png。如果中间任何环节失败，都会以自定义异常向上传递，方便上层决定重试还是记录日志。需要替换真实接口的字段名时，打开响应数据打印一遍，按照实际结构修改代码即可。