Bedrock新增长视频结构化解析模型

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。

每个请求都在两种分析模式之一中运行，通过 analysisMode 字段选择：

你可以通过两种方式向模型提供视频。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 或更高版本，并已安装 AWS SDK for Python（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 权限。

发送第一个请求

pip install boto3

以下示例在 general 模式下分析存储于 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"])

general 模式的请求字段：

对于短视频片段，你可以直接在请求中内联发送视频，而不必从 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

在 general 模式下，你可以使用 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：需要从每个片段中提取的值，每个字段都包含名称、类型（string、boolean、number、integer、array 或 timestamp）和描述。

下面的示例识别新闻播报中的每个内容板块，并为每个板块返回标题、摘要和关键主题：

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 以及你定义的字段。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 现已在美国西部（俄勒冈）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 构建的应用。