In production data pipelines, relying solely on prompt engineering for LLM agents often leads to inconsistent output formats, making downstream integration brittle and requiring constant manual oversight. This challenge is particularly acute when consuming dynamic, real-world feeds like engineering blogs, where content structure varies wildly. This post addresses the critical need for architecting agentic systems that not only interpret and act on information but also rigorously validate and self-correct their outputs, guaranteeing data integrity and significantly reducing operational overhead. If you're a data engineer, ML engineer, or Python developer building robust LLM applications, you'll learn how to leverage Claude's code generation capabilities with Pydantic AI's validation to create truly resilient, self-auditing workflows.
Key Takeaways
- LLM agents can generate executable Python code for structured data extraction, moving beyond simple JSON output.
- Pydantic AI is crucial for defining strict output schemas and validating LLM-generated data, catching inconsistencies early.
- Implementing a self-correction loop, where validation errors are fed back to the LLM, significantly improves output reliability.
- Executing LLM-generated code in production requires robust sandboxing and security measures to mitigate risks.
- Combining code generation with validation loops builds resilient agentic workflows essential for dynamic, real-world data feeds.
The Problem: The Brittleness of Unvalidated LLM Outputs
We've all been there: you build a clever LLM prompt, it works perfectly in testing, and then it hits production. Suddenly, your downstream systems are breaking because the LLM decided to add an extra field, change a data type, or just outright hallucinate. When dealing with dynamic sources like RSS feeds, where each entry might have subtle variations in how information is presented, this becomes a constant battle. Pure prompt engineering, even with JSON modes, often isn't enough to guarantee the strict, consistent output needed for reliable data pipelines. We need a mechanism for the agent to not just *produce* data, but to *understand* and *self-correct* against a predefined contract.
Data and Sources
For this demonstration, we'll be analyzing the GitHub Engineering RSS feed, a rich source of technical articles with varying structures and content. Our goal is to extract specific, structured information from each post.
- GitHub Engineering RSS Feed: https://github.blog/engineering/feed/
- Python `feedparser` Library: https://pypi.org/project/feedparser/
- Anthropic Claude API Documentation: https://docs.anthropic.com/claude/reference/messages_api
- Pydantic AI Documentation: https://docs.pydantic.dev/latest/usage/ai/
Data accessed on 2024-07-29.
Step 1 — Ingesting Dynamic Feed Data with Robustness
Before we can apply any agentic intelligence, we need to reliably pull the raw data. RSS feeds, while standardized, can still present issues like network errors, malformed XML, or unexpected content structures. Our ingestion layer needs to be robust.
The sub-problem here is fetching and parsing the RSS feed without our application crashing at the first sign of trouble. We'll use the `feedparser` library, which handles much of the complexity of RSS/Atom parsing, and wrap our network call in a `try-except` block to catch common `requests` exceptions. For this example, we'll focus on the first few entries to keep API calls manageable.
import feedparser
import requests
def fetch_feed_entries(url: str, limit: int = 5) -> list:
"""Fetches and parses an RSS feed, returning a limited number of entries."""
try:
# feedparser can directly handle URLs, but requests gives more control over headers/timeouts
# For simplicity, we'll let feedparser handle the fetch, but in production,
# you'd use requests with explicit timeouts and error handling.
feed = feedparser.parse(url)
if feed.bozo:
print(f"Warning: Malformed feed detected for {url}. Bozo exception: {feed.bozo_exception}")
return feed.entries[:limit]
except requests.exceptions.RequestException as e:
print(f"Network error fetching feed from {url}: {e}")
return []
except Exception as e:
print(f"An unexpected error occurred while parsing feed from {url}: {e}")
return []
This snippet sets up a function to fetch the RSS feed. While `feedparser.parse` can take a URL directly, in a real production system, you'd likely use `requests.get` with explicit timeouts, retries, and error handling, then pass the raw content to `feedparser.parse(xml_content)`. This gives you finer-grained control over network behavior.
Step 2 — Defining Production-Ready Output Schemas with Pydantic AI
The core of ensuring consistent LLM output lies in defining a strict schema. Pydantic is already a go-to for data validation in Python, and `pydantic-ai` extends this by making Pydantic models LLM-aware, allowing them to generate JSON schemas that LLMs can understand. This means we define our data contract once, and both our application and the LLM adhere to it.
The sub-problem here is to explicitly tell the LLM *exactly* what data format we expect. We'll define a `BlogPost` model that specifies the required fields, their types, and helpful descriptions for the LLM.
from pydantic import BaseModel, Field, ValidationError
from typing import Optional, List
class BlogPost(BaseModel):
"""
Represents a structured blog post entry with key extracted details.
This schema is used to guide the LLM's output and validate it.
"""
title: str = Field(description="The full title of the blog post.")
url: str = Field(description="The canonical URL to the blog post.")
summary: str = Field(description="A concise summary of the blog post's content, 2-3 sentences.")
main_topics: List[str] = Field(description="A list of 3-5 main technical topics covered in the post (e.g., 'Kubernetes', 'Python Performance').")
is_case_study: bool = Field(description="True if the post describes a specific problem and its solution, False otherwise.")
estimated_read_time_minutes: Optional[int] = Field(None, description="The estimated reading time in minutes, if mentioned or inferable.")
# We can get the JSON schema for Claude to understand
BLOG_POST_SCHEMA = BlogPost.model_json_schema()
Here, we define our `BlogPost` model. Notice the `Field` descriptions; these are crucial hints for the LLM to understand the intent behind each field. `BLOG_POST_SCHEMA` will be included in our prompt to Claude, acting as the blueprint for its code generation.
Step 3 — Orchestrating Claude's Code Generation for Structured Analysis
Instead of just asking Claude for JSON, we're going a step further: we're asking it to *generate Python code* that, when executed, produces a dictionary conforming to our `BlogPost` schema. This is where "Claude Code" comes into play, enabling a more dynamic and powerful agent. This approach demands that Claude not only understands the schema but also can translate that understanding into executable logic.
The sub-problem is instructing Claude to produce a runnable Python code block that extracts information from a raw RSS entry and formats it according to our Pydantic model. This generated code will then be executed by our agent.
import anthropic
import os
import json # For pretty printing schema
ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY")
if not ANTHROPIC_API_KEY:
raise ValueError("ANTHROPIC_API_KEY environment variable not set.")
client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY)
def generate_extraction_code(rss_entry: dict, schema: dict, previous_error: Optional[str] = None) -> str:
"""
Prompts Claude to generate Python code to extract structured data from an RSS entry
according to the provided schema. Includes self-correction feedback if a previous error occurred.
"""
schema_str = json.dumps(schema, indent=2)
system_prompt = (
"You are an expert Python data extraction agent. Your task is to generate a Python code block "
"that extracts specific information from a given RSS feed entry and returns it as a dictionary. "
"The dictionary MUST strictly conform to the provided Pydantic schema. "
"Your output must be ONLY the Python code block, enclosed in `` and ` ` tags. "
"The code should define and return a dictionary literal that matches the schema. "
"Do NOT include any imports or function definitions in the generated code itself, just the dictionary assignment and return."
)
user_message = (
f"Here is the RSS feed entry:\n\n\n{rss_entry}\n \n\n"
f"Here is the Pydantic schema you must adhere to:\n\n\n{schema_str}\n \n\n"
"Generate Python code to extract this information. Make sure to handle missing fields gracefully "
"and infer values if possible based on the schema descriptions. "
"For 'estimated_read_time_minutes', if not explicitly mentioned, try to infer it from the summary length (150 words/min is a rough estimate)."
)
if previous_error:
user_message += (
f"\n\nCRITICAL: Your previous attempt failed with the following validation error. "
f"Please review the error and correct your Python code to strictly conform to the schema:\n"
f"\n{previous_error}\n \n"
f"Ensure your generated dictionary is perfectly valid against the schema."
)
try:
response = client.messages.create(
model="claude-3-opus-20240229", # Or another suitable Claude model
max_tokens=1024,
system=system_prompt,
messages=[
{"role": "user", "content": user_message}
]
)
# Extract content from the response
if response.content:
generated_text = response.content[0].text
# Look for the Python code block
start_tag = ""
end_tag = " "
start_index = generated_text.find(start_tag)
end_index = generated_text.find(end_tag)
if start_index != -1 and end_index != -1:
return generated_text[start_index + len(start_tag):end_index].strip()
else:
print("Warning: Claude did not return a well-formatted Python code block.")
return ""
return ""
except anthropic.APIError as e:
print(f"Claude API error: {e}")
return ""
except Exception as e:
print(f"An unexpected error occurred during Claude API call: {e}")
return ""
Here, we build a detailed prompt for Claude, including the raw RSS entry and our Pydantic schema. The key instruction is to generate *only* the Python code block, enclosed in specific tags. This clear instruction helps Claude produce structured output that we can then parse. The `previous_error` parameter is our hook for self-correction, which we'll use in the next step.
For more on building autonomous agents with Claude's code plugins, you might find our previous post, Building an Autonomous Python Data Agent with Claude Code Plugins for Nepal’s Financial Feeds, insightful.
Step 4 — Implementing Self-Correction via Pydantic Validation Loops
This is where the "self-correcting" aspect truly comes alive. We execute the code Claude generates, and then we immediately validate its output using our Pydantic model. If validation fails, we don't give up; instead, we send the exact `ValidationError` message back to Claude and ask it to try again, leveraging its reasoning capabilities to fix its own mistakes.
The sub-problem is to safely execute potentially untrusted code and provide actionable feedback to the LLM for iterative refinement. This loop continues until a valid output is produced or a maximum number of retries is reached.
import time
MAX_RETRIES = 3
def execute_and_validate_code(code: str, schema: BaseModel) -> Optional[dict]:
"""
Executes the generated Python code in a restricted environment and validates its output
against the Pydantic schema.
"""
# CRITICAL: Executing arbitrary code from an LLM is a security risk.
# In a production environment, this MUST be done in a highly isolated and sandboxed environment.
# For this demonstration, we use a basic exec, but be aware of the implications.
# A safer approach might involve a dedicated microservice for code execution or
# using a strict allow-list for operations.
local_vars = {}
try:
exec(code, {}, local_vars)
extracted_data = local_vars.get('extracted_data') # Assuming the code sets a variable named 'extracted_data'
if not extracted_data:
raise ValueError("Generated code did not return an 'extracted_data' dictionary.")
validated_data = schema.model_validate(extracted_data)
return validated_data.model_dump()
except SyntaxError as e:
return {"error": f"Syntax Error in generated code: {e}"}
except NameError as e:
return {"error": f"Name Error in generated code: {e}"}
except ValidationError as e:
return {"error": f"Pydantic Validation Error: {e.json()}"}
except Exception as e:
return {"error": f"Runtime Error in generated code: {e}"}
def process_entry_with_self_correction(rss_entry: dict, schema: dict) -> Optional[dict]:
"""
Orchestrates the code generation, execution, and self-correction loop.
"""
previous_error = None
for attempt in range(MAX_RETRIES):
print(f" Attempt {attempt + 1}/{MAX_RETRIES} for '{rss_entry.get('title', 'Untitled')}'...")
generated_code = generate_extraction_code(rss_entry, schema, previous_error)
if not generated_code:
print(" Failed to generate code.")
previous_error = "LLM failed to generate any code or a valid code block."
time.sleep(1) # Small delay before retrying LLM call
continue
execution_result = execute_and_validate_code(generated_code, BlogPost)
if "error" in execution_result:
previous_error = execution_result["error"]
print(f" Validation/Execution failed: {previous_error}")
time.sleep(1) # Small delay before retrying LLM call
else:
print(f" Successfully extracted data on attempt {attempt + 1}.")
return execution_result
print(f" Failed to extract valid data after {MAX_RETRIES} attempts.")
return None
The `execute_and_validate_code` function is critical. It attempts to run the Python code generated by Claude. If the code itself has syntax errors, runtime errors, or if its output fails Pydantic validation, we capture that specific error. The `process_entry_with_self_correction` function then orchestrates the retry loop, passing the `previous_error` back to `generate_extraction_code`. This feedback loop allows Claude to learn from its mistakes in real-time and refine its code generation.
Step 5 — Navigating Production Realities: Execution Safety and Performance
Generating and executing code from an LLM, while powerful, introduces significant production challenges. Security is paramount: `exec()` is inherently dangerous if the generated code is untrusted. Performance and cost are also major considerations, especially with retry loops.
The sub-problem here is to acknowledge and mitigate the practical risks of this agentic approach. For security, in a real-world system, you would never use a bare `exec()` like this. Instead, consider:
- Sandboxed Execution: Run generated code in an isolated, containerized environment (e.g., Docker, gVisor, or even a separate microservice) with strict resource limits and no network/filesystem access.
- Code Review/Approval: For highly sensitive tasks, generated code might require human review before execution.
- Strict Allow-list: Restrict the Python modules and functions the generated code can use.
For performance and cost:
- Rate Limiting & Exponential Backoff: Implement robust retry mechanisms for API calls, with increasing delays between attempts, to handle transient network issues or rate limits.
- Context Window Management: Keep prompts concise. Sending full RSS entries and detailed schemas repeatedly can consume tokens quickly.
- Caching: Cache API responses or validated outputs to avoid re-processing identical data.
Our current script includes basic `time.sleep` for backoff, but a production system would use a more sophisticated retry library (e.g., `tenacity`). The `MAX_RETRIES` constant directly impacts cost and latency; careful tuning is required. This pattern, while powerful, is not cheap.
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
import feedparser
import requests
import anthropic
import os
import json
import time
from pydantic import BaseModel, Field, ValidationError
from typing import Optional, List, Dict, Any
# --- Configuration ---
ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY")
if not ANTHROPIC_API_KEY:
raise ValueError("ANTHROPIC_API_KEY environment variable not set. Please set it to your Claude API key.")
RSS_FEED_URL = "https://github.blog/engineering/feed/"
ENTRY_LIMIT = 3 # Limit for demonstration purposes
MAX_RETRIES = 3 # Max attempts for Claude to self-correct
CLAUDE_MODEL = "claude-3-opus-20240229" # Or another suitable Claude model
client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY)
# --- Pydantic Schema Definition ---
class BlogPost(BaseModel):
"""
Represents a structured blog post entry with key extracted details.
This schema is used to guide the LLM's output and validate it.
"""
title: str = Field(description="The full title of the blog post.")
url: str = Field(description="The canonical URL to the blog post.")
summary: str = Field(description="A concise summary of the blog post's content, 2-3 sentences.")
main_topics: List[str] = Field(description="A list of 3-5 main technical topics covered in the post (e.g., 'Kubernetes', 'Python Performance').")
is_case_study: bool = Field(description="True if the post describes a specific problem and its solution, False otherwise.")
estimated_read_time_minutes: Optional[int] = Field(None, description="The estimated reading time in minutes, if mentioned or inferable.")
# We can get the JSON schema for Claude to understand
BLOG_POST_SCHEMA = BlogPost.model_json_schema()
# --- Step 1: Ingesting Dynamic Feed Data with Robustness ---
def fetch_feed_entries(url: str, limit: int = 5) -> list:
"""Fetches and parses an RSS feed, returning a limited number of entries."""
print(f"Fetching feed from {url}...")
try:
feed = feedparser.parse(url)
if feed.bozo:
print(f"Warning: Malformed feed detected for {url}. Bozo exception: {feed.bozo_exception}")
print(f"Successfully fetched {len(feed.entries)} entries. Limiting to {limit}.")
return feed.entries[:limit]
except requests.exceptions.RequestException as e:
print(f"ERROR: Network error fetching feed from {url}: {e}")
return []
except Exception as e:
print(f"ERROR: An unexpected error occurred while parsing feed from {url}: {e}")
return []
# --- Step 3: Orchestrating Claude's Code Generation for Structured Analysis ---
def generate_extraction_code(rss_entry: dict, schema: dict, previous_error: Optional[str] = None) -> str:
"""
Prompts Claude to generate Python code to extract structured data from an RSS entry
according to the provided schema. Includes self-correction feedback if a previous error occurred.
"""
schema_str = json.dumps(schema, indent=2)
system_prompt = (
"You are an expert Python data extraction agent. Your task is to generate a Python code block "
"that extracts specific information from a given RSS feed entry and returns it as a dictionary. "
"The dictionary MUST strictly conform to the provided Pydantic schema. "
"Your output must be ONLY the Python code block, enclosed in `` and ` ` tags. "
"The code should define a variable named `extracted_data` which is a dictionary literal that matches the schema. "
"Do NOT include any imports or function definitions in the generated code itself, just the dictionary assignment."
)
user_message = (
f"Here is the RSS feed entry:\n\n\n{rss_entry}\n \n\n"
f"Here is the Pydantic schema you must adhere to:\n\n\n{schema_str}\n \n\n"
"Generate Python code to extract this information. Make sure to handle missing fields gracefully "
"and infer values if possible based on the schema descriptions. "
"For 'estimated_read_time_minutes', if not explicitly mentioned, try to infer it from the summary length (roughly 150 words per minute)."
"Your code should assign the resulting dictionary to a variable named `extracted_data`."
)
if previous_error:
user_message += (
f"\n\nCRITICAL: Your previous attempt failed with the following validation error. "
f"Please review the error and correct your Python code to strictly conform to the schema:\n"
f"\n{previous_error}\n \n"
f"Ensure your generated dictionary for `extracted_data` is perfectly valid against the schema."
)
try:
response = client.messages.create(
model=CLAUDE_MODEL,
max_tokens=2048, # Increased max_tokens for potentially longer code/responses
system=system_prompt,
messages=[
{"role": "user", "content": user_message}
]
)
if response.content:
generated_text = response.content[0].text
start_tag = ""
end_tag = " "
start_index = generated_text.find(start_tag)
end_index = generated_text.find(end_tag)
if start_index != -1 and end_index != -1:
return generated_text[start_index + len(start_tag):end_index].strip()
else:
print(" Warning: Claude did not return a well-formatted Python code block. Full response:")
print(f" {generated_text[:500]}...") # Print a snippet of the response
return ""
return ""
except anthropic.APIError as e:
print(f" ERROR: Claude API error: {e}")
return ""
except Exception as e:
print(f" ERROR: An unexpected error occurred during Claude API call: {e}")
return ""
# --- Step 4: Implementing Self-Correction via Pydantic Validation Loops ---
def execute_and_validate_code(code: str, schema_model: BaseModel) -> Dict[str, Any]:
"""
Executes the generated Python code in a restricted environment and validates its output
against the Pydantic schema. Returns a dictionary with either the validated data or an error.
"""
# CRITICAL: Executing arbitrary code from an LLM is a security risk.
# In a production environment, this MUST be done in a highly isolated and sandboxed environment.
# For this demonstration, we use a basic exec, but be aware of the implications.
# A safer approach might involve a dedicated microservice for code execution or
# using a strict allow-list for operations.
local_vars: Dict[str, Any] = {}
# Restrict builtins available to exec for basic safety, but this is not a sandbox.
# For example, disallow __import__ to prevent arbitrary module imports.
restricted_globals = {
"__builtins__": {
"print": print, "len": len, "str": str, "int": int, "float": float,
"bool": bool, "list": list, "dict": dict, "None": None, "True": True, "False": False,
"min": min, "max": max, "sum": sum, "round": round, "abs": abs,
# Add other necessary builtins, but be very selective
}
}
try:
exec(code, restricted_globals, local_vars)
extracted_data = local_vars.get('extracted_data')
if extracted_data is None:
raise ValueError("Generated code did not assign a dictionary to 'extracted_data'.")
if not isinstance(extracted_data, dict):
raise TypeError(f"Expected 'extracted_data' to be a dictionary, got {type(extracted_data).__name__}.")
validated_data = schema_model.model_validate(extracted_data)
return {"data": validated_data.model_dump()}
except SyntaxError as e:
return {"error": f"Syntax Error in generated code: {e}"}
except NameError as e:
return {"error": f"Name Error in generated code: {e}"}
except TypeError as e:
return {"error": f"Type Error in generated code: {e}"}
except ValueError as e:
return {"error": f"Value Error in generated code: {e}"}
except ValidationError as e:
return {"error": f"Pydantic Validation Error: {e.json()}"}
except Exception as e:
return {"error": f"Runtime Error in generated code: {e}. Code was:\n{code}"}
def process_entry_with_self_correction(rss_entry: dict, schema: dict) -> Optional[dict]:
"""
Orchestrates the code generation, execution, and self-correction loop.
"""
previous_error = None
for attempt in range(MAX_RETRIES):
print(f"\n Processing entry: '{rss_entry.get('title', 'Untitled')}' (Attempt {attempt + 1}/{MAX_RETRIES})")
# Introduce a small delay to respect API rate limits and avoid hammering
if attempt > 0:
time.sleep(2 ** attempt) # Exponential backoff
generated_code = generate_extraction_code(rss_entry, schema, previous_error)
if not generated_code:
print(" Failed to generate code from Claude. Retrying...")
previous_error = "LLM failed to generate any code or a valid code block."
continue
execution_result = execute_and_validate_code(generated_code, BlogPost)
if "error" in execution_result:
previous_error = execution_result["error"]
print(f" Validation/Execution failed: {previous_error}")
else:
print(f" Successfully extracted data on attempt {attempt + 1}.")
return execution_result["data"]
print(f" Failed to extract valid data after {MAX_RETRIES} attempts for '{rss_entry.get('title', 'Untitled')}'.")
return None
# --- Main execution ---
def main():
print("Starting agentic workflow for GitHub Engineering Feed...")
entries = fetch_feed_entries(RSS_FEED_URL, ENTRY_LIMIT)
if not entries:
print("No entries to process. Exiting.")
return
processed_posts = []
for i, entry in enumerate(entries):
print(f"\n--- Processing Entry {i+1}/{len(entries)} ---")
# Ensure 'summary' exists for the prompt, even if empty
if 'summary' not in entry:
entry['summary'] = entry.get('description', '') # Fallback to description if summary is missing
processed_data = process_entry_with_self_correction(entry, BLOG_POST_SCHEMA)
if processed_data:
processed_posts.append(processed_data)
print(json.dumps(processed_data, indent=2))
else:
print(f"Skipping entry due to repeated extraction failures: {entry.get('title', 'Untitled')}")
print("\n--- All Processed Posts ---")
if processed_posts:
for post in processed_posts:
print(json.dumps(post, indent=2))
else:
print("No posts were successfully processed.")
if __name__ == "__main__":
main()
Expected Output
When you run the script, you'll see output indicating the fetching of the RSS feed, then a detailed log for each entry. For each entry, it will show attempts to generate and execute code, and if validation fails, it will report the error and retry. Upon successful extraction, you'll see the validated JSON output for that post. Here's an example of what a successful extraction might look like (actual content will vary based on the latest GitHub Engineering posts):
Starting agentic workflow for GitHub Engineering Feed...
Fetching feed from https://github.blog/engineering/feed/...
Successfully fetched X entries. Limiting to 3.
--- Processing Entry 1/3 ---
Processing entry: 'Staying Evergreen: How we automated secure access and repo hygiene in a shared org' (Attempt 1/3)
Successfully extracted data on attempt 1.
{
"title": "Staying Evergreen: How we automated secure access and repo hygiene in a shared org",
"url": "https://github.blog/engineering/...",
"summary": "This post details how GitHub automated secure access and repository hygiene...",
"main_topics": [
"GitHub Features",
"Access Management",
"Repository Hygiene",
"Automation"
],
"is_case_study": true,
"estimated_read_time_minutes": 5
}
--- Processing Entry 2/3 ---
Processing entry: 'From 2,500 requests to 24: Rebuilding star history without the stargazers' (Attempt 1/3)
Successfully extracted data on attempt 1.
{
"title": "From 2,500 requests to 24: Rebuilding star history without the stargazers",
"url": "https://github.blog/engineering/...",
"summary": "GitHub rebuilt its repo star history API to provide trends more efficiently...",
"main_topics": [
"API Design",
"Performance Optimization",
"Database Engineering",
"Data Privacy"
],
"is_case_study": true,
"estimated_read_time_minutes": 7
}
--- Processing Entry 3/3 ---
Processing entry: 'Building Git infrastructure for agent-scale development' (Attempt 1/3)
Successfully extracted data on attempt 1.
{
"title": "Building Git infrastructure for agent-scale development",
"url": "https://github.blog/engineering/...",
"summary": "GitHub is rebuilding its Git infrastructure to support agent-scale software development...",
"main_topics": [
"Git",
"Infrastructure",
"Agent Development",
"Scalability"
],
"is_case_study": false,
"estimated_read_time_minutes": 6
}
--- All Processed Posts ---
[
{
"title": "Staying Evergreen: How we automated secure access and repo hygiene in a shared org",
"url": "https://github.blog/engineering/...",
"summary": "This post details how GitHub automated secure access and repository hygiene...",
"main_topics": [
"GitHub Features",
"Access Management",
"Repository Hygiene",
"Automation"
],
"is_case_study": true,
"estimated_read_time_minutes": 5
},
{
"title": "From 2,500 requests to 24: Rebuilding star history without the stargazers",
"url": "https://github.blog/engineering/...",
"summary": "GitHub rebuilt its repo star history API to provide trends more efficiently...",
"main_topics": [
"API Design",
"Performance Optimization",
"Database Engineering",
"Data Privacy"
],
"is_case_study": true,
"estimated_read_time_minutes": 7
},
{
"title": "Building Git infrastructure for agent-scale development",
"url": "https://github.blog/engineering/...",
"summary": "GitHub is rebuilding its Git infrastructure to support agent-scale software development...",
"main_topics": [
"Git",
"Infrastructure",
"Agent Development",
"Scalability"
],
"is_case_study": false,
"estimated_read_time_minutes": 6
}
]
Limitations and Tradeoffs
While powerful, this self-correcting agentic workflow is not a silver bullet and comes with significant tradeoffs:
- Security Risks of Code Execution: The biggest limitation is the inherent danger of executing LLM-generated code. Our current `exec()` approach is *not* production-ready for untrusted inputs. A robust solution requires dedicated sandboxing, which adds complexity and overhead.
- Latency and Cost: Each retry involves another LLM API call, which directly translates to increased latency and cost. For real-time applications or high-volume data, this approach might be too slow or expensive.
- LLM Hallucination/Stuck Loops: While self-correction helps, there's always a risk the LLM gets "stuck" in a loop, unable to produce valid code even after multiple attempts, or that it hallucinates an output that *looks* valid but is factually incorrect.
- Complexity: Introducing code generation, execution, and a feedback loop significantly increases the complexity of the system compared to simple prompt-based extraction. Debugging issues that span LLM reasoning, generated code, and Pydantic validation can be challenging.
- Dependency on LLM Quality: The effectiveness of self-correction heavily relies on the LLM's ability to understand and act on error messages. Smaller or less capable models may struggle more with this.
Frequently Asked Questions
Why use Claude to generate Python code instead of just asking for JSON output?
Asking Claude to generate Python code for structured data offers a higher degree of control and explicit instruction, especially when dealing with complex extraction logic or inferring values. It pushes the LLM to 'think' programmatically, which can sometimes lead to more robust and accurate structured output than simply requesting JSON, particularly when combined with self-correction. It also aligns with more advanced agentic patterns where the LLM is generating 'actions' (code) rather than just 'data'.
How can I make the `exec()` call safe in production?
The `exec()` call in the example is for demonstration only. In production, you must use a secure sandboxing solution. This could involve running the generated code in a lightweight container (like Docker), a virtual machine, a dedicated microservice with strict network and filesystem isolation, or specialized Python sandboxing libraries (though these are often difficult to make truly secure). The key is to prevent the generated code from accessing or modifying anything outside its intended scope.
What if the LLM never generates valid code?
We implement a `MAX_RETRIES` limit to prevent infinite loops. If the LLM fails to produce valid output after these attempts, the system should log the failure, potentially flag the problematic input for human review, and move on to the next item. It's a pragmatic recognition that LLMs are not infallible.
Is this approach more expensive than simple API calls?
Yes, potentially significantly more expensive. Each retry involves a new LLM API call, which incurs additional token usage and latency. The cost-effectiveness depends on the complexity of the extraction, the quality of the LLM, and the acceptable error rate. For high-value data where correctness is paramount, the increased cost might be justified by reduced manual intervention.
Can this be applied to other LLMs or structured output formats?
Absolutely. The core pattern of defining a schema (e.g., with Pydantic), prompting an LLM to adhere to it, and implementing a validation and self-correction loop is highly transferable. While the syntax for prompting and API calls would change for other LLMs (like OpenAI's GPT models), the underlying principles remain the same. You could also adapt it for other structured formats like XML or YAML, though JSON is typically the easiest for LLMs.
What I'd Change
While this workflow demonstrates a powerful self-correcting agent, my primary concern for production deployment would be hardening the code execution environment. The current `exec()` is a critical vulnerability. I would immediately refactor the `execute_and_validate_code` function to communicate with a dedicated, isolated microservice responsible for safely running untrusted Python code. This microservice would likely be containerized (e.g., a lightweight Docker container per execution request) with strict resource limits, no network access, and a read-only filesystem, preventing any malicious code from impacting the host system. This separation of concerns, even with the added network overhead, is a non-negotiable for building truly resilient and secure agentic systems in the real world. Furthermore, I'd integrate a more robust observability stack, logging every LLM input, output, and validation error, to quickly identify patterns in failures and continuously refine prompts or schemas. The ability to generate code is a superpower, but like all superpowers, it demands immense responsibility.