用 Strands Agents 在 AWS 上构建可对话的视频理解系统

用户问题
   │
   ▼
Strands Agent ─────► Amazon Bedrock：理解问题、规划调用、整理答案
   │
   ├───────────────► Amazon Transcribe：语音与对白
   │
   └───────────────► Amazon Rekognition：物体、人物、场景与时间点
                           │
                           ▼
                      Amazon S3 视频

python -m venv .venv
source .venv/bin/activate
pip install strands-agents boto3

export AWS_REGION=us-east-1
export VIDEO_BUCKET=my-video-bucket
export VIDEO_KEY=uploads/demo.mp4

import json
import os
import time
import urllib.request
import uuid

import boto3
from strands import Agent, tool

REGION = os.getenv("AWS_REGION", "us-east-1")
BUCKET = os.environ["VIDEO_BUCKET"]
KEY = os.environ["VIDEO_KEY"]

transcribe = boto3.client("transcribe", region_name=REGION)
rekognition = boto3.client("rekognition", region_name=REGION)


def wait(seconds: int = 5) -> None:
    time.sleep(seconds)


@tool
def transcribe_video() -> str:
    """Transcribe the speech in the configured S3 video and return its text."""
    job_name = f"video-agent-{uuid.uuid4().hex}"
    transcribe.start_transcription_job(
        TranscriptionJobName=job_name,
        Media={"MediaFileUri": f"s3://{BUCKET}/{KEY}"},
        IdentifyLanguage=True,
    )

    while True:
        response = transcribe.get_transcription_job(
            TranscriptionJobName=job_name
        )
        job = response["TranscriptionJob"]
        status = job["TranscriptionJobStatus"]

        if status == "COMPLETED":
            transcript_uri = job["Transcript"]["TranscriptFileUri"]
            with urllib.request.urlopen(transcript_uri, timeout=30) as result:
                payload = json.load(result)
            return payload["results"]["transcripts"][0]["transcript"]

        if status == "FAILED":
            return f"Transcription failed: {job.get('FailureReason', 'unknown')}"

        wait()


@tool
def detect_video_labels() -> str:
    """Detect objects and scenes in the configured video, including timestamps."""
    started = rekognition.start_label_detection(
        Video={"S3Object": {"Bucket": BUCKET, "Name": KEY}},
        MinConfidence=75,
    )
    job_id = started["JobId"]

    while True:
        page = rekognition.get_label_detection(
            JobId=job_id,
            MaxResults=1000,
            SortBy="TIMESTAMP",
        )
        status = page["JobStatus"]

        if status == "FAILED":
            return f"Label detection failed: {page.get('StatusMessage', 'unknown')}"
        if status == "SUCCEEDED":
            break
        wait()

    detections = []
    while True:
        for item in page.get("Labels", []):
            label = item["Label"]
            detections.append(
                {
                    "time_seconds": round(item["Timestamp"] / 1000, 2),
                    "label": label["Name"],
                    "confidence": round(label["Confidence"], 1),
                }
            )

        token = page.get("NextToken")
        if not token or len(detections) >= 300:
            break

        page = rekognition.get_label_detection(
            JobId=job_id,
            MaxResults=1000,
            SortBy="TIMESTAMP",
            NextToken=token,
        )

    return json.dumps(detections[:300], ensure_ascii=False)


SYSTEM_PROMPT = """
你是视频分析助手。根据问题只调用必要的工具：
- 涉及说话内容、主题、关键词时，调用 transcribe_video。
- 涉及画面中的物体、人物、场景或出现时间时，调用 detect_video_labels。
- 需要对齐画面与对白时，同时调用两个工具。
只能依据工具返回的证据回答。证据不足时明确说明，不要猜测。
涉及时间时使用 mm:ss 格式，并简要说明依据。
"""

agent = Agent(
    system_prompt=SYSTEM_PROMPT,
    tools=[transcribe_video, detect_video_labels],
)

question = input("请输入关于视频的问题：")
print(agent(question))

python video_agent.py

视频中什么时候出现了汽车？
演讲者主要讨论了什么？
画面出现电脑时，旁白正在说什么？

bucket + object_key + object_version + analysis_type + model_or_api_version

汽车约在 01:12 首次出现。
依据：Rekognition 标签 Car，置信度 96.4%。
相关旁白：转写片段 01:09–01:16。