TwelveLabs Pegasus 1.5 上线 Amazon Bedrock
Pegasus 1.5 from TwelveLabs is now available in Amazon Bedrock
TwelveLabs 的视频语言模型 Pegasus 1.5 已在 Amazon Bedrock 上线,可将长视频转为带时间戳的结构化 JSON 元数据。新版本支持按用户定义的片段规则返回所有匹配区间,新增视频分段、参考图多模态提示和免预索引的即时处理,单次请求可分析最长 2 小时视频,较 Pegasus 1.2 的 1 小时提升。
原文给出 Pegasus 1.5 在 Bedrock 上的调用方式与分段输出示例,可据此评估长视频结构化处理流程。
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。
先决条件
开始之前,请确保你具备:
- 一个具有 Amazon Bedrock 访问权限的有效 AWS 账户。
- Python 3.10 或更高版本,并安装了适用于 Python 的 AWS SDK(Boto3)。
- 针对
bedrock:InvokeModel和bedrock:InvokeModelWithResponseStream的 AWS Identity and Access Management(IAM)权限。 - 在 Amazon Bedrock 控制台中为 TwelveLabs Pegasus 1.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")由于 schema 只定义一次,相同的定义可以应用于整个视频库,为每个资产生成一致的元数据。
多模态提示:为查询添加视觉上下文
Pegasus 1.5 引入了多模态提示,因此你可以在文本提示词旁附带参考图像。当你想要询问模型关于某些特定物体、人物或场景时,这非常有用,因为这些内容展示出来比描述更容易。
例如,你可以提供一张产品图片,并要求模型识别该产品在视频中出现的每一个时刻。或者你可以提供特定徽标的参考图像,并要求模型检测品牌露出及其时间戳。此功能提升了视频分析的精度,并开辟了视觉品牌监测、产品植入分析和图像引导内容搜索等用例。
参考图像的附加方式因模式而异。在通用模式下,将它们放入 promptV2.mediaSources,与 promptV2.inputText 中的问题放在一起。在分割模式(analysisMode: time_based_metadata)下,将它们附加到 responseFormat.segmentDefinitions 内每个片段定义的 mediaSources 上。这样定义就可以同时匹配视觉参考和文本描述。每张图像需要一个 name、一个 media_type(值为 image),以及恰好一个来源:内联的 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 模型卡,了解功能支持、区域可用性和示例代码。
- 请通过 Amazon Bedrock 的 AWS re:Post 或您常用的 AWS Support 联系方式向我们发送反馈。
我们非常期待看到您使用 Amazon Bedrock 中的 TwelveLabs Pegasus 1.5 构建出什么。
关于作者
Ron Tal
Ron 是 TwelveLabs 的机器学习工程师,负责领导通过第三方平台(包括 Amazon Bedrock)部署 TwelveLabs 模型的推理工作。他拥有十多年构建机器学习平台和生产推理基础设施的经验,从创立 ML 平台团队到大规模提供基础模型服务。
Shannon Hong
Shannon 是 TwelveLabs 的产品经理,负责领导视频语言模型 Pegasus 的开发。她是牛津马丁学院 AI 治理倡议的研究附属成员,并曾与 Open Future Foundation 和 Creative Commons 合作开展 AI 民主治理研究。她的职业生涯始于 Scale AI 的产品经理,在那里她领导技术团队为自动驾驶汽车构建数据标注产品。Shannon 拥有牛津大学工商管理硕士学位,以及加州大学伯克利分校发展研究和数据科学学士学位。
Nate Stecz
Nate 是 AWS 的高级合作伙伴发展经理,帮助 AI 独立软件供应商构建和执行市场推广策略。Nate 在 AWS 工作 10 年,专注于帮助 AI 模型提供商构建、营销和销售其解决方案——涵盖 Amazon Bedrock 集成、AWS Marketplace 和现场赋能。业余时间,Nate 喜欢户外活动。您可以在 LinkedIn 上联系或关注 Nate。
Narcisse Zekpa
Narcisse 是一位常驻波士顿的首席解决方案架构师。他帮助美国东北部的客户通过 AWS 云上创新且可扩展的解决方案加速业务转型。他热衷于帮助组织利用高级分析和 AI 实现业务转型。工作之余,Narcisse 喜欢与家人共度时光、旅行和参加公路赛车。
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 是 Amazon SageMaker AI 的软件开发工程师。他通过在 Amazon SageMaker 上构建可扩展、安全且可观测的托管基础设施,帮助将领先模型提供商的基础模型带给客户。在 AWS 之外,他对机器人技术,尤其是自动驾驶汽车充满热情。工作之余,Badri 喜欢徒步、游泳和逛咖啡馆。
来源:AWS Machine Learning Blog · aws.amazon.com