From 0ce272e94b56e428b4dedb2a98b0261b61ef53cc Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 16 Oct 2026 10:02:53 +0800 Subject: [PATCH] =?UTF-8?q?=E6=AF=8F=E6=97=A5=E8=B4=A6=E6=B3=9B=E5=8C=96?= =?UTF-8?q?=E6=88=90=20Ledger(key)=EF=BC=88=E7=BB=86=E6=A1=88=2001=20?= =?UTF-8?q?=E5=81=9A=E6=B3=95=2017=EF=BC=89=EF=BC=9A=E6=A0=B8=E6=9F=A5?= =?UTF-8?q?=E3=80=81=E8=A3=81=E5=88=A4=E4=BB=A5=E5=89=8D=E5=90=84=E5=86=99?= =?UTF-8?q?=E4=B8=80=E4=BB=BD=E4=B8=80=E6=A8=A1=E4=B8=80=E6=A0=B7=E7=9A=84?= =?UTF-8?q?=E9=A2=84=E7=95=99=E3=80=81=E6=89=BE=E8=A1=A5=E3=80=81=E6=8C=89?= =?UTF-8?q?=E6=97=A5=E6=BB=9A=E5=8A=A8=EF=BC=8C=E8=AF=84=E6=B5=8B=E5=86=8D?= =?UTF-8?q?=E6=8A=84=E5=B0=B1=E6=98=AF=E7=AC=AC=E4=B8=89=E4=BB=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - engine/spend.py：Ledger(key) 一套代码，三个实例各记各的键——verify_spend（核查）、judge_spend（裁判）、eval_spend（评测，预留，设置和接口第 3 波加）；锁按（事件循环, 键）各一把，一本触顶不挤占另外两本 - 核查的模块函数（SPEND_KEY、daily_spend、charge、settle）照旧，委托给 verify_spend - engine/judge.py：_charge / _settle 和它自己那份 _today、_spend_of、模块级锁去掉，改用 spend.judge_spend；SPEND_KEY、daily_spend 照旧，裁判的预留、找补、上限行为不变 - 三本账按同一个本地日期滚动（spend._today）：test_evidence_judge 里模拟跨天改成换 spend._today（以前换的是裁判自己那份时钟，已经不在了） - 新增 tests/test_spend_ledger.py（三本账三个键、预留找补与上限、跨天、互不挤占、同一个时钟） --- backend/app/engine/judge.py | 64 ++--------- backend/app/engine/spend.py | 160 +++++++++++++++++---------- backend/tests/test_evidence_judge.py | 6 +- backend/tests/test_spend_ledger.py | 78 +++++++++++++ 4 files changed, 190 insertions(+), 118 deletions(-) create mode 100644 backend/tests/test_spend_ledger.py diff --git a/backend/app/engine/judge.py b/backend/app/engine/judge.py index 53c6dff4..c9f6bc90 100644 --- a/backend/app/engine/judge.py +++ b/backend/app/engine/judge.py @@ -37,7 +37,6 @@ import re import time from collections.abc import Callable, Iterable from dataclasses import asdict, dataclass -from datetime import datetime from typing import Any from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage @@ -68,6 +67,7 @@ from app.engine.evidence import ( ) from app.engine.expressions import CellError, column_kind, locate_cell from app.engine.merge_query import MERGE_SOURCE +from app.engine import spend from app.engine.nodes.llm import _split_structured, _usage_of from app.engine.state import message_text from app.providers import catalog as pricing @@ -855,8 +855,8 @@ JUDGE_DEFAULTS: dict[str, Any] = { "click_max_cost_usd": 0.01, # 探索运行里每点一次「请模型判断」的金额上限 "daily_max_usd": 2.0, # 全局每日裁判金额上限（按本地日期） } -#: 每日计数存在设置表的这一行：{date: 本地日期, usd, calls, unpriced_calls}。不从设置接口进出 -SPEND_KEY = "judge_spend" +#: 每日计数存在设置表的这一行：{date: 本地日期, usd, calls, unpriced_calls}。不从设置接口进出（账本见 engine/spend.py） +SPEND_KEY = spend.judge_spend.key _COUNTS = frozenset({"report_max_claims"}) _SECONDS = frozenset({"report_timeout_s"}) @@ -1024,61 +1024,13 @@ async def source_masks(catalog: dict[str, Any], *, loader: Callable[[str], Any] # 每日计数 # -------------------------------------------------------------------------- -_spend_lock = asyncio.Lock() - - -def _today() -> str: - """本地日期：每日上限按它滚动。""" - return datetime.now().astimezone().date().isoformat() - - -def _spend_of(row: Any) -> dict[str, Any]: - value = row.value if row is not None and isinstance(row.value, dict) else {} - today = _today() - if value.get("date") != today: - return {"date": today, "usd": 0.0, "calls": 0, "unpriced_calls": 0} - return {"date": today, "usd": float(value.get("usd") or 0.0), "calls": int(value.get("calls") or 0), - "unpriced_calls": int(value.get("unpriced_calls") or 0)} +#: 裁判这本每日账（engine/spend.py 的 Ledger：预留、找补、按本地日期滚动和核查、评测是同一套代码，各记各的键） +_LEDGER = spend.judge_spend async def daily_spend() -> dict[str, Any]: """今天（本地日期）裁判已经花了多少：{date, usd, calls, unpriced_calls}。估不出金额的调用只计次数。""" - async with SessionLocal() as session: - return _spend_of(await session.get(Setting, SPEND_KEY)) - - -async def _charge(amount: float, *, limit: float | None, priced: bool) -> str | None: - """记一次调用、先按估算预留金额。加上这笔会超过每日上限就不记，返回 None；记下了返回记在哪天。""" - async with _spend_lock: - async with SessionLocal() as session: - row = await session.get(Setting, SPEND_KEY) - spend = _spend_of(row) - if limit is not None and spend["usd"] + amount > limit + 1e-12: - return None - spend["usd"] = round(spend["usd"] + amount, 8) - spend["calls"] += 1 - spend["unpriced_calls"] += 0 if priced else 1 - if row is None: - session.add(Setting(key=SPEND_KEY, value=spend)) - else: - row.value = spend - await session.commit() - return spend["date"] - - -async def _settle(date: str, delta: float) -> None: - """调用结束后按实际用量找补预留的差额。跨了天就不找补：预留记在前一天，今天从零算。""" - if not delta: - return - async with _spend_lock: - async with SessionLocal() as session: - row = await session.get(Setting, SPEND_KEY) - spend = _spend_of(row) - if row is None or spend["date"] != date: - return - spend["usd"] = max(0.0, round(spend["usd"] + delta, 8)) - row.value = spend - await session.commit() + return await _LEDGER.daily() # -------------------------------------------------------------------------- @@ -1318,7 +1270,7 @@ async def run_request(request: JudgeRequest, *, spec: ModelSpec, budget: Budget, pending = pending[:len(batch)] pending = pending[len(batch):] charged = cost or 0.0 - day = await _charge(charged, limit=budget.daily_max_usd if priced else None, priced=priced) + day = await _LEDGER.charge(charged, limit=budget.daily_max_usd if priced else None, priced=priced) if day is None: # 估算之后、记账之前别的裁判先花掉了今天剩下的钱 stop([*batch, *pending], "daily_max_usd") pending = [] @@ -1353,7 +1305,7 @@ async def run_request(request: JudgeRequest, *, spec: ModelSpec, budget: Budget, inp, out = sum(t["input_tokens"] for t in tokens), sum(t["output_tokens"] for t in tokens) # 回复没带用量（有的网关不给、调用超时）就按估算记：宁可多记，不能让每日上限漏掉 actual = (pricing.estimate_cost(model_id, inp, out) if inp or out else charged) if priced else 0.0 - await _settle(day, actual - charged) + await _LEDGER.settle(day, actual - charged) spent += actual usage["input_tokens"] += inp usage["output_tokens"] += out diff --git a/backend/app/engine/spend.py b/backend/app/engine/spend.py index 0efa52a6..900324fd 100644 --- a/backend/app/engine/spend.py +++ b/backend/app/engine/spend.py @@ -1,11 +1,18 @@ -"""核查花费账：双盲复算、反方审稿、审批质询里的模型调用，按本地日期记一本每日账。 +"""模型调用的每日花费账：一本账一个键，按本地日期记，先预留、后找补（docs/design/next-round/01-效果度量与回归评测.md 做法 17）。 -照结论句裁判的每日计数做（judge.py 的 SPEND_KEY、_charge、_settle）：调用前按目录价估算、先预留，加上这笔会超过 -每日上限就不记；调用后按实际用量找补差额。账落在设置表里，按本地日期滚动。 +调用前按目录价估算、先预留，加上这笔会超过每日上限就不记；调用后按实际用量找补差额。账落在设置表里，按本地日期 +滚动，不从设置接口进出（api/settings.py 只给只读的「今天花了多少」）。 -**为什么和裁判分开记。** 核查是一个 agent 查库、来回好几轮，一次花的钱比裁判判一份报告多得多。记在同一本账上， -核查一多，当天的裁判就全因为「每日上限」判不了——报告的出具会因此降档，而降档的原因和这份报告毫无关系。两本账 -各管各的上限，互不挤占。 +**一套代码，几本账。** 以前核查（这个模块）和结论句裁判（judge.py 的 _charge、_settle）各写一份一模一样的读改写，评测 +要是再抄一份就是第三份。现在都是 Ledger(key) 的实例，各记各的键： + +- verify_spend：核查——双盲复算、反方审稿、审批质询里的模型调用； +- judge_spend：结论句裁判（engine/judge.py）； +- eval_spend：评测（预留，第 3 波的评测执行接上；设置和接口在那时加）。 + +**为什么各记各的。** 核查是一个 agent 查库、来回好几轮，一次花的钱比裁判判一份报告多得多；评测一次几十道题。记在同一本 +账上，一边一多，另一边当天就全因为「每日上限」做不了——报告的出具会因此降档，而降档的原因和这份报告毫无关系。几本账 +各管各的上限，互不挤占。三本按同一个本地日期滚动（_today）。 目录里没有价格的模型估不出金额：只计次数（unpriced_calls），金额上限对它不起作用，由调用方照实说明。 """ @@ -13,27 +20,26 @@ from __future__ import annotations import asyncio import weakref +from collections.abc import Callable from datetime import datetime from typing import Any from app.db.base import SessionLocal from app.db.models import Setting -#: 每日计数存在设置表的这一行：{date: 本地日期, usd, calls, unpriced_calls}。不从设置接口进出（api/settings.py） -SPEND_KEY = "verify_spend" - -_locks: "weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, asyncio.Lock]" = weakref.WeakKeyDictionary() +#: 每本账的读改写锁：{事件循环: {键: 锁}}。asyncio.Lock 不能跨循环用（测试每个用例一个循环），按循环各一组；同一个键 +#: 同一把——几个核查并行时各自预留，不串行的话两边读到同一个余额，会一起越过上限 +_locks: "weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, dict[str, asyncio.Lock]]" = weakref.WeakKeyDictionary() -def _lock() -> asyncio.Lock: - """读改写这一行时持的锁，按事件循环各一把（asyncio.Lock 不能跨循环用）。 - - 几个核查并行时各自预留：不串行的话两边读到同一个余额，会一起越过上限。 - """ +def _lock(key: str) -> asyncio.Lock: loop = asyncio.get_running_loop() - lock = _locks.get(loop) + group = _locks.get(loop) + if group is None: + group = _locks[loop] = {} + lock = group.get(key) if lock is None: - lock = _locks[loop] = asyncio.Lock() + lock = group[key] = asyncio.Lock() return lock @@ -42,57 +48,91 @@ def _today() -> str: return datetime.now().astimezone().date().isoformat() -def _spend_of(row: Any) -> dict[str, Any]: - value = row.value if row is not None and isinstance(row.value, dict) else {} - today = _today() - if value.get("date") != today: - return {"date": today, "usd": 0.0, "calls": 0, "unpriced_calls": 0} - return {"date": today, "usd": float(value.get("usd") or 0.0), "calls": int(value.get("calls") or 0), - "unpriced_calls": int(value.get("unpriced_calls") or 0)} +class Ledger: + """一本每日账，存在设置表 key 这一行：{date: 本地日期, usd, calls, unpriced_calls}。 + + today 是这本账的日历（测试换掉它来模拟跨天）；不给就按本模块的 _today，调用时才取，换掉 _today 对没给 today 的几本 + 账都生效。""" + + def __init__(self, key: str, *, today: Callable[[], str] | None = None) -> None: + self.key = key + self._clock = today + + def _date(self) -> str: + return self._clock() if self._clock is not None else _today() + + def _spend_of(self, row: Any) -> dict[str, Any]: + value = row.value if row is not None and isinstance(row.value, dict) else {} + today = self._date() + if value.get("date") != today: + return {"date": today, "usd": 0.0, "calls": 0, "unpriced_calls": 0} + return {"date": today, "usd": float(value.get("usd") or 0.0), "calls": int(value.get("calls") or 0), + "unpriced_calls": int(value.get("unpriced_calls") or 0)} + + async def daily(self) -> dict[str, Any]: + """今天（本地日期）这本账已经花了多少：{date, usd, calls, unpriced_calls}。估不出金额的调用只计次数。""" + async with SessionLocal() as session: + return self._spend_of(await session.get(Setting, self.key)) + + async def charge(self, amount: float, *, limit: float | None, priced: bool) -> str | None: + """记一次调用、先按估算预留金额。加上这笔会超过每日上限就不记，返回 None；记下了返回记在哪天（找补时要用）。 + + limit 为 None 表示不限。估不出金额的调用（priced=False）照记次数，金额记 0，调用方应传 limit=None： + 拿 0 去比上限，等于说它不花钱。 + """ + async with _lock(self.key): + async with SessionLocal() as session: + row = await session.get(Setting, self.key) + spend = self._spend_of(row) + if limit is not None and spend["usd"] + amount > limit + 1e-12: + return None + spend["usd"] = round(spend["usd"] + amount, 8) + spend["calls"] += 1 + spend["unpriced_calls"] += 0 if priced else 1 + if row is None: + session.add(Setting(key=self.key, value=spend)) + else: + row.value = spend + await session.commit() + return spend["date"] + + async def settle(self, date: str, delta: float) -> None: + """调用结束后按实际用量找补预留的差额。跨了天就不找补：预留记在前一天，今天从零算。""" + if not delta: + return + async with _lock(self.key): + async with SessionLocal() as session: + row = await session.get(Setting, self.key) + spend = self._spend_of(row) + if row is None or spend["date"] != date: + return + spend["usd"] = max(0.0, round(spend["usd"] + delta, 8)) + row.value = spend + await session.commit() + + +#: 核查、裁判、评测各一本 +verify_spend = Ledger("verify_spend") +judge_spend = Ledger("judge_spend") +eval_spend = Ledger("eval_spend") + +#: 核查那本账的老名字（verify_agent、设置接口、已有测试从这里取） +SPEND_KEY = verify_spend.key async def daily_spend() -> dict[str, Any]: - """今天（本地日期）核查已经花了多少：{date, usd, calls, unpriced_calls}。估不出金额的调用只计次数。""" - async with SessionLocal() as session: - return _spend_of(await session.get(Setting, SPEND_KEY)) + """今天核查已经花了多少（verify_spend.daily）。""" + return await verify_spend.daily() async def charge(amount: float, *, limit: float | None, priced: bool) -> str | None: - """记一次调用、先按估算预留金额。加上这笔会超过每日上限就不记，返回 None；记下了返回记在哪天（找补时要用）。 - - limit 为 None 表示不限。估不出金额的调用（priced=False）照记次数，金额记 0，调用方应传 limit=None： - 拿 0 去比上限，等于说它不花钱。 - """ - async with _lock(): - async with SessionLocal() as session: - row = await session.get(Setting, SPEND_KEY) - spend = _spend_of(row) - if limit is not None and spend["usd"] + amount > limit + 1e-12: - return None - spend["usd"] = round(spend["usd"] + amount, 8) - spend["calls"] += 1 - spend["unpriced_calls"] += 0 if priced else 1 - if row is None: - session.add(Setting(key=SPEND_KEY, value=spend)) - else: - row.value = spend - await session.commit() - return spend["date"] + """核查记一次调用（verify_spend.charge）。""" + return await verify_spend.charge(amount, limit=limit, priced=priced) async def settle(date: str, delta: float) -> None: - """调用结束后按实际用量找补预留的差额。跨了天就不找补：预留记在前一天，今天从零算。""" - if not delta: - return - async with _lock(): - async with SessionLocal() as session: - row = await session.get(Setting, SPEND_KEY) - spend = _spend_of(row) - if row is None or spend["date"] != date: - return - spend["usd"] = max(0.0, round(spend["usd"] + delta, 8)) - row.value = spend - await session.commit() + """核查找补（verify_spend.settle）。""" + await verify_spend.settle(date, delta) -__all__ = ["SPEND_KEY", "charge", "daily_spend", "settle"] +__all__ = ["Ledger", "SPEND_KEY", "charge", "daily_spend", "eval_spend", "judge_spend", "settle", "verify_spend"] diff --git a/backend/tests/test_evidence_judge.py b/backend/tests/test_evidence_judge.py index f46d6a62..c742ccf1 100644 --- a/backend/tests/test_evidence_judge.py +++ b/backend/tests/test_evidence_judge.py @@ -30,6 +30,7 @@ from app.core.events import EventType from app.db.base import SessionLocal from app.db.models import Setting from app.engine import judge as judge_mod +from app.engine import spend from app.engine.evidence import build_catalog, compose_doc, iter_units from app.engine.judge import ( BATCH_SIZE, @@ -503,7 +504,7 @@ async def test_unlimited_never_truncates(monkeypatch, stored): raw = "".join(f"第 {i} 个区域的销售额增长来自新客。[[see:m:gmv]]" for i in range(BATCH_SIZE + 5)) doc = compose_doc(raw, catalog) async with SessionLocal() as session: # 今天已经花掉很多：不限就是不限 - session.add(Setting(key=SPEND_KEY, value={"date": judge_mod._today(), "usd": 999.0, "calls": 9})) + session.add(Setting(key=SPEND_KEY, value={"date": spend._today(), "usd": 999.0, "calls": 9})) await session.commit() outcome = await judge_doc(doc, catalog, spec=SPEC, budget=UNLIMITED) assert [len(units) for units in judge.judged] == [BATCH_SIZE, 5] @@ -517,7 +518,8 @@ async def test_the_daily_cap_resets_the_next_day(monkeypatch, stored): cost = prepare(doc, catalog).cost(PRICED, prepare(doc, catalog).cands) budget = Budget(max_claims=None, max_cost_usd=None, timeout_s=None, daily_max_usd=cost * 1.05) day = {"today": "2026-09-28"} - monkeypatch.setattr(judge_mod, "_today", lambda: day["today"]) + # 裁判的账和核查、评测按同一个本地日期滚动（engine/spend.py 的 Ledger） + monkeypatch.setattr(spend, "_today", lambda: day["today"]) judge = Judge(monkeypatch) first = await judge_doc(doc, catalog, spec=SPEC, budget=budget) diff --git a/backend/tests/test_spend_ledger.py b/backend/tests/test_spend_ledger.py new file mode 100644 index 00000000..a847ed7e --- /dev/null +++ b/backend/tests/test_spend_ledger.py @@ -0,0 +1,78 @@ +"""每日账泛化成 Ledger(key)（docs/design/next-round/01-效果度量与回归评测.md 做法 17）。 + +核查、裁判以前各写一份一模一样的「预留 → 找补 → 按本地日期滚动」，评测要是再抄一份就是第三份。现在三本账是同一个 +Ledger 的三个实例，各记各的键、各管各的上限：一本触顶不挤占另外两本。裁判、核查原有的行为（模块级的 SPEND_KEY、 +daily_spend、charge / settle）不变，已有的测试照旧守着；这里守账本本身和互不挤占。 +""" +from __future__ import annotations + +import pytest +from sqlalchemy import delete + +from app.db.base import SessionLocal +from app.db.models import Setting +from app.engine import judge, spend +from app.engine.spend import Ledger, eval_spend, judge_spend, verify_spend + + +@pytest.fixture(autouse=True) +async def _clean(): + keys = [verify_spend.key, judge_spend.key, eval_spend.key, "test_ledger"] + async with SessionLocal() as session: + await session.execute(delete(Setting).where(Setting.key.in_(keys))) + await session.commit() + yield + async with SessionLocal() as session: + await session.execute(delete(Setting).where(Setting.key.in_(keys))) + await session.commit() + + +def test_three_books_three_keys(): + assert (verify_spend.key, judge_spend.key, eval_spend.key) == ("verify_spend", "judge_spend", "eval_spend") + # 老名字照旧指向各自那本：核查的模块函数、裁判的 SPEND_KEY / daily_spend + assert spend.SPEND_KEY == verify_spend.key + assert judge.SPEND_KEY == judge_spend.key + + +async def test_reserve_then_settle_and_the_limit(): + book = Ledger("test_ledger") + day = await book.charge(0.4, limit=1.0, priced=True) + assert day is not None + assert await book.charge(0.7, limit=1.0, priced=True) is None, "加上这笔超过上限就不记" + await book.settle(day, -0.1) + got = await book.daily() + assert got["usd"] == pytest.approx(0.3) and got["calls"] == 1 and got["unpriced_calls"] == 0 + # 估不出金额的照记次数，不比上限 + assert await book.charge(0.0, limit=None, priced=False) is not None + assert (await book.daily())["unpriced_calls"] == 1 + + +async def test_a_new_day_starts_from_zero(): + day = {"today": "2026-09-28"} + book = Ledger("test_ledger", today=lambda: day["today"]) + first = await book.charge(0.9, limit=1.0, priced=True) + assert first == "2026-09-28" + assert await book.charge(0.2, limit=1.0, priced=True) is None + day["today"] = "2026-09-29" + assert await book.charge(0.2, limit=1.0, priced=True) == "2026-09-29" + await book.settle(first, -0.5) # 跨了天的预留不找补：今天从零算 + assert (await book.daily())["usd"] == pytest.approx(0.2) + + +async def test_one_book_hitting_its_cap_does_not_squeeze_the_others(): + """核查花光了当天的额度，裁判、评测照常记：三本账互不挤占。""" + assert await verify_spend.charge(2.0, limit=2.0, priced=True) is not None + assert await verify_spend.charge(0.01, limit=2.0, priced=True) is None + assert await judge_spend.charge(1.0, limit=2.0, priced=True) is not None + assert await eval_spend.charge(1.5, limit=2.0, priced=True) is not None + assert (await judge.daily_spend())["usd"] == pytest.approx(1.0) + assert (await spend.daily_spend())["usd"] == pytest.approx(2.0) + assert (await eval_spend.daily())["usd"] == pytest.approx(1.5) + + +async def test_all_books_roll_over_on_the_same_clock(monkeypatch): + """三本账按同一个本地日期滚动（spend._today）：裁判不再自己另有一份时钟。""" + monkeypatch.setattr(spend, "_today", lambda: "2026-01-02") + assert await judge_spend.charge(0.1, limit=None, priced=True) == "2026-01-02" + assert (await judge.daily_spend())["date"] == "2026-01-02" + assert (await eval_spend.daily())["date"] == "2026-01-02"