跳到正文
原文
AWS Machine Learning Blog· Ron Tal·· 4 小时前AI 评分52

TwelveLabs Pegasus 1.5 现已在 Amazon Bedrock 提供

Pegasus 1.5 from TwelveLabs is now available in Amazon Bedrock

AI 导读

TwelveLabs 的视频语言模型 Pegasus 1.5 已在 Amazon Bedrock 提供,可一次分析最长两小时视频并返回带起止时间的 JSON 片段。

正文 · AI 翻译

2026年10月9日,我们宣布 Amazon Bedrock 上的 TwelveLabs Pegasus 1.5 现已可用。Pegasus 1.5 是一款视频语言模型,可将长视频转化为结构化、带时间戳的元数据。你描述工作流中重要的时刻,模型便会以 JSON 形式返回每一个匹配片段。

视频是一类重要且复杂的非结构化数据,但其中大部分仍无法搜索,也未得到充分利用。从这些内容中提取结构化、可操作的洞察,以往通常需要耗费大量人力的人工审阅,或依赖零散的工具链。

在早期版本的 Pegasus 中,你可以将模型指向一段视频片段、提出问题并获得答案。如今,Pegasus 1.5 会分析整段视频,找出每一个符合用户定义条件的片段,并返回结构化、带时间戳的元数据。它还引入了视频分段、结合参考图像的多模态提示,以及即时处理。

借助 Amazon Bedrock 上的 Pegasus 1.5,你可以直接通过 Amazon Simple Storage Service (Amazon S3) URI 或 base64 字符串分析视频,无需预先建立索引,并且你的数据仍然位于 AWS 数据边界之内。

在本文中,我们将介绍 Pegasus 1.5 的新功能、该模型在 Amazon Bedrock 上的封装方式,以及如何发送你的第一个请求。随后,我们将逐步讲解视频分段、结构化 JSON 输出和多模态提示。

Pegasus 1.5 的新功能

与先前版本相比,Pegasus 1.5 有以下变化:

  • 基于时间的元数据生成。 你定义对工作流重要的片段、事件或领域特定线索,例如编辑片段、体育比赛回合、发言人切换或品牌出镜。Pegasus 1.5 会以 JSON 形式返回每一个匹配片段,并填入开始时间、结束时间以及所请求的字段。
  • 多模态提示。 片段定义可以将文本描述与视觉参考相结合。这样便能定位特定产品、徽标或人物,以及他们在画面中正在做什么。
  • 即时处理。 无需摄取管道或预先索引步骤。视频直接从 Amazon S3 发送,结果在响应中返回。
  • 扩展的处理能力。 Pegasus 1.5 可在单次请求中分析最长两小时的视频,而 Pegasus 1.2 为最长一小时。

Pegasus 1.5 保留并增强了使 Pegasus 1.2 成为强大视频理解工具的核心能力,包括:

  • 视频转文本生成:根据视频内容创建详细的文本描述。
  • 细粒度视觉理解:分析视频帧中的物体、屏幕文字和数字内容。
  • 时间定位:准确识别视频中特定事件的时间戳。
  • 多模态理解:结合视觉、音频和文本信息进行全面分析。
  • 多语言支持:完整支持英语,并部分支持阿拉伯语、中文、法语、德语、意大利语、日语、韩语、葡萄牙语、俄语、西班牙语、泰语和越南语。

Pegasus 1.5 在 Amazon Bedrock 上的工作原理

Pegasus 1.5 接收一段视频,以及文本提示或一组片段定义,并返回文本或 JSON。你可以使用 InvokeModel API 在 bedrock-runtime 端点上调用它;如果希望在生成过程中以流式方式返回响应,则使用 InvokeModelWithResponseStream。

设置 值
模型 ID twelvelabs.pegasus-1-5-v1:0
AWS 区域 美国西部(俄勒冈),us-west-2
推理 按需
API InvokeModel, InvokeModelWithResponseStream

每个请求以两种分析模式之一运行,通过 analysisMode 字段选择:

分析模式 analysisMode 你发送 你获得
通用 general(默认) inputPrompt 中的文本提示 message 中的文本回答,或在你提供 schema 时返回 JSON
基于时间的元数据 time_based_metadata responseFormat 中的片段定义 包含开始和结束时间以及你所请求字段的 segments 列表

你可以通过两种方式让模型指向视频。S3 位置(s3Location)需要对象的 uri 和 bucketOwner 账户 ID,适合长视频。base64 字符串(base64String)将视频嵌入请求正文,最大 25 MB,适合短片段。

要按应用程序跟踪成本,请基于 Pegasus 1.5 模型创建应用程序推理配置文件,为其添加标签,并使用其 Amazon Resource Name (ARN) 代替模型 ID 进行调用。

Pegasus 1.5 入门

以下部分概述如何开始使用 Pegasus 1.5。

先决条件

开始之前,请确保你具备:

  1. 一个具有 Amazon Bedrock 访问权限的有效 AWS 账户。
  2. Python 3.10 或更高版本,并已安装适用于 Python 的 AWS SDK(Boto3)。
  3. 针对 bedrock:InvokeModel 和 bedrock:InvokeModelWithResponseStream 的 AWS Identity and Access Management (IAM) 权限。
  4. 在 Amazon Bedrock 控制台中为 TwelveLabs Pegasus 1.5 启用模型访问权限。
  5. 存储在 Amazon S3 存储桶中的视频,或可以内联发送的短片段。

以下策略授予对 Pegasus 1.5 的两项调用操作权限。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "bedrock:InvokeModel",
                "bedrock:InvokeModelWithResponseStream"
            ],
            "Resource": [
                "arn:aws:bedrock:us-west-2::foundation-model/twelvelabs.pegasus-1-5-v1:0"
            ]
        }
    ]
}

Boto3 使用你常用的 AWS 凭证对请求进行签名。你也可以通过在 AWS_BEARER_TOKEN_BEDROCK 环境变量中设置 Amazon Bedrock API 密钥来进行身份验证。对于生产环境,建议优先使用通过 aws-bedrock-token-generator 包从你的 IAM 凭证生成的短期密钥,因为它会自动过期。Bearer 令牌身份验证还需要 bedrock:CallWithBearerToken 权限。

发送你的第一个请求

安装 Boto3:

pip install boto3

以下示例以通用模式分析存储在 Amazon S3 中的视频。此示例使用全球跨区域推理(CRIS)端点。若要改为指向美国地理端点,请将端点 URL 中的“global.”替换为“us.”。

import boto3
import json
from botocore.config import Config

AWS_REGION = "us-west-2"
MODEL_ID = "global.twelvelabs.pegasus-1-5-v1:0"

bedrock_runtime = boto3.client(
    service_name="bedrock-runtime",
    region_name=AWS_REGION,
    config=Config(
        # keep a longer read_timeout for lengthier invocations such as video segmentation
        read_timeout=900,
        retries={"max_attempts": 0},
        connect_timeout=10,
    ),
)

# Analyze a video stored in Amazon S3
payload = {
    "inputPrompt": "Summarize this video and identify the key topics discussed.",
    "mediaSource": {
        "s3Location": {
            "uri": "s3://amzn-s3-demo-pegasus-15/your-video.mp4",
            "bucketOwner": "your-account-id"
        }
    },
    "temperature": 0.2,
    "maxOutputTokens": 4096,
    "includeMetering": True
}

response = bedrock_runtime.invoke_model(
    modelId=MODEL_ID,
    body=json.dumps(payload)
)
result = json.loads(response["body"].read())
print(result["message"])

响应在 message 中包含回答,在 input_tokens 和 output_tokens 中包含 token 计数,并在 stopReason 中包含生成结束的原因。stopReason 为 length 表示回答在完成之前已达到 maxOutputTokens,因此在依赖输出之前请先检查:

if result["stopReason"] == "length":
    print("The answer was cut off. Raise maxOutputTokens and try again.")
print("Tokens in:", result["input_tokens"], "out:", result["output_tokens"])

通用模式的请求字段:

