diff --git a/backend/app/api/datasources.py b/backend/app/api/datasources.py index df49d9f8..aad28a4f 100644 --- a/backend/app/api/datasources.py +++ b/backend/app/api/datasources.py @@ -72,12 +72,20 @@ class DataSourceOut(BaseModel): has_password: bool = False table_count: int = 0 schema_synced_at: str | None = None + #: 上次探查为什么没拿到表。"还没探查"和"探查失败了"是两回事： + #: 前者去点一下按钮就行，后者点了也没用，得先解决超时或者换 schema + schema_error: str = "" + #: 这台服务器上还有哪些库/schema 可选。探不到表时它就是下一步的线索 + available_schemas: list[str] = Field(default_factory=list) tools: list[str] = Field(default_factory=list) def _to_out(row: DataSource) -> DataSourceOut: - tables = (row.schema_cache or {}).get("tables") or {} + cache = row.schema_cache or {} + tables = cache.get("tables") or {} return DataSourceOut( + schema_error=str(cache.get("error") or "") if cache.get("failed") else "", + available_schemas=list(cache.get("available_schemas") or []), id=row.id, name=row.name, kind=row.kind, host=row.host, port=row.port, database=row.database, username=row.username, options=row.options or {}, readonly=row.readonly, description=row.description, enabled=row.enabled, @@ -190,7 +198,11 @@ async def introspect_source( row.schema_cache = await introspect_mod.introspect(row, schema=schema) except Exception as e: # noqa: BLE001 raise HTTPException(400, f"探查失败：{type(e).__name__}: {e}") from e - row.schema_synced_at = datetime.now(timezone.utc) + # 失败也要把缓存存下来——里面的 available_schemas 正是下一步的线索—— + # 但**不能盖上"已同步"的戳**。以前盖了，于是界面显示同步过、工具却说 + # "还没探查过"，两边都不说真话（run 554a0f92） + if not row.schema_cache.get("failed"): + row.schema_synced_at = datetime.now(timezone.utc) await session.commit() await session.refresh(row) return _to_out(row) diff --git a/backend/app/data/introspect.py b/backend/app/data/introspect.py index ad281c8c..0d55d80c 100644 --- a/backend/app/data/introspect.py +++ b/backend/app/data/introspect.py @@ -18,6 +18,7 @@ from datetime import datetime, timezone from typing import Any from sqlalchemy import inspect as sa_inspect +from sqlalchemy.engine.reflection import ObjectKind from app.data.engine import engines @@ -127,22 +128,39 @@ async def introspect(source: Any, *, schema: str | None = None) -> dict[str, Any truncated = len(names) + len(views) > _MAX_TABLES picked = (names + views)[:_MAX_TABLES] + # 批量反射接口（SQLAlchemy 2.0）。 + # + # **别指望它能治超时。** 实测 60 张表：逐表调用 121 次往返、批量 122 次 + # ——SQLAlchemy 的反射缓存本来就把 get_columns / get_pk_constraint 合并 + # 成了同一次底层查询，而 MySQL 方言压根没实现 get_multi_*，落到基类 + # 还是每表一次 SHOW CREATE TABLE。真正能少往返的是 PostgreSQL 这类 + # 原生实现了批量的方言。 + # + # 换成它是因为这是 2.0 的正经接口、少三层 try/except，不是因为它快。 + # shop 那次 "Lost connection to MySQL server during query" + # （run 554a0f92）多半是偶发的服务端超时——所以真正的对策不在这里， + # 在下面：失败要如实记成失败，让工具和界面都说真话。 + def _multi(fn: Any, **kw: Any) -> dict[Any, Any]: + try: + return dict(fn(schema=target_schema, filter_names=picked, + kind=ObjectKind.ANY, **kw)) + except NotImplementedError: + return {} # 表注释这类不是所有方言都支持 + except Exception: # noqa: BLE001 - 取不到不该让整次探查失败 + return {} + + columns_by = _multi(inspector.get_multi_columns) + pk_by = _multi(inspector.get_multi_pk_constraint) + comment_by = _multi(inspector.get_multi_table_comment) + tables: dict[str, Any] = {} for name in picked: - try: - columns = inspector.get_columns(name, schema=target_schema) - except Exception: # noqa: BLE001 - 单表失败不该让整次探查失败 - continue - try: - pk = (inspector.get_pk_constraint(name, schema=target_schema) or {}).get( - "constrained_columns" - ) or [] - except Exception: # noqa: BLE001 - pk = [] - try: - comment = (inspector.get_table_comment(name, schema=target_schema) or {}).get("text") - except Exception: # noqa: BLE001 - 不是所有方言都支持表注释 - comment = None + key = (target_schema, name) + columns = columns_by.get(key) + if not columns: + continue # 这张表反射不出来，跳过，别毁掉整次探查 + pk = (pk_by.get(key) or {}).get("constrained_columns") or [] + comment = (comment_by.get(key) or {}).get("text") tables[name] = { # 写 SQL 要用的全名。没有 schema 前缀时 Oracle 直接 ORA-00942—— @@ -171,11 +189,17 @@ async def introspect(source: Any, *, schema: str | None = None) -> dict[str, Any # schema 名填错是高频事故（大小写、拼写、或者根本不知道该填什么）， # 直接把驱动异常抛出去等于让人自己去猜。先看看连接本身好不好： # 能列出候选说明只是 schema 不对，这种情况给出可选项比报错有用得多。 + # + # 但**这里回来的是一次失败，不是"还没探查"**。以前两者存成了一模一样 + # 的形状（tables 空 + 一个没人读的 error 字段），于是工具对 agent 说 + # "还没有探查过结构，请在设置页里执行一次"——而用户明明执行过，是超时了。 + # run 554a0f92 里 agent 因此花了 6 次调用、29 秒自己重建 schema。 + # failed 这个标记就是为了让下游分得清。 candidates = await list_schemas(source) if not candidates: raise payload = {"tables": {}, "truncated": False, "total": 0, - "error": f"{type(exc).__name__}: {exc}"} + "failed": True, "error": f"{type(exc).__name__}: {exc}"} payload["synced_at"] = datetime.now(timezone.utc).isoformat() payload["schema"] = target @@ -206,6 +230,18 @@ def summary(source: Any, *, max_tables: int = 40, detail: bool = True) -> str: return _detailed_summary(source, max_tables=max_tables) +def why_empty(cache: dict[str, Any]) -> str: + """结构为什么是空的。一句话，给 agent 和用户看的是同一句。 + + 分清"还没探查"和"探查失败了"：前者该去点一下按钮，后者点了也没用， + 要先解决超时或换 schema。以前两者说的是同一句话。 + """ + if not cache.get("failed"): + return "还没有探查过结构" + err = str(cache.get("error") or "").strip() + return f"结构探查失败了：{err[:200]}" if err else "结构探查失败了" + + def _empty_summary(source: Any, cache: dict[str, Any]) -> str: hint = "" candidates = cache.get("available_schemas") or [] @@ -213,7 +249,7 @@ def _empty_summary(source: Any, cache: dict[str, Any]) -> str: hint = f"｜该账号名下没有对象，数据可能在这些 schema：{'、'.join(candidates[:8])}" return ( f"- {source.name}（{source.kind}）：{source.description or '未填说明'}" - f"｜尚未探查结构{hint}" + f"｜{why_empty(cache)}{hint}" ) @@ -255,9 +291,18 @@ def describe_table(source: Any, table: str) -> str: lowered = table.rsplit(".", 1)[-1].lower() meta = next((m for n, m in tables.items() if n.lower() == lowered), None) if not meta: + if not tables: + # 缓存是空的，说"里面没有这张表"就是在误导——真相是我们什么都不知道。 + # run 554a0f92 里 agent 一次问了 7 个表名，收到 7 条一模一样的 + # "没有 X。现有的对象：（还没探查过结构）"，白花一整步 + return ( + f"数据源「{source.name}」的结构信息不可用（{why_empty(cache)}），" + "所以无法判断有没有这张表。请直接查数据字典" + "（information_schema.tables / columns，SQLite 用 sqlite_master）。" + ) available = "、".join( m.get("qualified", n) for n, m in list(tables.items())[:30] - ) or "（还没探查过结构）" + ) return f"数据源「{source.name}」里没有 {table}。现有的对象：{available}" head = f"{'视图' if meta.get('is_view') else '表'} {meta.get('qualified', table)}" diff --git a/backend/app/tools/datasource.py b/backend/app/tools/datasource.py index 3dcc63fc..252b42f4 100644 --- a/backend/app/tools/datasource.py +++ b/backend/app/tools/datasource.py @@ -47,6 +47,34 @@ def tool_names(source: DataSource) -> list[str]: return [f"{QUERY_PREFIX}{source.name}", f"{SCHEMA_PREFIX}{source.name}"] +def _no_schema_note(source: DataSource) -> str: + """结构拿不到时，把实情和下一步一起交给 agent。 + + 以前这里只说"结构尚未探查，请在设置页里执行一次"。两个问题：它可能是错的 + （探查执行过了，是超时失败），而且它给的下一步 agent 根本做不到——agent + 进不了设置页。于是它只能自己摸：run 554a0f92 里先猜了 7 个表名、又发了 + 一条 SELECT 1 试连通性、再用三条 information_schema 把 schema 重建一遍， + 9 次工具调用花掉 6 次、29 秒之后才碰到真正要查的数据。 + + 那条自救路线是对的，只是它自己摸索了三轮才找到。直接告诉它。 + """ + cache = source.schema_cache or {} + note = f" {introspect.why_empty(cache)}。" + candidates = cache.get("available_schemas") or [] + if candidates: + # 这几个候选一直在缓存里躺着，只是以前没人交给 agent + note += ( + f"当前连的库是「{source.database or '未指定'}」，" + f"这台服务器上还有这些库可选：{'、'.join(candidates[:10])}。" + ) + note += ( + "**不要猜表名**——直接查数据字典自己确认，" + "例如 information_schema.tables / information_schema.columns" + "（SQLite 则是 sqlite_master），一条带 WHERE 的查询就能拿到表和字段。" + ) + return note + + def _query_description(source: DataSource) -> str: tables = introspect.table_names(source) head = f"在数据源「{source.name}」上执行 SQL 查询。" @@ -57,9 +85,9 @@ def _query_description(source: DataSource) -> str: listed = "、".join(tables[:25]) more = f" 等 {len(tables)} 张表" if len(tables) > 25 else "" head += f" 可用表：{listed}{more}。" + head += " 字段不确定时先查结构，不要猜字段名。" else: - head += " 结构尚未探查，先用同名的 db_schema 工具看看有哪些表。" - head += " 字段不确定时先查结构，不要猜字段名。" + head += _no_schema_note(source) return head @@ -139,10 +167,8 @@ def _make_schema_tool(source: DataSource) -> StructuredTool: return "\n\n".join(introspect.describe_table(source, t) for t in wanted) tables = introspect.table_names(source) if not tables: - return ( - f"数据源「{source.name}」还没有探查过结构。" - "请在设置页里对它执行一次结构探查。" - ) + # 把实情交出去，而不是让它去点一个它点不到的按钮 + return f"数据源「{source.name}」的结构信息不可用。" + _no_schema_note(source) return f"数据源「{source.name}」共 {len(tables)} 张表：\n" + "\n".join( f" {n}" for n in tables ) diff --git a/backend/tests/test_datasource.py b/backend/tests/test_datasource.py index 901fbf6e..0d332da0 100644 --- a/backend/tests/test_datasource.py +++ b/backend/tests/test_datasource.py @@ -15,6 +15,15 @@ import pytest from app.data.engine import build_url, run_query, test_connection from app.data.guard import QueryLimits, SqlRejected from app.data.introspect import describe_table, introspect, summary, table_names +from app.main import app as _fastapi_app +from httpx import ASGITransport, AsyncClient + + +@pytest.fixture +async def client(): + transport = ASGITransport(app=_fastapi_app) + async with AsyncClient(transport=transport, base_url="http://test") as c: + yield c @pytest.fixture @@ -72,7 +81,7 @@ async def test_summary_is_compact_enough_for_a_prompt(source): async def test_summary_says_so_when_never_introspected(source): """没探查过就明说，别让 Copilot 对着空气编表名。""" - assert "尚未探查" in summary(source) + assert "还没有探查过" in summary(source) async def test_describe_table_accepts_qualified_name(source): @@ -225,3 +234,103 @@ def test_schema_stats_sql_covers_every_supported_kind(): for kind in SUPPORTED_KINDS: assert kind in _SCHEMA_STATS_SQL or kind in ("sqlite", "mysql", "mariadb", "postgres", "postgresql") + + +# -------------------------------------------------------------------------- +# 探查失败 ≠ 还没探查 +# -------------------------------------------------------------------------- +# +# run 554a0f92：shop 的探查因为 "Lost connection to MySQL server during query" +# 失败了，但失败被存成了一个"成功但没表"的缓存，schema_synced_at 照常盖戳。 +# 于是 db_schema 工具对 agent 说"还没有探查过结构，请在设置页里执行一次"—— +# 一句错话，配一个 agent 根本做不到的动作。它只好自己摸：猜 7 个表名、发一条 +# SELECT 1 试连通性、再用三条 information_schema 把 schema 重建一遍。 +# 9 次工具调用花掉 6 次，前 29 秒没碰到一点真数据。 +# +# 而 available_schemas 那 10 个候选一直躺在缓存里，没人交给它。 + + +class _FailedCache: + """探查失败之后 schema_cache 的样子。""" + + name = "shop" + kind = "mysql" + database = "prod_shop" + description = "" + readonly = True + schema_cache = { + "tables": {}, "total": 0, "failed": True, + "error": "OperationalError: (2013, 'Lost connection to MySQL server during query')", + "available_schemas": ["prod_shop", "shop_dev", "toolsdb"], + } + + +class _NeverRun(_FailedCache): + """从来没探查过的样子。和上面必须说不一样的话。""" + + schema_cache: dict = {} + + +def test_why_empty_tells_the_two_cases_apart(): + from app.data.introspect import why_empty + + assert "失败" in why_empty(_FailedCache.schema_cache) + assert "Lost connection" in why_empty(_FailedCache.schema_cache) + assert why_empty({}) == "还没有探查过结构" + + +def test_schema_tool_tells_the_agent_the_truth(): + """给 agent 的那句话要是实情，而且要给一个它做得到的下一步。""" + from app.tools.datasource import _no_schema_note + + note = _no_schema_note(_FailedCache) + assert "失败" in note and "Lost connection" in note + # 候选一直在缓存里，以前没人交出去 + assert "prod_shop" in note and "toolsdb" in note + # agent 进不了设置页。它做得到的是自己查数据字典——那条路它最后也走通了， + # 只是自己摸索了三轮 + assert "information_schema" in note + assert "不要猜表名" in note + # 不能再说"请到设置页" + assert "设置页" not in note + + fresh = _no_schema_note(_NeverRun) + assert "还没有探查过" in fresh + assert "失败" not in fresh + + +def test_describe_table_does_not_claim_the_table_is_missing(): + """缓存是空的时候，说"里面没有这张表"是在误导——真相是我们什么都不知道。""" + from app.data.introspect import describe_table + + out = describe_table(_FailedCache, "user") + assert "无法判断" in out + assert "失败" in out + # 以前是 "数据源「shop」里没有 user。现有的对象：（还没探查过结构）" + assert "现有的对象" not in out + + +@pytest.mark.asyncio +async def test_introspect_failure_does_not_get_a_synced_stamp(client, monkeypatch): + """失败要把缓存留下（候选是下一步的线索），但不能盖"已同步"的戳。 + + 以前盖了，于是界面说同步过、工具说还没探查过，两边都不说真话。 + """ + import app.api.datasources as mod + + async def failed(source, schema=None): + return dict(_FailedCache.schema_cache) + + created = await client.post("/api/datasources", json={ + "name": "introspect_fail", "kind": "sqlite", "database": ":memory:", + }) + assert created.status_code == 201, created.text + sid = created.json()["id"] + + monkeypatch.setattr(mod.introspect_mod, "introspect", failed) + out = (await client.post(f"/api/datasources/{sid}/introspect")).json() + + assert out["schema_synced_at"] is None, "失败不该盖已同步的戳" + assert "Lost connection" in out["schema_error"] + assert out["available_schemas"] == ["prod_shop", "shop_dev", "toolsdb"] + assert out["table_count"] == 0 diff --git a/frontend/src/pages/DataSourcesTab.tsx b/frontend/src/pages/DataSourcesTab.tsx index 81831853..5ebb403e 100644 --- a/frontend/src/pages/DataSourcesTab.tsx +++ b/frontend/src/pages/DataSourcesTab.tsx @@ -234,11 +234,15 @@ function SourceCard({ row, onChanged, onEdit }: { setBusy('introspect') try { const r = await api.datasources.introspect(row.id) - toast(`探到 ${r.table_count} 个对象`, r.table_count ? 'ok' : 'error') - if (!r.table_count) { - // 大概率是 schema 没配对：只读账号名下往往什么都没有 - const s = await api.datasources.schema(row.id) - toast(s.summary?.includes('schema') ? s.summary.slice(0, 120) : '没探到对象，检查 schema 配置', 'error') + if (r.table_count) { + toast(`探到 ${r.table_count} 个对象`, 'ok') + } else if (r.schema_error) { + // 真失败了。以前这里一律说"检查 schema 配置"，而超时和填错 schema + // 要做的事完全不同，卡片上那段红字写了实情，这里指过去 + toast(`探查失败：${r.schema_error.slice(0, 100)}`, 'error') + } else { + // 连得上、也没报错，就是这个 schema 下真的没有对象 + toast('这个库里没有对象，多半是 schema 填的不对——卡片上列了可选的', 'error') } await onChanged() } catch (e: any) { @@ -285,6 +289,19 @@ function SourceCard({ row, onChanged, onEdit }: {
{row.description}
)} + {row.schema_error && ( +
+ 上次探查失败：{row.schema_error} + {!!row.available_schemas?.length && ( +
+ 这台服务器上可选的库：{row.available_schemas.slice(0, 10).join('、')} + ——当前连的是「{row.database || '未指定'}」，不对的话在「编辑」里改 +
+ )} +
+ )} +
{row.kind === 'sqlite' ? row.database : `${row.username ?? ''}@${row.host ?? ''}:${row.port ?? ''}`} @@ -294,6 +311,14 @@ function SourceCard({ row, onChanged, onEdit }: { + ) : row.schema_error ? ( + // "还没探查"和"探查失败了"是两回事：前者点一下按钮就行，后者点了 + // 也没用。以前两者显示成同一句，于是用户点了探查、看到还是"未探查"， + // 而 agent 那边也被告知"还没探查过"——没有一处说的是真话 + + 探查失败 + ) : ( 结构未探查 )}