diff --git a/backend/app/api/datasources.py b/backend/app/api/datasources.py index e91a0977..bea73628 100644 --- a/backend/app/api/datasources.py +++ b/backend/app/api/datasources.py @@ -22,6 +22,7 @@ from app.api.runs import actor_of from app.core.errors import AUTH, NETWORK, TIMEOUT, classify, explain, raw from app.core.config import settings from app.core.crypto import encrypt, mask +from app.data import catalog as data_catalog from app.data import introspect as introspect_mod from app.data import table_versions from app.data.engine import ( @@ -805,10 +806,13 @@ async def get_schema(source_id: str, table: str | None = None, raise HTTPException(409, str(e)) from e if table: meta = introspect_mod.find_table(row, table) + # 和 db_schema 工具同一个说法：有数据目录的表，字段清单后面附上目录 + entries = await data_catalog.read_catalog(session, found.id) return { "table": table, # 给模型看的那段文本照旧；界面画表格用下面的结构化字段 - "detail": introspect_mod.describe_table(row, table), + "detail": introspect_mod.describe_table(row, table, + catalog={n: e.notes for n, e in entries.items()}), "found": meta is not None, "qualified": meta.get("qualified", table) if meta else None, "kind": ("view" if meta.get("is_view") else "table") if meta else None, diff --git a/backend/app/data/introspect.py b/backend/app/data/introspect.py index 6824a99b..27ab77e1 100644 --- a/backend/app/data/introspect.py +++ b/backend/app/data/introspect.py @@ -14,6 +14,7 @@ """ from __future__ import annotations +from collections.abc import Mapping from datetime import datetime, timezone from typing import Any @@ -421,8 +422,14 @@ def table_columns(meta: dict[str, Any]) -> list[dict[str, Any]]: ] -def describe_table(source: Any, table: str) -> str: - """单表的完整字段说明。agent 调 db_schema 工具时返回这个。""" +def describe_table(source: Any, table: str, *, catalog: Mapping[str, Mapping[str, Any]] | None = None) -> str: + """单表的完整字段说明。agent 调 db_schema 工具时返回这个。 + + catalog 是这个源的数据目录 {表名: notes}（data/catalog.py），给了且这张表有可给模型看的内容时，在字段 + 清单后面附上目录：中文名、粒度、业务主键、业务日期、有效记录条件、每列的含义 / 单位 / 度量类型 / 码值、 + 关联关系，推断的项标出来。导入表格的源，表和列的说明是系统按核对结果生成的：说明照旧拼在表名和列后面， + 目录只追加说明没有覆盖的、或人工确认过的项（catalog.render_table_notes）。不给 catalog 时输出和以前一样。 + """ cache = source.schema_cache or {} tables = cache.get("tables") or {} # 模型可能传全名（ANALYTICS.V_TRIP_FACT），也可能只传表名，两种都认 @@ -457,6 +464,14 @@ def describe_table(source: Any, table: str) -> str: marks.append(col["comment"]) suffix = f" {'、'.join(marks)}" if marks else "" lines.append(f" {col['name']} {col['type']}{suffix}") + if catalog: + from app.data import catalog as catalog_mod + + key = catalog_mod.resolve_table_name(cache, table) + block = catalog_mod.render_table_notes(catalog.get(key) if key else None, meta=meta, + system_notes=catalog_mod.system_notes_source(source)) + if block: + lines.append(block) return "\n".join(lines) diff --git a/backend/app/tools/datasource.py b/backend/app/tools/datasource.py index 217fbf63..bd2350a7 100644 --- a/backend/app/tools/datasource.py +++ b/backend/app/tools/datasource.py @@ -21,6 +21,7 @@ import inspect import json import logging import weakref +from collections.abc import Mapping from pathlib import Path from typing import Any @@ -30,6 +31,7 @@ from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.events import EventType +from app.data import catalog as data_catalog from app.data import introspect, table_versions from app.data.engine import SnapshotTampered, masked_columns, query_timeout, run_query from app.data.guard import QueryLimits, SqlRejected, is_write @@ -154,6 +156,8 @@ RECIPE_QUERY_HINT = " 中文表名和列名请加双引号。查单个值时， #: 封存事件 → schema_snapshot → 快照清单 → 各期导入清单 → 原件或清除记录，每一跳都按内容哈希 _FROZEN_SCHEMA_KEYS = ("schema", "synced_at", "truncated", "total", "import_mode", "import_manifests", "snapshot_manifest") +#: 表结构快照里放数据目录的顶层键（_store_schema）。不来自 schema_cache：目录存在 catalog_notes 表里 +CATALOG_SNAPSHOT_KEY = "catalog" def _query_description(source: Any) -> str: @@ -218,14 +222,27 @@ async def build_datasource_tools( except table_versions.SnapshotMissing as e: tools.extend(_unavailable_tool(row, prefix, str(e)) for prefix in prefixes) continue + entries = await _catalog_of(session, row) for prefix in prefixes: if prefix == QUERY_PREFIX: - tools.append(_make_query_tool(source, ctx, fixed=fixed)) + tools.append(_make_query_tool(source, ctx, fixed=fixed, catalog=entries)) else: - tools.append(_make_schema_tool(source, fixed=fixed)) + tools.append(_make_schema_tool(source, fixed=fixed, catalog=entries)) return tools +async def _catalog_of(session: AsyncSession, row: DataSource) -> dict[str, data_catalog.CatalogEntry]: + """这个源的数据目录，建工具时读一次：db_schema 给模型看的和查询时冻结进表结构快照的是同一版。 + + 读不到（库还没建这张表之类）就当没有目录，工具照常可用——目录是附加说明，不能挡住查询。 + """ + try: + return await data_catalog.read_catalog(session, row.id) + except Exception: # noqa: BLE001 + logger.exception("读取数据源 %s 的数据目录失败，本次不附目录", row.name) + return {} + + # -------------------------------------------------------------------------- # 上传表格的版本 # -------------------------------------------------------------------------- @@ -414,10 +431,12 @@ def _unavailable_tool(row: DataSource, prefix: str, reason: str) -> StructuredTo # -------------------------------------------------------------------------- -def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False) -> StructuredTool: +def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False, + catalog: Mapping[str, data_catalog.CatalogEntry] | None = None) -> StructuredTool: """source 是手工源的 DataSource，或上传源绑定快照的 SourceView（字段同名）。 fixed：上传源的版本是这次运行固定的（找不到时的说法不同）。 + catalog：建工具时读到的数据目录，随表结构快照冻结（_store_schema）。 """ # 数据源在 options.query_timeout_s 里配了就用它，否则取缺省。数据库按它停下语句， # 引擎按它告诉界面上限是多少（metadata["timeout_s"]） @@ -449,7 +468,7 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False) -> S payload["source"] = source.name # 表结构快照：这次查询时数据源的结构冻结下来，报告里写的表名、字段名按它核对。事后数据源 # 重新探查、改了结构，已经跑完的运行核对的还是当时那一份。上传源冻结的是绑定快照里那份 - if schema := await _store_schema(source, ctx): + if schema := await _store_schema(source, ctx, catalog=catalog): payload["schema_artifact"] = schema # 查询快照进工件库：出具体系的数字回指要能下钻到"这个数是哪条 SQL 查出来的" try: @@ -465,9 +484,12 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False) -> S extra["mask_columns"] = mask if version := getattr(source, "snapshot_id", None): extra["data_version"] = version + # 工件引用的 meta 记下源和 SQL 里用到的表：数据目录按它数每张表被运行查询过几次 + # （catalog.table_usage），不用再逐个读工件内容。meta 不进内容哈希，快照本身一字不变 payload["artifact"] = await put_json( - {**payload, **extra} if extra else payload, kind="query_snapshot", + {**payload, **extra} if extra else payload, kind=data_catalog.QUERY_SNAPSHOT_KIND, run_id=ctx.run_id or "", node_id=ctx.node_id or "", + meta=data_catalog.query_snapshot_meta(source, str(payload.get("sql") or sql)), ) except Exception: # noqa: BLE001 - 存不下不影响查询本身 pass @@ -485,7 +507,8 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False) -> S ) -async def _store_schema(source: Any, ctx: ToolContext) -> str | None: +async def _store_schema(source: Any, ctx: ToolContext, *, + catalog: Mapping[str, data_catalog.CatalogEntry] | None = None) -> str | None: """把数据源此刻的 schema_cache 存成 schema_snapshot 工件，返回工件 id。 上传源传进来的是绑定快照的视图，schema_cache 就是快照里冻结的那份：查的是哪一版，冻结的 @@ -493,6 +516,10 @@ async def _store_schema(source: Any, ctx: ToolContext) -> str | None: 内容寻址：结构没变的话，同一次运行里查多少次都是同一件，文件只有一份。没探查过结构 （或者探查失败、一张表都没有）返回 None：没有东西可以冻结，报告也就不核对表名。 + + 数据目录一起冻结，放在顶层键 catalog 下：{表名: {version, notes（去掉驳回项）}}，只收这份结构里有的、 + 有目录的表（catalog.frozen_catalog）。这次运行的模型看到的是哪一版目录，事后按快照就能回溯。已有字段 + 一个不动；没有目录的源不加这个键，快照和以前一字不差；目录不变（版本和内容都不变）则快照不变。 """ cache = source.schema_cache or {} tables = cache.get("tables") @@ -503,15 +530,25 @@ async def _store_schema(source: Any, ctx: ToolContext) -> str | None: content = {"source": source.name, "tables": tables, **{k: cache[k] for k in _FROZEN_SCHEMA_KEYS if k in cache}} + meta: dict[str, Any] = {"source": source.name, "tables": len(tables)} + if frozen := data_catalog.frozen_catalog(catalog or {}, tables): + content[CATALOG_SNAPSHOT_KEY] = frozen + meta["catalog_tables"] = len(frozen) try: return await put_json(content, kind=SCHEMA_SNAPSHOT, run_id=ctx.run_id or "", node_id=ctx.node_id or "", - meta={"source": source.name, "tables": len(tables)}) + meta=meta) except Exception: # noqa: BLE001 - 存不下不影响查询本身，只是这次报告不核对表名 return None -def _make_schema_tool(source: Any, *, fixed: bool = False) -> StructuredTool: - """source 同 _make_query_tool：上传源的表结构取绑定快照里冻结的那份。""" +def _make_schema_tool(source: Any, *, fixed: bool = False, + catalog: Mapping[str, data_catalog.CatalogEntry] | None = None) -> StructuredTool: + """source 同 _make_query_tool：上传源的表结构取绑定快照里冻结的那份。 + + catalog：建工具时读到的数据目录，查单表时附在字段清单后面（introspect.describe_table）。 + """ + notes = {name: entry.notes for name, entry in (catalog or {}).items()} + async def _run(table: str | list[str] | None = None) -> str: if gone := _vanished(source, fixed): return f"{DATA_UNAVAILABLE}{gone}" @@ -520,7 +557,7 @@ def _make_schema_tool(source: Any, *, fixed: bool = False) -> StructuredTool: # 而模型两种都会写。为此拒绝一次调用，纯属浪费一步 wanted = [t.strip() for one in wanted for t in str(one).split(",") if t.strip()] if wanted: - return "\n\n".join(introspect.describe_table(source, t) for t in wanted) + return "\n\n".join(introspect.describe_table(source, t, catalog=notes) for t in wanted) tables = introspect.table_names(source) if not tables: # 把实情交出去，而不是让它去点一个它点不到的按钮 diff --git a/backend/tests/test_catalog_api.py b/backend/tests/test_catalog_api.py index f0f1080b..a66bb1d8 100644 --- a/backend/tests/test_catalog_api.py +++ b/backend/tests/test_catalog_api.py @@ -187,3 +187,13 @@ async def test_draft_without_schema_is_refused(client): r = await client.post(f"{_base(sid)}/draft", json={}) assert r.status_code == 409 and "探查结构" in r.json()["detail"] assert (await client.get(_base(sid))).json()["tables"] == [] + + +async def test_schema_endpoint_detail_carries_catalog(client, source_id): + """数据源页「表结构」里给模型看的那段文本和 db_schema 工具同一个说法，也附上目录。""" + before = (await client.get(f"/api/datasources/{source_id}/schema", params={"table": "visits"})).json() + assert catalog.NOTES_HEADING not in before["detail"] + await client.post(f"{_base(source_id)}/draft", json={"tables": ["visits"]}, headers=ACTOR) + after = (await client.get(f"/api/datasources/{source_id}/schema", params={"table": "visits"})).json() + assert catalog.NOTES_HEADING in after["detail"] and "gate_id → gates.id" in after["detail"] + assert after["detail"].startswith(before["detail"]) diff --git a/backend/tests/test_catalog_runtime.py b/backend/tests/test_catalog_runtime.py new file mode 100644 index 00000000..d51d8846 --- /dev/null +++ b/backend/tests/test_catalog_runtime.py @@ -0,0 +1,185 @@ +"""数据目录在运行时：db_schema 工具输出附上目录，查询时目录随表结构快照冻结进证据，查询快照记下用到的表。 + +- describe_table（db_schema 工具的输出）附上这张表的目录；推断的项标「推断，未确认」，驳回的不出现。 + 导入表格的源保留系统生成的说明，目录只追加说明没有覆盖的、或人工确认过的项。 +- 表结构快照新增顶层键 catalog：{表: {version, notes（去掉驳回项）}}。没有目录的源一个字都不加； + 目录不变则快照不变（内容寻址，哈希相同）。 +""" +from __future__ import annotations + +import json +import uuid +from types import SimpleNamespace + +from app.core.artifact_store import canonical_json, content_hash, load +from app.data import catalog +from app.data.catalog import make_item +from app.data.engine import engines +from app.data.introspect import describe_table, introspect +from app.db.base import SessionLocal +from app.db.models import Artifact, DataSource +from app.tools.datasource import _store_schema, build_datasource_tools +from app.tools.registry import ToolContext + +MARK = "（推断，未确认）" + + +def _rel(table, columns, to_table, to_columns, source="fk", status=None): + return {"id": catalog.relation_id(table, columns, to_table, to_columns), "columns": columns, "to_table": to_table, + "to_columns": to_columns, "cardinality": "many_to_one", "coverage": None, "source": source, + "status": status or catalog.initial_status(source)} + + +CACHE = { + "schema": "main", "synced_at": "2026-10-01T00:00:00+00:00", "truncated": False, "total": 2, + "tables": { + "visits": {"qualified": "visits", "schema": None, "is_view": False, "comment": None, "primary_key": ["id"], + "columns": [{"name": "id", "type": "INTEGER", "nullable": False, "comment": None}, + {"name": "park_id", "type": "INTEGER", "nullable": False, "comment": None}, + {"name": "visitor_count", "type": "INTEGER", "nullable": False, "comment": None}, + {"name": "status", "type": "INTEGER", "nullable": False, "comment": None}]}, + "parks": {"qualified": "parks", "schema": None, "is_view": False, "comment": None, "primary_key": ["id"], + "columns": [{"name": "id", "type": "INTEGER", "nullable": False, "comment": None}]}, + }, +} + +NOTES = { + "label": make_item("入园记录", "human"), + "grain": make_item("每张门票每次检票一行", "llm"), + "keys": make_item(["id"], "human"), + "business_date": make_item({"column": "visit_time", "rule": "按检票时间计"}, "human"), + "valid_filter": make_item("status = 1", "llm"), + "dedup": make_item("按票号去重", "llm", "rejected"), + "columns": {"visitor_count": {"meaning": make_item("这次检票进园的人数", "llm"), "unit": make_item("人", "human"), + "measure": make_item("flow", "llm")}, + "status": {"codes": make_item({"1": "有效", "0": "作废"}, "human")}}, + "relations": [_rel("visits", ["park_id"], "parks", ["id"])], +} + + +def _source(cache=CACHE, **kw) -> SimpleNamespace: + return SimpleNamespace(id=kw.pop("id", "src-x"), name=kw.pop("name", "scenic"), kind="sqlite", options={}, + readonly=True, description="", schema_cache=cache, origin=kw.pop("origin", "manual"), **kw) + + +# ---------------------------------------------------------------- describe_table + + +def test_describe_table_appends_catalog(): + text = describe_table(_source(), "visits", catalog={"visits": NOTES}) + head, _, block = text.partition(catalog.NOTES_HEADING) + assert head.startswith("表 visits") and " visitor_count INTEGER" in head # 原来的列清单照旧 + assert "中文名：入园记录" in block and f"粒度：每张门票每次检票一行{MARK}" in block + assert "业务主键：id" in block and "业务日期：visit_time；规则：按检票时间计" in block + assert f"有效记录条件：status = 1{MARK}" in block + assert f"visitor_count：这次检票进园的人数{MARK}；单位：人；度量类型：流量，可跨期加总{MARK}" in block + assert "status：码值：1=有效、0=作废" in block + assert "park_id → parks.id（多对一）" in block + assert "按票号去重" not in text + + +def test_describe_table_without_catalog_is_unchanged(): + plain = describe_table(_source(), "visits") + assert describe_table(_source(), "visits", catalog={}) == plain + assert describe_table(_source(), "visits", catalog={"parks": NOTES}) == plain + assert catalog.NOTES_HEADING not in plain + # 全名、大小写不一致也对得上目录 + assert catalog.NOTES_HEADING in describe_table(_source(), "MAIN.Visits", catalog={"visits": NOTES}) + + +def test_describe_table_keeps_system_notes_of_imported_tables(): + """导入表格：说明是按核对结果生成的。保留说明，目录不重复它、不和它抢话。""" + cache = {"import_mode": "recipe", "tables": {"日报": { + "qualified": "日报", "comment": "粒度：每个日期。不要把各入口相加", "primary_key": [], + "columns": [{"name": "日期", "type": "TEXT", "nullable": False, "comment": "格式 YYYY-MM-DD"}, + {"name": "客流", "type": "INTEGER", "nullable": True, "comment": "单位：人次"}]}}} + notes = {"label": make_item("客流日报", "llm"), "grain": make_item("每天一行", "llm"), + "columns": {"客流": {"unit": make_item("人", "llm"), "measure": make_item("flow", "llm")}, + "日期": {"meaning": make_item("统计日期", "human")}}} + text = describe_table(_source(cache, origin="upload"), "日报", catalog={"日报": notes}) + assert text.startswith("表 日报（粒度：每个日期。不要把各入口相加）") + assert "单位：人次" in text and "格式 YYYY-MM-DD" in text # 系统说明原样保留 + assert "每天一行" not in text # 表说明已写粒度：推断的粒度不追加 + assert "单位：人" + MARK not in text # 列说明已写单位：推断的单位不追加 + assert "客流日报" in text and "流量" in text # 说明没有覆盖的照常追加 + assert "日期：统计日期" in text # 人工确认过的追加 + + +# ---------------------------------------------------------------- 冻结进表结构快照 + + +def _entry(notes, version=1): + return {"visits": catalog.CatalogEntry("visits", notes, version)} + + +async def test_store_schema_freezes_catalog_and_is_content_addressed(): + ctx = SimpleNamespace(run_id="", node_id="") + plain = await _store_schema(_source(), ctx) + before = {"source": "scenic", "tables": CACHE["tables"], + **{k: CACHE[k] for k in ("schema", "synced_at", "truncated", "total")}} + assert plain == content_hash(canonical_json(before)) # 没有目录：和以前一字不差 + assert await _store_schema(_source(), ctx, catalog={}) == plain + + first = await _store_schema(_source(), ctx, catalog=_entry(NOTES, 3)) + frozen = load(first) + assert frozen["catalog"]["visits"]["version"] == 3 + assert "dedup" not in frozen["catalog"]["visits"]["notes"] # 驳回的不冻结 + assert frozen["tables"] == CACHE["tables"] # 已有字段的语义不变 + # 目录不变：同一件工件 + assert await _store_schema(_source(), ctx, catalog=_entry(json.loads(json.dumps(NOTES)), 3)) == first + # 目录变了（升了版本）：换一件 + assert await _store_schema(_source(), ctx, catalog=_entry(NOTES, 4)) != first + # 只有驳回项、或者表不在快照里：不加这个键 + assert await _store_schema(_source(), ctx, catalog={ + "visits": catalog.CatalogEntry("visits", {"dedup": make_item("x", "llm", "rejected")}, 1), + "gone": catalog.CatalogEntry("gone", {"label": make_item("x", "human")}, 1)}) == plain + + +def test_snapshot_readers_ignore_catalog_key(): + """读表结构快照的地方（证据台账的表结构条目、裁判摘录的口径行）只认自己的键，多一个 catalog 不影响。""" + from app.data.provenance import judge_lines + from app.engine.evidence import schema_entry_fields + + tables = {"日报": {"qualified": "日报", "comment": "粒度：每个日期", "primary_key": [], + "columns": [{"name": "客流", "type": "INTEGER", "nullable": True, "comment": "单位：人次"}]}} + base = {"source": "flow", "tables": tables, "schema": "main"} + with_catalog = {**base, "catalog": {"日报": {"version": 2, "notes": {"label": make_item("客流日报", "human")}}}} + assert schema_entry_fields(with_catalog) == schema_entry_fields(base) + query = {"data_version": "snap1", "columns": ["客流"], "sql": "SELECT 客流 FROM 日报"} + lines = judge_lines(query, base, lambda _id: None, tables=["日报"], hidden=set()) + assert lines and judge_lines(query, with_catalog, lambda _id: None, tables=["日报"], hidden=set()) == lines + + +# ---------------------------------------------------------------- 工具里串起来 + + +async def test_tools_show_and_freeze_catalog_and_record_query_tables(scenic_db): + async with SessionLocal() as session: + row = DataSource(name=f"cat_{uuid.uuid4().hex[:10]}", kind="sqlite", database=scenic_db) + session.add(row) + await session.commit() + row.schema_cache = await introspect(row) + await session.commit() + await catalog.write_entry(session, row.id, "visits", {"label": make_item("入园记录", "human"), + "grain": make_item("每张门票每次检票一行", "llm")}, + if_version=0, actor="王敏") + run_id = uuid.uuid4().hex + ctx = ToolContext(run_id=run_id, node_id="n1", data_versions=None) + tools = {t.name: t for t in await build_datasource_tools( + [f"db_query__{row.name}", f"db_schema__{row.name}"], ctx, session)} + try: + shown = await tools[f"db_schema__{row.name}"].coroutine(table="visits") + assert "中文名：入园记录" in shown and f"粒度：每张门票每次检票一行{MARK}" in shown + other = await tools[f"db_schema__{row.name}"].coroutine(table="parks") + assert catalog.NOTES_HEADING not in other + + result = json.loads(await tools[f"db_query__{row.name}"].coroutine( + sql="SELECT v.id FROM visits v JOIN parks p ON p.id = v.park_id LIMIT 3")) + frozen = load(result["schema_artifact"]) + assert frozen["catalog"] == {"visits": {"version": 1, "notes": { + "label": make_item("入园记录", "human"), "grain": make_item("每张门票每次检票一行", "llm")}}} + async with SessionLocal() as session: + ref = await session.get(Artifact, (result["artifact"], run_id)) + assert ref.meta == {"source_id": row.id, "source": row.name, "tables": ["visits", "parks"]} + finally: + await engines.invalidate(row.id)