Videoverständnis in Agenten

Videoverständnis für Agenten

OpenCV zur Frame-Extraktion installieren

# pip install opencv-python-headless
import cv2

def get_video_metadata(video_path: str) -> dict:
    cap = cv2.VideoCapture(video_path)
    if not cap.isOpened():
        raise IOError(f'Cannot open video: {video_path}')

    fps = cap.get(cv2.CAP_PROP_FPS)
    frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    duration_s = frame_count / fps if fps > 0 else 0
    cap.release()

    return {
        'fps': round(fps, 2),
        'frame_count': frame_count,
        'width': width,
        'height': height,
        'duration_seconds': round(duration_s, 2)
    }

meta = get_video_metadata('recording.mp4')
print(meta)

Frames alle N Sekunden extrahieren

import cv2
import os

def extract_frames(
    video_path: str,
    output_dir: str,
    every_n_seconds: float = 5.0
) -> list:
    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    frame_interval = int(fps * every_n_seconds)
    os.makedirs(output_dir, exist_ok=True)

    saved_frames = []
    frame_idx = 0
    while True:
        ret, frame = cap.read()
        if not ret:
            break
        if frame_idx % frame_interval == 0:
            timestamp = round(frame_idx / fps, 2)
            fname = f'{output_dir}/frame_{frame_idx:06d}_{timestamp}s.jpg'
            cv2.imwrite(fname, frame)
            saved_frames.append({'path': fname, 'timestamp': timestamp})
        frame_idx += 1

    cap.release()
    print(f'Extracted {len(saved_frames)} frames')
    return saved_frames

Erkennung von Szenenwechseln

import cv2
import numpy as np

def extract_scene_change_frames(
    video_path: str,
    output_dir: str,
    diff_threshold: float = 30.0
) -> list:
    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    os.makedirs(output_dir, exist_ok=True)
    prev_gray = None
    saved = []
    frame_idx = 0

    while True:
        ret, frame = cap.read()
        if not ret:
            break
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        if prev_gray is not None:
            diff = np.mean(np.abs(gray.astype(float) - prev_gray.astype(float)))
            if diff > diff_threshold:
                ts = round(frame_idx / fps, 2)
                fname = f'{output_dir}/scene_{frame_idx:06d}_{ts}s.jpg'
                cv2.imwrite(fname, frame)
                saved.append({'path': fname, 'timestamp': ts, 'diff': round(diff, 2)})
        prev_gray = gray
        frame_idx += 1

    cap.release()
    return saved

Eine Frame-Sequenz an die Vision API senden

import base64
import anthropic

def describe_video_frames(frame_info_list: list, query: str) -> str:
    client = anthropic.Anthropic(api_key='YOUR_API_KEY')
    content = []
    for fi in frame_info_list:
        with open(fi['path'], 'rb') as f:
            b64 = base64.standard_b64encode(f.read()).decode('utf-8')
        content.append({
            'type': 'text',
            'text': f'Frame at {fi["timestamp"]}s:'
        })
        content.append({
            'type': 'image',
            'source': {'type': 'base64', 'media_type': 'image/jpeg', 'data': b64}
        })
    content.append({'type': 'text', 'text': query})

    response = client.messages.create(
        model='claude-opus-4-5',
        max_tokens=1024,
        messages=[{'role': 'user', 'content': content}]
    )
    return response.content[0].text

Prompts für zeitliches Schlussfolgern

TEMPORAL_QUERY = (
    'You are analysing a video sequence. The frames above are in chronological order '
    'with their timestamps shown. Please:\n'
    '1. Describe what changed between the first and last frame\n'
    '2. Identify any notable events and when they occurred (timestamp)\n'
    '3. Summarise the overall activity shown in the video\n'
    '4. Estimate the total duration of the main activity\n'
    'Be specific about timestamps.'
)

# More targeted temporal query:
CHANGE_DETECTION_QUERY = (
    'Compare frame at {start_ts}s and frame at {end_ts}s. '
    'List all visible differences in bullet points. '
    'Categorise each change as: object_moved, object_added, '
    'object_removed, lighting_change, or camera_move.'
)

# Usage:
query = CHANGE_DETECTION_QUERY.format(start_ts=0.0, end_ts=20.0)

if __name__ == '__main__':
    print('Temporal reasoning query:')
    print(TEMPORAL_QUERY)
    print()
    print('Change detection query:')
    print(query)

Pipeline zur Videozusammenfassung

