代码里的确定性金融流程：改造前后与选库

步骤 1改造前：把整个月的数据交给模型，让它做对账。

团队最先搭出来的往往是这个版本，因为它在演示时能跑通：把两本账都放进提示词，要求给出匹配结果、未匹配的行和银行余额调节表。

"""BEFORE: hand the whole reconciliation to the model, numbers and all. Do not ship this."""
import json, subprocess, sys

PROMPT = """You are an accountant. Reconcile this September 2026 bank statement against the cash book.
Match each bank line to its ledger line, list everything unmatched, and prepare the bank
reconciliation: adjusted bank balance, adjusted book balance, and the difference.
Reply with JSON only:
{"matches": [["B..","L.."]], "unmatched_bank": ["B.."], "unmatched_ledger": ["L.."],
 "adjusted_bank_balance": "0.00", "adjusted_book_balance": "0.00", "difference": "0.00",
 "explanation": "..."}

DATA:
"""

它一共跑了 28 次，覆盖两种规模（每边 50 行和 406 行）和两种变体：一种导出数据里带期末余额，另一种需要模型自己把每一行加总。对照已知答案评分：

runs                                   28   Haiku 5.5, Sonnet 5.5, Opus 5.5
returned a difference                  25   every one of them: "0.00"
correct and complete                   21
  Sonnet, 406 lines, no totals          1   both closing balances +2,372.53; book total
                                            "derived from the bank total"
  Haiku, 50 and 406 lines               2   adjusted balances 1,245.30 below the correct
                                            figure; the 9.00 fix described, not applied
  Haiku, 406 lines                      1   listed 3 of 403 matched pairs
declined to compute                     3   Haiku, 406 lines, no totals: all nulls
keyed-error pair called a "match"   9 of 25 the same books, classified two ways

这些模型做这件事其实很在行。Opus 和 Sonnet 在大多数运行中都能手工把 406 行正确加总，而且每一次给出数字的运行都找到了那笔 9.00 的录入错误。问题出在输出没有告诉你的东西上。25 份答案都说差额是 0.00，其中包括两个期末余额都错了 2,372.53 的那一份——它的说明里承认，账簿总额是从银行一侧倒推出来的。算出来的平衡和声称出来的平衡，在 JSON 里看起来一模一样。在同一实验的一次预跑中，还有一次 Sonnet 运行返回了一个根本不存在的分类账 ID L390X，只在文字说明里才自我更正。

审阅者如果想知道哪些运行是错的，就只能把对账重做一遍。所以对账从一开始就应该是代码，模型只负责真正需要“读懂”的那部分。

步骤 2金额跨越任何边界时都以字符串传递，在内部以整数分存储。

MONEY = re.compile(r"-?\d+\.\d{2}")

def cents(s: str) -> int:
    """'-1254.30' -> -125430. Money crosses every boundary as a string, never a float."""
    if not isinstance(s, str) or not MONEY.fullmatch(s):
        raise ValueError(f"money must be a string like '-1254.30', got {s!r}")
    return int(s.replace(".", ""))

def fmt(c: int) -> str:
    return f"{'-' if c < 0 else ''}{abs(c) // 100}.{abs(c) % 100:02d}"

这个解析器是刻意写得很严格的，而且在本页中，还没有任何模型参与，它就已经证明了自己的价值。数据生成器的第一个版本把一个 Decimal 除以 100，写出了 7889.4，被 cents() 拒绝了。换成浮点数流水线，这个值会被悄无声息地带下去。常见的陷阱值得烂熟于心：0.1 + 0.2 等于 0.30000000000000004，round(2.675, 2) 等于 2.67，而 Decimal("100.00") / 3 量化后再乘回去得到 99.99——所以拆分金额时，除非把余数分配出去，否则就会丢一分钱。

有一个陷阱恰好就在模型边界上。把 Pydantic 字段声明为 Decimal 并不能保护你：如果模型发来的是一个 JSON 数字，JSON 解析器会先把它变成浮点数——12345678901234567.89 到手时成了 12345678901234568。在任何由模型填写的 schema 中，都要把金额定义为带正则约束的字符串，再在代码里转换。

