Wan2.1-UMT5模型服务化：使用RESTful API对外提供视频生成能力

Wan2.1-UMT5模型服务化：使用RESTful API对外提供视频生成能力

你是不是已经用Wan2.1-UMT5的WebUI界面玩得不亦乐乎了？自己输入描述，看着它生成一段段有趣的视频，确实很有成就感。但有没有想过，如果能把这种能力开放出去，让其他程序、网站或者移动应用也能调用它来生成视频，那会是什么场景？

比如，你的内容创作平台想给用户提供“文字变视频”的快捷功能；或者你的电商后台需要批量生成商品展示短视频。每次都让用户或运营人员手动打开WebUI去操作，显然不现实。这时候，就需要一个标准的、程序能直接调用的接口——也就是我们常说的API。

这篇文章，我就来手把手带你，把那个部署好的Wan2.1-UMT5 WebUI，包装成一个专业的RESTful API服务。我们会从最基础的接口设计开始，一步步实现任务提交、结果查询，再加上必不可少的API密钥认证和访问频率控制，最后还会教你如何生成一份机器和人都能看懂的API文档。学完这篇，你就能让任何应用都轻松拥有视频生成的能力了。

1. 准备工作与环境确认

在开始动手写代码之前，我们得先确保“地基”是稳固的。这里的地基，就是你之前已经部署好的Wan2.1-UMT5 WebUI服务。

首先，打开你的终端或命令行，确认你的WebUI服务正在健康运行。通常，你启动它的命令可能是这样的：

python webui.py --port 7860

你应该能在终端看到服务正常启动的日志，没有报错。然后，打开浏览器，访问http://你的服务器IP:7860（如果是本地，就是http://localhost:7860）。如果熟悉的WebUI界面能正常加载，并且你可以成功用它生成一段视频，那么恭喜你，准备工作就完成了一大半。

接下来，我们需要思考API层和WebUI层的关系。我们不打算直接修改WebUI本身的代码，那样可能会引入不必要的复杂性。一个更清晰、更稳定的架构是：构建一个全新的API服务，这个服务作为“中间人”或“代理”。当它收到外部的API请求时，它再去“模拟”用户操作，调用后端的WebUI服务来完成视频生成任务。这种解耦的设计，让API服务的迭代和WebUI的升级可以互不影响。

为了实现这个架构，我们需要安装几个关键的Python库。新建一个干净的目录作为你的API项目文件夹，然后在里面创建一个requirements.txt文件，内容如下：

fastapi==0.104.1 uvicorn[standard]==0.24.0 pydantic==2.5.0 requests==2.31.0 python-jose[cryptography]==3.3.0 passlib[bcrypt]==1.7.4 slowapi==0.1.8 python-multipart==0.0.6

这些库各自扮演着重要角色：

• FastAPI和Uvicorn：是我们构建现代、高性能API的框架和服务器。
• Pydantic：用于数据验证和设置管理，确保接口传入传出的数据格式都是正确的。
• Requests：让我们的API服务能够去调用后端的WebUI服务。
• python-jose和passlib：用来实现JWT（JSON Web Token）令牌认证，管理API密钥。
• slowapi：帮我们轻松实现接口限流，防止被过度调用。
• python-multipart：处理文件上传（虽然本文主要讲文本生成视频，但为扩展预留）。

在项目目录下，运行pip install -r requirements.txt来安装它们。环境准备好，我们就可以开始设计最核心的接口了。

2. 核心接口设计与实现：异步任务处理

视频生成是个耗时的过程，不可能让调用方一直等着。所以，我们采用“异步任务”的模式。这就像你去打印店打印一份厚文件，店员不会让你干等着，而是给你一个取件号，你可以先去忙别的，过会儿再来凭号取件。

我们的API也将遵循这个模式，设计两个核心接口：

1. POST /api/v1/generate：提交一个视频生成任务。接口立即返回一个唯一的task_id，而不是视频本身。
2. GET /api/v1/result/{task_id}：通过任务ID来查询这个任务的执行状态和结果。

