0Pricing
AI Agents · 课时

智能体中的视频理解

提取帧、总结视频并对视频片段进行时间推理

智能体中的视频理解 是 CoddyKit 上的免费 AI Agents 课时。 这是第 3 节课,共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分,你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

面向代理的视频理解

大多数 LLM 无法直接处理视频文件,但代理可以通过提取具有代表性的帧并将其作为图像发送来分析视频。随后,代理会基于这一视觉序列进行推理,以检测随时间发生的事件、变化和模式。

安装 OpenCV 以提取帧

OpenCV(cv2)是 Python 中用于视频处理的标准库。它可以打开视频文件,读取其元数据(帧率、分辨率和帧数),并将单个帧提取为 NumPy 数组。

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

每隔 N 秒提取帧

对于大多数视频理解任务,您不需要处理每一帧。每隔 N 秒采样一帧,就能得到视频内容的代表性摘要。以每秒 1 帧的速度采样时,60 秒的视频会生成 60 帧。

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

场景变化检测

您可以不采用均匀采样,而是在场景边界处提取帧,也就是视觉内容发生显著变化的时刻。这样每次 API 调用都能获得信息量更高的帧。请使用帧差分:如果连续帧之间的平均绝对差异超过阈值,就表示发生了场景变化。

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

向视觉 API 发送帧序列

在一条消息中向视觉 API 发送多个提取的帧。请包含每一帧的时间戳,以便模型推理时间顺序和持续时长。请在提示词中明确要求进行时间推理。

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

时间推理提示词

时间推理需要特定的提示词设计。请要求模型明确比较各帧,指出发生了哪些变化,估算活动的持续时间,并识别帧与帧之间的因果关系。

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)

视频摘要处理流程

完整的视频摘要处理流程是:提取帧 → 分批发送给视觉模型 → 收集每批摘要 → 综合最终摘要。分批可以避免超出上下文窗口限制。

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

跨帧跟踪对象

请要求视觉模型跟踪帧序列中的特定对象。例如:'在每一帧中,描述红色汽车的位置:左上、右上、中心、左下、右下。'这样无需计算机视觉跟踪算法,也能生成粗略的运动轨迹。

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

处理帧数限制

视觉 API 的上下文窗口限制了每次请求可发送的图像数量(通常为 5–20 张)。对于长视频,请采用分层方法:分别分析各个片段,然后使用第二次 LLM 调用,将片段摘要合并为最终摘要。

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

结合音频与视频分析

要获得最丰富的理解结果,请将帧级视觉分析与 Whisper 音频转录结合起来。代理可以将每个时间戳处看到的内容与说出的内容关联起来,从而实现会议分析或教程索引等强大功能。

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

清理与帧管理

提取的帧可能会迅速累积。处理完成后,请始终清理临时帧文件;同时实现一个以(视频路径、帧率)的哈希值为键的帧缓存,以避免对同一视频进行重复分析时再次提取帧。

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

知识检查

对于视频分析,为什么场景变化检测比均匀采样帧更合适?

回顾:代理中的视频理解

做得很好!本课要点如下:

  • OpenCV:使用 cap.read() 提取帧;每隔 N 秒采样,或在场景变化处采样
  • 帧序列:按时间顺序并带时间戳发送,以提供时间上下文
  • 分层分析:分成多个批次 → 片段摘要 → 最终综合
  • 结合分析:ffmpeg 提取音频;Whisper 进行转录;将转录内容与视觉帧交叉核对
  • 帧缓存:使用以哈希值为键的缓存目录,避免重复提取

下一步:跨模态推理——文本表达一种内容,而图像展示另一种内容时该如何处理。

常见问题解答

「智能体中的视频理解」课时是免费的吗?

是的 — 「智能体中的视频理解」的完整文本可在网页上免费阅读。要进行交互式练习(内置代码编辑器和全天候 AI 导师)并解锁 AI Agents 课程的其余内容,请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「智能体中的视频理解」这节课中我会学到什么?

提取帧、总结视频并对视频片段进行时间推理 你通过在浏览器中直接运行的动手代码来练习 AI Agents,全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗?

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者,你可以从这里开始或从头开始,按照自己的节奏学习。 这是第 3 节课,共 4 节。

「智能体中的视频理解」课时需要多长时间?

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动,所以你能稳步进步,并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗?

能。每节 AI Agents 课都包含内置代码编辑器,你可以在浏览器中直接编写并运行真实代码,并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 使用 Claude Vision 和 GPT-4V 的图像 + 文本智能体
  2. 音频 + 文本智能体工作流
  3. 智能体中的视频理解
  4. 跨模态推理模式
← 返回 AI Agents