TwelveLabs Pegasus 1.5 上线 Bedrock 支持视频结构化检索

{
    "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"
            ]
        }
    ]
}

pip install boto3

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
}

响应在 `message` 字段中携带答案，在 `input_tokens` 和 `output_tokens` 字段中携带 token 计数，并在 `stopReason` 字段中说明生成结束的原因。如果 `stopReason` 为 `length`，则表示答案在未完成之前达到了 `maxOutputTokens` 限制，因此在依赖输出结果之前请检查此字段：

```python
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"])

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"])

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()

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)

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": "换人与补时",
            "summary": "裁判宣布补时三分钟，双方球队进行换人调整，为可能到来的点球大战做准备。",
            "key_topics": ["换人", "补时", "点球大战准备", "战术调整"]
        },
        {
            "start_sec": "00:05:22",
            "end_sec": "00:08:57",
            "segment_title": "最后时刻的尝试与终场哨声",
            "summary": "比赛最后几分钟双方展开激烈对抗，多次出现险些破门的机会，最终裁判吹响终场哨，比赛以 3-3 平局收场。",
            "key_topics": ["终场哨", "平局"]
        },
        {
            "start_sec": "00:08:57",
            "end_sec": "00:09:29",
            "segment_title": "比赛结束与点球大战准备",
            "summary": "比赛以 3-3 平局告终，进入点球大战环节。双方球员握手致意，并为决定胜负的关键时刻做好准备。",
            "key_topics": ["比赛结束", "点球大战", "球员握手", "决赛"]
        }
    ],
    "definition_errors": [],
    "input_tokens": 249408,
    "output_tokens": 802,
    "stopReason": "stop",
    "finish_reason": "stop"
}

from collections import defaultdict

由于 schema 只需定义一次，相同的定义即可应用于整个视频库，从而为每个资产生成一致的元数据。

### 多模态提示：为查询添加视觉上下文

Pegasus 1.5 引入了多模态提示功能，允许你在文本提示中附带参考图像。当你希望询问模型关于特定物体、人物或场景的问题时，这非常有用，因为展示往往比描述更直观。

例如，你可以提供一张产品图片，要求模型识别视频中该产品出现的每一个时刻。或者，你可以提供特定 Logo 的参考图像，要求模型检测品牌露出及其时间戳。这一能力提升了视频分析的精度，并开启了诸如视觉品牌监控、植入广告分析以及基于图像的引导式内容搜索等应用场景。

参考图像的附加方式因模式而异。在通用模式下，将它们放在 `promptV2.mediaSources` 中，与 `promptV2.inputText` 中的问题并列。在分段模式（`analysisMode: time_based_metadata`）下，将它们附加到 `responseFormat.segmentDefinitions` 中每个分段定义的 `mediaSources` 上。这样，一个定义既可以匹配视觉参考，也可以匹配文本描述。每张图像需要一个 `name`、一个值为 `image` 的 `media_type`，以及恰好一个来源：要么是内联的 `base64String`，要么是包含 `uri` 和 `bucketOwner` 的 `s3Location`。单个请求最多可携带 30 张参考图像。

通用模式的请求会就参考图像中显示的物体向模型提问。以下示例 payload 和响应展示了通用模式下的用法。

```python
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"
}

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"
}