Building a VideoAgent-Style Multi-Agent System: Intent Parsing, Graph Planning, and Tool Routing for Video Editing Tasks

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":"", ' '"question":""}.') 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

[truncated for AI cost control]