Wan2.2-T2V-A5B后端服务开发：Python FastAPI框架高性能接口封装

你是不是也遇到过这样的情况？好不容易把一个强大的AI模型跑起来了，比如这个能根据文字生成视频的Wan2.2-T2V-A5B，但每次想用都得打开命令行，输入一堆参数，等半天才能看到结果。更别提想分享给同事或者集成到自己的应用里了，简直无从下手。

其实，给模型套上一个好用、稳定的“外壳”，让它变成一个随时可以调用的服务，才是真正发挥价值的关键。今天，我就来手把手带你，用Python里风头正劲的FastAPI框架，为Wan2.2-T2V-A5B模型打造一个生产级的RESTful API服务。我们不只是简单包装一下，而是要解决真实场景下的问题：怎么处理动辄几分钟的视频生成任务而不让请求超时？怎么保证服务在高并发下稳定运行？又怎么让团队其他成员能轻松使用和调试？

跟着这篇教程走完，你将得到一个具备异步任务处理、自动API文档、请求鉴权限流，并且能一键Docker化部署的完整后端服务。咱们不搞花架子，直接上干货，从零开始，一步步实现。

1. 项目蓝图与环境搭建

在动手写代码之前，我们先得把“厨房”收拾好。明确我们要做什么，以及需要哪些工具。

我们的目标很明确：构建一个Web服务，它接收一段文本描述，然后调用Wan2.2-T2V-A5B模型生成视频，最后把生成好的视频文件返回给用户。由于视频生成是个耗时操作，我们不能让用户在前端干等着，所以需要引入“异步任务”的概念。同时，服务要健壮、易维护、好扩展。

1.1 核心工具选型

这里是我们今天要用到的“兵器谱”：

• FastAPI: 作为我们的Web框架。它快如闪电（基于Starlette），写起来简单直观，还能自动生成漂亮的交互式API文档（Swagger UI），对开发者非常友好。
• Celery: 负责处理后台异步任务。当用户提交一个生成请求，FastAPI会快速响应“收到，正在处理”，然后把繁重的生成任务丢给Celery在后台慢慢跑。Celery会管理这些任务的队列、执行和状态。
• Redis: 作为Celery的“消息代理”和“结果后端”。简单理解，它就是Celery的任务中转站和任务结果的临时仓库。
• Uvicorn: 一个轻量级、超快的ASGI服务器，用来运行我们的FastAPI应用。
• Docker: 最后，我们用Docker把整个应用和环境打包，实现“一次构建，处处运行”。

1.2 一步到位的环境安装

假设你已经有了Python环境（建议3.8以上），我们创建一个新的项目目录，并安装所有依赖。打开你的终端，执行以下命令：

# 创建项目目录并进入
mkdir wan-t2v-api && cd wan-t2v-api

# 创建并激活虚拟环境（可选但推荐）
python -m venv venv
# Windows: venv\Scripts\activate
# Mac/Linux: source venv/bin/activate

# 安装核心依赖
pip install fastapi uvicorn

# 安装异步任务相关依赖
pip install celery redis

# 安装模型交互可能需要的依赖（根据Wan2.2-T2V-A5B的实际调用方式调整）
# 例如，如果是Hugging Face Transformers库
# pip install transformers torch

# 安装其他工具库
pip install python-multipart pydantic-settings

现在，你的基础环境就准备好了。我们还需要确保Redis服务已经启动并运行在本地（默认端口6379）。你可以通过Docker快速启动一个Redis：

docker run -d -p 6379:6379 --name redis-stack redis/redis-stack:latest

2. 从零构建FastAPI应用骨架

有了工具，我们开始搭房子的主体结构。首先创建项目的主要文件。

2.1 创建应用入口与配置

创建一个名为 main.py 的文件，这是FastAPI应用的启动入口。

# main.py
from fastapi import FastAPI, BackgroundTasks, HTTPException, Depends
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import Optional
import uuid
from datetime import datetime

# 导入我们将要创建的其他模块
from tasks import generate_video_task
from auth import verify_token
from config import settings

# 初始化FastAPI应用
app = FastAPI(
    title="Wan2.2-T2V-A5B 视频生成API服务",
    description="基于FastAPI封装的文本生成视频模型高性能接口",
    version="1.0.0"
)

