Haystack 集成 TwelveLabs：基于 Pegasus 视频理解与 Marengo 跨模态检索的完整实战指南

Haystack 集成 TwelveLabs：基于 Pegasus 视频理解与 Marengo 跨模态检索的完整实战指南

【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack

本篇技术指南围绕 Haystack 与 TwelveLabs 官方集成展开，系统讲解TwelveLabsVideoConverter（基于 Pegasus 视频语言模型将视频直接转为文本分析）、TwelveLabsDocumentEmbedder与TwelveLabsTextEmbedder（基于 Marengo 多模态嵌入模型）三个核心组件。读完本文，你将掌握如何在 Haystack 流水线中完成"视频 → 文本 → 向量 → 跨模态检索"的完整链路，直接上手构建支持用文字搜索视频内容的 RAG 与语义检索系统。

集成概览：三个组件、两大模型

TwelveLabs 集成（Python 包名twelvelabs-haystack）为 Haystack 提供了两个方向的视频多模态能力：

• Pegasus（视频语言模型）：直接分析视频的视觉内容与其自带音频（ASR 转录），输出文本分析结果（如描述 + 字幕），由TwelveLabsVideoConverter使用；
• Marengo（多模态嵌入模型）：把文本、图像、音频、视频映射到同一个共享向量空间，由TwelveLabsDocumentEmbedder和TwelveLabsTextEmbedder使用，支持跨模态检索（例如用文本查询去检索视频库）。

安装集成包：

pip install twelvelabs-haystack

三个组件的 API 密钥默认都从环境变量TWELVELABS_API_KEY读取（详见后文"密钥管理"一节）。没有密钥时，也可以传入api_key参数覆盖：

from haystack.utils import Secret from haystack_integrations.components.embedders.twelvelabs import TwelveLabsTextEmbedder embedder = TwelveLabsTextEmbedder(api_key=Secret.from_token("<your-api-key>"))

TwelveLabsVideoConverter：用 Pegasus 把视频变成文档

TwelveLabsVideoConverter位于索引流水线的最前端（在任何 PreProcessor 或嵌入器之前），输入视频源、输出documents。与"抽帧 + 独立转写"的传统视频处理链路不同，Pegasus 在云端即时分析视频的视觉画面及其自身音频 ASR 结果，一段源视频只产出一个 Document，其content即为 Pegasus 的分析文本（描述加字幕），不需要额外的抽帧或转写步骤。

视频源可以是公网可直连的视频 URL，也可以是本地文件路径（会上传到 TwelveLabs，上限 200 MB）。处理失败的单条视频源会被跳过并记录 warning，不会导致整批失败。

初始化参数

__init__( *, api_key: Secret = Secret.from_env_var("TWELVELABS_API_KEY"), model: str = DEFAULT_MODEL, prompt: str = DEFAULT_PROMPT, temperature: float = 0.2, max_tokens: int = 16384 ) -> None

run 方法

run( sources: list[str], meta: dict[str, Any] | list[dict[str, Any]] | None = None, ) -> dict[str, list[Document]]

• sources：视频源列表，元素为公网直连视频 URL 或本地文件路径；
• meta：可选的附加元数据——传单个字典则应用到所有输出 Document，传与sources等长的字典列表则按源一一对应；
• 返回：键为documents的字典，值是该批生成的 Document 列表。

每个产出 Document 的meta中会携带请求信息，包括source、asset_id、analysis_id、model、provider。默认模型为pegasus1.5。

单独使用

from haystack_integrations.components.converters.twelvelabs import TwelveLabsVideoConverter converter = TwelveLabsVideoConverter() result = converter.run(sources=["https://example.com/clip.mp4"]) document = result["documents"][0] print(document.content) # Pegasus 输出的描述 + 字幕 print(document.meta) # 包含 source, asset_id, analysis_id, model, provider

自定义提示词

prompt可以完全掌控 Pegasus 的输出形态，配合temperature与max_tokens微调：

from haystack_integrations.components.converters.twelvelabs import TwelveLabsVideoConverter converter = TwelveLabsVideoConverter( prompt="Summarize this video in three bullet points and list any products shown.", temperature=0.2, max_tokens=1024, ) result = converter.run(sources=["https://example.com/clip.mp4"]) print(result["documents"][0].content)

附加元数据

