Planner–Executor 与 Reviewer 模式

mkdir l6-reviewer
cd l6-reviewer

"""用固定订单数据执行计划，再按业务条件审查草稿。"""
ORDER = {"id": "O-17", "paid": 120, "shipped": False}

def plan():
    # 计划列出能实际执行的资料读取步骤和预期产物。
    return ["read_order", "read_refund_rule", "write_draft"]

def execute(steps, requested=150):
    facts, evidence = {}, []
    for step in steps:
        if step == "read_order":
            facts["paid"] = ORDER["paid"]
            facts["shipped"] = ORDER["shipped"]
            evidence.append("order:O-17")
        elif step == "read_refund_rule":
            facts["rule"] = "未发货可退已付金额"
            evidence.append("policy:R-1")
        elif step == "write_draft":
            if "paid" not in facts or "rule" not in facts:
                raise ValueError("写草稿前缺少订单或规则")
        else:
            raise ValueError("未知计划步骤")
    if not steps or steps[-1] != "write_draft":
        raise ValueError("计划缺少草稿产物")
    return {"amount": requested, "evidence": evidence, "facts": facts}

def review(draft):
    problems = []
    # Reviewer根据数据与规则检查，不能凭角色名称宣告正确。
    if set(draft["evidence"]) != {"order:O-17", "policy:R-1"}:
        problems.append("missing_evidence")
    paid = ORDER["paid"]
    if (draft["facts"].get("paid") != paid
            or draft["facts"].get("shipped") != ORDER["shipped"]):
        problems.append("facts_mismatch")
    if type(draft["amount"]) is not int or paid is None or not 0 < draft["amount"] <= paid:
        problems.append("amount_exceeds_paid")
    if draft["facts"].get("shipped") is not False or ORDER["shipped"] is not False:
        problems.append("eligibility_unknown")
    return problems

def run(max_rounds=2):
    if type(max_rounds) is not int or max_rounds < 1:
        raise ValueError("审查轮数必须是正整数")
    requested, trace = 150, []
    for number in range(1, max_rounds + 1):
        draft = execute(plan(), requested)
        problems = review(draft)
        trace.append(f"第{number}轮：金额={draft['amount']}；问题={','.join(problems) or '无'}")
        if not problems:
            return "approved", draft, trace
        # 只根据已经查到的订单事实修正，不增加凭空的退款额度。
        requested = draft["facts"]["paid"]
    return "needs_human", draft, trace

def main():
    status, draft, trace = run()
    print("计划：" + " → ".join(plan()))
    for row in trace:
        print(row)
    print(f"处理结果：{status}；候选退款={draft['amount']}元")
    print("依据：" + "、".join(draft["evidence"]))
    print("只给一轮：" + run(1)[0])

if __name__ == "__main__":
    main()

ls
python -X utf8 reviewer_demo.py

计划：read_order → read_refund_rule → write_draft
第1轮：金额=150；问题=amount_exceeds_paid
第2轮：金额=120；问题=无
处理结果：approved；候选退款=120元
依据：order:O-17、policy:R-1
只给一轮：needs_human

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