LangGraphを用いたAI動画生成アプリの構築

はじめに

こんにちは。KINTOテクノロジーズの JR.Liang です。

本記事では、AIと対話するだけでプレゼン動画が作れるアプリを題材に、生成AIアプリ向けのフレームワーク LangGraph を使ってこのアプリをどう構築するのかを解説します。特に、LangGraph の機能で複雑な処理の流れ（フロー）をどう制御したかに焦点を当て、使った機能を一つずつ説明していきます。

まず、私たちが作りたかったものはこんなアプリです。

• 使う人：生成AIも動画編集も、知識と経験が少ない人。
• やること：アプリの質問に答えていくと、簡単に自分のプレゼン動画が1本出来上がる。

ユーザー体験としては 「質問に答える → できた物を見て、気になれば直す → 完成」 という流れ、とてもシンプルです。こちらが、AI（LLM）部分の実行基盤として Claude Code（Agent SDK）を組み込んだアプリの操作イメージです。

左が動画プレビュー、右がチャット。対話 → 台本・構成確認 → 生成 → チャットで修正 → 完成の流れ

このシンプルな見た目を成り立たせる裏側の複雑さを、LangGraph が引き受けて管理しやすいようにしました。

生成AIアプリはなぜ「一直線」で書けないのか

通常のプログラムは「A を呼んで、次に B、次に C」と一直線に書けます。ところが生成AIが絡むと、返答・結果物が毎回ブレる（非決定的）ため、一回で完璧にならず、確認・修正の工程が欠かせません。

その結果、ユーザーが満足できる動画を生成するまでの過程は「行ったり来たり」する流れになります。

• 制作途中で確認したい：人間に確認させる
• 品質が悪い・修正したい：前の工程に戻ってやり直す
• ユーザーのさまざまな要望で：流れ内の工程が変わる
• いつでも編集を続けたい：中断・再開

これらの状態と分岐を整理し、管理するツールが LangGraph です。

LangGraph とは

LangGraph は、AIアプリの「処理の流れ」を組み立てるオープンソースのライブラリです。ChatGPT 連携で知られる LangChain チームが作っています。

特徴は、処理を 3つの要素の組み合わせ＝Graph として書けること。

• Node：1つの作業単位。この記事では「◯◯する係」と呼びます。
• Edge：Node から Node へ引く矢印。次にどこへ進むか。
• State（記録シート）：全 Node で共有する1枚の記録シート。各 Node が状態を書き込み、読み取ります。

Node を1つずつ実行して進むので、「分岐・ループ・途中で確認・中断再開」が絡まらずに書けます。この記事で使う技術は次の5つです。

ここからは、この5つの技術が どの流れで、何のために使われているか を1ステップずつ見ていきます。

全体の流れ

設計方針はこうです。複雑さは全部 LangGraph 側に仕込んで、ユーザーには理解しやすい流れを見せる。

ユーザーが体験するのは、次の5ステップです。

基本の流れとして各ステップを一問一答の形で進めます。加えて、どのステップからでも前に戻ってやり直せます（図の点線）。ユーザーの一言次第で、1つ前にも、さらに前にも戻れます。この「行ったり来たり」を破綻させずに扱えるのも LangGraph の強みです。

流れ① 何を作りたいか、対話で聞く

ユーザー体験

空欄のフォームではなく、1問ずつ答えるだけで動画制作に必要な情報：「内容・素材・尺・対象者」などを集めます。AIに不慣れでも、アプリが親切に案内してくれます。

使う技術：人への割り込み（interrupt）＋ 保存（checkpointer）

質問の度に実行を一時停止して人に聞き、答えをもらって続きます（Human-in-the-Loop）。答えは鵜呑みにせず、AIが「動画を作れる情報を聞き出したか」を判定し、足りなければ具体的に聞き直します。質問を1問＝1Node に分け、進むごとに保存するので、途中で閉じても続きから再開できます。

コードサンプル

from langgraph.graph import StateGraph, START
from langgraph.types import interrupt, Command


# 「話す内容を聞く係」のNode
def 話す内容を聞く係(state):
    回答 = interrupt("どんな内容の動画を作りたいですか？")  # 一時停止して人に聞く

    判定 = AIで確認("話す内容として十分か？", 回答)  # 答えが十分かAIが判定
    if not 判定.OK:  # 足りなければ…
        return Command(goto="話す内容を聞く係")  # 同じNodeに戻って聞き直し

    return {"企画書": 回答}  # OKなら記録シート（企画書）に保存 → 次へ


