Asian Beauty Z-Image Turbo 模型推理API封装教程：使用FastAPI构建标准化服务接口

Asian Beauty Z-Image Turbo 模型推理API封装教程：使用FastAPI构建标准化服务接口

如果你已经成功在本地部署了Asian Beauty Z-Image Turbo模型，看着它生成一张张精美的图片，是不是觉得很有成就感？但每次都要打开命令行，输入一堆参数，或者自己写脚本调用，总觉得不够方便，更别提分享给团队其他成员使用了。

这时候，一个标准化的API服务就显得尤为重要。它就像给你的模型装上一个标准插头，任何符合接口规范的程序都能轻松调用，实现远程访问、集成到其他系统、甚至对外提供服务。今天，我们就来聊聊如何用FastAPI这个现代、快速的Python Web框架，为你的Asian Beauty Z-Image Turbo模型打造一个既专业又好用的RESTful API服务。

通过这篇教程，你将学会如何设计清晰的请求和响应格式，如何优雅地处理图片生成这种耗时任务，如何自动生成漂亮的交互式API文档，以及如何为你的服务加上安全和流量控制的“门锁”。整个过程就像搭积木，一步步来，你会发现把模型能力“服务化”并没有想象中那么复杂。

1. 环境准备与项目初始化

在开始敲代码之前，我们需要先把“舞台”搭好。确保你已经有一个可以正常运行的Asian Beauty Z-Image Turbo模型环境。这里假设你已经通过openclaw或其他方式完成了本地部署。

1.1 创建项目目录与虚拟环境

首先，我们创建一个干净的项目目录，并建立独立的Python虚拟环境，避免依赖冲突。

# 创建项目目录并进入 mkdir asian_beauty_api && cd asian_beauty_api # 创建虚拟环境（这里使用venv，你也可以用conda） python -m venv venv # 激活虚拟环境 # 在Windows上： venv\Scripts\activate # 在Linux或Mac上： source venv/bin/activate

激活虚拟环境后，你的命令行提示符前通常会显示(venv)，表示你正在虚拟环境中工作。

1.2 安装核心依赖

接下来，安装我们构建API服务所需的库。核心就是FastAPI，以及用于运行服务的Uvicorn。同时，我们也会安装Pydantic（用于数据验证）和Python-multipart（用于处理文件上传）。

pip install fastapi uvicorn pydantic python-multipart

此外，你需要确保已经安装了运行Asian Beauty Z-Image Turbo模型所需的深度学习框架（如PyTorch）和模型依赖库。这部分因你的具体部署方式而异，请确保它们在你的虚拟环境中可用。

1.3 项目结构规划

一个好的项目结构能让代码更清晰，后期维护也更方便。我们先规划一下目录和文件：

asian_beauty_api/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI应用主入口 │ ├── api.py # API路由定义 │ ├── models.py # Pydantic数据模型（请求/响应） │ ├── core/ │ │ ├── __init__.py │ │ ├── config.py # 配置文件 │ │ └── security.py # 认证相关逻辑 │ ├── services/ │ │ ├── __init__.py │ │ └── inference.py # 模型推理服务封装 │ └── utils/ │ ├── __init__.py │ └── image_utils.py # 图片处理工具函数 ├── requirements.txt # 项目依赖列表 └── README.md # 项目说明

你可以先创建这些目录和空的__init__.py文件。我们接下来会一步步填充它们。

2. 设计API数据模型

API的核心是数据交换的约定。使用Pydantic来定义数据模型，能自动处理请求数据的验证、序列化和文档生成，非常省心。

打开app/models.py文件，我们来定义文生图（Text-to-Image）和图生图（Image-to-Image）的请求和响应模型。