字段 是否必需 说明
inputPrompt 是 关于视频的问题或指令
mediaSource.s3Location.uri 需要一个视频源 视频的 S3 URI
mediaSource.s3Location.bucketOwner 与 uri 一起使用 拥有该存储桶的 AWS 账户 ID
mediaSource.base64String 需要一个视频源 Base64 编码的视频,最大 25 MB
temperature 否 输出的随机性,范围为 0 到 1。默认值为 0.2
maxOutputTokens 否 要生成的最大 token 数,范围为 512 到 98,304
responseFormat 否 回答必须遵循的 JSON Schema
analysisMode 否 general(默认)或 time_based_metadata

对于短片段,你可以内联发送视频,而不是从 S3 发送:

import base64

with open("your-video.mp4", "rb") as f:
    video_b64 = base64.b64encode(f.read()).decode("utf-8")

payload["mediaSource"] = {"base64String": video_b64}
response = bedrock_runtime.invoke_model(modelId=MODEL_ID, body=json.dumps(payload))
print(json.loads(response["body"].read())["message"])

请注意,前面的示例使用了可选的 includeMetering 属性。includeMetering: True 会使 Pegasus 1.5 向非流式响应添加两个计费用量字段:outputTokenCount 和 inputVideoSecond(所分析视频的时长,以秒为单位)。

流式传输响应

对于交互式应用程序,InvokeModelWithResponseStream 会在生成过程中分块返回答案。每个块在 message 中包含下一段文本:

response = bedrock_runtime.invoke_model_with_response_stream(
    modelId=MODEL_ID,
    body=json.dumps(payload)
)
for event in response["body"]:
    chunk = json.loads(event["chunk"]["bytes"])
    print(chunk.get("message", ""), end="", flush=True)
print()

返回结构化 JSON

在通用模式下,你可以将回答约束为 JSON Schema,以便下游代码直接解析。将 responseFormat.type 设置为 json_schema,并将 schema 放在 jsonSchema 下。回答会以 JSON 字符串的形式出现在 message 中:

payload = {
    "inputPrompt": "Analyze this video and return a structured summary.",
    "mediaSource": {
        "s3Location": {
            "uri": "s3://amzn-s3-demo-pegasus-15/your-video.mp4",
            "bucketOwner": "your-account-id"
        }
    },
    "temperature": 0,
    "maxOutputTokens": 2048,
    "includeMetering": True,
    "responseFormat": {
        "type": "json_schema",
        "jsonSchema": {
            "type": "object",
            "properties": {
                "summary": {"type": "string"},
                "key_scenes": {"type": "array", "items": {"type": "string"}},
                "setting": {"type": "string"}
            },
            "required": ["summary", "key_scenes", "setting"]
        }
    }
}
response = bedrock_runtime.invoke_model(modelId=MODEL_ID, body=json.dumps(payload))
summary = json.loads(json.loads(response["body"].read())["message"])
print(summary["summary"])
for scene in summary["key_scenes"]:
    print(" -", scene)

视频分段:将视频转换为结构化数据

基于时间的元数据生成是 Pegasus 1.5 与早期版本差异最大的地方。你不再发送提示词,而是发送一个或多个片段定义,模型会返回视频中与每个定义相匹配的所有片段。

将 analysisMode 设为 time_based_metadata,将 responseFormat.type 设为 segment_definitions。这些定义取代了 inputPrompt,因此请省略该字段。每个定义包含:

  • id:定义的名称,会在其匹配的每个片段上返回。
  • description:匹配片段的样子。
  • fields:为每个片段提取的值,每个值都包含一个 name、一个 type(string、boolean、number、integer、array 或 timestamp)以及一个 description。

以下示例会识别新闻广播中的每个编辑段落,并为每个段落返回标题、摘要和关键主题:

payload = {
    "analysisMode": "time_based_metadata",
    "mediaSource": {
        "s3Location": {
            "uri": "s3://amzn-s3-demo-pegasus-15/news-broadcast.mp4",
            "bucketOwner": "your-account-id"
        }
    },
    "responseFormat": {
        "type": "segment_definitions",
        "segmentTimeFormat": "hh:mm:ss",
        "segmentDefinitions": [
            {
                "id": "editorial_section",
                "description": "A distinct editorial section of the broadcast, "
                "such as a new story, topic, or segment of the program.",
                "fields": [
                    {"name": "segment_title", "type": "string",
                     "description": "A descriptive title"},
                    {"name": "summary", "type": "string",
                     "description": "A brief summary of the segment content"},
                    {"name": "key_topics", "type": "array", "items": {"type": "string"},
                     "description": "Main topics discussed"}
                ]
            }
        ]
    },
    "temperature": 0,
    "maxOutputTokens": 32768,
    "includeMetering": True
}
response = bedrock_runtime.invoke_model(
    modelId=MODEL_ID,
    contentType="application/json",
    accept="application/json",
    body=json.dumps(payload)
)
result = json.loads(response["body"].read())
for segment in result["segments"]:
    print(segment["start_sec"], "-", segment["end_sec"], segment["segment_title"])
    print(" ", segment["summary"])
    print(" topics:", ", ".join(segment["key_topics"]))

将前面的示例代码用于体育片段进行测试时,可能会产生如下示例所示的输出。

{
    "segments": [
        {
            "start_sec": "00:04:24",
            "end_sec": "00:05:22",
            "segment_title": "Substitutions and Added Time",
            "summary": "The referee announces three minutes of added time, and both teams make substitutions in preparation for a potential penalty shootout.",
            "key_topics": ["Substitutions", "Added time", "Penalty shootout preparation", "Tactical changes"]
        },
        {
            "start_sec": "00:05:22",
            "end_sec": "00:08:57",
            "segment_title": "Last Gasp Attempts and Final Whistle",
            "summary": "The final minutes of the match see intense action from both sides, with several close calls before the referee blows the final whistle, ending the game in a 3-3 draw.",
            "key_topics": ["Final whistle", "Draw"]
        },
        {
            "start_sec": "00:08:57",
            "end_sec": "00:09:29",
            "segment_title": "Match Conclusion and Penalty Shootout Setup",
            "summary": "The match concludes with a 3-3 draw, leading to a penalty shootout. Players from both teams shake hands and prepare for the decisive moments.",
            "key_topics": ["Match conclusion", "Penalty shootout", "Player handshakes", "Final"]
        }
    ],
    "definition_errors": [],
    "input_tokens": 249408,
    "output_tokens": 802,
    "stopReason": "stop",
    "finish_reason": "stop"
}

每个片段都带有 start_sec、end_sec 以及你所定义的 fields。segmentTimeFormat 控制时间的书写方式:seconds 返回可用于算术运算的数字,而 hh:mm:ss 和 hh:mm:ss.fff 则以字符串形式返回可读的时间戳。响应中还包含一个 definition_errors 列表,当每个定义都已处理完毕时,该列表为空。

单个请求可以包含多个片段定义,例如一个用于编辑段落,另一个用于画面中的品牌出现。Pegasus 1.5 会在一个 segments 列表中返回所有这些定义的匹配结果,并用 _definition_id 标记每个片段,以便你区分它们。以下示例在前一个请求中添加了第二个定义,并对结果进行分组:

from collections import defaultdict

payload["responseFormat"]["segmentDefinitions"].append({
    "id": "brand_appearance",
    "description": "A shot where a company logo or branded product "
    "is clearly visible on screen.",
    "fields": [
        {"name": "brand", "type": "string", "description": "The brand shown"},
        {"name": "is_prominent", "type": "boolean",
         "description": "Whether the brand is the main focus of the shot"}
    ]
})
response = bedrock_runtime.invoke_model(modelId=MODEL_ID, body=json.dumps(payload))
result = json.loads(response["body"].read())
by_definition = defaultdict(list)
for segment in result["segments"]:
    by_definition[segment["_definition_id"]].append(segment)
for definition_id, segments in by_definition.items():
    print(definition_id, len(segments), "segments")

由于模式只需定义一次,相同的定义便可应用于整个视频库,为每项资产生成一致的元数据。

多模态提示:为查询添加视觉上下文

