ビデオ編集タスク向けの意図解析・グラフ計画・ツールルーティングを備えた VideoAgent スタイルのマルチエージェントシステムの構築
本記事は、LLM を活用したマルチエージェントシステム「VideoAgent」の構築チュートリアルを提供し、意図解析からグラフ計画、ツールルーティングに至るまでの実装コードと、FFmpeg や Whisper を組み合わせた具体的な動画編集ワークフローを公開している。
キーポイント
自律型マルチエージェント・パイプラインの構築
意図パーサー、エージェントライブラリ、ツールルーター、グラフプランナー、および依存関係修復用のテキスト勾配最適化器からなるコア・パイプラインを定義し、自然言語からの動画編集指示を実行可能な計画に変換する仕組みを示している。
API キー不要な軽量環境の実装
OpenAI や DeepSeek などの主要プロバイダーに対応しつつ、ローカル環境や Colab で API キーを設定せずに動作可能な構成を提供し、 Whisper による文字起こしやシーン検出など、多様な動画処理ツールを統合している。
包括的な動画処理機能の統合
キーフレームサンプリング、キャプション生成、クロスモーダルインデックス、ビート同期編集、最終レンダリングまでを含む一連の機能を、LLM の制御下で自動化する具体的な実装コードを提示している。
柔軟なLLMプロバイダー統合
OpenAI, DeepSeek, Anthropic, Geminiの複数プロバイダーに対応し、環境変数や設定ファイルから自動でAPIキーとモデルを切り替える汎用的なLLMクラスを実装しています。
堅牢な依存関係管理
Colab環境やローカル環境の両方で動作するよう、whisper, gTTS, numpy, pillowなどの外部ライブラリを動的にインストール・フォールバックさせる仕組みが組み込まれています。
構造化JSON出力の保証
LLMからの応答に対して正規表現を用いてコードブロック記法を除去し、JSONパースを試みるロジックを実装することで、システム間のデータ連携の信頼性を高めています。
VideoAgent ランタイムの初期設定と依存関係管理
LLM バックエンド、作業ディレクトリ、パッケージインストールフローを構成し、Colab やローカル環境でスムーズに動作するよう軽量な依存関係フォールバックを実装します。
重要な引用
We define an intent parser, an agent library, a tool router, a graph planner, and a textual-gradient optimizer that repairs missing dependencies in the execution graph.
By the end of the tutorial, we have a complete multi-agent video system that can answer questions about a video, summarize its content, generate a news-style overview, and produce edited video artifacts from natural-language instructions.
"Return assistant text, or None on any failure (caller falls back)."
"Chat + robust JSON extraction."
We create a shared helper layer for shell commands, pip installs, file paths, and environment detection so the notebook runs smoothly in Colab or locally.
We also define a unified LLM wrapper that supports OpenAI, DeepSeek, Anthropic, and Gemini while safely falling back to deterministic execution when no API key is available.
影響分析・編集コメントを表示
影響分析
このチュートリアルは、LLM を単なる情報検索ツールとしてではなく、複雑なマルチステップタスク（動画編集など）を実行する自律的なエージェントとして機能させるための具体的なアーキテクチャを示しており、開発者が即座に実装・検証できる点で業界への貢献度が大きい。特に、依存関係の自動修復やローカル環境での実行可能性を強調している点は、実社会での導入障壁を下げる重要な一歩となる。
編集コメント
本記事は、抽象的な概念論ではなく、実際にコードをコピペして動作させるための詳細な実装ガイドを提供しており、マルチエージェントシステムの開発者にとって極めて有用なリソースです。特に「テキスト勾配最適化器」による依存関係の自動修復機能は、複雑なワークフローの安定稼働に向けた画期的なアプローチと言えます。
本チュートリアルでは、動画の理解・検索・編集・再構築を支える中核となるエージェントパイプラインに焦点を当て、VideoAgent のワークフローを実行可能な形で再現します。まずは API キーが不要な軽量環境の構築から始めます。ここでは意図解析器（Intent Parser）、エージェントライブラリ、ツールルーティング機能、グラフプランナー、そして実行グラフ内の依存関係欠落を修復するテキスト勾配最適化器を定義します。
さらに、これらの計画コンポーネントを実際の動画処理ツールに接続します。具体的には FFmpeg、Whisper による文字起こし、シーン検出、キーフレームサンプリング、キャプション生成、クロスモーダルインデックスの構築と検索、トリミング、ビート同期編集、そして最終的なレンダリングなどです。
本チュートリアルの終了時には、動画に関する質問への回答や内容要約、ニュース形式の概要生成、自然言語による指示に基づく編集済み動画アートの作成が可能になる、完全なマルチエージェント動画システムが完成します。
CONFIG = {
"provider": "",
"api_key": "",
"base_url": "",
"model": "",
"max_shots": 4,
"opt_rounds": 4,
"run_demos": ["qa", "overview", "highlight", "beatsync"],
}
import os, sys, subprocess, json, math, re, wave, textwrap, shutil, time
from collections import defaultdict, deque
def _sh(cmd):
return subprocess.run(cmd, capture_output=True, text=True)
def _pip(pkgs):
_sh([sys.executable, "-m", "pip", "install", "-q", *pkgs])
IN_COLAB = "google.colab" in sys.modules
if IN_COLAB or os.environ.get("VA_INSTALL", "1") == "1":
for spec in (["openai-whisper"], ["gTTS"], ["sentence-transformers"]):
try:
_pip(spec)
except Exception as e:
print(f"[install] {spec} failed ({e}); a fallback will be used.")
try:
import numpy as np
except Exception:
_pip(["numpy"]); import numpy as np
try:
from PIL import Image, ImageDraw, ImageFont
except Exception:
_pip(["pillow"]); from PIL import Image, ImageDraw, ImageFont
HAS_FFMPEG = shutil.which("ffmpeg") is not None
WORK = "/content/va_workdir" if IN_COLAB else os.path.abspath("./va_workdir")
os.makedirs(WORK, exist_ok=True)
def wp(*a): return os.path.join(WORK, *a)
import urllib.request, urllib.error
class LLM:
DEFAULT = {
"openai": "gpt-4o-mini", "deepseek": "deepseek-chat",
"anthropic": "claude-3-5-sonnet-latest", "gemini": "gemini-1.5-flash",
}
def __init__(self, cfg):
self.provider = (cfg.get("provider") or "").lower().strip()
self.key = (cfg.get("api_key") or
os.environ.get(f"{self.provider.upper()}_API_KEY", "") or
os.environ.get("OPENAI_API_KEY", "") if self.provider else "")
self.model = cfg.get("model") or self.DEFAULT.get(self.provider, "")
self.base = cfg.get("base_url") or {
"openai": "https://api.openai.com/v1",
"deepseek": "https://api.deepseek.com/v1",
"anthropic": "https://api.anthropic.com/v1",
"gemini": "https://generativelanguage.googleapis.com/v1beta",
}.get(self.provider, "")
def available(self):
return bool(self.provider and self.key)
def _post(self, url, payload, headers):
data = json.dumps(payload).encode()
req = urllib.request.Request(url, data=data, headers=headers, method="POST")
with urllib.request.urlopen(req, timeout=90) as r:
return json.loads(r.read().decode())
def chat(self, system, user, temperature=0.2):
"""Return assistant text, or None on any failure (caller falls back)."""
if not self.available():
return None
try:
if self.provider in ("openai", "deepseek"):
out = self._post(
f"{self.base}/chat/completions",
{"model": self.model, "temperature": temperature,
"messages": [{"role": "system", "content": system},
{"role": "user", "content": user}]},
{"Content-Type": "application/json",
"Authorization": f"Bearer {self.key}"})
return out["choices"][0]["message"]["content"]
if self.provider == "anthropic":
out = self._post(
f"{self.base}/messages",
{"model": self.model, "max_tokens": 2000,
"system": system,
"messages": [{"role": "user", "content": user}]},
{"Content-Type": "application/json", "x-api-key": self.key,
"anthropic-version": "2023-06-01"})
return "".join(b.get("text", "") for b in out["content"])
if self.provider == "gemini":
out = self._post(
f"{self.base}/models/{self.model}:generateContent?key={self.key}",
{"system_instruction": {"parts": [{"text": system}]},
"contents": [{"parts": [{"text": user}]}]},
{"Content-Type": "application/json"})
return out["candidates"][0]["content"]["parts"][0]["text"]
except Exception as e:
print(f"[LLM] call failed, using fallback: {e}")
return None
def json(self, system, user):
"""Chat + robust JSON extraction."""
txt = self.chat(system, user)
if not txt:
return None
txt = re.sub(r"^
`(json)?|`$", "", txt.strip(), flags=re.M).strip()
for pat in (r"\[.*\]", r"\{.*\}"):
m = re.search(pat, txt, re.S)
if m:
try:
return json.loads(m.group(0))
except Exception:
pass
return None
llm = LLM(CONFIG)
まずは VideoAgent のランタイム、オプションの LLM バックエンド、作業ディレクトリ、パッケージインストールフロー、そして軽量な依存関係のフォールバック設定を行います。Colab 上でもローカル環境でもスムーズにノートブックが動作するように、シェルコマンドの実行、pip によるインストール、ファイルパスの管理、環境検出のための共通ヘルパーレイヤーを用意します。また、OpenAI、DeepSeek、Anthropic、Gemini に対応した統一された LLM ラッパーを定義し、API キーがない場合は安全に決定論的な実行へフォールバックする仕組みも実装しました。
意図の定義、エージェントライブラリ、グラフプランニング
Copy CodeCopiedUse a different Browser
ビデオ編集タスクにおける意図解析、グラフ計画、ツールルーティング
VideoAgent のようなマルチエージェントシステムを構築するには、まずユーザーの指示を最小限の必要な機能セットに分解する「意図パーサー」が必要です。このシステムは、音声抽出、文字起こし、リズム検出、シーン検出、キーフレームサンプリング、キャプション生成、クロスモーダルインデックス作成、ショットプランニング、視覚的検索、トリミング、要約、質問応答、ニュース概要、ビート同期編集、結合、レンダリングといった 17 の機能を定義しています。
各エージェントは特定の役割を担います。例えば「AudioExtractor」は動画から音声トラック（wav）を抽出し、「Transcriber」は Whisper ASR を使って音声からタイムスタンプ付きの文字起こしを行います。「RhythmDetector」はエネルギーピークに基づいてビートやカットポイントを特定し、「SceneDetector」はヒストグラムの差分を用いてショット境界を検出します。さらに「KeyframeSampler」は各シーンから代表となるキーフレームを抽出し、「Captioner」はゼロショット CLIP を活用してこれらの画像にキャプションを付与します。
高度な機能として「CrossModalIndexer」が、シーン、キャプション、文字起こしデータを用いてテキストと視覚のインデックスを構築します。これにより「RetrievalAgent」がクロスモーダルコサイン検索を行い、最適なシーンを特定できます。「ShotPlanner」は指示とキャプションに基づき、全体像を把握したストーリーボードサブクエリを生成します。
編集プロセスでは、「Trimmer」が抽出されたシーンに対して ffmpeg を用いた微調整トリミングを実行し、「VideoEditor」がクリップを結合して一つの編集済み動画を作成します。音楽に合わせた編集が必要な場合、「BeatSyncEditor」がリズムポイントとシーンを基盤としてビートグリッド上にカットを配置します。
また、要約や質問応答も可能です。「Summarizer」は文字起こしを要約してリキャップを作成し、「VideoQA」は文字起こしに基づいて質問に回答します。ニュース形式の概要が必要な場合は「NewsContentGenerator」が活用されます。最後に「Renderer」が最終エンコードと正規化処理を行い、完成した動画を出力します。
これらの機能を動的に組み合わせるために、システムは意図解析を行います。ユーザーの指示が入力されると、LLM が利用可能な場合は JSON 形式で必要な機能リストを返します。例えば、「質問への回答」を求める場合、明示的に「文字起こし」や「音声抽出」が指定されていなくても、それらが暗黙的に必要とされるため、システムは自動的にこれらを含めます。
LLM が利用できない場合や、特定のキーワードに基づいて決定論的なパーサーが動作します。指示文中に「about」や「of」が含まれていればクエリとして抽出され、「what does」「who」「when」などの疑問詞が含まれていれば質問応答機能が起動し、同様に文字起こしと音声抽出も必要となります。「summarize」「overview」「tldr」などのキーワードがあれば要約機能が発動し、ニュース関連の単語が含まれればニュース概要生成も追加されます。
「beat」「rhythm」「to the music」などの言葉が含まれる場合はビート同期編集モードとなり、「highlight」「montage」「reel」などの言葉が含まれる場合はハイライト動画作成モードが起動します。いずれの場合も、必要な機能セット（T）とパラメータ（instruction, query, question）が生成され、ツールルーティング関数が対応するエージェントを特定します。
最後に、グラフ構築プロセスでは、各エージェントの入力を他のエージェントの出力に接続します。ユーザー入力（動画パス、指示、質問、クエリ）は自由に利用可能ですが、それ以外の入力は必ず何らかのエージェントによって生成される必要があります。LLM が利用可能な場合、その能力を活用して最小限のエージェントセットで構成されたグラフを設計することも可能です。
このように、VideoAgent はユーザーの意図を理解し、必要な機能を自動的に選択・接続することで、複雑なビデオ編集タスクを効率的に実行します。
本システムの中核となる計画語彙を定義します。これには、ユーザーからの入力や専門的なエージェント登録情報、ターミナルエージェント、出力プロデューサーが含まれます。自然言語で記述された指示は、文字起こし、要約、検索、レンダリング、あるいはビート同期編集といった必要な動画処理機能へと解析されます。
その後、LLM が生成した計画に基づくか、最適化プロセスによって後から修正される決定論的なターミナルエージェントのみによる計画に基づいて、ツールのルーティングと初期のエージェントグラフの構築を行います。
テキスト勾配グラフの最適化と実行
def edges_of(nodes):
E = []
for a, nd in nodes.items():
for od in nd["outputs"]:
for link in od["links"]:
for tgt, param in link.items():
E.append((a, tgt, od["name"], param))
return E
def topo_order(nodes):
E = edges_of(nodes); indeg = {a: 0 for a in nodes}; adj = defaultdict(list)
for u, v, _, _ in E:
adj[u].append(v); indeg[v] += 1
q = deque([a for a in nodes if indeg[a] == 0]); order = []
while q:
u = q.popleft(); order.append(u)
for v in adj[u]:
indeg[v] -= 1
if indeg[v] == 0: q.append(v)
return order if len(order) == len(nodes) else None
def n_components(nodes):
adj = defaultdict(set)
for u, v, _, _ in edges_of(nodes):
adj[u].add(v); adj[v].add(u)
seen = set(); comp = 0
for a in nodes:
if a in seen: continue
comp += 1; st = [a]
while st:
x = st.pop()
if x in seen: continue
seen.add(x); st += [y for y in adj[x] if y not in seen]
return comp
def unsatisfied_inputs(nodes):
incoming = defaultdict(set)
for _u, v, _op, ip in edges_of(nodes):
incoming[v].add(ip)
miss = []
for a in nodes:
for inp in AGENTS[a]["inputs"]:
if inp in USER_INPUTS or inp in incoming[a]:
continue
miss.append((a, inp))
return miss
def assess(nodes, T):
order = topo_order(nodes); tau = 1 if order is not None else 0
covered = set()
for a in nodes: covered |= set(AGENTS[a]["caps"])
kappa = len(T & covered) / max(1, len(T))
E = edges_of(nodes)
chi = (sum(1 for _u, v, op, _ip in E if op in AGENTS[v]["inputs"]) / len(E)
if E else 1.0)
comp = n_components(nodes); miss = unsatisfied_inputs(nodes)
n = max(1, len(nodes))
L_struct = 0.5 * (1 - tau) + 0.5 * ((comp - 1) / n) + 0.5 * len(miss) / n
L_align = (1 - kappa) + 0.5 * len(miss) / n
return dict(tau=tau, kappa=round(kappa, 3), chi=round(chi, 3), comp=comp,
missing=miss, L_struct=round(L_struct, 3),
L_align=round(L_align, 3), order=order)
def textual_gradient(nodes, T):
"""Emit NL graph edits and apply them (insert producers / cover intents)."""
selected = set(nodes); grads = []
covered = set()
for a in nodes: covered |= set(AGENTS[a]["caps"])
for i in sorted(T - covered):
if i in TERMINALS and TERMINALS[i] not in selected:
selected.add(TERMINALS[i])
grads.append(f"insert {TERMINALS[i]} to cover uncovered intent {i}")
changed = True
while changed:
changed = False
for a, inp in unsatisfied_inputs(build_graph(selected)):
if inp in PRODUCER and PRODUCER[inp] not in selected:
selected.add(PRODUCER[inp]); changed = True
grads.append(f"insert {PRODUCER[inp]} -> supplies {inp} "
f"needed by {a} (satisfy data-flow)")
new = build_graph(selected)
if topo_order(new) is None:
grads.append("remove back-edge(s) to restore acyclicity (τ→1)")
return new, grads
def optimize_graph(nodes, T, Tmax=4, verbose=True):
history = []
for t in range(Tmax):
Q = assess(nodes, T)
if verbose:
print(f" round {t}: τ={Q['tau']} κ={Q['kappa']} χ={Q['chi']} "
f"components={Q['comp']} unmet_inputs={len(Q['missing'])} "
f"| L_struct={Q['L_struct']} L_align={Q['L_align']}")
history.append((t, Q))
if Q["L_struct"] == 0 and Q["L_align"] == 0:
if verbose: print(f" ✓ converged at round {t} "
f"(L_struct=L_align=0)")
return nodes, history
nodes, grads = textual_gradient(nodes, T)
if verbose:
for g in grads[:8]:
print(f" ∇_G {g}")
Q = assess(nodes, T)
if verbose:
print(f" round {Tmax}: τ={Q['tau']} κ={Q['kappa']} χ={Q['chi']} "
f"(stopped at Tmax)")
history.append((Tmax, Q))
return nodes, history
def execute_graph(nodes, seed, verbose=True):
order = topo_order(nodes)
assert order is not None, "cannot execute a cyclic graph"
bb = dict(seed)
for a in order:
kwargs = {inp: bb[inp] for inp in AGENTS[a]["inputs"] if inp in bb}
if verbose:
print(f" 
image {a}({', '.join(kwargs) or '—'})")
try:
out = AGENTS[a]"fn"
except Exception as e:
out = {"__error__": f"{a} failed: {e}"}
print(f" ! {a} error: {e}")
if not isinstance(out, dict):
out = {AGENTS[a]["outputs"][0]: out}
for o in AGENTS[a]["outputs"]:
if o in out: bb[o] = out[o]
if verbose:
for o in AGENTS[a]["outputs"]:
v = bb.get(o)
prev = (v if isinstance(v, str) else json.dumps(v)[:80]
if v is not None else "∅")
print(f" → {o}: {str(prev)[:80]}")
return bb, order
グラフの解析と最適化ロジックを実装し、エッジの有無、トポロジ順序、連結成分、入力不足、意図のカバレッジ、互換性をチェックします。システムが現在のグラフを実行可能かどうかを測定できるよう、τ（タウ）、κ（カッパ）、χ（カイ）を用いて構造的損失と整合性損失を計算しています。
原文を表示
In this tutorial, we build a runnable reconstruction of the VideoAgent workflow, focusing on the core agentic pipeline behind video understanding, retrieval, editing, and remaking. We start by configuring a lightweight environment that works without API keys. We define an intent parser, an agent library, a tool router, a graph planner, and a textual-gradient optimizer that repairs missing dependencies in the execution graph. We also connect these planning components to practical video-processing tools, including FFmpeg, Whisper-based transcription, scene detection, keyframe sampling, captioning, cross-modal indexing, retrieval, trimming, beat-synced editing, and final rendering. By the end of the tutorial, we have a complete multi-agent video system that can answer questions about a video, summarize its content, generate a news-style overview, and produce edited video artifacts from natural-language instructions.
Configuring the VideoAgent Runtime and Multi-Provider LLM Wrapper
Copy CodeCopiedUse a different Browser
CONFIG = {
"provider": "",
"api_key": "",
"base_url": "",
"model": "",
"max_shots": 4,
"opt_rounds": 4,
"run_demos": ["qa", "overview", "highlight", "beatsync"],
}
import os, sys, subprocess, json, math, re, wave, textwrap, shutil, time
from collections import defaultdict, deque
def _sh(cmd):
return subprocess.run(cmd, capture_output=True, text=True)
def _pip(pkgs):
_sh([sys.executable, "-m", "pip", "install", "-q", *pkgs])
IN_COLAB = "google.colab" in sys.modules
if IN_COLAB or os.environ.get("VA_INSTALL", "1") == "1":
for spec in (["openai-whisper"], ["gTTS"], ["sentence-transformers"]):
try:
_pip(spec)
except Exception as e:
print(f"[install] {spec} failed ({e}); a fallback will be used.")
try:
import numpy as np
except Exception:
_pip(["numpy"]); import numpy as np
try:
from PIL import Image, ImageDraw, ImageFont
except Exception:
_pip(["pillow"]); from PIL import Image, ImageDraw, ImageFont
HAS_FFMPEG = shutil.which("ffmpeg") is not None
WORK = "/content/va_workdir" if IN_COLAB else os.path.abspath("./va_workdir")
os.makedirs(WORK, exist_ok=True)
def wp(*a): return os.path.join(WORK, *a)
import urllib.request, urllib.error
class LLM:
DEFAULT = {
"openai": "gpt-4o-mini", "deepseek": "deepseek-chat",
"anthropic": "claude-3-5-sonnet-latest", "gemini": "gemini-1.5-flash",
}
def __init__(self, cfg):
self.provider = (cfg.get("provider") or "").lower().strip()
self.key = (cfg.get("api_key") or
os.environ.get(f"{self.provider.upper()}_API_KEY", "") or
os.environ.get("OPENAI_API_KEY", "") if self.provider else "")
self.model = cfg.get("model") or self.DEFAULT.get(self.provider, "")
self.base = cfg.get("base_url") or {
"openai": "https://api.openai.com/v1",
"deepseek": "https://api.deepseek.com/v1",
"anthropic": "https://api.anthropic.com/v1",
"gemini": "https://generativelanguage.googleapis.com/v1beta",
}.get(self.provider, "")
def available(self):
return bool(self.provider and self.key)
def _post(self, url, payload, headers):
data = json.dumps(payload).encode()
req = urllib.request.Request(url, data=data, headers=headers, method="POST")
with urllib.request.urlopen(req, timeout=90) as r:
return json.loads(r.read().decode())
def chat(self, system, user, temperature=0.2):
"""Return assistant text, or None on any failure (caller falls back)."""
if not self.available():
return None
try:
if self.provider in ("openai", "deepseek"):
out = self._post(
f"{self.base}/chat/completions",
{"model": self.model, "temperature": temperature,
"messages": [{"role": "system", "content": system},
{"role": "user", "content": user}]},
{"Content-Type": "application/json",
"Authorization": f"Bearer {self.key}"})
return out["choices"][0]["message"]["content"]
if self.provider == "anthropic":
out = self._post(
f"{self.base}/messages",
{"model": self.model, "max_tokens": 2000,
"system": system,
"messages": [{"role": "user", "content": user}]},
{"Content-Type": "application/json", "x-api-key": self.key,
"anthropic-version": "2023-06-01"})
return "".join(b.get("text", "") for b in out["content"])
if self.provider == "gemini":
out = self._post(
f"{self.base}/models/{self.model}:generateContent?key={self.key}",
{"system_instruction": {"parts": [{"text": system}]},
"contents": [{"parts": [{"text": user}]}]},
{"Content-Type": "application/json"})
return out["candidates"][0]["content"]["parts"][0]["text"]
except Exception as e:
print(f"[LLM] call failed, using fallback: {e}")
return None
def json(self, system, user):
"""Chat + robust JSON extraction."""
txt = self.chat(system, user)
if not txt:
return None
txt = re.sub(r"^`(json)?|`$", "", txt.strip(), flags=re.M).strip()
for pat in (r"\[.*\]", r"\{.*\}"):
m = re.search(pat, txt, re.S)
if m:
try:
return json.loads(m.group(0))
except Exception:
pass
return None
llm = LLM(CONFIG)
We begin by configuring the VideoAgent runtime, the optional LLM backend, the working directory, the package installation flow, and lightweight dependency fallbacks. We create a shared helper layer for shell commands, pip installs, file paths, and environment detection so the notebook runs smoothly in Colab or locally. We also define a unified LLM wrapper that supports OpenAI, DeepSeek, Anthropic, and Gemini while safely falling back to deterministic execution when no API key is available.
Defining Intents, the Agent Library, and Graph Planning
Copy CodeCopiedUse a different Browser
INTENTS = [
"audio_extraction", "transcription", "rhythm_detection", "scene_detection",
"keyframe_sampling", "captioning", "cross_modal_indexing", "shot_planning",
"visual_retrieval", "trimming", "summarization", "question_answering",
"news_overview", "beat_sync_edit", "concatenation", "rendering",
]
USER_INPUTS = {"video_path", "instruction", "question", "query"}
AGENTS = {
"AudioExtractor": dict(desc="Extract the audio track (wav) from a video.",
inputs=["video_path"], outputs=["audio_path"], caps=["audio_extraction"]),
"Transcriber": dict(desc="Whisper ASR: audio -> time-stamped transcript.",
inputs=["audio_path"], outputs=["transcript"], caps=["transcription"]),
"RhythmDetector": dict(desc="Energy-peak beat / cut-point detector.",
inputs=["audio_path"], outputs=["rhythm_points"], caps=["rhythm_detection"]),
"SceneDetector": dict(desc="Shot-boundary detection via histogram deltas.",
inputs=["video_path"], outputs=["scenes"], caps=["scene_detection"]),
"KeyframeSampler": dict(desc="Sample one representative keyframe per scene.",
inputs=["video_path", "scenes"], outputs=["keyframes"], caps=["keyframe_sampling"]),
"Captioner": dict(desc="Zero-shot CLIP captioning of keyframes.",
inputs=["keyframes"], outputs=["captions"], caps=["captioning"]),
"CrossModalIndexer": dict(desc="Build a CLIP text-visual index over scenes.",
inputs=["keyframes", "captions", "transcript"], outputs=["index"],
caps=["cross_modal_indexing"]),
"ShotPlanner": dict(desc="Global-aware storyboard sub-query generation.",
inputs=["instruction", "captions"], outputs=["storyboards"], caps=["shot_planning"]),
"RetrievalAgent": dict(desc="Cross-modal cosine retrieval of best scenes.",
inputs=["index", "storyboards"], outputs=["retrieved"], caps=["visual_retrieval"]),
"Trimmer": dict(desc="Fine-grained ffmpeg trimming of retrieved scenes.",
inputs=["retrieved", "video_path"], outputs=["clips"], caps=["trimming"]),
"VideoEditor": dict(desc="Concatenate clips into one edited video.",
inputs=["clips"], outputs=["edited_video"], caps=["concatenation"]),
"BeatSyncEditor": dict(desc="Assemble scene cuts onto the beat grid.",
inputs=["rhythm_points", "scenes", "video_path"], outputs=["edited_video"],
caps=["beat_sync_edit"]),
"Summarizer": dict(desc="Summarize the transcript into a recap.",
inputs=["transcript"], outputs=["summary"], caps=["summarization"]),
"VideoQA": dict(desc="Answer a question grounded in the transcript.",
inputs=["transcript", "question"], outputs=["answer"], caps=["question_answering"]),
"NewsContentGenerator": dict(desc="Write a news-style overview from transcript.",
inputs=["transcript", "instruction"], outputs=["overview"], caps=["news_overview"]),
"Renderer": dict(desc="Final encode/normalize pass -> final video.",
inputs=["edited_video"], outputs=["final_video"], caps=["rendering"]),
}
TERMINALS = {
"question_answering": "VideoQA", "summarization": "Summarizer",
"news_overview": "NewsContentGenerator", "visual_retrieval": "RetrievalAgent",
"beat_sync_edit": "BeatSyncEditor", "rendering": "Renderer",
}
PRODUCER = {}
for _a, _m in AGENTS.items():
for _o in _m["outputs"]:
PRODUCER.setdefault(_o, _a)
INTENT_SYS = (
"You are VideoAgent's Intent Parser. Decompose the user instruction into "
"the minimal set of required capabilities, choosing ONLY from this list: "
+ ", ".join(INTENTS) + ". Include BOTH explicit and necessary implicit "
"intents (e.g. answering a question implies transcription+audio_extraction). "
'Respond as JSON: {"intents":[...], "query":"<subject to retrieve or empty>", '
'"question":"<the question or empty>"}.')
def analyze_intents(instruction):
"""LLM intent parse if a key is set, else a faithful deterministic parser."""
if llm.available():
out = llm.json(INTENT_SYS, f"Instruction: {instruction}")
if isinstance(out, dict) and out.get("intents"):
T = {i for i in out["intents"] if i in INTENTS}
params = {"instruction": instruction,
"query": out.get("query", "") or "",
"question": out.get("question", "") or instruction}
if T:
return T, params
s = instruction.lower(); T = set()
params = {"instruction": instruction, "question": instruction, "query": ""}
m = re.search(r"about ([^.,;!?]+)", s) or re.search(r"of ([^.,;!?]+)", s)
if m: params["query"] = m.group(1).strip()
if any(w in s for w in ["?", "what does", "who ", "when ", "why ", "how ", "question"]):
T |= {"question_answering", "transcription", "audio_extraction"}
if any(w in s for w in ["summar", "overview", "recap", "tldr", "tl;dr", "digest"]):
T |= {"summarization", "transcription", "audio_extraction"}
if "news" in s or "overview" in s: T |= {"news_overview"}
is_beat = any(w in s for w in ["beat", "rhythm", "to the music",
"music video", "tempo", "synced", "sync "])
is_highlight = any(w in s for w in ["highlight", "montage", "supercut",
"reel", "best parts", "clips about",
"compile"])
if is_beat:
T |= {"beat_sync_edit", "rhythm_detection", "scene_detection",
"audio_extraction", "rendering"}
elif is_highlight:
T |= {"visual_retrieval", "cross_modal_indexing", "scene_detection",
"keyframe_sampling", "captioning", "shot_planning", "trimming",
"concatenation", "rendering", "audio_extraction", "transcription"}
if not T:
T |= {"summarization", "transcription", "audio_extraction"}
return T, params
def route_tools(T):
return {a for a, m in AGENTS.items() if set(m["caps"]) & T}
def build_graph(selected):
"""Wire each input to a producing agent's output (by param name)."""
nodes = {a: {"node": a,
"inputs": [{"name": x} for x in AGENTS[a]["inputs"]],
"outputs": [{"name": o, "links": []} for o in AGENTS[a]["outputs"]]}
for a in selected}
avail = {}
for a in selected:
for o in AGENTS[a]["outputs"]:
avail.setdefault(o, a)
for a in selected:
for inp in AGENTS[a]["inputs"]:
if inp in avail and avail[inp] != a:
for od in nodes[avail[inp]]["outputs"]:
if od["name"] == inp:
od["links"].append({a: inp})
return nodes
def naive_plan(T):
sel = {TERMINALS[i] for i in T if i in TERMINALS} or route_tools(T)
return build_graph(sel)
def llm_plan(T, instruction):
"""Optional: let the LLM draft the agent graph (Listing 5)."""
if not llm.available():
return None
lib = "\n".join(f'- {a}: {m["desc"]} | inputs={m["inputs"]} '
f'outputs={m["outputs"]}' for a, m in AGENTS.items())
sys_p = ("You are VideoAgent's Agent-Graph Designer. Using ONLY the agents "
"below, output a JSON list of nodes: "
'[{"node":NAME,"selected_inputs":[...],"selected_outputs":[...]}]. '
"Pick the minimal agents that satisfy the required intents and make "
"sure every non-user input is produced by some other selected agent.\n"
"USER INPUTS available for free: video_path, instruction, question, query.\n"
f"AGENT LIBRARY:\n{lib}")
usr = f"Instruction: {instruction}\nRequired intents: {sorted(T)}"
out = llm.json(sys_p, usr)
if isinstance(out, list) and out:
sel = {n.get("node") for n in out if n.get("node") in AGENTS}
if sel:
return build_graph(sel)
return None
We define the full intent space, user inputs, specialized agent registry, terminal agents, and output producers that constitute the system’s core planning vocabulary. We parse each natural-language instruction into required video capabilities such as transcription, summarization, retrieval, rendering, or beat-synced editing. We then route tools and construct an initial agent graph, either through an LLM-generated plan or a deterministic terminal-only plan that the optimizer later repairs.
Textual-Gradient Graph Optimization and Execution
Copy CodeCopiedUse a different Browser
def edges_of(nodes):
E = []
for a, nd in nodes.items():
for od in nd["outputs"]:
for link in od["links"]:
for tgt, param in link.items():
E.append((a, tgt, od["name"], param))
return E
def topo_order(nodes):
E = edges_of(nodes); indeg = {a: 0 for a in nodes}; adj = defaultdict(list)
for u, v, _, _ in E:
adj[u].append(v); indeg[v] += 1
q = deque([a for a in nodes if indeg[a] == 0]); order = []
while q:
u = q.popleft(); order.append(u)
for v in adj[u]:
indeg[v] -= 1
if indeg[v] == 0: q.append(v)
return order if len(order) == len(nodes) else None
def n_components(nodes):
adj = defaultdict(set)
for u, v, _, _ in edges_of(nodes):
adj[u].add(v); adj[v].add(u)
seen = set(); comp = 0
for a in nodes:
if a in seen: continue
comp += 1; st = [a]
while st:
x = st.pop()
if x in seen: continue
seen.add(x); st += [y for y in adj[x] if y not in seen]
return comp
def unsatisfied_inputs(nodes):
incoming = defaultdict(set)
for _u, v, _op, ip in edges_of(nodes):
incoming[v].add(ip)
miss = []
for a in nodes:
for inp in AGENTS[a]["inputs"]:
if inp in USER_INPUTS or inp in incoming[a]:
continue
miss.append((a, inp))
return miss
def assess(nodes, T):
order = topo_order(nodes); tau = 1 if order is not None else 0
covered = set()
for a in nodes: covered |= set(AGENTS[a]["caps"])
kappa = len(T & covered) / max(1, len(T))
E = edges_of(nodes)
chi = (sum(1 for _u, v, op, _ip in E if op in AGENTS[v]["inputs"]) / len(E)
if E else 1.0)
comp = n_components(nodes); miss = unsatisfied_inputs(nodes)
n = max(1, len(nodes))
L_struct = 0.5 * (1 - tau) + 0.5 * ((comp - 1) / n) + 0.5 * len(miss) / n
L_align = (1 - kappa) + 0.5 * len(miss) / n
return dict(tau=tau, kappa=round(kappa, 3), chi=round(chi, 3), comp=comp,
missing=miss, L_struct=round(L_struct, 3),
L_align=round(L_align, 3), order=order)
def textual_gradient(nodes, T):
"""Emit NL graph edits and apply them (insert producers / cover intents)."""
selected = set(nodes); grads = []
covered = set()
for a in nodes: covered |= set(AGENTS[a]["caps"])
for i in sorted(T - covered):
if i in TERMINALS and TERMINALS[i] not in selected:
selected.add(TERMINALS[i])
grads.append(f"insert {TERMINALS[i]} to cover uncovered intent {i}")
changed = True
while changed:
changed = False
for a, inp in unsatisfied_inputs(build_graph(selected)):
if inp in PRODUCER and PRODUCER[inp] not in selected:
selected.add(PRODUCER[inp]); changed = True
grads.append(f"insert {PRODUCER[inp]} -> supplies {inp} "
f"needed by {a} (satisfy data-flow)")
new = build_graph(selected)
if topo_order(new) is None:
grads.append("remove back-edge(s) to restore acyclicity (τ→1)")
return new, grads
def optimize_graph(nodes, T, Tmax=4, verbose=True):
history = []
for t in range(Tmax):
Q = assess(nodes, T)
if verbose:
print(f" round {t}: τ={Q['tau']} κ={Q['kappa']} χ={Q['chi']} "
f"components={Q['comp']} unmet_inputs={len(Q['missing'])} "
f"| L_struct={Q['L_struct']} L_align={Q['L_align']}")
history.append((t, Q))
if Q["L_struct"] == 0 and Q["L_align"] == 0:
if verbose: print(f" ✓ converged at round {t} "
f"(L_struct=L_align=0)")
return nodes, history
nodes, grads = textual_gradient(nodes, T)
if verbose:
for g in grads[:8]:
print(f" ∇_G {g}")
Q = assess(nodes, T)
if verbose:
print(f" round {Tmax}: τ={Q['tau']} κ={Q['kappa']} χ={Q['chi']} "
f"(stopped at Tmax)")
history.append((Tmax, Q))
return nodes, history
def execute_graph(nodes, seed, verbose=True):
order = topo_order(nodes)
assert order is not None, "cannot execute a cyclic graph"
bb = dict(seed)
for a in order:
kwargs = {inp: bb[inp] for inp in AGENTS[a]["inputs"] if inp in bb}
if verbose:
print(f" image {a}({', '.join(kwargs) or '—'})")
try:
out = AGENTS[a]"fn"
except Exception as e:
out = {"__error__": f"{a} failed: {e}"}
print(f" ! {a} error: {e}")
if not isinstance(out, dict):
out = {AGENTS[a]["outputs"][0]: out}
for o in AGENTS[a]["outputs"]:
if o in out: bb[o] = out[o]
if verbose:
for o in AGENTS[a]["outputs"]:
v = bb.get(o)
prev = (v if isinstance(v, str) else json.dumps(v)[:80]
if v is not None else "∅")
print(f" → {o}: {str(prev)[:80]}")
return bb, order
We implement the graph analysis and optimization logic that checks for edges, topological order, connected components, missing inputs, intent coverage, and compatibility. We compute the structural and alignment losses using τ, κ, and χ so the system can measure whether the current graph is executable and
関連記事
Google、ブラウザ上で WebGPU を経由して .tflite モデルを実行する JavaScript バインディング「LiteRT.js」をリリース
PrismML が Qwen3.6-27B の軽量版「Bonsai 27B」をリリース：ラップトップやスマートフォンで動作する 1 ビットおよび 3 値モデル
コード生成エージェント4種比較：Mistral Vibe、Claude Code、Cursor、Codexの実務タスク評価
今日のまとめ
AI日報で今日の重要ニュースをまとめ読み