yumoqing
9d3f0ff38c
新增模块: - app/eval_video.py: 场景视频多维度评估 (视觉质量25%, 运动流畅度20%, 语义一致性30%, 角色一致性15%, 时间连贯性10%) - app/eval_music.py: 音乐多维度评估 (节奏15%, 和弦10%, 听感20%, 情绪-歌词匹配20%, 指令匹配15%, 伴奏10%, 人声10%) 更新模块: - app/quality_gate.py: 集成多维度评估函数，支持降级 - app/ktv_adapter.py: 注册6个质量门控 handlers (步骤9-10, 11, 12, 13, 14-15, 17) 质量门控流程: 1. 执行 handler 2. 多维度评估 3. 不达标自动重试(最多3次) 4. 仍不达标创建 human_task 暂停任务等待人工决策
450 lines
15 KiB
Python
450 lines
15 KiB
Python
"""质量门控：评估-重试-人工介入闭环。
|
||
|
||
用于 KTV 产线关键步骤的质量保障。
|
||
|
||
架构：
|
||
1. 执行 handler → 2. 评估结果 → 3. 不达标则重试(最多3次) → 4. 仍不达标则暂停等待人工决策
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
from typing import Callable, Any, Optional, Dict, Tuple
|
||
|
||
logger = logging.getLogger("pipeline.quality_gate")
|
||
|
||
# 最大自动重试次数
|
||
MAX_RETRIES = 3
|
||
|
||
|
||
async def with_quality_gate(
|
||
task_id: str,
|
||
step_name: str,
|
||
version: int,
|
||
handler: Callable,
|
||
evaluator: Callable,
|
||
tenant_id: str = "",
|
||
input_data: Optional[dict] = None,
|
||
config: Optional[dict] = None,
|
||
) -> dict:
|
||
"""带质量门控的 handler 执行器。
|
||
|
||
参数:
|
||
handler: 原始步骤处理函数
|
||
evaluator: 评估函数 async def evaluator(output_data) -> (passed: bool, score: float, reason: str)
|
||
|
||
流程:
|
||
1. 执行 handler
|
||
2. 调用 evaluator 评估结果
|
||
3. 不通过则重试（最多 MAX_RETRIES 次）
|
||
4. 仍不通过则创建 human_task，暂停任务
|
||
|
||
返回:
|
||
handler 的输出数据（含 _quality_meta 元信息）
|
||
"""
|
||
input_data = input_data or {}
|
||
config = config or {}
|
||
|
||
last_output: Optional[dict] = None
|
||
last_reason = ""
|
||
|
||
for attempt in range(1, MAX_RETRIES + 1):
|
||
logger.info(f"[QualityGate] {step_name} 第 {attempt}/{MAX_RETRIES} 次执行")
|
||
|
||
try:
|
||
# 执行 handler
|
||
output = await handler(tenant_id, task_id, step_name, input_data, config)
|
||
last_output = output
|
||
|
||
# 评估结果
|
||
passed, score, reason = await evaluator(output)
|
||
|
||
logger.info(f"[QualityGate] {step_name} 评估: passed={passed}, score={score:.2f}, reason={reason}")
|
||
|
||
if passed:
|
||
# 通过，附加元信息返回
|
||
output["_quality_meta"] = {
|
||
"attempts": attempt,
|
||
"score": score,
|
||
"reason": reason,
|
||
"passed": True,
|
||
}
|
||
return output
|
||
|
||
last_reason = reason
|
||
|
||
except Exception as e:
|
||
logger.warning(f"[QualityGate] {step_name} 第 {attempt} 次执行异常: {e}")
|
||
last_reason = f"执行异常: {e}"
|
||
|
||
# 超过重试次数，创建人工任务
|
||
logger.warning(
|
||
f"[QualityGate] {step_name} 经过 {MAX_RETRIES} 次尝试仍不达标，创建人工任务"
|
||
)
|
||
|
||
await _create_quality_review_task(
|
||
task_id=task_id,
|
||
step_name=step_name,
|
||
version=version,
|
||
attempts=MAX_RETRIES,
|
||
last_output=last_output,
|
||
last_reason=last_reason,
|
||
)
|
||
|
||
# 抛出异常让 executor 检测到 WAITING 状态并暂停任务
|
||
raise QualityGatePausedError(
|
||
f"步骤 {step_name} 质量评估未通过，已暂停等待人工审核。原因: {last_reason}"
|
||
)
|
||
|
||
|
||
async def _create_quality_review_task(
|
||
task_id: str,
|
||
step_name: str,
|
||
version: int,
|
||
attempts: int,
|
||
last_output: dict,
|
||
last_reason: str,
|
||
):
|
||
"""创建质量审核人工任务。"""
|
||
from pipeline_service.storage import create_human_task
|
||
from pipeline_service.state import STATE_WAITING
|
||
from pipeline_service.storage import update_step_state
|
||
|
||
# 构造审核表单
|
||
form_schema = {
|
||
"title": f"步骤 {step_name} 质量审核",
|
||
"description": f"经过 {attempts} 次自动重试仍未通过质量评估",
|
||
"fields": [
|
||
{
|
||
"name": "last_reason",
|
||
"label": "失败原因",
|
||
"type": "textarea",
|
||
"readonly": True,
|
||
"default": last_reason,
|
||
},
|
||
{
|
||
"name": "last_output_summary",
|
||
"label": "最后一次输出摘要",
|
||
"type": "textarea",
|
||
"readonly": True,
|
||
"default": json.dumps(last_output, ensure_ascii=False, indent=2)[:2000] if last_output else "无",
|
||
},
|
||
{
|
||
"name": "decision",
|
||
"label": "决策",
|
||
"type": "select",
|
||
"options": [
|
||
{"value": "retry_manual", "label": "手动重试（调整参数后重新执行）"},
|
||
{"value": "accept", "label": "接受当前结果（忽略质量要求）"},
|
||
{"value": "abort", "label": "终止任务"},
|
||
],
|
||
"required": True,
|
||
},
|
||
{
|
||
"name": "notes",
|
||
"label": "备注说明",
|
||
"type": "textarea",
|
||
"required": False,
|
||
},
|
||
],
|
||
}
|
||
|
||
# 创建人工任务记录
|
||
await create_human_task(
|
||
task_id=task_id,
|
||
step_name=step_name,
|
||
version=version,
|
||
task_type="quality_review",
|
||
assignee_role="admin",
|
||
form_schema=form_schema,
|
||
timeout_hours=24,
|
||
)
|
||
|
||
# 设置步骤状态为等待
|
||
await update_step_state(task_id, step_name, STATE_WAITING)
|
||
|
||
logger.info(f"[QualityGate] 已创建人工审核任务: {task_id}/{step_name}")
|
||
|
||
|
||
class QualityGatePausedError(Exception):
|
||
"""质量门控暂停异常，用于通知 executor 暂停任务。"""
|
||
pass
|
||
|
||
|
||
# ─── 评估函数 ──────────────────────────────────────────────────────────────
|
||
|
||
async def eval_music_quality(output: dict) -> Tuple[bool, float, str]:
|
||
"""评估音乐生成结果（步骤 9-10）— 多维度评估。
|
||
|
||
7 个评估维度：
|
||
节奏质量 (15%) | 和弦进行 (10%) | 听感质量 (20%)
|
||
情绪-歌词匹配 (20%) | 指令匹配度 (15%)
|
||
伴奏质量 (10%) | 人声质量 (10%)
|
||
"""
|
||
music_path = output.get("music_path", "")
|
||
|
||
if not music_path or not os.path.exists(music_path):
|
||
return False, 0.0, f"音乐文件不存在: {music_path}"
|
||
|
||
file_size = os.path.getsize(music_path)
|
||
if file_size < 50 * 1024:
|
||
return False, 0.3, f"音乐文件过小 ({file_size/1024:.1f}KB < 50KB)"
|
||
|
||
# 调用多维度评估
|
||
try:
|
||
from app.eval_music import evaluate_music_quality
|
||
prompt = output.get("prompt", output.get("music_prompt", ""))
|
||
lyrics = output.get("lyrics", "")
|
||
|
||
result = await evaluate_music_quality(
|
||
audio_path=music_path,
|
||
prompt=prompt,
|
||
lyrics=lyrics,
|
||
use_demucs=True,
|
||
)
|
||
|
||
overall_score = result["overall_score"]
|
||
overall_reason = result["overall_reason"]
|
||
passed = overall_score >= 0.6
|
||
|
||
# 将维度评分存入 output 供后续参考
|
||
output["_eval_dimensions"] = result.get("dimensions", {})
|
||
|
||
return passed, overall_score, overall_reason
|
||
|
||
except ImportError:
|
||
logger.warning("eval_music 模块未安装，降级为基础评估")
|
||
# 降级：仅检查时长
|
||
import asyncio
|
||
proc = await asyncio.create_subprocess_shell(
|
||
f"ffprobe -v error -show_entries format=duration -of default=noprint_wrappers=1:nokey=1 '{music_path}'",
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE,
|
||
)
|
||
stdout, _ = await proc.communicate()
|
||
duration = float(stdout.decode().strip()) if stdout else 0
|
||
|
||
if duration < 30:
|
||
return False, 0.5, f"音乐时长不足 ({duration:.1f}s < 30s)"
|
||
|
||
score = min(1.0, duration / 180)
|
||
return True, score, f"音乐文件正常 ({duration:.1f}s, {file_size/1024:.0f}KB)"
|
||
|
||
|
||
async def eval_character_design(output: dict) -> Tuple[bool, float, str]:
|
||
"""评估角色设计结果（步骤 11）。
|
||
|
||
评估标准：
|
||
1. characters 列表非空
|
||
2. 每个角色包含 name, prompt/description, personality
|
||
3. prompt 描述长度合理（> 50 字符）
|
||
"""
|
||
characters = output.get("characters", [])
|
||
|
||
if not characters or not isinstance(characters, list):
|
||
return False, 0.0, "角色列表为空"
|
||
|
||
issues = []
|
||
for i, char in enumerate(characters):
|
||
if not isinstance(char, dict):
|
||
issues.append(f"角色 {i} 格式错误")
|
||
continue
|
||
|
||
name = char.get("name", "")
|
||
prompt = char.get("prompt", char.get("description", ""))
|
||
personality = char.get("personality", "")
|
||
|
||
if not name:
|
||
issues.append(f"角色 {i} 缺少 name")
|
||
if not prompt or len(prompt) < 50:
|
||
issues.append(f"角色 {i} 的 prompt 描述过短 ({len(prompt)} chars)")
|
||
if not personality:
|
||
issues.append(f"角色 {i} 缺少 personality")
|
||
|
||
if issues:
|
||
return False, 0.3, "; ".join(issues[:3])
|
||
|
||
return True, 0.9, f"角色设计完整 ({len(characters)} 个角色)"
|
||
|
||
|
||
async def eval_character_image(output: dict) -> Tuple[bool, float, str]:
|
||
"""评估角色图生成结果（步骤 12）。
|
||
|
||
评估标准：
|
||
1. character_images 列表非空
|
||
2. 每个图片文件存在且大小 > 10KB
|
||
3. 图片数量匹配角色数量
|
||
"""
|
||
char_images = output.get("character_images", [])
|
||
|
||
if not char_images or not isinstance(char_images, list):
|
||
return False, 0.0, "角色图列表为空"
|
||
|
||
missing = []
|
||
too_small = []
|
||
|
||
for item in char_images:
|
||
path = item.get("image_path", "")
|
||
name = item.get("name", "unknown")
|
||
|
||
if not path or not os.path.exists(path):
|
||
missing.append(name)
|
||
continue
|
||
|
||
size = os.path.getsize(path)
|
||
if size < 10 * 1024:
|
||
too_small.append(f"{name}({size/1024:.1f}KB)")
|
||
|
||
if missing:
|
||
return False, 0.3, f"图片缺失: {', '.join(missing)}"
|
||
if too_small:
|
||
return False, 0.5, f"图片过小: {', '.join(too_small)}"
|
||
|
||
return True, 0.9, f"角色图生成正常 ({len(char_images)} 张)"
|
||
|
||
|
||
async def eval_storyboard(output: dict) -> Tuple[bool, float, str]:
|
||
"""评估分镜脚本结果（步骤 13）。
|
||
|
||
评估标准：
|
||
1. storyboard 列表非空
|
||
2. 每个分镜包含 scene_id, start_time, end_time, description
|
||
3. 时间轴覆盖完整（无大段空白）
|
||
4. description 描述充分
|
||
"""
|
||
storyboard = output.get("storyboard", [])
|
||
|
||
if not storyboard or not isinstance(storyboard, list):
|
||
return False, 0.0, "分镜列表为空"
|
||
|
||
if len(storyboard) < 3:
|
||
return False, 0.3, f"分镜数量过少 ({len(storyboard)} < 3)"
|
||
|
||
issues = []
|
||
prev_end = 0
|
||
|
||
for i, scene in enumerate(storyboard):
|
||
if not isinstance(scene, dict):
|
||
issues.append(f"分镜 {i} 格式错误")
|
||
continue
|
||
|
||
# 必填字段检查
|
||
for field in ["scene_id", "start_time", "end_time", "description"]:
|
||
if field not in scene:
|
||
issues.append(f"分镜 {i} 缺少 {field}")
|
||
|
||
# 时间轴连续性
|
||
start = scene.get("start_time", 0)
|
||
end = scene.get("end_time", 0)
|
||
if start > prev_end + 5: # 允许 5 秒间隔
|
||
issues.append(f"分镜 {i} 时间轴有断层 ({prev_end}s → {start}s)")
|
||
prev_end = end
|
||
|
||
# description 质量
|
||
desc = scene.get("description", "")
|
||
if len(desc) < 20:
|
||
issues.append(f"分镜 {i} 描述过短 ({len(desc)} chars)")
|
||
|
||
if len(issues) > len(storyboard) // 2:
|
||
return False, 0.4, "; ".join(issues[:3])
|
||
|
||
score = 0.9 if not issues else max(0.6, 0.9 - 0.1 * len(issues))
|
||
return True, score, f"分镜脚本完整 ({len(storyboard)} 个分镜, {issues and '有'+str(len(issues))+'个问题' or '无问题'})"
|
||
|
||
|
||
async def eval_ktv_synthesis(output: dict) -> Tuple[bool, float, str]:
|
||
"""评估 KTV 合成结果（步骤 17）。
|
||
|
||
评估标准：
|
||
1. ktv_path 文件存在且大小 > 1MB
|
||
2. 视频时长 > 30 秒
|
||
3. 字幕文件存在
|
||
4. MTV 版本可选
|
||
"""
|
||
ktv_path = output.get("ktv_path", "")
|
||
subtitle_path = output.get("subtitle_path", "")
|
||
|
||
# 检查 KTV 视频
|
||
if not ktv_path or not os.path.exists(ktv_path):
|
||
return False, 0.0, f"KTV 视频不存在: {ktv_path}"
|
||
|
||
file_size = os.path.getsize(ktv_path)
|
||
if file_size < 1024 * 1024:
|
||
return False, 0.3, f"KTV 视频过小 ({file_size/1024/1024:.1f}MB < 1MB)"
|
||
|
||
# 用 ffprobe 检查时长
|
||
import asyncio
|
||
proc = await asyncio.create_subprocess_shell(
|
||
f"ffprobe -v error -show_entries format=duration -of default=noprint_wrappers=1:nokey=1 '{ktv_path}'",
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE,
|
||
)
|
||
stdout, _ = await proc.communicate()
|
||
duration = float(stdout.decode().strip()) if stdout else 0
|
||
|
||
if duration < 30:
|
||
return False, 0.5, f"KTV 视频时长不足 ({duration:.1f}s < 30s)"
|
||
|
||
# 检查字幕
|
||
if not subtitle_path or not os.path.exists(subtitle_path):
|
||
return False, 0.6, f"字幕文件缺失: {subtitle_path}"
|
||
|
||
score = min(1.0, duration / 180) # 3分钟满分
|
||
return True, score, f"KTV 合成正常 ({duration:.1f}s, {file_size/1024/1024:.1f}MB)"
|
||
|
||
|
||
async def eval_scene_video_quality(output: dict) -> Tuple[bool, float, str]:
|
||
"""评估场景视频生成结果（步骤 14）— 多维度评估。
|
||
|
||
5 个评估维度：
|
||
视觉质量 (25%) | 运动流畅度 (20%) | 语义一致性 (30%)
|
||
角色一致性 (15%) | 时间连贯性 (10%)
|
||
"""
|
||
scene_video_path = output.get("scene_video_path", "")
|
||
|
||
if not scene_video_path or not os.path.exists(scene_video_path):
|
||
return False, 0.0, f"场景视频不存在: {scene_video_path}"
|
||
|
||
file_size = os.path.getsize(scene_video_path)
|
||
if file_size < 1024 * 1024:
|
||
return False, 0.3, f"场景视频过小 ({file_size/1024/1024:.1f}MB < 1MB)"
|
||
|
||
# 调用多维度评估
|
||
try:
|
||
from app.eval_video import evaluate_scene_video_quality
|
||
prompt = output.get("prompt", output.get("scene_prompt", ""))
|
||
reference_image = output.get("reference_image", "")
|
||
|
||
result = await evaluate_scene_video_quality(
|
||
video_path=scene_video_path,
|
||
prompt=prompt,
|
||
reference_image=reference_image,
|
||
)
|
||
|
||
overall_score = result["overall_score"]
|
||
overall_reason = result["overall_reason"]
|
||
passed = overall_score >= 0.6
|
||
|
||
# 将维度评分存入 output 供后续参考
|
||
output["_eval_dimensions"] = result.get("dimensions", {})
|
||
|
||
return passed, overall_score, overall_reason
|
||
|
||
except ImportError:
|
||
logger.warning("eval_video 模块未安装，降级为基础评估")
|
||
# 降级：仅检查时长
|
||
import asyncio
|
||
proc = await asyncio.create_subprocess_shell(
|
||
f"ffprobe -v error -show_entries format=duration -of default=noprint_wrappers=1:nokey=1 '{scene_video_path}'",
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE,
|
||
)
|
||
stdout, _ = await proc.communicate()
|
||
duration = float(stdout.decode().strip()) if stdout else 0
|
||
|
||
if duration < 10:
|
||
return False, 0.5, f"场景视频时长不足 ({duration:.1f}s < 10s)"
|
||
|
||
score = min(1.0, duration / 30)
|
||
return True, score, f"场景视频正常 ({duration:.1f}s, {file_size/1024/1024:.1f}MB)"
|