fix(documentary): tighten prompt contract and config guards

This commit is contained in:
linyq 2026-04-03 01:14:41 +08:00
parent 1d148370c5
commit f9539eac8c
4 changed files with 69 additions and 2 deletions

3
.gitignore vendored
View File

@ -45,5 +45,4 @@ task.md
openspec/*
AGENTS.md
CLAUDE.md
tests/*
!tests/test_documentary_frame_analysis_service.py
tests/.pytest_cache/

View File

@ -11,6 +11,14 @@ class DocumentaryAnalysisConfig:
custom_prompt: str = ""
max_concurrency: int = 2
def __post_init__(self) -> None:
if self.frame_interval_seconds <= 0:
raise ValueError("frame_interval_seconds must be > 0")
if self.vision_batch_size <= 0:
raise ValueError("vision_batch_size must be > 0")
if self.max_concurrency <= 0:
raise ValueError("max_concurrency must be > 0")
@dataclass(slots=True)
class FrameBatchResult:

View File

@ -7,6 +7,16 @@ class DocumentaryFrameAnalysisService:
首先请详细描述每一帧的关键视觉信息包含主要内容人物动作和场景
然后基于所有帧的分析请用简洁的语言总结整个视频片段中发生的主要活动或事件流程
请务必使用 JSON 格式输出
JSON 必须包含以下键
- frame_observations: 数组且长度必须为 {frame_count}
- overall_activity_summary: 字符串描述整个批次主要活动
示例结构
{{
"frame_observations": [
{{"timestamp": "00:00:00,000", "observation": "画面描述"}}
],
"overall_activity_summary": "本批次主要活动总结"
}}
请务必不要遗漏视频帧我提供了 {frame_count} 张视频帧frame_observations 必须包含 {frame_count} 个元素
""".strip()

View File

@ -12,6 +12,8 @@ class DocumentaryFrameAnalysisServiceTests(unittest.TestCase):
self.assertIn("我提供了 3 张视频帧", prompt)
self.assertNotIn("%s", prompt)
self.assertIn("frame_observations", prompt)
self.assertIn("overall_activity_summary", prompt)
def test_parse_failed_batch_keeps_raw_response_and_time_range(self):
service = DocumentaryFrameAnalysisService()
@ -28,3 +30,51 @@ class DocumentaryFrameAnalysisServiceTests(unittest.TestCase):
self.assertEqual("not-json", batch.raw_response)
self.assertEqual("00:00:00,000-00:00:03,000", batch.time_range)
self.assertTrue(batch.fallback_summary)
def test_parse_failed_batch_uses_non_empty_fallback_when_raw_response_is_empty(self):
service = DocumentaryFrameAnalysisService()
batch = service._build_failed_batch_result(
batch_index=3,
raw_response="",
error_message="Empty model response",
frame_paths=["/tmp/keyframe_000001_000001000.jpg"],
time_range="00:00:03,000-00:00:06,000",
)
self.assertEqual("failed", batch.status)
self.assertEqual("", batch.raw_response)
self.assertTrue(batch.fallback_summary)
class DocumentaryAnalysisConfigTests(unittest.TestCase):
def test_config_rejects_non_positive_frame_interval(self):
with self.assertRaises(ValueError):
DocumentaryAnalysisConfig(
video_path="/tmp/demo.mp4",
frame_interval_seconds=0,
vision_batch_size=5,
vision_llm_provider="openai",
vision_model_name="gpt-4o-mini",
)
def test_config_rejects_non_positive_batch_size(self):
with self.assertRaises(ValueError):
DocumentaryAnalysisConfig(
video_path="/tmp/demo.mp4",
frame_interval_seconds=5,
vision_batch_size=0,
vision_llm_provider="openai",
vision_model_name="gpt-4o-mini",
)
def test_config_rejects_non_positive_max_concurrency(self):
with self.assertRaises(ValueError):
DocumentaryAnalysisConfig(
video_path="/tmp/demo.mp4",
frame_interval_seconds=5,
vision_batch_size=5,
vision_llm_provider="openai",
vision_model_name="gpt-4o-mini",
max_concurrency=0,
)