尧图网站建设 
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 的三方集成twelvelabs-haystack覆盖两个核心模型能力用 Pegasus 视频语言模型把视频就地分析为文本视觉 自带音频 ASR以及用 Marengo 多模态向量模型把文本、图像、音频、视频映射到同一向量空间。读完本文你将掌握三个组件的安装、参数、单独调用与管道编排方式能够直接搭建视频索引 跨模态语义检索的生产级 RAG 流水线。集成概览三个组件、两大模型TwelveLabs 集成在 Haystack 生态中属于独立安装的集成包由三个组件构成分属两大模型能力组件所属模块底层模型输入输出TwelveLabsVideoConvertercomponents.converters.twelvelabsPegasus视频语言模型视频 URL 或本地文件路径列表每个视频一个Document内容为分析文本TwelveLabsDocumentEmbeddercomponents.embedders.twelvelabsMarengo多模态向量模型Document列表带embedding的新Document副本TwelveLabsTextEmbeddercomponents.embedders.twelvelabsMarengo多模态向量模型单个字符串该字符串的嵌入向量安装方式统一为pip install twelvelabs-haystack三个组件默认都从TWELVELABS_API_KEY环境变量读取 API Key也支持在初始化时通过api_key参数显式传入用Secret包装。建议优先使用环境变量方式避免密钥泄露到代码或配置文件中。说明本集成组件源码位于独立的twelvelabs-haystack集成包不在本仓库核心代码内本文引用仓库内文件均指向 Haystack 核心库中与集成协同工作的基础组件与数据结构。环境准备与密钥管理在开始之前需要准备 TwelveLabs API Key可通过 TwelveLabs 官方 Playground 申请并完成环境变量设置export TWELVELABS_API_KEYyour-api-key如果你想在代码中显式传入密钥Haystack 提供了Secret封装定义于 auth.py支持Secret.from_token(...)或Secret.from_env_var(...)两种方式from haystack.utils import Secret from haystack_integrations.components.embedders.twelvelabs import TwelveLabsDocumentEmbedder embedder TwelveLabsDocumentEmbedder(api_keySecret.from_token(your-api-key))TwelveLabs 集成的三个组件在__init__中默认使用Secret.from_env_var(TWELVELABS_API_KEY)读取环境变量因此不传api_key也能直接工作。TwelveLabsVideoConverter视频原地转文本TwelveLabsVideoConverter使用 Pegasus 视频语言模型对每个源视频就地on the fly进行分析——同时理解视频画面visuals和它自身的音频转写ASR直接返回文本。每个源视频变成一个Document其content就是 Pegasus 的分析结果例如一段描述加一份转写无需单独的抽帧frame extraction或转写步骤。支持的视频来源有两类公开可访问的直接视频 URLdirect video URLs本地文件路径会上传到 TwelveLabs单个文件上限 200 MB。参数说明__init__( *, api_key: Secret Secret.from_env_var(TWELVELABS_API_KEY), model: str DEFAULT_MODEL, # Pegasus 模型名默认 pegasus1.5可选 pegasus1.2 prompt: str DEFAULT_PROMPT, # 发送给 Pegasus 的每段视频分析提示词 temperature: float 0.2, # 采样温度取值范围 0-1 max_tokens: int 16384 # 每次分析的最大输出 token 数 ) - None关键参数说明modelPegasus 模型名支持pegasus1.5或pegasus1.2默认pegasus1.5prompt控制分析内容与风格的分析提示词例如要求总结为三个要点并列出出现的商品temperature采样温度0-1值越低输出越确定默认0.2max_tokens每次分析的最大输出 token 数默认16384。单独使用最简单的用法是实例化后直接runfrom 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每个生成的Document会携带与请求相关的元数据包括source视频来源、asset_id、analysis_id、model所用模型、provider服务提供方等字段。使用自定义提示词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)为输出附加元数据run方法的meta参数支持两种用法——传单个字典应用到所有输出Document或传与sources对齐的字典列表实现逐源元数据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}, ) # 每个来源各自的元数据 result converter.run( sources[https://example.com/a.mp4, https://example.com/b.mp4], meta[{title: Clip A}, {title: Clip B}], )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列表。此外组件对单个来源的处理失败会跳过该来源并记录 warning单个坏源不会拖垮整批转换适合批量索引场景。序列化与反序列化与其他 Haystack 组件一致该组件实现了to_dict()序列化为字典与from_dict(data)从字典反序列化方便在Pipeline序列化、YAML 配置参见仓库 marshal/yaml.py等场景中完整保留配置。TwelveLabsDocumentEmbedder文档批量嵌入TwelveLabsDocumentEmbedder使用 Marengo 多模态向量模型对每个Document的content计算嵌入向量并写入Document.embedding字段对应 document.py 中定义的embedding: list[float]字段。返回的是输入文档的副本副本带上了embedding。Marengo 将文本、图像、音频、视频嵌入到同一个共享向量空间因此这些嵌入天然支持跨模态检索cross-modal retrieval。参数说明__init__( *, api_key: Secret Secret.from_env_var(TWELVELABS_API_KEY), model: str DEFAULT_MODEL, # Marengo 模型名默认 marengo3.0 prefix: str , # 嵌入前添加到每个文本开头的字符串 suffix: str , # 嵌入前添加到每个文本末尾的字符串 batch_size: int 32, # 每个批次的文档数批内 run_async 并发嵌入 progress_bar: bool True, # 是否显示进度条生产环境建议关闭以保持日志干净 meta_fields_to_embed: list[str] | None None, # 需要随文本一起嵌入的元数据字段名列表 embedding_separator: str \n # 拼接元数据字段与文档文本的分隔符 ) - None关键参数说明modelMarengo 模型名默认marengo3.0prefix/suffix嵌入前拼接到文本前后缀的字符串常用于提示词式区分文本类型batch_size每批处理的文档数量默认 32run_async会在批次内并发嵌入progress_bar是否显示进度条默认True生产部署中建议关闭以保持日志整洁meta_fields_to_embed需要随文档正文一起嵌入的元数据字段名列表embedding_separator拼接元数据与正文的分隔符默认换行符\n。嵌入元数据以提升检索效果文本文档通常带有元数据。如果某些元数据具有区分度和语义价值可以将其与正文一起嵌入从而改善检索效果。通过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]单独使用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, ...]run 与 run_asyncrun(documents: list[Document]) - dict[str, Any] run_async(documents: list[Document]) - dict[str, Any]输入documents文档列表嵌入其content输出键为documents输入副本已填充embedding与meta请求元数据记录所用模型的字典异常输入不是Document列表时抛出TypeErrorrun_async会在每个batch_size批次内并发嵌入文档适合批量索引场景下的异步提速。TwelveLabsTextEmbedder查询串嵌入TwelveLabsTextEmbedder使用 Marengo 将单个字符串通常是查询嵌入为向量。由于 Marengo 把文本、图像、音频、视频嵌入到同一共享向量空间该组件产出的文本嵌入可以直接与同一模型的图像/视频嵌入做余弦相似度比较从而实现跨模态检索。它的典型用途是在检索前把查询嵌入再去搜索一个充满 Marengo 嵌入的文档存储。参数说明__init__( *, api_key: Secret Secret.from_env_var(TWELVELABS_API_KEY), model: str DEFAULT_MODEL, # Marengo 模型名默认 marengo3.0 prefix: str , # 嵌入前添加到文本开头的字符串 suffix: str # 嵌入前添加到文本末尾的字符串 ) - None相比TwelveLabsDocumentEmbedder文本嵌入器更精简只保留api_key、model、prefix、suffix四个参数。单独使用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}run 与 run_asyncrun(text: str) - dict[str, Any] run_async(text: str) - dict[str, Any]输入text单个字符串输出键为embedding输入字符串的嵌入向量与meta请求元数据含所用模型的字典异常输入不是字符串时抛出TypeError。端到端实战视频索引 跨模态检索管道将三个组件串成完整流水线即可实现视频 → 文本分析 → 向量索引 → 文本查询检索视频的跨模态检索闭环。视频索引管道该索引管道先用 Pegasus 分析视频再用TwelveLabsDocumentEmbedder嵌入分析结果最后写入文档存储此处使用 Haystack 内置的 InMemoryDocumentStore并开启余弦相似度from haystack import Pipeline from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack.components.writers import DocumentWriter from haystack_integrations.components.converters.twelvelabs import ( TwelveLabsVideoConverter, ) from haystack_integrations.components.embedders.twelvelabs import ( TwelveLabsDocumentEmbedder, ) document_store InMemoryDocumentStore(embedding_similarity_functioncosine) 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]}})其中DocumentWriter实现位于 document_writer.py负责把嵌入后的文档写入文档存储。查询检索管道查询管道使用TwelveLabsTextEmbedder嵌入文本查询再交给嵌入检索器embedding_retriever.py从文档存储中召回最相似的文档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.embedders.twelvelabs import ( TwelveLabsDocumentEmbedder, TwelveLabsTextEmbedder, ) document_store InMemoryDocumentStore(embedding_similarity_functioncosine) 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 Pipeline() indexing_pipeline.add_component(embedder, TwelveLabsDocumentEmbedder()) indexing_pipeline.add_component(writer, DocumentWriter(document_storedocument_store)) indexing_pipeline.connect(embedder, writer) indexing_pipeline.run({embedder: {documents: documents}}) 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注意query_pipeline.connect(text_embedder.embedding, retriever.query_embedding)这一连接TwelveLabsTextEmbedder输出的embedding向量被直接接到检索器的query_embedding输入上。由于文本嵌入与文档嵌入出自同一 Marengo 共享向量空间即使用feline making music这样与原文不完全一致的描述性查询也能正确命中a cat playing piano这正是跨模态此处体现为语义等价检索的价值。三个组件的完整方法契约TwelveLabsVideoConverterto_dict() - dict[str, Any]序列化组件为字典from_dict(data: dict[str, Any]) - TwelveLabsVideoConverter从字典反序列化组件。TwelveLabsDocumentEmbedderto_dict() - dict[str, Any]序列化组件为字典from_dict(data: dict[str, Any]) - TwelveLabsDocumentEmbedder从字典反序列化组件run(documents: list[Document]) - dict[str, Any]嵌入文档列表run_async(documents: list[Document]) - dict[str, Any]异步嵌入文档列表批内并发。TwelveLabsTextEmbedderto_dict() - dict[str, Any]序列化组件为字典from_dict(data: dict[str, Any]) - TwelveLabsTextEmbedder从字典反序列化组件run(text: str) - dict[str, Any]嵌入单个字符串run_async(text: str) - dict[str, Any]异步嵌入单个字符串。这三个组件都完整实现 Haystack 组件的to_dict/from_dict序列化协议意味着你可以把包含它们的管道保存为 YAML/JSON 配置并在不同环境中重建参见 marshal/yaml.py 与 core/pipeline/base.py。常见问题与最佳实践API Key 管理始终优先使用TWELVELABS_API_KEY环境变量确需代码内传参时用Secret.from_token(...)包装避免明文密钥进入日志或版本库。视频来源限制仅支持公开可访问的直接视频 URL 或本地文件路径本地文件上传上限为 200 MB超大视频请先切片或改用直接 URL。单源失败不影响整批转换失败的单条来源会被跳过并记录 warning批量索引时可放心混入可能失效的 URL。生产环境关闭进度条TwelveLabsDocumentEmbedder(progress_barFalse)可避免海量文档嵌入时刷屏日志。批量异步提速大批量文档嵌入优先使用run_async借助batch_size批内并发显著缩短耗时。温度与 token 权衡视频分析需要事实准确性时降低temperature需要更长、更详尽的分析时调高max_tokens默认 16384 已很充裕。跨模态检索的基石务必让文档嵌入与查询嵌入使用同一个 Marengo 模型版本否则向量空间不一致余弦相似度比较将失去意义。延伸阅读组件速查本仓库的组件文档 twelvelabsvideoconverter.mdx、twelvelabsdocumentembedder.mdx、twelvelabstextembedder.mdx数据基类Document的定义见 document.py密钥封装Secret的from_token/from_env_var实现见 auth.py配套组件document_writer.py、embedding_retriever.py、InMemoryDocumentStore管道机制core/pipeline/base.py 与管道序列化支持 marshal/yaml.py。【免费下载链接】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），仅供参考 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 的三方集成twelvelabs-haystack覆盖两个核心模型能力用 Pegasus 视频语言模型把视频就地分析为文本视觉 自带音频 ASR以及用 Marengo 多模态向量模型把文本、图像、音频、视频映射到同一向量空间。读完本文你将掌握三个组件的安装、参数、单独调用与管道编排方式能够直接搭建视频索引 跨模态语义检索的生产级 RAG 流水线。集成概览三个组件、两大模型TwelveLabs 集成在 Haystack 生态中属于独立安装的集成包由三个组件构成分属两大模型能力组件所属模块底层模型输入输出TwelveLabsVideoConvertercomponents.converters.twelvelabsPegasus视频语言模型视频 URL 或本地文件路径列表每个视频一个Document内容为分析文本TwelveLabsDocumentEmbeddercomponents.embedders.twelvelabsMarengo多模态向量模型Document列表带embedding的新Document副本TwelveLabsTextEmbeddercomponents.embedders.twelvelabsMarengo多模态向量模型单个字符串该字符串的嵌入向量安装方式统一为pip install twelvelabs-haystack三个组件默认都从TWELVELABS_API_KEY环境变量读取 API Key也支持在初始化时通过api_key参数显式传入用Secret包装。建议优先使用环境变量方式避免密钥泄露到代码或配置文件中。说明本集成组件源码位于独立的twelvelabs-haystack集成包不在本仓库核心代码内本文引用仓库内文件均指向 Haystack 核心库中与集成协同工作的基础组件与数据结构。环境准备与密钥管理在开始之前需要准备 TwelveLabs API Key可通过 TwelveLabs 官方 Playground 申请并完成环境变量设置export TWELVELABS_API_KEYyour-api-key如果你想在代码中显式传入密钥Haystack 提供了Secret封装定义于 auth.py支持Secret.from_token(...)或Secret.from_env_var(...)两种方式from haystack.utils import Secret from haystack_integrations.components.embedders.twelvelabs import TwelveLabsDocumentEmbedder embedder TwelveLabsDocumentEmbedder(api_keySecret.from_token(your-api-key))TwelveLabs 集成的三个组件在__init__中默认使用Secret.from_env_var(TWELVELABS_API_KEY)读取环境变量因此不传api_key也能直接工作。TwelveLabsVideoConverter视频原地转文本TwelveLabsVideoConverter使用 Pegasus 视频语言模型对每个源视频就地on the fly进行分析——同时理解视频画面visuals和它自身的音频转写ASR直接返回文本。每个源视频变成一个Document其content就是 Pegasus 的分析结果例如一段描述加一份转写无需单独的抽帧frame extraction或转写步骤。支持的视频来源有两类公开可访问的直接视频 URLdirect video URLs本地文件路径会上传到 TwelveLabs单个文件上限 200 MB。参数说明__init__( *, api_key: Secret Secret.from_env_var(TWELVELABS_API_KEY), model: str DEFAULT_MODEL, # Pegasus 模型名默认 pegasus1.5可选 pegasus1.2 prompt: str DEFAULT_PROMPT, # 发送给 Pegasus 的每段视频分析提示词 temperature: float 0.2, # 采样温度取值范围 0-1 max_tokens: int 16384 # 每次分析的最大输出 token 数 ) - None关键参数说明modelPegasus 模型名支持pegasus1.5或pegasus1.2默认pegasus1.5prompt控制分析内容与风格的分析提示词例如要求总结为三个要点并列出出现的商品temperature采样温度0-1值越低输出越确定默认0.2max_tokens每次分析的最大输出 token 数默认16384。单独使用最简单的用法是实例化后直接runfrom 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每个生成的Document会携带与请求相关的元数据包括source视频来源、asset_id、analysis_id、model所用模型、provider服务提供方等字段。使用自定义提示词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)为输出附加元数据run方法的meta参数支持两种用法——传单个字典应用到所有输出Document或传与sources对齐的字典列表实现逐源元数据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}, ) # 每个来源各自的元数据 result converter.run( sources[https://example.com/a.mp4, https://example.com/b.mp4], meta[{title: Clip A}, {title: Clip B}], )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列表。此外组件对单个来源的处理失败会跳过该来源并记录 warning单个坏源不会拖垮整批转换适合批量索引场景。序列化与反序列化与其他 Haystack 组件一致该组件实现了to_dict()序列化为字典与from_dict(data)从字典反序列化方便在Pipeline序列化、YAML 配置参见仓库 marshal/yaml.py等场景中完整保留配置。TwelveLabsDocumentEmbedder文档批量嵌入TwelveLabsDocumentEmbedder使用 Marengo 多模态向量模型对每个Document的content计算嵌入向量并写入Document.embedding字段对应 document.py 中定义的embedding: list[float]字段。返回的是输入文档的副本副本带上了embedding。Marengo 将文本、图像、音频、视频嵌入到同一个共享向量空间因此这些嵌入天然支持跨模态检索cross-modal retrieval。参数说明__init__( *, api_key: Secret Secret.from_env_var(TWELVELABS_API_KEY), model: str DEFAULT_MODEL, # Marengo 模型名默认 marengo3.0 prefix: str , # 嵌入前添加到每个文本开头的字符串 suffix: str , # 嵌入前添加到每个文本末尾的字符串 batch_size: int 32, # 每个批次的文档数批内 run_async 并发嵌入 progress_bar: bool True, # 是否显示进度条生产环境建议关闭以保持日志干净 meta_fields_to_embed: list[str] | None None, # 需要随文本一起嵌入的元数据字段名列表 embedding_separator: str \n # 拼接元数据字段与文档文本的分隔符 ) - None关键参数说明modelMarengo 模型名默认marengo3.0prefix/suffix嵌入前拼接到文本前后缀的字符串常用于提示词式区分文本类型batch_size每批处理的文档数量默认 32run_async会在批次内并发嵌入progress_bar是否显示进度条默认True生产部署中建议关闭以保持日志整洁meta_fields_to_embed需要随文档正文一起嵌入的元数据字段名列表embedding_separator拼接元数据与正文的分隔符默认换行符\n。嵌入元数据以提升检索效果文本文档通常带有元数据。如果某些元数据具有区分度和语义价值可以将其与正文一起嵌入从而改善检索效果。通过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]单独使用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, ...]run 与 run_asyncrun(documents: list[Document]) - dict[str, Any] run_async(documents: list[Document]) - dict[str, Any]输入documents文档列表嵌入其content输出键为documents输入副本已填充embedding与meta请求元数据记录所用模型的字典异常输入不是Document列表时抛出TypeErrorrun_async会在每个batch_size批次内并发嵌入文档适合批量索引场景下的异步提速。TwelveLabsTextEmbedder查询串嵌入TwelveLabsTextEmbedder使用 Marengo 将单个字符串通常是查询嵌入为向量。由于 Marengo 把文本、图像、音频、视频嵌入到同一共享向量空间该组件产出的文本嵌入可以直接与同一模型的图像/视频嵌入做余弦相似度比较从而实现跨模态检索。它的典型用途是在检索前把查询嵌入再去搜索一个充满 Marengo 嵌入的文档存储。参数说明__init__( *, api_key: Secret Secret.from_env_var(TWELVELABS_API_KEY), model: str DEFAULT_MODEL, # Marengo 模型名默认 marengo3.0 prefix: str , # 嵌入前添加到文本开头的字符串 suffix: str # 嵌入前添加到文本末尾的字符串 ) - None相比TwelveLabsDocumentEmbedder文本嵌入器更精简只保留api_key、model、prefix、suffix四个参数。单独使用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}run 与 run_asyncrun(text: str) - dict[str, Any] run_async(text: str) - dict[str, Any]输入text单个字符串输出键为embedding输入字符串的嵌入向量与meta请求元数据含所用模型的字典异常输入不是字符串时抛出TypeError。端到端实战视频索引 跨模态检索管道将三个组件串成完整流水线即可实现视频 → 文本分析 → 向量索引 → 文本查询检索视频的跨模态检索闭环。视频索引管道该索引管道先用 Pegasus 分析视频再用TwelveLabsDocumentEmbedder嵌入分析结果最后写入文档存储此处使用 Haystack 内置的 InMemoryDocumentStore并开启余弦相似度from haystack import Pipeline from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack.components.writers import DocumentWriter from haystack_integrations.components.converters.twelvelabs import ( TwelveLabsVideoConverter, ) from haystack_integrations.components.embedders.twelvelabs import ( TwelveLabsDocumentEmbedder, ) document_store InMemoryDocumentStore(embedding_similarity_functioncosine) 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]}})其中DocumentWriter实现位于 document_writer.py负责把嵌入后的文档写入文档存储。查询检索管道查询管道使用TwelveLabsTextEmbedder嵌入文本查询再交给嵌入检索器embedding_retriever.py从文档存储中召回最相似的文档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.embedders.twelvelabs import ( TwelveLabsDocumentEmbedder, TwelveLabsTextEmbedder, ) document_store InMemoryDocumentStore(embedding_similarity_functioncosine) 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 Pipeline() indexing_pipeline.add_component(embedder, TwelveLabsDocumentEmbedder()) indexing_pipeline.add_component(writer, DocumentWriter(document_storedocument_store)) indexing_pipeline.connect(embedder, writer) indexing_pipeline.run({embedder: {documents: documents}}) 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注意query_pipeline.connect(text_embedder.embedding, retriever.query_embedding)这一连接TwelveLabsTextEmbedder输出的embedding向量被直接接到检索器的query_embedding输入上。由于文本嵌入与文档嵌入出自同一 Marengo 共享向量空间即使用feline making music这样与原文不完全一致的描述性查询也能正确命中a cat playing piano这正是跨模态此处体现为语义等价检索的价值。三个组件的完整方法契约TwelveLabsVideoConverterto_dict() - dict[str, Any]序列化组件为字典from_dict(data: dict[str, Any]) - TwelveLabsVideoConverter从字典反序列化组件。TwelveLabsDocumentEmbedderto_dict() - dict[str, Any]序列化组件为字典from_dict(data: dict[str, Any]) - TwelveLabsDocumentEmbedder从字典反序列化组件run(documents: list[Document]) - dict[str, Any]嵌入文档列表run_async(documents: list[Document]) - dict[str, Any]异步嵌入文档列表批内并发。TwelveLabsTextEmbedderto_dict() - dict[str, Any]序列化组件为字典from_dict(data: dict[str, Any]) - TwelveLabsTextEmbedder从字典反序列化组件run(text: str) - dict[str, Any]嵌入单个字符串run_async(text: str) - dict[str, Any]异步嵌入单个字符串。这三个组件都完整实现 Haystack 组件的to_dict/from_dict序列化协议意味着你可以把包含它们的管道保存为 YAML/JSON 配置并在不同环境中重建参见 marshal/yaml.py 与 core/pipeline/base.py。常见问题与最佳实践API Key 管理始终优先使用TWELVELABS_API_KEY环境变量确需代码内传参时用Secret.from_token(...)包装避免明文密钥进入日志或版本库。视频来源限制仅支持公开可访问的直接视频 URL 或本地文件路径本地文件上传上限为 200 MB超大视频请先切片或改用直接 URL。单源失败不影响整批转换失败的单条来源会被跳过并记录 warning批量索引时可放心混入可能失效的 URL。生产环境关闭进度条TwelveLabsDocumentEmbedder(progress_barFalse)可避免海量文档嵌入时刷屏日志。批量异步提速大批量文档嵌入优先使用run_async借助batch_size批内并发显著缩短耗时。温度与 token 权衡视频分析需要事实准确性时降低temperature需要更长、更详尽的分析时调高max_tokens默认 16384 已很充裕。跨模态检索的基石务必让文档嵌入与查询嵌入使用同一个 Marengo 模型版本否则向量空间不一致余弦相似度比较将失去意义。延伸阅读组件速查本仓库的组件文档 twelvelabsvideoconverter.mdx、twelvelabsdocumentembedder.mdx、twelvelabstextembedder.mdx数据基类Document的定义见 document.py密钥封装Secret的from_token/from_env_var实现见 auth.py配套组件document_writer.py、embedding_retriever.py、InMemoryDocumentStore管道机制core/pipeline/base.py 与管道序列化支持 marshal/yaml.py。【免费下载链接】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），仅供参考