from pydantic import BaseModel, Field, HttpUrl from typing import Optional, List, Union from enum import Enum # 定义一个枚举，列出支持的图片尺寸，避免用户输入无效值 class ImageSize(str, Enum): SQUARE_512 = "512x512" SQUARE_768 = "768x768" PORTRAIT_768 = "512x768" LANDSCAPE_768 = "768x512" # 文生图请求模型 class TextToImageRequest(BaseModel): """文生图请求参数""" prompt: str = Field(..., description="描述生成图片内容的文本提示词", example="一位优雅的亚洲女性，在樱花树下，电影质感") negative_prompt: Optional[str] = Field(None, description="不希望出现在图片中的内容", example="模糊，低质量，多余的手指") size: ImageSize = Field(default=ImageSize.SQUARE_768, description="生成图片的尺寸") num_images: int = Field(default=1, ge=1, le=4, description="一次性生成的图片数量，范围1-4") guidance_scale: float = Field(default=7.5, ge=1.0, le=20.0, description="指导强度，值越高越贴近提示词") num_inference_steps: int = Field(default=20, ge=1, le=100, description="去噪步数，影响生成质量和时间") # 使用Pydantic的配置类为模型生成更友好的schema示例 class Config: schema_extra = { "example": { "prompt": "一位优雅的亚洲女性，在樱花树下，电影质感", "negative_prompt": "模糊，低质量", "size": "768x768", "num_images": 2, "guidance_scale": 7.5, "num_inference_steps": 20 } } # 图生图请求模型 class ImageToImageRequest(BaseModel): """图生图请求参数""" prompt: str = Field(..., description="描述生成图片内容的文本提示词") init_image_url: Optional[HttpUrl] = Field(None, description="初始图片的URL地址") # 注意：实际中我们可能通过文件上传接收图片，这里用URL示例，文件上传会在路由中处理 strength: float = Field(default=0.75, ge=0.0, le=1.0, description="参考原图的强度，0.0完全忽略原图，1.0基本保持原图") negative_prompt: Optional[str] = None size: Optional[ImageSize] = None # 图生图可继承原图尺寸，也可指定 guidance_scale: float = Field(default=7.5, ge=1.0, le=20.0) num_inference_steps: int = Field(default=20, ge=1, le=100) # 统一的图片生成响应模型 class GeneratedImage(BaseModel): """单张生成图片的信息""" image_id: str = Field(..., description="图片唯一标识") image_url: Optional[str] = Field(None, description="图片访问URL（如果存储到服务器）") image_b64: Optional[str] = Field(None, description="图片的Base64编码字符串（直接返回时用）") class ImageGenerationResponse(BaseModel): """图片生成API的响应""" request_id: str = Field(..., description="本次请求的唯一ID") status: str = Field(..., description="请求状态，如'success', 'processing', 'failed'") images: List[GeneratedImage] = Field(..., description="生成的图片列表") processing_time: Optional[float] = Field(None, description="处理耗时，单位秒") model_info: dict = Field(default_factory=dict, description="模型相关信息")

这些模型定义了客户端需要发送什么数据，以及服务器会返回什么数据。Field函数里的description和example会自动出现在Swagger UI文档中，非常方便。

3. 封装模型推理服务

接下来是核心部分：如何调用你已经部署好的Asian Beauty Z-Image Turbo模型。我们在app/services/inference.py中创建一个服务类来封装这些逻辑。

这里的关键是，模型加载和推理可能比较耗时，我们要做好错误处理和资源管理。同时，考虑到API的并发请求，我们使用异步（async）方式来避免阻塞。

