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。你可以通过 bedrock-runtime 端点配合 InvokeModel API 调用它,或者当你希望在响应生成的同时进行流式传输时使用 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 |
A 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)。
- AWS Identity and Access Management (IAM) 权限,用于
bedrock:InvokeModel和bedrock:InvokeModelWithResponseStream. - 在以下位置启用了 TwelveLabs Pegasus 1.5 的模型访问 Amazon Bedrock 控制台.
- 存储在 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 包生成的短期密钥,因为它会自动过期。持有者令牌(Bearer-token)身份验证还需要 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 |
否 | 要生成的最大 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下。答案会在 message 作为 JSON 字符串:
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 与早期版本差异最大的地方。你无需发送提示词,而是发送一个或多个 segment 定义,模型会返回视频中与每个定义相匹配的所有片段。
将 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 列表,当所有定义都被处理时该列表为空。
单个请求可以包含多个 segment 定义,例如一个用于编辑栏目,另一个用于屏幕上的品牌露出。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 引入了多模态提示,因此你可以在文本提示的同时附上参考图像。当你想询问模型关于特定物体、人物或场景,而展示比描述更容易时,这会非常有用。
例如,你可以提供一张产品图片,并让模型识别该产品在视频中出现的每个时刻。或者,你可以提供某个特定标志的参考图像,并让模型检测品牌露出及其时间戳。此功能提升了视频分析的精确度,并开启了诸如视觉品牌监测、产品植入分析和图像引导内容搜索等用例。
参考图像的附加方式因模式而异。在 general 模式下,将它们放在 promptV2.mediaSources,以及以下内容中的问题 promptV2.inputText。在分段模式(analysisMode: time_based_metadata)下,将它们附加到 mediaSources 内每个分段定义的 responseFormat.segmentDefinitions。这样,一个定义既可以匹配视觉参考,也可以匹配文本描述。每张图片需要一个 name、一个 media_type 为 image,以及恰好一个来源:要么是内联 base64String ,要么是一个 s3Location ,其中包含 uri 和 bucketOwner。单个请求最多可携带 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 模型卡片 了解功能支持、区域可用性和示例代码。
- 请通过 AWS re:Post for Amazon Bedrock 或您常用的 AWS Support 联系方式向我们发送反馈。
我们期待看到您使用 Amazon Bedrock 中的 TwelveLabs Pegasus 1.5 构建的作品。
