Part 1 covered the 25 scripts I reach for every week — reliable, production-ready automation building blocks. This part covers the part that takes longer to figure out: how to chain them together when the control flow involves a decision.
Pure script-chaining is easy. Run A, pipe output to B, done. But real automation pipelines hit decision points: should this file be processed or skipped? Is this API response an error, a retry, or a success? Does this content need review or can it proceed?
Historically you handled this with if/elif
trees. This works fine when the decision logic is clear and stable. It falls apart when the decision requires judgment — parsing ambiguous output, classifying content that doesn't fit neat categories, handling the long tail of edge cases that if
statements don't cover.
The pattern I've landed on: use the scripts in the Cookbook for the deterministic parts of the pipeline (HTTP calls, file I/O, scheduling, retries), and use a small AI prompt for the judgment calls. The two fit together cleanly because neither is trying to do the other's job.
Here's a three-script pipeline before and after adding an AI decision layer.
Before — pure script chaining:
python http_client.py --url $API_ENDPOINT --output raw/
for f in raw/*.json; do
python csv_converter.py --input $f --output processed/
done
python file_archiver.py --source processed/ --destination archive/
This works if every file from the API is well-formed JSON that maps cleanly to CSV. In practice: some files have malformed encoding, some have unexpected schema changes, some are empty, some are error responses that look like successful ones. The script chain fails silently on these, or fails loudly and stops the whole pipeline.
After — with an AI triage step:
python http_client.py --url $API_ENDPOINT --output raw/
python prompt_runner.py \
--prompt "classify_file_quality" \
--input-dir raw/ \
--output triage_report.json
python pipeline_router.py \
--triage triage_report.json \
--good-dest processing/ \
--bad-dest review/ \
--skip-dest archive/
for f in processing/*.json; do
python csv_converter.py --input $f --output processed/
done
The AI triage step doesn't replace the Cookbook scripts — it routes between them. That's the pattern.
The triage prompt is small. Here's the actual prompt I use for file quality classification:
CLASSIFY_FILE_QUALITY = """You are a data quality classifier. Given a JSON file's contents, classify it as one of:
- PROCESS: well-formed, expected schema, ready for conversion
- REVIEW: unusual structure, possible schema change, needs human look
- SKIP: empty, error response, or duplicate of a file already processed
Respond with exactly one word: PROCESS, REVIEW, or SKIP.
Do not explain. Do not add caveats.
File contents:
{file_contents}
"""
Three outputs. No explanation. The AI is not being asked to think — it's being asked to classify. This is the right scope for an in-pipeline AI call.
The key constraints:
The pattern generalizes. You can use the same shape for:
The Cookbook includes prompt_runner.py
— a lightweight wrapper that handles the boilerplate for in-pipeline AI calls: API client initialization, token budgeting, error handling, output parsing, and retry logic.
Here's how it works:
python prompt_runner.py \
--prompt classify_file_quality \
--input-file raw/data_2026_03_23.json \
--output-file triage/data_2026_03_23.triage
python prompt_runner.py \
--prompt classify_file_quality \
--input-dir raw/ \
--output-dir triage/ \
--workers 4
The --prompt
flag takes a prompt name from prompts/
directory (where you store your project's prompt templates). It loads the template, substitutes the input content, calls the API, and writes structured output to the output file.
The batch mode runs workers in parallel (default 4) with automatic backoff on rate limit errors. For a 100-file batch, it typically runs in 30-60 seconds depending on file size and API response time.
The pattern I use for deciding whether a decision goes in code or in a prompt:
Write code if:
status == 200
)Use a prompt if:
The boundary I've found in practice: code handles the deterministic path, prompts handle the long tail. A well-designed pipeline has both.
The most useful pipeline I've built using this pattern: automated triage of incoming GitHub PR review requests.
The problem: I work on a codebase that gets 15-30 PRs per day. Not all of them need deep review — some are docs updates, typo fixes, or dependency bumps. But identifying which ones can be auto-approved vs. which ones need attention requires reading the diff.
The scripts involved:
webhook_receiver.py
— receives GitHub webhook, writes PR metadata + diff to a queue directoryprompt_runner.py
— classifies each PR using the triage promptpipeline_router.py
— routes based on classificationnotification_sender.py
— sends Slack notification with routing decisionThe triage prompt:
TRIAGE_PR = """You are a senior developer reviewing pull requests for triage priority.
Given a pull request diff and metadata, classify it as:
- AUTO: safe to auto-approve (docs, formatting, dependency updates with no API changes, comment additions)
- REVIEW: needs human review (logic changes, new features, API changes, security-sensitive code)
- URGENT: needs immediate review (rollback, hotfix, security patch)
Rules:
- If you can't determine the scope from the diff, classify as REVIEW
- If any test files are modified, classify as at least REVIEW
- If any auth, security, or credential-handling code is touched, classify as URGENT
Respond with exactly one word: AUTO, REVIEW, or URGENT.
PR title: {title}
Author: {author}
Files changed: {file_count}
Diff summary:
{diff}
"""
The routing logic:
AUTO → add "auto-approved" label, merge if checks pass
REVIEW → add to review queue, notify team on next batch
URGENT → page on-call immediately via notification_sender.py
What it does: Reduces manual triage from 20-30 minutes per day to 2-3 minutes. AUTO classifications are almost always right (>95%). URGENT classifications have had 0 false negatives in 6 months — it's conservative on purpose.
Running AI calls in a pipeline has a cost. The math that made this viable for me:
For most automation pipelines, the cost of AI triage is dominated by the time cost of writing and maintaining the if/elif
tree it replaces — which needs to be updated every time a new edge case appears.
The break-even is roughly: if you'd spend more than 10 minutes writing code to handle the edge case, the API call is cheaper.
The scripts most useful as AI pipeline components:
| Script | Role in AI pipeline |
|---|---|
http_client.py |
|
| Fetch data for AI processing; handles retries on API rate limits | |
file_watcher.py |
|
| Trigger AI triage when new files appear | |
webhook_receiver.py |
|
| Accept external events that feed into triage | |
queue_processor.py |
|
| Process AI-classified items from a queue | |
notification_sender.py |
|
| Send routing decisions and escalations | |
retry_runner.py |
|
| Wrap AI calls themselves — handles transient API errors | |
cron_wrapper.py |
|
| Schedule batch AI processing jobs |
The pipeline pattern:
http_client.py
or webhook_receiver.py
or file_watcher.py
prompt_runner.py
with your classification promptpipeline_router.py
based on triage outputnotification_sender.py
on completion or escalationYou don't always need all five stages. The triage step is what makes it AI-in-the-loop instead of pure script chaining.
If you have a pipeline that currently uses manual classification or a brittle if
tree, here's the migration path:
Identify the judgment call. Where in the pipeline does a human or messy code currently make a decision that requires reading unstructured input?
Define the output enum. What are the possible classifications? 2-4 choices is ideal. More than 6 and the prompt gets unreliable.
Write the prompt. Follow the pattern: role assignment → enumerated choices with definitions → rules for edge cases → required output format (one word). Test it on 10-20 representative examples before deploying.
Add prompt_runner.py to the pipeline. Replace the judgment call with a script call. Log outputs for the first week to validate accuracy.
Tune the rules. After seeing real output, refine the rules section of the prompt to handle edge cases you didn't anticipate. This is faster than updating code.
The prompts above are starting points — designed to be modified for your pipeline's specific enum and rules. The general shape is always the same: role assignment → enumerated choices with definitions → rules for edge cases → required one-word output format.
Once you have a working triage prompt, the marginal cost of adding classification steps to your pipeline is low. Each new decision point is another prompt + another routing branch in pipeline_router.py
.
One thing I keep running into as I expand these pipelines into production services: the pricing model matters as much as the architecture. Pipelines that run on a per-request basis have fundamentally different cost structures than pipelines that run on a per-user seat. If you're building toward a paid product, worth thinking about early.
I wrote a full breakdown of the six pricing models that work for AI agent products — including worked examples, a cost calculator, and the decision framework I actually use:
Pricing Your Python AI Agent SaaS in 2026 →
Six pricing models, Notion cost calculator, worked examples. $39 one-time.
And if you want Part 3 (failure modes, output validation, human-in-the-loop checkpoints) in your inbox when it drops — the list is here:
Part 3 is about failure modes — specifically, what happens when an AI step in your pipeline produces the wrong classification and you don't catch it. It covers: output validation, human-in-the-loop checkpoints for high-stakes decisions, logging patterns for pipeline observability, and how to build a correction loop that improves prompt accuracy over time.
If you're building AI-in-the-loop pipelines right now, leave a comment with what you're working on — I'm actively building from production use cases, and reader questions drive what goes in Part 3. (#discuss)