Skip to content
Home
gen-ai
prompting

Beyond the First Try: Engineering Self‑Correcting Prompts for Robust Data Extraction

Mausam

When I first tried to pull structured fields out of the Stripe Blog RSS feed, the LLM would happily spit out malformed JSON, drop a title, or hallucinate a link. In a production pipeline that feeds downstream analytics, those glitches cascade into broken dashboards and angry stakeholders. If you’re a data engineer who already knows how to call an LLM but still sees “oops, my JSON is broken” in the logs, this guide shows you how to turn a one‑shot prompt into a self‑correcting, multi‑stage extraction loop that guarantees schema‑compliant output—or a clean failure you can surface to observability.

Key Takeaways

  • Few‑shot examples plus a strict Pydantic schema dramatically reduce hallucinations.
  • Embedding a “refine‑if‑invalid” prompt inside a retry loop lets the LLM fix its own output without human intervention.
  • Explicit error handling (JSON decode, validation, LLM timeouts) combined with structured logging gives you observability that scales.
  • Self‑correcting prompts are cheap: the extra token budget is far lower than the cost of downstream data quality incidents.
  • When the LLM fails after N retries, fail fast and log the raw text so downstream services can decide to skip or flag the entry.

The Problem

Stripe publishes a daily RSS feed that mixes short summaries, rich HTML, and occasionally missing fields. My downstream analytics expect a clean JSON object per entry with title, summary, link, and published. A naive prompt (“Extract JSON with those keys”) works 60 % of the time, but the remaining 40 % produce malformed JSON, duplicate keys, or completely fabricated URLs—unacceptable for a production ETL that runs every hour.

Data and Sources

We pull the feed directly from Stripe’s public RSS endpoint: https://stripe.com/blog/feed.rss. The feed is parsed with feedparser, a battle‑tested library for RSS/Atom.
Data accessed on 2026-10-31.

Loading the Data

First we fetch the five most recent entries and keep only the fields we need for extraction.

import feedparser

def load_latest_entries(limit: int = 5):
    feed = feedparser.parse("https://stripe.com/blog/feed.rss")
    entries = []
    for e in feed.entries[:limit]:
        entries.append({
            "raw_title": e.title,
            "raw_summary": e.summary,
            "raw_link": e.link,
            "raw_published": e.published,
        })
    return entries

Step 1 — Naïve Extraction (Why It Fails)

A single‑shot prompt looks like this:

prompt = f\"\"\"Extract a JSON object with the keys:
title, summary, link, published.
Use the following raw text:
Title: {title}
Summary: {summary}
Link: {link}
Published: {published}
\"\"\"

The LLM often returns something close, but missing commas or quoting errors break json.loads. That’s why we need a schema guard.

Step 2 — Guarding with Pydantic

Define the contract the LLM must obey.

from pydantic import BaseModel, HttpUrl, ValidationError
from datetime import datetime

class BlogEntry(BaseModel):
    title: str
    summary: str
    link: HttpUrl
    published: datetime

If the LLM’s output violates any field, BlogEntry.parse_raw raises a ValidationError. We can catch that and trigger a correction.

Step 3 — In‑Prompt Self‑Correction Loop

The core idea: after a failed parse, feed the LLM its own output plus a clear “fix the JSON” instruction. We limit retries to three attempts to avoid endless loops.

import json, time, logging

logger = logging.getLogger(__name__)

def call_llm(prompt: str) -> str:
    """
    Placeholder for the real LLM call.
    Replace with openai.ChatCompletion.create or your hosted model.
    """
    # Simulated response for illustration:
    return "{\"title\": \"...\", \"summary\": \"...\", \"link\": \"...\", \"published\": \"...\"}"

def extract_with_correction(raw):
    base_prompt = f\"\"\"Extract a JSON object with these keys:
title, summary, link, published.
Use the raw fields below exactly as they appear.
---\nTitle: {raw['raw_title']}\nSummary: {raw['raw_summary']}\nLink: {raw['raw_link']}\nPublished: {raw['raw_published']}\n---\"\"\"
    
    attempt = 0
    last_response = ""
    while attempt < 3:
        attempt += 1
        try:
            response = call_llm(base_prompt if attempt == 1 else f\"\"\"The previous JSON was invalid:
{last_response}
Please return a corrected JSON that matches the schema exactly.\"\"\")
            # Try to load raw JSON first (some models wrap in markdown)
            cleaned = response.strip()
            if cleaned.startswith("```"):
                cleaned = cleaned.split("```")[1].strip()
            data = json.loads(cleaned)
            # Validate with Pydantic
            entry = BlogEntry.parse_obj(data)
            return entry
        except (json.JSONDecodeError, ValidationError) as exc:
            logger.warning("Attempt %d failed: %s", attempt, exc)
            last_response = response
            time.sleep(0.5)  # back‑off before retry
    # All retries exhausted
    logger.error("Extraction failed after %d attempts for title: %s", attempt, raw['raw_title'])
    return None

Step 4 — Productionizing: Retries, Observability, and Idempotence

Wrap the whole pipeline in a function that logs successes, failures, and latency metrics. Idempotence is achieved by using the entry’s canonical link as a primary key when persisting downstream.

import uuid

def process_feed():
    entries = load_latest_entries()
    results = []
    for raw in entries:
        start = time.time()
        entry = extract_with_correction(raw)
        latency = time.time() - start
        if entry:
            logger.info("✅ Extracted %s in %.2fs", entry.title, latency)
            results.append(entry.dict())
        else:
            logger.info("❌ Skipped entry with raw title: %s", raw['raw_title'])
    return results

Putting It Together

Now we tie loading, extraction, and logging into a single script that can be scheduled (Airflow, Prefect, or a simple cron). The script prints each validated JSON line; failures are logged to stderr.

if __name__ == "__main__":
    import logging, sys
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
        stream=sys.stdout,
    )
    extracted = process_feed()
    for obj in extracted:
        print(json.dumps(obj, ensure_ascii=False))

Complete Script

The full runnable script combining all steps:

#!/usr/bin/env python3
"""
Self‑correcting LLM extraction demo for Stripe Blog RSS.
Fetches the 5 latest entries, extracts a validated JSON payload,
and prints each successful object. Failures are logged.
"""

import feedparser
import json
import logging
import sys
import time
from datetime import datetime
from typing import List, Optional

from pydantic import BaseModel, HttpUrl, ValidationError

# ----------------------------------------------------------------------
# Schema definition
# ----------------------------------------------------------------------
class BlogEntry(BaseModel):
    title: str
    summary: str
    link: HttpUrl
    published: datetime

# ----------------------------------------------------------------------
# LLM placeholder – replace with your provider call
# ----------------------------------------------------------------------
def call_llm(prompt: str) -> str:
    """
    Simulated LLM response. In production replace this with
    openai.ChatCompletion.create, Cohere, Anthropic, etc.
    """
    # For demonstration we just echo a well‑formed JSON.
    # Real implementation would send `prompt` to the model.
    dummy = {
        "title": "Dummy Title",
        "summary": "Dummy summary extracted from raw text.",
        "link": "https://example.com/dummy",
        "published": "2026-10-31T00:00:00Z",
    }
    return json.dumps(dummy)

# ----------------------------------------------------------------------
# Data ingestion
# ----------------------------------------------------------------------
def load_latest_entries(limit: int = 5) -> List[dict]:
    feed = feedparser.parse("https://stripe.com/blog/feed.rss")
    entries = []
    for e in feed.entries[:limit]:
        entries.append({
            "raw_title": e.title,
            "raw_summary": e.summary,
            "raw_link": e.link,
            "raw_published": e.published,
        })
    return entries

# ----------------------------------------------------------------------
# Extraction with self‑correction
# ----------------------------------------------------------------------
def extract_with_correction(raw: dict) -> Optional[BlogEntry]:
    base_prompt = f\"\"\"Extract a JSON object with these keys:
title, summary, link, published.
Use the raw fields below exactly as they appear.
---\nTitle: {raw['raw_title']}\nSummary: {raw['raw_summary']}\nLink: {raw['raw_link']}\nPublished: {raw['raw_published']}\n---\"\"\"
    
    attempt = 0
    last_response = ""
    while attempt < 3:
        attempt += 1
        try:
            response = call_llm(base_prompt if attempt == 1 else f\"\"\"The previous JSON was invalid:
{last_response}
Please return a corrected JSON that matches the schema exactly.\"\"\")
            cleaned = response.strip()
            if cleaned.startswith("```"):
                cleaned = cleaned.split("```")[1].strip()
            data = json.loads(cleaned)
            entry = BlogEntry.parse_obj(data)
            return entry
        except (json.JSONDecodeError, ValidationError) as exc:
            logging.warning("Attempt %d failed: %s", attempt, exc)
            last_response = response
            time.sleep(0.5)  # simple back‑off
    logging.error("Extraction failed after %d attempts for title: %s", attempt, raw['raw_title'])
    return None

# ----------------------------------------------------------------------
# Orchestration
# ----------------------------------------------------------------------
def process_feed() -> List[dict]:
    entries = load_latest_entries()
    results = []
    for raw in entries:
        start = time.time()
        entry = extract_with_correction(raw)
        latency = time.time() - start
        if entry:
            logging.info("✅ Extracted %s in %.2fs", entry.title, latency)
            results.append(entry.dict())
        else:
            logging.info("❌ Skipped entry with raw title: %s", raw['raw_title'])
    return results

# ----------------------------------------------------------------------
# Entry point
# ----------------------------------------------------------------------
if __name__ == "__main__":
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
        stream=sys.stdout,
    )
    extracted = process_feed()
    for obj in extracted:
        print(json.dumps(obj, ensure_ascii=False))

Expected Output

Running python script.py prints one JSON line per successfully extracted entry, e.g.:

{"title":"Stripe on Stripe: What we learned moving $1B in vendor payments to Global Payouts","summary":"In early 2026, we began moving our vendor payments to Global Payouts...","link":"https://stripe.com/blog/global-payouts","published":"2026-10-30T12:45:00Z"}

Failed entries produce only log lines; no JSON is printed for them.

Limitations and Tradeoffs

This approach assumes the LLM can understand a “fix‑my‑JSON” instruction reliably. Very small models may not improve after the correction prompt, leading to wasted retries. The retry loop adds latency (≈1 s per failed attempt) and token cost; in high‑throughput pipelines you might batch multiple entries per LLM call to amortize overhead. Finally, schema validation is strict—if Stripe changes field names, the pipeline will start failing until the schema is updated.

Frequently Asked Questions

What if the LLM returns extra fields?

Pydantic will ignore unknown fields by default if you set extra = "ignore" in the model Config, but it’s safer to keep the schema tight and let the validation fail so you notice upstream changes.

Can I parallelize the extraction?

Yes. Wrap extract_with_correction in a thread pool or async task. Just make sure the LLM client you use is thread‑safe or use a connection pool.

How do I surface failures to downstream systems?

Emit a structured log entry or push a message to a dead‑letter queue (Kafka, SQS). Include the raw text and the last LLM response so an analyst can re‑run the entry manually.

Is there a way to reduce token usage for the correction prompt?

Cache the raw JSON you sent on the first attempt and only send the diff (the error message) in the second prompt. Most providers charge per token, so trimming the correction prompt can save 10‑15 % per entry.

Do I need a separate validation library?

Pydantic already gives you type coercion, URL validation, and datetime parsing. If you prefer a lighter footprint, marshmallow or dataclasses-json work, but they lack the built‑in error aggregation that makes debugging easier.

What I'd Change

If I were to rebuild this for a billion‑row nightly feed, I’d replace the per‑entry LLM call with a batched prompt that returns an array of JSON objects. That cuts token cost dramatically and lets me parallelize the validation step. I’d also instrument Prometheus metrics for success rate, latency, and token consumption so the ops team can set alerts before a schema drift silently degrades data quality. In short, the self‑correcting pattern is the right foundation, but scaling it requires batch‑level thinking and tighter observability.

Post a Comment

Hi! How can we help you? Send us a message and we'll get back to you.