Wiring the Pipeline: Sequential, Branching, Parallel and Retry

Module 5, Unit 2 | Lesson 2 of 3

By the end of this lesson, you will be able to:

• Implement conditional branch routing that sends records to different queues based on their content (K12, S10)
• Add parallel processing to a pipeline stage using Python's concurrent.futures (K12, S10)
• Write exponential backoff retry logic for transient failures in external API calls (K12, S10, B6)
• Choose the right orchestration pattern for a given pipeline stage and explain the trade-offs (K11, K12)

In L2.1 you made the workflow-vs-agent decision for each stage. Now you need to wire those stages together in ways that reflect how real data actually behaves. Not every record takes the same path. Some stages can run in parallel. External API calls fail transiently and need to retry. These are orchestration patterns — the structural shapes that determine how work flows through your pipeline.

The four patterns

1. Sequential

The simplest pattern: each stage completes before the next starts. Every record passes through every stage in order. This is already implemented in your pipeline.py runner.

Use sequential when stages depend on each other's output, ordering matters, and throughput is not a bottleneck.

2. Conditional branch

Different records take different paths based on their content. A pii_detected record goes to the compliance handler; a hallucination record goes to the ops team; clean records go straight to output.

The rule itself is small enough to name and keep on its own. Add this to stages.py, just above the placeholder stage_route — with the FlaggedIssue import, if stages.py does not already have it from L2.1:

from schema import FlaggedIssue


def route_for(item: dict[str, Any]) -> str:
    """Which queue one classified record belongs in. First match wins."""
    record = item["record"]
    if item.get("route_override") == "human_review":
        return "human_review"
    if record.flagged_issue == FlaggedIssue.PII_DETECTED:
        return "compliance"
    if record.flagged_issue in (
        FlaggedIssue.HALLUCINATION,
        FlaggedIssue.COST_OVERRUN,
    ):
        return "ops_review"
    return "standard"

Four tests, read top to bottom, and the first one that matches wins. Because the rule lives in its own function you can check where a single record would go without running a pipeline — and when the routing policy changes, there is exactly one place to change it.

Now replace the placeholder stage_route in stages.py with the version below. It makes no decisions of its own: it asks route_for about each record, keeps the four queues on the state, and records what the split looked like.

def stage_route(state: PipelineState) -> PipelineState:
    """Stage 4: Route classified records to the correct handler."""
    started = time.time()
    queues: dict[str, list[dict[str, Any]]] = {
        "human_review": [],
        "compliance": [],
        "ops_review": [],
        "standard": [],
    }

    for item in state.classified_records:
        queues[route_for(item)].append(item)

    routed = [
        {"route": name, **item}
        for name in ("human_review", "compliance", "ops_review", "standard")
        for item in queues[name]
    ]
    state.routed_records = routed
    state.human_review_queue = queues["human_review"]
    state.compliance_queue = queues["compliance"]
    state.ops_queue = queues["ops_review"]
    state.standard_queue = queues["standard"]

    state.stage_results.append(StageResult(
        stage_name="route",
        records_in=len(state.classified_records),
        records_out=len(routed),
        records_failed=0,
        duration_seconds=round(time.time() - started, 3),
        metadata={name: len(queue) for name, queue in queues.items()},
    ))

    print(
        f"[route]    {len(queues['human_review'])} human_review | "
        f"{len(queues['compliance'])} compliance | "
        f"{len(queues['ops_review'])} ops_review | "
        f"{len(queues['standard'])} standard"
    )
    return state

Counting the queues into metadata means the run summary can show the shape of the workload without anyone re-reading the records.

Also add the four queue fields to PipelineState in contracts.py. Here is the whole dataclass, so there is no doubt about where they go:

@dataclass
class PipelineState:
    """Accumulated state passed through the full pipeline."""
    run_id: str
    started_at: datetime
    source_file: str
    raw_rows: list[dict[str, Any]] = field(default_factory=list)
    valid_records: list[ToolUsageRecord] = field(default_factory=list)
    classified_records: list[dict[str, Any]] = field(default_factory=list)
    routed_records: list[dict[str, Any]] = field(default_factory=list)
    stage_results: list[StageResult] = field(default_factory=list)
    halted: bool = False
    halt_reason: str = ""
    human_review_queue: list[dict[str, Any]] = field(default_factory=list)
    compliance_queue: list[dict[str, Any]] = field(default_factory=list)
    ops_queue: list[dict[str, Any]] = field(default_factory=list)
    standard_queue: list[dict[str, Any]] = field(default_factory=list)

Click to enlarge

🔑 Key term — Conditional branch: An orchestration pattern where records are divided into separate processing paths based on their content or classification. Each branch can have different logic, different handlers, and different output destinations. The key requirement is that every record ends up in exactly one branch.

3. Parallel processing

Some pipeline operations are independent of each other and can run concurrently. If a stage calls an external API for each API-bound record — which your stage_classify will do in Unit 3 — running those calls sequentially means each record waits for the one before it. Running them in parallel collapses that to roughly the time taken by the slowest small batch.

Python's concurrent.futures handles this cleanly. Add this utility function to stages.py:

import concurrent.futures


def run_in_parallel(fn, items: list, max_workers: int = 4) -> tuple[list, list]:
    """
    Run fn(item) for each item in parallel using a thread pool.
    Returns (results, errors) where errors are dicts with item and exception.
    
    Use for I/O-bound tasks: API calls, file reads, network requests.
    Do NOT use for CPU-bound tasks — use ProcessPoolExecutor for those.
    """
    results = [None] * len(items)
    errors = []

    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_index = {
            executor.submit(fn, item): index
            for index, item in enumerate(items)
        }
        for future in concurrent.futures.as_completed(future_to_index):
            index = future_to_index[future]
            try:
                results[index] = future.result()
            except Exception as e:
                errors.append({
                    "index": index,
                    "item": items[index],
                    "error": f"{type(e).__name__}: {e}",
                })

    # Successful results only, still in input order. Every dropped slot has a
    # matching entry in `errors` — the two lists together account for every item.
    return [r for r in results if r is not None], errors

You will use this directly in stage_classify when the real LLM calls are added in Unit 3. The helper preserves input order in results, even though individual calls finish at different times. That makes later debugging much easier because record order does not change just because one API call was slower than another.

Coach Cora

Look carefully at that return statement: an item whose call raised is removed from the results list. Thirty records go in, twenty-eight come out, and nothing crashes. That is a deliberate design choice, and it is only defensible because the failure is recorded in errors — which the calling stage must then put somewhere durable. A parallel helper that drops failures and returns nothing about them is how records vanish from a compliance pipeline without anyone noticing. Whenever a function of yours can return fewer things than it was given, ask where the difference is written down. In L3.5 you will build a debugging framework whose very first step is hunting exactly this kind of gap — in your own code.

Curious Cat

What is the difference between ThreadPoolExecutor and ProcessPoolExecutor? Threads share memory and are suited to I/O-bound tasks — waiting for a network response, reading a file, calling an API. Processes have separate memory and are better for CPU-bound tasks — heavy computation, image processing, number crunching. For calling LLM APIs, which involves waiting for a network response, threads are the right choice. A ProcessPoolExecutor here would add overhead without benefit.

4. Retry with exponential backoff

API calls fail. Rate limits, network blips, and temporary server errors are not edge cases — they are normal operating conditions in any production system. A pipeline that gives up after one failure is fragile. A pipeline that retries immediately in a tight loop will make the rate limit problem worse.

The standard solution is exponential backoff with jitter: wait longer after each failure, and add a small random delay so all retrying clients do not hit the server at exactly the same moment.

Add this utility to stages.py:

import time
import random


def with_retry(fn, max_attempts: int = 3, base_delay: float = 1.0):
    """
    Call fn(), retrying up to max_attempts times on failure.
    Uses exponential backoff with jitter between attempts.
    
    Attempt 1 fails → wait ~1s
    Attempt 2 fails → wait ~2s  
    Attempt 3 fails → raise the exception
    """
    for attempt in range(1, max_attempts + 1):
        try:
            return fn()
        except Exception as e:
            if attempt == max_attempts:
                raise
            delay = base_delay * (2 ** (attempt - 1)) + random.uniform(0, 0.5)
            print(
                f"  [retry] attempt {attempt} failed: {e}. "
                f"Retrying in {delay:.1f}s..."
            )
            time.sleep(delay)

In Unit 3, every LLM call will be wrapped with with_retry. For now, you can test it works by wrapping a function that fails on the first two attempts:

# Quick test — run this in a Python shell
from stages import with_retry

counter = {"n": 0}

def flaky_function():
    counter["n"] += 1
    if counter["n"] < 3:
        raise ConnectionError(f"Simulated failure {counter['n']}")
    return "success"

result = with_retry(flaky_function, max_attempts=3, base_delay=0.1)
print(result)  # "success" after two retries

🔑 Key term — Exponential backoff with jitter: A retry strategy where wait times grow exponentially (1s, 2s, 4s, ...) and a small random value is added to each delay. The jitter prevents all retrying clients from synchronising — if a rate limit caused one failure, synchronised retries will re-trigger it. Jitter staggers the retries so the load is spread across time.

Coach Cora

Exponential backoff with jitter is the industry standard — it is used by AWS, Google Cloud, and every major API client library. If you look at the OpenAI Python SDK, it retries automatically using exactly this pattern. Understanding it means you know why your pipeline behaves well under load, and you can tune the parameters (max attempts, base delay) sensibly when something goes wrong in production.

Choosing the right pattern

Most real pipelines combine all of these. Your stage_route uses conditional branching. Your stage_classify in Unit 3 will use parallel processing and retry. L2.3 adds the human approval checkpoint.

Build activity

1. Replace the placeholder stage_route in stages.py with the conditional branch implementation above.
2. Add the four queue fields (human_review_queue, compliance_queue, ops_queue, standard_queue) to PipelineState in contracts.py.
3. Add run_in_parallel and with_retry to stages.py as utility functions.
4. Run python pipeline.py — the [route] line should now show the split across the route queues. Check pipeline_output.json to confirm the stage metadata records the queue counts.
5. Run the with_retry test in a Python shell. With max_attempts=3 the flaky function should succeed on its third attempt after two retries. Then re-run it with max_attempts=2 and confirm it gives up and re-raises the ConnectionError — max_attempts counts total attempts, not retries after the first.

Challenge Chase

The parallel utility uses a fixed max_workers=4. In production, the right number of workers depends on your API's rate limit, network latency, and batch size. Research the OpenAI rate limit tiers and design a function that calculates a safe max_workers value given a requests-per-minute limit and an estimated response time in seconds. At what point does adding more workers stop improving throughput?

The code from this lesson

The pipeline with routing, parallelism and retry wired in. The code files are assembled from the code blocks of this lesson and the ones before it, so they match what you were asked to write; the dataset and a short README come with them.

Take it if you would rather read and run the code than type it out. Typing it is still the better way to learn it — this is here so a typo cannot cost you the lesson.

Checklist

• stage_route implements real conditional branching — records split into human_review, compliance, ops_review and standard queues
• PipelineState in contracts.py has the four queue fields populated after routing
• pipeline_output.json stage metadata shows the correct human_review, compliance, ops_review and standard counts
• Running python pipeline.py shows the correct queue split in the [route] terminal output
• run_in_parallel is added to stages.py and I understand when to use ThreadPoolExecutor vs ProcessPoolExecutor
• with_retry is added to stages.py and I have tested that it retries correctly and raises on final failure

Quick Check

KSB evidence focus

• K12 — Understands how to set up, interact with and generate APIs, databases and spreadsheets. Retry logic with exponential backoff is the correct way to call any production API. The with_retry utility you wrote is directly reusable for the OpenAI API calls in Unit 3, and for any other external service your pipeline integrates with.

• S10 — Can carry out data processing tasks, including processing/cleaning, data transformations, and feature engineering. The conditional branch routing is applied data processing: records are classified by content and directed to different handling queues. The metadata captured in StageResult documents the processing decision for every run — that is production-grade data engineering practice.

Up next: Lesson 3 completes the Unit 2 architecture with two production essentials: a connector interface that abstracts where your pipeline writes its output, and a human approval checkpoint that pauses processing when records need review before continuing.