I recently faced a familiar challenge: how do you keep a content pipeline fresh and engaging without an army of human editors? Specifically, we needed to transform the firehose of technical blog posts into short, compelling video scripts, ready for automated production. The problem isn't just about extracting information; it's about synthesizing narratives, ensuring creative coherence, and, most critically, guaranteeing the output is consistently structured and validated to feed into our automated video generation tools. This isn't a job for a simple prompt; it's a job for an agentic system that can understand, create, and, crucially, self-correct. If you're wrestling with the unpredictable nature of LLM outputs and the rigid demands of downstream production systems, this post will walk you through building a resilient, self-correcting pipeline that bridges that gap, turning raw RSS feeds into production-ready video scripts.
Key Takeaways
- How to design a multi-agent LLM system for complex, creative content generation tasks.
- Strategies for enforcing structured output from LLMs using Pydantic and self-correction agents.
- Best practices for integrating LLM-generated content with external tools or APIs (e.g., OpenMontage).
- Techniques for robustly ingesting and processing dynamic, unstructured RSS feed data.
- Understanding the tradeoffs of agentic systems in terms of cost, latency, and creative control.
The Problem
The core problem isn't just generating content; it's generating production-ready content. Imagine you're pulling the latest technical insights from a blog like Cloudflare's, aiming to distill each post into a concise video script. A raw RSS feed is a stream of unstructured text, often with varying formats and content depth. Naively piping this into an LLM might give you something that looks like a script, but it will inevitably suffer from hallucinations, inconsistent formatting, or outright schema violations. Our video generation tool, let's call it OpenMontage, expects a very specific JSON structure – scenes, voiceovers, visual cues, timings. Any deviation means a broken pipeline, manual intervention, and wasted resources. This system needs to be robust, capable of understanding the source material, synthesizing a creative narrative, and then meticulously validating its own output against a strict schema, correcting itself when it fails.
Data and Sources
For this project, we're tapping into the pulse of current web infrastructure developments by using the Cloudflare Blog's RSS feed. This provides a dynamic, real-world data source that constantly updates, reflecting new announcements, technical deep dives, and industry insights.
- Cloudflare Blog RSS Feed: https://blog.cloudflare.com/rss/
feedparserofficial documentation: https://pythonhosted.org/feedparser/- OpenAI API documentation (or similar LLM provider): https://platform.openai.com/docs/api-reference
- Pydantic documentation: https://docs.pydantic.dev/
- (Hypothetical) OpenMontage API specification: This serves as an example of an external tool's strict input requirements, demonstrating how our validated script integrates.
Data accessed on YYYY-MM-DD. The data from Cloudflare Blog RSS is real-time. The content generated will reflect the latest posts available at the time of execution.
Step 1 — Ingesting Dynamic RSS Feeds with Resilience
The first hurdle in any automated content pipeline is reliably getting the source material. RSS feeds, while standardized, can be temperamental. Network hiccups, malformed XML, or even temporary service outages can derail the entire process. Our sub-problem here is to fetch and parse the latest content from a potentially volatile RSS feed without crashing the pipeline, while also gracefully handling errors.
To tackle this, I implemented a robust fetching mechanism using requests with retry logic and explicit error checking with feedparser. This ensures that transient network issues don't immediately fail our process, and that we detect malformed feeds early.
import requests
import feedparser
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def fetch_rss_feed(url: str, retries: int = 3, backoff_factor: float = 0.5) -> feedparser.FeedParserDict:
"""
Fetches and parses an RSS feed with retry logic for network resilience.
"""
session = requests.Session()
retry_strategy = Retry(
total=retries,
backoff_factor=backoff_factor,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["HEAD", "GET", "OPTIONS"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
try:
logging.info(f"Attempting to fetch RSS feed from: {url}")
response = session.get(url, timeout=10)
response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
feed = feedparser.parse(response.content)
if feed.bozo:
logging.warning(f"Malformed RSS feed detected from {url}: {feed.bozo_exception}")
# Depending on severity, you might want to raise an error or try to proceed
# For this example, we'll log and proceed, but in production, careful handling is key.
# Example: raise ValueError(f"Malformed RSS feed: {feed.bozo_exception}")
return feed
except requests.exceptions.Timeout:
logging.error(f"Request timed out while fetching RSS feed from {url}")
raise
except requests.exceptions.RequestException as e:
logging.error(f"Network error fetching RSS feed from {url}: {e}")
raise
except Exception as e:
logging.error(f"An unexpected error occurred during RSS feed fetching/parsing: {e}")
raise
Here, I'm using requests.Session with a custom HTTPAdapter configured for retries. The Retry strategy specifically targets common transient errors like rate limits (429) and server errors (5xx). After fetching the raw content, feedparser.parse() handles the XML parsing. Crucially, I check feed.bozo, which indicates if the feed was malformed. While I'm just logging a warning here, in a production environment, you might decide to quarantine or skip such entries, or even attempt a fallback parsing strategy. This builds directly on patterns for resilient API interaction we explored in Architecting Resilient API Sessions.
Step 2 — Defining the Production-Ready Video Script Schema
The biggest challenge with LLM-generated content is its inherent unpredictability. For integration with a downstream system like OpenMontage, we need ironclad guarantees about the output structure. This is where Pydantic shines. Our sub-problem is to define a strict, type-safe schema that the LLM must adhere to, ensuring every generated script is valid before it even leaves our agent's hands.
I defined a set of Pydantic models that precisely map to OpenMontage's expected input. This includes nested structures, type annotations, and validation rules for fields like minimum scene count or positive duration values.
from pydantic import BaseModel, Field, PositiveInt, ValidationError
from typing import List, Optional
class VisualCue(BaseModel):
"""
Represents a visual element or instruction for a scene.
"""
description: str = Field(..., description="A clear description of the visual element or action.")
type: str = Field("image", description="The type of visual cue (e.g., 'image', 'animation', 'text_overlay').")
duration_seconds: PositiveInt = Field(..., description="Duration in seconds this visual cue should be displayed.")
class Scene(BaseModel):
"""
Represents a single scene in the video script.
"""
scene_number: PositiveInt = Field(..., description="Unique identifier for the scene.")
voiceover_text: str = Field(..., min_length=20, description="The spoken narration for this scene.")
visual_cues: List[VisualCue] = Field(..., min_items=1, description="A list of visual elements for the scene.")
estimated_duration_seconds: PositiveInt = Field(..., description="The estimated total duration of the scene in seconds.")
class VideoScript(BaseModel):
"""
The complete video script, structured for automated video generation.
"""
title: str = Field(..., min_length=10, description="The title of the video.")
target_audience: str = Field(..., description="Who is this video primarily for?")
key_message: str = Field(..., min_length=20, description="The core takeaway message of the video.")
scenes: List[Scene] = Field(..., min_items=3, description="A list of individual scenes comprising the video.")
total_estimated_duration_seconds: PositiveInt = Field(..., description="The sum of all scene durations.")
This snippet defines VisualCue, Scene, and the top-level VideoScript. Notice the use of Field for descriptions, min_length, min_items, and PositiveInt. These are not just documentation; they are strict validation rules. If an LLM generates a script with zero scenes or a negative duration, Pydantic will raise a ValidationError, preventing invalid data from proceeding. This builds on our previous deep dive into structured data extraction with Architecting Resilient API Clients, extending it to generative outputs.
Step 3 — Architecting the Self-Correcting Multi-Agent Orchestrator
Even with a perfectly defined Pydantic schema, LLMs can be stubborn. They might produce malformed JSON, miss required fields, or simply get the types wrong. The sub-problem here is to design an agentic system that not only generates the script but also validates it and, if necessary, enters a feedback loop to self-correct until the output conforms to our schema.
My approach involves an AgenticVideoScriptGenerator class that orchestrates LLM calls. It first attempts to generate the script. If Pydantic validation fails, it constructs a specific "correction" prompt, feeding back the error message and the invalid output to the LLM, giving it another chance. This pattern is crucial for robust production systems.
import json
import time # For simulating LLM call latency
# Assume VideoScript, Scene, VisualCue are defined as above
class MockLLM:
"""
A mock LLM to simulate OpenAI API calls for demonstration purposes.
In a real scenario, this would be `openai.OpenAI().chat.completions.create`.
"""
def __init__(self, latency: float = 0.5):
self.latency = latency
def create_chat_completion(self, messages: List[dict], model: str = "gpt-4o", temperature: float = 0.7):
time.sleep(self.latency) # Simulate API call latency
# This is where you'd integrate with a real LLM like OpenAI
# For demonstration, we'll have a simplified mock response.
# The actual LLM would process `messages` and return JSON.
# For the initial generation, we simulate a good first attempt or a bad one.
# For correction, we'd expect it to use the error info.
# Example of a hardcoded mock response for a successful script
# In a real system, you'd make a call to `openai.OpenAI().chat.completions.create`
# and parse its `message.content`.
if "Please correct the following JSON" in messages[-1]["content"]:
logging.info("Mock LLM: Simulating correction attempt...")
# Simulate a corrected output
return {
"choices": [{
"message": {
"content": json.dumps({
"title": "Cloudflare's New Logpush Transformers: SQL for Your Logs",
"target_audience": "DevOps Engineers, SREs, Data Analysts",
"key_message": "Cloudflare's Logpush now allows filtering, reshaping, and redacting logs with SQL via Transformers, powered by Apache DataFusion.",
"scenes": [
{
"scene_number": 1,
"voiceover_text": "Managing vast log data can be a nightmare. Cloudflare's Logpush service helps, but what if you need to refine that data before it even leaves?",
"visual_cues": [
{"description": "Animated graphic of log data flowing into a funnel.", "type": "animation", "duration_seconds": 5}
],
"estimated_duration_seconds": 5
},
{
"scene_number": 2,
"voiceover_text": "Introducing Logpush Transformers! Now, you can use SQL to filter, reshape, and redact your log streams directly within Cloudflare.",
"visual_cues": [
{"description": "SQL query interface overlayed on Cloudflare dashboard.", "type": "image", "duration_seconds": 6}
],
"estimated_duration_seconds": 6
},
{
"scene_number": 3,
"voiceover_text": "Powered by Apache DataFusion, this feature ensures you only export the most relevant, secure data, saving on storage and processing.",
"visual_cues": [
{"description": "Diagram showing DataFusion processing logs.", "type": "animation", "duration_seconds": 7},
{"description": "Text overlay: 'Powered by Apache DataFusion'", "type": "text_overlay", "duration_seconds": 4}
],
"estimated_duration_seconds": 7
}
],
"total_estimated_duration_seconds": 18
})
}
}]
}
# Simulate an initial, potentially invalid, response
# This is where an LLM might make a mistake, e.g., missing a required field or wrong type.
# For example, let's make it miss 'total_estimated_duration_seconds' or have too few scenes initially.
logging.info("Mock LLM: Simulating initial script generation...")
return {
"choices": [{
"message": {
"content": json.dumps({
"title": "Cloudflare Logpush Gets SQL Power",
"target_audience": "Developers, Ops Teams",
"key_message": "Cloudflare Logpush now offers advanced log transformation with SQL.",
"scenes": [ # Intentionally too few scenes for initial generation to force correction
{
"scene_number": 1,
"voiceover_text": "Cloudflare Logpush just got a major upgrade, letting you filter and transform logs with SQL.",
"visual_cues": [
{"description": "Cloudflare logo and SQL code snippets.", "type": "image", "duration_seconds": 5}
],
"estimated_duration_seconds": 5
}
]
# "total_estimated_duration_seconds" is missing here to simulate an error
})
}
}]
}
class AgenticVideoScriptGenerator:
"""
Orchestrates LLM calls to generate and self-correct video scripts based on Pydantic schema.
"""
def __init__(self, llm_client: MockLLM, schema: type[BaseModel] = VideoScript, max_retries: int = 3):
self.llm_client = llm_client
self.schema = schema
self.max_retries = max_retries
self.system_prompt = (
"You are an expert video scriptwriter for technical content. "
"Your task is to convert a given article into a structured video script. "
"The output MUST be a JSON object strictly conforming to the following Pydantic schema. "
"Ensure all fields are present and correctly typed. Pay close attention to nested objects, "
"minimum lengths, and item counts for lists. The video should be engaging and informative. "
"Here is the Pydantic schema:\n\n"
f"```json\n{self.schema.model_json_schema(indent=2)}\n```\n"
)
self.correction_prompt_template = (
"The previous JSON output failed validation with the following error:\n"
"```\n{error_message}\n```\n"
"Please correct the following JSON to strictly adhere to the schema and fix the validation errors. "
"Return ONLY the corrected JSON object. Do not include any other text or explanation.\n"
"Invalid JSON:\n"
"```json\n{invalid_json}\n```\n"
)
def _call_llm(self, messages: List[dict]) -> str:
"""Helper to make LLM call and extract content."""
response = self.llm_client.create_chat_completion(messages=messages)
return response["choices"][0]["message"]["content"]
def generate_script_from_article(self, article_title: str, article_content: str) -> VideoScript:
"""
Generates a video script from an article, with self-correction.
"""
user_prompt = (
f"Generate a video script for the article titled '{article_title}'. "
"Focus on the key takeaways and present them in an engaging way, suitable for a technical audience. "
"Here is the article content:\n\n"
f"{article_content}\n\n"
"Ensure the output is a valid JSON object strictly following the provided schema. "
"Make sure to include at least 3 scenes."
)
messages = [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": user_prompt}
]
for attempt in range(self.max_retries):
logging.info(f"Attempt {attempt + 1}/{self.max_retries} to generate/correct script for '{article_title}'...")
try:
raw_json_output = self._call_llm(messages)
script_data = json.loads(raw_json_output)
# Calculate total duration if not provided by LLM (a common error)
if "total_estimated_duration_seconds" not in script_data and "scenes" in script_data:
script_data["total_estimated_duration_seconds"] = sum(s.get("estimated_duration_seconds", 0) for s in script_data["scenes"])
validated_script = self.schema.model_validate(script_data)
logging.info("Script successfully generated and validated.")
return validated_script
except json.JSONDecodeError as e:
error_message = f"JSON decoding failed: {e}"
logging.warning(error_message)
if attempt == self.max_retries - 1:
logging.error(f"Failed to generate valid JSON after {self.max_retries} attempts.")
raise
correction_user_prompt = self.correction_prompt_template.format(
error_message=error_message, invalid_json=raw_json_output
)
messages.append({"role": "assistant", "content": raw_json_output})
messages.append({"role": "user", "content": correction_user_prompt})
except ValidationError as e:
error_message = f"Pydantic validation failed:\n{e.json()}"
logging.warning(error_message)
if attempt == self.max_retries - 1:
logging.error(f"Failed to generate valid script after {self.max_retries} attempts.")
raise
correction_user_prompt = self.correction_prompt_template.format(
error_message=error_message, invalid_json=raw_json_output
)
messages.append({"role": "assistant", "content": raw_json_output})
messages.append({"role": "user", "content": correction_user_prompt})
except Exception as e:
logging.error(f"An unexpected error occurred during script generation: {e}")
raise
raise Exception("Failed to generate a valid video script after multiple attempts.")
The AgenticVideoScriptGenerator class initializes with a MockLLM (which would be replaced by a real OpenAI client in production) and our VideoScript schema. The system_prompt is critical: it embeds the Pydantic schema as a JSON string, explicitly instructing the LLM on the required output format. The generate_script_from_article method attempts generation, then uses a try-except block to catch both json.JSONDecodeError (if the LLM output isn't even valid JSON) and ValidationError (if it's valid JSON but doesn't match our schema). If an error occurs, it appends the error message and the faulty JSON to the conversation history with a dedicated correction prompt, giving the LLM context to fix its mistake. This iterative refinement is a cornerstone of robust agent design, something we touched upon in Beyond Simple Prompts.
Step 4 — Integrating with OpenMontage (Mocked for Output Validation)
With a validated VideoScript object in hand, the final step is to integrate it with our hypothetical downstream video production system, OpenMontage. The sub-problem here is to demonstrate a clean, type-safe interface for consuming our LLM's output, ensuring that the contract between our agent pipeline and the external tool is always met.
I've created a mock OpenMontageClient class. Its generate_video method simply accepts our Pydantic VideoScript object, prints its contents, and simulates a successful API call. This clearly shows how the validated object would be passed along.
# Assume VideoScript is defined as above
class OpenMontageClient:
"""
A mock client for the OpenMontage video generation API.
In a real scenario, this would make actual API calls.
"""
def generate_video(self, script: VideoScript) -> dict:
"""
Simulates sending a validated VideoScript to the OpenMontage API.
"""
logging.info(f"Sending validated script to OpenMontage for video generation: '{script.title}'")
# In a real scenario, you'd convert script.model_dump_json() to a POST request body
# and send it to the OpenMontage API endpoint.
# Example: requests.post("https://api.openmontage.com/v1/videos", json=script.model_dump())
logging.info("OpenMontage received script. Simulating successful video generation request.")
return {"status": "success", "video_id": "mock_video_123", "script_received": script.model_dump()}
The OpenMontageClient's generate_video method explicitly types its input as script: VideoScript. This isn't just for documentation; it's a compile-time guarantee that only fully validated, Pydantic-conforming script objects can be passed to it. This design pattern ensures that our rigorous validation efforts upstream aren't undone by an ill-defined interface, providing a robust contract for integration.
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
import requests
import feedparser
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import logging
import json
import time
from pydantic import BaseModel, Field, PositiveInt, ValidationError
from typing import List, Optional
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# --- Step 1: Ingesting Dynamic RSS Feeds with Resilience ---
def fetch_rss_feed(url: str, retries: int = 3, backoff_factor: float = 0.5) -> feedparser.FeedParserDict:
"""
Fetches and parses an RSS feed with retry logic for network resilience.
"""
session = requests.Session()
retry_strategy = Retry(
total=retries,
backoff_factor=backoff_factor,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["HEAD", "GET", "OPTIONS"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
try:
logging.info(f"Attempting to fetch RSS feed from: {url}")
response = session.get(url, timeout=10)
response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
feed = feedparser.parse(response.content)
if feed.bozo:
logging.warning(f"Malformed RSS feed detected from {url}: {feed.bozo_exception}")
return feed
except requests.exceptions.Timeout:
logging.error(f"Request timed out while fetching RSS feed from {url}")
raise
except requests.exceptions.RequestException as e:
logging.error(f"Network error fetching RSS feed from {url}: {e}")
raise
except Exception as e:
logging.error(f"An unexpected error occurred during RSS feed fetching/parsing: {e}")
raise
# --- Step 2: Defining the Production-Ready Video Script Schema ---
class VisualCue(BaseModel):
"""
Represents a visual element or instruction for a scene.
"""
description: str = Field(..., description="A clear description of the visual element or action.")
type: str = Field("image", description="The type of visual cue (e.g., 'image', 'animation', 'text_overlay').")
duration_seconds: PositiveInt = Field(..., description="Duration in seconds this visual cue should be displayed.")
class Scene(BaseModel):
"""
Represents a single scene in the video script.
"""
scene_number: PositiveInt = Field(..., description="Unique identifier for the scene.")
voiceover_text: str = Field(..., min_length=20, description="The spoken narration for this scene.")
visual_cues: List[VisualCue] = Field(..., min_items=1, description="A list of visual elements for the scene.")
estimated_duration_seconds: PositiveInt = Field(..., description="The estimated total duration of the scene in seconds.")
class VideoScript(BaseModel):
"""
The complete video script, structured for automated video generation.
"""
title: str = Field(..., min_length=10, description="The title of the video.")
target_audience: str = Field(..., description="Who is this video primarily for?")
key_message: str = Field(..., min_length=20, description="The core takeaway message of the video.")
scenes: List[Scene] = Field(..., min_items=3, description="A list of individual scenes comprising the video.")
total_estimated_duration_seconds: PositiveInt = Field(..., description="The sum of all scene durations.")
# --- Step 3: Architecting the Self-Correcting Multi-Agent Orchestrator ---
class MockLLM:
"""
A mock LLM to simulate OpenAI API calls for demonstration purposes.
In a real scenario, this would be `openai.OpenAI().chat.completions.create`.
"""
def __init__(self, latency: float = 0.5):
self.latency = latency
self.call_count = 0 # To simulate different responses for initial vs. correction
def create_chat_completion(self, messages: List[dict], model: str = "gpt-4o", temperature: float = 0.7):
time.sleep(self.latency)
self.call_count += 1
# Simulate an initial response that needs correction (e.g., too few scenes)
if self.call_count == 1:
logging.info("Mock LLM: Simulating initial script generation (will intentionally be invalid)...")
return {
"choices": [{
"message": {
"content": json.dumps({
"title": "Cloudflare Logpush Gets SQL Power",
"target_audience": "Developers, Ops Teams",
"key_message": "Cloudflare Logpush now offers advanced log transformation with SQL.",
"scenes": [ # Intentionally too few scenes for initial generation to force correction
{
"scene_number": 1,
"voiceover_text": "Cloudflare Logpush just got a major upgrade, letting you filter and transform logs with SQL.",
"visual_cues": [
{"description": "Cloudflare logo and SQL code snippets.", "type": "image", "duration_seconds": 5}
],
"estimated_duration_seconds": 5
}
]
# "total_estimated_duration_seconds" is missing here to simulate another error
})
}
}]
}
# Simulate a corrected response
logging.info("Mock LLM: Simulating correction attempt (will be valid)...")
return {
"choices": [{
"message": {
"content": json.dumps({
"title": "Cloudflare's New Logpush Transformers: SQL for Your Logs",
"target_audience": "DevOps Engineers, SREs, Data Analysts",
"key_message": "Cloudflare's Logpush now allows filtering, reshaping, and redacting logs with SQL via Transformers, powered by Apache DataFusion.",
"scenes": [
{
"scene_number": 1,
"voiceover_text": "Managing vast log data can be a nightmare. Cloudflare's Logpush service helps, but what if you need to refine that data before it even leaves?",
"visual_cues": [
{"description": "Animated graphic of log data flowing into a funnel.", "type": "animation", "duration_seconds": 5}
],
"estimated_duration_seconds": 5
},
{
"scene_number": 2,
"voiceover_text": "Introducing Logpush Transformers! Now, you can use SQL to filter, reshape, and redact your log streams directly within Cloudflare.",
"visual_cues": [
{"description": "SQL query interface overlayed on Cloudflare dashboard.", "type": "image", "duration_seconds": 6}
],
"estimated_duration_seconds": 6
},
{
"scene_number": 3,
"voiceover_text": "Powered by Apache DataFusion, this feature ensures you only export the most relevant, secure data, saving on storage and processing.",
"visual_cues": [
{"description": "Diagram showing DataFusion processing logs.", "type": "animation", "duration_seconds": 7},
{"description": "Text overlay: 'Powered by Apache DataFusion'", "type": "text_overlay", "duration_seconds": 4}
],
"estimated_duration_seconds": 7
}
],
"total_estimated_duration_seconds": 18
})
}
}]
}
class AgenticVideoScriptGenerator:
"""
Orchestrates LLM calls to generate and self-correct video scripts based on Pydantic schema.
"""
def __init__(self, llm_client: MockLLM, schema: type[BaseModel] = VideoScript, max_retries: int = 3):
self.llm_client = llm_client
self.schema = schema
self.max_retries = max_retries
self.system_prompt = (
"You are an expert video scriptwriter for technical content. "
"Your task is to convert a given article into a structured video script. "
"The output MUST be a JSON object strictly conforming to the following Pydantic schema. "
"Ensure all fields are present and correctly typed. Pay close attention to nested objects, "
"minimum lengths, and item counts for lists. The video should be engaging and informative. "
"Here is the Pydantic schema:\n\n"
f"```json\n{json.dumps(self.schema.model_json_schema(), indent=2)}\n```\n"
)
self.correction_prompt_template = (
"The previous JSON output failed validation with the following error:\n"
"```\n{error_message}\n```\n"
"Please correct the following JSON to strictly adhere to the schema and fix the validation errors. "
"Return ONLY the corrected JSON object. Do not include any other text or explanation.\n"
"Invalid JSON:\n"
"```json\n{invalid_json}\n```\n"
)
def _call_llm(self, messages: List[dict]) -> str:
"""Helper to make LLM call and extract content."""
response = self.llm_client.create_chat_completion(messages=messages)
return response["choices"][0]["message"]["content"]
def generate_script_from_article(self, article_title: str, article_content: str) -> VideoScript:
"""
Generates a video script from an article, with self-correction.
"""
user_prompt = (
f"Generate a video script for the article titled '{article_title}'. "
"Focus on the key takeaways and present them in an engaging way, suitable for a technical audience. "
"Here is the article content:\n\n"
f"{article_content}\n\n"
"Ensure the output is a valid JSON object strictly following the provided schema. "
"Make sure to include at least 3 scenes."
)
messages = [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": user_prompt}
]
raw_json_output = "" # Initialize for error reporting
for attempt in range(self.max_retries):
logging.info(f"Attempt {attempt + 1}/{self.max_retries} to generate/correct script for '{article_title}'...")
try:
raw_json_output = self._call_llm(messages)
script_data = json.loads(raw_json_output)
# Pydantic v2 handles `total_estimated_duration_seconds` calculation if it's a computed field
# or ensures it's present. For simplicity, if LLM misses it, we can calculate here
# before validation, or rely on Pydantic's strictness.
# Here, we'll rely on Pydantic to catch it, which will trigger a correction.
validated_script = self.schema.model_validate(script_data)
logging.info("Script successfully generated and validated.")
return validated_script
except json.JSONDecodeError as e:
error_message = f"JSON decoding failed: {e}"
logging.warning(error_message)
if attempt == self.max_retries - 1:
logging.error(f"Failed to generate valid JSON after {self.max_retries} attempts.")
raise
correction_user_prompt = self.correction_prompt_template.format(
error_message=error_message, invalid_json=raw_json_output
)
messages.append({"role": "assistant", "content": raw_json_output})
messages.append({"role": "user", "content": correction_user_prompt})
except ValidationError as e:
error_message = f"Pydantic validation failed:\n{e.json()}"
logging.warning(error_message)
if attempt == self.max_retries - 1:
logging.error(f"Failed to generate valid script after {self.max_retries} attempts.")
raise
correction_user_prompt = self.correction_prompt_template.format(
error_message=error_message, invalid_json=raw_json_output
)
messages.append({"role": "assistant", "content": raw_json_output})
messages.append({"role": "user", "content": correction_user_prompt})
except Exception as e:
logging.error(f"An unexpected error occurred during script generation: {e}")
raise
raise Exception("Failed to generate a valid video script after multiple attempts.")
# --- Step 4: Integrating with OpenMontage (Mocked for Output Validation) ---
class OpenMontageClient:
"""
A mock client for the OpenMontage video generation API.
In a real scenario, this would make actual API calls.
"""
def generate_video(self, script: VideoScript) -> dict:
"""
Simulates sending a validated VideoScript to the OpenMontage API.
"""
logging.info(f"Sending validated script to OpenMontage for video generation: '{script.title}'")
logging.info("OpenMontage received script. Simulating successful video generation request.")
return {"status": "success", "video_id": "mock_video_123", "script_received": script.model_dump()}
def main():
cloudflare_blog_rss_url = "https://blog.cloudflare.com/rss/"
# 1. Fetch RSS feed
try:
feed = fetch_rss_feed(cloudflare_blog_rss_url)
if not feed.entries:
logging.error("No entries found in the RSS feed.")
return
# Get the latest entry
latest_entry = feed.entries[0]
article_title = latest_entry.title
# Use summary or description; content can be HTML, so we'll simplify for the LLM
article_content = latest_entry.get('summary', latest_entry.get('description', 'No content available.')).strip()
# Basic HTML stripping for LLM input
article_content = requests.utils.unquote(article_content) # Unquote HTML entities
article_content = article_content.replace(" ", " ").replace("", "").replace("
", "").strip()
logging.info(f"\n--- Processing Latest Blog Post: '{article_title}' ---")
logging.info(f"Content snippet: {article_content[:200]}...")
except (requests.exceptions.RequestException, Exception) as e:
logging.error(f"Failed to fetch or parse RSS feed: {e}")
return
# 2. Initialize LLM agent and generate script
llm_client = MockLLM(latency=0.1) # Shorter latency for faster demo
script_generator = AgenticVideoScriptGenerator(llm_client=llm_client, schema=VideoScript, max_retries=2)
try:
generated_script = script_generator.generate_script_from_article(
article_title=article_title,
article_content=article_content
)
logging.info("\n--- Final Validated Video Script ---")
logging.info(generated_script.model_dump_json(indent=2))
# 3. Integrate with OpenMontage
openmontage_client = OpenMontageClient()
openmontage_response = openmontage_client.generate_video(generated_script)
logging.info(f"\n--- OpenMontage API Response ---")
logging.info(json.dumps(openmontage_response, indent=2))
except Exception as e:
logging.error(f"Failed to generate or process video script: {e}")
if __name__ == "__main__":
main()
Expected Output
When you run this script, you'll see a series of log messages detailing the RSS fetch, the mock LLM's initial (invalid) script generation, the self-correction process, and finally, the successful generation and validation of the VideoScript. The mock LLM is designed to fail on the first attempt and succeed on the second, demonstrating the self-correction loop.
202X-XX-XX XX:XX:XX,XXX - INFO - Attempting to fetch RSS feed from: https://blog.cloudflare.com/rss/
202X-XX-XX XX:XX:XX,XXX - INFO - --- Processing Latest Blog Post: 'Filter, reshape, and redact logs before export with Transformers: Logpush now supports Transformers, allowing teams to filter, redact, and reshape log streams using SQL before export. Powered by Apache DataFusion and' ---
202X-XX-XX XX:XX:XX,XXX - INFO - Content snippet: Logpush now supports Transformers, allowing teams to filter, redact, and reshape log streams using SQL before export. Powered by Apache DataFusion and...
202X-XX-XX XX:XX:XX,XXX - INFO - Attempt 1/2 to generate/correct script for 'Filter, reshape, and redact logs before export with Transformers: Logpush now supports Transformers, allowing teams to filter, redact, and reshape log streams using SQL before export. Powered by Apache DataFusion and'...
202X-XX-XX XX:XX:XX,XXX - INFO - Mock LLM: Simulating initial script generation (will intentionally be invalid)...
202X-XX-XX XX:XX:XX,XXX - WARNING - Pydantic validation failed:
[
{
"type": "too_short",
"loc": [
"scenes"
],
"msg": "List should have at least 3 items after validation, not 1",
"input": [
{
"scene_number": 1,
"voiceover_text": "Cloudflare Logpush just got a major upgrade, letting you filter and transform logs with SQL.",
"visual_cues": [
{
"description": "Cloudflare logo and SQL code snippets.",
"type": "image",
"duration_seconds": 5
}
],
"estimated_duration_seconds": 5
}
],
"url": "https://errors.pydantic.dev/2.5/v/too_short"
},
{
"type": "missing",
"loc": [
"total_estimated_duration_seconds"
],
"msg": "Field required",
"input": {
"title": "Cloudflare Logpush Gets SQL Power",
"target_audience": "Developers, Ops Teams",
"key_message": "Cloudflare Logpush now offers advanced log transformation with SQL.",
"scenes": [
{
"scene_number": 1,
"voiceover_text": "Cloudflare Logpush just got a major upgrade, letting you filter and transform logs with SQL.",
"visual_cues": [
{
"description": "Cloudflare logo and SQL code snippets.",
"type": "image",
"duration_seconds": 5
}
],
"estimated_duration_seconds": 5
}
]
},
"url": "https://errors.pydantic.dev/2.5/v/missing"
}
]
202X-XX-XX XX:XX:XX,XXX - INFO - Attempt 2/2 to generate/correct script for 'Filter, reshape, and redact logs before export with Transformers: Logpush now supports Transformers, allowing teams to filter, redact, and reshape log streams using SQL before export. Powered by Apache DataFusion and'...
202X-XX-XX XX:XX:XX,XXX - INFO - Mock LLM: Simulating correction attempt (will be valid)...
202X-XX-XX XX:XX:XX,XXX - INFO - Script successfully generated and validated.
202X-XX-XX XX:XX:XX,XXX - INFO - --- Final Validated Video Script ---
202X-XX-XX XX:XX:XX,XXX - INFO - {
"title": "Cloudflare's New Logpush Transformers: SQL for Your Logs",
"target_audience": "DevOps Engineers, SREs, Data Analysts",
"key_message": "Cloudflare's Logpush now allows filtering, reshaping, and redacting logs with SQL via Transformers, powered by Apache DataFusion.",
"scenes": [
{
"scene_number": 1,
"voiceover_text": "Managing vast log data can be a nightmare. Cloudflare's Logpush service helps, but what if you need to refine that data before it even leaves?",
"visual_cues": [
{
"description": "Animated graphic of log data flowing into a funnel.",
"type": "animation",
"duration_seconds": 5
}
],
"estimated_duration_seconds": 5
},
{
"scene_number": 2,
"voiceover_text": "Introducing Logpush Transformers! Now, you can use SQL to filter, reshape, and redact your log streams directly within Cloudflare.",
"visual_cues": [
{
"description": "SQL query interface overlayed on Cloudflare dashboard.",
"type": "image",
"duration_seconds": 6
}
],
"estimated_duration_seconds": 6
},
{
"scene_number": 3,
"voiceover_text": "Powered by Apache DataFusion, this feature ensures you only export the most relevant, secure data, saving on storage and processing.",
"visual_cues": [
{
"description": "Diagram showing DataFusion processing logs.",
"type": "animation",
"duration_seconds": 7
},
{
"description": "Text overlay: 'Powered by Apache DataFusion'",
"type": "text_overlay",
"duration_seconds": 4
}
],
"estimated_duration_seconds": 7
}
],
"total_estimated_duration_seconds": 18
}
202X-XX-XX XX:XX:XX,XXX - INFO - Sending validated script to OpenMontage for video generation: 'Cloudflare's New Logpush Transformers: SQL for Your Logs'
202X-XX-XX XX:XX:XX,XXX - INFO - OpenMontage received script. Simulating successful video generation request.
202X-XX-XX XX:XX:XX,XXX - INFO - --- OpenMontage API Response ---
202X-XX-XX XX:XX:XX,XXX - INFO - {
"status": "success",
"video_id": "mock_video_123",
"script_received": {
"title": "Cloudflare's New Logpush Transformers: SQL for Your Logs",
"target_audience": "DevOps Engineers, SREs, Data Analysts",
"key_message": "Cloudflare's Logpush now allows filtering, reshaping, and redacting logs with SQL via Transformers, powered by Apache DataFusion.",
"scenes": [
{
"scene_number": 1,
"voiceover_text": "Managing vast log data can be a nightmare. Cloudflare's Logpush service helps, but what if you need to refine that data before it even leaves?",
"visual_cues": [
{
"description": "Animated graphic of log data flowing into a funnel.",
"type": "animation",
"duration_seconds": 5
}
],
"estimated_duration_seconds": 5
},
{
"scene_number": 2,
"voiceover_text": "Introducing Logpush Transformers! Now, you can use SQL to filter, reshape, and redact your log streams directly within Cloudflare.",
"visual_cues": [
{
"description": "SQL query interface overlayed on Cloudflare dashboard.",
"type": "image",
"duration_seconds": 6
}
],
"estimated_duration_seconds": 6
},
{
"scene_number": 3,
"voiceover_text": "Powered by Apache DataFusion, this feature ensures you only export the most relevant, secure data, saving on storage and processing.",
"visual_cues": [
{
"description": "Diagram showing DataFusion processing logs.",
"type": "animation",
"duration_seconds": 7
},
{
"description": "Text overlay: 'Powered by Apache DataFusion'",
"type": "text_overlay",
"duration_seconds": 4
}
],
"estimated_duration_seconds": 7
}
],
"total_estimated_duration_seconds": 18
}
}
Limitations and Tradeoffs
While this agentic pipeline offers significant advantages in automation and structured output, it's not a silver bullet. Understanding its limitations and inherent tradeoffs is crucial for production deployment:
- LLM Hallucinations & Factual Accuracy: While self-correction helps enforce schema, it doesn't guarantee factual accuracy. LLMs can still generate plausible but incorrect information. For critical applications, human review remains indispensable to verify content against the source material.
- Cost and Latency: Each LLM call incurs API costs and contributes to end-to-end latency. A multi-agent, self-correcting system, by its very nature, might involve multiple calls per content item. This can quickly become expensive and slow, especially with high-volume feeds or complex generation tasks.
- Creative Control vs. Automation: Full automation can lead to generic or uninspired content. While LLMs are creative, their output can lack the nuanced storytelling or unique voice a human editor provides. Balancing agentic creativity with human oversight (e.g., for prompt engineering or final review) is crucial to maintain content quality.
- Schema Evolution: Changes to the
VideoScriptPydantic schema require careful updates to the LLM's system prompt and potentially re-training or extensive testing of the agents. A minor schema change can break the entire pipeline if the LLM isn't properly re-instructed. - Asset Sourcing: This system focuses solely on script generation. The actual video asset sourcing (finding relevant stock footage, generating images, recording voiceovers) and editing remain complex challenges not fully addressed by the mocked OpenMontage integration. It provides the blueprint, but not the building materials.
Frequently Asked Questions
How can I make this more cost-effective for production?
Strategies include using cheaper, faster LLMs (like smaller open-source models fine-tuned for your specific task) for initial drafts, and only escalating to larger, more capable models for validation or complex correction rounds. Caching successful outputs for identical or highly similar inputs can also reduce redundant LLM calls. Additionally, optimizing prompts to reduce token count directly reduces cost.
What if the LLM output consistently fails validation?
If the LLM consistently struggles, it's a strong indicator that your prompt instructions need refinement. Provide more detailed examples (few-shot prompting) in the system prompt, explicitly highlight difficult validation rules, or simplify your Pydantic schema if possible. For extreme cases, consider a more robust parsing library that can tolerate minor deviations and attempt to "fix" them before Pydantic validation.
Can this system generate videos in different languages?
Yes, by instructing the LLM to generate voiceover text and scene descriptions in the target language. You would need to ensure your LLM supports the desired language and that your prompt explicitly requests the output in that language. However, visual cues (e.g., text overlays in images) might need separate localization efforts.
How do I handle very long blog posts or articles?
For lengthy content exceeding typical LLM context windows, employ chunking and recursive summarization techniques. Break the article into smaller sections, summarize each section, and then synthesize these summaries for the final script generation. Alternatively, use LLMs with larger context windows, though these often come with higher costs.
Is OpenMontage a real library?
OpenMontage serves as a conceptual placeholder for any external video generation API or tool. Its inclusion in this post emphasizes the importance of defining a clear interface contract and ensuring the structured input required for such downstream systems. In a real-world scenario, you would replace it with clients for actual video APIs like Google Cloud Video Intelligence, AWS Elemental, or custom solutions.
What I'd Change
While the current pipeline demonstrates robust self-correction, for a true production system, I would introduce asynchronous processing and integrate a dedicated observability stack. Right now, each LLM call is blocking, which severely limits throughput. Moving to asyncio for both RSS fetching and LLM interactions would allow us to process multiple articles concurrently, drastically improving latency for batch operations. Furthermore, I'd implement detailed logging and metrics for each agent step: tracking LLM call counts, token usage, validation success/failure rates, and correction loop depths. This telemetry is invaluable for identifying prompt decay, cost spikes, and areas for optimization. Finally, the mock LLM would be replaced by a real, rate-limited OpenAI client, potentially wrapped in a circuit breaker pattern to prevent cascading failures during API outages. The goal isn't just to produce a script, but to build a transparent, scalable, and maintainable content factory.