2.1 定义数据模型与任务状态机

我们先在项目里创建一个models.py文件，用Pydantic来定义清晰的数据结构，这能让代码更健壮，FastAPI还能自动基于它生成文档。

from pydantic import BaseModel, Field from typing import Optional, Literal from enum import Enum class TaskStatus(str, Enum): """任务状态枚举""" PENDING = "pending" # 排队中 PROCESSING = "processing" # 生成中 SUCCESS = "success" # 成功 FAILED = "failed" # 失败 class VideoGenerateRequest(BaseModel): """视频生成请求体""" prompt: str = Field(..., min_length=5, max_length=500, description="视频描述文本，至少5个字符") negative_prompt: Optional[str] = Field(None, description="不希望出现在视频中的内容描述") duration: int = Field(default=5, ge=2, le=30, description="视频时长（秒），范围2-30") # 这里可以添加更多Wan2.1-UMT5支持的参数，如尺寸、帧率等 # width: int = Field(default=512, ge=256, le=1024) # height: int = Field(default=512, ge=256, le=1024) class Config: schema_extra = { "example": { "prompt": "一只可爱的猫咪在草地上追逐蝴蝶，阳光明媚", "negative_prompt": "模糊，丑陋，多只猫", "duration": 8 } } class TaskResponse(BaseModel): """任务提交响应""" task_id: str status: TaskStatus message: str = "任务已提交，请使用task_id查询结果" estimated_wait_time: Optional[int] = Field(None, description="预计等待时间（秒）") class TaskResultResponse(BaseModel): """任务结果查询响应""" task_id: str status: TaskStatus message: Optional[str] = None video_url: Optional[str] = Field(None, description="视频文件访问URL，仅当status为success时存在") error_detail: Optional[str] = Field(None, description="如果失败，错误详情") created_at: str finished_at: Optional[str] = None

2.2 实现任务队列与处理器

由于可能有多个任务同时提交，我们需要一个简单的内存队列来管理它们（对于生产环境，你可能需要考虑使用Redis或RabbitMQ这样的专业队列）。同时，我们需要一个“工人”在后台不断地从队列中取任务，并调用真正的WebUI服务。

创建一个task_manager.py文件：

