feat(llm): add TwelveLabs Pegasus as an optional content-aware clipping backend

[mohit-twelvelabs]

Hi! I'm Mohit, I work at TwelveLabs (@mohit-twelvelabs).

What this adds

An optional content-aware clipping backend for the LLM step, powered by TwelveLabs Pegasus (a video understanding model).

The existing LLM backends (qwen/gpt/g4f/deepseek) reason over the ASR transcript only. Pegasus reasons over the actual video — visuals and audio — so it can pick good highlight segments even when the transcript alone is ambiguous (action shots, scene changes, on-screen events, music, etc.).

It plugs into the same llm_inference dispatch via a new pegasus model prefix and returns segments in FunClip's existing N. [start-end] text format, so the 'AI Clip' button and extract_timestamps() work unchanged.

Why it helps FunClip

FunClip's smart-clipping is currently transcript-bound. Pegasus extends "LLM-based clipping" to genuinely multimodal clip selection without changing the UX: same flow, same output format, just a richer signal for choosing what's worth keeping.

Opt-in / non-breaking

• New file funclip/llm/twelvelabs_api.py; no existing backend touched.
• pegasus1.5 added to the model dropdown; the default model is unchanged.
• twelvelabs is only imported lazily inside the function, so the new dep is only needed if you actually select Pegasus.

How it was tested

• No-network unit tests for the URL/local-file routing and prompt format.
• A live, key-gated smoke test (TWELVELABS_API_KEY) that runs Pegasus over a public sample video and asserts text is returned — verified locally end to end against twelvelabs==1.2.8; the model returned correctly-formatted segments parseable by extract_timestamps().

You can grab a free API key at https://twelvelabs.io — there's a generous free tier.

[gemini-code-assist]

Code Review

This pull request integrates TwelveLabs Pegasus as an optional video-understanding LLM backend for content-aware clipping in FunClip, updating the UI, requirements, and documentation accordingly. The review feedback highlights a critical timestamp format mismatch between Pegasus's seconds-based output and FunClip's expected SRT format, suggesting a conversion helper to resolve it. Additionally, the reviewer recommends wrapping the Pegasus inference call in a try-except block for better error handling in the UI and adding a corresponding unit test for the timestamp conversion logic.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

Critical Bug: Timestamp Format Mismatch

There is a critical format mismatch between the output of the Pegasus model and the expectations of FunClip's extract_timestamps() function:

1. PEGASUS_SYSTEM_PROMPT instructs the model to return timestamps in seconds (e.g., [12.5-15.0]).
2. extract_timestamps() in funclip/utils/trans_utils.py uses a strict regular expression \[(\d{2}:\d{2}:\d{2},\d{2,3})\s*-\s*(\d{2}:\d{2}:\d{2},\d{2,3})\] which only matches standard SRT timestamp formats (e.g., [00:00:12,500-00:00:15,000]).

Because of this, extract_timestamps() will always return an empty list for Pegasus's output, and the clipping pipeline will fail to extract any segments.

To fix this without breaking other backends, we can intercept the Pegasus response text and convert any seconds-based timestamps (e.g., [12.5-15.0]) into the standard [HH:MM:SS,mmm-HH:MM:SS,mmm] format before returning it. Additionally, we should explicitly validate the presence of the API key to prevent generic errors.