builder = StateGraph(State)
builder.add_node("話す内容を聞く係", 話す内容を聞く係)
builder.add_node("使う素材を聞く係", 使う素材を聞く係)
builder.add_edge(START, "話す内容を聞く係")  # ← ここから開始（入口）
# … 残りの Node・Edge を追加 …
graph = builder.compile(checkpointer=checkpointer)  # 全部つないでから compile（各Nodeの後に自動保存）

使っている部品

• interrupt(質問)：一時停止して人に聞く
• Command(goto=…)：次に進む Node を指定（同じ Node を指定すれば聞き直し）
• compile(checkpointer=…)：Node ごとに保存

流れ② 台本・構成をAIが下書きし、確認して直せる

ユーザー体験

集めた情報から台本と構成の下書きをAIが作って見せます。ユーザーは 「OK」や「ここを直して」など自分の言葉で確認・修正の指示 ができます。

使う技術：人への割り込み（確認ゲート）＋ 動的な分岐

下書きを見せて一時停止し、返事を受け取ります。ここがポイントで、ユーザーの自由な一言の意味をAIが判定し、「どの Node に戻すか」を実行時に選びます（動的な分岐）。「2枚目のシーンを短くする」なら構成をやり直す、「柔らかい言い方にする」なら台本をやり直す、「OK」なら次へ。

コードサンプル

# 「確認ゲート」のNode
def 確認ゲート(state):
    返事 = interrupt("この台本でよろしいですか？ 直したいところがあれば教えてください")

    行き先 = AIで振り分け(返事, ["台本を直す", "構成を直す", "次へ"])  # 意味を判定
    return Command(goto=行き先)  # ← 返事次第で戻る先が変わる（動的な分岐）

使っている部品

• Command(goto=…)：次の Node を実行時に選ぶ。行き先を固定せず返事次第で変えられる＝これが「動的な分岐」。

流れ③ 計画（シーン構成）の確認とデザインの準備を同時に進める

ユーザー体験

「どんなシーン構成で組み立てるか」の計画を確認します。この確認の裏で次のステップのデザインを同時に進め、デザインの確認ができるまでの待ち時間を短縮します。

使う技術：並行実行

ポイントは、「計画の確認」と「デザインの準備」は互いに独立していること。計画の確認は人間の返事を待つ必要がありますが、デザインの下準備はその返事に関係なく進められます。そこで1つの Node から矢印（Edge）を2本同時に出し、片方が「人間の確認待ち」で止まっている裏で、もう片方が次に必要なデザインの下準備を先回りで計算しておきます。OK を出す頃には準備が済んでいるので、全体の作業時間が短くなります。

コードサンプル

# 構成ができたら 2 本の矢印を同時に出す（並行実行）
builder.add_edge("構成を作る係", "計画の確認")  # 片方：人間の確認待ち
builder.add_edge("構成を作る係", "デザイン先読み係")  # もう片方：裏でデザインを先読み


# 先読み係：確認待ちの裏でデザインを計算し、記録シートに置いておく
def デザイン先読み係(state):
    return {"先読みデザイン": デザインを生成(state)}


# 計画の確認がOKなら、次の「デザインの確認」へ進む
builder.add_edge("計画の確認", "デザインの確認")


# デザインの確認：先読み済みがあれば再計算せず、すぐ表示する
def デザインの確認(state):
    デザイン = state.get("先読みデザイン") or デザインを生成(state)
    interrupt(("このデザインでよろしいですか？", デザイン))
    return {}

使っている部品

• add_edge(元, 先)：矢印を1本引く。同じ元から2本のEdgeを設定することで、複数のNodeを並行して実行できる（＝並行実行）。
• state["先読みデザイン"]：先読みした結果を記録シートに残し、デザインの確認ですぐ取り出す（共有 State）。

流れ④ 動画生成して採点し、問題点・直すところをユーザーが確認

ユーザー体験

最初の動画を組み立て、ユーザーの要望に合わせて生成動画を採点します。ユーザー自身が「どこか自分の要望に合っていないところはないか」を確かめられるようになります。ユーザーは仕上がりを見て「OK」か「直して」を言うだけです。

使う技術：人への割り込み ＋ 共有 State ＋ ループ

「動画生成 → 採点 → 点数と問題点を提示 → ユーザーが『どこを直すか』を確認（割り込み） → 修正 → もう一度採点…」というループを回します。ポイントは採点のものさし：共有 State から、今までの流れで聞き取れたユーザーの要望を読み出し、できた動画を要望と照らし合わせて、点数とアドバイスを提示します。

