클라우드 환경에서의 비디오 인텔리전스 구현: TwelveLabs로 시작하는 AI 영상 분석 3부 – Strands Agent를 활용한 Agentic video engine구현

from strands import Agent

agent = Agent()
response = agent("Amazon Bedrock에 대해 알려줘")

pip install --upgrade boto3 botocore strands-agents

import boto3, json, time, uuid, hashlib  
from botocore.config import Config
from strands import Agent, tool  
from strands.models import BedrockModel  
from typing import Dict, Any  

REGION = 'us-east-1'  
S3_VECTORS_BUCKET = 'twl-marengo-3'      # S3 Vectors 버킷, 원하시는대로 변경 필요  
S3_VECTORS_INDEX = 'marengo3'             # 벡터 인덱스, 원하시는대로 변경 필요  
DYNAMODB_TABLE = 'strands-video-tasks'    # 작업 관리 테이블, 원하시는대로 변경 필요 
MARENGO_MODEL_ID = 'twelvelabs.marengo-embed-3-0-v1:0'   # 임베딩 모델  
PEGASUS_MODEL_ID = 'us.twelvelabs.pegasus-1-2-v1:0'       # 요약 모델  
ACCOUNT_ID = boto3.client('sts').get_caller_identity()['Account']  

# AWS 서비스 클라이언트  
# signature_version='v4'를 지정하여 StartAsyncInvoke에서 SigV4 서명 사용을 보장합니다.
# Bedrock API Key(Bearer Token) 인증 환경에서는 S3 자격증명이 전달되지 않아
# 비동기 호출이 실패할 수 있으므로, 명시적으로 SigV4를 지정합니다.
bedrock_runtime = boto3.client(
    'bedrock-runtime', region_name=REGION,
    config=Config(signature_version='v4')
)
s3 = boto3.client('s3', region_name=REGION)  
s3vectors = boto3.client('s3vectors', region_name=REGION)  
dynamodb = boto3.resource('dynamodb', region_name=REGION)  
task_table = dynamodb.Table(DYNAMODB_TABLE)  
transcribe = boto3.client('transcribe', region_name=REGION)  

# Strands에서 사용할 LLM 모델  
claude_model = BedrockModel(  
    model_id='us.anthropic.claude-sonnet-4-20250514-v1:0',  
    region_name=REGION  
)

# S3 Vectors 버킷 생성
try:
    s3vectors.create_vector_bucket(vectorBucketName=S3_VECTORS_BUCKET)
    print(f'버킷 생성 완료: {S3_VECTORS_BUCKET}')
except s3vectors.exceptions.ConflictException:
    print(f'기존 버킷 사용: {S3_VECTORS_BUCKET}')

# 벡터 인덱스 생성
try:
    s3vectors.create_index(
        vectorBucketName=S3_VECTORS_BUCKET,
        indexName=S3_VECTORS_INDEX,
        dataType='float32',
        dimension=EMBEDDING_DIMENSION,
        distanceMetric='cosine'
    )
    print(f'인덱스 생성 완료: {S3_VECTORS_INDEX}')
except s3vectors.exceptions.ConflictException:
    print(f'기존 인덱스 사용: {S3_VECTORS_INDEX}')

# DynamoDB Table 생성
try:
    table = dynamodb.create_table(
        TableName=DYNAMODB_TABLE,
        KeySchema=[{'AttributeName': 'task_id', 'KeyType': 'HASH'}],
        AttributeDefinitions=[{'AttributeName': 'task_id', 'AttributeType': 'S'}],
        BillingMode='PAY_PER_REQUEST'
    )
    table.wait_until_exists()
    print(f'테이블 생성 완료: {DYNAMODB_TABLE}')
except dynamodb.meta.client.exceptions.ResourceInUseException:
    table = dynamodb.Table(DYNAMODB_TABLE)
    print(f'기존 테이블 사용: {DYNAMODB_TABLE}')