步骤 3代码匹配一切能证明的，其余的拒绝猜测。

def exact_pass(bank, ledger, window=3):
    """Same amount to the cent, dates within the window, and unique in BOTH directions."""
    near = lambda b, l: b.cents == l.cents and abs((b.day - l.day).days) <= window
    pairs = []
    for b in sorted(bank, key=lambda x: x.id):
        ls = [l for l in ledger if near(b, l)]
        if len(ls) == 1 and sum(near(x, ls[0]) for x in bank) == 1:
            pairs.append((b.id, ls[0].id))
    return pairs

只有当金额精确到分相等、日期相差不超过三天，且任何一方都没有第二个候选时，配对才会被接受。真正关键的是最后一个条件。同一天的两笔 450.00 电汇各自都能看到两条可能的分类账记录，因此代码让这四行全部保持未匹配，而不是按位置把它们配上——规则引擎正是这样悄无声息地错误过账的。在 406 行的这个月里，这一轮匹配定下了 400 对，剩下 12 行。

步骤 4模型为剩下的行提出配对建议，由代码判定每条建议的性质。

class Proposal(BaseModel):
    bank_id: str
    ledger_id: str
    reason: str = Field(max_length=300)

class Proposals(BaseModel):
    proposals: list[Proposal]

LLM = Callable[[str], str]          # prompt in, JSON text out

PROPOSE = """These bank and cash-book lines did not match automatically. Propose bank->ledger
pairs that are the same transaction. Do not compute balances. Reply with JSON only:
{"proposals": [{"bank_id": "B..", "ledger_id": "L..", "reason": "..."}]}
"""

def propose(llm: LLM, bank, ledger) -> list[Proposal]:
    rows = lambda ls: [{"id": l.id, "date": l.day.isoformat(), "amount": fmt(l.cents), "text": l.text} for l in ls]
    text = llm(PROPOSE + json.dumps({"bank": rows(bank), "ledger": rows(ledger)}, sort_keys=True))
    return Proposals.model_validate_json(re.search(r"\{.*\}", text, re.S).group(0)).proposals

def dispose(p: Proposal, open_b: dict, open_l: dict, max_days=14) -> tuple[str, str]:
    b, l = open_b.get(p.bank_id), open_l.get(p.ledger_id)
    if b is None or l is None:
        return "rejected", "unknown or already-used id"
    if b.cents != l.cents:
        return "break", f"amounts differ by {fmt(b.cents - l.cents)}"
    if abs((b.day - l.day).days) > max_days:
        return "rejected", "dates too far apart"
    return "review", p.reason                      # a human confirms before it counts

模型看到的是 12 行而不是 812 行，它唯一可以返回的，是一组带理由的 ID 配对。它不需要计算任何东西。随后，代码给每条建议三种结果之一：未知或已被使用的 ID 被拒绝；金额不同的配对成为一个差异项，并附上精确差额；金额相等的配对作为待确认的匹配交给人工。模型说的任何话都不会被过账。在相同的剩余行上，Haiku 跑三次、Sonnet 跑三次：

                 Haiku 5.5 (x3)   Sonnet 5.5 (x3)
exact, by code        400              400
to review               3                3     both wires by payee; the 9-day Proseware payment
breaks                  1                1     B269/L005, difference -9.00
rejected                0                0
adjusted bank    289,941.09       289,941.09
adjusted book    289,941.09       289,941.09
can sign off          no               no     until a person confirms 3 pairs and the -9.00 entry

六次运行的结果完全相同。理由的措辞每次都不一样，但这些措辞都不属于账簿的一部分。较便宜的模型和较大的模型做得一样好，因为它被要求做的事情（读两段描述，判断它们是不是同一笔付款）是阅读任务，而不是计算。

步骤 5按规则对剩余项分类，计算调节表，并为签核设置门槛。