import asyncio import logging import time import base64 from io import BytesIO from typing import List, Optional from pathlib import Path # 假设你的模型调用方式如下（请根据你的实际部署调整） # 例如，如果你用的是diffusers库： # from diffusers import StableDiffusionPipeline # 或者是你自己封装的一个推理函数 # 这里我们用一个伪代码示例来说明结构 class ModelInferenceService: """模型推理服务封装""" def __init__(self, model_path: str): """ 初始化服务，加载模型。 model_path: 模型在本地磁盘的路径 """ self.logger = logging.getLogger(__name__) self.model_path = Path(model_path) self.model = None self.device = "cuda" # 或 "cpu"，根据你的环境 self._load_model() def _load_model(self): """加载模型到内存/显存""" self.logger.info(f"正在从 {self.model_path} 加载模型...") try: # 这里是伪代码，替换为你的实际模型加载逻辑 # 例如使用diffusers: # from diffusers import StableDiffusionPipeline # self.model = StableDiffusionPipeline.from_pretrained( # self.model_path, # torch_dtype=torch.float16 if self.device == "cuda" else torch.float32 # ).to(self.device) # 或者，如果你是通过openclaw部署的，可能有一个现成的API客户端或本地调用接口 # self.model = YourLocalModelWrapper(self.model_path) self.logger.info("模型加载成功。") # 模拟加载成功 self.model = {"status": "loaded", "name": "Asian Beauty Z-Image Turbo"} except Exception as e: self.logger.error(f"模型加载失败: {e}") raise RuntimeError(f"无法加载模型: {e}") async def text_to_image( self, prompt: str, negative_prompt: Optional[str] = None, height: int = 768, width: int = 768, num_images: int = 1, guidance_scale: float = 7.5, num_inference_steps: int = 20, **kwargs ) -> List[BytesIO]: """ 文生图异步推理函数。 返回一个BytesIO对象列表，每个对象包含一张生成的图片数据。 """ self.logger.info(f"开始文生图推理，提示词: {prompt[:50]}...") start_time = time.time() try: # 将同步的模型调用放到线程池中执行，避免阻塞事件循环 loop = asyncio.get_event_loop() images = await loop.run_in_executor( None, # 使用默认的线程池执行器 self._run_text_to_image_sync, prompt, negative_prompt, height, width, num_images, guidance_scale, num_inference_steps ) processing_time = time.time() - start_time self.logger.info(f"文生图推理完成，耗时: {processing_time:.2f}秒，生成 {len(images)} 张图片。") return images except Exception as e: self.logger.error(f"文生图推理出错: {e}") raise def _run_text_to_image_sync( self, prompt: str, negative_prompt: Optional[str], height: int, width: int, num_images: int, guidance_scale: float, num_inference_steps: int, ) -> List[BytesIO]: """ 同步的文生图推理函数，实际调用模型的地方。 注意：这个函数会在单独的线程中运行。 """ # 这里是伪代码，替换为你的实际模型调用逻辑 # 例如使用diffusers: # result = self.model( # prompt=prompt, # negative_prompt=negative_prompt, # height=height, # width=width, # num_images_per_prompt=num_images, # guidance_scale=guidance_scale, # num_inference_steps=num_inference_steps, # ).images # 模拟生成过程 import PIL.Image as Image import numpy as np simulated_images = [] for i in range(num_images): # 创建一个模拟的图片（实际中这里应调用模型） # 例如：image = result[i] # 将PIL Image转换为BytesIO img_byte_arr = BytesIO() # 创建一个简单的彩色图片作为示例 dummy_img = Image.new('RGB', (width, height), color=(i*50, 100, 150)) dummy_img.save(img_byte_arr, format='PNG') img_byte_arr.seek(0) simulated_images.append(img_byte_arr) return simulated_images async def image_to_image( self, prompt: str, init_image: BytesIO, # 上传的图片文件 strength: float = 0.75, negative_prompt: Optional[str] = None, guidance_scale: float = 7.5, num_inference_steps: int = 20, **kwargs ) -> List[BytesIO]: """图生图异步推理函数""" # 实现逻辑与文生图类似，增加对初始图片的处理 # 1. 将BytesIO转换为PIL Image或模型需要的格式 # 2. 调用模型的图生图功能 # 3. 返回结果 # 此处省略具体实现，结构同text_to_image pass @staticmethod def image_to_base64(image_bytes_io: BytesIO) -> str: """将BytesIO图片转换为Base64字符串""" image_bytes_io.seek(0) img_data = image_bytes_io.read() base64_str = base64.b64encode(img_data).decode('utf-8') return f"data:image/png;base64,{base64_str}" # 创建全局服务实例（在实际应用中，你可能需要考虑更复杂的生命周期管理） # 假设模型路径在环境变量或配置文件中 MODEL_PATH = "./models/asian_beauty_z_image_turbo" # 替换为你的实际路径 inference_service = ModelInferenceService(MODEL_PATH)

这个服务类做了几件关键事：一是封装了模型加载和调用细节；二是使用异步函数来处理耗时的推理任务，避免阻塞FastAPI的事件循环；三是提供了将图片转换为Base64格式的实用方法，方便在API响应中直接返回图片数据。

4. 构建FastAPI应用与路由

