TwelveLabs Pegasus 1.5 现已在 Amazon 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。

前提条件

开始之前，请确保你已具备：

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 中携带令牌计数，并在 stopReason 中携带生成结束的原因。length 的 stopReason 表示答案在完成之前达到了 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"])

通用模式的请求字段：

对于短视频片段，你可以内联发送视频而不是从 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、一个值为 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 在 Amazon CloudWatch 的 AWS/Bedrock 命名空间下按其模型 ID 发布指标，包括 Invocations、OutputTokenCount、InvocationLatency 和 InvocationClientErrors。可以使用这些指标来跟踪调用量和延迟。当你需要按应用细分使用情况时，可将它们与应用推理配置文件搭配使用。

用例

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

• 新闻和广播档案：团队可以一次性定义编辑片段、话题和发言人切换，然后对多年来的素材进行编目，这些素材此前只能通过转录文本关键词进行搜索。
• 体育转播商和联盟：在完整的比赛 footage 中检测战术动作、得分事件和其他关键时刻，加快集锦制作速度。
• 流媒体应用：场景级元数据支持基于时刻的推荐、情境化广告投放以及片段级品牌安全审查。

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

输入要求

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 已于今日在 US West (Oregon) AWS 区域的 Amazon Bedrock 上推出。查看完整的区域列表以了解未来的更新，并访问 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 构建出什么。

关于作者