From 8682deb447a132a4b96b8809796164a0c7c7bb7f Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Wed, 4 Nov 2026 08:37:52 +0800 Subject: [PATCH] =?UTF-8?q?=E8=B5=B7=E8=8D=89=E6=8C=89=E3=80=8C=E8=83=BD?= =?UTF-8?q?=E5=85=B1=E7=94=A8=E4=B8=80=E6=9D=A1=20SQL=20=E7=9A=84=E4=B8=80?= =?UTF-8?q?=E7=BB=84=E3=80=8D=E8=BF=9B=E8=A1=8C=EF=BC=8C=E5=B9=B6=E6=8A=A2?= =?UTF-8?q?=E6=95=91=E8=A2=AB=E6=88=AA=E6=96=AD=E7=9A=84=E5=9B=9E=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 实测：一条 SQL 枚举十几列 CASE WHEN 就把输出额度吃光，文档里十几条指标一条都写不出来， 界面只报「没有一条指标通过校验」。现在提示模型一次只起草同组指标（最多 6 条）、SQL 写短， 其余的进 uncovered；回复被截断时按完整片段抢救指标和 SQL，失败原因里写清模型实际交回了 什么形状，不再只说没有指标。 --- backend/app/api/ledger.py | 4 ++ backend/app/ledger/docdraft.py | 6 ++- backend/app/ledger/draftplan.py | 67 ++++++++++++++++++++++++-- backend/tests/test_ledger_draftplan.py | 27 +++++++++++ 4 files changed, 100 insertions(+), 4 deletions(-) diff --git a/backend/app/api/ledger.py b/backend/app/api/ledger.py index 6e9c0f70..831f2a16 100644 --- a/backend/app/api/ledger.py +++ b/backend/app/api/ledger.py @@ -661,9 +661,13 @@ async def _extract_in_background(draft_id: str) -> None: "sql": plan["sql"], "version": 1, "trial": {"run_id": "", "status": "running", "error": ""}, "dropped": problems, "truncated": result["truncated"], "model_id": result["model_id"], + "open_questions": plan["open_questions"], "uncovered": result.get("uncovered") or [], + "salvaged": bool(result.get("salvaged")), } if not plan["sql"] or not plan["metrics"]: error = _problems_text(problems) or "模型没有给出可用的取数 SQL 和指标" + if result.get("diagnostics"): + error = f"{error}（{result['diagnostics']}）" 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/app/ledger/docdraft.py b/backend/app/ledger/docdraft.py index 3c5879dd..711d150d 100644 --- a/backend/app/ledger/docdraft.py +++ b/backend/app/ledger/docdraft.py @@ -579,8 +579,12 @@ def merge_definition(card_metrics: list[dict], draft_metrics: list[dict]) -> tup return merged, warnings +#: 被截断的回复里抢回已经完整的指标对象（draftplan 生成取数草稿时也要用） +salvage_metrics = _salvage_metrics + + __all__ = [ "DEFINITION_FIELDS", "DraftUnavailable", "FORBIDDEN_FIELDS", "MAX_INPUT_CHARS", "METRIC_FIELDS", "MIN_QUOTE_CHARS", "STATUSES", "alignment", "extract", "merge_definition", "normalize_metrics", - "parse_reply", "prompt", "verify_quote", + "parse_reply", "prompt", "salvage_metrics", "verify_quote", ] diff --git a/backend/app/ledger/draftplan.py b/backend/app/ledger/draftplan.py index b0e49731..b11ff5e5 100644 --- a/backend/app/ledger/draftplan.py +++ b/backend/app/ledger/draftplan.py @@ -23,6 +23,7 @@ build_graph），单行结果取第 0 行的某一列；写成别的样子，卡 """ from __future__ import annotations +import json import re from collections.abc import Mapping from typing import Any @@ -59,6 +60,11 @@ _SYSTEM = """你是业务口径与取数模板的起草助手。读给定的业 5. 每条指标的定义必须带 citations：引用逐字抄原文，chunk 写材料里方括号内的 id。核对不上、或原文里没有依据的定义 不要写进指标，写进 open_questions。 6. 拿不准的不要猜，写进 open_questions，用一句问得清的话。不要编造文档里没有的数字、单位、生效日期。 +7. **一次只起草能共用同一条 SQL 的一组指标，最多 6 条。** 文档里其余的指标不要硬凑进这一条 SQL，把名称写进 + uncovered（字符串数组），下一轮再起草。不同粒度、不同表、不同时间字段的指标本来就不该共用一条 SQL。 +8. **sql 要短。** 能用 GROUP BY 和聚合表达的，不要为每个属性枚举一长串 CASE WHEN 列；输出额度有限，SQL 写长了， + 后面的指标就写不出来了。 +9. 写不下时宁可少写几条指标、把剩下的写进 uncovered，**也不要让 JSON 半截断掉**。 输出格式： { @@ -85,7 +91,8 @@ _SYSTEM = """你是业务口径与取数模板的起草助手。读给定的业 "open_questions": ["还没说清、需要业务确认的问题"] } ], - "open_questions": ["跨指标、以及整条 SQL 还没把握的问题"] + "open_questions": ["跨指标、以及整条 SQL 还没把握的问题"], + "uncovered": ["这次没有起草的指标名称（不同粒度、不同表、或写不下的）"] } """ @@ -184,6 +191,39 @@ def _fit(docs: list[dict], chunks: dict[str, dict], guidance: str, return kept, tables, truncated +def _salvage_value(text: str, key: str) -> Any: + """从被截断的回复里抢回一个完整的顶层字段（"sql": "…" 这样的值）。值本身不完整就返回 None。""" + at = text.find(f'"{key}"') + if at < 0: + return None + colon = text.find(":", at + len(key) + 2) + if colon < 0: + return None + start = colon + 1 + while start < len(text) and text[start] in " \t\r\n": + start += 1 + try: + value, _ = json.JSONDecoder().raw_decode(text, start) + except (json.JSONDecodeError, RecursionError): + return None + return value + + +def _salvage_reply(text: str) -> dict: + """JSON 被截断时，把还能完整读出来的部分抢回来。 + + 指标逐条抢（docdraft 按平衡括号找完整的指标对象），sql 和入口字段按完整的 JSON 值抢。抢不回来的留空， + 由调用方照实说「这次起草不完整」，而不是把半截回复当成没有指标。 + """ + return { + "metrics": docdraft.salvage_metrics(text), + "sql": _salvage_value(text, "sql") or "", + "entry_fields": _salvage_value(text, "entry_fields") or [], + "open_questions": _salvage_value(text, "open_questions") or [], + "uncovered": _salvage_value(text, "uncovered") or [], + } + + async def plan(session: Any, *, docs: list[dict], chunks: dict[str, dict], guidance: str = "", datasource: dict | None = None, tables_text: str = "", model: Any = None) -> dict: """一次模型调用，返回 {entry_fields, sql, metrics, open_questions, model_id, truncated}（未校验）。 @@ -205,15 +245,36 @@ async def plan(session: Any, *, docs: list[dict], chunks: dict[str, dict], guida kept, tables, truncated = _fit(docs, chunks, guidance, datasource, tables_text) reply = await model.ainvoke(prompt(docs, kept, guidance, datasource, tables)) - data = docdraft.parse_reply(message_text(reply)) + text = message_text(reply) + salvaged = False + try: + data = docdraft.parse_reply(text) + except ValueError: + # 回复被截断（SQL 写长了、指标太多）：能抢回多少算多少，别整份当成「没有指标」 + data, salvaged = _salvage_reply(text), True metrics = data.get("metrics") questions = data.get("open_questions") + uncovered = data.get("uncovered") + metrics_list = metrics if isinstance(metrics, list) else [] + diagnostics = "" + if salvaged: + diagnostics = f"回复被截断，从完整片段里抢救出 {len(metrics_list)} 条指标" + if not metrics_list: + keys = "、".join(str(k) for k in data) or "无" + shape = {"dict": "对象", "list": "数组", "str": "字符串", "NoneType": "空"}.get( + type(metrics).__name__, type(metrics).__name__) + diagnostics = (f"模型回复里没有可读的指标数组（metrics 是{shape}；顶层键：{keys}）" + + (f"；{diagnostics}" if diagnostics else "")) return { "entry_fields": data.get("entry_fields") if isinstance(data.get("entry_fields"), list) else [], "sql": data.get("sql") if isinstance(data.get("sql"), str) else "", - "metrics": metrics if isinstance(metrics, list) else [], + "metrics": metrics_list, "open_questions": [q for q in questions if isinstance(q, str) and q.strip()] if isinstance(questions, list) else [], + "uncovered": [q.strip() for q in uncovered if isinstance(q, str) and q.strip()] + if isinstance(uncovered, list) else [], + "salvaged": salvaged, + "diagnostics": diagnostics, "model_id": model_id, "truncated": truncated, } diff --git a/backend/tests/test_ledger_draftplan.py b/backend/tests/test_ledger_draftplan.py index 3e5f20ff..97b8fd94 100644 --- a/backend/tests/test_ledger_draftplan.py +++ b/backend/tests/test_ledger_draftplan.py @@ -198,3 +198,30 @@ async def test_plan_refuses_when_no_draft_model_is_available(monkeypatch): monkeypatch.setattr(recipe_ai, "resolve_draft_model", resolve) with pytest.raises(docdraft.DraftUnavailable, match="起草模型"): await draftplan.plan(None, docs=[], chunks={}) + + +async def test_plan_salvages_metrics_when_the_reply_is_cut_off(): + """SQL 把输出额度吃光、JSON 在指标数组里断掉：抢回完整的指标和 SQL，并照实说「回复被截断」。""" + text = ('{"entry_fields": [], "sql": "SELECT SUM(amount) AS gmv FROM orders", "metrics": [' + + json.dumps(raw()["metrics"][0], ensure_ascii=False) + ', {"id": "m2", "name": "写到这里就断了') + result = await draftplan.plan(None, docs=[], chunks={}, model=FakeModel(text)) + assert result["salvaged"] is True + assert len(result["metrics"]) == 1 and result["metrics"][0]["name"] == "销售额" + assert result["sql"].startswith("SELECT SUM(amount)") + assert "截断" in result["diagnostics"] + + +async def test_plan_explains_what_came_back_when_metrics_is_not_a_list(): + """指标为空时不能只说「没有一条通过校验」：要说出模型到底交回了什么形状。""" + model = FakeModel(json.dumps({"sql": "SELECT 1", "metrics": {}}, ensure_ascii=False)) + result = await draftplan.plan(None, docs=[], chunks={}, model=model) + assert result["metrics"] == [] + assert "metrics 是对象" in result["diagnostics"] + assert "sql" in result["diagnostics"] + + +async def test_plan_passes_uncovered_through(): + """一次只起草能共用一条 SQL 的一组，其余的写进 uncovered，下一轮再起草。""" + model = FakeModel(json.dumps({**raw(), "uncovered": ["月活买家"]}, ensure_ascii=False)) + result = await draftplan.plan(None, docs=[], chunks={}, model=model) + assert result["uncovered"] == ["月活买家"]