Pegasus 1.5 引入了多模态提示,因此你可以在文本提示旁附上参考图像。当你想向模型询问那些展示起来比描述更容易的特定物体、人物或场景时,这一功能非常有用。

例如,你可以提供一张产品图像,并要求模型识别该产品在视频中出现的每一时刻。你也可以提供特定徽标的参考图像,并要求模型检测品牌出现及其时间戳。此功能提高了视频分析的精度,并开辟了视觉品牌监测、产品植入分析和图像引导内容搜索等用例。

参考图像的附加方式因模式而异。在通用模式下,将它们放在 promptV2.mediaSources 中,与 promptV2.inputText 中的问题并列。在分段模式(analysisMode: time_based_metadata)下,将它们附加到 responseFormat.segmentDefinitions 内每个片段定义的 mediaSources 上。这样,定义既可以匹配视觉参考,也可以匹配文本描述。每张图像需要一个 name、一个值为 image 的 media_type,以及恰好一个来源:内联的 base64String,或带有 uri 和 bucketOwner 的 s3Location。单个请求最多可携带 30 张参考图像。

通用模式请求会向模型询问参考图像中所示的物体。以下示例载荷和响应提供了通用模式下的示例。

payload = {
    "analysisMode": "general",
    "mediaSource": {"s3Location": {"uri": "s3://...", "bucketOwner": "..."}},
    "promptV2": {
        "inputText": "Does the object in the reference image appear in this video?",
        "mediaSources": [
            {"name": "ref1", "media_type": "image", "base64String": "..."}
        ]
    }
    …
}
{
    "input_tokens": 144,
    "output_tokens": 11,
    "message": "Yes, the green canoe appears in the video.",
    "stopReason": "stop"
}

在分段模式下,参考图像位于定义内部,因此 Pegasus 仅返回该产品出现的片段。以下示例载荷和响应提供了分段模式下的示例。

payload = {
    "analysisMode": "time_based_metadata",
    "mediaSource": {"s3Location": {"uri": "s3://...", "bucketOwner": "..."}},
    "responseFormat": {
        "type": "segment_definitions",
        "segmentTimeFormat": "hh:mm:ss",
        "segmentDefinitions": [
            {
                "id": "product_appearance",
                "description": "A shot showing the product in the reference image",
                "fields": [
                    {"name": "what", "type": "string", "description": "what is on screen"}
                ],
                "mediaSources": [
                    {"name": "ref1", "media_type": "image",
                     "s3Location": {"uri": "s3://...", "bucketOwner": "..."}}
                ]
            }
        ]
    }
}
{
    "segments": [
        {"start_sec": "00:00:00", "end_sec": "00:00:02",
         "what": "A family of four in a green canoe, paddling on a calm river surrounded by towering red rock formations."},
        {"start_sec": "00:00:09", "end_sec": "00:00:10",
         "what": "The family back in the canoe, paddling away from the shore, with the children smiling and enjoying the serene surroundings."}
    ],
    "definition_errors": [],
    "input_tokens": 4498,
    "output_tokens": 292,
    "stopReason": "stop",
    "finish_reason": "stop"
}

监控使用情况

Pegasus 1.5 在其模型 ID 下的 AWS/Bedrock 命名空间中发布 Amazon CloudWatch 指标,包括 Invocations、OutputTokenCount、InvocationLatency 和 InvocationClientErrors。使用它们来跟踪调用量和延迟。当你需要按应用程序细分用量时,可将它们与应用程序推理配置文件配合使用。

用例

Pegasus 1.5 专为管理大量长视频的组织而设计:

  • 新闻与广播档案:团队只需一次性定义编辑片段、主题和发言人切换,即可对多年的影像资料进行编目,而这些资料此前只能通过文字稿关键词检索。
  • 体育转播机构与联赛:在完整比赛录像中检测比赛回合、得分事件和其他关键时刻,加快集锦制作。
  • 流媒体应用:场景级元数据支持基于时刻的推荐、情境化广告投放以及片段级品牌安全审核。

