数据目录 F3 / B4 服务端:没探查到的表说出来——目录清单返回 schema_truncated、schema_total;助手上下文的 total 用表结构缓存里的真实总数,截断时 context 操作带 explored,提示词写明还有几张没探查到、不认识的表名先查结构,挑表请求也写明只列了探查到的

This commit is contained in:
ThearchyHelios
2026-10-04 01:11:03 +08:00
parent 6c30becd94
commit 52aaaaf945
3 changed files with 176 additions and 6 deletions
+12 -1
View File
@@ -87,6 +87,10 @@ def _iso(value: Any) -> str | None:
return value.isoformat() if value is not None else None
def _int(value: Any) -> int:
return value if isinstance(value, int) and not isinstance(value, bool) and value > 0 else 0
def _item_value(notes: dict[str, Any], name: str) -> tuple[Any, str | None]:
item = notes.get(name)
if not isinstance(item, dict) or item.get("status") == "rejected":
@@ -169,7 +173,12 @@ def _invalid(e: catalog.CatalogInvalid) -> HTTPException:
@router.get("/{source_id}/catalog")
async def list_catalog(source_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
"""表清单:表结构里的每张表各一行(没有目录的也列),使用次数多的在前;目录还在、表已不在的排最后。"""
"""表清单:表结构里的每张表各一行(没有目录的也列),使用次数多的在前;目录还在、表已不在的排最后。
探查结构每个源最多取 200 张表(introspect._MAX_TABLES),多出来的不在表结构里、这里也列不出:schema_truncated
说明有没有截断,schema_total 是数据库里一共几张表(含视图),界面据此说明「只探查了前 200 张」。老缓存没记
总数时按探查到的张数算。
"""
_, source = await _resolved(session, source_id)
cache = source.schema_cache or {}
tables = cache.get("tables") or {}
@@ -184,6 +193,8 @@ async def list_catalog(source_id: str, session: AsyncSession = Depends(get_sessi
"system_notes": catalog.system_notes_source(source),
# 没有表结构时说明原因(尚未探查 / 探查失败),界面据此引导去探查
"schema_note": None if tables else introspect.why_empty(cache),
"schema_truncated": bool(cache.get("truncated")),
"schema_total": max(_int(cache.get("total")), len(tables)),
}
+48 -5
View File
@@ -78,6 +78,14 @@ SECTION_HEAD_PROMPT = "\n\n已接入的数据源(涉及取数时优先用它
NAMES_ONLY_PROMPT = " (对象较多,上面只列了名字。写 SQL 前用 db_schema 工具确认字段,别猜)"
INFERRED_PROMPT = (" 标「推断,未确认」的说明和连接条件是按列名、数据库注释或模型推断出来的,没有人确认过:"
"照抄前对照字段核实,拿不准就先用 db_schema 工具查")
#: 探查结构截断了的源(每个源最多取 200 张表,introspect._MAX_TABLES):没探查到的表不在表结构缓存里,上面列不出、
#: db_schema 也查不到。不说的话模型会以为库里就这么多表,用户说的表名对不上时就近找一张名字像的顶上
UNEXPLORED_PROMPT = (" (这个库共 {total} 张表,只探查了 {explored} 张,还有 {rest} 张没有探查到、不在上面。用户提到的表名"
"在上面找不到时,先用 db_schema__{source} 查;也查不到就查数据字典(information_schema.columns,"
"SQLite 用 sqlite_master)确认它的字段,别猜,也别拿名字相近的表顶替)")
#: 挑表请求里一个源的标题:截断了的写明只列了探查到的
PICK_SOURCE_PROMPT = "数据源「{source}」(共 {total} 张表)"
PICK_SOURCE_PARTIAL_PROMPT = "数据源「{source}」(共 {total} 张表,只探查了 {explored} 张,下面只列探查到的)"
SQL_RULES_PROMPT = (
"\n\n写 SQL 的硬性要求:\n"
# 这三条都是真实踩过的:Oracle 上漏 schema 前缀直接 ORA-00942,
@@ -177,6 +185,26 @@ class SourcePlan:
def tables(self) -> dict[str, Any]:
return (getattr(self.source, "schema_cache", None) or {}).get("tables") or {}
@property
def explored(self) -> int:
"""探查到、在表结构缓存里的表数。"""
return len(self.tables)
@property
def total(self) -> int:
"""这个源一共几张表(含视图):探查时数据库报的总数。老缓存没记总数时按探查到的算。"""
raw = (getattr(self.source, "schema_cache", None) or {}).get("total")
return max(raw if isinstance(raw, int) and not isinstance(raw, bool) else 0, self.explored)
@property
def unexplored(self) -> int:
"""探查截断了的话还有几张没探查到;没截断为 0。
只看 truncated:反射失败跳过的表(skipped)也算在 total 里,但那是「探查了、读不到」,不是没探查到。
"""
truncated = (getattr(self.source, "schema_cache", None) or {}).get("truncated")
return max(self.total - self.explored, 0) if truncated else 0
@property
def notes(self) -> dict[str, dict[str, Any]]:
"""给模型看的目录:去掉驳回项。"""
@@ -330,7 +358,9 @@ class ContextPlan:
continue
qualified = _qualifier(p.tables)
index = catalog.render_table_index(p.source.schema_cache, p.notes)
blocks.append(f"数据源「{p.source.name}」(共 {len(p.tables)} 张表):\n{index}")
head = (PICK_SOURCE_PARTIAL_PROMPT if p.unexplored else PICK_SOURCE_PROMPT).format(
source=p.source.name, total=p.total, explored=p.explored)
blocks.append(f"{head}:\n{index}")
if p.seeds:
used.append(f"数据源「{p.source.name}」的 " + "、".join(qualified(t) for t in p.seeds))
human = PICK_PROMPT.format(need=need.strip(), used=PICK_USED_PROMPT.format(tables=";".join(used)) if used
@@ -499,7 +529,13 @@ class SourceContext:
@property
def total(self) -> int:
return len(self.plan.tables)
"""这个源一共几张表:表结构缓存里记的真实总数(探查截断时比探查到的多),见 SourcePlan.total。"""
return self.plan.total
@property
def explored(self) -> int:
"""探查到的表数:没有表结构时为 0。"""
return self.plan.explored
def qualified(self, key: str) -> str:
return _qualifier(self.plan.tables)(key)
@@ -525,15 +561,18 @@ class DatasourceContext:
def op(self) -> dict[str, Any] | None:
"""告诉前端这一轮参考了哪些表:{"op": "context", "sources": [...], "elapsed_ms"?}。没有可查的表时为 None。
每个源一项:source 源名,tables 给了全字段的表(全名),selected_by,total 这个源一共几张表;挑表失败的
每个源一项:source 源名,tables 给了全字段的表(全名),selected_by,total 这个源一共几张表(表结构缓存里
记的真实总数);探查截断了的再带 explored(探查到几张),界面写「共 N 张,只探查了 200 张」;挑表失败的
带上 reason。只列了名字的表不进 tables:界面上说的「参考了 N 张表」是模型看得到字段的那些。
"""
rows = []
for s in self.sources:
if not s.total:
if not s.explored:
continue
row: dict[str, Any] = {"source": s.source.name, "tables": [s.qualified(t) for t in s.tables],
"selected_by": s.selected_by, "total": s.total}
if s.plan.unexplored:
row["explored"] = s.explored
if s.selected_by == "fallback" and self.reason:
row["reason"] = self.reason
rows.append(row)
@@ -564,6 +603,10 @@ class DatasourceContext:
logger.exception("数据源 %s 的上下文组装失败,本轮只列表名", s.source.name)
blocks.append(introspect.summary(s.source, detail=False))
names_only = True
if s.plan.unexplored:
# 探查截断了:上面列的只是探查到的那部分,说清还有多少张没探查到、遇到不认识的表名怎么办
blocks.append(UNEXPLORED_PROMPT.format(total=s.total, explored=s.explored, rest=s.plan.unexplored,
source=s.source.name))
text = "\n".join(blocks)
tail = []
if names_only:
@@ -635,7 +678,7 @@ def _detail_block(s: SourceContext, budget: list[int]) -> str:
目录按表放,放不下的表让模型用 db_schema 查(db_schema 的输出里带目录)。
"""
lines = [introspect.summary(s.source, detail=True)]
if not s.total:
if not s.explored:
return lines[0]
joins = _join_lines(s, _relation_graph(s.plan), list(s.plan.tables))
size = sum(len(line) + 1 for line in joins)
@@ -0,0 +1,116 @@
"""没探查到的表要说出来(数据目录完整性审查 F3 / B4)。
探查结构每个数据源最多取 200 张表(introspect._MAX_TABLES)。多出来的表不在表结构缓存里:目录页列不出,助手也
看不到字段。以前这件事哪儿都没说——目录清单只给探查到的 200 行,助手上下文里的「共几张表」是探查到的张数,
提示词里也不提。这组守的是:
- 目录清单接口返回 schema_truncated、schema_total(数据库里一共几张),界面据此说明「只探查了前 200 张」
- 助手上下文的 total 是表结构缓存里的真实总数;context 操作里带上 explored(探查到几张)
- 提示词里写明还有几张没探查到,遇到不认识的表名先查结构
- 没截断的库照旧,不多说一句
合成库:205 张空表,测试里临时建,用完连同数据源一起删掉。
"""
from __future__ import annotations
import sqlite3
import uuid
import pytest
from httpx import ASGITransport, AsyncClient
from app.db.base import SessionLocal
from app.db.models import DataSource
from app.main import app
from tests.fixtures.sources import drop_source
WIDE = 205
def _build(path, count: int) -> str:
db = sqlite3.connect(path)
for i in range(count):
db.execute(f"CREATE TABLE t{i:03d} (id INTEGER PRIMARY KEY, amount INTEGER)")
db.commit()
db.close()
return str(path)
@pytest.fixture
async def client():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
yield c
async def _source(client, path: str) -> tuple[str, str]:
name = f"wide_{uuid.uuid4().hex[:8]}"
r = await client.post("/api/datasources", json={"name": name, "kind": "sqlite", "database": path,
"readonly": True})
assert r.status_code in (200, 201), r.text
sid = r.json()["id"]
r = await client.post(f"/api/datasources/{sid}/introspect")
assert r.status_code == 200, r.text
return sid, name
@pytest.fixture
async def wide(client, tmp_path):
sid, name = await _source(client, _build(tmp_path / "wide.db", WIDE))
yield sid, name
await drop_source(sid)
@pytest.fixture
async def narrow(client, tmp_path):
sid, name = await _source(client, _build(tmp_path / "narrow.db", 3))
yield sid, name
await drop_source(sid)
async def _row(sid: str) -> DataSource:
async with SessionLocal() as session:
row = await session.get(DataSource, sid)
session.expunge(row)
return row
async def test_catalog_list_says_how_many_tables_were_not_introspected(client, wide):
sid, _ = wide
body = (await client.get(f"/api/datasources/{sid}/catalog")).json()
assert len(body["tables"]) == 200
assert body["schema_truncated"] is True and body["schema_total"] == WIDE
async def test_catalog_list_of_a_complete_schema(client, narrow):
sid, _ = narrow
body = (await client.get(f"/api/datasources/{sid}/catalog")).json()
assert len(body["tables"]) == 3
assert body["schema_truncated"] is False and body["schema_total"] == 3
async def test_assistant_context_counts_the_real_total(wide):
from app.api import copilot_context
sid, name = wide
plan = await copilot_context.plan_context([await _row(sid)])
ctx = plan.fallback("测试")
(row,) = ctx.op()["sources"]
assert row["total"] == WIDE and row["explored"] == 200
text = ctx.section()
assert f"共 {WIDE} 张表" in text and "只探查了 200 张" in text and f"还有 {WIDE - 200} 张没有探查到" in text
# 遇到不认识的表名先查结构,别猜
assert f"db_schema__{name}" in text and "information_schema" in text
# 挑表的请求里也写明只列了探查到的
pick = plan._pick_messages("各表的金额")[1][1]
assert f"共 {WIDE} 张表,只探查了 200 张" in pick
async def test_assistant_context_of_a_complete_schema_says_nothing_extra(narrow):
from app.api import copilot_context
sid, _ = narrow
plan = await copilot_context.plan_context([await _row(sid)])
ctx = await plan.resolve(None, need="x")
(row,) = ctx.op()["sources"]
assert row["total"] == 3 and "explored" not in row
assert "没有探查到" not in ctx.section()