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

CONFIG = {
   "supplier":  "",
   "api_key":   "",
   "base_url":  "",
   "mannequin":     "",
   "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, textual content=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"]):
       attempt:
           _pip(spec)
       besides Exception as e:
           print(f"[install] {spec} failed ({e}); a fallback shall be used.")
attempt:
   import numpy as np
besides Exception:
   _pip(["numpy"]); import numpy as np
attempt:
   from PIL import Image, ImageDraw, ImageFont
besides Exception:
   _pip(["pillow"]); from PIL import Image, ImageDraw, ImageFont
HAS_FFMPEG = shutil.which("ffmpeg") will not be None
WORK = "/content material/va_workdir" if IN_COLAB else os.path.abspath("./va_workdir")
os.makedirs(WORK, exist_ok=True)
def wp(*a): return os.path.be part of(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.supplier = (cfg.get("supplier") or "").decrease().strip()
       self.key = (cfg.get("api_key") or
                   os.environ.get(f"{self.supplier.higher()}_API_KEY", "") or
                   os.environ.get("OPENAI_API_KEY", "") if self.supplier else "")
       self.mannequin = cfg.get("mannequin") or self.DEFAULT.get(self.supplier, "")
       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.supplier, "")
   def obtainable(self):
       return bool(self.supplier and self.key)
   def _post(self, url, payload, headers):
       information = json.dumps(payload).encode()
       req = urllib.request.Request(url, information=information, headers=headers, methodology="POST")
       with urllib.request.urlopen(req, timeout=90) as r:
           return json.hundreds(r.learn().decode())
   def chat(self, system, consumer, temperature=0.2):
       """Return assistant textual content, or None on any failure (caller falls again)."""
       if not self.obtainable():
           return None
       attempt:
           if self.supplier in ("openai", "deepseek"):
               out = self._post(
                   f"{self.base}/chat/completions",
                   {"mannequin": self.mannequin, "temperature": temperature,
                    "messages": [{"role": "system", "content": system},
                                 {"role": "user", "content": user}]},
                   {"Content-Type": "software/json",
                    "Authorization": f"Bearer {self.key}"})
               return out["choices"][0]["message"]["content"]
           if self.supplier == "anthropic":
               out = self._post(
                   f"{self.base}/messages",
                   {"mannequin": self.mannequin, "max_tokens": 2000,
                    "system": system,
                    "messages": [{"role": "user", "content": user}]},
                   {"Content-Type": "software/json", "x-api-key": self.key,
                    "anthropic-version": "2023-06-01"})
               return "".be part of(b.get("textual content", "") for b in out["content"])
           if self.supplier == "gemini":
               out = self._post(
                   f"{self.base}/fashions/{self.mannequin}:generateContent?key={self.key}",
                   {"system_instruction": {"components": [{"text": system}]},
                    "contents": [{"parts": [{"text": user}]}]},
                   {"Content-Type": "software/json"})
               return out["candidates"][0]["content"]["parts"][0]["text"]
       besides Exception as e:
           print(f"[LLM] name failed, utilizing fallback: {e}")
       return None
   def json(self, system, consumer):
       """Chat + strong JSON extraction."""
       txt = self.chat(system, consumer)
       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:
               attempt:
                   return json.hundreds(m.group(0))
               besides Exception:
                   go
       return None
llm = LLM(CONFIG)

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", "query", "question"}
AGENTS = {
"AudioExtractor":     dict(desc="Extract the audio observe (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 by way of histogram deltas.",
   inputs=["video_path"], outputs=["scenes"], caps=["scene_detection"]),
"KeyframeSampler":    dict(desc="Sample one consultant 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 technology.",
   inputs=["instruction", "captions"], outputs=["storyboards"], caps=["shot_planning"]),
"RetrievalAgent":     dict(desc="Cross-modal cosine retrieval of finest 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 query grounded within the transcript.",
   inputs=["transcript", "question"], outputs=["answer"], caps=["question_answering"]),
"InformationContentGenerator": dict(desc="Write a news-style overview from transcript.",
   inputs=["transcript", "instruction"], outputs=["overview"], caps=["news_overview"]),
"Renderer":           dict(desc="Final encode/normalize go -> ultimate video.",
   inputs=["edited_video"], outputs=["final_video"], caps=["rendering"]),
}
TERMINALS = {
   "question_answering": "VideoQA", "summarization": "Summarizer",
   "news_overview": "InformationContentGenerator", "visual_retrieval": "RetrievalAgent",
   "beat_sync_edit": "BeatSyncEditor", "rendering": "Renderer",
}
PRODUCER = {}
for _a, _m in AGENTS.gadgets():
   for _o in _m["outputs"]:
       PRODUCER.setdefault(_o, _a)
