构建视频编辑多智能体系统：意图解析与图规划教程

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 BrowserCONFIG = { "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 BrowserINTENTS = [ "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", "i

原文超出正文长度上限，此处截断——上游还有内容，完整版见上方「原文 ↗」。