@tool
def create_video_embedding(s3_uri: str) -> Dict[str, Any]:
    '''비디오 임베딩 생성 (완료까지 대기 후 S3 Vectors 저장).'''
    task_id = str(uuid.uuid4())[:8]
    bucket = s3_uri.split('/')[2]
    key = '/'.join(s3_uri.split('/')[3:])

    # DynamoDB에 작업 기록
    task_table.put_item(Item={
        'task_id': task_id, 's3_uri': s3_uri,
        's3_bucket': bucket, 's3_key': key,
        'status': 'processing', 'created_at': int(time.time())
    })

    # Marengo 임베딩 비동기 호출
    response = bedrock_runtime.start_async_invoke(
        modelId=MARENGO_MODEL_ID,
        modelInput={
            'inputType': 'video',
            'video': {
                'mediaSource': {
                    's3Location': {'uri': s3_uri, 'bucketOwner': ACCOUNT_ID}
                },
                'embeddingOption': ['visual', 'audio'],
                'embeddingScope': ['clip'],
                'segmentation': {'method': 'fixed', 'fixed': {'durationSec': 6}}
            }
        },
        outputDataConfig={
            's3OutputDataConfig': {'s3Uri': f's3://{bucket}/embeddings/{task_id}/'}
        }
    )
    invocation_arn = response['invocationArn']

    # 완료까지 대기
    for _ in range(60):
        status = bedrock_runtime.get_async_invoke(
            invocationArn=invocation_arn
        )['status']
        if status == 'Completed':
            break
        time.sleep(10)

    # 결과 파싱 및 S3 Vectors에 저장
    output_uri = bedrock_runtime.get_async_invoke(
        invocationArn=invocation_arn
    )['outputDataConfig']['s3OutputDataConfig']['s3Uri']
    prefix = '/'.join(output_uri.split('/')[3:])
    objs = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)
    json_key = next(o['Key'] for o in objs['Contents'] if o['Key'].endswith('output.json'))
    data = json.loads(s3.get_object(Bucket=bucket, Key=json_key)['Body'].read())
    clips = data.get('data', [])

    # 벡터를 S3 Vectors에 배치 저장
    vectors = [{
        'key': f"{task_id}_{c['embeddingOption']}_{c['startSec']}_{c['endSec']}",
        'data': {'float32': c['embedding']},
        'metadata': {'task_id': task_id, 'startSec': c['startSec'], 'endSec': c['endSec']}
    } for c in clips]
    for i in range(0, len(vectors), 100):
        s3vectors.put_vectors(
            vectorBucketName=S3_VECTORS_BUCKET,
            indexName=S3_VECTORS_INDEX,
            vectors=vectors[i:i+100]
        )

    return {
        'task_id': task_id,
        'status': 'completed',
        'stored_clips': len(clips)
    }

@tool
def search_video_clips(query: str, top_k: int = 10) -> Dict[str, Any]:
    '''텍스트로 비디오 클립 검색. 결과에 video, timestamp, start_sec, end_sec 포함.'''
    # 텍스트 쿼리를 벡터로 변환
    response = bedrock_runtime.invoke_model(
        modelId=MARENGO_MODEL_ID,
        body=json.dumps({
            'inputType': 'text',
            'text': {'inputText': query}
        }),
        contentType='application/json'
    )
    emb = json.loads(response['body'].read())['data'][0]['embedding']

    # S3 Vectors에서 유사도 검색
    results = s3vectors.query_vectors(
        vectorBucketName=S3_VECTORS_BUCKET,
        indexName=S3_VECTORS_INDEX,
        queryVector={'float32': emb},
        topK=top_k,
        returnDistance=True,
        returnMetadata=True
    )

    clips = []
    for v in results.get('vectors', []):
        meta = v.get('metadata', {})
        task_info = task_table.get_item(Key={'task_id': meta.get('task_id')}).get('Item', {})
        start = int(meta.get('startSec', 0))
        end = int(meta.get('endSec', 0))
        clips.append({
            'video': task_info.get('s3_key', ''),
            's3_bucket': task_info.get('s3_bucket', ''),
            'timestamp': f'{start//60}:{start%60:02d}-{end//60}:{end%60:02d}',
            'start_sec': start,
            'end_sec': end,
        })

    return {'query': query, 'clips': clips}


@tool
def get_clip_playback_url(s3_bucket: str, s3_key: str, start_sec: int, end_sec: int) -> Dict[str, Any]:
    '''S3 영상의 특정 구간 재생 가능한 presigned URL 생성.'''
    base_url = s3.generate_presigned_url(
        'get_object', Params={'Bucket': s3_bucket, 'Key': s3_key}, ExpiresIn=3600
    )
    return {'playback_url': f'{base_url}#t={start_sec},{end_sec}'}

