From 8b2f01d15da004ee1c2ebbe70b6b3d07ff5ec973 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Wed, 4 Nov 2026 08:54:42 +0800 Subject: [PATCH] =?UTF-8?q?=E7=9B=AE=E5=BD=95=E6=8C=89=E6=96=87=E6=A1=A3?= =?UTF-8?q?=E7=82=B9=E5=88=B0=E7=9A=84=E8=A1=A8=E4=BC=98=E5=85=88=E5=B1=95?= =?UTF-8?q?=E5=BC=80=EF=BC=9B=E6=A8=A1=E5=9E=8B=E6=8F=90=E9=97=AE=E4=B8=8D?= =?UTF-8?q?=E5=86=8D=E7=AE=97=E8=B5=B7=E8=8D=89=E5=A4=B1=E8=B4=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 这个数据源的目录渲染出来十万字符，被截到两万四，kpi_info 的 attribute01…54 全在被砍掉的 后半段，模型看不到列就只能如实列问题——界面却把它记成「起草失败」。 现在文本按「概览 → 全量表索引 → 全量列名 → 文档点到的表展开」排，预算不够先丢最后一段， 列名一个不少。模型一条指标都没写、但把缺什么说清楚时，草稿落在 questions_open，问题原样 留下来，不再报失败。 --- backend/app/api/ledger.py | 62 ++++++++++++++++++++--- backend/tests/test_ledger_docdraft_api.py | 25 +++++++++ 2 files changed, 80 insertions(+), 7 deletions(-) diff --git a/backend/app/api/ledger.py b/backend/app/api/ledger.py index 831f2a16..8546e210 100644 --- a/backend/app/api/ledger.py +++ b/backend/app/api/ledger.py @@ -439,11 +439,51 @@ def _source_tables(source: DataSource) -> dict[str, Any]: return tables if isinstance(tables, dict) else {} -async def _schema_material(session: AsyncSession, source: DataSource) -> tuple[str, set[str]]: +def _focus_tables(tables: list[str], text: str, limit: int = 8) -> list[str]: + """文档里点到的表优先展开：按它在原文里第一次出现的位置排，没点到的按目录顺序补几张。 + + 目录动辄十万字符，全塞给模型会把文档和别的表挤掉。先给全量列名（见 _columns_digest），再把文档点到的 + 那几张表展开注释和目录——写 SQL 需要的列名一个不少，细节只给用得上的。 + """ + lowered = (text or "").lower() + hit: list[tuple[int, str]] = [] + for table in tables: + name = str(table).lower() + short = name.split(".")[-1] + at = lowered.find(name) + if at < 0: + at = lowered.find(short) + if at >= 0: + hit.append((at, table)) + hit.sort(key=lambda pair: pair[0]) + picked = [table for _, table in hit][:limit] + for table in tables: + if len(picked) >= min(limit, 5): + break + if table not in picked: + picked.append(table) + return picked + + +def _columns_digest(source: DataSource, tables: list[str]) -> str: + """每张表的列名清单（只给名字）：要写 SQL，先得知道有哪些列。注释和目录留给下面展开的那几张。""" + from app.data import introspect + + lines = ["各表的列（只列名字；下面展开的表另有类型、注释和数据目录）："] + for table in tables: + meta = introspect.find_table(source, table) + columns = [str(c.get("name")) for c in (meta or {}).get("columns") or [] if c.get("name")] + if columns: + lines.append(f"{table}: {', '.join(columns)}") + return "\n".join(lines) + + +async def _schema_material(session: AsyncSession, source: DataSource, focus_text: str = "" + ) -> tuple[str, set[str]]: """这个数据源给模型看的表结构 + 数据目录文本，以及 SQL 允许出现的表名集合。 - 给模型的是「库概览 + 紧凑表名索引 + 前几十张表的字段/目录」（schema_view 的现成渲染），表名一律照目录原文，模型照着抄。 - allowed_tables 收全部探查到的表（含 schema 前缀写法）：文本里被截掉详情的表也允许，因为索引里有名字。 + 文本按「概览 → 全量表索引 → 全量列名 → 文档点到的表展开」的顺序排：预算不够时先丢的是最后那一段， + 列名不会丢。表名一律照目录原文，模型照着抄。 """ from app.data import catalog, schema_view @@ -456,9 +496,8 @@ async def _schema_material(session: AsyncSession, source: DataSource) -> tuple[s lines = [schema_view.overview_line(source, catalog_tables=sum( 1 for entry in entries.values() if catalog.visible_notes(entry.notes)))] lines.append(schema_view.table_index(source, notes)) - for position, table in enumerate(tables): - if position >= 60: - break + lines.append(_columns_digest(source, tables)) + for table in _focus_tables(tables, focus_text): lines.append(schema_view.table_detail(source, table, catalog_notes=notes)) schema = cache.get("schema") allowed: set[str] = set() @@ -642,7 +681,8 @@ async def _extract_in_background(draft_id: str) -> None: draft.status = "failed" await session.commit() return - tables_text, allowed_tables = await _schema_material(session, source) + focus_text = "\n".join(str(c.get("content") or "") for c in chunks.values()) + tables_text, allowed_tables = await _schema_material(session, source, focus_text) if not allowed_tables: draft.error = "这个数据源还没有表结构：先探查表结构，再起草口径" draft.status = "failed" @@ -668,6 +708,14 @@ async def _extract_in_background(draft_id: str) -> None: error = _problems_text(problems) or "模型没有给出可用的取数 SQL 和指标" if result.get("diagnostics"): error = f"{error}（{result['diagnostics']}）" + if plan["open_questions"]: + # 模型把缺什么说清楚了（目录里没有这些列、业务上还没定）：这是一份要补信息的草稿，不是失败 + draft.payload = {**base, "trial": {"run_id": "", "status": "", "error": ""}} + draft.model_id = result["model_id"] or None + draft.error = "" + draft.status = "questions_open" + await session.commit() + return draft.payload = {**base, "trial": {"run_id": "", "status": "failed", "error": error}} draft.model_id = result["model_id"] or None draft.error = error diff --git a/backend/tests/test_ledger_docdraft_api.py b/backend/tests/test_ledger_docdraft_api.py index f90bab31..0c7eb05e 100644 --- a/backend/tests/test_ledger_docdraft_api.py +++ b/backend/tests/test_ledger_docdraft_api.py @@ -507,3 +507,28 @@ async def test_a_real_trial_run_feeds_the_trial_endpoint(client, documents, tmp_ if source is not None: await session.delete(source) await session.commit() + + +async def test_the_model_asking_questions_is_not_a_failure(client, documents, datasource, monkeypatch): + """模型说「目录里没有这些列、写不了 SQL」时，这是要补信息的草稿，不是失败：问题原样留下，不编指标。""" + doc = await documents("dd-questions") + source_id = await datasource() + fake_model(monkeypatch, json.dumps({ + "entry_fields": [], "sql": "", "metrics": [], + "open_questions": ["数据目录里没有 attribute01…54 的取值列，写不出时间类指标"], + }, ensure_ascii=False)) + + body = await extract(client, doc, source_id) + made = (await client.get(f"/api/ledger/drafts/{body['draft_id']}")).json() + assert made["status"] == "questions_open" and made["error"] == "" + assert made["open_questions"] == ["数据目录里没有 attribute01…54 的取值列，写不出时间类指标"] + assert made["metrics"] == [] and made["workflow_id"] is None + assert made["trial"]["run_id"] == "" + + +def test_the_schema_text_expands_the_tables_the_document_names_first(): + """目录十万字符、预算只有两万多：先展开文档点到的表，其余只靠列名清单认。""" + tables = ["orders", "refunds", "stores", "kpi_info", "attr_info"] + picked = api_ledger._focus_tables(tables, "指标取自 kpi_info，属性在 attr_info 里") + assert picked[:2] == ["kpi_info", "attr_info"] + assert set(picked) >= {"kpi_info", "attr_info"}