AIGC 创作大模型的工程化实践：多模态生成管线与可控性设计

from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Protocol

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

@dataclass
class GenRequest:
    modality: Modality
    prompt: str
    params: dict[str, Any] = field(default_factory=dict)
    seed: int | None = None

@dataclass
class GenResult:
    modality: Modality
    uri: str
    model: str
    tokens: int = 0
    cost: float = 0.0
    meta: dict[str, Any] = field(default_factory=dict)

class Generator(Protocol):
    def generate(self, req: GenRequest) -> GenResult: ...

import hashlib, json
from pathlib import Path

class PromptTemplate:
    def __init__(self, root: Path):
        self.root = root

    def render(self, name: str, version: str, **kw) -> tuple[str, str]:
        path = self.root / name / f"{version}.json"
        data = json.loads(path.read_text(encoding="utf-8"))
        fp = hashlib.sha256(
            json.dumps(data, sort_keys=True).encode()
        ).hexdigest()[:16]
        return data["template"].format(**kw), fp

TEMPLATE = """主体：{subject}
动作：{action}
风格：{style}
构图：{camera}
负面：文字、水印、Logo、真人、畸形、低清、复杂背景"""

from PIL import Image, ImageFilter
import numpy as np

def quality_score(path: Path) -> dict:
    img = Image.open(path).convert("RGBA")
    alpha = np.array(img.split()[-1])
    edges = np.array(img.convert("L").filter(ImageFilter.FIND_EDGES))
    ys, xs = np.where(alpha > 10)
    return {
        "sharpness": float(edges.var()),
        "coverage": float((alpha > 10).mean()),
        "margin_ok": bool(
            len(xs) and xs.min() > 4 and ys.min() > 4
            and xs.max() < img.width - 4
            and ys.max() < img.height - 4
        ),
        "size": img.size,
    }

def auto_filter(paths: list[Path], min_sharp: float = 80.0,
                min_cov: float = 0.05, max_cov: float = 0.85) -> list[Path]:
    return [
        p for p in paths
        if (s := quality_score(p))["sharpness"] >= min_sharp
        and min_cov <= s["coverage"] <= max_cov
        and s["margin_ok"]
    ]

import re

BAD = {"违法", "暴力", "色情", "歧视", "虚假", "最好", "第一"}
PII = re.compile(r"(\d{11}|\d{17}[\dXx]|[\w.+-]+@[\w-]+\.[\w.]+)")

def audit(text: str) -> None:
    if any(w in text for w in BAD):
        raise ValueError("含敏感或绝对化用语")
    if PII.search(text):
        raise ValueError("含个人信息，需脱敏")

def mask(text: str) -> str:
    return PII.sub("[REDACTED]", text)

def audit_commercial(meta: dict) -> list[str]:
    issues = []
    if not meta.get("model_license"):
        issues.append("缺少模型许可说明")
    if not meta.get("font_license"):
        issues.append("缺少字体授权")
    if meta.get("has_real_person") and not meta.get("portrait_license"):
        issues.append("含真人肖像但无授权")
    if meta.get("has_brand_logo"):
        issues.append("含品牌 Logo，需商标授权")
    return issues

import asyncio, logging, time, uuid

log = logging.getLogger("aigc")

async def run_one(req: GenRequest, gen: Generator,
                  sem: asyncio.Semaphore,
                  max_retries: int = 3) -> GenResult:
    async with sem:
        trace_id = uuid.uuid4().hex[:12]
        start = time.time()
        for attempt in range(max_retries + 1):
            try:
                result = gen.generate(req)
                audit(req.prompt)
                log.info("trace=%s modality=%s cost=%.2fs tokens=%d",
                         trace_id, req.modality, time.time() - start,
                         result.tokens)
                return result
            except Exception as e:
                if attempt >= max_retries:
                    raise
                await asyncio.sleep(2 ** attempt)
        raise RuntimeError("unreachable")

async def run_batch(requests: list[GenRequest], gen: Generator,
                    concurrency: int = 3) -> list[GenResult]:
    sem = asyncio.Semaphore(concurrency)
    return await asyncio.gather(
        *(run_one(r, gen, sem) for r in requests)
    )