From aeaca4ce6938eca9bda2002535cd4639630dd4e8 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Tue, 3 Nov 2026 19:38:00 +0800 Subject: [PATCH] =?UTF-8?q?=E8=AE=B0=E5=BD=95=E5=8F=A3=E5=BE=84=E8=B0=83?= =?UTF-8?q?=E7=94=A8=E5=AE=A1=E8=AE=A1=EF=BC=9Acompose=20=E4=B8=8E?= =?UTF-8?q?=E6=AD=A3=E5=BC=8F=E5=87=BA=E5=85=B7=E5=90=84=E8=AE=B0=E4=B8=80?= =?UTF-8?q?=E6=9D=A1=E5=8F=82=E6=95=B0=E6=8C=87=E7=BA=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/copilot.py | 8 +++ backend/app/api/copilot_gen.py | 9 ++++ backend/app/ledger/issue.py | 10 ++++ backend/app/ledger/usage.py | 87 ++++++++++++++++++++++++++++++ backend/tests/test_ledger_issue.py | 9 +++- backend/tests/test_ledger_usage.py | 55 +++++++++++++++++++ 6 files changed, 177 insertions(+), 1 deletion(-) create mode 100644 backend/app/ledger/usage.py create mode 100644 backend/tests/test_ledger_usage.py diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index b33c72a8..12fa575b 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -2076,6 +2076,14 @@ async def answer_ops(payload: GenerateIn, *, catalog_view: Any = None, analysis_ spent = usage.meter() async for op in usage.metered(steps(), spent): yield op + if applier.caliber_uses: + # 调用审计（caliber.use）：这一轮选用了哪几版口径、带什么参数。只记指纹，不落参数值；失败不影响这一轮交付 + from app.ledger.usage import audit_composed_uses + + try: + await audit_composed_uses(applier.caliber_uses, conversation_id=payload.conversation_id) + except Exception: # noqa: BLE001 - 审计是附加记录，不能挡住已经生成的图 + logger.warning("记录口径调用审计失败", exc_info=True) yield {"op": "usage", **spent.summary()} diff --git a/backend/app/api/copilot_gen.py b/backend/app/api/copilot_gen.py index 5d19c362..02557718 100644 --- a/backend/app/api/copilot_gen.py +++ b/backend/app/api/copilot_gen.py @@ -544,6 +544,9 @@ class GraphApplier: self.composer = composer #: 选用口径实例化出来的段：[(第一个取数节点, 全部节点 id)]，接入口用 self.composed: list[tuple[str, list[str]]] = [] + #: 这一轮选用过的口径（调用审计 caliber.use 用）：[{workflow_id, version, card, metrics, inputs}]。 + #: inputs 只在内存里，写审计时只留指纹，不落参数值 + self.caliber_uses: list[dict[str, Any]] = [] #: 已经落地的 use_caliber（ledger/compose.signature），接着写时重发的同一个不再落地 self._composed_ops: set[str] = set() #: 系统接上的连线（入口 → 取数）：模型再发一条一模一样的不重复加 @@ -643,6 +646,12 @@ class GraphApplier: if composed.op_id not in self._landed: self._landed.append(composed.op_id) self._wire() + self.caliber_uses.append({ + "workflow_id": composed.option.workflow_id, "version": int(composed.option.version), + "card": composed.option.node_id, + "metrics": [str(m.get("id")) for m in composed.metrics if m.get("id")], + "inputs": dict(composed.inputs or {}), + }) op["result"] = composed.result() return Applied(True, True) diff --git a/backend/app/ledger/issue.py b/backend/app/ledger/issue.py index ff742d6c..c66ba79c 100644 --- a/backend/app/ledger/issue.py +++ b/backend/app/ledger/issue.py @@ -392,6 +392,16 @@ async def issue(run_id: str, workflow_id: str, version: int, *, principal: Any, except BaseException: await quota.give_back(QUOTA, day=day) raise + try: + from app.ledger import usage as ledger_usage + + await ledger_usage.record_use( + session, workflow_id=workflow_id, version=int(version), how="issued", params=plan.inputs, + run_id=run_id, actor=actor, auto=auto) + await session.commit() + except Exception: # noqa: BLE001 - 审计是附加记录，正式运行已经发起，不能因此回滚 + logger.warning("记录口径调用审计失败（来源运行 %s，口径 %s@%s）", run_id, workflow_id, version, + exc_info=True) return launched, True diff --git a/backend/app/ledger/usage.py b/backend/app/ledger/usage.py new file mode 100644 index 00000000..70c6a222 --- /dev/null +++ b/backend/app/ledger/usage.py @@ -0,0 +1,87 @@ +"""口径调用审计（next-round 02 的后续）：谁在什么时候、用哪一版口径、带什么参数。 + +`caliber_match` 是封存后的等价证明，不是调用日志；一次 `use_caliber` 组合或一次按口径正式出具，在这里各记一条 +`AuditEvent`（action=caliber.use）。和账本其他部分一样不另建表：audit_events 已经按 subject 建了索引，按 +subject_kind=caliber、subject_id=「工作流@版本」查得到。 + +**只记参数指纹，不记参数值。** 参数可能是遮罩值、手机号这类不该再落一份的东西；记 hash 和字段名，排查「同一次调用 +用了哪组参数」够用，要复核参数去运行输入和比对工件里看（那里本来就有遮罩处理）。指标只记 id，不记算出来的值。 +""" +from __future__ import annotations + +import logging +from typing import Any, Iterable, Mapping + +from sqlalchemy.ext.asyncio import AsyncSession + +ACTION = "caliber.use" +SUBJECT_KIND = "caliber" +HOWS = ("composed", "issued") + +logger = logging.getLogger(__name__) + + +def params_fingerprint(values: Mapping[str, Any] | None) -> tuple[str, list[str]]: + """(参数指纹, 字段名)。空值不参与指纹：没填和填空串是同一组参数；字段名排序，保证同一组参数写法不同也相同。""" + from app.core.artifact_store import canonical_json, content_hash + + clean = {str(k): v for k, v in (values or {}).items() if v is not None and v != ""} + return content_hash(canonical_json(clean)), sorted(clean) + + +async def record_use(session: AsyncSession, *, workflow_id: str, version: int, how: str, + card: str | None = None, metrics: Iterable[str] = (), params: Mapping[str, Any] | None = None, + run_id: str | None = None, conversation_id: str | None = None, + actor: str | None = None, auto: bool | None = None) -> None: + """记一条调用。不提交：和调用方自己的事务一起提交。how 只能是 composed / issued。""" + from app.apps.audit import record + + if how not in HOWS: + raise ValueError(f"不认识的调用方式：{how}") + digest, fields = params_fingerprint(params) + data: dict[str, Any] = { + "workflow_id": workflow_id, + "version": int(version), + "how": how, + "params_hash": digest, + "param_fields": fields, + "metrics": [str(m) for m in dict.fromkeys(metrics)], + } + if card: + data["card"] = card + if run_id: + data["run_id"] = run_id + if conversation_id: + data["conversation_id"] = conversation_id + if auto is not None: + data["auto"] = bool(auto) + await record(session, action=ACTION, subject_kind=SUBJECT_KIND, + subject_id=f"{workflow_id}@{int(version)}", actor=actor, data=data) + + +async def audit_composed_uses(uses: Iterable[Mapping[str, Any]], *, conversation_id: str | None = None, + actor: str | None = None) -> int: + """一轮生成里所有 use_caliber 落成审计，返回记了几条。取数组合发生在建图阶段，没有运行可挂，只记对话。""" + uses = list(uses) + if not uses: + return 0 + from app.db.base import SessionLocal + + async with SessionLocal() as session: + for use in uses: + await record_use( + session, + workflow_id=str(use.get("workflow_id") or ""), + version=int(use.get("version") or 0), + how="composed", + card=str(use.get("card") or "") or None, + metrics=use.get("metrics") or (), + params=use.get("inputs") if isinstance(use.get("inputs"), Mapping) else None, + conversation_id=conversation_id, + actor=actor, + ) + await session.commit() + return len(uses) + + +__all__ = ["ACTION", "HOWS", "SUBJECT_KIND", "audit_composed_uses", "params_fingerprint", "record_use"] diff --git a/backend/tests/test_ledger_issue.py b/backend/tests/test_ledger_issue.py index 00c7f4df..44b204a5 100644 --- a/backend/tests/test_ledger_issue.py +++ b/backend/tests/test_ledger_issue.py @@ -21,7 +21,7 @@ from sqlalchemy import select from app.api.deps import Principal from app.db.base import SessionLocal -from app.db.models import Run, Setting, WorkflowVersion +from app.db.models import AuditEvent, Run, Setting, WorkflowVersion from app.engine.runner import run_manager from app.ledger import issue as ledger_issue from app.ledger import match as ledger_match @@ -130,6 +130,13 @@ async def test_an_independent_equivalent_answer_to_a_governed_caliber_is_issued_ origin = formal.origin assert origin["source_run"] == row.id and origin["caliber"] == f"{wf}@1" and origin["how"] == "independent" assert origin["auto"] is True and origin["binding_hash"] and origin["match"] + async with SessionLocal() as session: + rows = (await session.execute(select(AuditEvent).where( + AuditEvent.action == "caliber.use", AuditEvent.subject_id == f"{wf}@1"))).scalars().all() + [use] = [r for r in rows if r.data.get("run_id") == row.id] + assert use.data["how"] == "issued" and use.data["auto"] is True + assert use.data["param_fields"] == ["end", "start", "store"] and use.data["params_hash"] + assert "2026-09-01" not in str(use.data) body = (await client.get(f"/api/runs/{row.id}/ledger")).json() [action] = [a for a in body["actions"] if a["kind"] == "issue"] diff --git a/backend/tests/test_ledger_usage.py b/backend/tests/test_ledger_usage.py new file mode 100644 index 00000000..a137caf8 --- /dev/null +++ b/backend/tests/test_ledger_usage.py @@ -0,0 +1,55 @@ +"""口径调用审计（app/ledger/usage.py）：谁在什么时候用了哪一版、带什么参数。 + +- 参数只记指纹和字段名，不记值； +- composed（问数据里 use_caliber）和 issued（按口径正式出具）各记一条 caliber.use； +- 同一组参数写法不同（键顺序、空值）指纹相同；参数不同指纹不同。 +""" +from __future__ import annotations + +import uuid + +from sqlalchemy import select + +from app.db.base import SessionLocal +from app.db.models import AuditEvent +from app.ledger.usage import audit_composed_uses, params_fingerprint, record_use + + +def test_params_fingerprint_ignores_order_and_blanks_but_not_values(): + first, fields = params_fingerprint({"start": "2026-09-01", "end": "2026-09-08", "store": ""}) + second, _ = params_fingerprint({"end": "2026-09-08", "start": "2026-09-01", "store": None}) + assert first == second and fields == ["end", "start"] + other, _ = params_fingerprint({"start": "2026-09-01", "end": "2026-09-07"}) + assert other != first + + +async def test_record_use_keeps_a_fingerprint_not_the_values(): + wf = f"usage{uuid.uuid4().hex[:10]}" + async with SessionLocal() as session: + await record_use(session, workflow_id=wf, version=3, how="issued", card="card", metrics=["gmv", "gmv"], + params={"start": "2026-09-01", "end": "2026-09-08"}, run_id="run-1", auto=True, + actor="王敏") + await session.commit() + [event] = (await session.execute(select(AuditEvent).where( + AuditEvent.action == "caliber.use", AuditEvent.subject_id == f"{wf}@3"))).scalars().all() + assert event.actor == "王敏" and event.subject_kind == "caliber" + assert event.data["how"] == "issued" and event.data["metrics"] == ["gmv"] + assert event.data["run_id"] == "run-1" and event.data["auto"] is True + assert event.data["param_fields"] == ["end", "start"] and event.data["params_hash"] + assert "2026-09-01" not in str(event.data) and "2026-09-08" not in str(event.data) + + +async def test_composed_uses_are_written_as_one_audit_each(): + wf = f"usage{uuid.uuid4().hex[:10]}" + count = await audit_composed_uses([ + {"workflow_id": wf, "version": 1, "card": "card", "metrics": ["gmv"], + "inputs": {"start": "2026-09-01", "end": "2026-09-08"}}, + {"workflow_id": wf, "version": 2, "card": "card2", "metrics": ["orders"], "inputs": {}}, + ], conversation_id="conv-1") + assert count == 2 + async with SessionLocal() as session: + rows = list((await session.execute(select(AuditEvent).where( + AuditEvent.action == "caliber.use", AuditEvent.subject_kind == "caliber") + .where(AuditEvent.subject_id.in_([f"{wf}@1", f"{wf}@2"])))).scalars()) + assert {r.subject_id for r in rows} == {f"{wf}@1", f"{wf}@2"} + assert all(r.data["how"] == "composed" and r.data["conversation_id"] == "conv-1" for r in rows)