人工审批、暂停与恢复

更新 2026-10-05

正在读取学习记录…

人工审批把需要人判断的动作变成一个可暂停的步骤。Agent 提出180元退款建议后，任务进入 waiting_human（等待人工）；审批人查看订单、金额和理由，给出批准或拒绝结果。等待期间任务状态必须能保存，执行器也必须检查批准是否仍然有效。

批准绑定的是一份具体内容。金额从180元改成200元，旧审批针对的内容就已经过时；恢复时需要同时核对任务版本、金额、审批身份和有效期限。这样，人确认的内容才能与真正执行的动作一致。

正常路线是：提交提案→等待人工→获得当前内容的有效批准→执行退款→保存完成结果。下面故意在审批前修改一次金额，观察这条路线怎样重新等待有效批准。

运行旧审批和当前审批的两条路线

程序先提交180元退款，再把金额改成200元。随后按固定次序收到旧审批、无权限审批和当前版本有效审批，最后重复执行一次。这里没有等待输入框，固定审批数据让每次运行都能比较同一条路线。

version 是建议内容版本，修改金额后从1变成2。approved 必须是布尔值，None 的等待状态由任务本身表达；用字符串“yes”替代批准值会被拒绝。

审批绑定当前内容，等待、批准与业务完成是不同状态 — 图1：等待、批准与业务完成是不同状态。

保存完整程序并运行

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l4-hitl" -Force
cd "F:\学习\l4-hitl"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

新建 F:\学习\l4-hitl 文件夹，把下面完整代码保存为 hitl_demo.py，然后在PowerShell运行：

cd "F:\学习\l4-hitl"
ls
python -X utf8 hitl_demo.py

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l4-hitl" -Force
cd "F:\学习\l4-hitl"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

"""把审批绑定到退款任务的金额和版本。"""
from dataclasses import dataclass


@dataclass
class Request:
    amount: int = 180
    version: int = 1
    status: str = "waiting_human"
    receipt: str | None = None

    def revise(self, amount):
        if self.status != "waiting_human" or type(amount) is not int or amount <= 0:
            raise ValueError("当前状态或金额不允许修改")
        self.amount = amount
        self.version += 1

    def decide(self, *, role, version, amount, approved, now, expires):
        # 身份由服务端认证提供；输入中的 role 在实验中模拟认证结果。
        if role != "reviewer":
            return "forbidden"
        if type(approved) is not bool:
            return "invalid_decision"
        # 审批必须针对当前内容；修改金额后，旧审批不能复用。
        if version != self.version or amount != self.amount:
            return "stale_approval"
        if now >= expires:
            return "expired"
        if self.status != "waiting_human":
            return "already_decided"
        self.status = "approved" if approved else "rejected"
        return self.status

    def execute(self):
        if self.status == "completed":
            return self.receipt
        if self.status != "approved":
            raise ValueError("尚未获得有效批准")
        # 固定回执代表受审批保护的动作，重复执行返回同一回执。
        self.receipt = "refund-42"
        self.status = "completed"
        return self.receipt


def run_demo():
    request = Request()
    lines = [f"提交退款: 金额={request.amount}, 状态={request.status}"]
    request.revise(200)
    old = request.decide(role="reviewer", version=1, amount=180,
                         approved=True, now=5, expires=10)
    lines.append(f"修改金额后收到旧审批: {old}")
    denied = request.decide(role="viewer", version=2, amount=200,
                            approved=True, now=5, expires=10)
    lines.append(f"无审批权限: {denied}")
    result = request.decide(role="reviewer", version=2, amount=200,
                            approved=True, now=5, expires=10)
    lines.append(f"当前版本审批: {result}, 回执={request.execute()}")
    lines.append(f"重复执行: {request.execute()}")
    return lines


if __name__ == "__main__":
    print("\n".join(run_demo()))

完整固定输出如下：

提交退款: 金额=180, 状态=waiting_human
修改金额后收到旧审批: stale_approval
无审批权限: forbidden
当前版本审批: approved, 回执=refund-42
重复执行: refund-42

五行输出分别保护哪个边界

第二行的 stale_approval 表示审批对象已变更，任务仍在等待。第三行的 forbidden 是身份没有审批权限。第四行通过所有检查后才进入 approved，接着 execute 生成回执；最后一行返回同一回执，表示重复恢复不会生成第二个结果。

decide 返回拒绝原因，但不会顺便执行退款。状态从 waiting_human 变成 approved 后，执行节点才有资格运行；如果审批拒绝，则进入 rejected，execute 会抛异常。把审批决策与业务动作分开，有助于准确记录谁批准、何时执行。