在每种场景中,人员仍是工作流程的一部分。他们的工作从手工查找并标记每个时刻,转变为审核和完善模型生成的元数据。

输入要求

Pegasus 1.5 支持以下视频规格:

  • 时长:最长 2 小时。
  • 文件大小:来自 Amazon S3 时最大 2 GB,或以 base64 字符串形式最大 25 MB。
  • 分辨率:360×360 至 5184×2160。
  • 宽高比:介于 1:1 与 2.4:1 之间,或其倒数。
  • 格式:FFmpeg 支持的格式。

现已推出

TwelveLabs Pegasus 1.5 现已在美国西部(俄勒冈)AWS 区域的 Amazon Bedrock 上推出。请查看完整区域列表以了解后续更新,并访问Amazon Bedrock 定价页面了解定价信息。Pegasus 1.5 通过 AWS Marketplace 计费,因此其费用会显示在您的 AWS 账单中,并在 AWS Cost Explorer 中归入 TwelveLabs 而非 Amazon Bedrock。

开始使用:

我们期待看到您使用 Amazon Bedrock 中的 TwelveLabs Pegasus 1.5 构建的成果。


关于作者

Ron Tal

Ron Tal

Ron 是 TwelveLabs 的机器学习工程师,负责通过包括 Amazon Bedrock 在内的第三方平台部署 TwelveLabs 模型的推理服务。他在构建机器学习平台和生产级推理基础设施方面已有十余年经验,从组建机器学习平台团队到大规模提供基础模型服务均有涉猎。

Shannon Hong

Shannon Hong

Shannon 是 TwelveLabs 的产品经理,负责视频语言模型 Pegasus 的开发。她是 Oxford Martin School AI Governance Initiative 的研究附属人员,并曾与 Open Future Foundation 和 Creative Commons 合作开展人工智能民主治理研究。她的职业生涯始于在 Scale AI 担任产品经理,曾带领技术团队为自动驾驶汽车打造数据标注产品。Shannon 拥有牛津大学 M.B.A. 学位,以及加州大学伯克利分校发展研究与数据科学 B.A. 学位。

Nate Stecz

Nate Stecz

Nate 是 AWS 的高级合作伙伴发展经理,帮助 AI ISV 制定并执行上市策略。Nate 在 AWS 工作已有 10 年,专注于帮助 AI 模型提供商构建、营销和销售其解决方案,涵盖 Amazon Bedrock 集成、AWS Marketplace 以及一线赋能。闲暇时,Nate 喜欢户外活动。您可以通过 LinkedIn 与 Nate 联系或关注他。

Narcisse Zekpa

Narcisse Zekpa

Narcisse 是一位常驻波士顿的首席解决方案架构师。他帮助美国东北部的客户通过 AWS 云上创新且可扩展的解决方案,加速其业务转型。他热衷于运用高级分析和 AI,助力组织实现业务转型。当 Narcisse 不在构建时,他喜欢与家人共度时光、旅行以及公路赛车。

Suheel Farooq

Suheel Farooq

Suheel 是 AWS 的首席解决方案架构师,专注于人工智能、机器学习和生成式 AI。他帮助基础模型提供商客户在 AWS 上设计、构建、现代化并扩展其 AI/ML 和生成式 AI 工作负载——从将基础模型引入 Amazon Bedrock,到优化训练与推理以及构建 Agentic AI 解决方案。他的经验涵盖 AWS 的 AI/ML 和生成式 AI 产品组合,尤其是 Amazon Bedrock、Amazon Bedrock AgentCore 和 Amazon SageMaker AI。闲暇时,Suheel 喜欢健身和徒步。

Badri Raghunathan Srikumar

Badri Raghunathan Srikumar

Badri 是 Amazon SageMaker AI 的软件开发工程师。他通过在 Amazon SageMaker 上为其构建可扩展、安全且可观测的托管基础设施,帮助将领先模型提供商的基础模型带给客户。在 AWS 之外,他深深热衷于机器人技术,尤其是自动驾驶汽车。不工作时,Badri 喜欢徒步、游泳和逛咖啡馆。

来源:AWS Machine Learning Blog · aws.amazon.com