def tool_shot_planner(instrucciones, subtítulos): “””Subconsultas del guión gráfico con reconocimiento global de instrucciones + banco de subtítulos.””” bank = “; “.join(sorted({c[“caption”] para c en subtítulos})) if llm.available(): sys_p = (“Usted es el agente de planificación de tomas de VideoAgent. Dadas las ” “instrucciones del usuario y un banco de subtítulos de escena disponibles, escriba ” f”{CONFIG[‘max_shots’]} subconsultas breves del guión gráfico (una línea ” “cada una) que describen el contenido visual a recuperar, en ” ‘orden narrativo. Responda como lista JSON de cadenas.’) out = llm.json(sys_p, f”Instrucción: {instrucción}\nCaptions: {bank}”) if isinstance(out, list) and out: return {“storyboards”: [str(x) for x in out][:CONFIG[“max_shots”]]} m = re.buscar(r”acerca de ([^.,;!?]+)”, instrucción.lower()) subj = m.group(1).strip() si m else “el tema principal” late = [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”: tiempos[: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 y nb else 0.0 def tool_retrieval_agent(index, storyboards): “””Para cada guión gráfico, elija la mejor escena por coseno máximo(texto,visual).””” q_txt = embed_texts(guiones gráficos) q_img = q_txt elegido = []; usado = set() para i, sb en enumerar(storyboards): mejor, best_s = Ninguno, -1 para e en índice: s_t = _cos(q_txt[i]mi[“text_emb”]) s_v = _cos(q_img[i]mi[“img_emb”]) si e[“img_emb”] no es Nadie más -1 puntuación = max(s_t, s_v) si e[“scene_id”] en usado: puntaje -= 0.15 si puntaje > mejores_s: mejores_s, mejor = puntaje, e si mejor no es Ninguno: usado.add(mejor[“scene_id”]) elegido.append({“storyboard”: sb, “scene_id”: mejor[“scene_id”]”t”: mejor[“t”]”puntuación”: round(best_s, 3), “caption”: mejor[“caption”]}) return {“recuperado”: elegido} def tool_trimmer(recuperado, video_path, clip_len=2.0): d = wp(“clips”); os.makedirs(d, exist_ok=True) para f en os.listdir(d): os.remove(os.path.join(d, f)) meta = ff_probe(video_path); dur = meta[“dur”] o 6,0 clips = []
para i, r en enumerar (recuperado): 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]) si os.path.exists(out): clips.append(out) return {“clips”: clips} def _concat(clips, out): lst = wp(“concat.txt”) con open(lst, “w”) como f: para c en 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 y _concat(clips, out): return {“edited_video”: out} return {“edited_video”: clips[0] if clips else “”} def tool_beat_sync_editor(rhythm_points, scene, video_path): “””Corta la fuente en el beat grid, recorriendo las escenas detectadas.””” d = wp(“beat”); os.makedirs(d, exist_ok=True) para f en os.listdir(d): os.remove(os.path.join(d, f)) beats = sorted(set([0.0] + lista(puntos_ritmo))) segmentos = [(beats[i]late[i + 1]) para i en el rango(len(tiempos) – 1) si tiempos[i + 1] – latidos[i] > 0,15][:12]
clips = []
para i, (bs, be) en enumerar(segs): sc = escenas[i % len(scenes)]
src = (sc[“start”] + pb[“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]) si os.path.exists(out): clips.append(out) out = wp(“beatsync.mp4”) if clips y _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(“Usted es el resumidor de VideoAgent. Resuma la transcripción ” “en 3-4 oraciones, sencillas y objetivas.”, texto) if s: return {“summary”: s.strip()} sents = re.split(r”(?<=[.!?])\s+", texto) envía = [s for s in sents if len(s.split()) > 3] si no envía: return {"summary": "(no se detectó voz para resumir)"} picks = [sents[0]]+ ordenado(envía[1:]clave=lambda s: -len(s))[:2] return {"summary": " ".join(dict.fromkeys(selecciones))} def tool_video_qa(transcripción, pregunta): segs = transcript.get("segmentos", []); text = transcript.get("text", "") if llm.available() and text: ans = llm.chat("Usted es el agente VideoQA de VideoAgent. Responda SÓLO de la " "transcripción; si no lo sabe, dígalo. Sea conciso.", f"Transcripción:\n{text}\n\nPregunta: {pregunta}") if ans: return {"answer": ans.strip()} qtok = set(re.findall(r"[a-z0-9]+", question.lower())) puntuado = [] para (s, e, t) en segmentos: ov = len(qtok & set(re.findall(r"[a-z0-9]+", t.lower()))) if ov: scoring.append((ov, s, e, t)) scoring.sort(reverse=True) if not scoring: return {"answer": "No pude encontrar eso en el discurso del video."} top = scoring[:2] return {"respuesta": " ".unirse(f"[{_fmt_ts(s)}] {t}" para _o, s, _e, t en la parte superior)} def tool_news_overview(transcripción, instrucción): text = transcript.get("text", "").strip() if llm.available() and text: ov = llm.chat("Usted es NewsContentGenerator de VideoAgent. Escriba una " "resumen breve y coloquial de noticias (<=120 palabras) de la transcripción, " "que coincida con cualquier sugerencia de estilo en la instrucción.", f"Instrucción: {instrucción}\nTranscripción: {texto}") if ov: return {"overview": ov.strip()} base = tool_summarizer(transcripción)["summary"] return {"overview": "Aquí está el resumen: " + base} def tool_renderer(edited_video): si no edited_video o no 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": herramienta_audio_extractor, "Transcriber": herramienta_transcriber, "RhythmDetector": herramienta_rhythm_detector, "SceneDetector": herramienta_scene_detector, "KeyframeSampler": herramienta_keyframe_sampler, "Captioner": tool_captioner, "CrossModalIndexer": herramienta_cross_modal_indexer, "ShotPlanner": herramienta_shot_planner, "RetrievalAgent": herramienta_retrieval_agent, "Trimmer": herramienta_trimmer, "VideoEditor": herramienta_video_editor, "BeatSyncEditor": herramienta_beat_sync_editor, "Summarizer": herramienta_summarizer, "VideoQA": herramienta_video_qa, "NewsContentGenerator": tool_news_overview, "Renderer": tool_renderer, } para _a, _fn en _IMPL.items(): AGENTES[_a]["fn"] = _fn clase VideoAgent: def __init__(self, video_path): self.video = video_path def run(self, instrucción): print("\n" + "═" * 78) print("INSTRUCCIONES DE USUARIO:", instrucción) print("═" * 78) T, params = analyse_intents(instrucción) print("[1] Análisis de intención → intenciones requeridas T:") print(" ", ", ".join(ordenado(T))) if params.get("consulta"): print(" asunto de recuperación extraído:", repr(params["query"])) cand = herramientas_ruta(T) print(f"[2] Enrutamiento de herramientas → {len(cand)} los agentes candidatos coinciden con T:") print(" ", ", ".join(sorted(cand))) nodos = llm_plan(T, instrucción) if llm.available() else Ninguno origen = "LLM-drafted" si los nodos else "ingenuo (solo terminales)" si los nodos son Ninguno: nodos = naive_plan(T) print(f"[4] Construcción de gráficos → {origen} gráfico " f"({len(nodos)} nodos): {ordenados(nodos)}") print("[5] Optimización de gráfico de gradiente textual (τ, κ, χ):") nodos, historial = optimizar_grafo(nodos, T, Tmax=CONFIG["opt_rounds"]) orden = topo_order(nodos) print(f" Cadena final del agente: {' → '.join(order)}") print("[6] Ejecución de gráfico:") semilla = {"video_path": self.video, "instruction": instrucción, "question": params.get("question", instrucción), "query": params.get("query", "")} bb, orden = ejecutar_graph(nodos, semilla) resultado = {} para ingresar ("respuesta", "descripción general", "summary", "final_video", "edited_video"): si ingresa bb y bb[key]: resultado[key] = bb[key] print("\n── RESULTADO " + "─" * 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 y os.path.exists(v) else f" {k}: (vacío)") else: print(f" {k}:\n{textwrap.indent(textwrap.fill(str(v), 92), ' ')}") devuelve resultado, nodos, bb
f”a key moment about {subj}”,
f”a detailed close-up related to {subj}”,
f”a concluding shot about {subj}”]
return {“storyboards”: tiempos[: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 y nb else 0.0 def tool_retrieval_agent(index, storyboards): “””Para cada guión gráfico, elija la mejor escena por coseno máximo(texto,visual).””” q_txt = embed_texts(guiones gráficos) q_img = q_txt elegido = []; usado = set() para i, sb en enumerar(storyboards): mejor, best_s = Ninguno, -1 para e en índice: s_t = _cos(q_txt[i]mi[“text_emb”]) s_v = _cos(q_img[i]mi[“img_emb”]) si e[“img_emb”] no es Nadie más -1 puntuación = max(s_t, s_v) si e[“scene_id”] en usado: puntaje -= 0.15 si puntaje > mejores_s: mejores_s, mejor = puntaje, e si mejor no es Ninguno: usado.add(mejor[“scene_id”]) elegido.append({“storyboard”: sb, “scene_id”: mejor[“scene_id”]”t”: mejor[“t”]”puntuación”: round(best_s, 3), “caption”: mejor[“caption”]}) return {“recuperado”: elegido} def tool_trimmer(recuperado, video_path, clip_len=2.0): d = wp(“clips”); os.makedirs(d, exist_ok=True) para f en os.listdir(d): os.remove(os.path.join(d, f)) meta = ff_probe(video_path); dur = meta[“dur”] o 6,0 clips = []
para i, r en enumerar (recuperado): 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]) si os.path.exists(out): clips.append(out) return {“clips”: clips} def _concat(clips, out): lst = wp(“concat.txt”) con open(lst, “w”) como f: para c en 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 y _concat(clips, out): return {“edited_video”: out} return {“edited_video”: clips[0] if clips else “”} def tool_beat_sync_editor(rhythm_points, scene, video_path): “””Corta la fuente en el beat grid, recorriendo las escenas detectadas.””” d = wp(“beat”); os.makedirs(d, exist_ok=True) para f en os.listdir(d): os.remove(os.path.join(d, f)) beats = sorted(set([0.0] + lista(puntos_ritmo))) segmentos = [(beats[i]late[i + 1]) para i en el rango(len(tiempos) – 1) si tiempos[i + 1] – latidos[i] > 0,15][:12]
clips = []
para i, (bs, be) en enumerar(segs): sc = escenas[i % len(scenes)]
src = (sc[“start”] + pb[“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]) si os.path.exists(out): clips.append(out) out = wp(“beatsync.mp4”) if clips y _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(“Usted es el resumidor de VideoAgent. Resuma la transcripción ” “en 3-4 oraciones, sencillas y objetivas.”, texto) if s: return {“summary”: s.strip()} sents = re.split(r”(?<=[.!?])\s+", texto) envía = [s for s in sents if len(s.split()) > 3] si no envía: return {"summary": "(no se detectó voz para resumir)"} picks = [sents[0]]+ ordenado(envía[1:]clave=lambda s: -len(s))[:2] return {"summary": " ".join(dict.fromkeys(selecciones))} def tool_video_qa(transcripción, pregunta): segs = transcript.get("segmentos", []); text = transcript.get("text", "") if llm.available() and text: ans = llm.chat("Usted es el agente VideoQA de VideoAgent. Responda SÓLO de la " "transcripción; si no lo sabe, dígalo. Sea conciso.", f"Transcripción:\n{text}\n\nPregunta: {pregunta}") if ans: return {"answer": ans.strip()} qtok = set(re.findall(r"[a-z0-9]+", question.lower())) puntuado = [] para (s, e, t) en segmentos: ov = len(qtok & set(re.findall(r"[a-z0-9]+", t.lower()))) if ov: scoring.append((ov, s, e, t)) scoring.sort(reverse=True) if not scoring: return {"answer": "No pude encontrar eso en el discurso del video."} top = scoring[:2] return {"respuesta": " ".unirse(f"[{_fmt_ts(s)}] {t}" para _o, s, _e, t en la parte superior)} def tool_news_overview(transcripción, instrucción): text = transcript.get("text", "").strip() if llm.available() and text: ov = llm.chat("Usted es NewsContentGenerator de VideoAgent. Escriba una " "resumen breve y coloquial de noticias (<=120 palabras) de la transcripción, " "que coincida con cualquier sugerencia de estilo en la instrucción.", f"Instrucción: {instrucción}\nTranscripción: {texto}") if ov: return {"overview": ov.strip()} base = tool_summarizer(transcripción)["summary"] return {"overview": "Aquí está el resumen: " + base} def tool_renderer(edited_video): si no edited_video o no 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": herramienta_audio_extractor, "Transcriber": herramienta_transcriber, "RhythmDetector": herramienta_rhythm_detector, "SceneDetector": herramienta_scene_detector, "KeyframeSampler": herramienta_keyframe_sampler, "Captioner": tool_captioner, "CrossModalIndexer": herramienta_cross_modal_indexer, "ShotPlanner": herramienta_shot_planner, "RetrievalAgent": herramienta_retrieval_agent, "Trimmer": herramienta_trimmer, "VideoEditor": herramienta_video_editor, "BeatSyncEditor": herramienta_beat_sync_editor, "Summarizer": herramienta_summarizer, "VideoQA": herramienta_video_qa, "NewsContentGenerator": tool_news_overview, "Renderer": tool_renderer, } para _a, _fn en _IMPL.items(): AGENTES[_a]["fn"] = _fn clase VideoAgent: def __init__(self, video_path): self.video = video_path def run(self, instrucción): print("\n" + "═" * 78) print("INSTRUCCIONES DE USUARIO:", instrucción) print("═" * 78) T, params = analyse_intents(instrucción) print("[1] Análisis de intención → intenciones requeridas T:") print(" ", ", ".join(ordenado(T))) if params.get("consulta"): print(" asunto de recuperación extraído:", repr(params["query"])) cand = herramientas_ruta(T) print(f"[2] Enrutamiento de herramientas → {len(cand)} los agentes candidatos coinciden con T:") print(" ", ", ".join(sorted(cand))) nodos = llm_plan(T, instrucción) if llm.available() else Ninguno origen = "LLM-drafted" si los nodos else "ingenuo (solo terminales)" si los nodos son Ninguno: nodos = naive_plan(T) print(f"[4] Construcción de gráficos → {origen} gráfico " f"({len(nodos)} nodos): {ordenados(nodos)}") print("[5] Optimización de gráfico de gradiente textual (τ, κ, χ):") nodos, historial = optimizar_grafo(nodos, T, Tmax=CONFIG["opt_rounds"]) orden = topo_order(nodos) print(f" Cadena final del agente: {' → '.join(order)}") print("[6] Ejecución de gráfico:") semilla = {"video_path": self.video, "instruction": instrucción, "question": params.get("question", instrucción), "query": params.get("query", "")} bb, orden = ejecutar_graph(nodos, semilla) resultado = {} para ingresar ("respuesta", "descripción general", "summary", "final_video", "edited_video"): si ingresa bb y bb[key]: resultado[key] = bb[key] print("\n── RESULTADO " + "─" * 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 y os.path.exists(v) else f" {k}: (vacío)") else: print(f" {k}:\n{textwrap.indent(textwrap.fill(str(v), 92), ' ')}") devuelve resultado, nodos, bb