From 844b5ef4f6035aefacd7b4103e2da637a7beebe9 Mon Sep 17 00:00:00 2001 From: Mohit Varikuti Date: Thu, 25 Jun 2026 02:17:23 -0700 Subject: [PATCH] feat: Add TwelveLabs Pegasus video-to-Markdown converter Adds an opt-in converter that uses the TwelveLabs Pegasus video-understanding model to convert video files (.mp4, .mov, .webm, .mkv, .avi, .m4v) to Markdown. Unlike the speech-only built-in AudioConverter, Pegasus reads the actual frames and audio, so the output can also describe visual scenes and on-screen action. The converter is only registered when a TwelveLabs API key is provided (via the twelvelabs_api_key argument or the TWELVELABS_API_KEY environment variable), so default behavior is unchanged. Adds a --use-twelvelabs CLI flag, the optional [twelvelabs] dependency, README docs, and tests (mocked unit tests plus a live API smoke test gated on TWELVELABS_API_KEY). --- README.md | 34 +++ packages/markitdown/pyproject.toml | 2 + .../markitdown/src/markitdown/__main__.py | 36 +++ .../markitdown/src/markitdown/_markitdown.py | 21 ++ .../src/markitdown/converters/__init__.py | 2 + .../converters/_twelvelabs_converter.py | 163 +++++++++++++ .../tests/test_twelvelabs_converter.py | 222 ++++++++++++++++++ 7 files changed, 480 insertions(+) create mode 100644 packages/markitdown/src/markitdown/converters/_twelvelabs_converter.py create mode 100644 packages/markitdown/tests/test_twelvelabs_converter.py diff --git a/README.md b/README.md index aa2f58bb8..ffc6f2512 100644 --- a/README.md +++ b/README.md @@ -110,6 +110,7 @@ At the moment, the following optional dependencies are available: * `[az-content-understanding]` Installs dependencies for Azure Content Understanding * `[audio-transcription]` Installs dependencies for audio transcription of wav and mp3 files * `[youtube-transcription]` Installs dependencies for fetching YouTube video transcription +* `[twelvelabs]` Installs dependencies for video understanding with TwelveLabs Pegasus ### Plugins @@ -280,6 +281,39 @@ result = md.convert("example.jpg") print(result.text_content) ``` +### TwelveLabs (Video Understanding) + +The built-in audio/video converter only transcribes speech. [TwelveLabs](https://twelvelabs.io) +Pegasus is a video-understanding model that reads the actual frames and audio, so the +resulting Markdown can also describe visual scenes, on-screen action, and structure — not +just the spoken words. It is opt-in and only activates for video files (`.mp4`, `.mov`, +`.webm`, `.mkv`, `.avi`, `.m4v`) when an API key is supplied. All other formats continue to +use the existing converters unchanged. + +Install: `pip install 'markitdown[twelvelabs]'` + +From the command line (reads the `TWELVELABS_API_KEY` environment variable): + +```bash +export TWELVELABS_API_KEY="" +markitdown talk.mp4 --use-twelvelabs -o talk.md +``` + +In Python (pass the key explicitly or via `TWELVELABS_API_KEY`): + +```python +from markitdown import MarkItDown + +md = MarkItDown( + twelvelabs_api_key="", + twelvelabs_prompt="Summarize this video as Markdown.", # optional +) +result = md.convert("talk.mp4") +print(result.text_content) +``` + +You can grab a free API key at [twelvelabs.io](https://twelvelabs.io) — there's a generous free tier. + ### Docker ```sh diff --git a/packages/markitdown/pyproject.toml b/packages/markitdown/pyproject.toml index d4c20a402..9f9adafff 100644 --- a/packages/markitdown/pyproject.toml +++ b/packages/markitdown/pyproject.toml @@ -49,6 +49,7 @@ all = [ "azure-ai-documentintelligence", "azure-ai-contentunderstanding>=1.2.0b1", "azure-identity", + "twelvelabs>=1.2.8", ] pptx = ["python-pptx"] docx = ["mammoth~=1.11.0", "lxml"] @@ -61,6 +62,7 @@ youtube-transcription = ["youtube-transcript-api"] az-doc-intel = ["azure-ai-documentintelligence", "azure-identity"] # >=1.2.0b1 required for to_llm_input() helper used by ContentUnderstandingConverter az-content-understanding = ["azure-ai-contentunderstanding>=1.2.0b1", "azure-identity"] +twelvelabs = ["twelvelabs>=1.2.8"] [project.urls] Documentation = "https://github.com/microsoft/markitdown#readme" diff --git a/packages/markitdown/src/markitdown/__main__.py b/packages/markitdown/src/markitdown/__main__.py index ccb44b64b..8f392390f 100644 --- a/packages/markitdown/src/markitdown/__main__.py +++ b/packages/markitdown/src/markitdown/__main__.py @@ -2,6 +2,7 @@ # # SPDX-License-Identifier: MIT import argparse +import os import sys import codecs from typing import Any, Dict @@ -94,6 +95,13 @@ def main(): help="Use Azure Content Understanding to extract text. Requires --cu-endpoint.", ) + cloud_group.add_argument( + "--use-twelvelabs", + action="store_true", + dest="use_twelvelabs", + help="Use TwelveLabs Pegasus to convert video to Markdown. Requires the TWELVELABS_API_KEY environment variable.", + ) + parser.add_argument( "-e", "--endpoint", @@ -119,6 +127,18 @@ def main(): help="Comma-separated list of file types to route to Content Understanding (e.g., pdf,jpeg,mp4). If omitted, all supported types are routed.", ) + parser.add_argument( + "--twelvelabs-model", + type=str, + help="TwelveLabs Pegasus model name to use with --use-twelvelabs (e.g., pegasus1.5).", + ) + + parser.add_argument( + "--twelvelabs-prompt", + type=str, + help="Instruction sent to Pegasus describing how to summarize the video when using --use-twelvelabs.", + ) + parser.add_argument( "-p", "--use-plugins", @@ -241,6 +261,22 @@ def main(): cu_kwargs["cu_file_types"] = cu_types markitdown = MarkItDown(enable_plugins=args.use_plugins, **cu_kwargs) + elif args.use_twelvelabs: + twelvelabs_api_key = os.environ.get("TWELVELABS_API_KEY") + if twelvelabs_api_key is None: + _exit_with_error( + "The TWELVELABS_API_KEY environment variable is required when using --use-twelvelabs." + ) + elif args.filename is None: + _exit_with_error("Filename is required when using TwelveLabs.") + + tl_kwargs: Dict[str, Any] = {"twelvelabs_api_key": twelvelabs_api_key} + if args.twelvelabs_model is not None: + tl_kwargs["twelvelabs_model"] = args.twelvelabs_model + if args.twelvelabs_prompt is not None: + tl_kwargs["twelvelabs_prompt"] = args.twelvelabs_prompt + + markitdown = MarkItDown(enable_plugins=args.use_plugins, **tl_kwargs) else: markitdown = MarkItDown(enable_plugins=args.use_plugins) diff --git a/packages/markitdown/src/markitdown/_markitdown.py b/packages/markitdown/src/markitdown/_markitdown.py index f6aa4df0e..bc67305bf 100644 --- a/packages/markitdown/src/markitdown/_markitdown.py +++ b/packages/markitdown/src/markitdown/_markitdown.py @@ -40,6 +40,7 @@ DocumentIntelligenceConverter, ContentUnderstandingConverter, CsvConverter, + TwelveLabsConverter, ) from ._base_converter import DocumentConverter, DocumentConverterResult @@ -248,6 +249,26 @@ def enable_builtins(self, **kwargs) -> None: ContentUnderstandingConverter(**cu_args), ) + # Register the TwelveLabs (Pegasus) video converter at the top of the + # stack if an API key is provided (via kwarg or TWELVELABS_API_KEY). + twelvelabs_api_key = kwargs.get("twelvelabs_api_key") or os.getenv( + "TWELVELABS_API_KEY" + ) + if twelvelabs_api_key is not None: + tl_args: Dict[str, Any] = {"api_key": twelvelabs_api_key} + + tl_model = kwargs.get("twelvelabs_model") + if tl_model is not None: + tl_args["model_name"] = tl_model + + tl_prompt = kwargs.get("twelvelabs_prompt") + if tl_prompt is not None: + tl_args["prompt"] = tl_prompt + + self.register_converter( + TwelveLabsConverter(**tl_args), + ) + self._builtins_enabled = True else: warn("Built-in converters are already enabled.", RuntimeWarning) diff --git a/packages/markitdown/src/markitdown/converters/__init__.py b/packages/markitdown/src/markitdown/converters/__init__.py index 77f8b1acd..93299d8f2 100644 --- a/packages/markitdown/src/markitdown/converters/__init__.py +++ b/packages/markitdown/src/markitdown/converters/__init__.py @@ -27,6 +27,7 @@ ) from ._epub_converter import EpubConverter from ._csv_converter import CsvConverter +from ._twelvelabs_converter import TwelveLabsConverter __all__ = [ "PlainTextConverter", @@ -51,4 +52,5 @@ "ContentUnderstandingFileType", "EpubConverter", "CsvConverter", + "TwelveLabsConverter", ] diff --git a/packages/markitdown/src/markitdown/converters/_twelvelabs_converter.py b/packages/markitdown/src/markitdown/converters/_twelvelabs_converter.py new file mode 100644 index 000000000..4ed1fc8ab --- /dev/null +++ b/packages/markitdown/src/markitdown/converters/_twelvelabs_converter.py @@ -0,0 +1,163 @@ +import os +import sys +import base64 +from typing import Any, BinaryIO, List, Optional, Union + +from .._base_converter import DocumentConverter, DocumentConverterResult +from .._stream_info import StreamInfo +from .._exceptions import MissingDependencyException + +# Try loading the optional (but in this case, required) dependency. +# Save reporting of any exceptions for later. +_dependency_exc_info = None +try: + from twelvelabs import TwelveLabs + from twelvelabs.types.video_context import VideoContext_Base64String +except ImportError: + # Preserve the error and stack trace for later + _dependency_exc_info = sys.exc_info() + + # Define these names for type hinting when the package is not available + class TwelveLabs: # type: ignore[no-redef] + pass + + class VideoContext_Base64String: # type: ignore[no-redef] + pass + + +# Video container formats understood by the Pegasus video-understanding model. +ACCEPTED_MIME_TYPE_PREFIXES = [ + "video/mp4", + "video/quicktime", + "video/webm", + "video/x-matroska", + "video/x-msvideo", + "video/x-m4v", +] + +ACCEPTED_FILE_EXTENSIONS = [ + ".mp4", + ".mov", + ".webm", + ".mkv", + ".avi", + ".m4v", +] + +DEFAULT_PROMPT = ( + "Describe this video in detail as Markdown. Summarize what happens, " + "transcribe any spoken words, and note key visual scenes in order." +) + + +class TwelveLabsConverter(DocumentConverter): + """Converts video files to Markdown using the TwelveLabs Pegasus video-understanding model. + + This converter is opt-in: it is only registered when a TwelveLabs API key is + supplied (via the ``twelvelabs_api_key`` argument to ``MarkItDown`` or the + ``TWELVELABS_API_KEY`` environment variable). Pegasus reads the actual video + frames and audio, so the resulting Markdown can capture visual scenes in + addition to speech, unlike the speech-only built-in ``AudioConverter``. + """ + + def __init__( + self, + *, + api_key: Optional[str] = None, + model_name: str = "pegasus1.5", + prompt: Optional[str] = None, + max_tokens: int = 2048, + file_extensions: Optional[List[str]] = None, + mime_type_prefixes: Optional[List[str]] = None, + ): + """ + Initialize the TwelveLabsConverter. + + Args: + api_key: TwelveLabs API key. Falls back to the ``TWELVELABS_API_KEY`` + environment variable when not provided. + model_name: Pegasus model to use (e.g. ``"pegasus1.5"``). + prompt: Instruction sent to Pegasus describing how to summarize the + video. Defaults to a general "describe this video as Markdown" prompt. + max_tokens: Maximum number of tokens Pegasus may generate. + file_extensions: Override the list of accepted file extensions. + mime_type_prefixes: Override the list of accepted MIME-type prefixes. + """ + super().__init__() + + # Raise an error if the dependency is not available. This converter is + # only instantiated when explicitly requested, so failing here is correct. + if _dependency_exc_info is not None: + raise MissingDependencyException( + "TwelveLabsConverter requires the optional dependency [twelvelabs] (or [all]) to be installed. E.g., `pip install markitdown[twelvelabs]`" + ) from _dependency_exc_info[ + 1 + ].with_traceback( # type: ignore[union-attr] + _dependency_exc_info[2] + ) + + resolved_key = api_key or os.environ.get("TWELVELABS_API_KEY") + if not resolved_key: + raise ValueError( + "TwelveLabsConverter requires an API key. Pass twelvelabs_api_key=... " + "or set the TWELVELABS_API_KEY environment variable. " + "Grab a free key at https://twelvelabs.io." + ) + + self._model_name = model_name + self._prompt = prompt + self._max_tokens = max_tokens + self._file_extensions = ( + file_extensions if file_extensions is not None else ACCEPTED_FILE_EXTENSIONS + ) + self._mime_type_prefixes = ( + mime_type_prefixes + if mime_type_prefixes is not None + else ACCEPTED_MIME_TYPE_PREFIXES + ) + self._client = TwelveLabs(api_key=resolved_key) + + def accepts( + self, + file_stream: BinaryIO, + stream_info: StreamInfo, + **kwargs: Any, # Options to pass to the converter + ) -> bool: + mimetype = (stream_info.mimetype or "").lower() + extension = (stream_info.extension or "").lower() + + if extension in self._file_extensions: + return True + + for prefix in self._mime_type_prefixes: + if mimetype.startswith(prefix): + return True + + return False + + def convert( + self, + file_stream: BinaryIO, + stream_info: StreamInfo, + **kwargs: Any, # Options to pass to the converter + ) -> DocumentConverterResult: + prompt = kwargs.get("twelvelabs_prompt") or self._prompt or DEFAULT_PROMPT + + # Read the video bytes and base64-encode them for the analyze request. + cur_pos = file_stream.tell() + try: + base64_video = base64.b64encode(file_stream.read()).decode("ascii") + finally: + file_stream.seek(cur_pos) + + response = self._client.analyze( + model_name=self._model_name, + video=VideoContext_Base64String(base_64_string=base64_video), + prompt=prompt, + max_tokens=self._max_tokens, + ) + + markdown = response.data or "" + + title: Union[str, None] = stream_info.filename + return DocumentConverterResult(markdown=markdown.strip(), title=title) diff --git a/packages/markitdown/tests/test_twelvelabs_converter.py b/packages/markitdown/tests/test_twelvelabs_converter.py new file mode 100644 index 000000000..4865737b6 --- /dev/null +++ b/packages/markitdown/tests/test_twelvelabs_converter.py @@ -0,0 +1,222 @@ +"""Tests for TwelveLabsConverter. + +Covers accepts() routing, convert() via a mocked TwelveLabs client, the +MissingDependency / missing-key error paths, and registration. A single +live test against the real Pegasus API is gated on TWELVELABS_API_KEY and +skipped when it is not set. Follows the pattern of test_cu_converter.py. +""" + +import io +import os + +import pytest + +from markitdown.converters._twelvelabs_converter import ( + TwelveLabsConverter, + ACCEPTED_FILE_EXTENSIONS, + ACCEPTED_MIME_TYPE_PREFIXES, + DEFAULT_PROMPT, +) +from markitdown._stream_info import StreamInfo + + +def _make_converter(): + """Create a converter bypassing __init__ (no SDK / API key needed).""" + conv = TwelveLabsConverter.__new__(TwelveLabsConverter) + conv._file_extensions = ACCEPTED_FILE_EXTENSIONS + conv._mime_type_prefixes = ACCEPTED_MIME_TYPE_PREFIXES + conv._model_name = "pegasus1.5" + conv._prompt = None + conv._max_tokens = 2048 + return conv + + +# --------------------------------------------------------------------------- +# accepts() routing +# --------------------------------------------------------------------------- + + +class TestAccepts: + @pytest.mark.parametrize("ext", [".mp4", ".mov", ".webm", ".mkv", ".avi", ".m4v"]) + def test_accepts_video_extensions(self, ext): + conv = _make_converter() + assert conv.accepts(io.BytesIO(b""), StreamInfo(extension=ext)) + + @pytest.mark.parametrize( + "mime", ["video/mp4", "video/quicktime", "video/webm", "video/x-matroska"] + ) + def test_accepts_video_mimetypes(self, mime): + conv = _make_converter() + assert conv.accepts(io.BytesIO(b""), StreamInfo(mimetype=mime)) + + @pytest.mark.parametrize("ext", [".pdf", ".mp3", ".wav", ".jpg", ".txt", ".zip"]) + def test_rejects_non_video(self, ext): + conv = _make_converter() + assert not conv.accepts(io.BytesIO(b""), StreamInfo(extension=ext)) + + +# --------------------------------------------------------------------------- +# convert() with a mocked client +# --------------------------------------------------------------------------- + + +class _FakeResponse: + def __init__(self, data): + self.data = data + + +class _FakeClient: + def __init__(self, data="# Video\n\nA cat plays piano."): + self._data = data + self.last_call = None + + def analyze(self, **kwargs): + self.last_call = kwargs + return _FakeResponse(self._data) + + +class TestConvert: + def test_returns_markdown_from_pegasus(self): + conv = _make_converter() + conv._client = _FakeClient(data="# Video\n\nA cat plays piano.") + + result = conv.convert( + io.BytesIO(b"fake video bytes"), + StreamInfo(extension=".mp4", filename="cat.mp4"), + ) + + assert result.markdown == "# Video\n\nA cat plays piano." + assert result.title == "cat.mp4" + + def test_wires_model_and_prompt(self): + conv = _make_converter() + conv._model_name = "pegasus1.5" + conv._client = _FakeClient() + + conv.convert(io.BytesIO(b"data"), StreamInfo(extension=".mp4")) + + call = conv._client.last_call + assert call["model_name"] == "pegasus1.5" + assert call["prompt"] == DEFAULT_PROMPT + assert call["max_tokens"] == 2048 + # video is passed as a base64 VideoContext + assert call["video"].base_64_string # non-empty base64 payload + + def test_per_call_prompt_override(self): + conv = _make_converter() + conv._client = _FakeClient() + + conv.convert( + io.BytesIO(b"data"), + StreamInfo(extension=".mp4"), + twelvelabs_prompt="List every spoken sentence.", + ) + + assert conv._client.last_call["prompt"] == "List every spoken sentence." + + def test_stream_position_restored(self): + conv = _make_converter() + conv._client = _FakeClient() + stream = io.BytesIO(b"abc") + stream.read(1) # advance position + conv.convert(stream, StreamInfo(extension=".mp4")) + assert stream.tell() == 1 + + def test_none_data_yields_empty_markdown(self): + conv = _make_converter() + conv._client = _FakeClient(data=None) + result = conv.convert(io.BytesIO(b"data"), StreamInfo(extension=".mp4")) + assert result.markdown == "" + + +# --------------------------------------------------------------------------- +# Error paths +# --------------------------------------------------------------------------- + + +class TestErrors: + def test_missing_dependency_message(self): + import markitdown.converters._twelvelabs_converter as tl_module + from markitdown._exceptions import MissingDependencyException + from unittest.mock import patch + + import_error = ImportError("No module named 'twelvelabs'") + dependency_exc_info = (ImportError, import_error, None) + + with patch.object( + tl_module, "_dependency_exc_info", dependency_exc_info + ), pytest.raises(MissingDependencyException) as exc_info: + TwelveLabsConverter(api_key="fake") + + assert "twelvelabs" in str(exc_info.value) + assert exc_info.value.__cause__ is import_error + + def test_missing_api_key_raises(self, monkeypatch): + import markitdown.converters._twelvelabs_converter as tl_module + from unittest.mock import patch + + monkeypatch.delenv("TWELVELABS_API_KEY", raising=False) + # Pretend the SDK is installed so we reach the key check. + with patch.object(tl_module, "_dependency_exc_info", None): + with pytest.raises(ValueError, match="requires an API key"): + TwelveLabsConverter() + + +# --------------------------------------------------------------------------- +# Registration +# --------------------------------------------------------------------------- + + +class TestRegistration: + def test_registered_when_api_key_provided(self): + from unittest.mock import patch + import markitdown.converters._twelvelabs_converter as tl_module + + with patch.object(tl_module, "_dependency_exc_info", None), patch.object( + tl_module, "TwelveLabs" + ): + from markitdown import MarkItDown + from markitdown.converters import TwelveLabsConverter as TLC + + md = MarkItDown(twelvelabs_api_key="fake-key") + assert any(isinstance(reg.converter, TLC) for reg in md._converters) + + def test_not_registered_without_api_key(self, monkeypatch): + monkeypatch.delenv("TWELVELABS_API_KEY", raising=False) + from markitdown import MarkItDown + from markitdown.converters import TwelveLabsConverter as TLC + + md = MarkItDown() + assert not any(isinstance(reg.converter, TLC) for reg in md._converters) + + +# --------------------------------------------------------------------------- +# Live API smoke test (opt-in) +# --------------------------------------------------------------------------- + + +@pytest.mark.skipif( + os.environ.get("TWELVELABS_API_KEY") is None, + reason="TWELVELABS_API_KEY not set; skipping live TwelveLabs API test.", +) +def test_live_pegasus_analyze(): + """End-to-end smoke test against the real Pegasus API. + + Uses a short public sample video. Pegasus analysis can be slow, so this is + opt-in and only runs when TWELVELABS_API_KEY is set. + """ + import urllib.request + + sample_url = ( + "https://commondatastorage.googleapis.com/" + "gtv-videos-bucket/sample/ForBiggerBlazes.mp4" + ) + video_bytes = urllib.request.urlopen(sample_url, timeout=60).read() + + conv = TwelveLabsConverter() + result = conv.convert( + io.BytesIO(video_bytes), + StreamInfo(extension=".mp4", mimetype="video/mp4", filename="sample.mp4"), + ) + assert isinstance(result.markdown, str) + assert len(result.markdown) > 0