def classify(side: str, l: Line, period_end: date) -> str:
    if side == "ledger" and l.cents > 0 and (period_end - l.day).days <= 3:
        return "deposit_in_transit"
    if side == "ledger" and l.cents < 0 and l.text.lower().startswith("cheque"):
        return "outstanding_cheque"
    if side == "bank" and l.cents < 0 and re.search(r"\b(FEE|CHARGE)\b", l.text):
        return "bank_charge"
    if side == "bank" and l.cents > 0 and "INTEREST" in l.text:
        return "bank_interest"
    return "unexplained"

def reconcile(month: dict, llm: LLM | None, period_end=date(2026, 9, 30)) -> Report:
    bank, ledger = load(month["bank"]), load(month["ledger"])
    r = Report()
    # Every total is recomputed from the lines and tied to the statement's own figure.
    bank_close = cents(month["opening_balance_bank"]) + sum(l.cents for l in bank)
    book_close = cents(month["opening_balance_book"]) + sum(l.cents for l in ledger)
    if "closing_balance_bank" in month:
        assert bank_close == cents(month["closing_balance_bank"]), "statement does not foot"

    for b, l in exact_pass(bank, ledger):
        r.matched.append((b, l, "exact"))
    used_b = {m[0] for m in r.matched}; used_l = {m[1] for m in r.matched}
    open_b = {l.id: l for l in bank if l.id not in used_b}
    open_l = {l.id: l for l in ledger if l.id not in used_l}

    if llm and open_b and open_l:
        for p in propose(llm, sorted(open_b.values(), key=lambda x: x.id), sorted(open_l.values(), key=lambda x: x.id)):
            verdict, why = dispose(p, open_b, open_l)
            if verdict == "rejected":
                r.rejected.append((p.bank_id, p.ledger_id, why)); continue
            b, l = open_b.pop(p.bank_id), open_l.pop(p.ledger_id)
            if verdict == "break":
                r.breaks.append((b.id, l.id, b.cents - l.cents))
            else:
                r.review.append((b.id, l.id, why))

    r.items = ([("bank", l.id, classify("bank", l, period_end), l.cents) for l in open_b.values()] +
               [("ledger", l.id, classify("ledger", l, period_end), l.cents) for l in open_l.values()])
    timing = sum(c for side, _, cat, c in r.items if cat in ("deposit_in_transit", "outstanding_cheque"))
    bank_side = sum(c for side, _, cat, c in r.items if cat in ("bank_charge", "bank_interest"))
    correction = sum(d for _, _, d in r.breaks)
    r.adjusted_bank = bank_close + timing
    r.adjusted_book = book_close + bank_side + correction
    r.can_sign_off = (r.adjusted_bank == r.adjusted_book and not r.review and not r.breaks
                      and all(cat != "unexplained" for *_, cat, _ in r.items))
    check_invariants(bank, ledger, r)
    return r

差额之所以不能当作检查，原因在于算术本身。当两本账从同一期初余额出发时，任何一份把每条未匹配行都作为本方调节项列示的对账，都会天然平衡：调整后账面余额减去调整后银行余额，等于各匹配对差额之和，也就是零。STEP 1 中那两次结果少了 1,245.30 的 Haiku 运行正是这样做的，而且都报告了 0.00。所以在这里真正起作用的是另外几道控制：

• 对账单能够轧平。期末余额由各行重新计算，并与银行自己给出的数字做断言比对，因此漏掉一行就会让整次运行失败。
• 每条未匹配行都有一个来自规则的类别。在途存款是最后三天内的一条正数分类账记录，未兑现支票是一条属于支票的负数分类账记录。任何规则都认领不了的行就是 unexplained，单凭这一点就会阻止签核。
• 错误要做成分录，而不是调整。那笔 9.00 的差异项会产生一条账簿需要的更正分录，签核要等到它过账并获批之后。