import asyncio import uuid import time import logging from typing import Dict from models import TaskStatus, VideoGenerateRequest, TaskResultResponse import requests import json logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class TaskManager: def __init__(self, webui_base_url: str = "http://localhost:7860"): self.webui_base_url = webui_base_url.rstrip('/') self.tasks: Dict[str, dict] = {} # 内存中存储任务状态和结果 self.task_queue = asyncio.Queue() self._stop_event = asyncio.Event() # 启动后台任务处理器 asyncio.create_task(self._process_task_queue()) async def submit_task(self, request: VideoGenerateRequest) -> str: """提交一个新任务，返回task_id""" task_id = str(uuid.uuid4()) now = time.strftime("%Y-%m-%d %H:%M:%S") self.tasks[task_id] = { "status": TaskStatus.PENDING, "request": request.dict(), "created_at": now, "finished_at": None, "result": None, "error": None } # 将任务放入队列 await self.task_queue.put(task_id) logger.info(f"任务 {task_id} 已提交并进入队列。") return task_id async def _process_task_queue(self): """后台任务处理循环""" while not self._stop_event.is_set(): try: task_id = await asyncio.wait_for(self.task_queue.get(), timeout=1.0) await self._execute_single_task(task_id) except asyncio.TimeoutError: continue except Exception as e: logger.error(f"任务处理器发生错误: {e}") async def _execute_single_task(self, task_id: str): """执行单个任务：调用WebUI API生成视频""" task_info = self.tasks.get(task_id) if not task_info: return task_info["status"] = TaskStatus.PROCESSING request_data = task_info["request"] try: logger.info(f"开始处理任务 {task_id}: {request_data.get('prompt')[:50]}...") # 这里是关键：模拟WebUI的调用。 # 你需要根据Wan2.1-UMT5 WebUI实际提供的内部API或自动化方式来调用。 # 这里是一个假设的示例，实际情况可能需要分析WebUI的网络请求。 # 方法一：如果WebUI有内置的API（如--api启动参数） # api_url = f"{self.webui_base_url}/run/predict" # payload = { # "data": [ # request_data.get("prompt"), # request_data.get("negative_prompt", ""), # request_data.get("duration", 5) # ] # } # response = requests.post(api_url, json=payload) # 方法二：更通用的，使用requests模拟表单提交（如果WebUI是Gradio） # 你需要找到Gradio应用对应的API端点，通常是 /api/predict/ payload = { "data": json.dumps([ request_data.get("prompt"), request_data.get("negative_prompt", ""), request_data.get("duration", 5) ]) } # 注意：Gradio的API调用可能需要session或特定的headers，请根据实际情况调整。 response = requests.post(f"{self.webui_base_url}/api/predict/", data=payload, timeout=300) # 设置长超时 if response.status_code == 200: result = response.json() # 假设返回的data里包含视频文件路径或Base64数据 # 这里需要你解析Wan2.1-UMT5的实际返回格式 # 例如，它可能返回一个临时文件URL或Base64字符串 video_data = result.get("data", [])[0] if result.get("data") else None if video_data and isinstance(video_data, str) and video_data.startswith(‘http’): video_url = video_data else: # 如果不是URL，你可能需要将Base64数据保存为文件，并生成一个可访问的URL # 这里简化处理，假设我们有一个文件服务地址 video_url = f"/api/v1/videos/{task_id}.mp4" # 实际应将视频文件保存到磁盘，并记录路径 task_info["status"] = TaskStatus.SUCCESS task_info["result"] = {"video_url": video_url} task_info["finished_at"] = time.strftime("%Y-%m-%d %H:%M:%S") logger.info(f"任务 {task_id} 处理成功。") else: raise Exception(f"WebUI调用失败，状态码: {response.status_code}, 响应: {response.text}") except Exception as e: logger.error(f"处理任务 {task_id} 时出错: {e}") task_info["status"] = TaskStatus.FAILED task_info["error"] = str(e) task_info["finished_at"] = time.strftime("%Y-%m-%d %H:%M:%S") def get_task_result(self, task_id: str) -> Optional[TaskResultResponse]: """获取任务结果""" task_info = self.tasks.get(task_id) if not task_info: return None return TaskResultResponse( task_id=task_id, status=task_info["status"], message="任务已完成" if task_info["status"] in [TaskStatus.SUCCESS, TaskStatus.FAILED] else "任务处理中", video_url=task_info.get("result", {}).get("video_url") if task_info["status"] == TaskStatus.SUCCESS else None, error_detail=task_info.get("error"), created_at=task_info["created_at"], finished_at=task_info["finished_at"] ) async def stop(self): """停止任务处理器""" self._stop_event.set() # 全局任务管理器实例 task_manager = TaskManager()

重要提示：上面代码中_execute_single_task方法里调用WebUI的部分是关键，也是最需要你根据Wan2.1-UMT5 WebUI的实际接口进行调整的地方。你可能需要：

1. 查阅Wan2.1-UMT5的文档，看是否提供了直接的API。
2. 打开浏览器开发者工具（F12），在WebUI界面上操作一次生成，观察它向服务器发送了什么样的网络请求，然后模仿这个请求。

2.3 构建FastAPI主应用与核心接口

现在，让我们创建主应用文件main.py，把刚才设计的接口实现出来。