现在，我们把数据模型和推理服务组合起来，通过FastAPI暴露成HTTP接口。打开app/main.py和app/api.py。

首先，在app/main.py中创建FastAPI应用实例，并设置一些元数据，这些会显示在Swagger文档中。

from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware import logging from app.api import router as api_router from app.core.config import settings # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # 创建FastAPI应用实例 app = FastAPI( title="Asian Beauty Z-Image Turbo API", description="基于FastAPI封装的Asian Beauty Z-Image Turbo模型推理服务，支持文生图、图生图等功能。", version="1.0.0", docs_url="/docs", # Swagger UI地址 redoc_url="/redoc", # ReDoc地址 ) # 添加CORS中间件，允许前端跨域请求（根据需求调整） app.add_middleware( CORSMiddleware, allow_origins=["*"], # 生产环境应指定具体域名 allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 包含API路由 app.include_router(api_router, prefix="/api/v1") @app.on_event("startup") async def startup_event(): """应用启动时执行""" logger.info("Asian Beauty Z-Image Turbo API 服务启动中...") # 可以在这里进行一些初始化检查，比如模型是否加载成功 # from app.services.inference import inference_service # if inference_service.model is None: # logger.error("模型未正确加载，服务可能无法正常工作。") @app.on_event("shutdown") async def shutdown_event(): """应用关闭时执行""" logger.info("API服务正在关闭...") # 可以在这里进行资源清理 @app.get("/") async def root(): """根路径，返回简单的服务信息""" return { "service": "Asian Beauty Z-Image Turbo API", "version": "1.0.0", "docs": "/docs", "health_check": "/health" } @app.get("/health") async def health_check(): """健康检查端点""" # 可以添加更详细的健康状态，如模型状态、数据库连接等 return {"status": "healthy", "model_loaded": True}

接下来，在app/api.py中定义具体的API路由。这里我们会创建文生图和图生图两个核心端点。

