AutoGen agent team data pipeline automation works well in notebooks but breaks down when you add retries, observability, and schema evolution. This guide walks through the patterns that hold up at scale — structured state, explicit contracts, and failure isolation — so your pipelines survive contact with production data.
Define the contract before you write the agent
Most teams start by prompting agents to “figure it out.” That works for prototypes. In production, every agent needs a typed interface: input schema, output schema, side effects, and retry policy. Write these as Pydantic models first, then build agents that honor them.
# contracts.py
from pydantic import BaseModel, Field
from typing import Literal
from datetime import datetime
class ExtractionInput(BaseModel):
source_uri: str
cursor: str | None = None # for incremental reads
batch_size: int = Field(default=1000, ge=1, le=10000)
class ExtractionOutput(BaseModel):
records: list[dict]
next_cursor: str | None
extracted_at: datetime = Field(default_factory=datetime.utcnow)
source_checksum: str
class TransformInput(BaseModel):
raw_records: list[dict]
schema_version: int
class TransformOutput(BaseModel):
clean_records: list[dict]
rejected: list[dict]
schema_version: intAgents receive and emit these models. Validation happens at the boundary — if an agent emits invalid JSON, the orchestrator catches it before downstream agents see garbage.
Structure the team around failure domains
A monolithic agent that extracts, transforms, and loads is a single point of failure. Split by failure domain: one agent per external system boundary, one per transformation stage, one per load target. Each gets its own retry budget, timeout, and circuit breaker.
# agents/extraction_agent.py
from autogen import AssistantAgent
from contracts import ExtractionInput, ExtractionOutput
import httpx
import json
class ExtractionAgent(AssistantAgent):
def __init__(self, name: str, api_base: str, **kwargs):
super().__init__(name, **kwargs)
self.client = httpx.AsyncClient(base_url=api_base, timeout=30.0)
self.register_reply(self._handle_extraction, trigger=ExtractionInput)
async def _handle_extraction(self, msg: ExtractionInput) -> ExtractionOutput:
params = {"cursor": msg.cursor, "limit": msg.batch_size}
resp = await self.client.get("/v1/records", params=params)
resp.raise_for_status()
data = resp.json()
return ExtractionOutput(
records=data["records"],
next_cursor=data.get("next_cursor"),
source_checksum=data["checksum"]
)The orchestrator sequences these agents. It doesn’t contain business logic — it only handles routing, retries, and compensation.
Use structured state, not conversation history
AutoGen’s default pattern passes messages between agents. For pipelines, that’s the wrong abstraction. Messages are for coordination; state is for data. Keep a mutable pipeline state object that agents read and write. The orchestrator persists it after each stage.
# state.py
from pydantic import BaseModel
from typing import Optional
from contracts import ExtractionOutput, TransformOutput
from datetime import datetime
class PipelineState(BaseModel):
run_id: str
started_at: datetime
extraction: Optional[ExtractionOutput] = None
transform: Optional[TransformOutput] = None
load_status: Optional[str] = None
errors: list[dict] = []
retry_count: dict[str, int] = {}
def can_retry(self, stage: str, max_retries: int = 3) -> bool:
return self.retry_count.get(stage, 0) < max_retries
def record_retry(self, stage: str):
self.retry_count[stage] = self.retry_count.get(stage, 0) + 1Agents receive the state, mutate their slice, and return it. The orchestrator decides what runs next based on state.extraction is not None and state.errors.
Build the orchestrator as a state machine
Don’t rely on LLM reasoning to sequence stages. Write an explicit state machine. It’s deterministic, testable, and debuggable.
# orchestrator.py
from enum import Enum
from state import PipelineState
from agents.extraction_agent import ExtractionAgent
from agents.transform_agent import TransformAgent
from agents.load_agent import LoadAgent
class Stage(str, Enum):
EXTRACT = "extract"
TRANSFORM = "transform"
LOAD = "load"
COMPLETE = "complete"
FAILED = "failed"
TRANSITIONS = {
Stage.EXTRACT: Stage.TRANSFORM,
Stage.TRANSFORM: Stage.LOAD,
Stage.LOAD: Stage.COMPLETE,
}
class PipelineOrchestrator:
def __init__(self, extraction_agent: ExtractionAgent,
transform_agent: TransformAgent, load_agent: LoadAgent):
self.agents = {
Stage.EXTRACT: extraction_agent,
Stage.TRANSFORM: transform_agent,
Stage.LOAD: load_agent,
}
async def run(self, state: PipelineState) -> PipelineState:
current = self._current_stage(state)
while current != Stage.COMPLETE and current != Stage.FAILED:
agent = self.agents[current]
try:
state = await agent.process(state)
state.errors = [e for e in state.errors if e["stage"] != current.value]
current = TRANSITIONS[current]
except Exception as e:
state.errors.append({
"stage": current.value,
"error": str(e),
"timestamp": datetime.utcnow().isoformat()
})
if state.can_retry(current.value):
state.record_retry(current.value)
await self._backoff(state.retry_count[current.value])
continue
current = Stage.FAILED
return state
def _current_stage(self, state: PipelineState) -> Stage:
if state.load_status == "success":
return Stage.COMPLETE
if state.transform is not None:
return Stage.LOAD
if state.extraction is not None:
return Stage.TRANSFORM
return Stage.EXTRACTThis pattern lets you resume from any stage after a crash. The state object is your checkpoint.
Handle schema evolution with versioned transforms
Data schemas change. Your transform agent must handle multiple schema versions simultaneously. Never mutate the input — produce a new output with an explicit version bump.
# agents/transform_agent.py
from contracts import TransformInput, TransformOutput
from typing import Callable
TRANSFORMERS: dict[int, Callable[[list[dict]], list[dict]]] = {}
def register_transformer(version: int):
def decorator(fn):
TRANSFORMERS[version] = fn
return fn
return decorator
@register_transformer(1)
def transform_v1(records: list[dict]) -> list[dict]:
# legacy mapping
return [{"id": r["_id"], "value": r["val"]} for r in records]
@register_transformer(2)
def transform_v2(records: list[dict]) -> list[dict]:
# current mapping with new fields
return [{
"id": r["id"],
"value": r["value"],
"metadata": r.get("meta", {})
} for r in records]
class TransformAgent(AssistantAgent):
async def process(self, state: PipelineState) -> PipelineState:
input_data = TransformInput(
raw_records=state.extraction.records,
schema_version=state.extraction.schema_version
)
transformer = TRANSFORMERS.get(input_data.schema_version)
if not transformer:
raise ValueError(f"No transformer for schema version {input_data.schema_version}")
clean = transformer(input_data.raw_records)
# simple validation example
rejected = [r for r in clean if "id" not in r]
clean = [r for r in clean if "id" in r]
state.transform = TransformOutput(
clean_records=clean,
rejected=rejected,
schema_version=input_data.schema_version
)
return stateWhen a new schema arrives, add a new transformer. Old versions keep working. Deploy the new transformer alongside the old one — no big bang migration.
Isolate external calls with circuit breakers
Agents that call APIs, databases, or file systems need circuit breakers. Without them, a slow downstream service cascades into thread exhaustion across your whole pipeline.
# resilience.py
import asyncio
from enum import Enum
from dataclasses import dataclass, field
from time import monotonic
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
@dataclass
class CircuitBreaker:
failure_threshold: int = 5
recovery_timeout: float = 30.0
state: CircuitState = CircuitState.CLOSED
failures: int = 0
last_failure_time: float = 0
_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
async def call(self, coro):
async with self._lock:
if self.state == CircuitState.OPEN:
if monotonic() - self.last_failure_time > self.recovery_timeout:
self.state = CircuitState.HALF_OPEN
else:
raise CircuitOpenError("Circuit breaker is open")
try:
result = await coro
async with self._lock:
self.failures = 0
self.state = CircuitState.CLOSED
return result
except Exception as e:
async with self._lock:
self.failures += 1
self.last_failure_time = monotonic()
if self.failures >= self.failure_threshold:
self.state = CircuitState.OPEN
raise
class CircuitOpenError(Exception):
passWrap every external call. The extraction agent gets a breaker for the source API. The load agent gets one for the warehouse. They fail fast and independently.
Add observability that actually helps debugging
Logs are not observability. Emit structured events at stage boundaries: stage started, stage completed, stage retried, stage failed. Include the run ID, stage name, record counts, and latency.
# observability.py
import structlog
from contextvars import ContextVar
from state import PipelineState
run_context: ContextVar[dict] = ContextVar("run_context", default={})
logger = structlog.get_logger()
def bind_run(state: PipelineState):
run_context.set({
"run_id": state.run_id,
"extraction_records": len(state.extraction.records) if state.extraction else 0,
"transform_records": len(state.transform.clean_records) if state.transform else 0,
})
async def log_stage_start(stage: str):
ctx = run_context.get()
logger.info("stage_started", stage=stage, **ctx)
async def log_stage_complete(stage: str, duration_ms: float, record_count: int):
ctx = run_context.get()
logger.info("stage_completed", stage=stage, duration_ms=duration_ms,
record_count=record_count, **ctx)
async def log_stage_retry(stage: str, attempt: int, error: str):
ctx = run_context.get()
logger.warning("stage_retry", stage=stage, attempt=attempt, error=error, **ctx)Wire these into the orchestrator. When a pipeline stalls at 2 AM, you can query by run_id and see exactly which stage retried how many times.
Common pitfalls
Letting agents decide control flow. The LLM should transform data, not decide whether to retry or skip a stage. Keep control flow in Python.
Passing full datasets in messages. AutoGen messages have size limits. Pass references (S3 keys, database cursors, temp table names) instead of inline data. The state object holds references; agents materialize only what they need.
Ignoring idempotency. Every stage must be idempotent or have a deduplication key. The extraction cursor is your idempotency key for reads. The load stage needs a primary key or upsert logic.
Treating schema as implicit. If the transform agent “figures out” the schema from examples, it will drift. Explicit versioned transformers prevent silent corruption.
Single-threaded orchestration. The orchestrator above runs stages sequentially. For independent partitions (by date, by tenant, by region), fan out: spawn multiple orchestrator instances each with a partition-specific state. Coordinate with a parent workflow that tracks partition completion.
Tradeoffs worth knowing
| Choice | Gain | Cost |
|---|---|---|
| Explicit state machine | Deterministic, resumable, testable | More boilerplate than “let the LLM decide” |
| Versioned transformers | Safe schema evolution | Maintain multiple transformer functions |
| Circuit breakers per agent | Failure isolation | More configuration, tuning thresholds |
| Structured events over logs | Queryable debugging | Requires log aggregation infrastructure |
| Partition-level parallelism | Throughput scales horizontally | Complexity in partition assignment and coordination |
Where n4n.ai fits
If your agents call multiple model providers — one for extraction reasoning, another for classification, a third for summarization — routing them through a single OpenAI-compatible endpoint reduces client complexity. n4n.ai forwards provider cache-control hints and honors routing directives, so you can pin specific stages to specific models without changing agent code.
What to build next
- Add a dead letter queue for rejected records. The transform agent writes rejects to a separate table/bucket with the error reason. A separate review pipeline handles them.
- Implement backfill support. The orchestrator should accept a date range or cursor range and run partitions in parallel with a concurrency limit.
- Add data quality assertions. After transform, run Great Expectations or custom checks. Fail the pipeline if null rates exceed thresholds.
- Build a replay tool. Given a run_id and a fixed transformer, re-run transform and load stages without re-extracting.
The patterns here — typed contracts, explicit state machines, versioned transforms, circuit breakers, structured observability — are the difference between a demo that works on clean data and a pipeline that runs unattended for months. Start with the contracts. Everything else follows.