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_keySecret.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参数类型默认值说明api_keySecretTWELVELABS_API_KEY环境变量TwelveLabs API 密钥modelstrDEFAULT_MODELPegasus 模型名可选pegasus1.5或pegasus1.2promptstrDEFAULT_PROMPT发送给 Pegasus 的分析提示词temperaturefloat0.2采样温度取值范围 0–1max_tokensint16384每次分析的最大输出 token 数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( promptSummarize this video in three bullet points and list any products shown., temperature0.2, max_tokens1024, ) 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参数类型默认值说明api_keySecretTWELVELABS_API_KEY环境变量TwelveLabs API 密钥modelstrDEFAULT_MODELMarengo 模型名prefixstr嵌入前附加到文本开头的字符串suffixstr嵌入前附加到文本末尾的字符串batch_sizeint32每批处理的 Document 数run_async时同批内并发嵌入progress_barboolTrue是否显示进度条生产环境可关闭以保持日志干净meta_fields_to_embedlist[str] | NoneNone需要与文档正文一起参与嵌入的 meta 字段名列表embedding_separatorstr\n拼接 meta 字段到正文时使用的分隔符嵌入元数据提升检索质量文本文档往往携带元数据若其中包含语义独特且有区分度的字段可以让它们参与嵌入以改善检索。通过meta_fields_to_embed指定字段名即可from haystack import Document from haystack_integrations.components.embedders.twelvelabs import TwelveLabsDocumentEmbedder doc Document(contentsome 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_asyncrun(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(contenta 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参数类型默认值说明api_keySecretTWELVELABS_API_KEY环境变量TwelveLabs API 密钥modelstrDEFAULT_MODELMarengo 模型名prefixstr嵌入前附加到文本开头的字符串suffixstr嵌入前附加到文本末尾的字符串run 与 run_asyncrun(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(texta 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_functioncosine) # —— 索引流水线视频 → Pegasus 文本分析 → Marengo 向量 → 写入文档库 —— indexing_pipeline Pipeline() indexing_pipeline.add_component(converter, TwelveLabsVideoConverter()) indexing_pipeline.add_component(embedder, TwelveLabsDocumentEmbedder()) indexing_pipeline.add_component(writer, DocumentWriter(document_storedocument_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_storedocument_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(contenta cat playing piano), Document(contenta dog catching a frisbee at the beach), Document(contenta 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解析时按顺序读取候选环境变量找到第一个已设置的变量即返回其值strictTrue时若全部未设置会直接抛错避免密钥缺失被静默吞掉。而TokenSecretSecret.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.5Marengo 默认marengo3.0。异步能力两个 Embedder 均提供run_asyncbatch_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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考