from fastapi import APIRouter, HTTPException, UploadFile, File, Form, Depends, BackgroundTasks from fastapi.responses import JSONResponse import uuid import time from typing import Optional from app.models import ( TextToImageRequest, ImageToImageRequest, ImageGenerationResponse, GeneratedImage, ImageSize ) from app.services.inference import inference_service from app.core.security import verify_api_key # 我们稍后会实现简单的认证 router = APIRouter() # 一个简单的内存缓存，用于存储任务状态（生产环境建议用Redis等） task_status_cache = {} @router.post("/generate/text-to-image", response_model=ImageGenerationResponse) async def generate_image_from_text( request: TextToImageRequest, # 使用Depends添加API Key认证（可选） # api_key: str = Depends(verify_api_key) ): """ 文生图接口。 根据文本描述生成图片。 """ request_id = str(uuid.uuid4()) try: start_time = time.time() # 调用异步推理服务 images_bytes = await inference_service.text_to_image( prompt=request.prompt, negative_prompt=request.negative_prompt, height=int(request.size.value.split('x')[1]), # 从"768x768"解析 width=int(request.size.value.split('x')[0]), num_images=request.num_images, guidance_scale=request.guidance_scale, num_inference_steps=request.num_inference_steps, ) processing_time = time.time() - start_time # 构建响应数据 generated_images = [] for idx, img_bytes in enumerate(images_bytes): # 这里可以选择将图片保存到文件服务器并返回URL，或者直接返回Base64 # 示例：返回Base64 image_b64 = inference_service.image_to_base64(img_bytes) generated_images.append( GeneratedImage( image_id=f"{request_id}_{idx}", image_b64=image_b64, # image_url=f"https://your-cdn.com/images/{request_id}_{idx}.png" # 如果保存到文件服务器 ) ) response = ImageGenerationResponse( request_id=request_id, status="success", images=generated_images, processing_time=round(processing_time, 2), model_info={"name": "Asian Beauty Z-Image Turbo", "version": "1.0"} ) return response except Exception as e: raise HTTPException(status_code=500, detail=f"图片生成失败: {str(e)}") @router.post("/generate/image-to-image") async def generate_image_from_image( prompt: str = Form(..., description="描述生成图片内容的文本提示词"), init_image: UploadFile = File(..., description="初始图片文件"), strength: float = Form(0.75, ge=0.0, le=1.0), negative_prompt: Optional[str] = Form(None), guidance_scale: float = Form(7.5), num_inference_steps: int = Form(20), # api_key: str = Depends(verify_api_key) ): """ 图生图接口。 根据初始图片和文本描述生成新图片。 注意：这里使用Form和File处理混合数据（文件上传+表单字段）。 """ # 检查上传的文件类型 if not init_image.content_type.startswith("image/"): raise HTTPException(status_code=400, detail="上传的文件必须是图片格式") request_id = str(uuid.uuid4()) try: # 读取上传的图片文件 image_data = await init_image.read() from io import BytesIO init_image_bytes = BytesIO(image_data) # 调用图生图推理服务 # 注意：这里需要你实现image_to_image方法 # generated_images_bytes = await inference_service.image_to_image(...) # 模拟响应 processing_time = 2.5 # 模拟处理时间 # 构建响应（这里用文生图的响应结构模拟） response = ImageGenerationResponse( request_id=request_id, status="success", images=[GeneratedImage(image_id=request_id, image_b64="data:image/png;base64,...")], processing_time=processing_time, model_info={"name": "Asian Beauty Z-Image Turbo", "version": "1.0"} ) return response except Exception as e: raise HTTPException(status_code=500, detail=f"图片生成失败: {str(e)}") @router.post("/generate/async/text-to-image") async def generate_image_async( request: TextToImageRequest, background_tasks: BackgroundTasks, # api_key: str = Depends(verify_api_key) ): """ 异步文生图接口。 提交生成任务，立即返回任务ID，生成完成后通过回调或查询获取结果。 适用于生成时间较长的场景。 """ request_id = str(uuid.uuid4()) # 将任务状态初始化为处理中 task_status_cache[request_id] = { "status": "processing", "request": request.dict(), "result": None, "created_at": time.time() } # 将耗时的推理任务添加到后台任务中 background_tasks.add_task( process_async_generation, request_id, request ) return JSONResponse( status_code=202, # Accepted content={ "request_id": request_id, "status": "accepted", "message": "任务已提交，正在处理中", "check_status_url": f"/api/v1/tasks/{request_id}/status" } ) async def process_async_generation(request_id: str, request: TextToImageRequest): """后台处理异步生成任务""" try: # 这里调用实际的推理逻辑 # images_bytes = await inference_service.text_to_image(...) # 模拟处理耗时 await asyncio.sleep(5) # 更新任务状态为完成 task_status_cache[request_id].update({ "status": "completed", "result": {"message": "图片生成完成"}, # 这里应存储实际结果 "completed_at": time.time() }) except Exception as e: # 更新任务状态为失败 task_status_cache[request_id].update({ "status": "failed", "error": str(e), "completed_at": time.time() }) @router.get("/tasks/{task_id}/status") async def get_task_status(task_id: str): """查询异步任务状态""" if task_id not in task_status_cache: raise HTTPException(status_code=404, detail="任务不存在") task_info = task_status_cache[task_id] return { "task_id": task_id, "status": task_info["status"], "created_at": task_info["created_at"], "completed_at": task_info.get("completed_at"), "result": task_info.get("result"), "error": task_info.get("error") }

这段代码创建了几个关键端点：同步的文生图接口、图生图接口（支持文件上传），以及一个异步文生图接口。异步接口适用于生成时间可能较长的场景，它立即返回一个任务ID，客户端可以轮询任务状态或等待回调。

5. 添加基础认证与限流

开放API服务，安全和稳定性是必须考虑的。我们来实现一个简单的API Key认证和基于IP的请求限流。

在app/core/security.py中：