コードサンプル

# 「採点する係」のNode（採点だけ。割り込みは持たない）
def 採点する係(state):
    点数, 問題点 = AIで採点(
        state["video"], state["企画書"]
    )  # 記録シートの要望を"ものさし"に採点

    if 点数.合格:  # 要望どおり
        return Command(goto="完成の確認")  # → ループを抜ける

    # 点数と問題点を共有 State に載せて、確認役の Node へ渡す
    return Command(
        goto="ユーザーに確認する係",
        update={"点数": 点数, "問題点": 問題点},
    )


# 「ユーザーに確認する係」のNode（interrupt を先頭に置く）
def ユーザーに確認する係(state):
    # 点数と問題点を見せて、ユーザーに「どこを直すか」を聞く（割り込み）
    直す指示 = interrupt(
        ("点数と問題点はこちらです。どこを直しますか？", state["点数"], state["問題点"])
    )
    return Command(
        goto="修正する係",  # → 直してもう一度採点（ループ）
        update={"直す指示": 直す指示},
    )

使っている部品

• state["企画書"]：今までの流れで保存した要望を読み出す（共有 State）
• interrupt(…)：点数と問題点を見せて「どこを直すか」を聞く
• Command(goto=…)：「修正する係」に戻してループを作る

流れ⑤ 動画を書き出して完成

ユーザー体験

最後に「この動画で完成にしますか？」と一度だけ確認します。

使う技術：人への割り込み（最終ゲート）

最終確認も一時停止のゲート。止まってユーザーに聞き、承認されたら動画を書き出して完成です。

コードサンプル

from langgraph.graph import END


def 完成の確認(state):
    interrupt("この動画で完成にしますか？")  # 承認されるまで進まない
    return {}


builder.add_edge("完成の確認", "書き出す係")  # 承認後は書き出しへ（分岐なし）
builder.add_edge("書き出す係", END)  # 書き出しが終わったら完成

実装で難しかった点：ユーザーのバラバラな要望を、どの段階でも適切に受け止めて対応できるか

実装において一番難しかったのが、ユーザーが自由に打ち込む一言の適切な扱いです。言い方は無数にあるため、同じ「直して」でも戻るべき工程（Node）は違う場合があります。

理想は、ユーザーが今どのステップに居ても、どんな要望でも、それに合わせて処理できること。ユーザー体験をできるだけ簡単にするために、その一言を「どの Node（処理の係）に戻して直すか」を正しく選ぶのが重要ポイントです。そのため「自由文を解釈し、ちょうどいい係（ステップ）に戻す」必要がありました。

私たちが考えた解決方法の軸は 「行き先の候補はコードが決めて、意味の解釈はAIに任せる」 こと。

• 戻れる候補を宣言 → AIがどう解釈しても、宣言外へは飛べない（暴走しない）。
• 自由文の意味はAIが分類 → 候補から一番近いものを選ぶ。
• 迷ったら聞き返す → 曖昧な推測をせず、もう一度確認する。

戻せる先は、今居るステップごとに変える：候補は確認ゲートごとに別々に宣言してあり、今止まっているゲートの候補が、そのまま戻せる範囲になります。

コードサンプル

from typing import Literal


# 戻れる候補を宣言 → 変な所へは飛ばせない
def 確認ゲート(
    state,
) -> Command[
    Literal[
        "軽い修正の係",
        "シーン単位の修正係",
        "構成を作り直す係",
        "全部作り直す係",
        "次へ",
    ]
]:
    返事 = interrupt("直したいところはありますか？")
    判定 = AIで振り分け(
        返事,
        ["軽い修正の係", "シーン単位の修正係", "構成を作り直す係", "全部作り直す係", "次へ"],
    )

    if 判定.曖昧:  # 迷ったら
        return Command(goto="確認ゲート")  # もう一度聞き返す
    return Command(goto=判定.行き先)  # はっきりしていれば最短のNodeへ

最後に

このアプリでユーザーに見せるのは「質問に答える → 確認して直す → 完成」という簡単な流れ。その裏にある 分岐・やり直し・確認・中断からの再開 という複雑さを、LangGraph（共有 State・動的な分岐・ループ・割り込み・保存）がまるごと引き受けてくれました。

おかげで私たちは、どこで人に確認するか・どこでAIに処理させるかというアプリならではの判断の設計に集中できました。生成AIを組み込んだアプリを開発する際に、本記事で用いた概念と技術が参考になれば幸いです。