TwelveLabs 비디오 에이전트

TwelveLabs 비디오 에이전트 (TwelveLabs Video Agent)

TwelveLabs Pegasus를 사용해 비디오를 이해하는 Pydantic AI 에이전트 예시예요.

이 예시가 보여주는 것:

여기서 아이디어는 "비디오 분석가" 에이전트예요. 사용자가 비디오(URL이 주어짐)에 대해 질문하면, 에이전트는 analyze_video 도구를 사용해 비디오 이해 모델인 TwelveLabs Pegasus를 호출해서 답을 얻어요. LLM은 비디오에 대해 무엇을 물어볼지 결정하고, Pegasus가 실제 비디오 이해를 수행해요.

예시 실행하기

TWELVELABS_API_KEY 환경 변수로 TwelveLabs API 키를 설정해야 해요. 무료 키는 twelvelabs.io에서 받을 수 있어요 — 관대한 무료 티어가 있어요.

예시 에이전트는 openai:gpt-5-mini에서 실행되므로, OPENAI_API_KEY 환경 변수로 OpenAI API 키도 설정해야 해요.

선택적으로 VIDEO_URL을 설정해 에이전트가 여러분의 공개적으로 접근 가능한 비디오를 가리키게 할 수 있어요. 설정하지 않으면 짧은 공개 샘플 클립이 사용돼요.

의존성 설치와 환경 변수 설정이 끝나면 실행하세요:

터미널

python -m pydantic_ai_examples.twelvelabs_video_agent

터미널

uv run -m pydantic_ai_examples.twelvelabs_video_agent

예제 코드

twelvelabs_video_agent.py

from __future__ import annotations as _annotations

import asyncio
import os
from dataclasses import dataclass

import logfire
from twelvelabs import AsyncTwelveLabs
from twelvelabs.types import VideoContext_Url

from pydantic_ai import Agent, RunContext

# 'if-token-present' means nothing will be sent (and the example will work) if you don't have logfire configured
logfire.configure(send_to_logfire='if-token-present')
logfire.instrument_pydantic_ai()

# A public sample video used when the user doesn't provide one. The URL must point at a
# video file TwelveLabs can fetch directly; set VIDEO_URL to use your own.
DEFAULT_VIDEO_URL = 'https://commondatastorage.googleapis.com/gtv-videos-bucket/sample/ElephantsDream.mp4'


@dataclass
class Deps:
    twelvelabs: AsyncTwelveLabs
    video_url: str


video_agent = Agent(
    'openai:gpt-5-mini',
    instructions=(
        'You help users understand a video. '
        'Use the `analyze_video` tool to ask the video-understanding model questions, '
        'then answer the user concisely based on what it returns.'
    ),
    deps_type=Deps,
    retries=2,
)


@video_agent.tool
async def analyze_video(ctx: RunContext[Deps], prompt: str) -> str:
    """Analyze the video with TwelveLabs Pegasus and return a text answer.

    Args:
        ctx: The context.
        prompt: What to ask about the video, e.g. "Summarize this video" or
            "What objects appear in the first 10 seconds?".
    """
    response = await ctx.deps.twelvelabs.analyze(
        model_name='pegasus1.5',
        video=VideoContext_Url(url=ctx.deps.video_url),
        prompt=prompt,
        max_tokens=2048,
    )
    return response.data or ''


async def main():
    api_key = os.environ.get('TWELVELABS_API_KEY')
    if not api_key:
        raise RuntimeError(
            'Set TWELVELABS_API_KEY to run this example. '
            'Grab a free key at https://twelvelabs.io.'
        )
    video_url = os.environ.get('VIDEO_URL', DEFAULT_VIDEO_URL)

    async with AsyncTwelveLabs(api_key=api_key) as client:
        deps = Deps(twelvelabs=client, video_url=video_url)
        result = await video_agent.run(
            'Give me a one-sentence summary of this video.', deps=deps
        )
        print('Response:', result.output)


if __name__ == '__main__':
    asyncio.run(main())

출처: 문서

더 알아보기 (Learn more)