From 2df591d05f7b7d62e9cba021aa5d12616b06a3cc Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 14 Sep 2026 10:10:20 +0800 Subject: [PATCH] =?UTF-8?q?=E5=B7=A5=E4=BB=B6=E5=BA=93=EF=BC=9A=E5=86=85?= =?UTF-8?q?=E5=AE=B9=E5=AF=BB=E5=9D=80=E5=AD=98=E5=82=A8=E3=80=81=E8=8A=82?= =?UTF-8?q?=E7=82=B9=E8=BE=93=E5=87=BA=E4=B8=8E=E5=B7=A5=E5=85=B7=E5=BF=AB?= =?UTF-8?q?=E7=85=A7=E8=90=BD=E7=9B=98=E3=80=81=E8=BF=90=E8=A1=8C=E6=B8=85?= =?UTF-8?q?=E5=8D=95=E5=93=88=E5=B8=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 事件流里的输出是给画布看的截断预览，出具溯源需要完整证据。 新增 data/artifacts/ 内容寻址工件库（sha256 即地址，取回时复验哈希， 篡改即报错）；节点完整输出与工具调用快照（query+结果集）自动落盘， 事件只携带工件 id。运行终态对全部事件计算清单哈希存入 Run， 事后对事件流的增删改会与它对不上。 graph_hash 剔除 viewport 后计算——挪画布不改变图的身份。 Run/Workflow/WorkflowVersion/Approval 预置分级与治理列， init_db 附带 SQLite 轻量列迁移。 --- backend/app/core/artifact_store.py | 110 +++++++++++++++++++++++++++++ backend/app/core/events.py | 4 ++ backend/app/db/base.py | 26 +++++++ backend/app/db/models.py | 36 ++++++++++ backend/app/engine/compiler.py | 22 ++++++ backend/app/engine/nodes/llm.py | 12 ++++ backend/app/engine/nodes/tools.py | 13 +++- backend/app/engine/runner.py | 25 +++++++ 8 files changed, 247 insertions(+), 1 deletion(-) create mode 100644 backend/app/core/artifact_store.py diff --git a/backend/app/core/artifact_store.py b/backend/app/core/artifact_store.py new file mode 100644 index 00000000..cb9b1e5c --- /dev/null +++ b/backend/app/core/artifact_store.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +import hashlib +import json +from pathlib import Path +from typing import Any + +from app.core.config import settings + +# -------------------------------------------------------------------------- +# 内容寻址的工件库 +# +# 事件流里的输出是给画布看的截断预览；这里存的是完整证据。 +# 工件按内容 sha256 寻址：同一份内容只存一份，且任何篡改都会改变地址本身， +# 这比"相信数据库没被改过"强得多。出具溯源、周对比 diff、基线回归、 +# 轨迹提模板，全部踩在这一层上。 +# -------------------------------------------------------------------------- + + +def canonical_json(obj: Any) -> str: + """规范化 JSON：键排序、紧凑分隔符。同一数据永远得到同一字节串，哈希才稳定。""" + return json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str) + + +def content_hash(data: bytes | str) -> str: + if isinstance(data, str): + data = data.encode() + return hashlib.sha256(data).hexdigest() + + +def graph_hash(graph: dict[str, Any]) -> str: + """图定义的指纹。formal run 钉住的就是这个值。 + + viewport 是纯视觉状态，挪一下画布不该改变图的身份，剔除后再哈希。 + """ + slim = {k: v for k, v in graph.items() if k != "viewport"} + return content_hash(canonical_json(slim)) + + +def _path_of(artifact_id: str) -> Path: + # 按前两位分桶，避免单目录塞几万个文件 + return settings.data_dir / "artifacts" / artifact_id[:2] / f"{artifact_id}.json" + + +async def put_json( + obj: Any, + *, + kind: str, + run_id: str | None = None, + node_id: str | None = None, + meta: dict[str, Any] | None = None, +) -> str: + """存一个 JSON 工件，返回其内容哈希（即工件 id）。 + + 幂等：同内容重复写只落一次盘；同一工件被多个 run 引用时， + 数据库里各记一行引用（溯源需要知道"谁产出的"），文件只有一份。 + """ + from app.db.base import SessionLocal + from app.db.models import Artifact + + text = canonical_json(obj) + artifact_id = content_hash(text) + path = _path_of(artifact_id) + if not path.exists(): + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(".tmp") + tmp.write_text(text, encoding="utf-8") + tmp.rename(path) # 原子落盘，写一半崩了不会留下半个工件 + + async with SessionLocal() as session: + existing = await session.get(Artifact, (artifact_id, run_id or "")) + if existing is None: + session.add( + Artifact( + id=artifact_id, + run_id=run_id or "", + kind=kind, + node_id=node_id, + size=len(text.encode()), + meta=meta or {}, + ) + ) + await session.commit() + return artifact_id + + +def load(artifact_id: str) -> Any | None: + """按 id 取回工件内容。取回时重新校验哈希——工件库的承诺就是内容不可变。""" + if not artifact_id or not artifact_id.isalnum(): + return None + path = _path_of(artifact_id) + if not path.exists(): + return None + text = path.read_text(encoding="utf-8") + if content_hash(text) != artifact_id: + raise ValueError(f"工件 {artifact_id[:12]}… 内容与哈希不符，疑似被篡改") + return json.loads(text) + + +def manifest_hash(rows: list[tuple[int, str, str | None, dict[str, Any]]]) -> str: + """把一次运行的全部事件压成一个清单哈希。 + + rows: [(seq, type, node_id, data)]。存在 Run 上之后，事后任何对 + 事件流的增删改都能被这一个值戳穿。 + """ + lines = [ + f"{seq}|{etype}|{node_id or ''}|{content_hash(canonical_json(data))}" + for seq, etype, node_id, data in sorted(rows, key=lambda r: r[0]) + ] + return content_hash("\n".join(lines)) diff --git a/backend/app/core/events.py b/backend/app/core/events.py index 70c992ce..cd34ea59 100644 --- a/backend/app/core/events.py +++ b/backend/app/core/events.py @@ -47,6 +47,10 @@ class EventType(StrEnum): HUMAN_REQUESTED = "human.requested" HUMAN_RESOLVED = "human.resolved" + # 出具与口径治理 + ISSUANCE = "issuance" # 出具契约的判定结果（档位 / 缺数据 / 未回指数字） + CALIBER_UPGRADE = "caliber.upgrade" # 钉住的方法卡有新版本，按声明的策略处置 + # 其他 LOG = "log" USAGE = "usage" # token / 成本累计 diff --git a/backend/app/db/base.py b/backend/app/db/base.py index b7edfdd9..42707ef8 100644 --- a/backend/app/db/base.py +++ b/backend/app/db/base.py @@ -62,8 +62,34 @@ async def get_session() -> AsyncIterator[AsyncSession]: yield session +# 轻量迁移：create_all 只建新表，不给已有表加列。SQLite 的 ADD COLUMN +# 便宜且带默认值回填，够用；等真需要改列类型再考虑 alembic。 +_COLUMN_MIGRATIONS: list[tuple[str, str, str]] = [ + ("runs", "run_class", "TEXT DEFAULT 'exploratory'"), + ("runs", "version", "INTEGER"), + ("runs", "version_hash", "VARCHAR(64)"), + ("runs", "manifest_hash", "VARCHAR(64)"), + ("runs", "started_by", "VARCHAR(100)"), + ("workflows", "status", "TEXT DEFAULT 'draft'"), + ("workflows", "published_version", "INTEGER"), + ("workflows", "published_by", "VARCHAR(100)"), + ("workflow_versions", "graph_hash", "VARCHAR(64)"), + ("approvals", "resolved_by", "VARCHAR(100)"), +] + + +def _migrate(conn) -> None: # pragma: no cover - 同步回调 + from sqlalchemy import text + + for table, column, ddl in _COLUMN_MIGRATIONS: + cols = {row[1] for row in conn.execute(text(f"PRAGMA table_info({table})"))} + if cols and column not in cols: + conn.execute(text(f"ALTER TABLE {table} ADD COLUMN {column} {ddl}")) + + async def init_db() -> None: from app.db import models # noqa: F401 确保模型已注册到 metadata async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) + await conn.run_sync(_migrate) diff --git a/backend/app/db/models.py b/backend/app/db/models.py index 114869f0..20893b68 100644 --- a/backend/app/db/models.py +++ b/backend/app/db/models.py @@ -84,6 +84,11 @@ class Workflow(Base, TimestampMixin): tags: Mapped[list[Any]] = mapped_column(default=list) version: Mapped[int] = mapped_column(Integer, default=1) is_template: Mapped[bool] = mapped_column(default=False) + # 治理状态机：draft（随便改）→ published（有正式版本可跑）→ governed（发布需过治理 lint） + status: Mapped[str] = mapped_column(String(20), default="draft") + # 正式运行默认解析到这个版本；为空表示还没发布过 + published_version: Mapped[int | None] = mapped_column(Integer, default=None) + published_by: Mapped[str | None] = mapped_column(String(100), default=None) versions: Mapped[list["WorkflowVersion"]] = relationship( back_populates="workflow", cascade="all, delete-orphan", lazy="selectin" @@ -101,6 +106,8 @@ class WorkflowVersion(Base, TimestampMixin): version: Mapped[int] = mapped_column(Integer) graph: Mapped[dict[str, Any]] = mapped_column(default=dict) note: Mapped[str] = mapped_column(Text, default="") + # 图内容的 sha256 指纹（剔除 viewport）。formal run 钉住的就是它。 + graph_hash: Mapped[str | None] = mapped_column(String(64), default=None) workflow: Mapped[Workflow] = relationship(back_populates="versions") @@ -130,6 +137,14 @@ class Run(Base, TimestampMixin): started_at: Mapped[datetime | None] = mapped_column(default=None) finished_at: Mapped[datetime | None] = mapped_column(default=None) last_seq: Mapped[int] = mapped_column(Integer, default=0) + # formal：从不可变的 WorkflowVersion 发起，可复现、可出具 + # exploratory：画布试跑 / 裸 graph / 追问，结果只标探索性 + run_class: Mapped[str] = mapped_column(String(20), default="exploratory") + version: Mapped[int | None] = mapped_column(Integer, default=None) + version_hash: Mapped[str | None] = mapped_column(String(64), default=None) + # 结束时对全部事件计算的清单哈希，事后改动事件流会被它戳穿 + manifest_hash: Mapped[str | None] = mapped_column(String(64), default=None) + started_by: Mapped[str | None] = mapped_column(String(100), default=None) events: Mapped[list["RunEvent"]] = relationship( back_populates="run", cascade="all, delete-orphan" @@ -169,6 +184,27 @@ class Approval(Base, TimestampMixin): status: Mapped[str] = mapped_column(String(20), default="pending", index=True) response: Mapped[dict[str, Any]] = mapped_column(default=dict) resolved_at: Mapped[datetime | None] = mapped_column(default=None) + # 责任归属：谁批的。没有 actor 的治理是空转。 + resolved_by: Mapped[str | None] = mapped_column(String(100), default=None) + + +class Artifact(Base, TimestampMixin): + """工件引用表。 + + 文件本体按内容哈希存在 data/artifacts/ 下（同内容只一份）； + 这张表记录"哪个 run 的哪个节点产出了它"——溯源需要的正是这层归属。 + 复合主键：同一工件被多个 run 产出时各记一行引用。 + """ + + __tablename__ = "artifacts" + __table_args__ = (Index("ix_artifacts_run", "run_id"),) + + id: Mapped[str] = mapped_column(String(64), primary_key=True) + run_id: Mapped[str] = mapped_column(String(32), primary_key=True, default="") + kind: Mapped[str] = mapped_column(String(32), default="node_output") # node_output | tool_snapshot | metric_set + node_id: Mapped[str | None] = mapped_column(String(64), default=None) + size: Mapped[int] = mapped_column(Integer, default=0) + meta: Mapped[dict[str, Any]] = mapped_column(default=dict) # -------------------------------------------------------------------------- diff --git a/backend/app/engine/compiler.py b/backend/app/engine/compiler.py index 55aa9898..b1b2ce4f 100644 --- a/backend/app/engine/compiler.py +++ b/backend/app/engine/compiler.py @@ -71,11 +71,33 @@ def _wrap(node: GraphNode, run_ctx: RunContext) -> Callable[[GraphState], Awaita try: updates = await runner(state, ctx) elapsed = int((time.perf_counter() - started) * 1000) + + # 完整输出落工件库。事件里只放截断预览是给画布看的； + # 出具溯源、周对比、轨迹提模板需要的是这份完整证据。 + artifact_id: str | None = None + payload = (updates.get("nodes") or {}).get(node.id) + if payload is not None: + from app.core.artifact_store import put_json + + try: + artifact_id = await put_json( + payload, + kind="node_output", + run_id=run_ctx.run_id, + node_id=node.id, + meta={"type": str(node.type), "attempt": attempt + 1}, + ) + if isinstance(payload, dict): + payload["__artifact__"] = artifact_id + except Exception: # noqa: BLE001 - 工件写失败不该毁掉运行本身 + artifact_id = None + ctx.emit( EventType.NODE_FINISHED, duration_ms=elapsed, attempt=attempt + 1, preview=_preview(updates, node.id), + artifact=artifact_id, ) trail = { "node_id": node.id, diff --git a/backend/app/engine/nodes/llm.py b/backend/app/engine/nodes/llm.py index 4197e2d8..5c39809b 100644 --- a/backend/app/engine/nodes/llm.py +++ b/backend/app/engine/nodes/llm.py @@ -326,12 +326,24 @@ async def run_agent(state: GraphState, ctx: NodeContext) -> dict[str, Any]: content = f"工具执行失败：{type(e).__name__}: {e}" ok = False elapsed = int((time.perf_counter() - started) * 1000) + snapshot_id: str | None = None + if ok: + from app.core.artifact_store import put_json + + try: + snapshot_id = await put_json( + {"tool": name, "args": args, "result": content}, + kind="tool_snapshot", run_id=ctx.run.run_id, node_id=ctx.node.id, + ) + except Exception: # noqa: BLE001 + snapshot_id = None ctx.emit( EventType.TOOL_END if ok else EventType.TOOL_ERROR, tool=name, call_id=call_id, duration_ms=elapsed, preview=content[:2000], + artifact=snapshot_id, ) transcript.append( {"tool": name, "args": args, "ok": ok, "duration_ms": elapsed, "result": content[:4000]} diff --git a/backend/app/engine/nodes/tools.py b/backend/app/engine/nodes/tools.py index 72ff00c1..151f7cef 100644 --- a/backend/app/engine/nodes/tools.py +++ b/backend/app/engine/nodes/tools.py @@ -67,8 +67,19 @@ async def run_tool(state: GraphState, ctx: NodeContext) -> dict[str, Any]: result = {"error": f"{type(e).__name__}: {e}"} elapsed = int((time.perf_counter() - started) * 1000) + # 取数快照：query（args）和结果集一起进工件库，正式出具时数字回指的就是它 + from app.core.artifact_store import put_json + + try: + snapshot_id = await put_json( + {"tool": name, "args": args, "result": result}, + kind="tool_snapshot", run_id=ctx.run.run_id, node_id=ctx.node.id, + ) + except Exception: # noqa: BLE001 + snapshot_id = None ctx.emit(EventType.TOOL_END, tool=name, duration_ms=elapsed, - preview=json.dumps(result, ensure_ascii=False, default=str)[:2000]) + preview=json.dumps(result, ensure_ascii=False, default=str)[:2000], + artifact=snapshot_id) updates: dict[str, Any] = {"nodes": {ctx.node.id: result}} var_name = ctx.cfg("assign_to", "") diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index e849d992..56826db1 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -100,6 +100,13 @@ class RunManager: ) await session.commit() + async def note( + self, run_id: str, event_type: EventType | str, *, node_id: str | None = None, + **data: Any, + ) -> None: + """给 API 层用的事件入口（比如 formal 启动时记录口径升版处置）。""" + await self._emit(run_id, event_type, node_id=node_id, data=data) + # ---------------- 启动 ---------------- async def start( @@ -111,6 +118,10 @@ class RunManager: workflow_name: str = "", memory_scope: str = "default", collection: str = "default", + run_class: str = "exploratory", + version: int | None = None, + version_hash: str | None = None, + started_by: str | None = None, ) -> Run: spec = GraphSpec.model_validate(graph) report = validate_graph(spec) @@ -126,6 +137,10 @@ class RunManager: status="queued", graph=graph, input=input_payload, + run_class=run_class, + version=version, + version_hash=version_hash, + started_by=started_by, ) run.thread_id = run.id # 一个 run 一条 checkpoint 线程 session.add(run) @@ -343,6 +358,16 @@ class RunManager: run.usage = usage if status != "interrupted": run.finished_at = datetime.now(timezone.utc) + # 终态时对全部已落库事件计算清单哈希：事后任何对事件流的 + # 增删改都会与这个值对不上。中断态不算——事件还会继续追加。 + from app.core.artifact_store import manifest_hash as _manifest + + rows = await session.execute( + select(RunEvent.seq, RunEvent.type, RunEvent.node_id, RunEvent.data) + .where(RunEvent.run_id == run_id) + .order_by(RunEvent.seq) + ) + run.manifest_hash = _manifest([tuple(r) for r in rows]) await session.commit() if status == "succeeded":