Build Self-Managing Data Pipelines With an LLM Agent

Six-hour data pipeline. Spot termination. Job crashes. 45 minutes of compute lost. Engineer paged at 2 AM.

This isn't a tooling problem — it's a decision-making problem. And humans don't scale.

This guide shows you how to build an LLM-powered agent that makes these decisions autonomously: checkpoint timing, instance selection, and failure recovery. In our test environment, this approach reduced cost and operational toil compared with a manual baseline.

What you'll build: An autonomous orchestrator using Python, AWS, Terraform, and Claude.

The Problem: Pipelines Need Constant Babysitting

Long-running data pipelines on spot instances require constant judgment calls:

• When should I checkpoint? (Too often = overhead, too rare = lost work)
• Which instance types are stable right now?
• Should I scale up or ride it out?
• Is this termination warning real or noise?

Manual monitoring is unsustainable. Teams typically choose between:

1. Avoid spot instances → Pay significantly more for on-demand
2. Use simple rules → Require manual intervention when rules fail edge cases
3. Accept failures → Lose hours of compute regularly

An LLM agent can evaluate these tradeoffs in real-time and adapt to conditions that are cumbersome to capture with static rules alone.

Prerequisites

Before implementing this pattern, make sure you have:

• An AWS account with access to EC2, Auto Scaling, S3, CloudWatch, and Budgets
• IAM roles/policies for read-only observation mode and controlled execution mode
• Python 3.10+ for the orchestrator runtime
• Terraform 1.5+ for infrastructure guardrails and reproducible deployment
• A clear rollback/runbook path before enabling autonomous actions in production

Quick Start Repository Layout

costagent/
├── src/
│   ├── monitor.py
│   ├── planner.py
│   ├── validator.py
│   └── executor.py
├── infra/
│   ├── main.tf
│   ├── autoscaling.tf
│   └── budgets.tf
├── scripts/
│   ├── run_observation_mode.sh
│   └── replay_decisions.py
└── tests/
    ├── test_validator.py
    └── test_integration.py

Architecture Overview

The system runs a continuous loop — monitoring state, asking the LLM for decisions, validating for safety, then executing:

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│    State     │────▶│  LLM Planner │────▶│  Validator   │
│   Monitor    │     │   (Claude)   │     │  (Clamping)  │
└──────────────┘     └──────────────┘     └──────┬───────┘
       ▲                                         │
       │                                         ▼
       │                                  ┌──────────────┐
       └──────────────────────────────────│   Executor   │
                                          │  (AWS SDK)   │
                                          └──────────────┘

The key insight: The LLM proposes decisions, while deterministic guardrails constrain execution risk.

Step 1: Build the State Monitor

First, gather everything the LLM needs to make informed decisions:

@dataclass
class PipelineState:
    active_instances: List[InstanceInfo]
    spot_prices: Dict[str, float]  # Current prices by type
    termination_notices: List[str]  # 2-min warnings
    pipeline_progress: float  # 0.0 to 1.0
    minutes_since_checkpoint: int
    budget_remaining: float

The monitor queries EC2 APIs every 30–60 seconds and formats this into a prompt the LLM can reason about.

Step 2: Build the LLM Planner

The planner sends current state to the LLM and gets back a structured decision:

@dataclass
class AgentDecision:
    action: str  # SCALE_UP, SCALE_DOWN, CHECKPOINT, MIGRATE, WAIT
    instance_type: Optional[str]
    instance_count: Optional[int]
    should_checkpoint: bool
    reasoning: str  # LLM explains its thinking

Example LLM reasoning:

"Spot prices for c6i.4xlarge spiked 40% in the last hour, and we received a termination notice. Checkpointing now and migrating to c5.2xlarge which has stable pricing. Progress is at 73% so we can't afford to lose work."

This is the kind of nuanced, multi-factor reasoning that's impossible with static rules.

Step 3: Build the Safety Validator (Critical)

Never trust LLM output directly. The validator ensures any decision is safe and executable:

