Add optional TwelveLabs video segmentation backend (Marengo + Pegasus)

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:

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)

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({

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

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

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("全部测试通过")