| """ |
| 多维度加权决策模型路由器与质量降级漏斗 |
| Author: XiaoZhe (Commercial Contact: janejulius119@gmail.com / WeChat: julius119) |
| """ |
|
|
| import logging |
| from dataclasses import dataclass |
| from typing import Dict, List, Tuple, Optional |
|
|
| logger = logging.getLogger("VerseFlow.Router") |
|
|
| @dataclass |
| class ModelProfile: |
| model_id: str |
| display_name: str |
| param_count: float |
| max_resolution: Tuple[int, int] |
| max_duration: float |
| fp8_vram_gb: float |
| has_native_audio: bool |
| identity_consistency: float |
| motion_quality: float |
| face_quality: float |
| text_compliance: float |
| audio_sync: float |
| speed_score: float |
|
|
| @dataclass |
| class RoutingRequest: |
| scene_type: str |
| duration: float |
| resolution: Tuple[int, int] |
| requires_audio: bool |
| requires_text: bool |
| identity_critical: bool |
| deadline_s: float |
|
|
| class ModelRouter: |
| """6 维度加权决策模型路由器""" |
| SCENE_WEIGHTS = { |
| "product_demo": {"identity_consistency": 0.25, "text_compliance": 0.25, "motion_quality": 0.20, "speed_score": 0.30}, |
| "talking_head": {"face_quality": 0.40, "identity_consistency": 0.30, "audio_sync": 0.20, "speed_score": 0.10}, |
| "b_roll": {"speed_score": 0.40, "motion_quality": 0.30, "identity_consistency": 0.30}, |
| "brand_ad": {"text_compliance": 0.30, "identity_consistency": 0.30, "audio_sync": 0.20, "motion_quality": 0.20} |
| } |
|
|
| def __init__(self, models: Optional[List[ModelProfile]] = None): |
| self.models = models or self._get_default_profiles() |
|
|
| def _get_default_profiles(self) -> List[ModelProfile]: |
| return [ |
| ModelProfile("minimax_h3", "MiniMax H3", 33.0, (2560, 1440), 15.0, 38.0, True, 9.0, 8.5, 9.0, 9.5, 9.5, 3.0),[cite: 1] |
| ModelProfile("wan2.2_14b", "Wan2.2 14B", 14.0, (1280, 720), 10.0, 28.0, False, 8.5, 9.0, 8.5, 7.5, 0.0, 4.0),[cite: 1] |
| ModelProfile("hunyuan_video_1_5", "Hunyuan 1.5", 8.3, (1280, 720), 12.0, 24.0, False, 8.0, 8.0, 9.5, 7.0, 0.0, 5.0),[cite: 1] |
| ModelProfile("ltx_2_3", "LTX 2.3", 22.0, (1920, 1080), 10.0, 16.0, True, 7.0, 8.0, 7.5, 6.0, 8.0, 9.0),[cite: 1] |
| ModelProfile("wan2.2_5b", "Wan2.2 5B", 5.0, (854, 480), 5.0, 8.0, False, 6.0, 7.0, 6.5, 6.0, 0.0, 9.5),[cite: 1] |
| ] |
|
|
| def route(self, request: RoutingRequest) -> List[Tuple[str, float, str]]: |
| candidates = [] |
| weights = self.SCENE_WEIGHTS.get(request.scene_type, self.SCENE_WEIGHTS["b_roll"]) |
|
|
| for model in self.models: |
| if request.duration > model.max_duration: |
| continue |
| if request.resolution[0] > model.max_resolution[0] or request.resolution[1] > model.max_resolution[1]: |
| continue |
|
|
| score = 0.0 |
| for attr, weight in weights.items(): |
| val = getattr(model, attr, 5.0) |
| score += val * weight |
|
|
| if request.requires_audio and not model.has_native_audio: |
| score *= 0.6 |
|
|
| reason = f"场景匹配得分: {score:.2f} | Model={model.display_name}" |
| candidates.append((model.model_id, score, reason)) |
|
|
| candidates.sort(key=lambda x: x[1], reverse=True) |
| return candidates |
|
|
| class QualityFunnelController: |
| """三级质量控制漏斗""" |
| @staticmethod |
| def get_stage_config(quality_tier: str) -> Dict[str, Any]: |
| if quality_tier == "draft": |
| return {"model_id": "wan2.2_5b", "resolution": (854, 480), "max_attempts": 1}[cite: 1] |
| elif quality_tier == "refine": |
| return {"model_id": "auto_route", "resolution": (1280, 720), "max_attempts": 3}[cite: 1] |
| else: |
| return {"model_id": "minimax_h3", "resolution": (1920, 1080), "max_attempts": 5}[cite: 1] |
