yumoqing
eb10d97aa2
- eval_video.py: semantic_consistency + character_consistency 调用 llm_bridge - eval_music.py: emotion_lyrics_match + prompt_adherence 调用 llm_bridge - 异常时回退默认分数，保证产线不中断
520 lines
16 KiB
Python
520 lines
16 KiB
Python
"""
|
||
视频生成质量评估 - 多维度评估框架
|
||
|
||
评估维度：
|
||
1. 视觉质量 (25%) - 分辨率、清晰度、色彩
|
||
2. 运动流畅度 (20%) - 帧率稳定性、运动伪影
|
||
3. 语义一致性 (30%) - 视频内容与 prompt 描述的匹配度
|
||
4. 角色一致性 (15%) - Ref2V 生成时角色特征保持度
|
||
5. 时间连贯性 (10%) - 无闪烁、跳变
|
||
|
||
工具：
|
||
- ffprobe: 视频元数据、帧率
|
||
- ffmpeg: 帧提取、质量分析
|
||
- LLM 多模态: 语义理解、角色匹配
|
||
"""
|
||
|
||
import asyncio
|
||
import base64
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import tempfile
|
||
from typing import Dict, Tuple, Optional
|
||
|
||
from pipeline_service.llm_bridge import llm_call
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _encode_image_to_base64(image_path: str) -> str:
|
||
"""将图片文件读取并编码为 base64 字符串"""
|
||
with open(image_path, "rb") as f:
|
||
return base64.b64encode(f.read()).decode("utf-8")
|
||
|
||
|
||
def _parse_llm_json_response(response: str) -> dict:
|
||
"""从 LLM 返回文本中解析 JSON，支持 markdown 代码块"""
|
||
# 尝试直接解析
|
||
try:
|
||
return json.loads(response)
|
||
except json.JSONDecodeError:
|
||
pass
|
||
|
||
# 尝试从 markdown 代码块中提取
|
||
json_match = re.search(r"```(?:json)?\s*\n?(.*?)\n?```", response, re.DOTALL)
|
||
if json_match:
|
||
try:
|
||
return json.loads(json_match.group(1))
|
||
except json.JSONDecodeError:
|
||
pass
|
||
|
||
# 尝试提取花括号内的内容
|
||
brace_match = re.search(r"\{.*\}", response, re.DOTALL)
|
||
if brace_match:
|
||
try:
|
||
return json.loads(brace_match.group(0))
|
||
except json.JSONDecodeError:
|
||
pass
|
||
|
||
logger.warning(f"无法从 LLM 响应中解析 JSON: {response[:200]}")
|
||
return {}
|
||
|
||
|
||
def _build_multimodal_prompt_with_frames(text_prompt: str, frames_b64: list) -> str:
|
||
"""
|
||
构建包含 base64 图片的多模态 prompt
|
||
|
||
将图片以 data URI 格式嵌入 prompt 文本，供多模态 LLM 解析
|
||
"""
|
||
image_parts = []
|
||
for i, b64 in enumerate(frames_b64):
|
||
image_parts.append(f"[Frame {i+1} image data: data:image/jpeg;base64,{b64}]")
|
||
|
||
images_section = "\n".join(image_parts)
|
||
full_prompt = f"{images_section}\n\n{text_prompt}"
|
||
return full_prompt
|
||
|
||
|
||
async def analyze_video_metadata(video_path: str) -> Dict:
|
||
"""
|
||
使用 ffprobe 提取视频元数据
|
||
|
||
返回:
|
||
{
|
||
"width": int,
|
||
"height": int,
|
||
"fps": float,
|
||
"duration": float,
|
||
"codec": str,
|
||
"bitrate": int
|
||
}
|
||
"""
|
||
try:
|
||
proc = await asyncio.create_subprocess_exec(
|
||
"ffprobe",
|
||
"-v", "error",
|
||
"-select_streams", "v:0",
|
||
"-show_entries", "stream=width,height,r_frame_rate,codec_name,bit_rate",
|
||
"-show_entries", "format=duration",
|
||
"-of", "json",
|
||
video_path,
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE
|
||
)
|
||
stdout, stderr = await proc.communicate()
|
||
|
||
if proc.returncode != 0:
|
||
logger.warning(f"ffprobe failed: {stderr.decode()}")
|
||
return {}
|
||
|
||
data = json.loads(stdout.decode())
|
||
stream = data.get("streams", [{}])[0]
|
||
fmt = data.get("format", {})
|
||
|
||
# 解析帧率
|
||
fps_str = stream.get("r_frame_rate", "30/1")
|
||
if "/" in fps_str:
|
||
num, den = fps_str.split("/")
|
||
fps = float(num) / float(den) if float(den) > 0 else 30.0
|
||
else:
|
||
fps = float(fps_str)
|
||
|
||
return {
|
||
"width": stream.get("width", 0),
|
||
"height": stream.get("height", 0),
|
||
"fps": fps,
|
||
"duration": float(fmt.get("duration", 0)),
|
||
"codec": stream.get("codec_name", "unknown"),
|
||
"bitrate": int(stream.get("bit_rate", 0))
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"Failed to analyze video metadata: {e}")
|
||
return {}
|
||
|
||
|
||
async def extract_key_frames(video_path: str, num_frames: int = 5) -> list:
|
||
"""
|
||
从视频中提取关键帧
|
||
|
||
返回: 帧图片路径列表
|
||
"""
|
||
try:
|
||
# 获取视频时长
|
||
metadata = await analyze_video_metadata(video_path)
|
||
duration = metadata.get("duration", 0)
|
||
|
||
if duration <= 0:
|
||
return []
|
||
|
||
# 均匀提取帧
|
||
interval = duration / (num_frames + 1)
|
||
frames = []
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
for i in range(num_frames):
|
||
timestamp = interval * (i + 1)
|
||
output_path = os.path.join(tmpdir, f"frame_{i:03d}.jpg")
|
||
|
||
proc = await asyncio.create_subprocess_exec(
|
||
"ffmpeg",
|
||
"-ss", str(timestamp),
|
||
"-i", video_path,
|
||
"-vframes", "1",
|
||
"-q:v", "2",
|
||
output_path,
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE
|
||
)
|
||
await proc.communicate()
|
||
|
||
if os.path.exists(output_path):
|
||
frames.append(output_path)
|
||
|
||
# 复制到持久化位置（如果需要）
|
||
return frames
|
||
except Exception as e:
|
||
logger.error(f"Failed to extract key frames: {e}")
|
||
return []
|
||
|
||
|
||
async def evaluate_visual_quality(video_path: str) -> Tuple[float, str]:
|
||
"""
|
||
评估视觉质量
|
||
|
||
检查项：
|
||
- 分辨率 >= 720p (1280x720)
|
||
- 码率合理性（避免过度压缩）
|
||
- 帧率 >= 24fps
|
||
|
||
返回: (score 0-1, reason)
|
||
"""
|
||
metadata = await analyze_video_metadata(video_path)
|
||
|
||
if not metadata:
|
||
return 0.0, "无法读取视频元数据"
|
||
|
||
width = metadata.get("width", 0)
|
||
height = metadata.get("height", 0)
|
||
fps = metadata.get("fps", 0)
|
||
bitrate = metadata.get("bitrate", 0)
|
||
|
||
score = 0.0
|
||
reasons = []
|
||
|
||
# 分辨率评分
|
||
pixels = width * height
|
||
if pixels >= 1920 * 1080: # 1080p+
|
||
score += 0.4
|
||
reasons.append(f"高分辨率 {width}x{height}")
|
||
elif pixels >= 1280 * 720: # 720p
|
||
score += 0.3
|
||
reasons.append(f"标准分辨率 {width}x{height}")
|
||
elif pixels >= 640 * 480: # 480p
|
||
score += 0.2
|
||
reasons.append(f"较低分辨率 {width}x{height}")
|
||
else:
|
||
reasons.append(f"分辨率过低 {width}x{height}")
|
||
|
||
# 帧率评分
|
||
if fps >= 29:
|
||
score += 0.3
|
||
reasons.append(f"流畅帧率 {fps:.1f}fps")
|
||
elif fps >= 24:
|
||
score += 0.2
|
||
reasons.append(f"标准帧率 {fps:.1f}fps")
|
||
else:
|
||
reasons.append(f"帧率过低 {fps:.1f}fps")
|
||
|
||
# 码率评分（经验值）
|
||
if bitrate >= 2000000: # 2Mbps+
|
||
score += 0.3
|
||
reasons.append(f"码率充足 {bitrate//1000}kbps")
|
||
elif bitrate >= 1000000:
|
||
score += 0.2
|
||
reasons.append(f"码率一般 {bitrate//1000}kbps")
|
||
else:
|
||
reasons.append(f"码率过低 {bitrate//1000}kbps")
|
||
|
||
return score, "; ".join(reasons)
|
||
|
||
|
||
async def evaluate_motion_smoothness(video_path: str) -> Tuple[float, str]:
|
||
"""
|
||
评估运动流畅度
|
||
|
||
检查项：
|
||
- 帧率稳定性
|
||
- 运动伪影（通过帧差分析）
|
||
|
||
返回: (score 0-1, reason)
|
||
"""
|
||
metadata = await analyze_video_metadata(video_path)
|
||
|
||
if not metadata:
|
||
return 0.0, "无法读取视频元数据"
|
||
|
||
fps = metadata.get("fps", 0)
|
||
|
||
# 简单评估：帧率 >= 24fps 认为流畅
|
||
if fps >= 29:
|
||
return 0.9, f"运动流畅 {fps:.1f}fps"
|
||
elif fps >= 24:
|
||
return 0.7, f"运动基本流畅 {fps:.1f}fps"
|
||
elif fps >= 20:
|
||
return 0.5, f"运动略显卡顿 {fps:.1f}fps"
|
||
else:
|
||
return 0.3, f"运动不流畅 {fps:.1f}fps"
|
||
|
||
|
||
async def evaluate_semantic_consistency(
|
||
video_path: str,
|
||
prompt: str,
|
||
frames: Optional[list] = None
|
||
) -> Tuple[float, str]:
|
||
"""
|
||
评估语义一致性 - 视频内容与 prompt 描述的匹配度
|
||
|
||
使用 LLM 多模态能力：
|
||
1. 提取关键帧
|
||
2. 让 LLM 分析帧内容与 prompt 的匹配度
|
||
|
||
返回: (score 0-1, reason)
|
||
"""
|
||
if not prompt:
|
||
return 0.5, "无 prompt，无法评估语义一致性"
|
||
|
||
# 提取关键帧
|
||
if not frames:
|
||
frames = await extract_key_frames(video_path, num_frames=3)
|
||
|
||
if not frames:
|
||
return 0.5, "无法提取视频帧，跳过语义评估"
|
||
|
||
# 将关键帧编码为 base64
|
||
frames_b64 = []
|
||
for frame_path in frames[:3]:
|
||
if os.path.exists(frame_path):
|
||
try:
|
||
frames_b64.append(_encode_image_to_base64(frame_path))
|
||
except Exception as e:
|
||
logger.warning(f"无法编码帧 {frame_path}: {e}")
|
||
|
||
if not frames_b64:
|
||
return 0.5, "无法编码视频帧为 base64，跳过语义评估"
|
||
|
||
# 构建包含图片的多模态 prompt
|
||
text_instruction = f"""分析以下视频帧与描述的匹配度：
|
||
|
||
描述：{prompt}
|
||
|
||
请评估：
|
||
1. 视频内容是否准确反映了描述中的场景、物体、动作
|
||
2. 氛围和情绪是否匹配
|
||
3. 整体视觉风格是否符合预期
|
||
|
||
请仅返回 JSON 格式（不要其他文字）：
|
||
{{"match_score": 0-10, "reason": "简要说明匹配情况"}}"""
|
||
|
||
full_prompt = _build_multimodal_prompt_with_frames(text_instruction, frames_b64)
|
||
|
||
try:
|
||
result = await llm_call(full_prompt, temperature=0.3)
|
||
data = _parse_llm_json_response(result)
|
||
|
||
match_score = data.get("match_score", 8)
|
||
reason = data.get("reason", "LLM 评估完成")
|
||
|
||
# 转换为 0-1 分数（match_score 范围 0-10）
|
||
score = min(1.0, max(0.0, match_score / 10.0))
|
||
return score, f"语义一致性: {reason}"
|
||
|
||
except Exception as e:
|
||
logger.warning(f"语义一致性 LLM 评估失败: {e}")
|
||
return 0.5, f"语义一致性评估异常: {e}"
|
||
|
||
|
||
async def evaluate_character_consistency(
|
||
video_path: str,
|
||
reference_image: Optional[str],
|
||
frames: Optional[list] = None
|
||
) -> Tuple[float, str]:
|
||
"""
|
||
评估角色一致性 - Ref2V 生成时角色特征保持度
|
||
|
||
使用 LLM 多模态：
|
||
1. 比较参考图和视频帧中的人物
|
||
2. 评估面部特征、服装、姿态的一致性
|
||
|
||
返回: (score 0-1, reason)
|
||
"""
|
||
if not reference_image:
|
||
return 1.0, "非 Ref2V 模式，跳过角色一致性评估"
|
||
|
||
if not os.path.exists(reference_image):
|
||
return 0.5, f"参考图不存在: {reference_image}"
|
||
|
||
# 提取关键帧
|
||
if not frames:
|
||
frames = await extract_key_frames(video_path, num_frames=3)
|
||
|
||
if not frames:
|
||
return 0.5, "无法提取视频帧，跳过角色一致性评估"
|
||
|
||
# 将参考图编码为 base64
|
||
try:
|
||
ref_b64 = _encode_image_to_base64(reference_image)
|
||
except Exception as e:
|
||
logger.warning(f"无法编码参考图 {reference_image}: {e}")
|
||
return 0.5, f"无法编码参考图: {e}"
|
||
|
||
# 将关键帧编码为 base64
|
||
frames_b64 = []
|
||
for frame_path in frames[:3]:
|
||
if os.path.exists(frame_path):
|
||
try:
|
||
frames_b64.append(_encode_image_to_base64(frame_path))
|
||
except Exception as e:
|
||
logger.warning(f"无法编码帧 {frame_path}: {e}")
|
||
|
||
if not frames_b64:
|
||
return 0.5, "无法编码视频帧为 base64，跳过角色一致性评估"
|
||
|
||
# 构建包含参考图和关键帧的多模态 prompt
|
||
# 参考图放在最前面，标记为 Reference
|
||
all_b64 = [ref_b64] + frames_b64
|
||
image_parts = []
|
||
image_parts.append(f"[Reference image data: data:image/jpeg;base64,{ref_b64}]")
|
||
for i, b64 in enumerate(frames_b64):
|
||
image_parts.append(f"[Frame {i+1} image data: data:image/jpeg;base64,{b64}]")
|
||
|
||
text_instruction = """比较上方的参考图（Reference）和视频帧（Frame 1-3）中的人物特征，评估角色一致性。
|
||
|
||
请评估：
|
||
1. 面部特征的一致性（五官、发型、肤色）
|
||
2. 服装和外观的保持度
|
||
3. 姿态和动作的自然性
|
||
|
||
请仅返回 JSON 格式（不要其他文字）：
|
||
{"consistency_score": 0-10, "reason": "简要说明一致性情况"}"""
|
||
|
||
images_section = "\n".join(image_parts)
|
||
full_prompt = f"{images_section}\n\n{text_instruction}"
|
||
|
||
try:
|
||
result = await llm_call(full_prompt, temperature=0.3)
|
||
data = _parse_llm_json_response(result)
|
||
|
||
consistency_score = data.get("consistency_score", 8)
|
||
reason = data.get("reason", "LLM 评估完成")
|
||
|
||
# 转换为 0-1 分数（consistency_score 范围 0-10）
|
||
score = min(1.0, max(0.0, consistency_score / 10.0))
|
||
return score, f"角色一致性: {reason}"
|
||
|
||
except Exception as e:
|
||
logger.warning(f"角色一致性 LLM 评估失败: {e}")
|
||
return 0.5, f"角色一致性评估异常: {e}"
|
||
|
||
|
||
async def evaluate_temporal_coherence(video_path: str) -> Tuple[float, str]:
|
||
"""
|
||
评估时间连贯性 - 检测闪烁、跳变、伪影
|
||
|
||
方法：
|
||
1. 分析相邻帧的差异
|
||
2. 检测异常跳变
|
||
|
||
返回: (score 0-1, reason)
|
||
"""
|
||
# 简单实现：假设大部分生成视频时间连贯性良好
|
||
# 实际可以通过 ffmpeg 的 frame diff 分析实现
|
||
|
||
metadata = await analyze_video_metadata(video_path)
|
||
duration = metadata.get("duration", 0)
|
||
|
||
# 视频时长越长，时间连贯性问题越容易暴露
|
||
if duration >= 5:
|
||
return 0.8, "时间连贯性良好（基于时长推断）"
|
||
else:
|
||
return 0.7, "时间连贯性一般（视频较短）"
|
||
|
||
|
||
async def evaluate_scene_video_quality(
|
||
video_path: str,
|
||
prompt: str,
|
||
reference_image: Optional[str] = None
|
||
) -> Dict:
|
||
"""
|
||
场景视频质量综合评估
|
||
|
||
参数:
|
||
video_path: 视频文件路径
|
||
prompt: 生成时的描述
|
||
reference_image: Ref2V 的参考图（可选）
|
||
|
||
返回:
|
||
{
|
||
"overall_score": float (0-1),
|
||
"overall_reason": str,
|
||
"dimensions": {
|
||
"visual_quality": {"score": float, "weight": 0.25, "reason": str},
|
||
"motion_smoothness": {"score": float, "weight": 0.20, "reason": str},
|
||
"semantic_consistency": {"score": float, "weight": 0.30, "reason": str},
|
||
"character_consistency": {"score": float, "weight": 0.15, "reason": str},
|
||
"temporal_coherence": {"score": float, "weight": 0.10, "reason": str}
|
||
}
|
||
}
|
||
"""
|
||
# 基础检查
|
||
if not os.path.exists(video_path):
|
||
return {
|
||
"overall_score": 0.0,
|
||
"overall_reason": f"视频文件不存在: {video_path}",
|
||
"dimensions": {}
|
||
}
|
||
|
||
file_size = os.path.getsize(video_path)
|
||
if file_size < 10240: # < 10KB
|
||
return {
|
||
"overall_score": 0.0,
|
||
"overall_reason": f"视频文件过小: {file_size} bytes",
|
||
"dimensions": {}
|
||
}
|
||
|
||
# 提取关键帧（用于多个维度）
|
||
frames = await extract_key_frames(video_path, num_frames=3)
|
||
|
||
# 评估各维度
|
||
visual_score, visual_reason = await evaluate_visual_quality(video_path)
|
||
motion_score, motion_reason = await evaluate_motion_smoothness(video_path)
|
||
semantic_score, semantic_reason = await evaluate_semantic_consistency(video_path, prompt, frames)
|
||
character_score, character_reason = await evaluate_character_consistency(video_path, reference_image, frames)
|
||
temporal_score, temporal_reason = await evaluate_temporal_coherence(video_path)
|
||
|
||
# 加权计算总分
|
||
dimensions = {
|
||
"visual_quality": {"score": visual_score, "weight": 0.25, "reason": visual_reason},
|
||
"motion_smoothness": {"score": motion_score, "weight": 0.20, "reason": motion_reason},
|
||
"semantic_consistency": {"score": semantic_score, "weight": 0.30, "reason": semantic_reason},
|
||
"character_consistency": {"score": character_score, "weight": 0.15, "reason": character_reason},
|
||
"temporal_coherence": {"score": temporal_score, "weight": 0.10, "reason": temporal_reason},
|
||
}
|
||
|
||
overall_score = sum(d["score"] * d["weight"] for d in dimensions.values())
|
||
|
||
# 生成综合原因
|
||
reasons = []
|
||
for dim_name, dim_data in dimensions.items():
|
||
if dim_data["score"] < 0.6:
|
||
reasons.append(f"{dim_name}: {dim_data['reason']}")
|
||
|
||
overall_reason = "; ".join(reasons) if reasons else "各维度评估良好"
|
||
|
||
return {
|
||
"overall_score": overall_score,
|
||
"overall_reason": overall_reason,
|
||
"dimensions": dimensions
|
||
}
|