AI動画生成をWeb APIから呼び出そうとしたとき、最初は普通の同期APIとして実装していました。
@app.post("/generate")
def generate_video(request: GenerateRequest):
output = run_video_generation(request.prompt)
return {"output": output}
動作確認だけならこれでも問題ありません。
しかし、動画生成に数分かかるようになると扱いにくくなりました。
- HTTP接続を長時間維持する必要がある
- ブラウザ側でタイムアウトしやすい
- 同時リクエストを制御しにくい
- GPUが処理中なのか停止しているのか分からない
- エラー時に生成状態が残らない
そこで、動画生成処理をHTTPリクエストから分離することにしました。
今回はRedisやCeleryを使わず、まずPythonとFastAPIだけで最小構成を作ります。
最終的なAPI
クライアントは動画生成を依頼します。
POST /jobs
サーバーは動画完成まで待たず、job_idだけ返します。
{
"job_id": "f78a6b02",
"status": "queued"
}
クライアントは別のAPIで状態を確認します。
GET /jobs/f78a6b02
処理中なら、
{
"job_id": "f78a6b02",
"status": "running"
}
完了後は、
{
"job_id": "f78a6b02",
"status": "completed",
"output": "outputs/f78a6b02.mp4"
}
という形です。
構造は単純です。
Client
|
| POST /jobs
v
FastAPI
|
v
Queue
|
v
Worker
|
v
Video Generator
|
v
Output
1. Jobの状態を定義する
まず生成ジョブを表すモデルを作ります。
from enum import Enum
from pydantic import BaseModel
class JobStatus(str, Enum):
QUEUED = "queued"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
class GenerateRequest(BaseModel):
prompt: str
class Job(BaseModel):
id: str
prompt: str
status: JobStatus
output: str | None = None
error: str | None = None
今回は最小構成なのでDBは使いません。
Jobはメモリ上の辞書に保存します。
jobs: dict[str, Job] = {}
当然ですが、プロセスを再起動すると消えます。
ここではジョブキューの構造を確認することを優先します。
2. Queueを用意する
Python標準のqueue.Queueを使います。
from queue import Queue
job_queue = Queue()
生成リクエストが来たら、動画をその場で作らずQueueに入れます。
import uuid
@app.post("/jobs")
def create_job(request: GenerateRequest):
job_id = uuid.uuid4().hex[:8]
job = Job(
id=job_id,
prompt=request.prompt,
status=JobStatus.QUEUED
)
jobs[job_id] = job
job_queue.put(job_id)
return {
"job_id": job_id,
"status": job.status
}
これでHTTPレスポンスはすぐ返せます。
3. Workerで動画生成を実行する
次にQueueからJobを取り出すWorkerを作ります。
import threading
def worker():
while True:
job_id = job_queue.get()
job = jobs[job_id]
job.status = JobStatus.RUNNING
try:
output = run_video_generation(
job_id=job.id,
prompt=job.prompt
)
job.output = output
job.status = JobStatus.COMPLETED
except Exception as e:
job.error = str(e)
job.status = JobStatus.FAILED
finally:
job_queue.task_done()
FastAPI起動時にWorkerも開始します。
@app.on_event("startup")
def start_worker():
thread = threading.Thread(
target=worker,
daemon=True
)
thread.start()
これで、
HTTP request
↓
Queue
↓
Worker
↓
Generation
に分離できました。
4. 生成処理は別関数にする
重要なのはAPI側にモデル固有の処理を書きすぎないことでした。
def run_video_generation(
job_id: str,
prompt: str
) -> str:
output_path = f"outputs/{job_id}.mp4"
# Call video generation backend here.
return output_path
実際にはこの部分から、
- ComfyUI API
- ローカルPython pipeline
- 別GPUサーバー
- 外部generation API
などを呼び出します。
最近はLTX 2.5のような動画モデルも確認していますが、API側は特定モデルに依存させないようにしています。
例えば、
class VideoBackend:
def generate(
self,
prompt: str,
output_path: str
) -> None:
raise NotImplementedError
というinterfaceを用意しておけば、
class ComfyUIBackend(VideoBackend):
...
class LocalModelBackend(VideoBackend):
...
class RemoteAPIBackend(VideoBackend):
...
のように差し替えられます。
Web APIから見ると、
FastAPI
↓
VideoBackend
↓
実際の生成環境
という関係になります。
モデルを変更するたびにHTTP APIまで書き直さなくて済みます。
5. Job状態を取得するAPI
クライアントから状態を確認できるようにします。
from fastapi import HTTPException
@app.get("/jobs/{job_id}")
def get_job(job_id: str):
job = jobs.get(job_id)
if job is None:
raise HTTPException(
status_code=404,
detail="Job not found"
)
return job
これだけでフロントエンド側は、
const response = await fetch(`/jobs/${jobId}`);
const job = await response.json();
if (job.status === "completed") {
console.log(job.output);
}
のように状態を確認できます。
6. GPUが1枚ならWorkerもまず1つ
ここでWorkerを大量に作れば速くなるかというと、動画生成ではそう単純ではありません。
GPUが1枚しかない環境で複数の重い生成処理を同時実行すると、VRAM不足になる可能性があります。
そのため最初は、
Queue
|
v
Worker × 1
|
v
GPU × 1
という構成にしました。
つまりAPIには複数のリクエストを送れますが、GPU処理は順番に実行します。
例えば3件来た場合、
job-a running
job-b queued
job-c queued
となります。
job-aが終われば、
job-a completed
job-b running
job-c queued
になります。
動画生成ではこの方が扱いやすいケースがあります。
7. Queueの位置も返したくなった
使っていると「queued」だけでは情報が足りなくなりました。
待っているのが自分だけなのか、前に10件あるのか分からないからです。
簡易的には、
{
"job_id": job.id,
"status": job.status,
"queue_size": job_queue.qsize()
}
のような情報も返せます。
ただしqsize()は「自分が何番目か」を正確に表すものではありません。
本番でpositionを管理するなら、Queue自体の設計をもう少しきちんと作る必要があります。
8. この構成の限界
今回のコードはあくまで最小構成です。
特に大きな問題があります。
Jobがメモリにしかない
jobs = {}
なのでサーバーを再起動すると全部消えます。
Workerも同じプロセス
FastAPIとgeneration workerが分離されていません。
複数サーバーに対応できない
API serverを複数立てると、それぞれ別のQueueを持ってしまいます。
Retryがない
一度失敗したJobを自動的に再実行する仕組みもありません。
本番環境なら、
FastAPI
↓
Redis
↓
Celery / RQ / custom worker
↓
GPU worker
↓
Object Storage
のような構成を検討した方がよさそうです。
ただ、最初からここまで構築すると「本当に必要なのか」が分からないままシステムだけ大きくなります。
今回はまず最小構成で、
HTTPリクエストと長時間のGPU処理を分離する
ところまで実装しました。
まとめ
最初の実装は、
POST /generate
↓
数分待つ
↓
video.mp4
でした。
現在は、
POST /jobs
↓
job_id
↓
Queue
↓
Worker
↓
GPU generation
↓
GET /jobs/{id}
という形にしています。
小さな変更ですが、動画生成のような長時間タスクをWebアプリに組み込むときはかなり扱いやすくなりました。
特に良かったのは、動画モデルとAPIを分離できたことです。
モデルを変更しても、
backend.generate(...)
の裏側だけ差し替えればよく、Job APIの仕様は変わりません。
次に改善するとしたら、Redisを入れてJobを永続化し、GPU WorkerをFastAPIプロセスから完全に分離する予定です。