INTENT_SYS = (
   "You are VideoAgent's Intent Parser. Decompose the consumer instruction into "
   "the minimal set of required capabilities, selecting ONLY from this listing: "
   + ", ".be part of(INTENTS) + ". Include BOTH express and needed implicit "
   "intents (e.g. answering a query implies transcription+audio_extraction). "
   'Respond as JSON: {"intents":[...], "question":"<topic to retrieve or empty>", '
   '"query":"<the query or empty>"}.')
def analyze_intents(instruction):
   """LLM intent parse if a secret is set, else a devoted deterministic parser."""
   if llm.obtainable():
       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,
                     "question": out.get("question", "") or "",
                     "query": out.get("query", "") or instruction}
           if T:
               return T, params
   s = instruction.decrease(); T = set()
   params = {"instruction": instruction, "query": instruction, "question": ""}
   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 "information" 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.gadgets() if set(m["caps"]) & T}
def build_graph(chosen):
   """Wire every enter to a producing agent's output (by param title)."""
   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 chosen}
   avail = {}
   for a in chosen:
       for o in AGENTS[a]["outputs"]:
           avail.setdefault(o, a)
   for a in chosen:
       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.obtainable():
       return None
   lib = "n".be part of(f'- {a}: {m["desc"]} | inputs={m["inputs"]} '
                   f'outputs={m["outputs"]}' for a, m in AGENTS.gadgets())
   sys_p = ("You are VideoAgent's Agent-Graph Designer. Using ONLY the brokers "
            "beneath, output a JSON listing of nodes: "
            '[{"node":NAME,"selected_inputs":[...],"selected_outputs":[...]}]. '
            "Pick the minimal brokers that fulfill the required intents and make "
            "certain each non-user enter is produced by another chosen agent.n"
            "USER INPUTS obtainable for free: video_path, instruction, query, question.n"
            f"AGENT LIBRARY:n{lib}")
   usr = f"Instruction: {instruction}nRequired intents: {sorted(T)}"
   out = llm.json(sys_p, usr)
   if isinstance(out, listing) and out:
       sel = {n.get("node") for n in out if n.get("node") in AGENTS}
       if sel:
           return build_graph(sel)
   return None

def edges_of(nodes):
   E = []
   for a, nd in nodes.gadgets():
       for od in nd["outputs"]:
           for hyperlink in od["links"]:
               for tgt, param in hyperlink.gadgets():
                   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(listing)
   for u, v, _, _ in E:
       adj[u].append(v); indeg[v] += 1
   q = deque([a for a in nodes if indeg[a] == 0]); order = []
   whereas 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: proceed
       comp += 1; st = [a]
       whereas st:
           x = st.pop()
           if x in seen: proceed
           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]:
               proceed
           miss.append((a, inp))
   return miss
def assess(nodes, T):
   order = topo_order(nodes); tau = 1 if order will not be None else 0
   coated = set()
   for a in nodes: coated |= set(AGENTS[a]["caps"])
   kappa = len(T & coated) / 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=spherical(kappa, 3), chi=spherical(chi, 3), comp=comp,
               lacking=miss, L_struct=spherical(L_struct, 3),
               L_align=spherical(L_align, 3), order=order)
def textual_gradient(nodes, T):
   """Emit NL graph edits and apply them (insert producers / cowl intents)."""
   chosen = set(nodes); grads = []
   coated = set()
   for a in nodes: coated |= set(AGENTS[a]["caps"])
   for i in sorted(T - coated):
       if i in TERMINALS and TERMINALS[i] not in chosen:
           chosen.add(TERMINALS[i])
           grads.append(f"insert `{TERMINALS[i]}` to cowl uncovered intent `{i}`")
   modified = True
   whereas modified:
       modified = False
       for a, inp in unsatisfied_inputs(build_graph(chosen)):
           if inp in PRODUCER and PRODUCER[inp] not in chosen:
               chosen.add(PRODUCER[inp]); modified = True
               grads.append(f"insert `{PRODUCER[inp]}` -> provides `{inp}` "
                            f"wanted by `{a}` (fulfill data-flow)")
   new = build_graph(chosen)
   if topo_order(new) is None:
       grads.append("take away back-edge(s) to revive acyclicity (τ→1)")
   return new, grads
def optimize_graph(nodes, T, Tmax=4, verbose=True):
   historical past = []
   for t in vary(Tmax):
       Q = assess(nodes, T)
       if verbose:
           print(f"    spherical {t}: τ={Q['tau']} κ={Q['kappa']} χ={Q['chi']} "
                 f"parts={Q['comp']} unmet_inputs={len(Q['missing'])} "
                 f"| L_struct={Q['L_struct']} L_align={Q['L_align']}")
       historical past.append((t, Q))
       if Q["L_struct"] == 0 and Q["L_align"] == 0:
           if verbose: print(f"    ✓ converged at spherical {t} "
                             f"(L_struct=L_align=0)")
           return nodes, historical past
       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"    spherical {Tmax}: τ={Q['tau']} κ={Q['kappa']} χ={Q['chi']} "
             f"(stopped at Tmax)")
   historical past.append((Tmax, Q))
   return nodes, historical past
