Python Automation Cookbook, Part 2: Using AI to make decisions in your automation pipelines

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.

---

The shape of an AI-in-the-loop pipeline

Here's a three-script pipeline before and after adding an AI decision layer.

Before — pure script chaining:

# Download files
python http_client.py --url $API_ENDPOINT --output raw/

# Process each file
for f in raw/*.json; do
    python csv_converter.py --input $f --output processed/
done

# Archive
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:

# Download files
python http_client.py --url $API_ENDPOINT --output raw/

# AI triage: classify each file before processing
python prompt_runner.py \
  --prompt "classify_file_quality" \
  --input-dir raw/ \
  --output triage_report.json

# Route based on triage result
python pipeline_router.py \
  --triage triage_report.json \
  --good-dest processing/ \
  --bad-dest review/ \
  --skip-dest archive/

# Process clean files only
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.

---

What the AI step actually does

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:

1. Single-word output — downstream code parses this; ambiguity breaks the pipeline
2. No reasoning trail — you don't need it and it slows the call
3. Enumerated choices — if you give the model open-ended output, you'll get open-ended answers

The pattern generalizes. You can use the same shape for:

• Classifying log entries (ERROR / WARNING / INFO / NOISE)
• Scoring extracted data quality (HIGH / MEDIUM / LOW / REJECT)
• Routing customer messages to the right queue (BILLING / SUPPORT / FEEDBACK / IGNORE)
• Deciding whether a scraped page contains the content type you're looking for

---

The prompt_runner.py script

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:

# Usage:
python prompt_runner.py \
  --prompt classify_file_quality \
  --input-file raw/data_2026_03_23.json \
  --output-file triage/data_2026_03_23.triage

# Batch mode (processes all files in a directory):
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.

---

When to use a prompt vs. when to write code

The pattern I use for deciding whether a decision goes in code or in a prompt:

Write code if:

• The classification is fully enumerable in advance (status == 200)
• The inputs are structured and deterministic (dates, numbers, known enum values)
• You need the decision to be auditable without an API call
• You're running at high volume where API costs matter

Use a prompt if:

• The classification requires reading unstructured text
• The input format varies in ways you can't predict
• Edge cases exist that you haven't mapped yet
• Getting it 80% right automatically is better than getting it 0% right manually

The boundary I've found in practice: code handles the deterministic path, prompts handle the long tail. A well-designed pipeline has both.

---

A real example: automated PR review triage

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:

1. webhook_receiver.py — receives GitHub webhook, writes PR metadata + diff to a queue directory
2. prompt_runner.py — classifies each PR using the triage prompt
3. pipeline_router.py — routes based on classification
4. notification_sender.py — sends Slack notification with routing decision

The 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:

# pipeline_router.py takes the triage output and acts on it
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.

---

The cost question

Running AI calls in a pipeline has a cost. The math that made this viable for me:

• Average classification call: ~200 input tokens + 1 output token
• At current API pricing: < $0.001 per call
• 100-file daily batch: < $0.10/day

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.

---

Combining Part 1 scripts with the AI layer

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:

1. Collect — http_client.py or webhook_receiver.py or file_watcher.py
2. Triage — prompt_runner.py with your classification prompt
3. Route — pipeline_router.py based on triage output
4. Process — whichever scripts fit the routed path
5. Notify — notification_sender.py on completion or escalation

You don't always need all five stages. The triage step is what makes it AI-in-the-loop instead of pure script chaining.

---

Building your first AI triage step

If you have a pipeline that currently uses manual classification or a brittle if tree, here's the migration path:

1. Identify the judgment call. Where in the pipeline does a human or messy code currently make a decision that requires reading unstructured input?

2. Define the output enum. What are the possible classifications? 2-4 choices is ideal. More than 6 and the prompt gets unreliable.

3. 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.

4. Add prompt_runner.py to the pipeline. Replace the judgment call with a script call. Log outputs for the first week to validate accuracy.

5. 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 prompt library

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.

---

If you're building agent SaaS on top of these pipelines

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:

Subscribe →

---

What part 3 covers

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)

---

Comments

[Vinicius Pereira]

The split between deterministic ops and judgment calls is the part most people get wrong, so good to see it stated this plainly. One thing I would add from running these in production: the single-word label hides its own uncertainty. A confident-wrong SKIP is byte-identical to a correct one, so silent misclassifications never surface until something downstream breaks.

Cheap fix that stays inside your token budget: have the model emit the label plus a coarse confidence (high/low is enough) and route on both. Anything low-confidence gets bumped to REVIEW regardless of what it picked. That turns invisible errors into a queue a human can watch, which is also the backbone of the correction loop you are saving for Part 3.

Small parsing note: "exactly one word" still misfires (a trailing period, or "I'd say REVIEW"), so validate against the enum and treat a non-member as REVIEW rather than crashing the router. Fail toward the human.