from fastapi import FastAPI, HTTPException, Depends, status from fastapi.middleware.cors import CORSMiddleware from contextlib import asynccontextmanager import asyncio from models import VideoGenerateRequest, TaskResponse, TaskResultResponse, TaskStatus from task_manager import task_manager import time # 应用生命周期管理 @asynccontextmanager async def lifespan(app: FastAPI): # 启动时 print("API服务启动...") yield # 关闭时 print("API服务关闭，清理任务队列...") await task_manager.stop() await asyncio.sleep(1) # 等待任务处理器结束 app = FastAPI( title="Wan2.1-UMT5 视频生成API服务", description="提供异步视频生成能力，基于Wan2.1-UMT5模型。", version="1.0.0", lifespan=lifespan ) # 添加CORS中间件，允许前端应用调用 app.add_middleware( CORSMiddleware, allow_origins=["*"], # 生产环境应指定具体域名 allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.post("/api/v1/generate", response_model=TaskResponse, status_code=status.HTTP_202_ACCEPTED, # 202表示已接受处理 summary="提交视频生成任务", tags=["视频生成"]) async def submit_generation_task(request: VideoGenerateRequest): """ 提交一个视频生成任务。 由于视频生成需要时间，此接口会立即返回一个任务ID (task_id)。 你需要使用这个task_id去查询任务状态和获取结果。 """ try: task_id = await task_manager.submit_task(request) # 简单估算等待时间（例如：队列中任务数 * 平均处理时间） queue_size = task_manager.task_queue.qsize() estimated_wait = queue_size * 30 # 假设每个任务平均30秒 return TaskResponse( task_id=task_id, status=TaskStatus.PENDING, message="视频生成任务已加入处理队列。", estimated_wait_time=estimated_wait if estimated_wait > 0 else None ) except Exception as e: raise HTTPException(status_code=500, detail=f"提交任务失败: {str(e)}") @app.get("/api/v1/result/{task_id}", response_model=TaskResultResponse, summary="查询任务结果", tags=["任务查询"]) async def get_task_result(task_id: str): """ 根据任务ID查询视频生成任务的状态和结果。 - **task_id**: 提交任务时返回的唯一任务标识符。 """ result = task_manager.get_task_result(task_id) if not result: raise HTTPException(status_code=404, detail="任务ID不存在") return result if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

现在，你可以运行python main.py启动你的API服务了（默认在8000端口）。用工具（如curl或Postman）测试一下：

1. 提交任务：

   curl -X POST "http://localhost:8000/api/v1/generate" \ -H "Content-Type: application/json" \ -d '{ "prompt": "星空下的宁静湖泊，有流星划过", "duration": 10 }'

   你会得到一个包含task_id的响应。

2. 查询结果：

   curl "http://localhost:8000/api/v1/result/刚才得到的task_id"

   根据任务状态，你会看到pending,processing,success或failed的信息。

核心功能已经跑通了！但一个真正能对外提供的服务，还需要安全和控制措施。

3. 加固服务：认证、限流与文件服务

任何人都能随意调用你的API生成视频，可能会产生不必要的成本和安全风险。我们来加上两道“锁”。

3.1 API密钥认证

我们采用常见的Bearer Token（JWT）方式。创建一个auth.py文件。

from datetime import datetime, timedelta from typing import Optional from jose import JWTError, jwt from passlib.context import CryptContext from fastapi import HTTPException, status, Depends from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials import secrets # 用于演示，生产环境应从安全的环境变量或配置中心读取 SECRET_KEY = secrets.token_urlsafe(32) # 生成一个随机的密钥 ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7 # 令牌有效期（例如7天） # 模拟一个用户/API密钥数据库。生产环境应使用真实数据库。 fake_users_db = { "client_app_01": { "api_key": "sk_test_123456789abcdef", # 模拟的API Key "hashed_key": None, # 这里我们直接对比明文，生产环境应哈希存储 "is_active": True, } } pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") security = HTTPBearer() def verify_api_key(api_key: str) -> bool: """验证API Key是否有效""" for user, info in fake_users_db.items(): if info["is_active"] and info["api_key"] == api_key: return True return False def create_access_token(data: dict, expires_delta: Optional[timedelta] = None): """创建JWT访问令牌""" to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)): """依赖项：验证请求中的Bearer Token""" credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="无效的认证凭证", headers={"WWW-Authenticate": "Bearer"}, ) token = credentials.credentials try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) api_key: str = payload.get("sub") if api_key is None: raise credentials_exception except JWTError: raise credentials_exception if not verify_api_key(api_key): raise credentials_exception return api_key # 一个简单的登录端点，用于交换API Key为JWT Token（可选） @app.post("/api/v1/auth/token") async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()): # 这里form_data.username可以当作api_key传入 api_key = form_data.username if not verify_api_key(api_key): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="无效的API Key", ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={"sub": api_key}, expires_delta=access_token_expires ) return {"access_token": access_token, "token_type": "bearer"}