def _convert_seconds_to_hhmmss_mmm(seconds_str):
    try:
        seconds = float(seconds_str)
    except ValueError:
        return seconds_str
    hours = int(seconds // 3600)
    minutes = int((seconds % 3600) // 60)
    secs = int(seconds % 60)
    millis = int(round((seconds - int(seconds)) * 1000))
    return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}"


def _convert_response_timestamps(text):
    import re
    if not isinstance(text, str):
        return text
    def repl(match):
        start_seq = match.group(1)
        end_seq = match.group(2)
        return f"[{_convert_seconds_to_hhmmss_mmm(start_seq)}-{_convert_seconds_to_hhmmss_mmm(end_seq)}]"
    return re.sub(r'\[(\d+(?:\.\d+)?)\s*-\s*(\d+(?:\.\d+)?)\]', repl, text)


def call_twelvelabs_pegasus(apikey,
                            video,
                            model="pegasus1.5",
                            prompt=None,
                            max_tokens=2048):
    """Run TwelveLabs Pegasus content-aware analysis over a video.

    Unlike the ASR-text based LLM backends, Pegasus reasons over the actual
    video (visuals + audio), which lets it pick highlight segments even when
    the transcript alone is ambiguous. Returns Pegasus' text response, which
    is expected to follow the "N. [start-end] text" segment format.

    Args:
        apikey: TwelveLabs API key (https://twelvelabs.io, free tier available).
        video: a public video URL or a local video file path.
        model: Pegasus model name, e.g. "pegasus1.5".
        prompt: optional extra instruction appended to the default prompt.
        max_tokens: max tokens for the Pegasus response.
    """
    from twelvelabs import TwelveLabs

    api_key = apikey or os.environ.get("TWELVELABS_API_KEY")
    if not api_key:
        raise ValueError("TwelveLabs API key is required. Please provide it via the UI or set the TWELVELABS_API_KEY environment variable.")

    client = TwelveLabs(api_key=api_key)
    video_context = _resolve_video_context(client, video)

    full_prompt = PEGASUS_SYSTEM_PROMPT
    if prompt is not None and len(prompt.strip()):
        full_prompt = full_prompt + "\n\n" + prompt.strip()

    response = client.analyze(
        model_name=model,
        video=video_context,
        prompt=full_prompt,
        max_tokens=max_tokens,
    )
    logging.info("TwelveLabs Pegasus inference done.")
    return _convert_response_timestamps(response.data)

[gemini-code-assist]

Robustness: Add Exception Handling for Pegasus Inference

Unlike standard text-based LLM APIs, the Pegasus pipeline involves video asset uploads, processing, and polling, which have more potential failure modes (e.g., network issues, invalid files, or API key errors). Wrapping the call in a try-except block ensures that any exceptions are caught and a friendly error message is returned to the Gradio UI instead of crashing the thread.

Suggested change

	if model.startswith('pegasus'):
	# TwelveLabs Pegasus reasons over the actual video (visuals + audio)
	# rather than the ASR transcript, so it needs the video source.
	if video_input is None:
	logging.error("Pegasus requires a video input; please upload a video first.")
	return "Please upload a video before running Pegasus inference."
	return call_twelvelabs_pegasus(apikey, video_input, model=model, prompt=system_content)
	if model.startswith('pegasus'):
	# TwelveLabs Pegasus reasons over the actual video (visuals + audio)
	# rather than the ASR transcript, so it needs the video source.
	if video_input is None:
	logging.error("Pegasus requires a video input; please upload a video first.")
	return "Please upload a video before running Pegasus inference."
	try:
	return call_twelvelabs_pegasus(apikey, video_input, model=model, prompt=system_content)
	except Exception as e:
	logging.error(f"Pegasus inference failed: {e}")
	return f"Error running Pegasus inference: {e}"

[gemini-code-assist]

Testing: Add Unit Test for Timestamp Conversion

To ensure the new timestamp conversion logic works correctly and handles edge cases (such as single-digit seconds, decimals, and multiple segments), we should add a dedicated unit test.

Suggested change

	def test_default_prompt_requests_parseable_segment_format(self):
	self.assertIn("[start_time-end_time]", PEGASUS_SYSTEM_PROMPT)
	def test_default_prompt_requests_parseable_segment_format(self):
	self.assertIn("[start_time-end_time]", PEGASUS_SYSTEM_PROMPT)
	def test_timestamp_conversion(self):
	from llm.twelvelabs_api import _convert_response_timestamps
	raw_response = "1. [12.5-15.0] description\n2. [120.45-135] another segment"
	expected = "1. [00:00:12,500-00:00:15,000] description\n2. [00:02:00,450-00:02:15,000] another segment"
	self.assertEqual(_convert_response_timestamps(raw_response), expected)

[LauraGPT]

Thanks @mohit-twelvelabs — this is a clean, well-scoped contribution. 🙌

I reviewed and verified it locally:

• Optional & non-intrusive: twelvelabs is imported lazily inside call_twelvelabs_pegasus / _resolve_video_context, so users who don't use Pegasus pay nothing; the existing qwen/gpt/g4f/moonshot/deepseek backends are untouched, and pegasus is just an added prefix in SUPPORT_LLM_PREFIX with a clear guard (requires a video input, returns a helpful message otherwise).
• Tests pass: python -m unittest tests.test_twelvelabs_pegasus → 3 passed (URL-context / missing-file / prompt-format), live smoke test correctly skipUnless TWELVELABS_API_KEY.
• Fits the repo convention: the dep goes in requirements.txt alongside the other backend deps, API key via apikey or os.environ["TWELVELABS_API_KEY"].

The value is real: the existing backends reason over the ASR transcript only, while Pegasus reasons over the actual video (visuals + audio), so content-aware clipping can use cues a transcript can't capture. Merging — thanks for adding it!