From 4c37cec1dc617434f60baf5ae9e646ac22b8cee7 Mon Sep 17 00:00:00 2001 From: Mohit Varikuti Date: Tue, 23 Jun 2026 21:13:15 -0700 Subject: [PATCH] support TwelveLabs as an optional video segmentation backend Add an opt-in segmentation backend that analyzes video frames and audio directly via TwelveLabs (Marengo retrieval + Pegasus analysis), instead of relying on Whisper transcripts. Useful for speech-light content. Returns the same {start, end, summary, tags} structure so the existing clip/export flow is unchanged. Enabled via SEGMENTER_TYPE=twelvelabs; default behavior unchanged. --- README.md | 22 ++++ config.py | 18 +++ modules/processing_queue.py | 29 +++-- modules/twelvelabs_processor.py | 175 +++++++++++++++++++++++++++++ requirements.txt | 1 + tests/test_twelvelabs_processor.py | 76 +++++++++++++ web/gradio_ui.py | 19 +++- 7 files changed, 327 insertions(+), 13 deletions(-) create mode 100644 modules/twelvelabs_processor.py create mode 100644 tests/test_twelvelabs_processor.py diff --git a/README.md b/README.md index 58f6569..68c0fa4 100644 --- a/README.md +++ b/README.md @@ -19,6 +19,7 @@ language queries. - **SRT Export**: Generate subtitles with accurate timestamps and custom line length - **Batch Processing**: find a specific topic across multiple files - **Re-analysis**: Experiment with different prompts without reprocessing audio +- **Optional TwelveLabs backend**: Segment directly from video frames + audio (Marengo retrieval + Pegasus analysis) for speech-light content — opt-in via `SEGMENTER_TYPE=twelvelabs` ## ⚙️ Installation @@ -63,6 +64,27 @@ export DOUBAO_1_5_PRO_API_KEY=your_doubao_api_key 5. (Optional)set up gradio temp file directory: set os.environ['GRADIO_TEMP_DIR'] in config.py file. +6. (Optional) Use TwelveLabs as the segmentation backend: + +By default PreenCut segments video by running Whisper transcription and asking an +LLM to split the transcript. As an opt-in alternative you can use +[TwelveLabs](https://twelvelabs.io) to segment directly from the video frames and +audio — useful when there is little or no speech (sports, b-roll, surveillance, music +videos). It uses **Marengo** for semantic retrieval and **Pegasus** for clip +summaries/tags, and returns the same `{start, end, summary, tags}` structure, so the +existing analysis/clip/export flow is unchanged. + +```bash +export SEGMENTER_TYPE=twelvelabs +export TWELVELABS_API_KEY=your_twelvelabs_api_key +# optional overrides: +# export TWELVELABS_INDEX_NAME=preencut # index reused across runs +# export TWELVELABS_MAX_CLIPS=10 # max segments per file +``` + +Grab a free API key at https://twelvelabs.io — there's a generous free tier. +Leave `SEGMENTER_TYPE` unset (or `llm`) to keep the default behavior. + ## 🚀 Usage 1. Start the Gradio interface: diff --git a/config.py b/config.py index 462e587..e779e5e 100644 --- a/config.py +++ b/config.py @@ -82,6 +82,24 @@ def get_device_config(): } ] +# 视频分段后端配置 +# 'llm': 默认行为，基于Whisper字幕用大模型分段 +# 'twelvelabs': 可选后端，使用 TwelveLabs (Marengo检索 + Pegasus分析) 直接分析视频画面 +SEGMENTER_TYPE = os.getenv('SEGMENTER_TYPE', 'llm') # llm, twelvelabs + +# TwelveLabs配置 (仅在 SEGMENTER_TYPE='twelvelabs' 时使用) +# 申请免费API key: https://twelvelabs.io (有免费额度) +TWELVELABS_API_KEY_ENV_NAME = 'TWELVELABS_API_KEY' +TWELVELABS_INDEX_NAME = os.getenv('TWELVELABS_INDEX_NAME', 'preencut') +TWELVELABS_MAX_CLIPS = int(os.getenv('TWELVELABS_MAX_CLIPS', '10')) +TWELVELABS_MODEL_OPTIONS = [ + { + "label": "TwelveLabs (Marengo + Pegasus)", + "search_model": "marengo3.0", + "analyze_model": "pegasus1.2", + } +] + # 创建必要的目录 for folder in [TEMP_FOLDER, OUTPUT_FOLDER]: os.makedirs(folder, exist_ok=True) diff --git a/modules/processing_queue.py b/modules/processing_queue.py index 766436a..df4762e 100644 --- a/modules/processing_queue.py +++ b/modules/processing_queue.py @@ -8,7 +8,7 @@ from modules.llm_processor import LLMProcessor from modules.video_processor import VideoProcessor from modules.word_segmenter import WordSegmenter -from config import SPEECH_RECOGNIZER_TYPE +from config import SPEECH_RECOGNIZER_TYPE, SEGMENTER_TYPE from typing import List, Dict, Optional from utils import clear_cache @@ -68,7 +68,12 @@ def _process_queue(self): file_results = [] llm_model = task_result.get("llm_model") temperature = task_result.get("temperature") - llm = LLMProcessor(llm_model, temperature) + if SEGMENTER_TYPE == 'twelvelabs': + from modules.twelvelabs_processor import TwelveLabsProcessor + segmenter = TwelveLabsProcessor(llm_model) + else: + segmenter = None + llm = LLMProcessor(llm_model, temperature) if task_result['enable_alignment']: word_segmenter = WordSegmenter() @@ -106,13 +111,19 @@ def _process_queue(self): del aligner clear_cache() - # 调用大模型进行分段 - print("调用大模型进行分段...") - llm_inputs = [{key: segment.get(key) for key in - ["start", "end", "text"]} for segment in - result["segments"]] - segments = llm.segment_video(llm_inputs, prompt) - print(f"大模型分段完成，段数: {len(segments)}") + # 进行分段 + if segmenter is not None: + # TwelveLabs 直接分析视频画面（不依赖字幕） + print("调用 TwelveLabs 进行视频分段...") + segments = segmenter.segment_video(file_path, prompt) + else: + # 默认：基于字幕用大模型分段 + print("调用大模型进行分段...") + llm_inputs = [{key: segment.get(key) for key in + ["start", "end", "text"]} for segment in + result["segments"]] + segments = llm.segment_video(llm_inputs, prompt) + print(f"分段完成，段数: {len(segments)}") # 保存结果 file_results.append({ diff --git a/modules/twelvelabs_processor.py b/modules/twelvelabs_processor.py new file mode 100644 index 0000000..fbf6f73 --- /dev/null +++ b/modules/twelvelabs_processor.py @@ -0,0 +1,175 @@ +import json +import os + +from twelvelabs import TwelveLabs +from config import ( + TWELVELABS_MODEL_OPTIONS, + TWELVELABS_API_KEY_ENV_NAME, + TWELVELABS_MAX_CLIPS, + TWELVELABS_INDEX_NAME, +) +from typing import List, Dict, Optional + + +class TwelveLabsProcessor: + """使用 TwelveLabs 进行视频分段（可选后端）。 + + 与 LLMProcessor 不同，本处理器直接分析视频画面与音频： + - Marengo (search): 当用户提供自然语言提示时，按语义检索相关片段，得到 start/end 时间。 + - Pegasus (analyze): 为每个片段（或整段视频）生成一句话摘要和主题标签。 + + 返回结构与 LLMProcessor.segment_video 一致：[{start, end, summary, tags}]， + 因此可以无缝接入现有的剪辑/导出流程。 + """ + + def __init__(self, llm_model: str): + for model in TWELVELABS_MODEL_OPTIONS: + if model['label'] == llm_model: + self.search_model = model['search_model'] + self.analyze_model = model['analyze_model'] + break + else: + # 允许在未显式配置 label 时使用默认模型 + self.search_model = 'marengo3.0' + self.analyze_model = 'pegasus1.2' + + self.api_key = os.getenv(TWELVELABS_API_KEY_ENV_NAME) + if not self.api_key: + raise ValueError( + f"TwelveLabs API key is not set, " + f"please set the {TWELVELABS_API_KEY_ENV_NAME} environment variable" + ) + self.client = TwelveLabs(api_key=self.api_key) + self.max_clips = TWELVELABS_MAX_CLIPS + + def _ensure_index(self) -> str: + """复用或创建一个同时包含 Marengo 与 Pegasus 模型的索引。""" + for index in self.client.indexes.list(): + if index.index_name == TWELVELABS_INDEX_NAME: + return index.id + + created = self.client.indexes.create( + index_name=TWELVELABS_INDEX_NAME, + models=[ + {"model_name": self.search_model, + "model_options": ["visual", "audio"]}, + {"model_name": self.analyze_model, + "model_options": ["visual", "audio"]}, + ], + ) + return created.id + + def _index_video(self, index_id: str, video_path: str) -> str: + """上传并索引视频，返回 video_id（索引过程可能较慢）。""" + with open(video_path, "rb") as f: + task = self.client.tasks.create(index_id=index_id, video_file=f) + task = self.client.tasks.wait_for_done(task_id=task.id, sleep_interval=5) + if task.status != "ready": + raise RuntimeError(f"TwelveLabs 索引失败, 状态: {task.status}") + return task.video_id + + def _pegasus_segments(self, video_id: str, + prompt: Optional[str]) -> List[Dict]: + """用 Pegasus 对整段视频进行分段，得到 [{start, end, summary, tags}]。""" + instruction = prompt + "。" if prompt else "" + full_prompt = ( + f"{instruction}" + "请将这个视频分成不超过" + f"{self.max_clips}个有意义的片段，每个片段主题连贯。" + "为每个片段给出开始时间(秒)、结束时间(秒)、一句话内容摘要和1-3个主题标签。" + "严格返回JSON数组，每个元素包含四个键：start, end, summary, tags(字符串数组)。" + "只返回JSON，不要其它文字。" + ) + res = self.client.analyze( + model_name=self.analyze_model, + video_id=video_id, + prompt=full_prompt, + max_tokens=2048, + ) + return _parse_segments(res.data) + + def _marengo_windows(self, index_id: str, video_id: str, + prompt: str) -> List[tuple]: + """用 Marengo 语义检索，返回与提示相关的 (start, end) 时间窗。""" + pager = self.client.search.query( + index_id=index_id, + query_text=prompt, + search_options=["visual", "audio"], + group_by="clip", + page_limit=self.max_clips, + ) + windows = [] + for item in pager: + if item.video_id != video_id: + continue + if item.start is None or item.end is None: + continue + windows.append((float(item.start), float(item.end))) + return windows + + def segment_video(self, video_path: str, + prompt: Optional[str] = None) -> List[Dict]: + """对视频进行语义分段，返回 [{start, end, summary, tags}]。 + + 与 LLMProcessor.segment_video 不同，这里的第一个参数是视频文件路径， + 因为 TwelveLabs 直接分析视频画面而非字幕文本。 + + 流程： + 1. 上传并索引视频(同时建立 Marengo 与 Pegasus 模型索引)。 + 2. Pegasus(analyze) 生成带摘要和标签的视频分段。 + 3. 若提供了提示词，再用 Marengo(search) 语义检索相关时间窗， + 只保留与之重叠的片段，实现按需检索。 + """ + index_id = self._ensure_index() + video_id = self._index_video(index_id, video_path) + + segments = self._pegasus_segments(video_id, prompt) + + if prompt: + windows = self._marengo_windows(index_id, video_id, prompt) + if windows: + segments = [ + seg for seg in segments + if _overlaps_any(seg["start"], seg["end"], windows) + ] + return segments + + +def _overlaps_any(start: float, end: float, windows: List[tuple]) -> bool: + """片段 [start, end] 是否与任一检索窗口重叠。""" + for w_start, w_end in windows: + if start < w_end and end > w_start: + return True + return False + + +def _strip_code_fence(text: str) -> str: + text = text.strip() + if text.startswith("```json"): + text = text[7:] + elif text.startswith("```"): + text = text[3:] + if text.endswith("```"): + text = text[:-3] + return text.strip() + + +def _parse_segments(text: str) -> List[Dict]: + """解析整段视频的分段结果。""" + try: + segments = json.loads(_strip_code_fence(text)) + except (json.JSONDecodeError, AttributeError): + raise ValueError(f"处理 TwelveLabs 结果出错, 返回:{text}") + + normalized = [] + for seg in segments: + tags = seg.get("tags", []) + if isinstance(tags, str): + tags = [tags] + normalized.append({ + "start": float(seg["start"]), + "end": float(seg["end"]), + "summary": seg.get("summary", ""), + "tags": tags, + }) + return normalized diff --git a/requirements.txt b/requirements.txt index 05bb78d..1d1414b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,6 +3,7 @@ faster-whisper==1.1.1 ctranslate2==4.4.0 ffmpeg-python==0.2.0 openai==1.91.0 +twelvelabs>=1.2.8 # 可选的 TwelveLabs 分段后端 (SEGMENTER_TYPE=twelvelabs) fastapi==0.115.13 soundfile==0.13.1 jieba diff --git a/tests/test_twelvelabs_processor.py b/tests/test_twelvelabs_processor.py new file mode 100644 index 0000000..810f22c --- /dev/null +++ b/tests/test_twelvelabs_processor.py @@ -0,0 +1,76 @@ +"""TwelveLabs 分段后端的测试。 + +无网络的解析单测可直接运行： + python -m tests.test_twelvelabs_processor +或使用 pytest： + pytest tests/test_twelvelabs_processor.py + +设置 TWELVELABS_API_KEY 后会额外跑一个 Marengo 检索能力的连通性冒烟测试， +未设置时自动跳过。 +""" +import os +import sys + +# 允许从仓库根目录直接运行 +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from modules.twelvelabs_processor import ( + _strip_code_fence, + _overlaps_any, + _parse_segments, +) + + +def test_strip_code_fence(): + assert _strip_code_fence('```json\n{"a": 1}\n```') == '{"a": 1}' + assert _strip_code_fence('```\n[]\n```') == '[]' + assert _strip_code_fence('{"a": 1}') == '{"a": 1}' + + +def test_overlaps_any(): + windows = [(0.0, 5.0), (10.0, 12.0)] + assert _overlaps_any(1.0, 3.0, windows) is True # 完全落入第一个窗口 + assert _overlaps_any(4.0, 11.0, windows) is True # 跨越两个窗口 + assert _overlaps_any(6.0, 9.0, windows) is False # 落在窗口之间 + assert _overlaps_any(1.0, 3.0, []) is False # 无窗口 + + +def test_parse_segments(): + raw = ( + '[{"start": 0, "end": 5.5, "summary": "开场", "tags": ["介绍"]},' + ' {"start": 5.5, "end": 10, "summary": "演示", "tags": "产品"}]' + ) + segs = _parse_segments(raw) + assert len(segs) == 2 + assert segs[0]["start"] == 0.0 and segs[0]["end"] == 5.5 + assert segs[1]["tags"] == ["产品"] # 字符串标签归一化为列表 + + +def test_parse_segments_invalid_raises(): + try: + _parse_segments("not json at all") + except ValueError: + pass + else: + raise AssertionError("无效输入应抛出 ValueError") + + +def test_marengo_smoke(): + """连通性冒烟测试：Marengo 文本嵌入返回向量。需要 TWELVELABS_API_KEY。""" + if not os.getenv("TWELVELABS_API_KEY"): + print("SKIP test_marengo_smoke: TWELVELABS_API_KEY 未设置") + return + from twelvelabs import TwelveLabs + client = TwelveLabs(api_key=os.environ["TWELVELABS_API_KEY"]) + emb = client.embed.create(model_name="marengo3.0", text="a cat playing piano") + vector = emb.text_embedding.segments[0].float_ + assert len(vector) > 0 + print(f"OK test_marengo_smoke: 向量维度={len(vector)}") + + +if __name__ == "__main__": + for name, fn in sorted(globals().items()): + if name.startswith("test_") and callable(fn): + fn() + print(f"PASS {name}") + print("全部测试通过") diff --git a/web/gradio_ui.py b/web/gradio_ui.py index f7ca0a2..2de3ec1 100644 --- a/web/gradio_ui.py +++ b/web/gradio_ui.py @@ -364,13 +364,24 @@ def reanalyze_with_prompt(task_id: str, reanalyze_llm_model: str, try: # 使用新提示重新处理 - from modules.llm_processor import LLMProcessor - llm = LLMProcessor(reanalyze_llm_model, temperature) + from config import SEGMENTER_TYPE + if SEGMENTER_TYPE == 'twelvelabs': + from modules.twelvelabs_processor import TwelveLabsProcessor + segmenter = TwelveLabsProcessor(reanalyze_llm_model) + llm = None + else: + from modules.llm_processor import LLMProcessor + segmenter = None + llm = LLMProcessor(reanalyze_llm_model, temperature) updated_results = [] for file_data in task_result["result"]: - new_segments = llm.segment_video(file_data["align_result"], - new_prompt) + if segmenter is not None: + new_segments = segmenter.segment_video(file_data["filepath"], + new_prompt) + else: + new_segments = llm.segment_video(file_data["align_result"], + new_prompt) updated_results.append({ "filename": file_data["filename"], "filepath": file_data["filepath"],