def execute_graph(nodes, seed, verbose=True):
   order = topo_order(nodes)
   assert order will not be None, "can not execute a cyclic graph"
   bb = dict(seed)
   for a so as:
       kwargs = {inp: bb[inp] for inp in AGENTS[a]["inputs"] if inp in bb}
       if verbose:
           print(f"     {a}({', '.be part of(kwargs) or '—'})")
       attempt:
           out = AGENTS[a]["fn"](**kwargs)
       besides 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 will not be None else "∅")
               print(f"        → {o}: {str(prev)[:80]}")
   return bb, order

_ff = "ffmpeg"; _fp = "ffprobe"
def ff_probe(path):
   r = _sh([_fp, "-v", "error", "-select_streams", "v:0", "-show_entries",
            "stream=r_frame_rate,width,height,nb_frames", "-show_entries",
            "format=duration", "-of", "json", path])
   attempt:
       j = json.hundreds(r.stdout)
       st = j["streams"][0]; num, den = st["r_frame_rate"].cut up("/")
       return dict(dur=float(j["format"]["duration"]),
                   fps=float(num) / float(den),
                   w=int(st["width"]), h=int(st["height"]))
   besides Exception:
       return dict(dur=0.0, fps=24.0, w=0, h=0)
def tool_audio_extractor(video_path):
   out = wp("audio.wav")
   _sh([_ff, "-y", "-i", video_path, "-vn", "-acodec", "pcm_s16le",
        "-ar", "16000", "-ac", "1", "-loglevel", "error", out])
   return {"audio_path": out if os.path.exists(out) else ""}
_whisper_model = None
def _load_whisper():
   world _whisper_model
   if _whisper_model will not be None: return _whisper_model
   attempt:
       import whisper
       _whisper_model = whisper.load_model("base")
   besides Exception as e:
       print(f"      [Transcriber] Whisper unavailable ({e}); "
             f"utilizing injected script fallback.")
       _whisper_model = False
   return _whisper_model
FALLBACK_SCRIPT = None
def tool_transcriber(audio_path):
   m = _load_whisper()
   if m and audio_path and os.path.exists(audio_path):
       attempt:
           res = m.transcribe(audio_path, fp16=False)
           segs = [(float(s["start"]), float(s["end"]), s["text"].strip())
                   for s in res.get("segments", [])]
           full = res.get("textual content", "").strip()
           if segs:
               return {"transcript": {"textual content": full, "segments": segs}}
       besides Exception as e:
           print(f"      [Transcriber] error {e}; falling again.")
   if FALLBACK_SCRIPT:
       full = " ".be part of(t for _s, _e, t in FALLBACK_SCRIPT)
       return {"transcript": {"textual content": full, "segments": listing(FALLBACK_SCRIPT)}}
   return {"transcript": {"textual content": "", "segments": []}}
def _read_wav_energy(path, hop=0.05):
   """RMS vitality envelope from a wav utilizing solely the stdlib `wave` module."""
   attempt:
       w = wave.open(path, "rb")
       sr = w.getframerate(); n = w.getnframes()
       uncooked = w.readframes(n); w.shut()
       x = np.frombuffer(uncooked, dtype=np.int16).astype(np.float32)
       step = max(1, int(sr * hop)); env = []
       for i in vary(0, len(x) - step, step):
           env.append(float(np.sqrt(np.imply(x[i:i + step] ** 2))))
       env = np.array(env) / (np.max(env) + 1e-9)
       return env, hop
   besides Exception:
       return np.array([]), hop
def tool_rhythm_detector(audio_path):
   env, hop = _read_wav_energy(audio_path)
   if len(env) < 3:
       pts = [round(t, 2) for t in np.arange(0.5, 6.0, 0.8)]
       return {"rhythm_points": pts}
   thr = float(np.imply(env) + 0.6 * np.std(env))
   beats = []
   for i in vary(1, len(env) - 1):
       if env[i] > thr and env[i] >= env[i - 1] and env[i] > env[i + 1]:
           t = i * hop
           if not beats or t - beats[-1] > 0.25:
               beats.append(spherical(t, 2))
   if len(beats) < 2:
       beats = [round(t, 2) for t in np.arange(0.5, len(env) * hop, 0.8)]
   return {"rhythm_points": beats}