# 添加CORS中间件，方便前端调用
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境应替换为具体的前端地址
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 定义请求数据模型
class VideoGenRequest(BaseModel):
    """视频生成请求体"""
    prompt: str = Field(..., min_length=5, max_length=500, description="视频描述文本")
    negative_prompt: Optional[str] = Field(None, description="不希望出现在视频中的内容")
    duration_seconds: Optional[int] = Field(5, ge=1, le=30, description="视频时长（秒）")
    resolution: Optional[str] = Field("512x512", description="视频分辨率，如 '512x512'")

# 定义响应数据模型
class TaskResponse(BaseModel):
    """任务提交响应"""
    task_id: str
    status: str
    message: str
    submit_time: datetime

class TaskStatusResponse(BaseModel):
    """任务状态查询响应"""
    task_id: str
    status: str  # PENDING, STARTED, SUCCESS, FAILURE
    result: Optional[str] = None  # 成功时为视频文件URL或路径，失败时为错误信息
    created_at: datetime
    updated_at: Optional[datetime] = None

# 一个简单的内存字典，用于模拟任务状态存储（生产环境应用数据库，如Redis或SQL数据库）
task_status_store = {}

# 根路径，健康检查
@app.get("/")
async def root():
    return {"message": "Wan2.2-T2V-A5B API Service is running."}

@app.get("/health")
async def health_check():
    return {"status": "healthy", "timestamp": datetime.utcnow()}

2.2 实现核心生成接口

接下来，我们在 main.py 中继续添加最核心的文本生成视频接口。这个接口负责接收用户请求，创建异步任务，并立即返回一个任务ID。

# main.py (续)

@app.post("/api/v1/generate", response_model=TaskResponse)
async def generate_video(
    request: VideoGenRequest,
    background_tasks: BackgroundTasks,
    # 使用依赖注入进行简单的鉴权（示例）
    token: str = Depends(verify_token)
):
    """
    提交视频生成任务。
    由于生成过程耗时，本接口会立即返回一个任务ID，后续可通过该ID查询任务状态和结果。
    """
    # 生成唯一任务ID
    task_id = str(uuid.uuid4())
    current_time = datetime.utcnow()

    # 初始化任务状态
    task_status_store[task_id] = {
        "status": "PENDING",
        "request": request.dict(),
        "created_at": current_time,
        "updated_at": current_time
    }

    # 将任务加入Celery后台队列
    # 注意：这里我们调用的是Celery任务函数，但Celery会异步执行
    task = generate_video_task.apply_async(args=[task_id, request.dict()])

    # 将Celery的异步任务ID也关联存储，方便后续查询
    task_status_store[task_id]["celery_task_id"] = task.id

    return TaskResponse(
        task_id=task_id,
        status="accepted",
        message="视频生成任务已提交，请使用task_id查询进度。",
        submit_time=current_time
    )

@app.get("/api/v1/task/{task_id}", response_model=TaskStatusResponse)
async def get_task_status(task_id: str):
    """根据任务ID查询生成状态和结果"""
    task_info = task_status_store.get(task_id)
    if not task_info:
        raise HTTPException(status_code=404, detail="任务不存在")

    # 这里可以扩展：如果任务状态还是PENDING或STARTED，可以去查询Celery后端获取最新状态
    # 例如：task = AsyncResult(task_id, app=celery_app)
    #       task_info['status'] = task.status

    return TaskStatusResponse(
        task_id=task_id,
        status=task_info["status"],
        result=task_info.get("result"),
        created_at=task_info["created_at"],
        updated_at=task_info.get("updated_at")
    )

3. 集成Celery处理异步任务

现在，我们来处理重头戏——异步任务。创建 tasks.py 和 celery_app.py 文件。

3.1 配置Celery应用

首先创建 celery_app.py，配置Celery，并指定使用Redis作为消息队列。

# celery_app.py
from celery import Celery
import os

# 从环境变量或配置文件中读取Redis地址
redis_url = os.getenv("REDIS_URL", "redis://localhost:6379/0")

# 创建Celery应用实例
celery_app = Celery(
    "wan_t2v_worker",
    broker=redis_url,          # 消息代理，用于发送任务
    backend=redis_url,         # 结果后端，用于存储任务结果
)

# 配置Celery
celery_app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='UTC',
    enable_utc=True,
    # 设置任务路由（可选）
    task_routes = {
        'tasks.generate_video_task': {'queue': 'video_generation'},
    },
)

# 自动发现任务模块
celery_app.autodiscover_tasks(['tasks'])

3.2 编写视频生成任务

然后创建 tasks.py，这里包含实际调用Wan2.2-T2V-A5B模型的代码。注意：你需要根据该模型具体的Python调用方式来实现 run_model_inference 函数。