谁有权审批是一个策略问题，用策略来表达比用 if 语句更好。在 Cedar（cedarpy 4.12.2）中，permit(principal, action == Action::"approve", resource) when { resource.maker != principal }; 在对账服务试图审批自己生成的更正分录时返回 Deny，而对一位具名的财务主管返回 Allow。

def check_invariants(bank, ledger, r: Report) -> None:
    seen_b = [m[0] for m in r.matched + r.review + r.breaks] + [i[1] for i in r.items if i[0] == "bank"]
    seen_l = [m[1] for m in r.matched + r.review + r.breaks] + [i[1] for i in r.items if i[0] == "ledger"]
    assert sorted(seen_b) == sorted(l.id for l in bank), "every bank line accounted for exactly once"
    assert sorted(seen_l) == sorted(l.id for l in ledger), "every ledger line accounted for exactly once"
    amt = {l.id: l.cents for l in bank + ledger}
    assert all(amt[b] == amt[l] for b, l, _ in r.matched + r.review), "a match is equal to the cent"

步骤 6记录模型的回复，让审计师的重跑就是同一次运行。

审计师会要求你在 12 月重新执行 9 月的对账。靠再问一次模型是做不到的，因为没有任何一家托管服务商保证两次给出相同的回复（边界篇列出了各家分别提供什么）。因此，可复现性来自记录，而不是采样参数。

class Recorded:
    """Same model + same prompt -> the stored reply, so an auditor's rerun is byte-identical."""
    def __init__(self, llm: LLM | None, model: str, path: str):
        self.llm, self.model, self.path = llm, model, path
        try: self.store = json.load(open(path))
        except FileNotFoundError: self.store = {}
        self.calls = 0
    def __call__(self, prompt: str) -> str:
        key = hashlib.sha256(f"{self.model}\0{prompt}".encode()).hexdigest()
        if key not in self.store:
            if self.llm is None:
                raise LookupError(f"no recorded reply for {key[:12]}; replay cannot call a model")
            self.store[key] = self.llm(prompt); self.calls += 1
            json.dump(self.store, open(self.path, "w"), indent=1, sort_keys=True)
        return self.store[key]

键涵盖模型名称和完整的提示词，而提示词由排序后的行构建，所以同一个月总是生成同一个键。不挂接模型的回放会返回存储的回复，调用次数为零，并产出完全相同的报告。找不到记录的回放会直接报错，而不是悄悄去调用模型。把这个存储和该期间的工作底稿放在一起保存。它是证据，证明财务主管批准的那些建议正是模型当时给出的。

步骤 7模型撰写文字时，其中的每个数字都必须是代码算出来的。

对于 CFO 要读的说明文字，模型确实很有用。它可以根据报告来写这段说明，前提是你要检查它写了什么。

def figures(report) -> set[str]:
    """The only numbers the commentary may state, as normalised strings."""
    vals = [report.adjusted_bank, report.adjusted_book] + [c for *_, c in report.items] + [d for *_, d in report.breaks]
    out = {fmt(abs(v)) for v in vals}
    out |= {str(len(report.matched)), str(len(report.review)), str(len(report.breaks)), "0.00"}
    return out

def unsupported(text: str, allowed: set[str]) -> list[str]:
    bad = []
    for raw in NUM.findall(text):
        n = raw.strip("()$-").replace(",", "")
        if re.fullmatch(r"20\d\d", n):        # a year, not an amount
            continue
        if "." in n:
            n = str(Decimal(n).quantize(Decimal("0.01"))) if len(n.split(".")[1]) <= 2 else n
        if n not in allowed:
            bad.append(raw)

十段说明（五段 Haiku、五段 Sonnet），每段都只根据报告中的数字撰写，没有出现任何无依据的数字。当模型是被“给予”数字而不是被“索要”数字时，这正是预期的结果，而这道检查让你可以放心依赖它。在测试套件中，同一个函数能在一句本应写 9.00 和 3 的句子里，标记出埋入的“90.00”和“12 items”。检查中的金额以规范化的 Decimal 字符串比较，从不使用浮点数。