@tool
def summarize_video(s3_uri: str, prompt: str = '이 영상을 챕터별로 3문장 정도로 요약해줘') -> Dict[str, Any]:
    '''Pegasus 모델로 영상 요약 생성.'''
    response = bedrock_runtime.invoke_model(
        modelId=PEGASUS_MODEL_ID,
        body=json.dumps({
            'inputPrompt': prompt,
            'mediaSource': {
                's3Location': {'uri': s3_uri, 'bucketOwner': ACCOUNT_ID}
            }
        }),
        contentType='application/json'
    )
    result = json.loads(response['body'].read())
    return {'s3_uri': s3_uri, 'summary': result.get('message', result)}

def _get_transcript_data(s3_uri: str):
    '''S3에서 자막 JSON 데이터를 가져오는 헬퍼 함수.'''
    bucket = s3_uri.split('/')[2]
    job_name = f"transcript-{hashlib.md5(s3_uri.encode()).hexdigest()[:8]}"
    return json.loads(
        s3.get_object(Bucket=bucket, Key=f'transcripts/{job_name}.json')['Body'].read()
    )

def _ensure_transcript(s3_uri: str) -> Dict[str, Any]:
    '''자막이 없으면 Amazon Transcribe로 생성하는 헬퍼 함수.'''
    bucket = s3_uri.split('/')[2]
    job_name = f"transcript-{hashlib.md5(s3_uri.encode()).hexdigest()[:8]}"
    output_key = f'transcripts/{job_name}.json'
    try:
        status = transcribe.get_transcription_job(
            TranscriptionJobName=job_name
        )['TranscriptionJob']['TranscriptionJobStatus']
    except transcribe.exceptions.BadRequestException:
        transcribe.start_transcription_job(
            TranscriptionJobName=job_name,
            Media={'MediaFileUri': s3_uri},
            MediaFormat='mp4',
            LanguageCode='ko-KR',
            OutputBucketName=bucket,
            OutputKey=output_key
        )
        status = 'IN_PROGRESS'
    if status == 'IN_PROGRESS':
        while status == 'IN_PROGRESS':
            time.sleep(5)
            status = transcribe.get_transcription_job(
                TranscriptionJobName=job_name
            )['TranscriptionJob']['TranscriptionJobStatus']
    return {'transcript_file': f's3://{bucket}/{output_key}'}

@tool
def get_transcript_file(s3_uri: str) -> Dict[str, Any]:
    '''자막 파일 S3 경로 반환 (없으면 생성).'''
    return _ensure_transcript(s3_uri)

@tool
def get_transcript(s3_uri: str) -> Dict[str, Any]:
    '''자막을 타임스탬프와 함께 테이블 형식으로 반환. 없으면 자동 생성.'''
    try:
        data = _get_transcript_data(s3_uri)
    except:
        _ensure_transcript(s3_uri)
        data = _get_transcript_data(s3_uri)

    # 5초 단위로 그룹핑
    grouped = {}
    for item in data['results']['items']:
        if item['type'] == 'pronunciation':
            sec = int(float(item.get('start_time', 0)))
            key = f"{sec//5*5//60}:{sec//5*5%60:02d}"
            grouped[key] = grouped.get(key, '') + ' ' + item['alternatives'][0]['content']

    return {'transcript': [{'시간': k, '자막': v.strip()} for k, v in grouped.items()]}


@tool
def get_keywords(s3_uri: str) -> Dict[str, Any]:
    '''영상 자막에서 키워드 추출.'''
    try:
        data = _get_transcript_data(s3_uri)
    except:
        _ensure_transcript(s3_uri)
        data = _get_transcript_data(s3_uri)
    transcript = ' '.join([
        i['alternatives'][0]['content']
        for i in data['results']['items']
        if i['type'] == 'pronunciation'
    ])[:2000]

    # Claude Haiku로 키워드 추출
    response = bedrock_runtime.invoke_model(
        modelId='us.anthropic.claude-3-5-haiku-20241022-v1:0',
        body=json.dumps({
            'anthropic_version': 'bedrock-2023-05-31',
            'max_tokens': 256,
            'messages': [{
                'role': 'user',
                'content': f'핵심 키워드 10개 JSON 배열로만:\n{transcript}'
            }]
        }),
        contentType='application/json'
    )
    return {'keywords': json.loads(response['body'].read())['content'][0]['text']}

from strands.handlers.callback_handler import PrintingCallbackHandler

