こんにちは、ノバセル24新卒エンジニアの秦です。
本記事はノバセル テクノ場 出張版2025 Advent Calendar 2025の4日目の記事になります。
はじめに
OpenAIのAPIは、現時点では「動画ファイルそのもの」を直接入力として受け付けるインターフェースを持っていません。そのため、動画の内容をAIで解析したい場合、システム側で「動画を音声と映像フレーム（静止画）に分解し、それぞれをテキストや画像として入力する」という前処理が不可欠になります。
また、動画を解析する場合、動画の長さ自体もボトルネックになります。特に数分〜数十分に及ぶ動画の場合、リアルタイムAPI（同期処理）では大量のトークン消費によるコスト増加や、HTTPタイムアウトの発生といった課題がつきまといます。
本記事では、これらの課題を解決するために、Pythonで前処理（分割）を行い、OpenAIのBatch APIを活用して長時間動画を効率的かつ低コストに非同期処理するための実装フローを紹介します。
Batch APIを利用することで、処理完了まで時間はかかりますが（24時間以内）、通常のAPI利用に比べて50%のコスト削減が可能になります。特に大量の動画アーカイブを処理する場合や、夜間にまとめて解析を行うようなユースケースでは、このコストメリットは非常に大きくなります。
想定読者
- OpenAI APIを使って動画解析を行いたいエンジニア
- 長時間動画の処理コストやタイムアウトに悩んでいる方
- Batch APIの具体的な実装イメージ（特にサイズ制限への対処）を知りたい方
この記事でわかること
- 動画を音声と映像に分割するPython/FFmpegの実装方法
- OpenAI Batch APIでのバッチ登録方法とサイズ制限への対応
- 非同期処理における結果取得とエラーハンドリング
全体フロー
動画解析の全体パイプラインは以下の通りです。
graph TD
Video[動画ファイル] -->|FFmpegで抽出| Audio[音声データ]
Video -->|FFmpegで切り出し| Frames[映像フレーム画像]
Audio -->|分割| AudioChunks[音声チャンク]
AudioChunks -->|Whisper API| Transcript[文字起こしテキスト]
Transcript -->|JSONL整形| AudioBatch[音声バッチファイル]
Frames -->|Base64エンコード & JSONL整形| VideoBatch[映像バッチファイル]
AudioBatch -->|アップロード & バッチ登録| BatchProcess[バッチ処理プロセス]
VideoBatch -->|アップロード & バッチ登録| BatchProcess
BatchProcess -->|待機 & ポーリング| CompletedResult[解析結果ファイル]
CompletedResult -->|ダウンロード & 紐付け| DB[(データベース)]
- 動画取得: 解析対象の動画ファイルを取得します。
- メディア分割: 動画を「音声」と「映像（フレーム）」に分離・加工します。
- 文字起こし: 音声データを文字起こし（Transcription）します。
- Batch API登録: 映像フレームと文字起こし結果をJSONL形式に整形し、OpenAI Batch APIに登録します。
- 結果取得: バッチ処理完了後、結果を取得しDB等に保存します。
特に「メディア分割」と「Batch API登録」のステップにおいて、OpenAIの制限（ファイルサイズやリクエストサイズ）を考慮した設計が必要です。
1. メディア分割の実装
動画ファイルをそのままAIに投げるのではなく、PythonとFFmpegを使って適切な形式に加工していきます。
音声処理フロー（抽出・分割・文字起こし）
動画内の音声をAIで解析するためには、まず音声データをテキスト化（文字起こし）する必要があります。しかし、OpenAIのWhisper APIには25MBというファイルサイズ制限があり、長時間の動画データはそのままではアップロードできない場合があります。
また、プロダクトの要件として「処理全体の完了時間を短縮したい」という事情もありました。巨大な1つのファイルを順次処理するよりも、分割して複数のリクエストとして並列処理させる方が、トータルの待ち時間を短く抑えられるためです。
そこで、以下のステップで処理を行っていきます。
- 抽出: 動画から音声を抜き出す（16kHz, モノラル）。
- 分割: FFmpegの
segmentオプション等を使い、一定時間（例：300秒）ごとに分割する。 - 文字起こし: 分割した各チャンクをWhisper API（同期）でテキスト化する。
Batch APIには音声データ（バイナリ）を直接投げられないため、この文字起こし工程までは同期APIで事前に行っておく必要がある点に注意してください。
import subprocess from pathlib import Path from openai import OpenAI client = OpenAI() def process_audio(video_path: str, output_dir: Path, chunk_seconds: int = 300): output_dir.mkdir(parents=True, exist_ok=True) audio_path = output_dir / "audio.wav" # 1. 動画から音声を抽出 subprocess.run([ "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-i", video_path, "-vn", "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1", str(audio_path) ], check=True) # 2. 音声を一定時間ごとに分割 segment_pattern = output_dir / "chunk_%03d.wav" subprocess.run([ "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-i", str(audio_path), "-f", "segment", "-segment_time", str(chunk_seconds), "-c", "copy", str(segment_pattern) ], check=True) # 3. 分割したチャンクを文字起こし transcriptions = [] for chunk_path in sorted(output_dir.glob("chunk_*.wav")): with open(chunk_path, "rb") as audio_file: transcript = client.audio.transcriptions.create( model="whisper-1", file=audio_file, response_format="text" ) transcriptions.append({ "chunk": chunk_path.name, "text": transcript }) return transcriptions
映像フレームの切り出し
次に、映像解析のために動画から静止画（フレーム）を切り出します。ここでは1秒間に1枚（1fps）のペースでPNG形式として抽出する例を示します。
必要以上に画像サイズが大きいと消費トークン数が無駄になるため、要件に応じてJPEGへの変更やリサイズを行うと良いと思います。
def extract_frames(video_path: str, output_dir: Path, interval_seconds: float = 1.0): output_dir.mkdir(parents=True, exist_ok=True) frame_pattern = output_dir / "frame_%06d.png" subprocess.run([ "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-i", video_path, "-vf", f"fps=1/{interval_seconds}", str(frame_pattern) ], check=True) return sorted(output_dir.glob("frame_*.png"))
2. OpenAI Batch APIへの登録
抽出したデータをBatch API用の形式（JSONL）に変換し、登録します。
Batch APIの仕組みとサイズ制限
Batch APIは、複数のAPIリクエストを1つのJSONLファイルにまとめてアップロードし、非同期で処理させる仕組みです。「ファイル作成 → アップロード → バッチ作成 → 待機 → 結果取得」というフローになります。
実装時に意識すべき制限（2025年12月時点）は以下の2点です。
- 1リクエストの上限: 50 MB
- 1バッチファイルの上限: 200 MB
実装戦略：なぜ「1フレーム1リクエスト」なのか？
今回の実装では、「映像は1リクエスト1フレームとし、それを大量に束ねて200MBごとのバッチファイルを作成する」という方針を採用しました。
もし「1リクエストに可能な限り画像を詰め込む」アプローチを取った場合、「リクエストへのパッキング（50MB制限）」と「バッチへのパッキング（200MB制限）」という2重のサイズ計算が必要になります。さらに、サーバーのリソース節約のために画像を生成しながら順次バッチ化するようなストリーム処理を行う場合、この計算ロジックは非常に複雑になってしまいます。
「1フレーム1リクエスト」であれば、画像1枚が50MBを超えることはまずないため、リクエスト単体のサイズ超過を気にせず、単純にバッチファイルの容量（200MB）だけを管理すれば良くなり、実装が大幅にシンプルになるためです。
バッチファイルの生成コード例
以下は、映像フレームをBase64エンコードし、制限サイズを超えないようにJSONLを分割生成する実装例です。
（音声テキスト用も同様に body 内にテキストを含めて生成します）
import json import base64 from pathlib import Path MAX_BATCH_BYTES = 170_000_000 # 200MBに対し余裕を持たせて170MB MAX_REQUESTS_PER_BATCH = 50_000 def create_visual_batch_files(frames: list[Path], output_dir: Path): current_batch = [] current_size = 0 batch_index = 0 for i, frame_path in enumerate(frames): # 画像をBase64エンコード image_bytes = frame_path.read_bytes() image_b64 = base64.b64encode(image_bytes).decode("ascii") data_url = f"data:image/png;base64,{image_b64}" # Batch APIリクエストオブジェクト request_item = { "custom_id": f"frame-{i}", "method": "POST", "url": "/v1/chat/completions", "body": { "model": "gpt-5-mini", "messages": [ {"role": "system", "content": "この画像のシーンを詳細に描写してください。"}, { "role": "user", "content": [{"type": "image_url", "image_url": {"url": data_url}}] } ], "max_tokens": 300 } } # JSONLの1行としてのサイズを計算 json_line = json.dumps(request_item, ensure_ascii=False) line_size = len(json_line.encode("utf-8")) # 制限チェック（サイズ or 件数） if (current_size + line_size > MAX_BATCH_BYTES) or (len(current_batch) >= MAX_REQUESTS_PER_BATCH): _write_batch_file(current_batch, output_dir / f"batch_visual_{batch_index}.jsonl") current_batch = [] current_size = 0 batch_index += 1 current_batch.append(request_item) current_size += line_size # 残りのリクエストを書き出し if current_batch: _write_batch_file(current_batch, output_dir / f"batch_visual_{batch_index}.jsonl") def _write_batch_file(requests, path): with open(path, "w", encoding="utf-8") as f: for req in requests: f.write(json.dumps(req, ensure_ascii=False) + "\n")
3. 結果の取得とハンドリング
Batch APIは非同期処理であるため、「バッチジョブの登録」の後に「バッチジョブの状態確認（ポーリング）」を一定間隔で行い、バッチジョブ完了を確認後に「結果ファイルの取得」を行うという2段階のプロセスが必要になります。
基本的にボーリングでは、バッチのステータスを定期的に確認し、completed になるのを待つことになります。しかし、ここで一つ注意点があります。
それは、「ステータスが completed ＝ 成功」とは限らない ということです。
稀に、バッチファイルのフォーマット不正などが原因で、ステータスは completed なのに output_file_id（成功結果）が null で、error_file_id（失敗詳細）だけが返ってくるケースがあります。
そのため、単にステータスだけを見て「完了したから結果を取りに行こう！」と判断するのは危険です。これを成功とみなして処理を進めると、結果ファイルが存在しないため後続の処理でエラーになってしまいます。必ず output_file_id の有無や request_counts.failed もあわせて確認するようにしてください。
import time from openai import OpenAI client = OpenAI() def wait_for_batch_completion(batch_id: str, polling_interval: int = 60): while True: batch = client.batches.retrieve(batch_id) print(f"Batch {batch_id} status: {batch.status}") if batch.status == "completed": # 成功ファイルがあるか確認 if not batch.output_file_id: raise Exception(f"Batch completed but no output file. Errors: {batch.error_file_id}") return batch elif batch.status == "failed": raise Exception(f"Batch failed: {batch.errors}") elif batch.status in ["cancelled", "expired"]: raise Exception(f"Batch ended with status: {batch.status}") time.sleep(polling_interval)
結果の保存
ダウンロードした結果ファイル（JSONL）には、各行にリクエスト時の custom_id（例: frame-0）が含まれています。
これをキーにして元の動画フレームや音声チャンクと紐付けを行い、DB等に保存します。今回は映像と音声を別バッチで処理したため、それぞれの結果リストを custom_id ベースでソート・マージして利用します。
4. 今後の展望
前述の「1フレーム1リクエスト」戦略は実装をシンプルにしますが、以下のような課題があります。
- リクエストごとに指示用のプロンプトが必要になる
- 1枚ごとの解釈になるため前後のフレーム関係を踏まえられない
- 音声とも切り離されているため、音声と映像を紐づけて解析ができない
これらを解決するアプローチとして、1バッチ1リクエストにする案を考えています。 バッチの上限は気にせず、1バッチには1リクエストだけを含める形にし、1リクエストに50MBを超えない程度に複数画像を入れ込みます。
これにより、AIが複数のフレームを一度に見られるため「動画の流れ」を考慮した解析が可能になり、音声も1フレームごとに切り分けるよりは1バッチに含めやすくなります。
まとめ
PythonとBatch APIを利用して長時間動画を処理する際のポイントは以下の通りです。
- 非同期処理の活用: 即時性が不要ならBatch APIでコストを大幅（50%）に削減できる。
- 事前分割: FFmpeg等で音声・映像を適切に分割・加工し、音声は先に文字起こししておく。
- サイズ制限への対応: 「1フレーム1リクエスト」等の戦略で、1リクエスト50MB、1バッチ200MBの制限を回避する実装を行う。
- 確実な結果取得:
completedステータスを鵜呑みにせず、必ず出力ファイルの有無を確認する。
動画のAI処理を実装する際、この記事が少しでもお役に立てば嬉しいです