步骤 8测试：把设计的承诺写下来。

def test_model_proposals_are_disposed_by_code():
    (b1, l1), (b2, l2) = NAMED["wires"]
    r = reconcile(MONTH, fake({"proposals": [
        {"bank_id": b1, "ledger_id": l1, "reason": "same payee"},
        {"bank_id": TP_B, "ledger_id": TP_L, "reason": "same vendor"},      # amounts differ by 9.00
        {"bank_id": b2, "ledger_id": "L390X", "reason": "typo"},          # no such ledger line
    ]}))
    assert [x[:2] for x in r.review] == [(b1, l1)]
    assert r.breaks == [(TP_B, TP_L, -900)]
    assert r.rejected == [(b2, "L390X", "unknown or already-used id")]
    assert not r.can_sign_off

def test_dropping_a_line_fails_the_tie_to_the_statement():
    m = copy.deepcopy(MONTH); m["bank"].pop()
    with pytest.raises(AssertionError, match="does not foot"):
        reconcile(m, None)

def test_input_order_does_not_change_the_result():
    m = copy.deepcopy(MONTH); random.Random(7).shuffle(m["bank"]); random.Random(8).shuffle(m["ledger"])
    a, b = reconcile(MONTH, None), reconcile(m, None)
    assert (sorted(a.matched), sorted(a.items), a.adjusted_bank) == (sorted(b.matched), sorted(b.items), b.adjusted_bank)

def test_replay_is_identical_and_makes_no_call(tmp_path):
    path = str(tmp_path / "replies.json")
    live = Recorded(fake({"proposals": [{"bank_id": TP_B, "ledger_id": TP_L, "reason": "x"}]}), "m", path)
    first = reconcile(MONTH, live)
    replay = Recorded(None, "m", path)
    second = reconcile(MONTH, replay)
    assert (live.calls, replay.calls) == (1, 0)
    assert first == second
    with pytest.raises(LookupError):
        reconcile(MONTH, Recorded(None, "other-model", path))

完整的测试套件有八个测试。按顺序，它们分别检查：

• 金额解析器会拒绝 0.1、"7889.4"、"1.234" 和 "1,254.30"；
• 在 Decimal 字段中，JSON 数字确实会丢失分位；
• 仅靠代码时，两笔相同的电汇都保持未匹配；
• 幻觉出来的 ID、金额不符的配对和有效配对各自得到正确的结果；
• 删掉一条银行流水，会导致与对账单的勾稽失败；
• 打乱输入顺序不会改变任何结果；
• 回放结果完全一致，且不发起任何调用；
• 说明文字中的无依据数字会被捕获。

这些测试都不调用模型。代替模型的是一个返回固定回复的假对象，而这恰恰是重点：这些保证是代码的属性，所以可以像测试代码一样测试它们。

步骤 9库：各自适用于什么，以及我们踩过的坑。

下面每个库都在 2026-10-22 安装并实际试用过。标注为“实测”的行为是我们自己跑出来的。

金额与算术。

• Python decimal：正确的基础类型，前提是始终从字符串构造值，并用显式的舍入模式量化。默认上下文是 ROUND_HALF_EVEN，即银行家舍入，所以 Decimal("2.665") 舍入为 2.66，而四舍五入得到 2.67。有些税务和开票规则要求四舍五入，因此要把舍入模式做成一个具名参数。
• 整数最小货币单位：本页采用的方式。用 BIGINT 存分，或者在 Postgres 中用 NUMERIC(p,s) 列，绝不用 float8。babel 2.18.0 提供每种货币的精度：JPY 为 0 位小数，BHD 为 3 位。
• py-moneyed 3.0：能阻止你把 USD 和 EUR 相加，但不会按货币精度量化（Money("10.005", "USD") 仍是 10.005），而且最后一次发布是在 2022 年。
• JavaScript：dinero.js 2.0.2（经历多年 alpha 版本后，自 2026 年 3 月起稳定）使用整数最小货币单位，并提供 allocate。decimal.js 10.6.0 和 big.js 7.0.1 负责任意精度算术。

