From da041e54f44ff40699ef55d8cfdb72fdfa4280c0 Mon Sep 17 00:00:00 2001 From: Mustansir Rangwala <119410651+mustansirr@users.noreply.github.com> Date: Mon, 29 Jun 2026 10:47:40 +0530 Subject: [PATCH 1/2] feat: Add pipeline diagnostics and fix MathTex indexing rule --- backend/app/agents/nodes/audio_generator.py | 73 ++++++ backend/app/agents/nodes/coder.py | 10 + backend/app/agents/prompts/coder_prompts.py | 21 +- backend/app/agents/state.py | 2 + backend/app/agents/workflow.py | 11 +- backend/app/sandbox/renderer.py | 47 ++++ backend/app/sandbox/stitcher.py | 166 +++++++++++-- backend/app/services/tts_service.py | 82 +++++++ backend/requirements.txt | 2 + docs/original_workflow.md | 247 ++++++++++++++++++++ 10 files changed, 638 insertions(+), 23 deletions(-) create mode 100644 backend/app/agents/nodes/audio_generator.py create mode 100644 backend/app/services/tts_service.py create mode 100644 docs/original_workflow.md diff --git a/backend/app/agents/nodes/audio_generator.py b/backend/app/agents/nodes/audio_generator.py new file mode 100644 index 0000000..77e50e2 --- /dev/null +++ b/backend/app/agents/nodes/audio_generator.py @@ -0,0 +1,73 @@ +""" +Audio Generator Node. + +This LangGraph node generates TTS audio for the current scene +and records the precise audio duration into the state. +""" + +import logging +from pathlib import Path + +from app.agents.state import AgentState +from app.services.tts_service import generate_scene_audio +from app.config import get_settings + +logger = logging.getLogger(__name__) + + +async def generate_audio_node(state: AgentState) -> AgentState: + """ + Generate TTS audio and subtitles for the current scene. + + Args: + state: Current agent state. + + Returns: + Updated AgentState with scene_audio_durations. + """ + scene_index = state.get("current_scene_index", 0) + scripts = state.get("scripts", []) + video_id = state.get("video_id") + + if not scripts or scene_index >= len(scripts): + logger.error(f"No script found for scene {scene_index}") + return state + + current_script = scripts[scene_index] + narration_text = current_script.get("narration", "") + + settings = get_settings() + storage_path = Path(getattr(settings, "storage_path", "/app/storage")) + output_dir = storage_path / video_id / f"scene_{scene_index}" + + if not narration_text.strip(): + logger.warning(f"Empty narration for scene {scene_index}, duration is 0") + durations = state.get("scene_audio_durations", {}).copy() + durations[scene_index] = 0.0 + return {**state, "scene_audio_durations": durations} + + try: + duration, audio_path, vtt_path = await generate_scene_audio( + narration_text=narration_text, + output_dir=output_dir, + scene_index=scene_index + ) + + # Update state with the exact duration + durations = state.get("scene_audio_durations", {}).copy() + durations[scene_index] = duration + + logger.info(f"Audio generated for scene {scene_index}: {duration}s") + return {**state, "scene_audio_durations": durations} + + except Exception as e: + logger.exception(f"Failed to generate audio for scene {scene_index}: {e}") + # In case of failure, provide a fallback duration estimate based on words + words = len(narration_text.split()) + estimated_duration = max(3.0, words * 0.4) + + durations = state.get("scene_audio_durations", {}).copy() + durations[scene_index] = estimated_duration + + logger.warning(f"Using fallback duration {estimated_duration}s for scene {scene_index}") + return {**state, "scene_audio_durations": durations} diff --git a/backend/app/agents/nodes/coder.py b/backend/app/agents/nodes/coder.py index 396f0f6..c876c70 100644 --- a/backend/app/agents/nodes/coder.py +++ b/backend/app/agents/nodes/coder.py @@ -114,9 +114,19 @@ async def generate_code(state: AgentState) -> dict: llm = create_llm("coder", temperature=0.2) # Create prompt with few-shot examples + # Get audio duration for current scene (default to 5.0 if missing) + scene_audio_durations = state.get("scene_audio_durations", {}) + audio_duration = scene_audio_durations.get(scene_index, 5.0) + + # --- TASK 2: VERIFY AUDIO SYNCHRONIZATION --- + logger.info(f"=== SYNCHRONIZATION DIAGNOSTIC ===") + logger.info(f"Coder read audio_duration={audio_duration:.3f}s from state for scene {scene_index+1}") + logger.info(f"==================================") + prompt = create_coder_prompt( visual_description=visual_description, narration=narration, + audio_duration=audio_duration, include_examples=True, ) diff --git a/backend/app/agents/prompts/coder_prompts.py b/backend/app/agents/prompts/coder_prompts.py index a1f0acb..1edd25d 100644 --- a/backend/app/agents/prompts/coder_prompts.py +++ b/backend/app/agents/prompts/coder_prompts.py @@ -42,6 +42,16 @@ WRONG: `self.play(Write(MathTex("...")).next_to(...))` CORRECT: `obj = MathTex("...").next_to(...)` then `self.play(Write(obj))` +CRITICAL TEXT AND LAYOUT RULES: +1. PREVENT TEXT OVERLAP: Texts should NEVER overlap with each other. Use `VGroup(...).arrange(DOWN, buff=0.5)` or `.next_to(...)` to position texts relative to one another. +2. TEXT INSIDE OBJECTS: If an object (like a Rectangle or Circle) needs to have a title or name inside it, the text MUST appear completely inside it and NOT spill out. Use `text.scale_to_fit_width(object.width - 0.5)` or `text.scale_to_fit_height(object.height - 0.5)` or set an appropriately small `font_size`. +3. TEXT CONTAINMENT: Same goes with any text that should appear inside an object, it should NOT appear outside the component. +4. TEXT READABILITY: A solid component/shape should NOT be put on top of text, as the text will not be easily readable. Always render text *after* or *on top of* background objects. +5. STRICT VISUAL PLAN: All the animations and visualizations MUST make sense and strictly follow the provided visual plan step-by-step. Do not invent unrelated visuals. +6. MATHTEX INDEXING: To animate parts of an equation separately, you MUST pass them as separate string arguments to `MathTex`. + - WRONG: `eq = MathTex("a^2 + b^2 = c^2")` followed by `self.play(Write(eq[1]))` (This causes IndexError because the whole equation is `eq[0]`). + - CORRECT: `eq = MathTex("a^2", "+", "b^2", "=", "c^2")` followed by `self.play(Write(eq[1]))` (Now `+` is `eq[1]`). + FORBIDDEN — DO NOT USE: 1. SVGMobject() — no SVG files exist in the render environment. 2. ImageMobject() — no image files exist in the render environment. @@ -109,6 +119,7 @@ def get_few_shot_examples(snippets_dir: Optional[Path] = None) -> str: def create_coder_prompt( visual_description: str, narration: str, + audio_duration: float, include_examples: bool = True ) -> str: """ @@ -117,6 +128,7 @@ def create_coder_prompt( Args: visual_description: What should appear on screen (from scripter). narration: The narration text (for timing reference). + audio_duration: The precise length of the TTS audio in seconds. include_examples: Whether to include few-shot examples. Returns: @@ -133,7 +145,14 @@ def create_coder_prompt( prompt_parts.append("SCENE TO ANIMATE:") prompt_parts.append(f"Visual Description: {visual_description}") - prompt_parts.append(f"Narration (for timing reference): {narration}") + prompt_parts.append(f"Narration (for context): {narration}") + prompt_parts.append(f"AUDIO DURATION: {audio_duration:.3f} seconds.") + prompt_parts.append( + "\nCRITICAL TIMING RULE: " + f"This scene's audio narration is EXACTLY {audio_duration:.3f} seconds long. " + "Ensure all `run_time` for animations and `self.wait()` pauses add up " + f"perfectly to exactly {audio_duration:.3f} seconds so the video matches the audio." + ) prompt_parts.append("\nGenerate the Manim code for this scene.") return "\n".join(prompt_parts) diff --git a/backend/app/agents/state.py b/backend/app/agents/state.py index 9ab3cf7..bff9fe6 100644 --- a/backend/app/agents/state.py +++ b/backend/app/agents/state.py @@ -65,6 +65,7 @@ class AgentState(TypedDict): # Code generation (for Sanika's coder node) # ========================================================================= current_scene_index: int + scene_audio_durations: dict[int, float] generated_codes: List[str] # ========================================================================= @@ -107,6 +108,7 @@ def create_initial_state( user_feedback=None, # Code generation current_scene_index=0, + scene_audio_durations={}, generated_codes=[], # Execution last_render_error=None, diff --git a/backend/app/agents/workflow.py b/backend/app/agents/workflow.py index 1f6bfb0..b11e5dd 100644 --- a/backend/app/agents/workflow.py +++ b/backend/app/agents/workflow.py @@ -17,6 +17,7 @@ from app.agents.nodes.scripter import write_scripts from app.agents.nodes.human_review import wait_for_approval from app.agents.nodes.coder import generate_code +from app.agents.nodes.audio_generator import generate_audio_node from app.sandbox.renderer import execute_and_check from app.sandbox.stitcher import finalize_video @@ -58,6 +59,7 @@ def create_workflow(): workflow.add_node("planner", plan_scenes) workflow.add_node("scripter", write_scripts) workflow.add_node("human_review", wait_for_approval) + workflow.add_node("audio_generator", generate_audio_node) workflow.add_node("coder", generate_code) workflow.add_node("renderer", execute_and_check) workflow.add_node("finalize", finalize_video) @@ -73,11 +75,14 @@ def create_workflow(): "human_review", route_after_review, { - "approved": "coder", # Go to code generation + "approved": "audio_generator", # Go to audio generation "rejected": END, } ) + # After audio generation, go to coder + workflow.add_edge("audio_generator", "coder") + # After coder, go to renderer workflow.add_edge("coder", "renderer") @@ -86,8 +91,8 @@ def create_workflow(): "renderer", route_after_render, { - "retry": "coder", # Error occurred, retry with reflector - "next_scene": "coder", # Move to next scene + "retry": "coder", # Error occurred, retry with coder + "next_scene": "audio_generator", # Move to next scene "finalize": "finalize", # All scenes done } ) diff --git a/backend/app/sandbox/renderer.py b/backend/app/sandbox/renderer.py index 2be6dca..be6f948 100644 --- a/backend/app/sandbox/renderer.py +++ b/backend/app/sandbox/renderer.py @@ -11,6 +11,7 @@ import logging from pathlib import Path from uuid import UUID +import subprocess from app.agents.state import AgentState from app.sandbox.executor import ManimExecutor @@ -90,6 +91,21 @@ async def upload_to_storage( return public_url +def _get_video_duration(file_path: Path) -> float: + """Get the duration of a video file using ffprobe.""" + try: + probe_cmd = [ + "ffprobe", "-v", "error", "-show_entries", + "format=duration", "-of", + "default=noprint_wrappers=1:nokey=1", str(file_path) + ] + dur = subprocess.check_output(probe_cmd, timeout=10).decode('utf-8').strip() + return float(dur) + except Exception as e: + logger.error(f"Failed to get video duration: {e}") + return 0.0 + + async def execute_and_check(state: AgentState) -> AgentState: """ Execute Manim code and check the result. @@ -135,6 +151,37 @@ async def execute_and_check(state: AgentState) -> AgentState: logger.info( f"Render successful for video {video_id}, scene {scene_index}" ) + + # --- TASK 1: SCENE-LEVEL DIAGNOSTICS LOGGING --- + try: + scripts = state.get("scripts", []) + narration = scripts[scene_index].get("narration", "") if scene_index < len(scripts) else "" + + audio_dur = state.get("scene_audio_durations", {}).get(scene_index, 0.0) + has_audio = audio_dur > 0 + + video_path = Path(result["video_path"]) + video_dur = _get_video_duration(video_path) + + diff = abs(audio_dur - video_dur) + status = "PASS" if diff <= 0.5 else "WARNING" + + logger.info("=== SCENE-LEVEL DIAGNOSTICS ===") + logger.info(f"Scene number: {scene_index + 1}") + logger.info(f"Narration generated: '{narration[:50]}...'") + logger.info(f"Audio generated: {'Yes' if has_audio else 'No'}") + logger.info(f"Audio duration: {audio_dur:.3f}s") + logger.info(f"Subtitle generated: {'Yes' if has_audio else 'No'}") + logger.info(f"Code generated: Yes") + logger.info(f"Video rendered: Yes") + logger.info(f"Render retries: {state.get('retry_count', 0)}") + logger.info(f"Video duration: {video_dur:.3f}s") + logger.info(f"Absolute difference: {diff:.3f}s") + logger.info(f"Final status: {status}") + logger.info("===============================") + except Exception as e: + logger.error(f"Failed to log diagnostics: {e}") + # ----------------------------------------------- try: # Upload to Supabase Storage diff --git a/backend/app/sandbox/stitcher.py b/backend/app/sandbox/stitcher.py index 1c343ac..5168a08 100644 --- a/backend/app/sandbox/stitcher.py +++ b/backend/app/sandbox/stitcher.py @@ -8,6 +8,7 @@ import asyncio import logging import subprocess +import re from pathlib import Path from uuid import UUID @@ -137,7 +138,9 @@ async def stitch_videos(self, video_id: str) -> str: # Download all segments and validate segment_paths = [] for scene in sorted(rendered_scenes, key=lambda s: s.scene_order): - local_path = work_dir / f"scene_{scene.scene_order}.mp4" + local_video_path = work_dir / f"scene_{scene.scene_order}.mp4" + local_audio_path = work_dir / f"scene_{scene.scene_order}.mp3" + local_vtt_path = work_dir / f"scene_{scene.scene_order}.vtt" logger.info( f"Downloading scene {scene.scene_order} from: " @@ -146,16 +149,64 @@ async def stitch_videos(self, video_id: str) -> str: success = await self.download_file( scene.video_segment_url, - local_path, + local_video_path, ) - if success and self._validate_video_file(local_path): - file_size = local_path.stat().st_size + # Audio and VTT are generated by TTS service and saved in the storage path. + # We don't download them from Supabase; we grab them from the local storage + # where audio_generator put them. + src_audio_path = self.storage_path / video_id / f"scene_{scene.scene_order - 1}" / f"scene_{scene.scene_order - 1}.mp3" + src_vtt_path = self.storage_path / video_id / f"scene_{scene.scene_order - 1}" / f"scene_{scene.scene_order - 1}.vtt" + + if src_audio_path.exists(): + local_audio_path.write_bytes(src_audio_path.read_bytes()) + else: + logger.warning(f"Audio not found at {src_audio_path}") + + if src_vtt_path.exists(): + local_vtt_path.write_bytes(src_vtt_path.read_bytes()) + else: + logger.warning(f"VTT not found at {src_vtt_path}") + + if success and self._validate_video_file(local_video_path): + file_size = local_video_path.stat().st_size logger.info( f"Scene {scene.scene_order} downloaded OK " f"({file_size} bytes)" ) - segment_paths.append(local_path) + + logger.info(f"=== STITCHER DIAGNOSTIC: SCENE {scene.scene_order} ===") + logger.info(f"Scene video selected: {local_video_path.name} ({'EXISTS' if local_video_path.exists() else 'MISSING'})") + logger.info(f"Scene audio selected: {local_audio_path.name} ({'EXISTS' if local_audio_path.exists() else 'MISSING'})") + logger.info(f"Subtitle selected: {local_vtt_path.name} ({'EXISTS' if local_vtt_path.exists() else 'MISSING'})") + logger.info("========================================") + + # Mux the video and audio + muxed_path = work_dir / f"scene_muxed_{scene.scene_order}.mp4" + mux_cmd = [ + "ffmpeg", "-y", + "-i", str(local_video_path) + ] + if local_audio_path.exists(): + mux_cmd.extend(["-i", str(local_audio_path)]) + mux_cmd.extend(["-c:v", "copy", "-c:a", "aac"]) + else: + mux_cmd.extend(["-c:v", "copy"]) + mux_cmd.append(str(muxed_path)) + + try: + result = await asyncio.get_event_loop().run_in_executor( + None, lambda: subprocess.run(mux_cmd, capture_output=True, text=True, timeout=60) + ) + if result.returncode == 0 and muxed_path.exists(): + segment_paths.append((muxed_path, local_vtt_path)) + logger.info(f"Mux success for scene {scene.scene_order}: True") + else: + logger.error(f"Failed to mux scene {scene.scene_order}: {result.stderr}") + logger.info(f"Mux success for scene {scene.scene_order}: False") + except Exception as e: + logger.error(f"Error muxing scene {scene.scene_order}: {e}") + logger.info(f"Mux success for scene {scene.scene_order}: False") else: logger.warning( f"Failed to download/validate scene {scene.scene_order}" @@ -168,25 +219,23 @@ async def stitch_videos(self, video_id: str) -> str: f"Successfully downloaded {len(segment_paths)}/{len(rendered_scenes)} segments" ) - # If only one segment, skip concat and upload directly + # If only one segment, skip concat and just use it if len(segment_paths) == 1: logger.info("Only one segment, skipping FFmpeg concat") - output_path = segment_paths[0] + output_path = segment_paths[0][0] + final_vtt_path = self._merge_vtts(segment_paths, work_dir) else: # Create FFmpeg concat file concat_file = work_dir / "concat.txt" with open(concat_file, "w") as f: - for path in segment_paths: - # Use absolute paths and escape single quotes + for path, _ in segment_paths: escaped_path = str(path.absolute()).replace("'", "'\\''") f.write(f"file '{escaped_path}'\n") logger.info(f"Concat file contents:") logger.info(concat_file.read_text()) - # Run FFmpeg to concatenate with re-encoding for robustness - # -c copy only works if all inputs have identical codec params; - # re-encoding handles different resolutions, framerates, etc. + # Run FFmpeg to concatenate (already muxed, can often use copy) output_path = work_dir / "final_video.mp4" ffmpeg_cmd = [ "ffmpeg", "-y", @@ -196,7 +245,7 @@ async def stitch_videos(self, video_id: str) -> str: "-c:v", "libx264", # Re-encode video for compatibility "-preset", "fast", # Balance speed vs compression "-crf", "23", # Quality (lower = better, 23 is default) - "-c:a", "aac", # Re-encode audio (if any) + "-c:a", "aac", # Re-encode audio "-movflags", "+faststart", # Enable streaming playback str(output_path), ] @@ -223,6 +272,8 @@ async def stitch_videos(self, video_id: str) -> str: except subprocess.TimeoutExpired: raise RuntimeError("FFmpeg timed out after 5 minutes") + + final_vtt_path = self._merge_vtts(segment_paths, work_dir) if not output_path.exists(): raise RuntimeError("FFmpeg did not produce output file") @@ -231,9 +282,18 @@ async def stitch_videos(self, video_id: str) -> str: logger.info( f"FFmpeg stitching complete: {output_path} ({output_size} bytes)" ) + + logger.info("=== STITCHER FINAL DIAGNOSTIC ===") + logger.info("Final concatenation success: True") + logger.info(f"Output video: {output_path.name}") + logger.info(f"Output VTT: {final_vtt_path.name if final_vtt_path and final_vtt_path.exists() else 'None'}") + logger.info("=================================") # Upload final video to Supabase Storage - final_url = await self._upload_final_video(output_path, video_id) + final_url = await self._upload_final_file(output_path, video_id, "final.mp4", "video/mp4") + if final_vtt_path and final_vtt_path.exists(): + vtt_url = await self._upload_final_file(final_vtt_path, video_id, "final.vtt", "text/vtt") + logger.info(f"VTT uploaded to {vtt_url}") # Update database with final URL await supabase_client.set_final_video_url(UUID(video_id), final_url) @@ -241,11 +301,79 @@ async def stitch_videos(self, video_id: str) -> str: logger.info(f"Video stitching complete: {final_url}") return final_url - - async def _upload_final_video( + + def _parse_time(self, t_str: str) -> float: + """Parse VTT time string (HH:MM:SS.mmm) to seconds.""" + h, m, s = t_str.split(":") + return int(h) * 3600 + int(m) * 60 + float(s) + + def _format_time(self, seconds: float) -> str: + """Format seconds to VTT time string (HH:MM:SS.mmm).""" + h = int(seconds // 3600) + m = int((seconds % 3600) // 60) + s = seconds % 60 + return f"{h:02d}:{m:02d}:{s:06.3f}" + + def _merge_vtts(self, segment_paths: list[tuple[Path, Path]], work_dir: Path) -> Path: + """Merge VTT files, shifting timestamps by cumulative duration.""" + final_vtt_path = work_dir / "final.vtt" + + cumulative_duration = 0.0 + + with open(final_vtt_path, "w", encoding="utf-8") as f: + f.write("WEBVTT\n\n") + + for mux_path, vtt_path in segment_paths: + if not vtt_path.exists(): + logger.warning(f"VTT missing: {vtt_path}") + # Estimate duration using ffprobe for the video segment + try: + probe_cmd = ["ffprobe", "-v", "error", "-show_entries", + "format=duration", "-of", + "default=noprint_wrappers=1:nokey=1", str(mux_path)] + dur = subprocess.check_output(probe_cmd).decode('utf-8').strip() + cumulative_duration += float(dur) + except Exception: + pass + continue + + content = vtt_path.read_text(encoding="utf-8") + lines = content.splitlines() + + max_time_in_this_vtt = 0.0 + + # regex to match VTT timestamps: 00:00:00.000 --> 00:00:01.000 + ts_pattern = re.compile(r"(\d{2}:\d{2}:\d{2}\.\d{3})\s*-->\s*(\d{2}:\d{2}:\d{2}\.\d{3})") + + for line in lines: + if line.startswith("WEBVTT"): + continue + + match = ts_pattern.search(line) + if match: + start_str, end_str = match.groups() + start_sec = self._parse_time(start_str) + cumulative_duration + end_sec = self._parse_time(end_str) + cumulative_duration + + max_time_in_this_vtt = max(max_time_in_this_vtt, self._parse_time(end_str)) + + shifted_start = self._format_time(start_sec) + shifted_end = self._format_time(end_sec) + f.write(f"{shifted_start} --> {shifted_end}\n") + else: + f.write(line + "\n") + + f.write("\n") + cumulative_duration += max_time_in_this_vtt + + return final_vtt_path + + async def _upload_final_file( self, local_path: Path, video_id: str, + filename: str, + content_type: str ) -> str: """ Upload the final video to Supabase Storage. @@ -261,21 +389,21 @@ async def _upload_final_video( settings = get_settings() file_content = local_path.read_bytes() - storage_path = f"videos/{video_id}/final.mp4" + storage_path = f"videos/{video_id}/{filename}" bucket_name = getattr(settings, "storage_bucket", "video-segments") try: client.storage.from_(bucket_name).upload( storage_path, file_content, - file_options={"content-type": "video/mp4"}, + file_options={"content-type": content_type}, ) except Exception as e: if "Duplicate" in str(e) or "already exists" in str(e).lower(): client.storage.from_(bucket_name).update( storage_path, file_content, - file_options={"content-type": "video/mp4"}, + file_options={"content-type": content_type}, ) else: raise diff --git a/backend/app/services/tts_service.py b/backend/app/services/tts_service.py new file mode 100644 index 0000000..042a4d1 --- /dev/null +++ b/backend/app/services/tts_service.py @@ -0,0 +1,82 @@ +""" +TTS Service using edge-tts. +""" + +import asyncio +import logging +import subprocess +from pathlib import Path +from typing import Tuple + +from mutagen.mp3 import MP3 + +logger = logging.getLogger(__name__) + + +async def generate_scene_audio( + narration_text: str, + output_dir: Path, + scene_index: int, + voice: str = "en-US-AriaNeural" +) -> Tuple[float, Path, Path]: + """ + Generate TTS audio and subtitles for a scene. + + Args: + narration_text: The text to be spoken. + output_dir: Directory to save the outputs. + scene_index: The index of the scene. + voice: The edge-tts voice to use. + + Returns: + Tuple containing (duration_in_seconds, audio_path, subtitle_path) + """ + output_dir.mkdir(parents=True, exist_ok=True) + + audio_path = output_dir / f"scene_{scene_index}.mp3" + vtt_path = output_dir / f"scene_{scene_index}.vtt" + + # Run edge-tts via subprocess + cmd = [ + "edge-tts", + "--voice", voice, + "--text", narration_text, + "--write-media", str(audio_path), + "--write-subtitles", str(vtt_path) + ] + + logger.info(f"Generating TTS for scene {scene_index}...") + + try: + # Run in executor to avoid blocking the event loop + result = await asyncio.get_event_loop().run_in_executor( + None, + lambda: subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=60 + ) + ) + + if result.returncode != 0: + logger.error(f"edge-tts failed: {result.stderr}") + raise RuntimeError(f"TTS generation failed: {result.stderr}") + + if not audio_path.exists(): + raise FileNotFoundError(f"edge-tts did not create {audio_path}") + + # Use mutagen to get the exact duration of the MP3 + audio = MP3(str(audio_path)) + duration = float(audio.info.length) + + logger.info(f"TTS generated for scene {scene_index}, duration: {duration:.3f}s") + + return duration, audio_path, vtt_path + + except subprocess.TimeoutExpired: + logger.error(f"edge-tts timed out for scene {scene_index}") + raise RuntimeError("TTS generation timed out") + except Exception as e: + logger.exception(f"Error during TTS generation: {e}") + raise diff --git a/backend/requirements.txt b/backend/requirements.txt index e5dfb02..6dd683e 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -28,3 +28,5 @@ httpx>=0.26.0 # Testing pytest>=7.4.0 pytest-asyncio>=0.21.0 +edge-tts>=6.1.12 +mutagen>=1.47.0 diff --git a/docs/original_workflow.md b/docs/original_workflow.md new file mode 100644 index 0000000..9399fcd --- /dev/null +++ b/docs/original_workflow.md @@ -0,0 +1,247 @@ +# LangGraph Multi-Agent Workflow Documentation + +## 1. High-Level Overview + +In this project, **LangGraph** orchestrates the multi-agent AI pipeline responsible for converting a user's prompt (e.g., a topic or concept) into a fully rendered, synchronized, educational video. + +LangGraph handles the following responsibilities: +- **State Management**: Passing a single, strongly-typed state object (`AgentState`) between various independent agents and services. +- **Workflow Routing**: Defining the exact execution order, loops (e.g., scene-by-scene processing), and retry paths for transient errors. +- **Human-in-the-Loop (HITL)**: Persisting the workflow state to disk/memory and pausing execution (`interrupt_before`) so a human can review and approve AI-generated scripts before costly code generation and video rendering occur. +- **Agent Orchestration**: Coordinating specialized nodes—Planner, Scripter, Coder, Audio Generator, Renderer, and Stitcher—so each handles a distinct part of the video creation process. + +## 2. Graph Structure + +The entire graph is defined as a `StateGraph(AgentState)` in `app/agents/workflow.py`. + +### Entry Point +- `retrieve_info` (Context Retrieval) + +### Nodes +- **`retrieve_info`**: Fetches RAG context from the syllabus PDF. +- **`planner`**: Creates the structured scene plan. +- **`scripter`**: Generates narration and visual descriptions. +- **`human_review`**: Wait state for user approval. +- **`audio_generator`**: Generates TTS audio and measures duration. +- **`coder`**: Generates Manim Python code. +- **`renderer`**: Executes Manim code to produce `.mp4` video segments. +- **`finalize`**: Stitches audio, video, and VTT subtitles together. + +### Edges +- `retrieve_info` -> `planner` +- `planner` -> `scripter` +- `scripter` -> `human_review` +- `audio_generator` -> `coder` +- `coder` -> `renderer` +- `finalize` -> `END` + +### Conditional Edges +- **From `human_review`**: + - If `approved` -> `audio_generator` + - If `rejected` -> `END` +- **From `renderer`**: + - If `retry` (error with retries left) -> `coder` + - If `next_scene` (more scenes remain) -> `audio_generator` + - If `finalize` (all scenes complete) -> `finalize` + +### Loops & Retry Paths +- **Scene Iteration Loop**: After a successful `renderer` execution, if there are more scenes, it loops back to `audio_generator` -> `coder` -> `renderer` for the next scene index. +- **Retry Path**: If `renderer` encounters an error and `retry_count < 3`, it routes back to `coder` to attempt generating fixed code. + +### Exit Conditions +- If the human user rejects the script in `human_review`. +- When all scenes have been processed and `finalize` completes stitching. + +## 3. State Object + +The workflow state is defined in `app/agents/state.py` using `TypedDict`. + +### `AgentState` +- `video_id` (str, required): UUID of the Supabase video record. Written at start, read by all nodes. +- `user_prompt` (str, required): The user's input topic. Written at start, read by `retrieve_info`, `planner`. +- `syllabus_context` (str, required): RAG context. Written by `retrieve_info`, read by `planner`. +- `video_title` (str, required): Title of the video. Written by `planner`, read by `scripter`. +- `topic_breakdown` (List[str], required): Learning objectives. Written by `planner`. +- `scene_plans` (List[ScenePlan], required): Planned scene structure. Written by `planner`, read by `scripter`. +- `scripts` (List[SceneScript], required): Narration and visual descriptions. Written by `scripter`, read by `coder`, `audio_generator`, `renderer`. +- `user_approved` (bool, required): Whether scripts were approved. Written by API/resume, read by `human_review`. +- `user_feedback` (Optional[str], optional): Feedback on rejection. Written by API/resume, read by `human_review`. +- `current_scene_index` (int, required): Current scene being processed. Written by `coder` (increments), read by `audio_generator`, `coder`, `renderer`. +- `scene_audio_durations` (dict[int, float], required): Measured TTS duration per scene. Written by `audio_generator`, read by `coder`. +- `generated_codes` (List[str], required): List of all generated Manim code strings. Written by `coder`, read by `renderer`. +- `last_render_error` (Optional[str], optional): Error from execution sandbox. Written by `renderer`, read by `coder` (for retries) and workflow router. +- `retry_count` (int, required): Current retry attempt for rendering. Written by `renderer`, read by workflow router. +- `all_scenes_done` (bool, required): Flag indicating all scenes are rendered. Written by `coder` / `renderer`, read by workflow router. + +### `ScenePlan` +- `scene_number` (int): Sequential index. +- `title` (str): Scene title. +- `key_concepts` (List[str]): Concepts to cover. +- `visual_type` (str): Type of visual. +- `duration_seconds` (int): Estimated length. + +### `SceneScript` +- `scene_order` (int): Sequential index. +- `narration` (str): Spoken script. +- `visual_description` (str): Prompt for Manim code. +- `duration_estimate` (int): Estimated duration in seconds. + +## 4. Node Documentation + +### `retrieve_context_node` (`app/agents/nodes/context.py`) +- **Purpose**: Fetches syllabus context from uploaded documents using RAG. +- **Inputs**: `video_id`, `user_prompt` +- **Outputs**: `syllabus_context` +- **Important Functions**: `retrieve_context` from `rag_service.py` + +### `plan_scenes` (`app/agents/nodes/planner.py`) +- **Purpose**: Generates a structured breakdown and timeline (scenes) for the video. +- **Inputs**: `video_id`, `user_prompt`, `syllabus_context` +- **Outputs**: `video_title`, `topic_breakdown`, `scene_plans` +- **Important Functions**: `_parse_json_response`, `update_video_status` +- **Prompt Used**: `PLANNER_SYSTEM_PROMPT` and `create_planner_prompt` +- **Model Used**: Configured via `create_llm("planner", temperature=0.7)`. Default expected is `llama-3.3-70b-versatile` (Groq). Uses JSON structured parsing manually via string formatting and parsing. + +### `write_scripts` (`app/agents/nodes/scripter.py`) +- **Purpose**: Generates the exact narration and visual descriptions for every planned scene. +- **Inputs**: `video_id`, `scene_plans`, `video_title` +- **Outputs**: `scripts` +- **Important Functions**: `create_scene` (Saves to DB), `_parse_json_response` +- **Prompt Used**: `SCRIPTER_SYSTEM_PROMPT` and `create_scripter_prompt` +- **Model Used**: Configured via `create_llm("scripter", temperature=0.7)`. Manual JSON parsing. + +### `wait_for_approval` (`app/agents/nodes/human_review.py`) +- **Purpose**: Processes the HITL resume signal. If approved, preps for coding. If rejected, fails the video. +- **Inputs**: `video_id`, `user_approved`, `user_feedback` +- **Outputs**: Resets `current_scene_index`, `retry_count`, `generated_codes` +- **Important Functions**: `update_video_status` + +### `generate_audio_node` (`app/agents/nodes/audio_generator.py`) +- **Purpose**: Pre-computes the TTS `.mp3` and `.vtt` to measure exact duration. +- **Inputs**: `current_scene_index`, `scripts`, `video_id` +- **Outputs**: `scene_audio_durations` +- **Important Functions**: `generate_scene_audio` from `tts_service.py` + +### `generate_code` (`app/agents/nodes/coder.py`) +- **Purpose**: Generates Python Manim code tailored precisely to the audio duration and visual description. +- **Inputs**: `video_id`, `current_scene_index`, `scripts`, `scene_audio_durations` +- **Outputs**: `generated_codes`, increments `current_scene_index` +- **Important Functions**: `clean_code_response`, `update_scene_code` +- **Prompt Used**: `CODER_SYSTEM_PROMPT` and `create_coder_prompt` +- **Model Used**: Configured via `create_llm("coder", temperature=0.2)`. + +### `execute_and_check` (`app/sandbox/renderer.py`) +- **Purpose**: Sandboxes and executes the generated Python code to render an MP4 segment via Manim. +- **Inputs**: `generated_codes`, `current_scene_index` (implicitly via list length), `video_id`, `scripts` +- **Outputs**: Uploads raw silent `.mp4` to Supabase DB. Updates `last_render_error`, `retry_count`, `all_scenes_done`. +- **Important Functions**: `ManimExecutor.execute()` + +### `finalize_video` (`app/sandbox/stitcher.py`) +- **Purpose**: Downloads rendered scenes, muxes them with TTS audio, concatenates them into a final video, merges and offsets `.vtt` subtitle files, and uploads results. +- **Inputs**: `video_id` +- **Outputs**: Updates DB `final_video_url`. +- **Important Functions**: `VideoStitcher.stitch_video()`, `_merge_vtts()` + +## 5. Execution Flow + +1. API calls `start_workflow`, initializing `AgentState` with user prompt. Workflow hits `retrieve_info`. +2. `retrieve_info` writes `syllabus_context`. +3. `planner` generates `scene_plans` and updates `AgentState`. +4. `scripter` uses `scene_plans` to populate `scripts`. The graph pauses at the `interrupt_before=["human_review"]` node. +5. User calls `resume_workflow` via API. Execution resumes at `human_review`. +6. If approved, graph routes to `audio_generator`. +7. `audio_generator` generates `.mp3` for scene 0 and records duration. +8. `coder` uses duration to generate Manim code for scene 0. `current_scene_index` increments to 1. +9. `renderer` executes code for scene 0. +10. `route_after_render` conditionally routes back to `audio_generator` for scene 1. +11. Steps 7-10 loop until all scenes are processed. +12. `route_after_render` routes to `finalize`. +13. `finalize` builds the final artifact and workflow hits `END`. + +## 6. Error Handling + +- **Coder Retries**: The `coder.py` node has an internal `for attempt in range(MAX_RETRIES)` loop equipped with exponential backoff designed to handle transient provider issues (e.g., rate limits, HTTP 524, 529, 429). +- **Renderer Retries**: If the executed code fails in `renderer.py` (e.g., Manim syntax error), it sets `last_render_error` and increments `retry_count`. The graph's conditional edge routes back to `coder`. The `coder` can (if implemented) read `last_render_error` to fix its code, and tries again up to 3 times. +- **Fallback Behavior**: If RAG fails, `retrieve_info` swallows the exception and returns an empty context string. If TTS fails, `audio_generator` falls back to estimating duration based on word count (0.4s per word). If parsing JSON fails in Planner/Scripter, it throws a `ValueError` (or generates a placeholder script in Scripter) to prevent crashing the chain blindly. + +## 7. Graph Visualization + +```mermaid +graph TD + START((START)) --> retrieve_info + retrieve_info --> planner + planner --> scripter + scripter --> human_review + + human_review -- approved --> audio_generator + human_review -- rejected --> END((END)) + + audio_generator --> coder + coder --> renderer + + renderer -- retry --> coder + renderer -- next_scene --> audio_generator + renderer -- finalize --> finalize + + finalize --> END((END)) +``` + +```mermaid +sequenceDiagram + participant User + participant API + participant Graph as LangGraph Workflow + participant Planner + participant Scripter + participant Coder + participant Renderer + participant Stitcher + + User->>API: POST /api/videos + API->>Graph: start_workflow() + Graph->>Planner: plan_scenes() + Graph->>Scripter: write_scripts() + Graph-->>API: Paused (interrupt_before) + + User->>API: POST /api/videos/{id}/approve + API->>Graph: resume_workflow() + + loop Per Scene + Graph->>Coder: audio_generator() + generate_code() + Graph->>Renderer: execute_and_check() + alt Syntax Error + Renderer-->>Coder: Retry (max 3) + end + end + + Graph->>Stitcher: finalize_video() + Stitcher-->>User: Final MP4 & VTT +``` + +## 8. File Structure + +- `app/agents/workflow.py`: Graph definition, routing logic, and workflow lifecycle functions (`start_workflow`, `resume_workflow`). +- `app/agents/state.py`: Definition of `AgentState`, `SceneScript`, and `ScenePlan` TypedDicts. +- `app/agents/nodes/context.py`: RAG integration node. +- `app/agents/nodes/planner.py`: Planner LLM execution node. +- `app/agents/nodes/scripter.py`: Scripter LLM execution node. +- `app/agents/nodes/human_review.py`: State processing after HITL resume. +- `app/agents/nodes/audio_generator.py`: TTS generation and duration tracking node. +- `app/agents/nodes/coder.py`: Manim code generation node. +- `app/sandbox/renderer.py`: Local execution sandbox for rendering Manim scenes. +- `app/sandbox/stitcher.py`: FFmpeg assembly logic for video, audio, and subtitles. +- `app/agents/prompts/*.py`: Specialized system prompts and template constructors for each agent. + +## 9. Dependencies + +- **LLM Factory** (`app/services/llm_factory.py`): Abstracted function to instantiate `langchain` chat models dynamically using either Groq or OpenRouter, falling back to global settings if agent-specific settings are missing. +- **Supabase Client** (`app/services/supabase_client.py`): Database helpers to actively persist the generated videos, plans, scripts, and status updates as they happen inside the nodes. +- **RAG Service** (`app/services/rag_service.py`): Used by context node to embed and query vector storage. +- **TTS Service** (`app/services/tts_service.py`): Leverages `edge-tts` and `mutagen.mp3` for precise audio synthesis and time duration calculations. +- **ManimExecutor** (`app/sandbox/executor.py`): Wraps python `subprocess` / Docker to safely execute the generated Python code and capture errors/logs. + +## 10. Notes + +- The workflow heavily leverages Supabase for persistence. Nodes often push updates to the database (e.g. `update_video_status()`, `create_scene()`) independently of the LangGraph state. +- The `human_review` node requires `MemorySaver()` checkpointer in LangGraph to function since `interrupt_before` halts process execution entirely until triggered externally. +- Code generation retry logic (`coder.py`) for API transient errors is isolated entirely within the `generate_code` node and does NOT iterate through the wider LangGraph edges. LangGraph's routing (`retry` edge) is reserved exclusively for handling execution/rendering errors (like Python Tracebacks from Manim). From 0bc7e39c7fab32d15e2fce4f046bd63590624616 Mon Sep 17 00:00:00 2001 From: Mustansir Rangwala <119410651+mustansirr@users.noreply.github.com> Date: Mon, 29 Jun 2026 22:23:10 +0530 Subject: [PATCH 2/2] fix: make workflow start non-blocking and improve coder prompts --- backend/app/agents/prompts/coder_prompts.py | 1 + backend/app/api/routes/videos.py | 40 +++++++++++---------- 2 files changed, 23 insertions(+), 18 deletions(-) diff --git a/backend/app/agents/prompts/coder_prompts.py b/backend/app/agents/prompts/coder_prompts.py index 1edd25d..f5f87b1 100644 --- a/backend/app/agents/prompts/coder_prompts.py +++ b/backend/app/agents/prompts/coder_prompts.py @@ -61,6 +61,7 @@ 6. If a scene calls for a complex diagram (e.g., an ant, a human body, a logo), approximate it using combinations of Circle, Ellipse, Rectangle, Line, Arc, Polygon, Arrow, Dot, CurvedArrow, and other built-in shapes. +7. DO NOT hallucinate shapes like `DashedCircle` or `DashedSquare`. For dashed borders, wrap a built-in shape in `DashedVMobject`, e.g., `DashedVMobject(Circle())` or `DashedVMobject(Square())`. COMMON PATTERNS: - Text: Text("content", font_size=36) diff --git a/backend/app/api/routes/videos.py b/backend/app/api/routes/videos.py index cdb7d5c..2193aab 100644 --- a/backend/app/api/routes/videos.py +++ b/backend/app/api/routes/videos.py @@ -7,7 +7,7 @@ from typing import List, Optional from uuid import UUID -from fastapi import APIRouter, HTTPException, Query, status +from fastapi import APIRouter, HTTPException, Query, status, BackgroundTasks from app.models.schemas import ( SceneResponse, @@ -103,33 +103,37 @@ async def create_video_request( and scripting, then pause for human review. """, ) -async def start_video_workflow(video_id: str): +async def start_video_workflow(video_id: str, background_tasks: BackgroundTasks): """Start the generation workflow for a video.""" video_uuid = validate_uuid(video_id, "video_id") # Verify video exists video = await get_video_or_404(video_uuid) - # Start the LangGraph workflow - from app.agents.workflow import start_workflow - try: - await start_workflow( - video_id=str(video_uuid), - user_prompt=video.prompt, - syllabus_context="" # retrieve_context_node will fetch from RAG - ) - except Exception as e: + async def _run_start_workflow_bg(): + from app.agents.workflow import start_workflow import logging - logging.error(f"Workflow error for video {video_id}: {e}") - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"Workflow failed to start: {str(e)}" - ) + try: + await start_workflow( + video_id=str(video_uuid), + user_prompt=video.prompt, + syllabus_context="" # retrieve_context_node will fetch from RAG + ) + except Exception as e: + logging.error(f"Workflow error for video {video_id}: {e}") + from app.services.supabase_client import update_video_status + try: + await update_video_status(video_uuid, VideoStatus.FAILED) + except Exception as update_err: + logging.error(f"Failed to update status to FAILED: {update_err}") + + # Start the LangGraph workflow in the background + background_tasks.add_task(_run_start_workflow_bg) return { "video_id": str(video_uuid), - "status": VideoStatus.WAITING_APPROVAL.value, - "message": "Scripts ready for review" + "status": VideoStatus.PLANNING.value, + "message": "Workflow started in background" }