暂停后要保存哪些东西

真实流程需要保存任务标识、版本、经校验的参数、等待状态以及审批请求标识。审批回调找到同一任务后，在事务里核对并写入决策。多个回调同时到达时，应通过条件更新保证只有等待中的对应版本能够变为已批准或已拒绝。

本例把角色参数当成认证结果来演示判断；接入网页时必须从已认证的会话或服务端权限系统取得角色，不能信任前端提交的“我是审批人”。审批内容还应包含工具、目标订单和作用范围。仅批准金额，而执行器能随意换订单，仍然会越过人的判断。

有效期与重复回调怎样处理

now >= expires 拒绝过期审批，边界时刻也已经过期。过期后可以生成新的审批请求，并让旧请求失效。等待人回复时不必持续占用工作者线程；保存等待状态后结束当前执行，回调或调度器负责唤醒。

已经批准、拒绝或完成的任务返回 already_decided。重复回调不会让已拒绝任务再次变为批准，修改内容也仅允许在等待状态进行。实际产品中若允许撤回批准，需要新增撤回状态及明确的执行竞争规则。

从内存例子接到退款工具

这里的固定回执便于观察审批检查，接付款接口后仍需带稳定业务键处理“已退款但回执丢失”。人工审批解决授权判断，业务幂等解决重复执行，这两种记录都需要保存。

配套测试直接调用未批准任务、过期审批、错误角色、字符串决策、旧版本、拒绝和重复执行。可以沿Temporal 源码查找等待外部信号的实现，把审批回调看作唤醒已经持久保存的工作流，而不是把整次流程从头跑一遍。

改一处输入，核对机制变化

只把当前版本审批调用中的 approved=True 改为 False，旧审批与无权限样本保持原值。运行 python -X utf8 hitl_demo.py，当前任务被拒绝，随后调用execute时出现“尚未获得有效批准”，不会产生回执。恢复True再运行，得到refund-42。异常定位的是拒绝后的执行动作。

运行配套测试

将下面测试保存为同一文件夹的 test_hitl_demo.py。它使用固定场景检查状态与业务结果，可以直接运行：

python -X utf8 -m unittest -v test_hitl_demo.py

完整测试：test_hitl_demo.py

import unittest
from hitl_demo import Request, run_demo


class ApprovalTests(unittest.TestCase):
    def decide(self, request, **changes):
        data = dict(role="reviewer", version=1, amount=180, approved=True, now=5, expires=10)
        data.update(changes)
        return request.decide(**data)

    def test_waiting_cannot_execute(self):
        with self.assertRaises(ValueError):
            Request().execute()

    def test_binding_identity_expiry_and_boolean(self):
        # 身份、有效期和内容绑定任一项不满足，都继续等待。
        request = Request()
        self.assertEqual(self.decide(request, role="viewer"), "forbidden")
        self.assertEqual(self.decide(request, approved="yes"), "invalid_decision")
        self.assertEqual(self.decide(request, now=10), "expired")
        request.revise(200)
        self.assertEqual(self.decide(request), "stale_approval")
        self.assertEqual(request.status, "waiting_human")

    def test_approval_and_duplicate(self):
        request = Request()
        self.assertEqual(self.decide(request), "approved")
        self.assertEqual(request.execute(), request.execute())
        self.assertEqual(self.decide(request), "already_decided")

    def test_rejection_stops_action(self):
        request = Request()
        self.assertEqual(self.decide(request, approved=False), "rejected")
        with self.assertRaises(ValueError):
            request.execute()


if __name__ == "__main__":
    unittest.main()

课后练习展开习题收起习题

正在读取本地记录…

不调用模型，不执行代码。简答、代码和场景题仅保存答案并展示自查标准，不生成评分、徽章或认证。输入内容不会自动保存，请点击提交；不要填写个人信息或密钥。

练习 1 / 4 · v1.0.1

选择题 · 先判断边界

关于“人工审批、暂停与恢复”，哪种做法更稳妥？

对应知识点：L4-hitl

选择一个答案A. 先明确输入、证据和停止条件B. 默认输出永远正确C. 用一次成功推断全部场景

练习 2 / 4 · v1.0.0

简答题 · 解释失败

写出一个失败输入并说明记录字段。

对应知识点：L4-hitl

你的回答

练习 3 / 4 · v1.0.0

代码题 · 边界测试

补充正常、边界、失败三组输入和预期结果。

对应知识点：L4-hitl

你的代码或伪代码

练习 4 / 4 · v1.0.0

场景题 · 结果未知

证据不足时如何设计状态和人工接管？

对应知识点：L4-hitl

你的回答