Where Reasoning Belongs in an Agentic Data Pipeline

Building a two-level loop where LLMs schedule and rules execute.

Image generated by Gemini

I had an Airflow operator that fetched netflow data from our network observability API across 70+ Kubernetes clusters every day. It worked, mostly.

Netflow data is network traffic: which workloads are sending bytes where, and how much. It’s the raw input for cross-zone and cross-region cost attribution: it tells you which services are generating expensive traffic, not just which ones are busy. The API returns workload objects nested with per-destination byte counts, and the responses get large; query too wide a window and the API returns a 500. You get nothing for that window rather than degraded data.

The window sizes were hardcoded: 15 minutes for the busy clusters, 60 minutes for everything else. Those numbers came from trial and error months earlier and nobody touched them after that. New clusters defaulted to 60 minutes. You’d find out the number was wrong when the Hive partition had gaps. Not from a good monitoring system, but from someone on the analytics team asking why their numbers looked off.

There was no prioritization. The operator processed clusters in whatever order the API returned them. Four of those clusters were critical: production traffic, cost attribution downstream. If they happened to run last and hit rate limits, they failed quietly. Nothing in the code knew they mattered more than the rest.

And the operator had no memory. If a cluster needed 20-minute windows to return clean data three days in a row, the operator didn’t know that. It tried 60 minutes, got throttled, fell back, and you manually updated a dict somewhere in the codebase. Then you forgot about it until the next new cluster.

The Two-Level Loop

The agent replaces the operator’s execute loop with two levels: an outer loop that reasons about what to do next, and an inner loop that actually does it.

The outer loop asks what to do next. The inner loop doesn’t ask anything. (image by the author via Gemini)

The outer loop is where the agent lives. It looks at which clusters are pending, how much time is left, what the API health looks like, and picks the next action: fetch this cluster, pause, or abandon a low-priority one if time is running short. That decision happens once per cluster, not once per API call.

The inner loop is deterministic. Given a cluster, it fetches windows sequentially, handles 429 and 500 backoff, and shrinks or grows the window size based on what it sees. No reasoning required, just rules. When something breaks that pattern (sustained throttling, an error the rules don’t cover, time running critically short), the inner loop hands back to the outer loop and the agent decides what to do next.

That separation matters. Window fetching is mechanical: halve the window on throttling, retry on server errors, move on. There’s no judgment call there. But deciding which cluster to prioritize when you have 40 minutes left and 30 clusters pending, that’s a judgment call. That’s where the agent earns its place.

At the end of each run, the window size that worked gets written back to the state. The next day, the agent starts each cluster where yesterday left off, not back at 60 minutes to discover again.

Critical clusters get an additional hard guard: the agent cannot abandon them regardless of what it decides. That’s enforced in code, not in the prompt.

The One Architectural Decision That Matters

The agent makes decisions at the cluster level. Not the window level, not the API call level. The cluster level.

One LLM call per cluster. Everything below that line is deterministic. (image by the author)

I was tempted to put the LLM inside the tight loop: let it decide whether to retry this specific window, what size to use next, and whether this particular 429 was serious. That’s where the interesting stuff was happening.

But that’s also where I’d have spent 200 tokens per API call and still gotten worse decisions than a simple if-else. Window fetching is mechanical. The rules for it are known. There’s no new information an agent brings to “got a 429, should I wait 2 seconds?” The answer is yes. Always yes. You don’t need reasoning for that.

Where you do need reasoning: across clusters, across time, under constraint. Which of the 30 pending clusters should I fetch next, given that I have 45 minutes left and the API has been flaky for the last 10? Should I skip a low-priority staging cluster to preserve time for a prod cluster that hasn’t started yet? Those questions require a view of the global state, not the local state. That’s the outer loop.

The state dict that gets passed to the agent reflects this. It contains per-cluster status, coverage percentages, API health trends, and remaining time. It does not contain per-window details. The agent doesn’t know which specific 15-minute window just failed. It knows that a cluster is 60% covered and the API has been returning 429s at a 25% rate for the last 10 minutes. That’s enough to make a scheduling decision. It’s not enough to micromanage a retry.

That’s the design. Here’s what the first run actually looked like.

What the First Run Taught Me

The agent ran in production for the first time. 68 clusters. 4-hour window. It completed 2.

The logs showed why: the LLM call timed out once, at attempt 4, and the error handler re-raised immediately. No retry, no recovery. Just exit. The 66 clusters that hadn’t started yet never got touched.

I wrote the agent error handler assuming one failure meant something was seriously wrong. In production, one timeout means the API had a bad moment. The system should have shrugged and tried again.

The fix: track consecutive failures, not total failures. Retry up to three times with a 15-second sleep between attempts. Only abort after three in a row. One timeout is noise. Three in a row is a signal.

The second bug was quieter. The run summary reported 3,318 rows fetched. The Parquet file had 64,482. The metric was off by a factor of 20.

I was counting the wrong thing. The API returns workload objects, each with a list of destinations. I was counting workloads. The actual rows in the output are destinations, one per workload-destination pair. The number looked plausible enough that I almost didn’t check.

One line fix: sum(len(item.get(“destinations”, [])) for item in result[“rows”]).

Two bugs, both in the scaffolding around the agent logic rather than the agent logic itself. The outer loop, the inner loop, the state management: those worked. What failed was the error handling at the seams.

Where It Earned Its Place (And Where It Didn’t)

One run, one case where the agent caught something the alternatives wouldn’t have.

A sandbox cluster spiked traffic mid-run. Not dramatically: the kind of spike that degrades API response times on that cluster specifically while every other cluster stays clean. The global 429 health check didn’t fire: aggregate error rate was fine. A per-cluster rule would have required anticipating a sandbox spike. I hadn’t. Sandbox clusters are usually quiet. The agent saw the combination: low-priority tier, isolated degradation on a cluster that shouldn’t be generating load, 429 rate climbing. It deprioritized the cluster and shifted time to prod clusters that were still clean. The 500s never started.

No threshold would have caught that. The signal was in the combination, not any single metric.

The other case: window shrinking on 429s. When the API rate-limits a cluster, the inner loop halves the window size and retries. That’s an if-else. It was always going to be an if-else. I considered routing that decision through the agent (different window sizes for different cluster tiers, dynamic backoff based on history) and then I didn’t, because the rule is simple and the rule is right. Halve it. Wait. Try again. No reasoning required.