from fastapi import HTTPException, status, Request from fastapi.security import APIKeyHeader from typing import Optional import time # 简单的API Key验证（生产环境应从数据库或配置中心读取） VALID_API_KEYS = { "test_key_123": {"name": "测试客户端", "rate_limit": 10}, # 每分钟10次 "client_key_abc": {"name": "正式客户端A", "rate_limit": 60}, } # 存储请求记录，用于限流（生产环境建议用Redis） request_records = {} api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False) async def verify_api_key(api_key: Optional[str] = None) -> dict: """验证API Key""" if not api_key: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="API Key缺失", ) if api_key not in VALID_API_KEYS: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="无效的API Key", ) return VALID_API_KEYS[api_key] async def rate_limit_middleware(request: Request, call_next): """简单的速率限制中间件""" client_ip = request.client.host api_key = request.headers.get("x-api-key") # 如果提供了API Key，使用API Key的限流配置 if api_key and api_key in VALID_API_KEYS: client_id = f"api_key:{api_key}" limit = VALID_API_KEYS[api_key]["rate_limit"] else: # 否则使用IP地址限流（更严格） client_id = f"ip:{client_ip}" limit = 5 # 每分钟5次 current_time = time.time() minute_window = int(current_time / 60) # 每分钟一个窗口 # 初始化或清理旧记录 if client_id not in request_records: request_records[client_id] = {} # 清理一分钟前的记录 request_records[client_id] = { window: count for window, count in request_records[client_id].items() if window >= minute_window - 1 # 保留最近2个窗口 } # 获取当前窗口的请求计数 current_count = request_records[client_id].get(minute_window, 0) # 检查是否超过限制 if current_count >= limit: raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail=f"请求过于频繁，请稍后再试。限制: {limit}次/分钟", headers={"Retry-After": "60"} ) # 更新计数 request_records[client_id][minute_window] = current_count + 1 # 继续处理请求 response = await call_next(request) return response

然后，在app/main.py中注册这个中间件：

# 在app/main.py的FastAPI应用创建后添加 from app.core.security import rate_limit_middleware app.middleware("http")(rate_limit_middleware)

这样，你的API就具备了基础的认证和限流能力。在生产环境中，你可能需要更复杂的方案，比如使用Redis存储请求计数、集成OAuth2等。

6. 运行与测试API服务

一切就绪，让我们启动服务并测试一下。

6.1 启动服务

在项目根目录下，运行：

uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

--reload参数会在代码更改时自动重启服务，适合开发环境。看到类似下面的输出，说明服务启动成功：

INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit) INFO: Started reloader process [12345] using WatchFiles INFO: Started server process [12346] INFO: Waiting for application startup. INFO: Application startup complete.

6.2 访问交互式API文档

FastAPI自动生成了交互式API文档，这是它的一大亮点。打开浏览器，访问：

• Swagger UI:http://localhost:8000/docs
• ReDoc:http://localhost:8000/redoc

在Swagger UI界面，你可以看到所有定义好的API端点，点击"Try it out"按钮，可以直接在浏览器里测试API，无需额外工具。

6.3 使用Python客户端测试

你也可以写一个简单的Python脚本来测试API：

import requests import json import base64 from PIL import Image from io import BytesIO # API基础地址 BASE_URL = "http://localhost:8000/api/v1" # 测试文生图 def test_text_to_image(): url = f"{BASE_URL}/generate/text-to-image" payload = { "prompt": "一位优雅的亚洲女性，在樱花树下，电影质感，细节丰富", "negative_prompt": "模糊，低质量，多余的手指", "size": "768x768", "num_images": 1, "guidance_scale": 7.5, "num_inference_steps": 20 } headers = { "Content-Type": "application/json", # 如果启用了认证，需要添加API Key # "X-API-Key": "test_key_123" } response = requests.post(url, json=payload, headers=headers) if response.status_code == 200: result = response.json() print(f"请求ID: {result['request_id']}") print(f"处理时间: {result['processing_time']}秒") # 如果有Base64图片数据，可以解码查看 if result['images'] and result['images'][0].get('image_b64'): # 提取Base64数据（去掉data:image/png;base64,前缀） base64_data = result['images'][0]['image_b64'].split(',')[1] image_data = base64.b64decode(base64_data) # 用PIL打开图片 image = Image.open(BytesIO(image_data)) image.show() # 显示图片 # 或者保存到文件 # image.save("generated_image.png") return result else: print(f"请求失败: {response.status_code}") print(response.text) return None # 测试图生图（需要准备一张图片文件） def test_image_to_image(image_path: str): url = f"{BASE_URL}/generate/image-to-image" with open(image_path, "rb") as f: files = {"init_image": f} data = { "prompt": "将图片转换为动漫风格", "strength": 0.6, "guidance_scale": 7.5, } headers = { # "X-API-Key": "test_key_123" } response = requests.post(url, files=files, data=data, headers=headers) if response.status_code == 200: result = response.json() print(f"图生图成功: {result}") return result else: print(f"请求失败: {response.status_code}") print(response.text) return None if __name__ == "__main__": # 测试文生图 print("测试文生图接口...") result = test_text_to_image() # 测试图生图（需要准备一张图片） # print("\n测试图生图接口...") # test_image_to_image("input_image.jpg")

