When I first tried pulling structured insights from Cloudflare’s blog feed, the static prompt I wrote stopped working as soon as the author changed the article layout. The result was a flood of malformed JSON and a growing maintenance burden. If you’re building a daily digest or a knowledge graph from any evolving text source, you need an extraction agent that can spot its own mistakes, learn from them, and adapt on the fly. In this post I’ll walk you through a production‑ready, self‑healing pipeline that does exactly that, using only open‑source tools and a single LLM call per iteration.
Key Takeaways
- Use a two‑stage LLM workflow: an extractor for raw fields and a critic that judges factual fidelity.
- Iterative prompt refinement driven by critic feedback turns a one‑shot extraction into a robust loop.
- Tenacity‑based retries, idempotent feed processing, and structured logging give the system production readiness.
The Problem
Static parsers and single‑shot LLM calls break as soon as a blog’s formatting or vocabulary shifts. You end up with inconsistent schemas, hallucinated values, and a maintenance nightmare. The Cloudflare Blog RSS feed, with its ever‑changing post titles, links, and summaries, is a perfect testbed for this problem. I needed a system that could ingest the feed, extract a rich set of fields, validate them, and self‑correct whenever the extraction diverges from reality.
Data and Sources
Real data comes from the Cloudflare Blog RSS feed: https://blog.cloudflare.com/rss/. The feed is parsed with feedparser. Schema validation uses pydantic. LLM calls are made via the OpenAI Python client. Data accessed on 2026-10-22.
Loading the Data
We fetch the RSS feed, parse the entries, and keep only the most recent five posts for quick iteration. The `feedparser.parse` function returns a list of entries with `title`, `link`, and `summary` fields.
import feedparser
def load_feed(url: str, limit: int = 5):
"""Retrieve and parse the RSS feed, returning the latest entries."""
feed = feedparser.parse(url)
return feed.entries[:limit]
The Core Logic
The core logic is split into three layers: extraction, critique, and correction. Each layer is a small, testable function that can be swapped out or extended.
1️⃣ Defining the Target Schema
We use pydantic to declare the exact shape of the data we want. The schema guides the extractor and the critic.
from pydantic import BaseModel, Field, ValidationError, validator
from typing import Optional, List
class CloudflarePostInsight(BaseModel):
title: str = Field(..., description="Post title")
link: str = Field(..., description="URL to the full article")
keywords: List[str] = Field(..., description="Top keywords extracted from the summary")
key_takeaway_summary: str = Field(..., description="A concise takeaway")
sentiment: str = Field(..., description="Positive, Neutral, or Negative")
product_mentioned: Optional[str] = Field(None, description="If a Cloudflare product is mentioned")
@validator('sentiment')
def sentiment_must_be_valid(cls, v):
if v not in {"Positive", "Neutral", "Negative"}:
raise ValueError("sentiment must be Positive, Neutral, or Negative")
return v
2️⃣ Initial Extraction
We ask the LLM to output a JSON snippet that matches the schema. Immediately we attempt to parse that JSON into a pydantic model. A `ValidationError` signals a structural mismatch.
import openai
import json
def extract_fields(text: str, schema: BaseModel) -> CloudflarePostInsight:
prompt = f"""
Extract the following fields as a JSON object, following the schema:
{schema.schema_json(indent=2)}
Text:
{text}
"""
response = openai.ChatCompletion.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0,
max_tokens=512,
response_format={"type": "json_object"},
)
try:
data = json.loads(response.choices[0].message.content)
return CloudflarePostInsight(**data)
except ValidationError as ve:
raise ValueError(f"Extraction failed validation: {ve}")
except json.JSONDecodeError as je:
raise ValueError(f"Extraction returned invalid JSON: {je}")
3️⃣ Self‑Reflection Critic
The critic receives the original text, the attempted extraction, and the schema. It returns a `Feedback` model that flags issues and suggests prompt tweaks.
class Feedback(BaseModel):
success: bool = Field(..., description="Was the extraction accurate?")
reasons: List[str] = Field(..., description="Why or why not")
prompt_tweak: Optional[str] = Field(None, description="Suggested prompt addition")
def critic(text: str, extraction: CloudflarePostInsight) -> Feedback:
prompt = f"""
You are a critic. Evaluate the following extraction against the original text.
Original text:
{text}
Extraction:
{extraction.json(indent=2)}
Check for factual accuracy, completeness, and whether the key takeaway truly reflects the text.
Respond with a JSON object matching this schema:
{Feedback.schema_json(indent=2)}
"""
response = openai.ChatCompletion.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0,
max_tokens=256,
response_format={"type": "json_object"},
)
return Feedback(**json.loads(response.choices[0].message.content))
4️⃣ Self‑Correction Loop
If the critic flags an issue, we modify the extraction prompt by appending the suggested tweak and retry. We limit retries to avoid infinite loops.
def self_healing_extraction(text: str, schema: BaseModel, max_attempts: int = 3) -> CloudflarePostInsight:
attempt = 0
tweak = ""
while attempt < max_attempts:
try:
extraction = extract_fields(text + tweak, schema)
feedback = critic(text, extraction)
if feedback.success:
return extraction
if feedback.prompt_tweak:
tweak += f"\n\nAdditional instruction: {feedback.prompt_tweak}"
except Exception as e:
print(f"Attempt {attempt+1} failed: {e}")
attempt += 1
raise RuntimeError(f"Extraction failed after {max_attempts} attempts")
5️⃣ Production Hardening
We wrap network and API calls with tenacity for exponential backoff, keep a set of processed entry IDs for idempotency, and log each decision path.
from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type
import logging, hashlib
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
@retry(wait=wait_exponential(multiplier=1, min=2, max=10), stop=stop_after_attempt(3),
retry=retry_if_exception_type((openai.error.OpenAIError, requests.exceptions.RequestException)))
def safe_openai_call(*args, **kwargs):
return openai.ChatCompletion.create(*args, **kwargs)
@retry(wait=wait_exponential(multiplier=1, min=2, max=10), stop=stop_after_attempt(3),
retry=retry_if_exception_type(requests.exceptions.RequestException))
def safe_fetch(url):
return feedparser.parse(url)
def hash_entry(entry):
return hashlib.sha256((entry.title + entry.link).encode()).hexdigest()
processed_hashes = set()
Putting It Together
The main routine fetches the feed, skips already processed entries, runs the self‑healing extraction, and stores results in a JSON file. It also generates a simple bar plot of sentiment counts for visual sanity checks.
import json
import matplotlib.pyplot as plt
import os
def main():
feed_url = "https://blog.cloudflare.com/rss/"
entries = load_feed(feed_url)
insights = []
for entry in entries:
entry_hash = hash_entry(entry)
if entry_hash in processed_hashes:
logging.info(f"Skipping already processed entry: {entry.title}")
continue
try:
insight = self_healing_extraction(entry.summary, CloudflarePostInsight)
insights.append(insight.dict())
processed_hashes.add(entry_hash)
logging.info(f"Successfully extracted: {entry.title}")
except Exception as e:
logging.error(f"Failed to process {entry.title}: {e}")
# Persist results
with open("insights.json", "w") as f:
json.dump(insights, f, indent=2)
# Plot sentiment distribution
sentiments = [i["sentiment"] for i in insights]
counts = {s: sentiments.count(s) for s in set(sentiments)}
plt.bar(counts.keys(), counts.values())
plt.title("Sentiment Distribution")
plt.xlabel("Sentiment")
plt.ylabel("Count")
plt.savefig("sentiment_distribution.png")
logging.info("Saved sentiment plot to sentiment_distribution.png")
if __name__ == "__main__":
main()
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
# self_healing_extractor.py
import os
import json
import hashlib
import logging
import feedparser
import openai
import matplotlib.pyplot as plt
from pydantic import BaseModel, Field, ValidationError, validator
from typing import Optional, List
from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type
import requests
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
# ---------- Schema ----------
class CloudflarePostInsight(BaseModel):
title: str = Field(..., description="Post title")
link: str = Field(..., description="URL to the full article")
keywords: List[str] = Field(..., description="Top keywords extracted from the summary")
key_takeaway_summary: str = Field(..., description="A concise takeaway")
sentiment: str = Field(..., description="Positive, Neutral, or Negative")
product_mentioned: Optional[str] = Field(None, description="If a Cloudflare product is mentioned")
@validator('sentiment')
def sentiment_must_be_valid(cls, v):
if v not in {"Positive", "Neutral", "Negative"}:
raise ValueError("sentiment must be Positive, Neutral, or Negative")
return v
class Feedback(BaseModel):
success: bool = Field(..., description="Was the extraction accurate?")
reasons: List[str] = Field(..., description="Why or why not")
prompt_tweak: Optional[str] = Field(None, description="Suggested prompt addition")
# ---------- Utilities ----------
def load_feed(url: str, limit: int = 5):
feed = feedparser.parse(url)
return feed.entries[:limit]
def hash_entry(entry):
return hashlib.sha256((entry.title + entry.link).encode()).hexdigest()
processed_hashes = set()
# ---------- Tenacity wrappers ----------
@retry(wait=wait_exponential(multiplier=1, min=2, max=10), stop=stop_after_attempt(3),
retry=retry_if_exception_type((openai.error.OpenAIError, requests.exceptions.RequestException)))
def safe_openai_call(*args, **kwargs):
return openai.ChatCompletion.create(*args, **kwargs)
@retry(wait=wait_exponential(multiplier=1, min=2, max=10), stop=stop_after_attempt(3),
retry=retry_if_exception_type(requests.exceptions.RequestException))
def safe_fetch(url):
return feedparser.parse(url)
# ---------- Core Logic ----------
def extract_fields(text: str, schema: BaseModel) -> CloudflarePostInsight:
prompt = f"""
Extract the following fields as a JSON object, following the schema:
{schema.schema_json(indent=2)}
Text:
{text}
"""
response = safe_openai_call(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0,
max_tokens=512,
response_format={"type": "json_object"},
)
try:
data = json.loads(response.choices[0].message.content)
return CloudflarePostInsight(**data)
except ValidationError as ve:
raise ValueError(f"Extraction failed validation: {ve}")
except json.JSONDecodeError as je:
raise ValueError(f"Extraction returned invalid JSON: {je}")
def critic(text: str, extraction: CloudflarePostInsight) -> Feedback:
prompt = f"""
You are a critic. Evaluate the following extraction against the original text.
Original text:
{text}
Extraction:
{extraction.json(indent=2)}
Check for factual accuracy, completeness, and whether the key takeaway truly reflects the text.
Respond with a JSON object matching this schema:
{Feedback.schema_json(indent=2)}
"""
response = safe_openai_call(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0,
max_tokens=256,
response_format={"type": "json_object"},
)
return Feedback(**json.loads(response.choices[0].message.content))
def self_healing_extraction(text: str, schema: BaseModel, max_attempts: int = 3) -> CloudflarePostInsight:
attempt = 0
tweak = ""
while attempt < max_attempts:
try:
extraction = extract_fields(text + tweak, schema)
feedback = critic(text, extraction)
if feedback.success:
return extraction
if feedback.prompt_tweak:
tweak += f"\n\nAdditional instruction: {feedback.prompt_tweak}"
except Exception as e:
logging.warning(f"Attempt {attempt+1} failed: {e}")
attempt += 1
raise RuntimeError(f"Extraction failed after {max_attempts} attempts")
# ---------- Main ----------
def main():
feed_url = "https://blog.cloudflare.com/rss/"
entries = load_feed(feed_url)
insights = []
for entry in entries:
entry_hash = hash_entry(entry)
if entry_hash in processed_hashes:
logging.info(f"Skipping already processed entry: {entry.title}")
continue
try:
insight = self_healing_extraction(entry.summary, CloudflarePostInsight)
insights.append(insight.dict())
processed_hashes.add(entry_hash)
logging.info(f"Successfully extracted: {entry.title}")
except Exception as e:
logging.error(f"Failed to process {entry.title}: {e}")
# Persist results
with open("insights.json", "w") as f:
json.dump(insights, f, indent=2)
# Plot sentiment distribution
sentiments = [i["sentiment"] for i in insights]
counts = {s: sentiments.count(s) for s in set(sentiments)}
plt.bar(counts.keys(), counts.values())
plt.title("Sentiment Distribution")
plt.xlabel("Sentiment")
plt.ylabel("Count")
plt.savefig("sentiment_distribution.png")
logging.info("Saved sentiment plot to sentiment_distribution.png")
if __name__ == "__main__":
main()
Expected Output
Running the script will produce two files in the current directory:
insights.json– a list of structured insights for the latest five Cloudflare posts.sentiment_distribution.png– a bar chart showing how many posts were classified as Positive, Neutral, or Negative.
Limitations and Tradeoffs
• The approach relies on an external LLM; costs can grow quickly with many posts. • Tenacity retries add latency; for high‑frequency feeds you may need to batch requests. • The critic model may over‑penalize subtle deviations, leading to unnecessary prompt tweaks. • Idempotency is only at the entry level; if the same article is republished with a different summary, the hash will change and the system will re‑process it. • Sentiment extraction is naive; a dedicated sentiment model would yield more reliable results.
Frequently Asked Questions
Why use a separate critic LLM instead of relying solely on pydantic validation?
Pydantic catches only structural errors—missing keys or wrong types. It cannot judge whether the extracted summary truly reflects the source text or if a product name was hallucinated. The critic adds semantic verification.
How do I prevent runaway API costs when scaling this agent to thousands of documents?
Implement caching keyed by the content hash of each RSS entry. If an entry’s hash has been seen before, skip the extraction entirely. Also batch LLM calls or use cheaper models for the critic.
Can this pattern be used with local open‑source models via Ollama?
Yes. Replace openai.ChatCompletion.create with a compatible Ollama endpoint that supports JSON output. Ensure the local model can generate structured responses.
What if the RSS feed stops providing summaries?
Fall back to fetching the full article via requests and performing a short extractive summarization before feeding it to the extractor.
How do I handle entries with extremely long summaries?
Chunk the summary into 2000‑token blocks, run the extractor on each chunk, then merge the results. Keep track of partial extractions to avoid duplication.
What I'd Change
In production I’d replace the simple in‑memory processed_hashes set with a persistent cache like Redis or SQLite, so the agent survives restarts. I’d also move the critic logic into a separate microservice, allowing us to swap in a more sophisticated semantic verifier without touching the extraction core. Finally, I’d instrument the pipeline with distributed tracing (OpenTelemetry) so I can see the latency of each LLM call and the frequency of retries, making it easier to spot bottlenecks as the feed volume grows.