github-featured/pathwaycom--llm-app
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

production-harden TwelveLabs video RAG components (#10508)
lint PR / linter (push) Has been cancelled

Browse Source 
GitOrigin-RevId: bf7b49270b31b4e793d220f5f2b791b34e3521e5
This commit is contained in:
Sergey Kulik
2026-07-05 19:22:33 +02:00
committed by Manul from Pathway
parent 1cfa84a223
commit beb17f1ea0
6 changed files with 26 additions and 583 deletions
Download Patch File Download Diff File Expand all files Collapse all files
templates/video_rag_twelvelabs/.env.example
+4
View File
@@ -4,3 +4,7 @@ TWELVELABS_API_KEY=
# OpenAI API key, used by the question-answering LLM.
OPENAI_API_KEY=
# Pathway license key, required by the TwelveLabsVideoParser.
# Get a free key at https://pathway.com/features
PATHWAY_LICENSE_KEY=
templates/video_rag_twelvelabs/README.md
+11 -15
View File
@@ -46,13 +46,13 @@ video files ─▶ pw.io.fs.read (bytes)
─▶ REST API on :8000
```
The TwelveLabs components live in the local
[`pathway_twelvelabs`](pathway_twelvelabs/__init__.py) package:
The TwelveLabs components are part of the Pathway LLM xpack:
- `TwelveLabsVideoParser` — a Pathway parser (`pw.UDF`) that uploads the video
bytes as a TwelveLabs asset and asks Pegasus to describe it.
- `MarengoEmbedder` — a Pathway embedder (`BaseEmbedder`) that calls the Marengo
embedding endpoint.
- [`TwelveLabsVideoParser`](https://pathway.com/developers/api-docs/pathway-xpacks-llm/parsers#pathway.xpacks.llm.parsers.TwelveLabsVideoParser)
— a Pathway parser that uploads the video bytes as a TwelveLabs asset and asks
Pegasus to describe it.
- [`MarengoEmbedder`](https://pathway.com/developers/api-docs/pathway-xpacks-llm/embedders#pathway.xpacks.llm.embedders.MarengoEmbedder)
— a Pathway embedder that calls the Marengo embedding endpoint.
Both are wired in entirely through [`app.yaml`](app.yaml), so you can swap models,
prompts, the data source, or the LLM without touching any Python.
@@ -76,12 +76,14 @@ prompts, the data source, or the LLM without touching any Python.
- A TwelveLabs API key. Get a free one at [twelvelabs.io](https://twelvelabs.io) —
there is a generous free tier.
- An OpenAI API key for the question-answering LLM.
- A Pathway license key, required by the `TwelveLabsVideoParser`. Get a free one
at [pathway.com/features](https://pathway.com/features).
Copy `.env.example` to `.env` and fill in your keys:
```bash
cp .env.example .env
# edit .env and set TWELVELABS_API_KEY and OPENAI_API_KEY
# edit .env and set TWELVELABS_API_KEY, OPENAI_API_KEY and PATHWAY_LICENSE_KEY
```
Put one or more videos (e.g. `.mp4`, `.mov`) into the `data/` directory.
@@ -112,11 +114,5 @@ curl -X POST http://localhost:8000/v1/pw_ai_answer \
## Tests
A small test suite lives in [`test_twelvelabs.py`](test_twelvelabs.py). The
no-network tests run without any credentials; the live Marengo embedding test is
skipped unless `TWELVELABS_API_KEY` is set:
```bash
pip install -r requirements.txt pytest
TWELVELABS_API_KEY=... pytest templates/video_rag_twelvelabs/test_twelvelabs.py
```
The TwelveLabs components are tested as part of the Pathway core test suite, in
[`python/pathway/xpacks/llm/tests/test_twelvelabs.py`](https://github.com/pathwaycom/pathway/blob/main/python/pathway/xpacks/llm/tests/test_twelvelabs.py).
templates/video_rag_twelvelabs/app.py
+7 -8
View File
@@ -3,23 +3,22 @@
# Copyright © 2026 Pathway
import logging
import os
from pathlib import Path
from warnings import warn
import pathway as pw
# `pathway_twelvelabs` registers the TwelveLabsVideoParser and MarengoEmbedder
# classes so they can be referenced from `app.yaml` via the `!` YAML tags.
import pathway_twelvelabs # noqa: F401
from dotenv import load_dotenv
from pathway.xpacks.llm.question_answering import BaseRAGQuestionAnswerer
from pathway.xpacks.llm.servers import QASummaryRestServer
from pydantic import BaseModel, ConfigDict, InstanceOf
# To use advanced features with Pathway Live Data Framework Scale, get your free license key from
# https://pathway.com/features and paste it below.
# To use Pathway Live Data Framework Community, comment out the line below.
pw.set_license_key("demo-license-key-with-telemetry")
# The TwelveLabsVideoParser used by this template is a Pathway Scale feature.
# Get your free license key from https://pathway.com/features and set it in the
# PATHWAY_LICENSE_KEY environment variable (see .env.example) or paste it below.
pw.set_license_key(
os.environ.get("PATHWAY_LICENSE_KEY", "demo-license-key-with-telemetry")
)
logging.basicConfig(
level=logging.INFO,
templates/video_rag_twelvelabs/app.yaml
+4 -2
View File
@@ -32,15 +32,17 @@ $sources:
# The parser turns each video into text using the TwelveLabs Pegasus model.
# The TWELVELABS_API_KEY environment variable must be set (see .env.example).
# You can customize the prompt to extract exactly what your application needs.
$parser: !pathway_twelvelabs.TwelveLabsVideoParser
$parser: !pw.xpacks.llm.parsers.TwelveLabsVideoParser
model: "pegasus1.5"
max_tokens: 2048
cache_strategy: !pw.udfs.DefaultCache {}
# Uncomment so that a single malformed video does not halt the pipeline:
# on_error: "skip"
# prompt: "Describe this video, focusing on the products that appear and any prices shown."
# The embedder converts the parsed text into 512-dimensional multimodal
# embeddings using the TwelveLabs Marengo model.
$embedder: !pathway_twelvelabs.MarengoEmbedder
$embedder: !pw.xpacks.llm.embedders.MarengoEmbedder
model: "marengo3.0"
cache_strategy: !pw.udfs.DefaultCache {}
retry_strategy: !pw.udfs.ExponentialBackoffRetryStrategy {}
templates/video_rag_twelvelabs/pathway_twelvelabs/__init__.py
-309
View File
@@ -1,309 +0,0 @@
# Copyright © 2026 Pathway
"""TwelveLabs components for the Pathway Live Data Framework.
This module provides two opt-in building blocks that let Pathway pipelines work
directly with video:
* :class:`TwelveLabsVideoParser` - a Pathway parser (``pw.UDF``) that turns raw
video bytes into text using TwelveLabs' `Pegasus
<https://docs.twelvelabs.io/docs/concepts/models/pegasus>`_ video-understanding
model. The resulting text can be chunked, embedded and indexed by any of the
standard Pathway RAG components, exactly like the output of the built-in PDF
parsers.
* :class:`MarengoEmbedder` - a Pathway embedder (``BaseEmbedder``) backed by
TwelveLabs' `Marengo <https://docs.twelvelabs.io/docs/concepts/models/marengo>`_
multimodal embedding model. It returns 512-dimensional vectors that live in the
same embedding space for text, image, audio and video, which makes it a natural
choice when indexing the text produced by ``TwelveLabsVideoParser``.
Both components require the official ``twelvelabs`` Python SDK (``>=1.2.8``) and a
TwelveLabs API key. The key is read from the ``TWELVELABS_API_KEY`` environment
variable unless it is passed explicitly to the constructor.
"""
import asyncio
import logging
import os
import time
import numpy as np
import pathway as pw
from pathway import udfs
from pathway.xpacks.llm.embedders import BaseEmbedder
DEFAULT_PEGASUS_MODEL = "pegasus1.5"
DEFAULT_MARENGO_MODEL = "marengo3.0"
DEFAULT_PROMPT = (
"Describe this video in detail. Summarize what happens, who and what appears, "
"the setting, any spoken or on-screen text, and the overall topic. "
"Write the description so it can be used to answer questions about the video."
)
logger = logging.getLogger(__name__)
def _resolve_api_key(api_key: str | None) -> str:
key = api_key or os.environ.get("TWELVELABS_API_KEY")
if not key:
raise ValueError(
"TwelveLabs API key is missing. Pass `api_key=...` or set the "
"`TWELVELABS_API_KEY` environment variable."
)
return key
def _build_client(api_key: str | None):
try:
from twelvelabs import TwelveLabs
except ImportError as e:
raise ImportError(
"The `twelvelabs` package is required to use the TwelveLabs components. "
"Install it with `pip install twelvelabs>=1.2.8`."
) from e
return TwelveLabs(api_key=_resolve_api_key(api_key))
def _build_async_client(api_key: str | None):
try:
from twelvelabs import AsyncTwelveLabs
except ImportError as e:
raise ImportError(
"The `twelvelabs` package is required to use the TwelveLabs components. "
"Install it with `pip install twelvelabs>=1.2.8`."
) from e
return AsyncTwelveLabs(api_key=_resolve_api_key(api_key))
class TwelveLabsVideoParser(pw.UDF):
"""Parse videos into text using the TwelveLabs Pegasus model.
The parser uploads the incoming video bytes to TwelveLabs as an asset, waits
for the asset to be ready, and then asks Pegasus to produce a textual
description of the video using ``prompt``. The returned text is suitable for
chunking, embedding and indexing by the standard Pathway RAG components.
By default the uploaded asset is deleted once the analysis finishes (even if
the analysis fails), so repeated runs do not flood the TwelveLabs asset list.
Set ``delete_assets=False`` to keep the assets around for reuse or
inspection; in that case the emitted ``twelvelabs_asset_id`` metadata refers
to a live, retrievable asset.
Args:
prompt: Instruction sent to Pegasus describing what to extract from the
video. Defaults to a generic, RAG-oriented description prompt.
model: Pegasus model name. Defaults to ``"pegasus1.5"``.
api_key: TwelveLabs API key. If ``None``, the SDK reads it from the
``TWELVELABS_API_KEY`` environment variable.
max_tokens: Maximum number of tokens Pegasus may generate. Defaults to 2048.
temperature: Sampling temperature for Pegasus. Defaults to ``None`` (SDK default).
asset_poll_interval: Seconds between asset-readiness checks. Defaults to 5.
asset_timeout: Maximum number of seconds to wait for an uploaded asset to
become ready before raising. Defaults to 600.
delete_assets: If ``True`` (the default), the uploaded asset is deleted
after the analysis completes, so repeated runs do not accumulate
assets in your TwelveLabs account. When ``True``, the emitted
``twelvelabs_asset_id`` metadata is omitted because the asset no
longer exists. Set to ``False`` to keep assets (e.g. for reuse or
debugging), in which case the id is included in the metadata.
cache_strategy: Pathway caching strategy. To enable caching, pass a valid
:py:class:`~pathway.udfs.CacheStrategy`. Defaults to ``None``.
Example:
>>> import pathway as pw # doctest: +SKIP
>>> from pathway_twelvelabs import TwelveLabsVideoParser # doctest: +SKIP
>>> parser = TwelveLabsVideoParser() # doctest: +SKIP
"""
def __init__(
self,
prompt: str = DEFAULT_PROMPT,
model: str = DEFAULT_PEGASUS_MODEL,
api_key: str | None = None,
max_tokens: int = 2048,
temperature: float | None = None,
asset_poll_interval: float = 5.0,
asset_timeout: float = 600.0,
delete_assets: bool = True,
cache_strategy: udfs.CacheStrategy | None = None,
):
super().__init__(cache_strategy=cache_strategy)
self.prompt = prompt
self.model = model
self.max_tokens = max_tokens
self.temperature = temperature
self.asset_poll_interval = asset_poll_interval
self.asset_timeout = asset_timeout
self.delete_assets = delete_assets
self._api_key = api_key
self._client = None
@property
def client(self):
if self._client is None:
self._client = _build_client(self._api_key)
return self._client
def _upload_asset(self, contents: bytes) -> str:
"""Upload video bytes and return the asset id once it is ready."""
asset = self.client.assets.create(
method="direct", file=("video.mp4", contents), filename="video.mp4"
)
deadline = time.monotonic() + self.asset_timeout
while asset.status not in ("ready", "failed"):
if time.monotonic() > deadline:
raise TimeoutError(
f"TwelveLabs asset {asset.id} was not ready after "
f"{self.asset_timeout}s (last status: {asset.status})."
)
time.sleep(self.asset_poll_interval)
asset = self.client.assets.retrieve(asset.id)
if asset.status == "failed":
raise RuntimeError(f"TwelveLabs asset {asset.id} failed to process.")
return asset.id
def __wrapped__(self, contents: bytes, **kwargs) -> list[tuple[str, dict]]:
from twelvelabs.types.video_context import VideoContext_AssetId
asset_id = self._upload_asset(contents)
try:
logger.info("Analyzing TwelveLabs asset %s with Pegasus...", asset_id)
analyze_kwargs: dict = dict(
model_name=self.model,
video=VideoContext_AssetId(asset_id=asset_id),
prompt=self.prompt,
max_tokens=self.max_tokens,
)
if self.temperature is not None:
analyze_kwargs["temperature"] = self.temperature
response = self.client.analyze(**analyze_kwargs)
text = response.data or ""
finally:
if self.delete_assets:
# Remove the per-run asset so repeated runs do not flood the
# TwelveLabs asset list. Best-effort: a cleanup failure must not
# mask the analysis result (or an analysis error above).
try:
self.client.assets.delete(asset_id)
except Exception: # noqa: BLE001
logger.warning("Failed to delete TwelveLabs asset %s.", asset_id)
# When the asset has been deleted the id no longer resolves, so only
# surface it in the metadata when the asset is kept around.
metadata = {} if self.delete_assets else {"twelvelabs_asset_id": asset_id}
return [(text, metadata)]
def __call__(self, contents: pw.ColumnExpression, **kwargs) -> pw.ColumnExpression:
"""Parse the video document.
Args:
contents: Column with the raw bytes of each video.
Returns:
A column with a list of ``(text, metadata)`` pairs for each video.
When ``delete_assets=False`` the metadata records the TwelveLabs
``twelvelabs_asset_id`` used for the analysis; with the default
``delete_assets=True`` the asset is removed afterwards and the id is
omitted (it would no longer resolve).
"""
return super().__call__(contents, **kwargs)
class MarengoEmbedder(BaseEmbedder):
"""Embed text using the TwelveLabs Marengo multimodal embedding model.
Marengo returns 512-dimensional embeddings in a shared multimodal space, so the
text it produces is directly comparable with image, audio and video embeddings
from the same model. This makes it a natural retriever embedder for pipelines
that index video with :class:`TwelveLabsVideoParser`.
Args:
model: Marengo model name. Defaults to ``"marengo3.0"``.
api_key: TwelveLabs API key. If ``None``, the SDK reads it from the
``TWELVELABS_API_KEY`` environment variable.
capacity: Maximum number of concurrent operations. Defaults to ``None``
(no specific limit).
retry_strategy: Strategy for handling retries. Defaults to
:py:class:`~pathway.udfs.ExponentialBackoffRetryStrategy`.
cache_strategy: Pathway caching strategy. Defaults to ``None``.
Example:
>>> import pathway as pw # doctest: +SKIP
>>> from pathway_twelvelabs import MarengoEmbedder # doctest: +SKIP
>>> embedder = MarengoEmbedder() # doctest: +SKIP
"""
def __init__(
self,
*,
model: str = DEFAULT_MARENGO_MODEL,
api_key: str | None = None,
capacity: int | None = None,
retry_strategy: (
udfs.AsyncRetryStrategy | None
) = pw.udfs.ExponentialBackoffRetryStrategy(),
cache_strategy: udfs.CacheStrategy | None = None,
):
executor = udfs.async_executor(capacity=capacity, retry_strategy=retry_strategy)
# Marengo embeds one text per request, so keep batches at size 1.
super().__init__(
executor=executor, cache_strategy=cache_strategy, max_batch_size=1
)
self.model = model
self._api_key = api_key
self._client = None
self._aclient = None
@property
def client(self):
if self._client is None:
self._client = _build_client(self._api_key)
return self._client
@property
def aclient(self):
if self._aclient is None:
self._aclient = _build_async_client(self._api_key)
return self._aclient
def get_embedding_dimension(self, **kwargs) -> int:
"""Return the embedding dimension (512 for Marengo).
This is a one-time, setup-time probe: Pathway calls it once while
building the index, not on the per-document hot path. The single
synchronous request issued here is therefore intentional and acceptable
(the actual embedding hot path runs asynchronously via ``__wrapped__``).
The base implementation probes ``__wrapped__`` with a single string and
takes ``len`` of the result; since this embedder always returns a list of
vectors, probe with a one-element list and measure the first vector
instead (mirroring Pathway's ``SentenceTransformerEmbedder``).
"""
return len(self._embed_one("."))
def _embed_one(self, text: str) -> np.ndarray:
"""Synchronous single-text embed, used only for the setup-time probe."""
response = self.client.embed.create(model_name=self.model, text=text)
vector = response.text_embedding.segments[0].float_
return np.array(vector, dtype=np.float32)
async def _aembed_one(self, text: str) -> np.ndarray:
resp = await self.aclient.embed.create(model_name=self.model, text=text)
vector = resp.text_embedding.segments[0].float_
return np.array(vector, dtype=np.float32)
async def __wrapped__(self, inputs: list[str], **kwargs) -> list[np.ndarray]:
"""Embed the given texts with Marengo.
Marengo embeds one text per request, so the requests are issued
concurrently on the async TwelveLabs client (``AsyncTwelveLabs``) rather
than serially, keeping the embedding hot path non-blocking.
Args:
inputs: the strings to embed.
Returns:
A list of 512-dimensional ``numpy`` arrays, one per input string.
"""
return list(await asyncio.gather(*[self._aembed_one(t) for t in inputs]))
templates/video_rag_twelvelabs/test_twelvelabs.py
-249
View File
@@ -1,249 +0,0 @@
# Copyright © 2026 Pathway
"""Tests for the TwelveLabs video-RAG components.
The no-network tests stub the TwelveLabs SDK and run without any credentials.
The live test is skipped unless ``TWELVELABS_API_KEY`` is set in the environment.
Run with::
pytest templates/video_rag_twelvelabs/test_twelvelabs.py
"""
import os
import sys
import types
import numpy as np
import pytest
# Importing `pathway.xpacks.llm.embedders` runs the xpacks package __init__,
# which eagerly imports sibling submodules (parsers, document_store, ...) that
# in turn pull in heavy, optional document-parsing dependencies (docling,
# unstructured, ...). Those are present in the `pathwaycom/pathway` Docker image
# used to run this template but are not needed to exercise the TwelveLabs
# components. If they are unavailable, stub the siblings so the import chain
# succeeds in a lightweight test environment.
try: # pragma: no cover - only triggers when optional deps are missing
import pathway.xpacks.llm.embedders # noqa: F401
except ImportError:
for _name in (
"parsers",
"document_store",
"question_answering",
"rerankers",
"servers",
"splitters",
"vector_store",
"llms",
"prompts",
):
_full = f"pathway.xpacks.llm.{_name}"
sys.modules.setdefault(_full, types.ModuleType(_full))
from pathway_twelvelabs import ( # noqa: E402
DEFAULT_MARENGO_MODEL,
DEFAULT_PEGASUS_MODEL,
MarengoEmbedder,
TwelveLabsVideoParser,
)
class _FakeSegment:
def __init__(self, vector):
self.float_ = vector
class _FakeTextEmbedding:
def __init__(self, vector):
self.segments = [_FakeSegment(vector)]
class _FakeEmbeddingResponse:
def __init__(self, vector):
self.text_embedding = _FakeTextEmbedding(vector)
class _FakeEmbed:
def __init__(self, vector):
self._vector = vector
self.calls = []
def create(self, *, model_name, text):
self.calls.append((model_name, text))
return _FakeEmbeddingResponse(self._vector)
class _FakeAsset:
def __init__(self, id, status):
self.id = id
self.status = status
class _FakeAssets:
def __init__(self):
self.uploaded = None
self.deleted = []
def create(self, *, method, file, filename):
self.uploaded = (method, filename)
return _FakeAsset("asset-123", "ready")
def retrieve(self, asset_id):
return _FakeAsset(asset_id, "ready")
def delete(self, asset_id):
self.deleted.append(asset_id)
class _FakeAnalyzeResponse:
def __init__(self, data):
self.data = data
class _FakeClient:
def __init__(self, *, vector=None, analyze_text="a description"):
self.embed = _FakeEmbed(vector or [0.0] * 512)
self.assets = _FakeAssets()
self._analyze_text = analyze_text
self.analyze_calls = []
def analyze(self, **kwargs):
self.analyze_calls.append(kwargs)
return _FakeAnalyzeResponse(self._analyze_text)
# --- No-network unit tests -------------------------------------------------
class _FakeAsyncEmbed:
def __init__(self, vector):
self._vector = vector
self.calls = []
async def create(self, *, model_name, text):
self.calls.append((model_name, text))
return _FakeEmbeddingResponse(self._vector)
class _FakeAsyncClient:
def __init__(self, *, vector=None):
self.embed = _FakeAsyncEmbed(vector or [0.0] * 512)
def test_embedder_returns_vector_array():
embedder = MarengoEmbedder()
embedder._client = _FakeClient(vector=list(range(512)))
out = embedder._embed_one("a red car")
assert isinstance(out, np.ndarray)
assert out.shape == (512,)
assert out.dtype == np.float32
assert embedder._client.embed.calls == [(DEFAULT_MARENGO_MODEL, "a red car")]
def test_embedder_defaults():
embedder = MarengoEmbedder()
assert embedder.model == DEFAULT_MARENGO_MODEL
# Marengo embeds a single text per request.
assert embedder.max_batch_size == 1
def test_embedder_wrapped_is_async_and_concurrent():
# The hot path runs on the async client and returns one array per input.
import asyncio
embedder = MarengoEmbedder()
embedder._aclient = _FakeAsyncClient(vector=list(range(512)))
out = asyncio.run(embedder.__wrapped__(["a red car", "a blue boat"]))
assert len(out) == 2
for arr in out:
assert isinstance(arr, np.ndarray)
assert arr.shape == (512,)
assert arr.dtype == np.float32
assert embedder._aclient.embed.calls == [
(DEFAULT_MARENGO_MODEL, "a red car"),
(DEFAULT_MARENGO_MODEL, "a blue boat"),
]
def test_embedding_dimension_probe_returns_512():
# `BaseEmbedder.get_embedding_dimension` probes `__wrapped__` with a single
# string (not a list); the index factory relies on this returning the true
# vector size, so `__wrapped__` must handle a bare string input.
embedder = MarengoEmbedder()
embedder._client = _FakeClient(vector=[0.0] * 512)
assert embedder.get_embedding_dimension() == 512
def test_video_parser_uploads_then_analyzes_and_deletes_asset():
# Default: delete_assets=True -> asset is removed and id omitted from metadata.
parser = TwelveLabsVideoParser(prompt="What happens?")
parser._client = _FakeClient(analyze_text="A red car drives on a highway.")
out = parser.__wrapped__(b"fake-video-bytes")
assert out == [("A red car drives on a highway.", {})]
# Asset was uploaded via the direct method...
assert parser._client.assets.uploaded[0] == "direct"
# ...deleted afterwards so runs don't flood the asset list...
assert parser._client.assets.deleted == ["asset-123"]
# ...and Pegasus was called with the right model, prompt and asset.
(call,) = parser._client.analyze_calls
assert call["model_name"] == DEFAULT_PEGASUS_MODEL
assert call["prompt"] == "What happens?"
assert call["video"].asset_id == "asset-123"
def test_video_parser_keeps_asset_when_disabled():
parser = TwelveLabsVideoParser(delete_assets=False)
parser._client = _FakeClient(analyze_text="desc")
out = parser.__wrapped__(b"bytes")
assert out == [("desc", {"twelvelabs_asset_id": "asset-123"})]
assert parser._client.assets.deleted == []
def test_video_parser_failed_asset_raises():
parser = TwelveLabsVideoParser()
client = _FakeClient()
client.assets.create = lambda **kw: _FakeAsset("a", "failed")
parser._client = client
with pytest.raises(RuntimeError):
parser.__wrapped__(b"bytes")
# Failure happens during upload (before analyze); nothing was deleted.
assert client.assets.deleted == []
def test_video_parser_deletes_asset_even_when_analyze_raises():
parser = TwelveLabsVideoParser()
client = _FakeClient()
def _boom(**kwargs):
raise RuntimeError("pegasus exploded")
client.analyze = _boom
parser._client = client
with pytest.raises(RuntimeError):
parser.__wrapped__(b"bytes")
# try/finally still cleaned up the uploaded asset.
assert client.assets.deleted == ["asset-123"]
# --- Live smoke test (requires TWELVELABS_API_KEY) -------------------------
@pytest.mark.skipif(
not os.environ.get("TWELVELABS_API_KEY"),
reason="TWELVELABS_API_KEY not set; skipping live TwelveLabs call",
)
def test_marengo_live_embedding_is_512_dim():
embedder = MarengoEmbedder()
vector = embedder._embed_one("a red car driving on a highway")
assert vector.shape == (512,)