def _sample_grid(path, fps=2, measurement=(64, 36)):
   d = wp("sf"); os.makedirs(d, exist_ok=True)
   for f in os.listdir(d): os.take away(os.path.be part of(d, f))
   _sh([_ff, "-y", "-i", path, "-vf",
        f"fps={fps},scale={size[0]}:{measurement[1]}", "-loglevel", "error",
        os.path.be part of(d, "spercent04d.png")])
   fs = sorted(os.listdir(d))
   arr = [np.asarray(Image.open(os.path.join(d, f)).convert("RGB")) for f in fs]
   return arr, fps
def _hist(a):
   h = [np.histogram(a[:, :, c], bins=16, vary=(0, 255))[0] for c in vary(3)]
   v = np.concatenate(h).astype(float); return v / (v.sum() + 1e-9)
def tool_scene_detector(video_path):
   frames, fps = _sample_grid(video_path, fps=2)
   meta = ff_probe(video_path); dur = meta["dur"] or (len(frames) / fps)
   if len(frames) < 2:
       return {"scenes": [{"id": 0, "start": 0.0, "end": round(dur, 2)}]}
   diffs = [float(np.abs(_hist(frames[i]) - _hist(frames[i - 1])).sum())
            for i in vary(1, len(frames))]
   thr = float(np.imply(diffs) + 1.0 * np.std(diffs))
   begins = [0.0] + [(i + 1) / fps for i, d in enumerate(diffs) if d > thr]
   begins = sorted(set(spherical(s, 2) for s in begins))
   scenes = []
   for okay, s in enumerate(begins):
       e = begins[k + 1] if okay + 1 < len(begins) else spherical(dur, 2)
       if e - s >= 0.4:
           scenes.append({"id": len(scenes), "begin": s, "finish": spherical(e, 2)})
   if not scenes:
       scenes = [{"id": 0, "start": 0.0, "end": round(dur, 2)}]
   return {"scenes": scenes}
def tool_keyframe_sampler(video_path, scenes):
   d = wp("keyframes"); os.makedirs(d, exist_ok=True)
   kfs = []
   for sc in scenes:
       mid = (sc["start"] + sc["end"]) / 2.0
       out = os.path.be part of(d, f"kf_{sc['id']:03d}.png")
       _sh([_ff, "-y", "-ss", f"{mid:.2f}", "-i", video_path, "-frames:v", "1",
            "-vf", "scale=224:-1", "-loglevel", "error", out])
       kfs.append({"scene_id": sc["id"], "t": spherical(mid, 2),
                   "path": out if os.path.exists(out) else ""})
   return {"keyframes": kfs}
_clip = None
def _load_clip():
   world _clip
   if _clip will not be None: return _clip
   attempt:
       from sentence_transformers import SentenceTransformer
       _clip = SentenceTransformer("clip-ViT-B-32")
   besides Exception as e:
       print(f"      [CLIP] unavailable ({e}); utilizing lexical-hash embeddings.")
       _clip = False
   return _clip
def _hash_embed(textual content, dim=256):
   """Deterministic bag-of-words hashing embedding (offline fallback)."""
   v = np.zeros(dim, dtype=np.float32)
   for tok in re.findall(r"[a-z0-9]+", (textual content or "").decrease()):
       v[hash(tok) % dim] += 1.0
   n = np.linalg.norm(v); return v / n if n else v
def embed_texts(texts):
   m = _load_clip()
   if m:
       attempt:
           return np.asarray(m.encode(listing(texts), normalize_embeddings=True))
       besides Exception:
           go
   return np.stack([_hash_embed(t) for t in texts])
def embed_images(paths):
   m = _load_clip()
   legitimate = [p for p in paths if p and os.path.exists(p)]
   if m and legitimate:
       attempt:
           imgs = [Image.open(p).convert("RGB") for p in valid]
           emb = np.asarray(m.encode(imgs, normalize_embeddings=True))
           out = {}; j = 0
           for p in paths:
               if p and os.path.exists(p): out[p] = emb[j]; j += 1
           return out
       besides Exception:
           go
   return {}
CAPTION_BANK = ["a title card with text", "a person speaking to camera",
               "a bright colorful scene", "a dark moody scene",
               "an outdoor landscape", "a user interface screenshot",
               "an abstract graphic", "a close-up shot"]
def tool_captioner(keyframes):
   m = _load_clip(); caps = []
   if m:
       attempt:
           lab_emb = embed_texts(CAPTION_BANK)
           paths = [k["path"] for okay in keyframes]
           img_emb = embed_images(paths)
           for okay in keyframes:
               e = img_emb.get(okay["path"])
               if e will not be None:
                   sims = lab_emb @ e
                   caps.append({"scene_id": okay["scene_id"],
                                "caption": CAPTION_BANK[int(np.argmax(sims))]})
               else:
                   caps.append({"scene_id": okay["scene_id"],
                                "caption": "a video scene"})
           return {"captions": caps}
       besides Exception:
           go
   for okay in keyframes:
       c = "a video scene"
       if okay["path"] and os.path.exists(okay["path"]):
           a = np.asarray(Image.open(okay["path"]).convert("RGB"))
           b = a.imply(); r, g, bl = a[..., 0].imply(), a[..., 1].imply(), a[..., 2].imply()
           hue = ("reddish" if r > g and r > bl else
                  "greenish" if g > r and g > bl else "bluish")
           c = f"a {'brilliant' if b > 128 else 'darkish'} {hue} scene"
       caps.append({"scene_id": okay["scene_id"], "caption": c})
   return {"captions": caps}