from haystack_integrations.components.converters.twelvelabs import TwelveLabsVideoConverter converter = TwelveLabsVideoConverter() # 所有源使用同一份元数据 result = converter.run( sources=["https://example.com/a.mp4", "https://example.com/b.mp4"], meta={"campaign": "demo"}, ) # 每个源各自的元数据（列表须与 sources 对齐） result = converter.run( sources=["https://example.com/a.mp4", "https://example.com/b.mp4"], meta=[{"title": "Clip A"}, {"title": "Clip B"}], )

TwelveLabsDocumentEmbedder：为文档批量计算 Marengo 向量

TwelveLabsDocumentEmbedder对每个 Document 的content计算 Marengo 嵌入，并写入Document.embedding字段。这些向量是文档库做嵌入检索的必要前提：检索时把查询向量与文档向量比较（余弦相似度），找出最相关的文档。由于 Marengo 将文本、图像、音频、视频映射进同一个向量空间，这些嵌入天然支持跨模态检索。默认模型为marengo3.0。

初始化参数

__init__( *, api_key: Secret = Secret.from_env_var("TWELVELABS_API_KEY"), model: str = DEFAULT_MODEL, prefix: str = "", suffix: str = "", batch_size: int = 32, progress_bar: bool = True, meta_fields_to_embed: list[str] | None = None, embedding_separator: str = "\n" ) -> None

嵌入元数据提升检索质量

文本文档往往携带元数据，若其中包含语义独特且有区分度的字段，可以让它们参与嵌入以改善检索。通过meta_fields_to_embed指定字段名即可：

from haystack import Document from haystack_integrations.components.embedders.twelvelabs import TwelveLabsDocumentEmbedder doc = Document(content="some text", meta={"title": "relevant title", "page number": 18}) embedder = TwelveLabsDocumentEmbedder(meta_fields_to_embed=["title"]) docs_w_embeddings = embedder.run(documents=[doc])["documents"]

run 与 run_async

run(documents: list[Document]) -> dict[str, Any] run_async(documents: list[Document]) -> dict[str, Any]

• documents：待嵌入的 Document 列表（对每个元素的content计算嵌入）；
• 返回：含documents（输入副本并填充了embedding）与meta（请求元信息，含所用模型）两个键的字典；
• run_async在batch_size指定的每个批次内并发嵌入；
• 两个方法在输入不是 Document 列表时都会抛出TypeError。

单独使用示例：

from haystack import Document from haystack_integrations.components.embedders.twelvelabs import TwelveLabsDocumentEmbedder doc = Document(content="a cat playing piano") document_embedder = TwelveLabsDocumentEmbedder() result = document_embedder.run(documents=[doc]) print(result["documents"][0].embedding) # [-0.043398008, -0.025287028, -0.0061081843, ...]

TwelveLabsTextEmbedder：为查询字符串计算 Marengo 向量

TwelveLabsTextEmbedder把单个字符串（例如用户查询）转成向量，典型位置在查询/RAG 流水线中、嵌入型 Retriever 之前。它适合嵌入单条文本；要嵌入文档列表请使用TwelveLabsDocumentEmbedder。默认模型同样是marengo3.0。

由于 Marengo 的共享向量空间，文本嵌入与同一模型的图像、音频、视频嵌入可直接用余弦相似度比较——例如用文本查询搜索视频集合。这正是构建跨模态检索的关键。

初始化参数

__init__( *, api_key: Secret = Secret.from_env_var("TWELVELABS_API_KEY"), model: str = DEFAULT_MODEL, prefix: str = "", suffix: str = "" ) -> None

run 与 run_async

run(text: str) -> dict[str, Any] run_async(text: str) -> dict[str, Any]

• text：待嵌入的字符串；
• 返回：含embedding（输入字符串的向量）与meta（请求元信息，含所用模型）两个键的字典；
• 两个方法在输入不是字符串时都会抛出TypeError。

from haystack_integrations.components.embedders.twelvelabs import TwelveLabsTextEmbedder text_embedder = TwelveLabsTextEmbedder() result = text_embedder.run(text="a cat playing piano") print(result["embedding"]) # [-0.043398008, -0.025287028, -0.0061081843, ...] print(result["meta"]) # {'model': 'marengo3.0'}

端到端实战：视频 RAG 索引与跨模态查询

把三个组件串起来，即可构建"视频 → 分析文本 → 向量 → 文本查询召回视频"的完整系统。索引流水线用 Pegasus 分析视频、用 Marengo 嵌入分析结果，再写入以余弦相似度检索的InMemoryDocumentStore；查询流水线用TwelveLabsTextEmbedder嵌入查询，交给InMemoryEmbeddingRetriever召回。