# tasks.py
from celery_app import celery_app
import time
import os
from datetime import datetime
from typing import Dict
import sys
sys.path.append('.')
from main import task_status_store  # 导入状态存储，实际生产应用数据库

# 模拟或实际调用模型的地方
def run_model_inference(prompt: str, negative_prompt: str, duration: int, resolution: str) -> str:
    """
    调用Wan2.2-T2V-A5B模型生成视频。
    此处为示例，你需要替换为真实的模型调用代码。
    返回生成视频的本地文件路径或云端存储URL。
    """
    print(f"[模型调用] 开始生成视频: prompt='{prompt}'")
    # 模拟一个耗时的生成过程
    time.sleep(30)  # 假设生成需要30秒

    # 示例：假设生成一个文件，实际中应保存到特定目录或上传到对象存储
    # 这里返回一个模拟路径
    output_filename = f"generated_video_{int(time.time())}.mp4"
    output_path = f"/app/outputs/{output_filename}"
    
    # 确保输出目录存在（示例）
    os.makedirs(os.path.dirname(output_path), exist_ok=True)
    # 实际应调用模型并保存文件到 output_path
    # from your_model_package import WanT2VModel
    # model = WanT2VModel()
    # model.generate(prompt=prompt, output_path=output_path, ...)

    print(f"[模型调用] 视频生成完成: {output_path}")
    return output_path

@celery_app.task(bind=True, name='tasks.generate_video_task')
def generate_video_task(self, task_id: str, request_data: Dict):
    """Celery后台任务：执行视频生成"""
    try:
        # 1. 更新任务状态为 STARTED
        if task_id in task_status_store:
            task_status_store[task_id]["status"] = "STARTED"
            task_status_store[task_id]["updated_at"] = datetime.utcnow()

        # 2. 解析请求参数
        prompt = request_data.get("prompt", "")
        negative_prompt = request_data.get("negative_prompt")
        duration = request_data.get("duration_seconds", 5)
        resolution = request_data.get("resolution", "512x512")

        # 3. 调用模型生成视频（核心）
        video_file_path = run_model_inference(prompt, negative_prompt, duration, resolution)

        # 4. 任务成功，更新状态和结果
        if task_id in task_status_store:
            task_status_store[task_id]["status"] = "SUCCESS"
            task_status_store[task_id]["result"] = video_file_path  # 或可访问的URL
            task_status_store[task_id]["updated_at"] = datetime.utcnow()

        return {"success": True, "file_path": video_file_path, "task_id": task_id}

    except Exception as e:
        # 5. 任务失败，更新状态
        error_msg = str(e)
        if task_id in task_status_store:
            task_status_store[task_id]["status"] = "FAILURE"
            task_status_store[task_id]["result"] = f"生成失败: {error_msg}"
            task_status_store[task_id]["updated_at"] = datetime.utcnow()
        # 通知Celery任务失败
        self.update_state(state='FAILURE', meta={'exc_type': type(e).__name__, 'exc_message': error_msg})
        raise  # 重新抛出异常，Celery会记录

4. 增强服务：鉴权、限流与配置

一个生产级的服务还需要安全性和可配置性。我们简单实现几个关键功能。

4.1 简单的请求鉴权

创建 auth.py，实现一个简单的API Key验证。

# auth.py
from fastapi import HTTPException, Security, Depends
from fastapi.security import APIKeyHeader
from typing import Optional

# 定义API Key头部名称
API_KEY_NAME = "X-API-Key"
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)

# 模拟一个有效的API Key存储（生产环境应从数据库或配置中心读取）
VALID_API_KEYS = {"your-secret-api-key-123456", "test-key-789"}

async def verify_token(api_key: Optional[str] = Security(api_key_header)):
    """验证API Key的依赖函数"""
    if not api_key:
        raise HTTPException(status_code=403, detail="未提供API Key")
    if api_key not in VALID_API_KEYS:
        raise HTTPException(status_code=403, detail="无效的API Key")
    return api_key

4.2 应用配置管理

创建 config.py，使用Pydantic来管理配置，方便从环境变量读取。

# config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    """应用配置"""
    app_name: str = "Wan2.2-T2V-A5B API"
    redis_url: str = "redis://localhost:6379/0"
    api_key: str = "your-secret-api-key-123456"
    # 模型相关配置
    model_cache_dir: str = "./model_cache"
    output_dir: str = "./outputs"
    # 限流配置
    rate_limit_per_minute: int = 10

    class Config:
        env_file = ".env"  # 从 .env 文件加载配置