账簿。

• beancount 3.2.3：会拒绝不平衡的交易。实测：写成两位小数的 balance 断言会容忍一分钱的差异，所以 -1199.99 USD 在实际为 -1200.00 时也通过了。写成 -1199.99 ~ 0.00 USD 才是精确比对。许可证为 GPL-2.0。
• TigerBeetle 0.17.9：专为借贷记账打造的数据库，其两阶段挂起转账可以对应到制单-复核，转账 ID 具备幂等性。它仍处于 0.x 阶段，所以要把客户端和服务器版本一起锁定。
• Formance Ledger 和 Blnk：可自行部署的账簿服务。Blnk 自带一个带匹配规则的对账引擎。
• python-accounting 2.0.0：不同时装好 MySQL 和 Postgres 客户端库就无法安装。

差异比对与数据校验。

• datacompy 1.1.0（Capital One）：按键比对两份导出数据的最快方式。实测：1.x 版本移除了 datacompy.Compare，所以要用 PandasCompare 并设置 abs_tol=0。它能处理 Decimal 列。
• pandera 0.34.1、Great Expectations 1.24.0 和 Pointblank 1.1.1：在流水线的数据表进入对账之前先做校验。Great Expectations 正在转由 Fivetran 管理。
• Splink 5.0.0：用于交易对手名称的概率性匹配。把它的分数送入一个确定性的阈值，绝不直接用于过账。第 5 版有破坏性的 API 变更。

规则、策略与证明。

• GoRules ZEN（zen-engine 2.1.3）：财务主管看得懂的决策表。实测：数字返回 Python 时是浮点数（10000.10 返回为 10000.1），所以进出都用整数分。
• Cedar 和 OPA：用于“谁可以审批什么”。Cedar 的制单-复核规则见 STEP 5。
• Z3（z3-solver 5.3.1.0）：用编码好的规则检验模型给出的结论。实测：在费用政策为 2%、上限 25.00，余额为 1,500.00 的条件下，“费用是 35.00”这一断言返回 unsat。
• AWS Bedrock Automated Reasoning checks：这一思路的托管版本，自 2025 年 8 月起正式可用。它只报告发现，从不拦截内容。它仅支持英语、不支持流式输出，其文档也提醒，复利这类非线性算术可能返回 TOO_COMPLEX。用它来处理资格和政策类问题，而不是计算金额。

与模型交互。

• 原生结构化输出、Instructor 1.17.0、PydanticAI 和 BAML：它们都能让你拿到合法的结构，但没有一个能让你拿到正确的数字。Instructor 的“重试直到合法”循环会让输出最终合法，同时多出需要记录的调用。BAML 的宽松解析会把格式错误的回复强行凑成结构。无论哪种，都要在下游校验。
• Outlines 1.3.3：用于自托管模型的约束解码。
• Guardrails AI 和 NeMo Guardrails：用于安全与格式检查。LLM Guard 的代码仓库已归档。

持久化执行。Temporal、DBOS 或 Restate 会把每次模型调用记录为一个步骤，因此恢复的工作流会回放记录下来的回复，而不是再次调用模型。这就是基础设施层面的 STEP 6。Restate 的服务器采用 BSL 1.1 许可证。

• 从这里起步：配合严格字符串解析器的整数分、一轮拒绝歧义的精确匹配、只返回 ID 配对的模型、负责分类的代码、一条审批策略，以及一个回复记录存储。这些全部只需 Python 标准库加上 Pydantic。
• 当智能体的输出会过账时，再引入账簿（TigerBeetle、Formance 或 Blnk），不要提前。
• 当模型对政策作出断言时，再引入 Z3 或 Automated Reasoning，例如资格、费用减免或限额，而不是提出匹配建议的场景。
• 永远不要引入让 LLM 评审来裁决某个数字。边界篇解释了原因。