def summarise_video(
    video_path: str,
    every_n_seconds: float = 10.0,
    batch_size: int = 5
) -> str:
    import anthropic
    client = anthropic.Anthropic(api_key='YOUR_API_KEY')

    # Extract frames
    frames = extract_frames(video_path, '/tmp/video_frames', every_n_seconds)

    # Process in batches
    batch_summaries = []
    for i in range(0, len(frames), batch_size):
        batch = frames[i:i + batch_size]
        ts_range = f'{batch[0]["timestamp"]}s - {batch[-1]["timestamp"]}s'
        query = f'Summarise what happens in this video segment ({ts_range}).'
        summary = describe_video_frames(batch, query)
        batch_summaries.append(f'[{ts_range}] {summary}')

    # Synthesise
    all_summaries = '\n\n'.join(batch_summaries)
    result = client.messages.create(
        model='claude-opus-4-5',
        max_tokens=512,
        messages=[{'role': 'user', 'content':
            f'Here are segment-level video summaries:\n\n{all_summaries}\n\n'
            'Write a single coherent summary of the entire video.'
        }]
    )
    return result.content[0].text

Objektverfolgung über mehrere Frames

def track_object_across_frames(
    frames: list,
    object_description: str
) -> list:
    import anthropic, base64
    client = anthropic.Anthropic(api_key='YOUR_API_KEY')
    trajectory = []

    for fi in frames:
        with open(fi['path'], 'rb') as f:
            b64 = base64.standard_b64encode(f.read()).decode('utf-8')
        response = client.messages.create(
            model='claude-opus-4-5',
            max_tokens=128,
            messages=[{
                'role': 'user',
                'content': [
                    {'type': 'image', 'source': {'type': 'base64',
                      'media_type': 'image/jpeg', 'data': b64}},
                    {'type': 'text', 'text':
                      f'Where is the {object_description} in this frame? '
                      'Answer with one of: top-left, top-right, centre, '
                      'bottom-left, bottom-right, not-visible.'}
                ]
            }]
        )
        trajectory.append({
            'timestamp': fi['timestamp'],
            'position': response.content[0].text.strip()
        })
    return trajectory

Umgang mit Frame-Anzahllimits

MAX_FRAMES_PER_REQUEST = 8

def hierarchical_video_analysis(frames: list, final_query: str) -> str:
    import anthropic
    client = anthropic.Anthropic(api_key='YOUR_API_KEY')

    # Level 1: analyse each chunk
    chunk_summaries = []
    for i in range(0, len(frames), MAX_FRAMES_PER_REQUEST):
        chunk = frames[i:i + MAX_FRAMES_PER_REQUEST]
        ts = f'{chunk[0]["timestamp"]}s-{chunk[-1]["timestamp"]}s'
        summary = describe_video_frames(
            chunk, f'What happens in this segment ({ts})?'
        )
        chunk_summaries.append(f'Segment {ts}: {summary}')

    # Level 2: synthesise all chunk summaries
    combined = '\n'.join(chunk_summaries)
    result = client.messages.create(
        model='claude-opus-4-5',
        max_tokens=512,
        messages=[{
            'role': 'user',
            'content': f'Video segments:\n{combined}\n\n{final_query}'
        }]
    )
    return result.content[0].text

Audio- und Videoanalyse kombinieren

def full_video_analysis(video_path: str) -> dict:
    import subprocess

    # Step 1: Extract audio track
    audio_path = '/tmp/video_audio.mp3'
    subprocess.run([
        'ffmpeg', '-i', video_path, '-q:a', '0', '-map', 'a',
        audio_path, '-y'
    ], capture_output=True)

    # Step 2: Transcribe audio
    transcript_data = transcribe_with_timestamps(audio_path)

    # Step 3: Extract key frames
    frames = extract_frames(video_path, '/tmp/key_frames', every_n_seconds=15.0)

    # Step 4: Visual summary
    visual_summary = hierarchical_video_analysis(
        frames, 'Summarise the visual content.'
    )

    return {
        'transcript': transcript_data['full_text'],
        'transcript_segments': transcript_data['segments'],
        'visual_summary': visual_summary,
        'frame_count': len(frames)
    }

Aufräumen und Frame-Verwaltung

import os
import shutil
import hashlib

FRAME_CACHE_DIR = '/tmp/frame_cache'

def get_cache_key(video_path: str, every_n_seconds: float) -> str:
    content = f'{video_path}:{every_n_seconds}'
    return hashlib.md5(content.encode()).hexdigest()[:12]

def get_or_extract_frames(video_path: str, every_n_seconds: float) -> list:
    key = get_cache_key(video_path, every_n_seconds)
    cache_dir = os.path.join(FRAME_CACHE_DIR, key)

    if os.path.exists(cache_dir):
        print(f'Cache hit: {key}')
        files = sorted(os.listdir(cache_dir))
        return [
            {'path': os.path.join(cache_dir, f),
             'timestamp': float(f.split('_')[-1].replace('s.jpg', ''))}
            for f in files if f.endswith('.jpg')
        ]
    return extract_frames(video_path, cache_dir, every_n_seconds)

def clear_frame_cache():
    if os.path.exists(FRAME_CACHE_DIR):
        shutil.rmtree(FRAME_CACHE_DIR)
        print('Frame cache cleared')

Wissensüberprüfung

Zusammenfassung: Videoverständnis in Agenten