SYSTEM_PROMPT = '''당신은 영상 분석 에이전트입니다. 한국어로 응답하세요.

## 사용 가능한 도구

1. **create_video_embedding(s3_uri)**: 영상 임베딩 생성 및 벡터 DB 저장
2. **search_video_clips(query, top_k)**: 텍스트로 비디오 클립 검색
3. **get_clip_playback_url(s3_bucket, s3_key, start_sec, end_sec)**: 재생 URL 생성
4. **summarize_video(s3_uri, prompt)**: Pegasus 모델로 영상 요약
5. **get_transcript(s3_uri)**: 자막 조회 (없으면 자동 생성)
6. **get_keywords(s3_uri)**: 자막에서 키워드 추출

## 도구 선택 가이드라인
- 임베딩 생성 또는 요약 → summarize_video, create_video_embedding
- 클립 검색 또는 특정 장면 찾기 → search_video_clips
- 자막, 키워드 → get_transcript, get_keywords
- 복잡한 요청은 여러 도구를 순차적으로 사용
'''

orchestrator = Agent(
    model=claude_model,
    tools=[create_video_embedding, search_video_clips,
           get_clip_playback_url, summarize_video,
           get_transcript, get_keywords],
    system_prompt=SYSTEM_PROMPT,
    callback_handler=PrintingCallbackHandler()
)


이제 자연어로 복합 요청을 보낼 수 있습니다. 사용할 예시 영상 uri를 사용하여 검색해볼 수 있습니다.

response = orchestrator(
    "s3://my-bucket/sample-video.mp4 영상 임베딩해줘"
    "그 후에 골대에 공이 들어가는 장면 찾아줘"
)

# 영상 분석 전문 에이전트
video_analysis_agent = Agent(
    model=claude_model,
    tools=[create_video_embedding, summarize_video],
    system_prompt='당신은 영상 분석 전문 에이전트입니다. 영상 임베딩 생성과 요약을 담당합니다.'
)

# 검색 전문 에이전트
search_agent = Agent(
    model=claude_model,
    tools=[search_video_clips, get_clip_playback_url],
    system_prompt='당신은 영상 검색 전문 에이전트입니다. 클립 검색과 재생 URL 생성을 담당합니다.'
)

# 자막 처리 전문 에이전트
transcript_agent = Agent(
    model=claude_model,
    tools=[get_transcript_file, get_transcript, get_keywords],
    system_prompt='당신은 자막 처리 전문 에이전트입니다. 자막 생성, 조회, 키워드 추출을 담당합니다.'
)

@tool
def video_analysis_tool(query: str) -> str:
    """
    영상 분석 도구 (임베딩 생성, 요약)

    Use this tool for video embedding creation and summarization.

    Args:
        query: The analysis request (String)

    Returns:
        Analysis results (String)
    """
    response = video_analysis_agent(query)
    return str(response)

@tool
def video_search_tool(query: str) -> str:
    """
    영상 검색 도구 (클립 검색, 재생 URL 생성)

    Use this tool for searching video clips and generating playback URLs.

    Args:
        query: The search request (String)

    Returns:
        Search results with playback links (String)
    """
    response = search_agent(query)
    return str(response)

@tool
def transcript_tool(query: str) -> str:
    """
    자막 처리 도구 (자막 생성/조회, 키워드 추출)

    Use this tool for transcript generation, retrieval, and keyword extraction.

    Args:
        query: The transcript request (String)

    Returns:
        Transcript or keyword results (String)
    """
    response = transcript_agent(query)
    return str(response)

from strands.agent.conversation_manager import SlidingWindowConversationManager

ORCHESTRATOR_PROMPT = """
You are the Video Processing Orchestrator Agent. Always respond in Korean.

Tool Selection Guidelines: 
- For embedding creation or video summarization → video_analysis_tool
- For searching clips or specific scenes → video_search_tool
- For transcripts, subtitles, or keywords → transcript_tool
- For complex requests, use multiple tools in sequence 
"""

orchestrator = Agent(
    system_prompt=ORCHESTRATOR_PROMPT,
    tools=[video_analysis_tool, video_search_tool, transcript_tool], #사용할 에이전트 지정
    model=claude_model,
    callback_handler=PrintingCallbackHandler(),
    conversation_manager=SlidingWindowConversationManager(
        window_size=20,
        should_truncate_results=True
    )
)

response = orchestrator(
    "s3://my-bucket/sample-video.mp4 내용 요약해주고 키워드 찾아줘."
    "그 후에 자막을 1분까지만 보여줘."
)