AI 制作生成的专业工程化实践：统一生成流水线与代码实现

import os, asyncio, uuid
from enum import Enum
from typing import Any, Dict
from pydantic import BaseModel, Field
from openai import AsyncOpenAI

class Modality(str, Enum):
    text = "text"
    image = "image"
    video = "video"
    audio = "audio"

class GenTask(BaseModel):
    task_id: str = Field(default_factory=lambda: uuid.uuid4().hex)
    modality: Modality
    prompt: str
    params: Dict[str, Any] = {}
    status: str = "pending"
    retries: int = 0

queue: asyncio.Queue[GenTask] = asyncio.Queue()

async def submit(task: GenTask):
    await queue.put(task)
    return task.task_id

async def worker():
    while True:
        task = await queue.get()
        try:
            task.status = "running"
            result = await dispatch(task)
            result = audit(task, result)
            task.status = "succeeded"
            print(task.task_id, result)
        except Exception as e:
            task.retries += 1
            if task.retries < 3:
                await queue.put(task)
            else:
                task.status = "failed"
                print("failed", task.task_id, e)
        finally:
            queue.task_done()

client = AsyncOpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)

async def gen_text(task: GenTask) -> dict:
    resp = await client.chat.completions.create(
        model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
        messages=[{"role": "user", "content": task.prompt}],
        temperature=task.params.get("temperature", 0.7),
    )
    return {"text": resp.choices[0].message.content}

async def gen_image(task: GenTask) -> dict:
    # 实际调用即梦、通义万相、Stable Diffusion 等官方 API
    return {"image_url": "outputs/image.png"}

async def gen_video(task: GenTask) -> dict:
    # 实际调用可灵、即梦、Runway 等官方 API
    return {"video_path": "outputs/video.mp4"}

async def dispatch(task: GenTask) -> dict:
    if task.modality == Modality.text:
        return await gen_text(task)
    if task.modality == Modality.image:
        return await gen_image(task)
    if task.modality == Modality.video:
        return await gen_video(task)
    raise ValueError("不支持的模态")

BAD_WORDS = {"违法", "暴力", "色情"}  # 示例，生产环境应接入审核服务

def audit(task: GenTask, result: dict) -> dict:
    text = str(result)
    if any(w in text for w in BAD_WORDS):
        raise ValueError("审核未通过")
    return result

async def compose_video(image_pattern: str, audio: str, out: str):
    cmd = [
        "ffmpeg", "-y",
        "-framerate", "12",
        "-i", image_pattern,
        "-i", audio,
        "-c:v", "libx264",
        "-pix_fmt", "yuv420p",
        "-shortest", out,
    ]
    proc = await asyncio.create_subprocess_exec(*cmd)
    await proc.wait()

from fastapi import FastAPI, HTTPException
app = FastAPI(title="AI Generation Pipeline")

@app.on_event("startup")
async def startup():
    for _ in range(3):
        asyncio.create_task(worker())

@app.post("/tasks")
async def create_task(task: GenTask):
    if len(task.prompt) > 2000:
        raise HTTPException(400, "提示词过长")
    task_id = await submit(task)
    return {"task_id": task_id, "status": "pending"}