def _align_transcript_to_scenes(transcript, scenes):
   """Assign every ASR section to the scene containing its midpoint."""
   txt = {sc["id"]: "" for sc in scenes}
   for (s, e, t) in transcript.get("segments", []):
       mid = (s + e) / 2.0
       for sc in scenes:
           if sc["start"] <= mid < sc["end"]:
               txt[sc["id"]] = (txt[sc["id"]] + " " + t).strip(); break
   return txt
def tool_cross_modal_indexer(keyframes, captions, transcript):
   cap_by = {c["scene_id"]: c["caption"] for c in captions}
   scenes = [{"id": k["scene_id"], "t": okay["t"], "path": okay["path"]} for okay in keyframes]
   for i, sc in enumerate(scenes):
       sc["start"] = 0.0 if i == 0 else (scenes[i - 1]["t"] + sc["t"]) / 2.0
       sc["end"] = (sc["t"] + scenes[i + 1]["t"]) / 2.0 if i + 1 < len(scenes) else sc["t"] + 2.0
   tr = _align_transcript_to_scenes(transcript, scenes)
   img_emb = embed_images([sc["path"] for sc in scenes])
   text_docs = [f'{cap_by.get(sc["id"], "")}. {tr.get(sc["id"], "")}'.strip()
                for sc in scenes]
   txt_emb = embed_texts(text_docs)
   index = []
   for i, sc in enumerate(scenes):
       index.append({"scene_id": sc["id"], "t": sc["t"],
                     "caption": cap_by.get(sc["id"], ""),
                     "transcript": tr.get(sc["id"], ""),
                     "text_emb": txt_emb[i].tolist(),
                     "img_emb": (img_emb[sc["path"]].tolist()
                                 if sc["path"] in img_emb else None)})
   return {"index": index}

def tool_shot_planner(instruction, captions):
   """Global-aware storyboard sub-queries from instruction + caption financial institution."""
   financial institution = "; ".be part of(sorted({c["caption"] for c in captions}))
   if llm.obtainable():
       sys_p = ("You are VideoAgent's Shot-Planning Agent. Given the consumer "
                "instruction and a financial institution of obtainable scene captions, write "
                f"{CONFIG['max_shots']} quick storyboard sub-queries (one line "
                "every) describing the visible content material to retrieve, in narrative "
                'order. Respond as JSON listing of strings.')
       out = llm.json(sys_p, f"Instruction: {instruction}nCaptions: {financial institution}")
       if isinstance(out, listing) and out:
           return {"storyboards": [str(x) for x in out][:CONFIG["max_shots"]]}
   m = re.search(r"about ([^.,;!?]+)", instruction.decrease())
   subj = m.group(1).strip() if m else "the primary topic"
   beats = [f"opening shot introducing {subj}",
            f"a key moment about {subj}",
            f"a detailed close-up related to {subj}",
            f"a concluding shot about {subj}"]
   return {"storyboards": beats[:CONFIG["max_shots"]]}
def _cos(a, b):
   a = np.asarray(a); b = np.asarray(b)
   na, nb = np.linalg.norm(a), np.linalg.norm(b)
   return float(a @ b / (na * nb)) if na and nb else 0.0
def tool_retrieval_agent(index, storyboards):
   """For every storyboard, choose one of the best scene by max(textual content,visible) cosine."""
   q_txt = embed_texts(storyboards)
   q_img = q_txt
   chosen = []; used = set()
   for i, sb in enumerate(storyboards):
       finest, best_s = None, -1
       for e in index:
           s_t = _cos(q_txt[i], e["text_emb"])
           s_v = _cos(q_img[i], e["img_emb"]) if e["img_emb"] will not be None else -1
           rating = max(s_t, s_v)
           if e["scene_id"] in used: rating -= 0.15
           if rating > best_s:
               best_s, finest = rating, e
       if finest will not be None:
           used.add(finest["scene_id"])
           chosen.append({"storyboard": sb, "scene_id": finest["scene_id"],
                          "t": finest["t"], "rating": spherical(best_s, 3),
                          "caption": finest["caption"]})
   return {"retrieved": chosen}
