Building an agentic data pipeline that self-corrects on errors

A data pipeline that breaks silently is worse than one that breaks loudly. A pipeline that self-corrects is better than both. LLMs make self-correcting pipelines practical for the first time: instead of hand-coding recovery logic for every failure mode, you can give the model the error context and let it diagnose and retry intelligently.

This guide covers the architecture and implementation patterns for agentic pipelines that catch errors, reason about them, and recover — without human intervention for the common cases.

The anatomy of a self-correcting pipeline

A self-correcting pipeline has three layers that traditional pipelines lack:

1. Error detection — structured validation of intermediate outputs, not just exception catching
2. Error diagnosis — determining what went wrong and why
3. Recovery orchestration — deciding whether to retry, transform the input, skip, or escalate

The key insight is that errors in LLM pipelines are usually recoverable. Unlike a database failure that requires human intervention, an LLM producing malformed JSON can often self-correct if given its error and another attempt.

Structured output validation as the first line of defence

Most LLM errors in production are schema violations, not exceptions. The model returns something that looks right but fails validation. Catch these explicitly:

from pydantic import BaseModel, ValidationError
from anthropic import Anthropic
import json

client = Anthropic()

class PipelineError(Exception):
    def __init__(self, stage: str, error_type: str, details: str, recoverable: bool):
        self.stage = stage
        self.error_type = error_type
        self.details = details
        self.recoverable = recoverable
        super().__init__(f"[{stage}] {error_type}: {details}")

class ExtractionResult(BaseModel):
    entities: list[str]
    sentiment: str  # "positive", "negative", "neutral"
    key_facts: list[str]
    confidence: float

def extract_with_validation(text: str) -> ExtractionResult:
    """Extract data with schema validation and structured error reporting."""
    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=1024,
        tools=[{
            "name": "extract_data",
            "description": "Extract structured data from text",
            "input_schema": ExtractionResult.model_json_schema(),
        }],
        tool_choice={"type": "tool", "name": "extract_data"},
        messages=[{"role": "user", "content": f"Extract data from: {text}"}],
    )

    tool_use = next(
        (b for b in response.content if b.type == "tool_use"), None
    )

    if not tool_use:
        raise PipelineError(
            stage="extraction",
            error_type="no_tool_use",
            details="Model did not call the extraction tool",
            recoverable=True,
        )

    try:
        return ExtractionResult(**tool_use.input)
    except ValidationError as e:
        raise PipelineError(
            stage="extraction",
            error_type="schema_violation",
            details=str(e),
            recoverable=True,
        )

The self-correction loop

When an error is recoverable, feed it back to the model with explicit correction instructions:

from typing import TypeVar, Callable, Optional
import time

T = TypeVar("T")

def with_self_correction(
    fn: Callable[[], T],
    max_attempts: int = 3,
    correction_prompt: Optional[Callable[[Exception], str]] = None,
) -> T:
    """
    Execute a function with LLM-powered self-correction on failure.
    On each failure, the error is logged and the attempt retries with
    additional context about what went wrong.
    """
    last_error = None

    for attempt in range(max_attempts):
        try:
            return fn()
        except PipelineError as e:
            last_error = e
            if not e.recoverable:
                raise

            if attempt < max_attempts - 1:
                wait = 2 ** attempt  # Exponential backoff
                print(f"Attempt {attempt + 1} failed: {e}. Retrying in {wait}s...")
                time.sleep(wait)
        except Exception as e:
            raise PipelineError(
                stage="unknown",
                error_type=type(e).__name__,
                details=str(e),
                recoverable=False,
            ) from e

    raise last_error


def extract_with_correction(text: str, previous_error: str = "") -> ExtractionResult:
    """
    Extraction function that accepts previous error context for correction.
    """
    correction_context = ""
    if previous_error:
        correction_context = f"""
PREVIOUS ATTEMPT FAILED:
{previous_error}

Please correct the issues above in your response."""

    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=1024,
        tools=[{
            "name": "extract_data",
            "description": "Extract structured data from text",
            "input_schema": ExtractionResult.model_json_schema(),
        }],
        tool_choice={"type": "tool", "name": "extract_data"},
        messages=[{
            "role": "user",
            "content": f"Extract data from: {text}{correction_context}"
        }],
    )

    tool_use = next(b for b in response.content if b.type == "tool_use")
    return ExtractionResult(**tool_use.input)

Stateful pipeline orchestration

For multi-stage pipelines, track state so partial progress survives failures:

from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime

class StageStatus(str, Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    SKIPPED = "skipped"

@dataclass
class StageResult:
    stage_name: str
    status: StageStatus
    output: object = None
    error: str = None
    attempts: int = 0
    duration_ms: int = 0

@dataclass
class PipelineRun:
    pipeline_id: str
    input_data: str
    stages: list[StageResult] = field(default_factory=list)
    started_at: datetime = field(default_factory=datetime.utcnow)
    completed_at: datetime = None

    def get_stage(self, name: str) -> Optional[StageResult]:
        return next((s for s in self.stages if s.stage_name == name), None)

    def is_stage_done(self, name: str) -> bool:
        stage = self.get_stage(name)
        return stage is not None and stage.status == StageStatus.COMPLETED


class SelfCorrectingPipeline:
    def __init__(self, max_retries: int = 3):
        self.max_retries = max_retries
        self.stages: list[tuple[str, Callable]] = []

    def add_stage(self, name: str, fn: Callable) -> "SelfCorrectingPipeline":
        self.stages.append((name, fn))
        return self

    def run(self, input_data: str) -> PipelineRun:
        run = PipelineRun(
            pipeline_id=f"run-{int(datetime.utcnow().timestamp())}",
            input_data=input_data,
        )

        current_output = input_data

        for stage_name, stage_fn in self.stages:
            # Resume from checkpoint if stage already completed
            if run.is_stage_done(stage_name):
                current_output = run.get_stage(stage_name).output
                continue

            stage_result = StageResult(
                stage_name=stage_name,
                status=StageStatus.RUNNING,
            )
            run.stages.append(stage_result)

            start = datetime.utcnow()
            last_error = ""

            for attempt in range(self.max_retries):
                try:
                    stage_result.attempts = attempt + 1
                    output = stage_fn(current_output, last_error)
                    stage_result.status = StageStatus.COMPLETED
                    stage_result.output = output
                    current_output = output
                    break
                except PipelineError as e:
                    last_error = f"{e.error_type}: {e.details}"
                    if not e.recoverable or attempt == self.max_retries - 1:
                        stage_result.status = StageStatus.FAILED
                        stage_result.error = last_error
                        run.completed_at = datetime.utcnow()
                        return run
                    time.sleep(2 ** attempt)

            stage_result.duration_ms = int(
                (datetime.utcnow() - start).total_seconds() * 1000
            )

        run.completed_at = datetime.utcnow()
        return run

LLM-powered error diagnosis

For non-obvious failures, ask the model to diagnose the error before retrying:

def diagnose_and_recover(
    stage_name: str,
    input_data: str,
    error: PipelineError,
    context: dict,
) -> dict:
    """
    Ask an LLM to diagnose a pipeline error and suggest a recovery strategy.
    """
    diagnosis_response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": f"""A pipeline stage failed. Diagnose the issue and
suggest a recovery strategy.

STAGE: {stage_name}
ERROR TYPE: {error.error_type}
ERROR DETAILS: {error.details}
INPUT SAMPLE: {str(input_data)[:500]}
CONTEXT: {json.dumps(context, default=str)}

Return JSON: {{
  "root_cause": "brief explanation",
  "recovery_strategy": "retry" | "transform_input" | "skip" | "escalate",
  "input_transformation": "how to modify input before retrying, if applicable",
  "confidence": 0.0 to 1.0
}}"""
        }]
    )

    text = diagnosis_response.content[0].text
    if "```" in text:
        text = text.split("```")[1].replace("json", "").strip()

    return json.loads(text)

When to stop retrying

Self-correction has limits. A well-designed pipeline knows when to stop:

UNRECOVERABLE_ERRORS = {
    "rate_limit_exceeded",    # Back off at the scheduler level, not here
    "context_length_exceeded", # Input is too long; needs chunking, not retrying
    "invalid_api_key",         # Configuration issue, not transient
    "content_policy_violation", # Input itself is the problem
}

def should_retry(error: PipelineError, attempt: int, max_attempts: int) -> bool:
    if not error.recoverable:
        return False
    if error.error_type in UNRECOVERABLE_ERRORS:
        return False
    if attempt >= max_attempts - 1:
        return False
    return True

Always implement a dead-letter queue for records that exhaust retries. Failed records should be preserved with their error context for human review or offline reprocessing — silently dropping them is almost never the right choice.

Observability and dead-letter queues

A self-correcting pipeline that you cannot observe is a black box. Instrument every stage so you can answer: how often does each stage fail on the first attempt, which errors trigger the most retries, and which records eventually exhaust retries and land in the dead-letter queue?

import logging
from dataclasses import dataclass, asdict

logger = logging.getLogger("pipeline")

@dataclass
class PipelineMetrics:
    pipeline_id: str
    total_records: int = 0
    successful: int = 0
    recovered: int = 0      # Succeeded after at least one retry
    dead_lettered: int = 0
    stage_failures: dict = None

    def __post_init__(self):
        if self.stage_failures is None:
            self.stage_failures = {}

    def record_stage_failure(self, stage: str, error_type: str):
        key = f"{stage}.{error_type}"
        self.stage_failures[key] = self.stage_failures.get(key, 0) + 1

    def log_summary(self):
        recovery_rate = (
            self.recovered / max(self.recovered + self.dead_lettered, 1) * 100
        )
        logger.info(
            "Pipeline run complete",
            extra={
                **asdict(self),
                "recovery_rate_pct": round(recovery_rate, 1),
            }
        )


def run_pipeline_with_metrics(
    pipeline: SelfCorrectingPipeline,
    records: list[str],
    dead_letter_path: str,
) -> PipelineMetrics:
    """
    Run a pipeline over a list of records, routing failures to a dead-letter file.
    """
    metrics = PipelineMetrics(
        pipeline_id=f"batch-{int(datetime.utcnow().timestamp())}",
        total_records=len(records),
    )
    dead_letter_records = []

    for record in records:
        run = pipeline.run(record)
        failed_stages = [s for s in run.stages if s.status == StageStatus.FAILED]
        retried_stages = [s for s in run.stages if s.attempts > 1]

        if failed_stages:
            metrics.dead_lettered += 1
            for stage in failed_stages:
                metrics.record_stage_failure(stage.stage_name, stage.error or "unknown")
            dead_letter_records.append({
                "input": record,
                "pipeline_id": run.pipeline_id,
                "failed_stage": failed_stages[0].stage_name,
                "error": failed_stages[0].error,
            })
        else:
            metrics.successful += 1
            if retried_stages:
                metrics.recovered += 1

    # Write dead-letter records for offline review
    with open(dead_letter_path, "w") as f:
        json.dump(dead_letter_records, f, indent=2, default=str)

    metrics.log_summary()
    return metrics

With per-stage failure counts broken out by error type, you can spot patterns quickly. If extraction.schema_violation is consistently the top failure, that signals a prompt or schema mismatch worth fixing at the source rather than relying on retries.

Input chunking for context-length errors

One of the most common unrecoverable errors in LLM pipelines is context_length_exceeded. The right fix is not to retry with the same input — it is to chunk the input and process it in parts. Make this a first-class recovery strategy rather than an afterthought:

def chunk_text(text: str, max_chars: int = 4000, overlap: int = 200) -> list[str]:
    """
    Split text into overlapping chunks to preserve context at boundaries.
    """
    chunks = []
    start = 0
    while start < len(text):
        end = min(start + max_chars, len(text))
        chunks.append(text[start:end])
        start = end - overlap if end < len(text) else end
    return chunks


def extract_with_chunking(text: str) -> list[ExtractionResult]:
    """
    Handle long inputs by chunking and merging extraction results.
    Falls back to chunked processing when context length is exceeded.
    """
    try:
        result = extract_with_validation(text)
        return [result]
    except PipelineError as e:
        if e.error_type != "context_length_exceeded":
            raise

        logger.info(f"Input too long ({len(text)} chars), switching to chunked mode")
        chunks = chunk_text(text)
        results = []

        for i, chunk in enumerate(chunks):
            try:
                result = extract_with_validation(chunk)
                results.append(result)
            except PipelineError as chunk_error:
                logger.warning(f"Chunk {i} failed: {chunk_error}")
                # Continue with remaining chunks rather than aborting

        return results

Merging results across chunks depends on your schema. For entity extraction, union the entity lists. For sentiment, take a majority vote or average confidence-weighted scores. Design your ExtractionResult schema with chunk-merging in mind from the start.

The self-correcting pipeline pattern adds complexity, but the payoff is a system that handles the messy reality of LLM outputs without requiring you to hand-code every failure case. For high-volume processing where manual intervention does not scale, it is essential.

Parallel stage execution

Some pipeline stages are independent of each other and can run concurrently. Parallelism reduces total wall-clock time, but requires careful error handling because failures in one branch must not silently suppress failures in another:

import asyncio
from typing import Any

async def run_stage_async(
    stage_name: str,
    stage_fn: Callable,
    input_data: str,
    max_retries: int = 3,
) -> StageResult:
    """Run a single pipeline stage asynchronously with retry logic."""
    result = StageResult(stage_name=stage_name, status=StageStatus.RUNNING)
    last_error = ""

    for attempt in range(max_retries):
        try:
            result.attempts = attempt + 1
            output = await asyncio.to_thread(stage_fn, input_data, last_error)
            result.status = StageStatus.COMPLETED
            result.output = output
            return result
        except PipelineError as e:
            last_error = f"{e.error_type}: {e.details}"
            if not e.recoverable or attempt == max_retries - 1:
                result.status = StageStatus.FAILED
                result.error = last_error
                return result
            await asyncio.sleep(2 ** attempt)

    return result


async def run_parallel_stages(
    stages: list[tuple[str, Callable, str]],  # (name, fn, input)
    max_retries: int = 3,
) -> list[StageResult]:
    """
    Run multiple independent stages concurrently.
    All stages run regardless of individual failures — collect all results.
    """
    tasks = [
        run_stage_async(name, fn, inp, max_retries)
        for name, fn, inp in stages
    ]
    # gather with return_exceptions=False since we handle errors inside run_stage_async
    results = await asyncio.gather(*tasks)
    return list(results)


# Usage: fan out to three independent extraction stages, then merge
async def parallel_extraction_pipeline(text: str) -> dict:
    stages = [
        ("extract_entities", extract_entities_stage, text),
        ("extract_sentiment", extract_sentiment_stage, text),
        ("extract_keywords", extract_keywords_stage, text),
    ]
    results = await run_parallel_stages(stages)

    merged = {}
    for r in results:
        if r.status == StageStatus.COMPLETED:
            merged[r.stage_name] = r.output
        else:
            merged[r.stage_name] = None  # Downstream code handles missing fields

    return merged

Parallel execution works well when stages operate on the same input (fan-out pattern) rather than sequential stages where each feeds the next. For sequential stages, parallelism helps when you can overlap the next stage's preparation with the current stage's LLM call.

Testing self-correcting pipelines

The retry and correction logic is the most important part of the pipeline and the hardest to test with real LLM calls. Use dependency injection to replace the LLM client with a deterministic fake that produces failures on demand:

from unittest.mock import MagicMock, patch
import pytest

class FakeLLMClient:
    """
    A fake LLM client that returns preset responses in sequence.
    Use to test retry logic without making real API calls.
    """
    def __init__(self, responses: list):
        self._responses = iter(responses)
        self.call_count = 0

    def messages_create(self, **kwargs):
        self.call_count += 1
        response = next(self._responses)
        if isinstance(response, Exception):
            raise response
        return response

    @property
    def messages(self):
        # Mimic the client.messages.create() call pattern
        mock = MagicMock()
        mock.create = self.messages_create
        return mock


def make_tool_response(data: dict) -> MagicMock:
    """Build a fake API response containing a tool_use block."""
    block = MagicMock()
    block.type = "tool_use"
    block.input = data
    response = MagicMock()
    response.content = [block]
    return response


def test_self_correction_retries_on_schema_violation():
    """Pipeline should retry after schema_violation and succeed on second attempt."""
    bad_response = make_tool_response({
        "entities": ["Apple"],
        "sentiment": "INVALID_VALUE",  # Not in allowed set
        "key_facts": [],
        "confidence": 0.8,
    })
    good_response = make_tool_response({
        "entities": ["Apple"],
        "sentiment": "positive",
        "key_facts": ["Apple releases new product"],
        "confidence": 0.9,
    })

    fake_client = FakeLLMClient([bad_response, good_response])

    with patch("mymodule.client", fake_client):
        result = extract_with_correction("Apple released a new product today.")

    assert result.sentiment == "positive"
    assert fake_client.call_count == 2  # First attempt failed, second succeeded


def test_pipeline_dead_letters_after_max_retries():
    """After max retries, the stage should be marked FAILED, not raise."""
    always_bad = make_tool_response({
        "entities": [],
        "sentiment": "BROKEN",
        "key_facts": [],
        "confidence": -1.0,  # Invalid
    })

    fake_client = FakeLLMClient([always_bad, always_bad, always_bad])
    pipeline = SelfCorrectingPipeline(max_retries=3)
    pipeline.add_stage("extract", lambda text, err: extract_with_validation(text))

    with patch("mymodule.client", fake_client):
        run = pipeline.run("some input text")

    assert run.stages[0].status == StageStatus.FAILED
    assert run.stages[0].attempts == 3

Testing with deterministic fakes lets you cover failure scenarios that are hard to trigger reliably with real API calls: rate limits, malformed responses, partial tool outputs, and schema violations on specific fields. Aim for unit tests that cover every branch of should_retry and at least one integration test per stage using the fake client.

Choosing the right recovery strategy per error type

Not every error warrants the same recovery path. A structured mapping of error types to strategies keeps your recovery logic predictable and avoids the trap of retrying blindly:

Implement this mapping as a lookup table rather than a chain of if-else conditions. It is easier to extend, easier to test, and makes the recovery policy explicit and auditable. When a new error type appears in your dead-letter queue, add it to the table with a deliberate strategy rather than letting it fall through to a default retry.