结论:用yield实现中断式数据流、闭包封装可配置逻辑,组合成无全局状态、可暂停恢复、一次定义多处复用的业务算子;yield支持按需执行与内存友好分步输出,闭包固化配置避免参数重复传递与逻辑混杂。

直接说结论:用 yield 实现中断式数据流,用 闭包 封装可配置的业务逻辑,两者组合就能写出不依赖全局状态、随时暂停恢复、一次定义多处复用的业务算子。
yield 让算子具备“按需执行”能力
普通函数一调用就跑完全部逻辑,而生成器函数遇到 yield 就停住,把当前结果交出去,等下次被触发再接着往下走。这天然适合需要分阶段处理、中间可能要等待或判断的业务场景。
- 比如一个订单审核算子,不是一次性走完所有校验,而是每通过一项就 yield 一个状态,外部可以决定是继续、跳过还是终止
- yield 不返回完整列表,只返回当前步骤的输出,内存友好,尤其适合处理日志流、传感器数据、分页接口响应等持续产出的数据
- 不要手动调用 next() —— for 循环或 list() 自动驱动它,更安全也更符合直觉
闭包让算子自带“可配置上下文”
闭包就是外层函数返回内层函数,内层函数能记住外层函数的参数和变量。这意味着你可以提前把业务规则、阈值、路径、API token 等“环境信息”固化进去,生成一个专属算子实例。
- 例如:
def fraud_checker(threshold=0.8): def check(amount, score): return score > threshold and amount > 1000 return check—— 调用fraud_checker(0.9)就得到一个更严格的新算子 - 闭包避免了每次调用都传一堆重复参数,也防止逻辑和配置混在一起,改规则只需换闭包,不碰核心流程
- 注意别在闭包里引用后续会变的变量(比如循环中的 i),否则所有闭包实例可能共享最后一个值
组合起来:一个带中断能力的风控算子示例
下面是一个真实可用的组合写法:
def risk_pipeline(min_amount, max_retry):
# 闭包封装业务配置
def validate_order(order):
# yield 实现分步中断
yield "checking amount"
if order["amount"] yield "checking user history"
if order.get("retry_count", 0) > max_retry:
yield "exceed max retry"
return
yield "approved"
return {"status": "success", "order_id": order["id"]}
return validate_order
使用:生成一个具体算子
checker = risk_pipeline(min_amount=50, max_retry=2)
对单个订单执行,可随时中断
for step in checker({"id": "ORD-001", "amount": 30, "retry_count": 0}): print(step) # 输出 checking amount → amount too low
这个算子既可嵌入 for 循环批量处理,也可单步调试;配置(min_amount/max_retry)由闭包固化,行为(每步校验什么)由 yield 定义;不用类、不改全局、不建对象,干净利落。
进阶提示:配合 send() 实现双向通信
如果外部需要在中断点动态注入新信息(比如人工复核后返回审批意见),可以用 send() 向生成器传值:
- yield 变成表达式:
decision = yield "awaiting review" - 调用方用
gen.send("approved")把结果送回去,生成器继续执行 - 这种“暂停-决策-继续”模式特别适合工作流引擎、审批链、对话机器人等强交互场景