settings = Settings()

4.3 添加请求速率限制

在 main.py 中，我们可以使用 slowapi 或 fastapi-limiter 来实现简单的限流。这里以 slowapi 为例（需额外安装 pip install slowapi）。

# 在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)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# 在需要限流的接口上添加装饰器
from slowapi.errors import RateLimitExceeded
@app.post("/api/v1/generate")
@limiter.limit("10/minute")  # 限制每分钟10次请求
async def generate_video(...):
    # ... 原有代码

5. 打包与部署：走向生产环境

代码写好了，怎么让它能在任何地方稳定运行呢？Docker化是最好的选择。

5.1 编写Dockerfile

在项目根目录创建 Dockerfile。

# Dockerfile
# 使用官方Python镜像作为基础
FROM python:3.10-slim

# 设置工作目录
WORKDIR /app

# 设置环境变量，防止Python输出被缓冲
ENV PYTHONUNBUFFERED=1

# 安装系统依赖（如果需要编译某些Python包）
RUN apt-get update && apt-get install -y \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# 复制依赖文件并安装
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 创建输出目录
RUN mkdir -p /app/outputs

# 暴露端口（FastAPI默认运行在8000端口）
EXPOSE 8000

# 启动命令：同时启动FastAPI服务和Celery Worker
# 注意：生产环境建议使用Supervisor或K8s分别管理两个进程
CMD uvicorn main:app --host 0.0.0.0 --port 8000 &
    celery -A celery_app worker --loglevel=info --concurrency=2 -Q video_generation

5.2 编写docker-compose.yml

为了更方便地管理多个服务（Web应用、Celery Worker、Redis），我们使用Docker Compose。创建 docker-compose.yml。

# docker-compose.yml
version: '3.8'

services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    command: redis-server --appendonly yes

  web:
    build: .
    ports:
      - "8000:8000"
    depends_on:
      - redis
    environment:
      - REDIS_URL=redis://redis:6379/0
    volumes:
      - ./outputs:/app/outputs  # 挂载输出目录，方便查看生成的文件
      - ./model_cache:/app/model_cache # 挂载模型缓存目录
    command: >
      sh -c "uvicorn main:app --host 0.0.0.0 --port 8000"

  worker:
    build: .
    depends_on:
      - redis
      - web
    environment:
      - REDIS_URL=redis://redis:6379/0
    volumes:
      - ./outputs:/app/outputs
      - ./model_cache:/app/model_cache
    command: celery -A celery_app worker --loglevel=info --concurrency=2 -Q video_generation

volumes:
  redis_data:

5.3 一键启动与测试

现在，一切就绪。在项目根目录下，执行一个命令就能启动所有服务：

docker-compose up --build

服务启动后，打开你的浏览器，访问 http://localhost:8000/docs。你会看到FastAPI自动生成的Swagger UI界面，所有接口一目了然，并且可以在这里直接进行测试。

1. 点击 /api/v1/generate 接口的 “Try it out” 按钮。
2. 输入一段文本描述，比如 “A cat playing with a ball of yarn in a sunny living room.”。
3. 点击 “Execute”。你会立刻收到一个包含 task_id 的响应。
4. 复制这个 task_id，在 /api/v1/task/{task_id} 接口中查询任务状态。一开始会是 PENDING 或 STARTED，等待一段时间（模拟的30秒）后，会变成 SUCCESS，并返回视频文件的路径。

6. 总结与后续

走完这一趟，我们从头构建了一个具备生产级潜力的AI模型服务。它不再是黑箱脚本，而是一个有清晰接口、能处理并发、有状态管理、并且易于部署的Web服务。FastAPI的简洁高效，加上Celery的异步任务能力，让处理Wan2.2-T2V-A5B这类耗时模型变得非常顺畅。

当然，这只是一个坚实的起点。在实际项目中，你可能还需要考虑更多，比如用数据库（如PostgreSQL）持久化任务状态、集成对象存储（如AWS S3或MinIO）来管理生成的视频文件、实现更完善的用户认证与授权、添加更细致的监控和日志。但有了这个框架，这些功能都可以像搭积木一样逐步添加进去。

下次当你有一个新的AI模型需要服务化时，不妨直接套用这个模板，它能帮你省下大量从零搭建的时间。希望这篇教程能切实地帮到你，如果有任何问题或新的想法，欢迎一起交流探讨。

---

获取更多AI镜像

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