def tool_trimmer(retrieved, video_path, clip_len=2.0):
   d = wp("clips"); os.makedirs(d, exist_ok=True)
   for f in os.listdir(d): os.take away(os.path.be part of(d, f))
   meta = ff_probe(video_path); dur = meta["dur"] or 6.0
   clips = []
   for i, r in enumerate(retrieved):
       s = max(0.0, r["t"] - clip_len / 2.0)
       s = min(s, max(0.0, dur - clip_len))
       out = os.path.be part of(d, f"clip_{i:03d}.mp4")
       _sh([_ff, "-y", "-ss", f"{s:.2f}", "-i", video_path, "-t", f"{clip_len:.2f}",
            "-c:v", "libx264", "-pix_fmt", "yuv420p", "-an",
            "-vf", "scale=480:270:force_original_aspect_ratio=decrease,"
                   "pad=480:270:(ow-iw)/2:(oh-ih)/2",
            "-loglevel", "error", out])
       if os.path.exists(out):
           clips.append(out)
   return {"clips": clips}
def _concat(clips, out):
   lst = wp("concat.txt")
   with open(lst, "w") as f:
       for c in clips:
           f.write(f"file '{os.path.abspath(c)}'n")
   _sh([_ff, "-y", "-f", "concat", "-safe", "0", "-i", lst,
        "-c:v", "libx264", "-pix_fmt", "yuv420p", "-loglevel", "error", out])
   return os.path.exists(out)
def tool_video_editor(clips):
   out = wp("edited.mp4")
   if clips and _concat(clips, out):
       return {"edited_video": out}
   return {"edited_video": clips[0] if clips else ""}
def tool_beat_sync_editor(rhythm_points, scenes, video_path):
   """Cut the supply onto the beat grid, biking by detected scenes."""
   d = wp("beat"); os.makedirs(d, exist_ok=True)
   for f in os.listdir(d): os.take away(os.path.be part of(d, f))
   beats = sorted(set([0.0] + listing(rhythm_points)))
   segs = [(beats[i], beats[i + 1]) for i in vary(len(beats) - 1)
           if beats[i + 1] - beats[i] > 0.15][:12]
   clips = []
   for i, (bs, be) in enumerate(segs):
       sc = scenes[i % len(scenes)]
       src = (sc["start"] + sc["end"]) / 2.0
       dur = min(be - bs, 1.2)
       out = os.path.be part of(d, f"b_{i:03d}.mp4")
       _sh([_ff, "-y", "-ss", f"{src:.2f}", "-i", video_path, "-t", f"{dur:.2f}",
            "-c:v", "libx264", "-pix_fmt", "yuv420p", "-an",
            "-vf", "scale=480:270", "-loglevel", "error", out])
       if os.path.exists(out): clips.append(out)
   out = wp("beatsync.mp4")
   if clips and _concat(clips, out):
       return {"edited_video": out}
   return {"edited_video": clips[0] if clips else ""}
def _fmt_ts(x):
   return f"{int(x // 60):02d}:{int(x % 60):02d}"
def tool_summarizer(transcript):
   textual content = transcript.get("textual content", "").strip()
   if llm.obtainable() and textual content:
       s = llm.chat("You are VideoAgent's summariser. Summarise the transcript "
                    "in 3-4 sentences, plain and factual.", textual content)
       if s: return {"abstract": s.strip()}
   sents = re.cut up(r"(?<=[.!?])s+", textual content)
   sents = [s for s in sents if len(s.split()) > 3]
   if not sents:
       return {"abstract": "(no speech detected to summarise)"}
   picks = [sents[0]] + sorted(sents[1:], key=lambda s: -len(s))[:2]
   return {"abstract": " ".be part of(dict.fromkeys(picks))}
def tool_video_qa(transcript, query):
   segs = transcript.get("segments", []); textual content = transcript.get("textual content", "")
   if llm.obtainable() and textual content:
       ans = llm.chat("You are VideoAgent's VideoQA agent. Answer ONLY from the "
                      "transcript; if unknown, say so. Be concise.",
                      f"Transcript:n{textual content}nnQuestion: {query}")
       if ans: return {"reply": ans.strip()}
   qtok = set(re.findall(r"[a-z0-9]+", query.decrease()))
   scored = []
   for (s, e, t) in segs:
       ov = len(qtok & set(re.findall(r"[a-z0-9]+", t.decrease())))
       if ov: scored.append((ov, s, e, t))
   scored.kind(reverse=True)
   if not scored:
       return {"reply": "I could not discover that within the video's speech."}
   prime = scored[:2]
   return {"reply": " ".be part of(f"[{_fmt_ts(s)}] {t}" for _o, s, _e, t in prime)}