from haystack import Document, Pipeline from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack.components.writers import DocumentWriter from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever from haystack_integrations.components.converters.twelvelabs import TwelveLabsVideoConverter from haystack_integrations.components.embedders.twelvelabs import ( TwelveLabsDocumentEmbedder, TwelveLabsTextEmbedder, ) document_store = InMemoryDocumentStore(embedding_similarity_function="cosine") # —— 索引流水线：视频 → Pegasus 文本分析 → Marengo 向量 → 写入文档库 —— indexing_pipeline = Pipeline() indexing_pipeline.add_component("converter", TwelveLabsVideoConverter()) indexing_pipeline.add_component("embedder", TwelveLabsDocumentEmbedder()) indexing_pipeline.add_component("writer", DocumentWriter(document_store=document_store)) indexing_pipeline.connect("converter", "embedder") indexing_pipeline.connect("embedder", "writer") indexing_pipeline.run({"converter": {"sources": ["https://example.com/clip.mp4"]}}) # —— 查询流水线：文本查询 → Marengo 向量 → 嵌入检索 —— query_pipeline = Pipeline() query_pipeline.add_component("text_embedder", TwelveLabsTextEmbedder()) query_pipeline.add_component("retriever", InMemoryEmbeddingRetriever(document_store=document_store)) query_pipeline.connect("text_embedder.embedding", "retriever.query_embedding") result = query_pipeline.run({"text_embedder": {"text": "feline making music"}}) print(result["retriever"]["documents"][0].content) # a cat playing piano

纯文本场景同样成立：先对文档批量嵌入并写入文档库，再以文本查询检索：

documents = [ Document(content="a cat playing piano"), Document(content="a dog catching a frisbee at the beach"), Document(content="a timelapse of a city skyline at night"), ] indexing_pipeline.run({"embedder": {"documents": documents}})

序列化：to_dict 与 from_dict

三个组件都实现了标准的 Haystack 序列化协议，便于把组件配置写入 YAML 或从字典恢复，从而支持流水线的保存与复用：

• to_dict() -> dict[str, Any]：将组件序列化为字典；
• from_dict(data: dict[str, Any]) -> <Component>：从字典反序列化出组件实例。

结合 Haystack 的流水线反序列化机制，你可以把 TwelveLabs 组件声明在 YAML 流水线文件中。需要注意：api_key以Secret形式序列化时，推荐使用环境变量类型（Secret.from_env_var），因为 token 类型的 Secret 本身不可序列化（见 haystack/utils/auth.py）。

密钥管理：为什么推荐环境变量

从源码 haystack/utils/auth.py 可以看到，Secret.from_env_var("TWELVELABS_API_KEY")创建的是EnvVarSecret：解析时按顺序读取候选环境变量，找到第一个已设置的变量即返回其值；strict=True时若全部未设置会直接抛错，避免密钥缺失被静默吞掉。而TokenSecret（Secret.from_token(...)）以字符串形式保存 token，且不可序列化。

因此官方使用文档的建议是：优先把TWELVELABS_API_KEY设为环境变量，而不是在参数中硬编码。这既能让组件配置可序列化、可安全入库，也能避免密钥通过print、日志或 traceback 泄露（token 的__repr__会被脱敏为<redacted>）。

使用建议与限制

• 组件定位：TwelveLabsVideoConverter放在索引流水线最前、PreProcessor 或嵌入器之前；TwelveLabsDocumentEmbedder放在DocumentWriter之前；TwelveLabsTextEmbedder放在嵌入型 Retriever 之前。
• 视频源限制：仅支持公网可直连的视频 URL 或本地文件路径，上传上限 200 MB。
• 容错行为：单条视频源处理失败只跳过并告警，不会拖垮整批任务。
• 模型版本：Pegasus 可选pegasus1.5/pegasus1.2，默认pegasus1.5；Marengo 默认marengo3.0。
• 异步能力：两个 Embedder 均提供run_async，batch_size内的文档并发嵌入，适合高吞吐索引场景；生产环境可关闭progress_bar保持日志整洁。
• 检索前提：查询向量与文档向量必须来自同一 Marengo 模型，才能保证共享向量空间内的余弦相似度比较有效。

至此，你已拥有在 Haystack 中构建完整视频多模态检索系统的三个核心组件：Pegasus 负责"看懂"视频，Marengo 负责"统一度量"，而 Haystack 流水线负责把它们与文档库、检索器无缝衔接。更详细的组件参考见 TwelveLabs 集成 API 文档，以及各组件使用文档：TwelveLabsVideoConverter、TwelveLabsDocumentEmbedder、TwelveLabsTextEmbedder。

【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack