Python 自动化：用 AI 替代条件判断逻辑

第一部分介绍了我每周都会用到的 25 个脚本——可靠的、生产级别的自动化构建块。这一部分讨论需要花更多时间才能解决的问题：当控制流涉及决策时，如何将它们链接在一起。

纯脚本链接很容易。运行 A，将输出管道输入到 B，完成。但真实的自动化管道会遇到决策点：这个文件应该被处理还是跳过？这个 API 响应是错误、重试还是成功？这个内容需要审查还是可以继续进行？

历史上你用 if/elif 树来处理这个问题。当决策逻辑清晰且稳定时，这种方法运行得很好。但当决策需要判断时就失效了——解析模糊的输出、分类不符合整洁类别的内容、处理 if 语句无法覆盖的长尾边界情况。

我最后采用的模式是：对管道的确定性部分使用 Cookbook 中的脚本（HTTP 调用、文件 I/O、调度、重试），而对判断调用使用一个小的 AI 提示。两者配合得很好，因为都没有试图做对方的工作。

AI 内循环管道的形状

下面是添加 AI 决策层之前和之后的三脚本管道。

之前——纯脚本链接：

# 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/

如果来自 API 的每个文件都是格式正确的 JSON，并且能够清晰地映射到 CSV，这就能正常工作。实际上：某些文件的编码格式不正确，某些文件有意外的模式更改，某些文件为空，某些是错误响应但看起来像成功的响应。脚本链在这些情况下会无声地失败，或者大声失败并停止整个管道。

之后——带有 AI 分类步骤：

# 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

AI 分类步骤不会替代 Cookbook 脚本——它在它们之间进行路由。这就是这个模式。

AI 步骤实际做什么

分类提示很小。这是我用于文件质量分类的实际提示：

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}
"""

三个输出。没有解释。AI 没有被要求思考——它被要求分类。这是管道内 AI 调用的合适范围。

单词输出——下游代码解析这个；模糊性会破坏管道

没有推理路径——你不需要它，它会减慢调用速度

枚举选项——如果你给模型开放式输出，你会得到开放式答案

这个模式是可以推广的。你可以用相同的形状处理：

对日志条目进行分类（ERROR / WARNING / INFO / NOISE）

给提取的数据质量评分（HIGH / MEDIUM / LOW / REJECT）

将客户消息路由到正确的队列（BILLING / SUPPORT / FEEDBACK / IGNORE）

决定爬取的页面是否包含你正在寻找的内容类型

prompt_runner.py 脚本

Cookbook 包含 prompt_runner.py——一个轻量级包装器，处理管道内 AI 调用的样板代码：API 客户端初始化、令牌预算、错误处理、输出解析和重试逻辑。

# 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

--prompt 标志从 prompts/ 目录中获取提示名称（你在那里存储项目的提示模板）。它加载模板，替换输入内容，调用 API，并将结构化输出写入输出文件。

批处理模式并行运行工作器（默认 4 个），在速率限制错误时自动退避。对于 100 个文件的批次，它通常在 30-60 秒内运行，具体取决于文件大小和 API 响应时间。

何时使用提示与何时写代码

我用于决定决策是否进入代码或提示的模式：

使用代码：

分类在提前时是完全可枚举的（status == 200）

输入是结构化的且确定性的（日期、数字、已知的枚举值）

你需要决策在没有 API 调用的情况下是可审计的

你以高量运行，其中 API 成本很重要

使用提示：

分类需要读取非结构化文本

输入格式以你无法预测的方式变化

边界情况存在，但你还没有映射过

自动正确得 80% 比手动正确得 0% 要好

我在实践中发现的边界是：代码处理确定性路径，提示处理长尾。一个设计良好的管道两者都有。

实际例子：自动化 PR 审查分类

我使用这个模式构建的最有用的管道：自动分类传入的 GitHub PR 审查请求。

问题：我在一个每天收到 15-30 个 PR 的代码库上工作。并非所有 PR 都需要深度审查——有些是文档更新、拼写错误修复或依赖更新。但要识别哪些可以自动批准，哪些需要注意，需要读取 diff。

涉及的脚本：

webhook_receiver.py——接收 GitHub webhook，将 PR 元数据 + diff 写入队列目录

prompt_runner.py——使用分类提示对每个 PR 进行分类

pipeline_router.py——根据分类进行路由

notification_sender.py——发送 Slack 通知和路由决策

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}
"""

# 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

它所做的：将手动分类从每天 20-30 分钟减少到 2-3 分钟。AUTO 分类的准确率几乎总是正确的（>95%）。URGENT 分类在 6 个月内有 0 个假阳性——它有意是保守的。

在管道中运行 AI 调用是有成本的。这对我来说可行的数学：

平均分类调用：~200 输入令牌 + 1 输出令牌

按当前 API 定价：< $0.001 每次调用

100 文件日批量：< $0.10/天

对于大多数自动化管道，AI 分类的成本由编写和维护它替代的 if/elif 树的时间成本主导——每当出现新的边界情况时，这个树都需要更新。

损益平衡大约是：如果你会花超过 10 分钟来编写代码来处理边界情况，API 调用会更便宜。

将第一部分脚本与 AI 层结合

作为 AI 管道组件最有用的脚本：

管道模式：

收集——http_client.py 或 webhook_receiver.py 或 file_watcher.py

分类——prompt_runner.py 使用你的分类提示

路由——pipeline_router.py 基于分类输出

处理——适合路由路径的任何脚本

通知——notification_sender.py 在完成或升级时

你不总是需要所有五个阶段。分类步骤是使其成为 AI 内循环而不是纯脚本链接的原因。

构建你的第一个 AI 分类步骤

如果你有一个当前使用手动分类或脆弱 if 树的管道，这是迁移路径：

识别判断调用。 管道中的哪个位置是人类或混乱的代码当前做出需要读取非结构化输入的决策？

识别判断调用。 管道中的哪个位置是人类或混乱的代码当前做出需要读取非结构化输入的决策？

定义输出枚举。 可能的分类是什么？2-4 个选择是理想的。超过 6 个，提示就变得不可靠。

定义输出枚举。 可能的分类是什么？2-4 个选择是理想的。超过 6 个，提示就变得不可靠。

写提示。 遵循模式：角色分配 → 带定义的枚举选择 → 边界情况规则 → 必需的输出格式（一个单词）。在部署前在 10-20 个代表性示例上测试它。

写提示。 遵循模式：角色分配 → 带定义的枚举选择 → 边界情况规则 → 必需的输出格式（一个单词）。在部署前在 10-20 个代表性示例上测试它。

将 prompt_runner.py 添加到管道。 用脚本调用替代判断调用。在第一周记录输出以验证准确性。

将 prompt_runner.py 添加到管道。 用脚本调用替代判断调用。在第一周记录输出以验证准确性。

调整规则。 看到真实输出后，精化提示的规则部分来处理你没有预期的边界情况。这比更新代码要快。

调整规则。 看到真实输出后，精化提示的规则部分来处理你没有预期的边界情况。这比更新代码要快。

上面的提示是起点——设计来为你的管道的特定枚举和规则修改。一般形状总是相同的：角色分配 → 带定义的枚举选择 → 边界情况规则 → 必需的单词输出格式。

一旦你有了一个工作的分类提示，将分类步骤添加到管道的边际成本就很低。每个新的决策点是另一个提示 + pipeline_router.py 中的另一个路由分支。

如果你在这些管道之上构建 agent SaaS

当我将这些管道扩展到生产服务时，我一直遇到的一件事：定价模型与架构一样重要。基于每个请求运行的管道与基于每个用户座位运行的管道有根本不同的成本结构。如果你在构建付费产品，值得提前思考。

我写了六个适用于 AI agent 产品的定价模型的完整分解——包括实际工作示例、成本计算器和我实际使用的决策框架：

Pricing Your Python AI Agent SaaS in 2026 → Six pricing models, Notion cost calculator, worked examples. $39 one-time.

如果你想在第 3 部分（故障模式、输出验证、人机检查点）发布时收到——列表在这里：

第 3 部分讲的是故障模式——具体来说，当你管道中的 AI 步骤产生错误分类而你没有捕捉到时会发生什么。它涵盖：输出验证、对于高风险决策的人机检查点、管道可观测性的日志模式，以及如何构建一个纠正循环来随时间改进提示准确性。

如果你现在正在构建 AI 内循环管道，请留下评论说你在做什么——我在从生产用例积极构建，读者问题推动第 3 部分的内容。(#discuss)

对于进一步的行动，你可能考虑屏蔽这个人和/或报告滥用