def tool_news_overview(transcript, instruction):
   textual content = transcript.get("textual content", "").strip()
   if llm.obtainable() and textual content:
       ov = llm.chat("You are VideoAgent's InformationContentGenerator. Write a quick, "
                     "colloquial information overview (<=120 phrases) of the transcript, "
                     "matching any type hints within the instruction.",
                     f"Instruction: {instruction}nTranscript: {textual content}")
       if ov: return {"overview": ov.strip()}
   base = tool_summarizer(transcript)["summary"]
   return {"overview": "Here's the rundown: " + base}
def tool_renderer(edited_video):
   if not edited_video or not os.path.exists(edited_video):
       return {"final_video": ""}
   out = wp("ultimate.mp4")
   r = _sh([_ff, "-y", "-i", edited_video, "-c:v", "libx264", "-pix_fmt",
            "yuv420p", "-movflags", "+faststart", "-loglevel", "error", out])
   return {"final_video": out if os.path.exists(out) else edited_video}
_IMPL = {
   "AudioExtractor": tool_audio_extractor, "Transcriber": tool_transcriber,
   "RhythmDetector": tool_rhythm_detector, "SceneDetector": tool_scene_detector,
   "KeyframeSampler": tool_keyframe_sampler, "Captioner": tool_captioner,
   "CrossModalIndexer": tool_cross_modal_indexer, "ShotPlanner": tool_shot_planner,
   "RetrievalAgent": tool_retrieval_agent, "Trimmer": tool_trimmer,
   "VideoEditor": tool_video_editor, "BeatSyncEditor": tool_beat_sync_editor,
   "Summarizer": tool_summarizer, "VideoQA": tool_video_qa,
   "InformationContentGenerator": tool_news_overview, "Renderer": tool_renderer,
}
for _a, _fn in _IMPL.gadgets():
   AGENTS[_a]["fn"] = _fn
class VideoAgent:
   def __init__(self, video_path):
       self.video = video_path
   def run(self, instruction):
       print("n" + "═" * 78)
       print("USER INSTRUCTION:", instruction)
       print("═" * 78)
       T, params = analyze_intents(instruction)
       print("[1] Intent Analysis → required intents T:")
       print("    ", ", ".be part of(sorted(T)))
       if params.get("question"):
           print("     extracted retrieval topic:", repr(params["query"]))
       cand = route_tools(T)
       print(f"[2] Tool Routing → {len(cand)} candidate brokers match T:")
       print("    ", ", ".be part of(sorted(cand)))
       nodes = llm_plan(T, instruction) if llm.obtainable() else None
       origin = "LLM-drafted" if nodes else "naive (terminals solely)"
       if nodes is None:
           nodes = naive_plan(T)
       print(f"[4] Graph Construction → {origin} graph "
             f"({len(nodes)} nodes): {sorted(nodes)}")
       print("[5] Textual-Gradient Graph Optimization (τ, κ, χ):")
       nodes, historical past = optimize_graph(nodes, T, Tmax=CONFIG["opt_rounds"])
       order = topo_order(nodes)
       print(f"    Final agent chain: {' → '.be part of(order)}")
       print("[6] Graph Execution:")
       seed = {"video_path": self.video, "instruction": instruction,
               "query": params.get("query", instruction),
               "question": params.get("question", "")}
       bb, order = execute_graph(nodes, seed)
       consequence = {}
       for key in ("reply", "overview", "abstract", "final_video", "edited_video"):
           if key in bb and bb[key]:
               consequence[key] = bb[key]
       print("n── RESULT " + "─" * 68)
       for okay, v in consequence.gadgets():
           if okay in ("final_video", "edited_video"):
               print(f"  {okay}: {v}   ({os.path.getsize(v)//1024} KB)"
                     if v and os.path.exists(v) else f"  {okay}: (empty)")
           else:
               print(f"  {okay}:n{textwrap.indent(textwrap.fill(str(v), 92), '    ')}")
       return consequence, nodes, bb