6.4 使用cURL测试

如果你习惯命令行，也可以用cURL测试：

# 测试文生图 curl -X POST "http://localhost:8000/api/v1/generate/text-to-image" \ -H "Content-Type: application/json" \ -d '{ "prompt": "一位优雅的亚洲女性，在樱花树下，电影质感", "size": "768x768", "num_images": 1 }' # 测试异步接口 curl -X POST "http://localhost:8000/api/v1/generate/async/text-to-image" \ -H "Content-Type: application/json" \ -d '{ "prompt": "测试异步生成", "size": "512x512" }' # 查询异步任务状态（替换{task_id}为实际ID） curl "http://localhost:8000/api/v1/tasks/{task_id}/status"

7. 部署与优化建议

当你的API在本地测试通过后，就可以考虑部署到生产环境了。这里有一些建议：

7.1 使用Gunicorn部署（Linux/macOS）

对于生产环境，建议使用Gunicorn作为WSGI服务器，配合Uvicorn工作进程：

pip install gunicorn gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000

• -w 4: 使用4个工作进程
• -k uvicorn.workers.UvicornWorker: 使用Uvicorn工作器

7.2 使用Docker容器化

创建Dockerfile可以方便部署：

FROM python:3.9-slim WORKDIR /app # 复制依赖文件并安装 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY app/ ./app/ # 暴露端口 EXPOSE 8000 # 启动命令 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

然后构建和运行：

docker build -t asian-beauty-api . docker run -p 8000:8000 asian-beauty-api

7.3 性能优化建议

1. 模型预热：在应用启动时预加载模型，避免第一次请求时加载。
2. 结果缓存：对相同的请求参数缓存生成结果，减少重复计算。
3. 异步处理：对于耗时操作，使用Celery或RQ等任务队列，避免阻塞API。
4. CDN存储：生成的图片可以存储到对象存储（如S3、MinIO）并通过CDN分发，而不是直接返回Base64。
5. 监控与日志：集成Prometheus监控和结构化日志（如JSON格式），方便问题排查。
6. 配置管理：使用环境变量或配置文件管理敏感信息（如API Keys、模型路径）。

7.4 安全加固

1. HTTPS：生产环境一定要使用HTTPS。
2. API Key轮换：定期更换API Key。
3. 输入验证：除了Pydantic验证，还可以添加额外的安全检查。
4. 请求日志：记录所有请求，但注意不要记录敏感数据。
5. 限流策略：根据业务需求调整限流策略，可以按用户、按IP、按端点等多维度限流。

8. 总结

走完这一趟，你应该已经成功将你的Asian Beauty Z-Image Turbo模型封装成了一个标准的RESTful API服务。我们从头搭建了项目结构，设计了清晰的数据模型，封装了模型推理逻辑，实现了同步和异步的API端点，还添加了基础的认证和限流功能。

整个过程下来，你会发现FastAPI确实让API开发变得简单高效。它的自动文档生成、数据验证、依赖注入等特性，大大减少了重复工作。而将模型能力通过API暴露出来，带来的好处是显而易见的：前后端分离、多语言调用、服务化部署都变得可能。

当然，这只是一个起点。在实际生产环境中，你可能还需要考虑更多方面，比如更完善的错误处理、更细粒度的权限控制、更强大的监控告警等。但有了这个基础框架，后续的扩展就会顺利很多。

最后，别忘了测试你的API，确保它在各种边界条件下都能稳定工作。然后，你就可以放心地把它集成到你的应用里，或者分享给团队成员使用了。模型的能力不再局限于你的本地环境，而是可以通过网络被任何授权的客户端调用，这才是AI模型真正发挥价值的方式。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。