TwelveLabs Pegasus 1.5 在 Amazon Bedrock 上线
Pegasus 1.5 from TwelveLabs is now available in Amazon Bedrock
TwelveLabs 的视频语言模型 Pegasus 1.5 已在 Amazon Bedrock 美国西部(俄勒冈)区域上线,可分析最长 2 小时的视频并返回带时间戳的结构化片段。新功能包括按用户定义条件生成元数据、结合参考图像进行多模态提示,以及无需预索引的按需处理;视频可通过 Amazon S3 URI 或 base64 字符串提交。
Pegasus 1.5 将长视频分析扩展为按自定义条件提取带时间戳的结构化片段,并支持参考图像提示;文中还给出了调用方式与输入限制。
2026 年 10 月 9 日,我们宣布 TwelveLabs’ Pegasus 1.5 在 Amazon Bedrock 上正式可用。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 中的文本回答;如果您提供了架构,则返回 JSON |
| 基于时间的元数据 | time_based_metadata |
responseFormat 中的片段定义 |
包含起始和结束时间以及您所请求字段的 segments 列表 |
您可以通过两种方式将视频提供给模型。S3 位置(s3Location)需要对象的 uri 和 bucketOwner 账户 ID,适用于长视频。Base64 字符串(base64String)会将视频嵌入请求正文,最大为 25 MB,适用于短片段。
若要按应用跟踪成本,请基于 Pegasus 1.5 模型创建应用推理配置文件,为其添加标签,并使用其 Amazon 资源名称(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 密钥进行身份验证。对于生产环境,建议使用通过 IAM 凭证生成的短期密钥,并使用 aws-bedrock-token-generator 包,因为该密钥会自动过期。持有者令牌身份验证还需要 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 中包含令牌计数,并在 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 |
否 | 要生成的最大令牌数,范围为 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,并将架构放在 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 会在 AWS/Bedrock 命名空间下,以其模型 ID 发布 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 今日起已在 Amazon Bedrock 的美国西部(俄勒冈)AWS 区域推出。请查看完整区域列表,了解未来更新;有关定价信息,请访问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 模型推理。他在构建机器学习平台和生产级推理基础设施方面拥有十多年的经验,曾创立机器学习平台团队,也曾大规模提供基础模型服务。
Shannon Hong
Shannon 是 TwelveLabs 的产品经理,负责 Pegasus 视频语言模型的开发。她是 Oxford Martin School AI Governance Initiative 的研究附属成员,并与 Open Future Foundation 和 Creative Commons 合作开展过人工智能民主治理研究。她的职业生涯始于 Scale AI 的产品经理一职,在那里她领导技术团队为自动驾驶汽车构建数据标注产品。Shannon 拥有牛津大学工商管理硕士学位,以及加州大学伯克利分校发展研究与数据科学文学学士学位。
Nate Stecz
Nate 是 AWS 的高级合作伙伴开发经理,帮助 AI 独立软件供应商制定并执行上市策略。Nate 在 AWS 工作了 10 年,专注于帮助 AI 模型提供商构建、营销和销售解决方案,涵盖 Amazon Bedrock 集成、AWS Marketplace 和一线团队赋能。闲暇时,Nate 喜欢户外活动。您可以在LinkedIn上联系或关注 Nate。
Narcisse Zekpa
Narcisse 是一位常驻波士顿的首席解决方案架构师。他通过 AWS Cloud 上创新且可扩展的解决方案,帮助美国东北部的客户加快业务转型。他热衷于运用高级分析和 AI 助力组织实现业务转型。Narcisse 不忙于构建解决方案时,喜欢与家人共度时光、旅行和公路赛车。
Suheel Farooq
Suheel 是 AWS 的首席解决方案架构师,专注于人工智能、机器学习和生成式 AI。他帮助基础模型提供商客户在 AWS 上设计、构建、现代化并扩展 AI/ML 和生成式 AI 工作负载——从将基础模型引入 Amazon Bedrock,到优化训练和推理,再到构建智能体 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