DEMO_SCRIPT = [
   ("BREAKING TECH NEWS", (18, 22, 46),
    "Breaking tech news: a major lab has released a new image generation model."),
   ("THE MODEL: GPT-4o", (128, 28, 40),
    "The system, called GPT-4o, can generate and edit images directly from text prompts."),
   ("IMAGE GENERATION", (16, 78, 60),
    "Its native image generation renders text inside pictures far more accurately than before."),
   ("LIVE DEMO", (58, 40, 92),
    "In the live demo, the model turned a rough sketch into a polished product photo in seconds."),
   ("WHAT IT MEANS", (30, 30, 34),
    "Analysts say this pushes multimodal AI closer to everyday creative work."),
]
def build_demo_video():
   world FALLBACK_SCRIPT
   W, H, FPS, sec = 480, 270, 24, 2.0
   fdir = wp("democ_frames"); os.makedirs(fdir, exist_ok=True)
   for f in os.listdir(fdir): os.take away(os.path.be part of(fdir, f))
   idx = 0; scene_times = []
   for title, col, _n in DEMO_SCRIPT:
       t0 = idx / FPS
       for fr in vary(int(FPS * sec)):
           im = Image.new("RGB", (W, H), col); d = ImageDraw.Draw(im)
           x = int((fr / (FPS * sec)) * (W - 70))
           d.rectangle([x, H - 42, x + 56, H - 22], fill=(235, 235, 235))
           d.textual content((22, 26), title, fill=(255, 255, 255))
           d.textual content((22, H - 66), f"t={idx/FPS:0.2f}s", fill=(210, 210, 210))
           im.save(os.path.be part of(fdir, f"f{idx:05d}.png")); idx += 1
       scene_times.append((t0, idx / FPS))
   silent = wp("demo_silent.mp4")
   _sh([_ff, "-y", "-framerate", str(FPS), "-i", os.path.join(fdir, "f%05d.png"),
        "-pix_fmt", "yuv420p", "-c:v", "libx264", "-loglevel", "error", silent])
   narrated = wp("demo.mp4"); have_audio = False
   attempt:
       from gtts import gTTS
       adir = wp("narr"); os.makedirs(adir, exist_ok=True)
       components = []
       for i, (_t, _c, n) in enumerate(DEMO_SCRIPT):
           mp3 = os.path.be part of(adir, f"n{i}.mp3"); gTTS(n).save(mp3); components.append(mp3)
       lst = wp("narr.txt")
       with open(lst, "w") as f:
           for i, mp3 in enumerate(components):
               f.write(f"file '{os.path.abspath(mp3)}'n")
       joined = os.path.be part of(adir, "joined.mp3")
       _sh([_ff, "-y", "-f", "concat", "-safe", "0", "-i", lst,
            "-loglevel", "error", joined])
       _sh([_ff, "-y", "-i", silent, "-i", joined, "-c:v", "copy",
            "-c:a", "aac", "-shortest", "-loglevel", "error", narrated])
       have_audio = os.path.exists(narrated)
   besides Exception as e:
       print(f"[demo] gTTS narration unavailable ({e}); utilizing silent video "
             f"+ injected transcript.")
   if not have_audio:
       narrated = silent
       FALLBACK_SCRIPT = [(scene_times[i][0], scene_times[i][1], DEMO_SCRIPT[i][2])
                          for i in vary(len(DEMO_SCRIPT))]
   else:
       FALLBACK_SCRIPT = [(scene_times[i][0], scene_times[i][1], DEMO_SCRIPT[i][2])
                          for i in vary(len(DEMO_SCRIPT))]
   meta = ff_probe(narrated)
   print(f"[demo] constructed {narrated}  ({meta['dur']:.1f}s, "
         f"{meta['w']}x{meta['h']}, audio={'sure' if have_audio else 'no'})")
   return narrated
def preview(path, width=480):
   """Display an mp4 inline in Colab/Jupyter (no-op elsewhere)."""
   attempt:
       from IPython.show import HTML, show
       import base64
       with open(path, "rb") as f:
           b64 = base64.b64encode(f.learn()).decode()
       show(HTML(f'<video width="{width}" controls '
                    f'src="information:video/mp4;base64,{b64}"></video>'))
   besides Exception as e:
       print(f"[preview] {path} (inline show unavailable: {e})")
DEMO_INSTRUCTIONS = {
   "qa":        "What does the video say about GPT-4o picture technology?",
   "overview":  "Give me a quick, colloquial information overview of this tech video.",
   "spotlight": "Make a spotlight montage concerning the picture technology mannequin.",
   "beatsync":  "Create a quick beat-synced montage from this footage.",
}
def predominant():
   print("VideoAgent — devoted reconstruction")
   print("LLM backend:",
         f"{CONFIG['provider']} ({llm.mannequin})" if llm.obtainable()
         else "NONE → deterministic brokers (paste a key in CONFIG to go LLM-driven)")
   if not HAS_FFMPEG:
       print("!! ffmpeg not discovered — set up it (Colab has it by default).")
       return
   video = build_demo_video()
   agent = VideoAgent(video)
   outputs = {}
   for title in CONFIG["run_demos"]:
       if title not in DEMO_INSTRUCTIONS:
           proceed
       res, _nodes, _bb = agent.run(DEMO_INSTRUCTIONS[name])
       outputs[name] = res
   print("n" + "═" * 78)
   print("ALL DEMOS COMPLETE. Artifacts are underneath:", WORK)
   for title, res in outputs.gadgets():
       for okay, v in res.gadgets():
           if okay.endswith("video") and v and os.path.exists(v):
               print(f"  [{name}] {okay}: {v}")
   print("nTo use YOUR OWN video:  VideoAgent('/content material/my.mp4').run('...')")
   print("To go LLM-driven: set CONFIG['provider']/['api_key'] and re-run.")
if __name__ == "__main__":
   predominant()