def tool_shot_planner(instruction, captions):
"""Global-aware storyboard sub-queries from instruction + caption bank."""
bank = "; ".join(sorted({c["caption"] for c in captions}))
if llm.available():
sys_p = ("You are VideoAgent's Shot-Planning Agent. Given the user "
"instruction and a bank of available scene captions, write "
f"{CONFIG['max_shots']} short storyboard sub-queries (one line "
"each) describing the visual content to retrieve, in narrative "
'order. Respond as JSON list of strings.')
out = llm.json(sys_p, f"Instruction: {instruction}\nCaptions: {bank}")
if isinstance(out, list) and out:
return {"storyboards": [str(x) for x in out][:CONFIG["max_shots"]]}
m = re.search(r"about ([^.,;!?]+)", instruction.lower())
subj = m.group(1).strip() if m else "the main subject"
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 each storyboard, pick the best scene by max(text,visual) cosine."""
q_txt = embed_texts(storyboards)
q_img = q_txt
chosen = []; used = set()
for i, sb in enumerate(storyboards):
best, 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"] is not None else -1
score = max(s_t, s_v)
if e["scene_id"] in used: score -= 0.15
if score > best_s:
best_s, best = score, e
if best is not None:
used.add(best["scene_id"])
chosen.append({"storyboard": sb, "scene_id": best["scene_id"],
"t": best["t"], "score": round(best_s, 3),
"caption": best["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.remove(os.path.join(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.join(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 source onto the beat grid, cycling through detected scenes."""
d = wp("beat"); os.makedirs(d, exist_ok=True)
for f in os.listdir(d): os.remove(os.path.join(d, f))
beats = sorted(set([0.0] + list(rhythm_points)))
segs = [(beats[i], beats[i + 1]) for i in range(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.join(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):
text = transcript.get("text", "").strip()
if llm.available() and text:
s = llm.chat("You are VideoAgent's summariser. Summarise the transcript "
"in 3-4 sentences, plain and factual.", text)
if s: return {"summary": s.strip()}
sents = re.split(r"(?<=[.!?])\s+", text)
sents = [s for s in sents if len(s.split()) > 3]
if not sents:
return {"summary": "(no speech detected to summarise)"}
picks = [sents[0]] + sorted(sents[1:], key=lambda s: -len(s))[:2]
return {"summary": " ".join(dict.fromkeys(picks))}
def tool_video_qa(transcript, question):
segs = transcript.get("segments", []); text = transcript.get("text", "")
if llm.available() and text:
ans = llm.chat("You are VideoAgent's VideoQA agent. Answer ONLY from the "
"transcript; if unknown, say so. Be concise.",
f"Transcript:\n{text}\n\nQuestion: {question}")
if ans: return {"answer": ans.strip()}
qtok = set(re.findall(r"[a-z0-9]+", question.lower()))
scored = []
for (s, e, t) in segs:
ov = len(qtok & set(re.findall(r"[a-z0-9]+", t.lower())))
if ov: scored.append((ov, s, e, t))
scored.sort(reverse=True)
if not scored:
return {"answer": "I couldn't find that in the video's speech."}
top = scored[:2]
return {"answer": " ".join(f"[{_fmt_ts(s)}] {t}" for _o, s, _e, t in top)}
def tool_news_overview(transcript, instruction):
text = transcript.get("text", "").strip()
if llm.available() and text:
ov = llm.chat("You are VideoAgent's NewsContentGenerator. Write a short, "
"colloquial news overview (<=120 words) of the transcript, "
"matching any style hints in the instruction.",
f"Instruction: {instruction}\nTranscript: {text}")
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("final.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,
"NewsContentGenerator": tool_news_overview, "Renderer": tool_renderer,
}
for _a, _fn in _IMPL.items():
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(" ", ", ".join(sorted(T)))
if params.get("query"):
print(" extracted retrieval subject:", repr(params["query"]))
cand = route_tools(T)
print(f"[2] Tool Routing → {len(cand)} candidate agents match T:")
print(" ", ", ".join(sorted(cand)))
nodes = llm_plan(T, instruction) if llm.available() else None
origin = "LLM-drafted" if nodes else "naive (terminals only)"
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, history = optimize_graph(nodes, T, Tmax=CONFIG["opt_rounds"])
order = topo_order(nodes)
print(f" Final agent chain: {' → '.join(order)}")
print("[6] Graph Execution:")
seed = {"video_path": self.video, "instruction": instruction,
"question": params.get("question", instruction),
"query": params.get("query", "")}
bb, order = execute_graph(nodes, seed)
result = {}
for key in ("answer", "overview", "summary", "final_video", "edited_video"):
if key in bb and bb[key]:
result[key] = bb[key]
print("\n── RESULT " + "─" * 68)
for k, v in result.items():
if k in ("final_video", "edited_video"):
print(f" {k}: {v} ({os.path.getsize(v)//1024} KB)"
if v and os.path.exists(v) else f" {k}: (empty)")
else:
print(f" {k}:\n{textwrap.indent(textwrap.fill(str(v), 92), ' ')}")
return result, nodes, bb
Credit: Source link