def clamp_decision(decision, state):
    return ValidDecision(
        action=decision.action if decision.action in VALID_ACTIONS else "WAIT",
        instance_type=decision.instance_type if decision.instance_type in ALLOWED_TYPES else DEFAULT,
        count=min(max(decision.instance_count or 0, 0), MAX_INSTANCES),
        checkpoint=state.minutes_since_checkpoint > 30 or decision.should_checkpoint,
    )

Clamping rules:

Key principle: Prefer safe transformation over hard rejection. The pipeline can continue operating safely even when model output is imperfect.

Step 4: Build the Executor

The executor applies validated decisions to AWS:

• CHECKPOINT: Save state to S3, record timestamp
• SCALE_UP: Launch spot instances with ManagedBy: costagent tag
• SCALE_DOWN: Terminate oldest instances first
• MIGRATE: Checkpoint → terminate → launch new type

All operations are idempotent and logged for debugging.

Step 5: Add Infrastructure Guardrails

Even if your code has bugs, infrastructure limits protect you:

resource "aws_autoscaling_group" "pipeline_workers" {
  max_size = 50  # Hard ceiling - cannot exceed
  min_size = 0

  mixed_instances_policy {
    instances_distribution {
      spot_max_price = "1.00"  # Price ceiling
    }
  }
}

Three layers of defense:

1. Application: Clamping function (milliseconds)
2. Infrastructure: Terraform limits (can't be bypassed)
3. Cloud: AWS Budgets hard stop at limit

In practice, this layered design reduces blast radius by requiring multiple independent failures before severe impact.

Step 6: Run in Observation Mode First

Before enabling autonomous execution, watch the decisions:

def main_loop():
    while pipeline_running:
        state     = monitor.get_state()
        decision  = planner.get_decision(state)
        validated = validator.clamp(decision)

        # Week 1-2: Log only
        logger.info(f"Would execute: {validated}")
        # executor.run(validated)  # Disabled initially

        time.sleep(60)

Observation checklist:

• Decisions make sense for the scenario
• Clamping triggers are logged (shows edge cases)
• Decision latency is acceptable (10–20 sec typical)
• No unexpected patterns in recommendation log
• Post-review signoff complete before enabling execution

Practical Outcomes

In internal testing, this architecture reduced manual intervention and improved interruption handling compared with a manual spot-operations workflow.

The practical gains came from:

1. Deterministic safety guardrails around model output
2. Faster reactions to spot interruptions and changing price conditions
3. Better checkpoint discipline during long-running jobs
4. Less operator toil during failure recovery

Actual cost and reliability outcomes vary by workload shape, instance availability, checkpoint cadence, and service limits in each environment.

Common Production Pitfalls

• Skipping observation mode: Going straight to execution increases operational risk.
• Weak IAM boundaries: Over-permissive roles can turn a bad decision into a large blast radius.
• No checkpoint restore drills: Checkpointing is only useful if restore is tested regularly.
• Unbounded instance choices: Keep an allowlist of tested instance families per workload.
• No decision audit trail: Persist state, model output, clamp result, and action outcome for every cycle.

Troubleshooting

LLM decisions are too slow (>30 sec).

• Check API latency to Anthropic
• Reduce prompt size by summarizing history
• Consider caching similar decisions

Clamping triggers constantly.

• Review allowed instance types list
• Check if LLM prompt has correct constraints
• May need to expand action space

Pipeline still failing.

• Check executor error logs
• Verify IAM permissions for EC2/S3
• Ensure checkpoint restore works end-to-end

Key Takeaways

Infrastructure operations are gradually moving from dashboard-heavy monitoring to systems that can make bounded, auditable decisions.

1. Autonomy is the goal, cost savings are the side effect — Once pipelines make their own decisions, optimization happens continuously without human bottlenecks.
2. Guardrails unlock trust — You don't need to trust AI judgment. You need constraints that make bad judgment survivable.
3. Observe, then automate — Two weeks of shadow mode builds confidence and catches edge cases before they matter.
4. Transform, never reject — Clamping keeps systems running. A suboptimal decision beats a halted pipeline every time.

This pattern — LLM reasoning within deterministic bounds — will define the next generation of infrastructure automation. Start small, prove value, expand scope.

Get the Complete Project

This guide covers the architecture. For the full implementation:

• Complete Python source code
• Terraform infrastructure modules
• Deployment scripts and configuration
• Testing framework