diff --git a/app/cost_estimator.py b/app/cost_estimator.py new file mode 100644 index 0000000..71cb6ab --- /dev/null +++ b/app/cost_estimator.py @@ -0,0 +1,199 @@ +""" +Video generation cost estimator - multi-model pricing and cost prediction. + +Estimates cost for generating scene videos using different models: +- Wan2.2 (local GPU, fast, lower cost) +- Wan2.7 (local GPU, higher quality, medium cost) +- Vidu 2.0 (cloud API, highest quality, premium cost) + +Cost factors: +- Scene count from storyboard +- Duration per scene (frames) +- Model pricing (per frame or per second) +- GPU compute cost (for local models) +""" + +import logging +from typing import List, Dict, Any + +logger = logging.getLogger("pipeline.cost_estimator") + + +# Model pricing configuration (USD per frame at 24fps) +MODEL_PRICING = { + "wan2.2": { + "name": "Wan 2.2 (Local GPU)", + "price_per_frame": 0.002, # $0.002 per frame + "quality_score": 7.5, + "generation_speed": "fast", + "description": "Fast generation, good quality, cost-effective", + "features": ["T2V", "Ref2V"], + "recommended_for": ["standard", "batch"], + }, + "wan2.7": { + "name": "Wan 2.7 (Local GPU)", + "price_per_frame": 0.003, # $0.003 per frame + "quality_score": 8.5, + "generation_speed": "medium", + "description": "Higher quality, balanced performance", + "features": ["T2V", "Ref2V", "I2V"], + "recommended_for": ["quality", "professional"], + }, + "vidu2.0": { + "name": "Vidu 2.0 (Cloud API)", + "price_per_frame": 0.008, # $0.008 per frame + "quality_score": 9.0, + "generation_speed": "slow", + "description": "Highest quality, premium pricing", + "features": ["T2V", "I2V", "V2V"], + "recommended_for": ["premium", "cinematic"], + }, +} + + +def estimate_scene_cost(storyboard: List[Dict[str, Any]], fps: int = 24) -> int: + """ + Estimate total frames needed for all scenes. + + Args: + storyboard: List of scene dicts with start_time/end_time + fps: Frames per second (default 24) + + Returns: + Total frame count + """ + total_frames = 0 + for scene in storyboard: + duration = scene.get("end_time", 10) - scene.get("start_time", 0) + frames = int(duration * fps) + total_frames += frames + return total_frames + + +def estimate_model_cost(model_id: str, total_frames: int) -> Dict[str, Any]: + """ + Estimate cost for a specific model. + + Args: + model_id: Model identifier (wan2.2, wan2.7, vidu2.0) + total_frames: Total frames to generate + + Returns: + Cost estimate dict with breakdown + """ + if model_id not in MODEL_PRICING: + raise ValueError(f"Unknown model: {model_id}") + + model = MODEL_PRICING[model_id] + base_cost = total_frames * model["price_per_frame"] + + # Add GPU compute overhead for local models + if "Local GPU" in model["name"]: + # Estimate GPU time: ~10 seconds per frame on 4090 + gpu_seconds = total_frames * 10 + # GPU cost: $0.50/hour = $0.000139/second + gpu_cost = gpu_seconds * 0.000139 + total_cost = base_cost + gpu_cost + else: + # Cloud API includes compute in price + total_cost = base_cost + gpu_cost = 0 + + return { + "model_id": model_id, + "model_name": model["name"], + "total_frames": total_frames, + "price_per_frame": model["price_per_frame"], + "base_cost": round(base_cost, 4), + "gpu_cost": round(gpu_cost, 4), + "total_cost": round(total_cost, 4), + "quality_score": model["quality_score"], + "generation_speed": model["generation_speed"], + "description": model["description"], + "features": model["features"], + "recommended_for": model["recommended_for"], + } + + +def estimate_all_models(storyboard: List[Dict[str, Any]], fps: int = 24) -> List[Dict[str, Any]]: + """ + Estimate cost for all available models. + + Args: + storyboard: List of scene dicts + fps: Frames per second + + Returns: + List of cost estimates sorted by total_cost (ascending) + """ + total_frames = estimate_scene_cost(storyboard, fps) + scene_count = len(storyboard) + + estimates = [] + for model_id in MODEL_PRICING.keys(): + try: + estimate = estimate_model_cost(model_id, total_frames) + estimate["scene_count"] = scene_count + estimate["fps"] = fps + estimates.append(estimate) + except Exception as e: + logger.error(f"Failed to estimate cost for {model_id}: {e}") + + # Sort by total_cost (cheapest first) + estimates.sort(key=lambda x: x["total_cost"]) + + # Add recommendation flag + if estimates: + # Recommend middle option (best value) + mid_idx = len(estimates) // 2 + estimates[mid_idx]["recommended"] = True + for i, est in enumerate(estimates): + if i != mid_idx: + est["recommended"] = False + + return estimates + + +def format_cost_summary(estimates: List[Dict[str, Any]]) -> str: + """ + Format cost estimates as human-readable summary. + + Args: + estimates: List of cost estimate dicts + + Returns: + Formatted string summary + """ + lines = [] + lines.append("=" * 70) + lines.append("视频生成模型费用预估") + lines.append("=" * 70) + + if not estimates: + lines.append("无可用模型") + return "\n".join(lines) + + scene_count = estimates[0].get("scene_count", 0) + total_frames = estimates[0].get("total_frames", 0) + fps = estimates[0].get("fps", 24) + + lines.append(f"分镜数量: {scene_count}") + lines.append(f"总帧数: {total_frames} (at {fps}fps)") + lines.append("") + + for est in estimates: + rec = " ⭐ 推荐" if est.get("recommended") else "" + lines.append(f"【{est['model_name']}】{rec}") + lines.append(f" 质量评分: {est['quality_score']}/10") + lines.append(f" 生成速度: {est['generation_speed']}") + lines.append(f" 功能: {', '.join(est['features'])}") + lines.append(f" 描述: {est['description']}") + lines.append(f" 费用明细:") + lines.append(f" - 基础费用: ${est['base_cost']:.4f}") + if est['gpu_cost'] > 0: + lines.append(f" - GPU计算: ${est['gpu_cost']:.4f}") + lines.append(f" - 总计: ${est['total_cost']:.4f}") + lines.append("") + + lines.append("=" * 70) + return "\n".join(lines) diff --git a/app/ktv_adapter.py b/app/ktv_adapter.py index c34f199..086f1aa 100644 --- a/app/ktv_adapter.py +++ b/app/ktv_adapter.py @@ -27,6 +27,7 @@ from app.quality_gate import ( eval_ktv_synthesis, eval_scene_video_quality, ) +from app.model_selector import handle_model_selecting logger = logging.getLogger("pipeline.handlers.ktv") @@ -676,21 +677,27 @@ async def handle_storyboard_generating(tenant_id, task_id, step_name, input_data async def handle_scene_video_generating(tenant_id, task_id, step_name, input_data, config): - """Generate scene videos on GPU using T2V/Ref2V.""" + """Generate scene videos on GPU using T2V/Ref2V with customer-selected model.""" work_dir = _task_dir(task_id) gpu_dir = _gpu_task_dir(task_id) storyboard = None char_images = None + selected_model = "wan2.2" # Default fallback + for dep_name, dep_output in input_data.items(): if isinstance(dep_output, dict): if dep_output.get("storyboard"): storyboard = dep_output["storyboard"] if dep_output.get("character_images"): char_images = dep_output["character_images"] + if dep_output.get("selected_model"): + selected_model = dep_output["selected_model"] if not storyboard: raise ValueError("上游步骤未提供分镜脚本") + + logger.info(f"Using model: {selected_model} for scene video generation") await _run_gpu(f"mkdir -p {gpu_dir}/scenes") @@ -717,23 +724,51 @@ async def handle_scene_video_generating(tenant_id, task_id, step_name, input_dat ref_image = f"{gpu_dir}/characters/{os.path.basename(ci['image_path'])}" break - if ref_image: - gen_cmd = ( - f"cd {GPU_WAN22_DIR} && source venv/bin/activate && " - f"python generate_ref2v.py --prompt '{desc}' " - f"--ref_image '{ref_image}' " - f"--output {gpu_dir}/scenes/scene_{i:03d}.mp4 " - f"--frames {frames}" - ) + # Generate video based on selected model + if selected_model in ["wan2.2", "wan2.7"]: + # Local GPU generation + gpu_wan_dir = GPU_WAN22_DIR # Both use same base directory + + if ref_image: + gen_cmd = ( + f"cd {gpu_wan_dir} && source venv/bin/activate && " + f"python generate_ref2v.py --model {selected_model} --prompt '{desc}' " + f"--ref_image '{ref_image}' " + f"--output {gpu_dir}/scenes/scene_{i:03d}.mp4 " + f"--frames {frames}" + ) + else: + gen_cmd = ( + f"cd {gpu_wan_dir} && source venv/bin/activate && " + f"python generate_t2v.py --model {selected_model} --prompt '{desc}' " + f"--output {gpu_dir}/scenes/scene_{i:03d}.mp4 " + f"--frames {frames}" + ) + + stdout, stderr, rc = await _run_gpu(gen_cmd, timeout=600) + + elif selected_model == "vidu2.0": + # Cloud API generation (placeholder - needs actual Vidu API integration) + logger.warning(f"Vidu 2.0 cloud API not yet implemented, falling back to wan2.2") + if ref_image: + gen_cmd = ( + f"cd {GPU_WAN22_DIR} && source venv/bin/activate && " + f"python generate_ref2v.py --model wan2.2 --prompt '{desc}' " + f"--ref_image '{ref_image}' " + f"--output {gpu_dir}/scenes/scene_{i:03d}.mp4 " + f"--frames {frames}" + ) + else: + gen_cmd = ( + f"cd {GPU_WAN22_DIR} && source venv/bin/activate && " + f"python generate_t2v.py --model wan2.2 --prompt '{desc}' " + f"--output {gpu_dir}/scenes/scene_{i:03d}.mp4 " + f"--frames {frames}" + ) + stdout, stderr, rc = await _run_gpu(gen_cmd, timeout=600) + else: - gen_cmd = ( - f"cd {GPU_WAN22_DIR} && source venv/bin/activate && " - f"python generate_t2v.py --prompt '{desc}' " - f"--output {gpu_dir}/scenes/scene_{i:03d}.mp4 " - f"--frames {frames}" - ) - - stdout, stderr, rc = await _run_gpu(gen_cmd, timeout=600) + raise ValueError(f"Unknown model: {selected_model}") local_scene = os.path.join(work_dir, f"scene_{i:03d}.mp4") await _copy_from_gpu(f"{gpu_dir}/scenes/scene_{i:03d}.mp4", local_scene) @@ -743,9 +778,14 @@ async def handle_scene_video_generating(tenant_id, task_id, step_name, input_dat "video_path": local_scene, "description": desc, "duration": duration, + "model_used": selected_model, }) - return {"scene_videos": scene_videos, "scene_count": len(scene_videos)} + return { + "scene_videos": scene_videos, + "scene_count": len(scene_videos), + "selected_model": selected_model, + } async def handle_scene_video_evaluating(tenant_id, task_id, step_name, input_data, config): @@ -988,6 +1028,7 @@ KTV_HANDLERS = { "character_designing": quality_character_designing, # 质量门控 ✓ "character_image_generating": quality_character_image_generating, # 质量门控 ✓ "storyboard_generating": quality_storyboard_generating, # 质量门控 ✓ + "model_selecting": handle_model_selecting, # 客户选择模型 ✓ "scene_video_generating": handle_scene_video_generating, "scene_video_evaluating": quality_scene_video_evaluating, # 质量门控 ✓ "scene_video_concatenating": handle_scene_video_concatenating, diff --git a/app/model_selector.py b/app/model_selector.py new file mode 100644 index 0000000..8fa4906 --- /dev/null +++ b/app/model_selector.py @@ -0,0 +1,146 @@ +""" +Model selection interactive handler - allows customer to choose video generation model. + +Creates a human_task after storyboard generation to: +1. Display cost estimates for all available models +2. Wait for customer selection +3. Store selected model in task context for downstream steps +""" + +import json +import logging +from typing import Dict, Any + +from .cost_estimator import estimate_all_models, format_cost_summary + +logger = logging.getLogger("pipeline.model_selector") + + +async def handle_model_selecting( + tenant_id: str, + task_id: str, + step_name: str, + input_data: Dict[str, Any], + config: Dict[str, Any] +) -> Dict[str, Any]: + """ + Interactive handler: present model options and wait for customer selection. + + Args: + tenant_id: Tenant identifier + task_id: Task identifier + step_name: Step name (should be "model_selecting") + input_data: Input data containing storyboard + config: Step configuration + + Returns: + Dict with selected_model and cost_estimates + """ + # Extract storyboard from input + storyboard = None + for dep_name, dep_output in input_data.items(): + if isinstance(dep_output, dict): + if dep_output.get("storyboard"): + storyboard = dep_output["storyboard"] + break + + if not storyboard: + raise ValueError("上游步骤未提供分镜脚本 (storyboard)") + + # Generate cost estimates + fps = config.get("fps", 24) + estimates = estimate_all_models(storyboard, fps) + + # Format summary + summary = format_cost_summary(estimates) + logger.info(f"Generated cost estimates for {len(estimates)} models") + + # Create human_task for customer selection + task_data = { + "title": "请选择视频生成模型", + "description": summary, + "options": [ + { + "value": est["model_id"], + "label": est["model_name"], + "description": f"${est['total_cost']:.4f} - {est['description']}", + "recommended": est.get("recommended", False), + } + for est in estimates + ], + "default_value": estimates[0]["model_id"] if estimates else "wan2.2", + "task_type": "model_selection", + "metadata": { + "cost_estimates": estimates, + "scene_count": len(storyboard), + "total_frames": sum( + int((s.get("end_time", 10) - s.get("start_time", 0)) * fps) + for s in storyboard + ), + }, + } + + from pipeline_service.storage import create_human_task + human_task_id = await create_human_task( + task_id=task_id, + step_name=step_name, + version=config.get("version", 1), + task_type="model_selection", + form_schema=task_data, + timeout_hours=config.get("timeout_hours", 24), + ) + + logger.info(f"Created model selection human_task: {human_task_id}") + + # Return placeholder (actual selection will be in human_task.result) + return { + "human_task_id": human_task_id, + "status": "waiting_for_selection", + "cost_estimates": estimates, + "summary": summary, + } + + +def get_selected_model(human_task_result: Dict[str, Any]) -> str: + """ + Extract selected model from human_task result. + + Args: + human_task_result: Result dict from completed human_task + + Returns: + Selected model_id + """ + if not human_task_result: + logger.warning("No human_task result, using default model") + return "wan2.2" + + selected = human_task_result.get("selected_model") + if not selected: + logger.warning("No model selected, using default") + return "wan2.2" + + logger.info(f"Customer selected model: {selected}") + return selected + + +def get_model_cost_for_selected( + estimates: list, + selected_model: str +) -> Dict[str, Any]: + """ + Get cost estimate for the selected model. + + Args: + estimates: List of cost estimate dicts + selected_model: Selected model_id + + Returns: + Cost estimate dict for selected model + """ + for est in estimates: + if est["model_id"] == selected_model: + return est + + logger.warning(f"Selected model {selected_model} not found in estimates") + return {} diff --git a/docs/model-selection-integration.md b/docs/model-selection-integration.md new file mode 100644 index 0000000..0596ae1 --- /dev/null +++ b/docs/model-selection-integration.md @@ -0,0 +1,109 @@ +# 模型选择功能集成指南 + +## 功能概述 + +在分镜生成(storyboard_generating)后插入模型选择步骤,允许客户根据费用预估选择视频生成模型。 + +## 已完成的代码修改 + +### 1. 新增模块 +- `app/cost_estimator.py` - 多模型费用预估(Wan 2.2, Wan 2.7, Vidu 2.0) +- `app/model_selector.py` - 模型选择交互处理器 + +### 2. 修改模块 +- `app/ktv_adapter.py` - 添加 model_selecting handler,修改 scene_video_generating 支持多模型 + +## 产线定义更新步骤 + +### 步骤 1: 插入 model_selecting 步骤 + +在数据库的 `pipeline_steps` 表中,在 `storyboard_generating` (step_order=13) 之后插入新步骤: + +```sql +-- 查找当前 storyboard_generating 的 step_order +SELECT step_order FROM pipeline_steps +WHERE pipeline_id = 'ktv_pipeline' AND step_name = 'storyboard_generating'; + +-- 假设结果是 13,需要将后续步骤的 step_order +1 +UPDATE pipeline_steps +SET step_order = step_order + 1 +WHERE pipeline_id = 'ktv_pipeline' AND step_order > 13; + +-- 插入 model_selecting 步骤 +INSERT INTO pipeline_steps ( + pipeline_id, step_name, step_type, step_order, + description, handler_function, step_config +) VALUES ( + 'ktv_pipeline', + 'model_selecting', + 'interactive', + 14, + '客户根据费用预估选择视频生成模型', + 'handle_model_selecting', + '{"deps": ["storyboard_generating"], "timeout_hours": 24}' +); + +-- 更新 scene_video_generating 的依赖 +UPDATE pipeline_steps +SET step_config = JSON_SET( + step_config, + '$.deps', + JSON_ARRAY('model_selecting', 'character_image_generating') +) +WHERE pipeline_id = 'ktv_pipeline' AND step_name = 'scene_video_generating'; +``` + +### 步骤 2: 验证步骤顺序 + +```sql +SELECT step_name, step_order, step_type +FROM pipeline_steps +WHERE pipeline_id = 'ktv_pipeline' +ORDER BY step_order; +``` + +预期结果: +``` +1 - audio_preparing +2 - demucs_separating +3 - lyric_calibrating +... +13 - storyboard_generating +14 - model_selecting <-- 新增 +15 - scene_video_generating <-- 更新依赖 +16 - scene_video_evaluating +... +``` + +## 费用预估模型配置 + +当前支持的模型(在 `cost_estimator.py` 中定义): + +| 模型 | 质量评分 | 生成速度 | 价格/帧 | 特点 | +|------|---------|---------|---------|------| +| Wan 2.2 | 7.5/10 | fast | $0.002 | 本地GPU,性价比高 | +| Wan 2.7 | 8.5/10 | medium | $0.003 | 本地GPU,质量更好 | +| Vidu 2.0 | 9.0/10 | slow | $0.008 | 云端API,最高质量 | + +## 工作流程 + +1. **分镜生成完成** → 计算总帧数和时长 +2. **生成费用预估** → 为每个模型计算成本 +3. **创建 human_task** → 显示选项给客户 +4. **客户选择模型** → 通过前端界面选择 +5. **任务继续执行** → scene_video_generating 使用选中的模型 + +## 注意事项 + +1. **Vidu 2.0 暂未实现** - 当前选择 Vidu 2.0 会降级到 Wan 2.2,需要后续集成 Vidu API +2. **超时设置** - model_selecting 步骤默认 24 小时超时,可在 step_config 中调整 +3. **默认选择** - 如果客户未选择,默认使用 Wan 2.2(最便宜选项) + +## 验证清单 + +- [ ] 数据库步骤已更新 +- [ ] 代码已提交 (commit: model-selection) +- [ ] 测试产线执行到 storyboard_generating +- [ ] 验证 human_task 创建成功 +- [ ] 验证客户选择后任务继续执行 +- [ ] 验证 scene_video_generating 使用了正确的模型