然后，修改main.py中的接口，添加依赖项：

from auth import get_current_user @app.post("/api/v1/generate", ...) async def submit_generation_task( request: VideoGenerateRequest, current_user: str = Depends(get_current_user) # 添加认证依赖 ): # 现在只有携带有效Token的请求才能调用 # 你可以用current_user记录是谁调用的 print(f"用户 {current_user} 提交了任务") # ... 其余代码不变 ...

3.2 接口访问限流

防止同一个API密钥在短时间内疯狂调用，拖垮服务。我们使用slowapi和redis（可选）来实现。这里先用内存限流演示。

在main.py中新增：

from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded # 初始化限流器 limiter = Limiter(key_func=get_remote_address) # 根据IP限流，也可用`lambda: current_user`根据用户限流 app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # 然后将限流装饰器加到需要限流的接口上 @app.post("/api/v1/generate") @limiter.limit("5/minute") # 限制每分钟最多5次调用 async def submit_generation_task(request: VideoGenerateRequest, current_user: str = Depends(get_current_user)): # ... 函数体不变 ...

3.3 提供生成的视频文件

任务成功后，我们返回了一个video_url。你需要实现这个文件服务端点，让用户能真正下载到视频。在main.py中添加：

from fastapi.responses import FileResponse import os # 假设视频文件保存在这个目录 VIDEO_STORAGE_PATH = "./generated_videos" @app.get("/api/v1/videos/{filename}") async def get_video_file(filename: str): """提供生成的视频文件下载""" file_path = os.path.join(VIDEO_STORAGE_PATH, filename) if not os.path.exists(file_path): raise HTTPException(status_code=404, detail="视频文件未找到") return FileResponse(file_path, media_type='video/mp4', filename=filename)

同时，记得在task_manager.py的_execute_single_task方法中，成功生成视频后，将文件保存到VIDEO_STORAGE_PATH目录下，并以task_id或其他唯一名称命名。

4. 生成清晰的API文档与总结

FastAPI的一个巨大优势就是能自动生成交互式API文档。你启动服务后，访问以下两个地址就能看到：

• http://localhost:8000/docs：Swagger UI提供的交互式文档，可以在这里直接尝试调用接口。
• http://localhost:8000/redoc：ReDoc提供的另一种风格的文档，更简洁美观。

为了让文档更专业，我们还可以导出OpenAPI规范文件，用于导入到其他API管理平台。在你的项目根目录创建一个generate_openapi.py脚本：

import json from main import app # 生成OpenAPI规范 openapi_schema = app.openapi() # 保存为JSON文件 with open("openapi.json", "w") as f: json.dump(openapi_schema, f, indent=2) print("OpenAPI规范已保存到 openapi.json")

运行这个脚本，你就会得到一个标准的openapi.json文件，它可以被Postman、Apifox等工具直接导入。

整套流程走下来，我们从一个只能手动操作的WebUI，构建出了一个具备生产级雏形的API服务。它具备了异步任务处理、认证鉴权、访问控制等关键特性。实际部署时，你还需要考虑更多，比如使用Nginx做反向代理和负载均衡、用Redis作为任务队列和限流存储、添加更详细的日志监控、以及制定清晰的API使用计费策略等。

开发过程中最关键的步骤，是厘清你的API服务如何与底层的Wan2.1-UMT5 WebUI进行通信，这需要你仔细研究WebUI的交互方式。一旦这个桥梁打通，剩下的就是标准的服务化工程实践了。希望这篇教程能为你打开一扇门，让你强大的模型能力可以更灵活、更广泛地服务于各种应用场景。

获取更多AI镜像

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