1. 项目概述
这个视频生成系统是一个典型的AI应用开发项目,它巧妙地将多个开源工具和框架整合在一起,构建了一个从文本输入到视频输出的完整工作流。作为一名长期从事AI应用开发的工程师,我发现这种架构设计在实际业务场景中非常实用,尤其适合需要快速搭建原型或中小规模部署的场景。
核心组件包括:
- ComfyUI:作为视频生成引擎,负责执行具体的AI模型推理工作
- Dify:提供大模型能力编排和Prompt工程管理
- LangChain:处理复杂的文本解析和结构化输出
- FastAPI:构建高效的后端服务
- Docker:实现环境标准化和快速部署
这种架构最大的优势在于它的模块化设计,每个组件都可以独立升级或替换。比如当有新的视频生成模型发布时,只需调整ComfyUI的工作流配置,而不需要改动其他部分的代码。
2. 环境准备与部署
2.1 服务器基础配置
在开始部署前,我们需要准备一台至少满足以下配置的服务器:
- Ubuntu 20.04/22.04 LTS
- NVIDIA显卡(建议RTX 3090及以上)
- 32GB以上内存
- 100GB以上可用磁盘空间
提示:如果预算有限,可以考虑使用云服务商的GPU实例,但需要注意视频生成对显存的要求较高,建议选择24GB以上显存的显卡。
2.2 组件安装与配置
2.2.1 Docker环境安装
bash复制# 卸载旧版本(如有)
sudo apt-get remove docker docker-engine docker.io containerd runc
# 安装依赖
sudo apt-get update
sudo apt-get install -y \
ca-certificates \
curl \
gnupg \
lsb-release
# 添加Docker官方GPG密钥
sudo mkdir -p /etc/apt/keyrings
curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo gpg --dearmor -o /etc/apt/keyrings/docker.gpg
# 设置仓库
echo \
"deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.gpg] https://download.docker.com/linux/ubuntu \
$(lsb_release -cs) stable" | sudo tee /etc/apt/sources.list.d/docker.list > /dev/null
# 安装Docker引擎
sudo apt-get update
sudo apt-get install -y docker-ce docker-ce-cli containerd.io docker-compose-plugin
# 验证安装
sudo docker run hello-world
2.2.2 NVIDIA容器工具包安装
bash复制# 添加NVIDIA容器工具包仓库
distribution=$(. /etc/os-release;echo $ID$VERSION_ID) \
&& curl -fsSL https://nvidia.github.io/libnvidia-container/gpgkey | sudo gpg --dearmor -o /usr/share/keyrings/nvidia-container-toolkit-keyring.gpg \
&& curl -fsSL https://nvidia.github.io/libnvidia-container/$distribution/libnvidia-container.list | \
sed 's#deb https://#deb [signed-by=/usr/share/keyrings/nvidia-container-toolkit-keyring.gpg] https://#g' | \
sudo tee /etc/apt/sources.list.d/nvidia-container-toolkit.list
# 安装NVIDIA容器运行时
sudo apt-get update
sudo apt-get install -y nvidia-container-toolkit
sudo nvidia-ctk runtime configure --runtime=docker
sudo systemctl restart docker
# 验证安装
sudo docker run --rm --gpus all nvidia/cuda:11.6.2-base-ubuntu20.04 nvidia-smi
3. 核心组件详解
3.1 ComfyUI工作流配置
ComfyUI的核心在于工作流(Workflow)的设计。对于视频生成,我们需要配置一个完整的工作流,通常包括以下几个关键节点:
- 文本编码器:将输入的文本转换为模型可理解的嵌入向量
- 视频生成模型:如Stable Video Diffusion或类似的模型
- 风格控制节点:调整视频的艺术风格
- 时长控制节点:设置生成视频的帧数和时长
- 后处理节点:包括降噪、色彩校正等处理
- 输出节点:指定视频保存路径和格式
一个典型的JSON工作流配置如下:
json复制{
"3": {
"inputs": {
"text": "海边日落",
"clip": ["CLIPTextEncode", 1]
},
"class_type": "CLIPTextEncode",
"_meta": {
"title": "正向提示词"
}
},
"5": {
"inputs": {
"ckpt_name": "svd_xt.safetensors",
"vae_name": "vae-ft-mse-840000-ema-pruned.safetensors"
},
"class_type": "CheckpointLoaderSimple",
"_meta": {
"title": "加载模型"
}
},
"7": {
"inputs": {
"width": 1024,
"height": 576,
"batch_size": 1,
"frames": 24,
"fps": 12,
"motion_bucket_id": 127,
"noise_aug_strength": 0.02
},
"class_type": "VideoDiffusionSettings",
"_meta": {
"title": "视频参数设置"
}
}
}
3.2 Dify应用配置
Dify的核心价值在于它提供了可视化的Prompt管理和工作流编排能力。在Dify中配置视频生成应用时,我们需要:
- 创建新的"文本生成"类型应用
- 在Prompt编排中添加以下关键组件:
- 用户输入解析模板
- 参数提取规则
- 输出格式定义
一个典型的Prompt模板如下:
code复制你是一个专业的视频参数生成助手。请根据用户描述生成视频生成所需的参数。
要求:
1. 主题:简洁描述视频内容,不超过10个字
2. 风格:从[治愈、科技、卡通、电影、写实]中选择最匹配的
3. 时长:整数,单位秒,范围5-30秒
4. 分辨率:从[480p,720p,1080p]中选择
用户需求:{{input}}
请以JSON格式返回结果,只包含主题、风格、时长、分辨率四个字段。
3.3 FastAPI后端优化
原始代码中的FastAPI实现已经比较完整,但在生产环境中还需要考虑以下优化点:
- 异步任务处理:使用Celery替代BackgroundTasks,支持任务重试和分布式处理
- API限流:防止恶意请求
- 认证鉴权:添加API密钥验证
- 日志记录:详细的请求和错误日志
- 性能监控:集成Prometheus指标
优化后的任务处理代码示例:
python复制from celery import Celery
from fastapi import FastAPI, HTTPException
from fastapi.security import APIKeyHeader
import logging
# 初始化Celery
celery_app = Celery(
'video_tasks',
broker='redis://redis:6379/0',
backend='redis://redis:6379/1'
)
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# API密钥验证
api_key_header = APIKeyHeader(name='X-API-Key')
app = FastAPI()
@app.post("/api/generate-video")
async def generate_video(
user_prompt: str,
x_api_key: str = Depends(api_key_header)
):
# 验证API密钥
if not validate_api_key(x_api_key):
raise HTTPException(status_code=403, detail="Invalid API Key")
try:
# 创建Celery任务
task = generate_video_task.delay(user_prompt)
logger.info(f"New video generation task: {task.id}")
return {
"code": 200,
"msg": "视频生成任务已提交",
"data": {"task_id": task.id}
}
except Exception as e:
logger.error(f"Task submission failed: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@celery_app.task(bind=True)
def generate_video_task(self, user_prompt: str):
try:
# 生成视频参数
params = generate_video_params(user_prompt)
# 调用ComfyUI生成视频
video_path = generate_video_with_comfyui(params, self.request.id)
return {
"status": "completed",
"video_path": video_path
}
except Exception as e:
logger.error(f"Video generation failed: {str(e)}")
raise self.retry(exc=e, countdown=60)
4. 高级功能实现
4.1 视频生成进度跟踪
在实际应用中,用户通常希望看到视频生成的实时进度。我们可以通过以下方式实现:
- ComfyUI进度API:ComfyUI提供了
/progress端点,可以获取当前任务的进度 - WebSocket推送:前端通过WebSocket连接获取实时进度更新
- Redis发布/订阅:后端将进度更新发布到Redis频道
实现代码示例:
python复制from fastapi import WebSocket
import json
import redis
# WebSocket进度推送
@app.websocket("/ws/progress/{task_id}")
async def websocket_progress(
websocket: WebSocket,
task_id: str
):
await websocket.accept()
redis_client = redis.Redis(host='redis', port=6379, db=0)
pubsub = redis_client.pubsub()
pubsub.subscribe(f'progress:{task_id}')
try:
while True:
message = pubsub.get_message(ignore_subscribe_messages=True)
if message:
progress = json.loads(message['data'])
await websocket.send_json(progress)
await asyncio.sleep(0.1)
except Exception as e:
logger.error(f"WebSocket error: {str(e)}")
finally:
await websocket.close()
pubsub.close()
# 在视频生成任务中添加进度推送
def generate_video_with_comfyui(params: dict, task_id: str) -> str:
# ...原有代码...
# 添加进度监控
def update_progress(progress):
redis_client = redis.Redis(host='redis', port=6379, db=0)
redis_client.publish(
f'progress:{task_id}',
json.dumps({
"progress": progress,
"status": "running"
})
)
# 在ComfyUI调用中添加进度回调
response = requests.post(
COMFYUI_API_URL,
json=comfyui_workflow,
headers={"Content-Type": "application/json"},
stream=True
)
# 解析进度更新
for line in response.iter_lines():
if line:
data = json.loads(line)
if "progress" in data:
update_progress(data["progress"])
# ...其余代码...
4.2 多模型支持
为了提供更丰富的视频生成选项,我们可以扩展系统以支持多种视频生成模型:
- 模型路由:根据用户需求自动选择最合适的模型
- 模型热加载:无需重启服务即可加载新模型
- 资源管理:根据模型需求动态分配GPU资源
实现示例:
python复制MODEL_REGISTRY = {
"svd": {
"name": "Stable Video Diffusion",
"workflow": "svd_workflow.json",
"requirements": {
"vram": 16,
"resolution": ["480p", "720p", "1080p"]
}
},
"animatediff": {
"name": "AnimateDiff",
"workflow": "animatediff_workflow.json",
"requirements": {
"vram": 12,
"resolution": ["480p", "720p"]
}
}
}
def select_model(params: dict) -> str:
# 根据参数选择最合适的模型
for model_id, model_info in MODEL_REGISTRY.items():
if params["分辨率"] in model_info["requirements"]["resolution"]:
return model_id
return "svd" # 默认回退
def generate_video_with_comfyui(params: dict, task_id: str) -> str:
# 选择模型
model_id = select_model(params)
model_info = MODEL_REGISTRY[model_id]
# 加载对应的工作流
with open(f"workflows/{model_info['workflow']}", "r") as f:
workflow = json.load(f)
# 更新工作流参数
workflow["prompt"]["3"]["inputs"]["text"] = params["主题"]
workflow["prompt"]["5"]["inputs"]["style"] = params["风格"]
# ...其他参数更新...
# 调用ComfyUI
response = requests.post(
COMFYUI_API_URL,
json=workflow,
headers={"Content-Type": "application/json"}
)
# ...其余代码...
5. 性能优化与监控
5.1 GPU资源管理
视频生成对GPU资源消耗很大,需要合理管理:
- 显存监控:实时监控GPU显存使用情况
- 任务排队:当显存不足时自动排队任务
- 自动清理:定期清理显存中的残留数据
实现代码:
python复制import pynvml
def get_gpu_status():
pynvml.nvmlInit()
device_count = pynvml.nvmlDeviceGetCount()
status = []
for i in range(device_count):
handle = pynvml.nvmlDeviceGetHandleByIndex(i)
mem_info = pynvml.nvmlDeviceGetMemoryInfo(handle)
status.append({
"index": i,
"total_mem": mem_info.total,
"used_mem": mem_info.used,
"free_mem": mem_info.free
})
pynvml.nvmlShutdown()
return status
def can_accept_task(required_vram: int) -> bool:
status = get_gpu_status()
for gpu in status:
if gpu["free_mem"] >= required_vram * 1024 * 1024 * 1024:
return True
return False
@app.post("/api/generate-video")
async def generate_video(user_prompt: str):
params = generate_video_params(user_prompt)
model_id = select_model(params)
required_vram = MODEL_REGISTRY[model_id]["requirements"]["vram"]
if not can_accept_task(required_vram):
raise HTTPException(
status_code=429,
detail="系统当前负载过高,请稍后再试"
)
# ...其余代码...
5.2 Prometheus监控集成
为了全面掌握系统运行状态,我们可以集成Prometheus监控:
- 指标定义:定义关键性能指标
- 中间件:添加FastAPI中间件收集请求指标
- 暴露端点:提供/metrics端点供Prometheus抓取
实现示例:
python复制from prometheus_client import Counter, Gauge, Histogram, generate_latest
from fastapi import Response
# 定义指标
REQUEST_COUNT = Counter(
'video_api_requests_total',
'Total number of API requests',
['method', 'endpoint', 'status_code']
)
REQUEST_LATENCY = Histogram(
'video_api_request_latency_seconds',
'API request latency',
['method', 'endpoint']
)
GPU_MEMORY_USAGE = Gauge(
'video_api_gpu_memory_usage_bytes',
'GPU memory usage',
['gpu_index']
)
# 监控中间件
@app.middleware("http")
async def monitor_requests(request: Request, call_next):
start_time = time.time()
method = request.method
endpoint = request.url.path
try:
response = await call_next(request)
REQUEST_COUNT.labels(method, endpoint, response.status_code).inc()
REQUEST_LATENCY.labels(method, endpoint).observe(time.time() - start_time)
return response
except Exception as e:
REQUEST_COUNT.labels(method, endpoint, 500).inc()
raise e
# 指标端点
@app.get("/metrics")
async def metrics():
# 更新GPU指标
status = get_gpu_status()
for gpu in status:
GPU_MEMORY_USAGE.labels(gpu["index"]).set(gpu["used_mem"])
return Response(
content=generate_latest(),
media_type="text/plain"
)
6. 安全与权限控制
6.1 API密钥管理
生产环境中必须实现严格的API访问控制:
- 密钥生成:使用安全的随机算法生成API密钥
- 密钥存储:加密存储密钥哈希值
- 访问限制:基于密钥实施速率限制
实现示例:
python复制import secrets
import hashlib
from fastapi.security import APIKeyHeader
from fastapi import Depends, HTTPException
api_key_header = APIKeyHeader(name="X-API-Key")
# 模拟数据库存储
API_KEYS_DB = {}
def generate_api_key() -> str:
key = secrets.token_urlsafe(32)
hashed_key = hashlib.sha256(key.encode()).hexdigest()
API_KEYS_DB[hashed_key] = {
"created_at": datetime.utcnow(),
"rate_limit": 10 # 每分钟10次请求
}
return key
def validate_api_key(api_key: str) -> bool:
hashed_key = hashlib.sha256(api_key.encode()).hexdigest()
return hashed_key in API_KEYS_DB
# 速率限制中间件
from fastapi import Request
from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
@app.post("/api/generate-video")
@limiter.limit("10/minute")
async def generate_video(
request: Request,
user_prompt: str,
x_api_key: str = Depends(api_key_header)
):
if not validate_api_key(x_api_key):
raise HTTPException(status_code=403, detail="Invalid API Key")
# ...其余代码...
6.2 视频内容审核
为防止生成不当内容,应添加内容审核机制:
- 文本审核:在LangChain处理前审核用户输入
- 图像审核:对生成的视频关键帧进行审核
- 日志记录:完整记录生成请求和结果
实现示例:
python复制from transformers import pipeline
# 初始化审核模型
text_classifier = pipeline(
"text-classification",
model="Hate-speech-CNERG/dehatebert-mono-english"
)
image_classifier = pipeline(
"image-classification",
model="google/vit-base-patch16-224"
)
def moderate_text(text: str) -> bool:
result = text_classifier(text)[0]
return result["label"] == "NON_HATE" and result["score"] > 0.9
def moderate_video(video_path: str) -> bool:
# 提取关键帧
frames = extract_key_frames(video_path, num_frames=3)
for frame in frames:
result = image_classifier(frame)[0]
if result["label"] == "NSFW" and result["score"] > 0.7:
return False
return True
@app.post("/api/generate-video")
async def generate_video(user_prompt: str):
# 文本审核
if not moderate_text(user_prompt):
raise HTTPException(
status_code=400,
detail="输入内容不符合安全规范"
)
# ...生成视频...
# 视频内容审核
if not moderate_video(video_path):
os.remove(video_path) # 删除违规视频
raise HTTPException(
status_code=400,
detail="生成内容不符合安全规范"
)
# ...其余代码...
7. 生产环境部署建议
7.1 高可用架构
对于生产环境,建议采用以下架构确保高可用性:
- 负载均衡:使用Nginx或Traefik作为入口负载均衡器
- 服务冗余:关键组件(ComfyUI、Dify)部署多个实例
- 数据库集群:使用PostgreSQL集群替代SQLite
- 分布式存储:使用S3或MinIO存储生成的视频文件
优化后的docker-compose.yml示例:
yaml复制version: '3.8'
services:
# 负载均衡器
traefik:
image: traefik:v2.10
ports:
- "80:80"
- "443:443"
volumes:
- /var/run/docker.sock:/var/run/docker.sock
- ./traefik.yml:/etc/traefik/traefik.yml
networks:
- video-network
# ComfyUI集群
comfyui1:
image: comfyui/comfyui:latest
deploy:
replicas: 2
environment:
- NVIDIA_VISIBLE_DEVICES=all
volumes:
- ./comfyui:/app
- ./generated_videos:/app/output
networks:
- video-network
labels:
- "traefik.enable=true"
- "traefik.http.services.comfyui.loadbalancer.server.port=8188"
- "traefik.http.routers.comfyui.rule=PathPrefix(`/comfyui`)"
# Dify集群
dify-api:
image: langgenius/dify-api:latest
deploy:
replicas: 2
environment:
- DATABASE_URL=postgresql://postgres:password@postgres:5432/dify
- REDIS_URL=redis://redis:6379/0
networks:
- video-network
depends_on:
- postgres
- redis
# PostgreSQL集群
postgres:
image: postgres:15
environment:
POSTGRES_PASSWORD: password
POSTGRES_DB: dify
volumes:
- postgres_data:/var/lib/postgresql/data
networks:
- video-network
# Redis集群
redis:
image: redis:7
command: redis-server --appendonly yes
volumes:
- redis_data:/data
networks:
- video-network
# 对象存储
minio:
image: minio/minio
volumes:
- minio_data:/data
environment:
MINIO_ROOT_USER: minioadmin
MINIO_ROOT_PASSWORD: minioadmin
command: server /data --console-address ":9001"
networks:
- video-network
volumes:
postgres_data:
redis_data:
minio_data:
networks:
video-network:
driver: bridge
7.2 备份与恢复策略
为确保数据安全,应实施以下备份策略:
- 数据库备份:每日全量备份+WAL日志
- 配置文件备份:版本控制所有配置文件
- 视频文件备份:定期归档到冷存储
- 灾难恢复计划:文档化恢复流程
备份脚本示例:
bash复制#!/bin/bash
# 数据库备份
docker exec postgres pg_dump -U postgres -d dify > dify_backup_$(date +%Y%m%d).sql
# 配置文件备份
tar czf config_backup_$(date +%Y%m%d).tar.gz \
./comfyui/config.json \
./dify/config.yaml \
./docker-compose.yml \
./nginx.conf
# 上传到S3
aws s3 cp dify_backup_$(date +%Y%m%d).sql s3://backup-bucket/
aws s3 cp config_backup_$(date +%Y%m%d).tar.gz s3://backup-bucket/
# 清理旧备份
find . -name "*.sql" -mtime +7 -exec rm {} \;
find . -name "*.tar.gz" -mtime +7 -exec rm {} \;
8. 常见问题排查
8.1 ComfyUI相关问题
问题1:视频生成失败,日志显示"CUDA out of memory"
解决方案:
- 检查工作流复杂度,简化节点数量
- 降低生成分辨率或视频时长
- 在docker-compose中限制容器内存使用
- 添加
--lowvram参数启动ComfyUI
问题2:生成的视频出现闪烁或伪影
解决方案:
- 调整视频生成模型的噪声参数
- 增加关键帧间隔
- 添加后处理节点进行稳定化处理
- 尝试不同的视频编码参数
8.2 Dify相关问题
问题1:LangChain调用返回意外结果
解决方案:
- 检查Prompt模板中的变量替换
- 验证模型温度(temperature)参数设置
- 添加输出格式验证逻辑
- 在Dify中测试Prompt效果
问题2:API响应缓慢
解决方案:
- 检查模型加载是否使用了GPU加速
- 增加Dify实例数量
- 优化Prompt复杂度
- 添加缓存层缓存常见请求结果
8.3 部署相关问题
问题1:Docker容器启动失败
排查步骤:
- 查看容器日志:
docker logs <container_id> - 检查端口冲突:
netstat -tulnp | grep <port> - 验证卷挂载权限:
docker inspect <container_id> | grep Mounts - 检查GPU驱动兼容性:
nvidia-smi和docker run --gpus all nvidia/cuda:11.6.2-base-ubuntu20.04 nvidia-smi
问题2:Nginx返回502错误
排查步骤:
- 检查后端服务是否正常运行
- 验证Nginx代理配置中的upstream地址
- 检查防火墙设置
- 查看Nginx错误日志:
docker exec nginx cat /var/log/nginx/error.log
9. 性能调优指南
9.1 ComfyUI性能优化
-
工作流优化:
- 合并相似功能的节点
- 移除不必要的中间节点
- 使用缓存节点存储中间结果
-
模型优化:
- 使用量化模型减少显存占用
- 启用xFormers加速注意力计算
- 使用TensorRT优化模型推理
-
系统级优化:
- 启用CUDA Graph减少内核启动开销
- 调整Docker容器资源限制
- 使用--no-half-vae参数避免精度问题
9.2 FastAPI性能优化
-
异步处理:
- 使用async/await避免阻塞调用
- 将CPU密集型任务卸载到工作进程
- 使用httpx替代requests进行异步HTTP调用
-
数据库优化:
- 添加连接池
- 优化查询语句
- 添加适当的索引
-
缓存策略:
- 对频繁访问的数据添加Redis缓存
- 实现ETag缓存验证
- 对静态资源启用CDN加速
优化后的异步HTTP调用示例:
python复制import httpx
async def generate_video_with_comfyui(params: dict, task_id: str) -> str:
async with httpx.AsyncClient(timeout=60.0) as client:
try:
response = await client.post(
COMFYUI_API_URL,
json=comfyui_workflow,
headers={"Content-Type": "application/json"}
)
response.raise_for_status()
return f"{VIDEO_SAVE_DIR}/{task_id}.mp4"
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error: {e.response.status_code}")
raise
except httpx.RequestError as e:
logger.error(f"Request failed: {str(e)}")
raise
10. 扩展与定制
10.1 自定义视频风格
要实现自定义视频风格,可以考虑以下方法:
- LoRA适配器:训练特定风格的LoRA模型
- 文本反转:创建自定义的文本嵌入
- ControlNet:使用参考图像控制风格
- IPAdapter:通过图像提示调整风格
实现示例:
python复制def apply_custom_style(workflow: dict, style_name: str) -> dict:
# 加载预定义的风格配置
with open(f"styles/{style_name}.json", "r") as f:
style_config = json.load(f)
# 应用风格到工作流
for node_id, node_params in style_config.items():
if node_id in workflow["prompt"]:
workflow["prompt"][node_id]["inputs"].update(node_params)
return workflow
# 在生成视频时调用
workflow = load_workflow("base_workflow.json")
workflow = apply_custom_style(workflow, params["风格"])
10.2 多模态输入支持
扩展系统以支持更多输入类型:
- 图像输入:允许上传参考图像
- 音频输入:根据音频生成匹配的视频
- 视频输入:视频风格迁移或编辑
图像输入处理示例:
python复制from PIL import Image
import io
@app.post("/api/generate-video-with-image")
async def generate_video_with_image(
prompt: str,
image: UploadFile = File(...)
):
# 读取上传的图像
image_data = await image.read()
img = Image.open(io.BytesIO(image_data))
# 调整大小并保存
img = img.resize((512, 512))
img_path = f"./temp/{uuid.uuid4()}.png"
img.save(img_path)
# 更新工作流添加图像输入节点
workflow = load_workflow("image_to_video.json")
workflow["prompt"]["5"]["inputs"]["image"] = img_path
# ...其余生成逻辑...
10.3 商业API集成
对于需要更高稳定性的场景,可以考虑集成商业API:
- OpenAI视频API:当开源模型效果不佳时作为备选
- AWS/Azure AI服务:利用云服务商的托管AI服务
- 专业视频API:如RunwayML等专业视频生成服务
集成示例:
python复制class VideoGenerator:
def __init__(self, config):
self.config = config
async def generate(self, params: dict) -> str:
if self.config["use_commercial"]:
return await self._generate_commercial(params)
else:
return await self._generate_local(params)
async def _generate_local(self, params: dict) -> str:
# 使用本地ComfyUI生成
pass
async def _generate_commercial(self, params: dict) -> str:
# 使用商业API生成
if self.config["provider"] == "openai":
client = AsyncOpenAI(api_key=self.config["api_key"])
response = await client.video.generate(
prompt=params["prompt"],
model="video-model-alpha",
size="1024x576",
duration=params["duration"]
)
return response.data[0].url
elif self.config["provider"] == "runway":
# RunwayML集成
pass