diff --git a/backend/app/api/ledger.py b/backend/app/api/ledger.py index 8546e210..4716904f 100644 --- a/backend/app/api/ledger.py +++ b/backend/app/api/ledger.py @@ -21,8 +21,9 @@ - POST /api/ledger/drafts/extract、GET /api/ledger/drafts、GET/PATCH /api/ledger/drafts/{id}、POST …/{id}/dismiss、 POST …/{id}/attach、GET …/{id}/trial（router，搭建者）：从业务文档起草口径。主路径一次到位：模型读文档和数据目录， 产出定义 + 取数 SQL + 入口字段 + 指标表达式（app/ledger/draftplan.py），服务端校验收成草稿图、建草稿工作流和 v1， - 后台自动只读试跑；草稿页给「定义 + 数 + 原文引用」。attach 保留为辅助入口：已经有跑对的答案时让它认这份定义 - （app/ledger/docattach.py）。 + 后台自动只读试跑；草稿页给「定义 + 数 + 原文引用」。模型一轮只起草一组指标，一轮覆盖不完就带着 uncovered 续跑， + 一组一个取数节点挂到同一张口径卡上（见 _extract_in_background）。attach 保留为辅助入口：已经有跑对的答案时让它认 + 这份定义（app/ledger/docattach.py）。 这几个新接口都不放给访问密钥（auth/keys.KEY_ROUTES 不登记）。 """ from __future__ import annotations @@ -640,12 +641,86 @@ def _doc_title(docs: list[Document]) -> str: return (f"{title}等 {len(docs)} 份文档" if len(docs) > 1 else title)[:200] +#: 一次起草最多续跑几轮：模型一轮只给一组指标，文档里的指标分成几组由它自己切，覆盖完就停 +MAX_ROUNDS = 6 +#: 一张草稿图最多收几条指标：再多，一张口径卡和一次试跑都读不动 +MAX_METRICS = 40 + + +def _merged_fields(blocks: list[dict[str, Any]]) -> list[dict[str, Any]]: + """各组声明的入口字段按名字合并去重（先出现的为准），给 payload 看：它是整张图的入口。 + + 建图时 draftplan.build_graph 还会再合一次，并在同名冲突时抛 ValueError；这里只负责让界面看到一份像样的 entry_fields。 + """ + merged: dict[str, dict[str, Any]] = {} + for block in blocks: + for field in block.get("entry_fields") or []: + if not isinstance(field, dict): + continue + name = str(field.get("name") or "").strip() + if name and name not in merged: + merged[name] = dict(field) + return list(merged.values()) + + +def _fresh_metrics(metrics: list[dict[str, Any]], covered: list[str]) -> list[dict[str, Any]]: + """这一轮里还没起草过的指标，按名字认；同一轮重复交同名指标的只留第一条。 + + 提示词里已经写了「这些起草过、别重复」，但模型偶尔照抄一遍上一轮的指标。照单全收会出现同名两条指标，口径卡上 + 读的是哪条也说不清，所以在这里挡掉。 + """ + seen = set(covered) + fresh: list[dict[str, Any]] = [] + for metric in metrics: + name = str(metric.get("name") or "").strip() + if name: + if name in seen: + continue + seen.add(name) + fresh.append(metric) + return fresh + + +def _draft_payload(*, entries: list[dict[str, Any]], blocks: list[dict[str, Any]], problems: list[dict[str, Any]], + questions: list[str], uncovered: list[str], progress: dict[str, int], + doc_truncated: bool, catalog_truncated: bool, model_id: str, + salvaged: bool) -> dict[str, Any]: + """草稿 payload 的形状：整份指标、各组 SQL、进度。起草中途也按它写回，页面靠 progress 显示第几轮、覆盖了几条。 + + blocks 是组数（不是内容）；sql 把各组的取数拼在一起（组间空一行），人看的是「这次一共要跑哪几条」。truncated 仍是 + doc_truncated 与 catalog_truncated 的「或」，老前端只认这一个键。 + """ + return { + "documents": entries, + "metrics": [m for block in blocks for m in block["metrics"]], + "entry_fields": _merged_fields(blocks), + "sql": "\n\n".join(str(block["sql"]) for block in blocks), + "version": 1, + "blocks": len(blocks), + "progress": progress, + "trial": {"run_id": "", "status": "running", "error": ""}, + "dropped": problems, + "truncated": doc_truncated or catalog_truncated, + "doc_truncated": doc_truncated, + "catalog_truncated": catalog_truncated, + "model_id": model_id, + "open_questions": questions, + "uncovered": uncovered, + "salvaged": salvaged, + } + + async def _extract_in_background(draft_id: str) -> None: - """后台起草：读文档的段落和数据目录、调一次模型、校验收成草稿图、建草稿工作流、再跑一次试跑。 + """后台起草：读文档的段落和数据目录，分组反复调模型直到覆盖完，收成一张多取数的草稿图、建工作流、再跑一次试跑。 用新的会话：请求那一份在响应发出后就关了，后台接着用它写库是错的。没有数据源或没有表结构时起草必须失败——SQL 只能 对照真实目录写，没有目录就不让模型猜。个别指标不合格只丢那一条、写进 dropped / open_questions；SQL 本身不合格才整份 失败（图跑不了，留着也没有意义）。 + + **为什么一轮接一轮地跑。** 模型一轮只交一组能共用同一条 SQL 的指标（提示词里限 6 条），39 条指标的文档一轮必然覆盖 + 不完。以前只跑一轮，其余全进 uncovered，草稿就停在 6 条。现在把已起草的名字（covered）交回提示词、按 uncovered 续跑， + 每轮收一组；直到模型不再给出新指标、uncovered 空了、或到轮数/条数上限。每轮都先写一次 payload 的 progress 再 + commit：一轮要等一次模型调用，页面得能看见进展，而不是停在「起草中」。 """ from app.core.artifact_store import graph_hash from app.db.models import WorkflowVersion @@ -692,63 +767,105 @@ async def _extract_in_background(draft_id: str) -> None: guidance = str(origin.get("guidance") or "") alignment: dict[str, list[Any]] = {"new": [], "duplicate": [], "conflict": []} + # 每组一组指标：{sql, entry_fields, metrics}，循环结束时拼成一张图 + blocks: list[dict[str, Any]] = [] + problems: list[dict[str, Any]] = [] + covered: list[str] = [] + questions: list[str] = [] + uncovered: list[str] = [] + progress: dict[str, int] = {} + diagnostics = "" + doc_truncated = False + catalog_truncated = False + salvaged = False + model_id = "" + + def state() -> dict[str, Any]: + return _draft_payload(entries=entries, blocks=blocks, problems=problems, questions=questions, + uncovered=uncovered, progress=progress, doc_truncated=doc_truncated, + catalog_truncated=catalog_truncated, model_id=model_id, salvaged=salvaged) + try: - result = await draftplan.plan(session, docs=entries, chunks=chunks, guidance=guidance, - datasource=datasource, tables_text=tables_text) - plan, problems = draftplan.normalize_plan(result, chunks, allowed_tables=allowed_tables) - base = { - "documents": entries, "metrics": plan["metrics"], "entry_fields": plan["entry_fields"], - "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"]: + for rnd in range(1, MAX_ROUNDS + 1): + result = await draftplan.plan(session, docs=entries, chunks=chunks, guidance=guidance, + datasource=datasource, tables_text=tables_text, covered=covered) + plan, round_problems = draftplan.normalize_plan(result, chunks, allowed_tables=allowed_tables) + fresh = _fresh_metrics(plan["metrics"], covered) + if plan["sql"] and fresh: + blocks.append({"sql": plan["sql"], "entry_fields": plan["entry_fields"], "metrics": fresh}) + covered += [m["name"] for m in fresh if m.get("name")] + problems += round_problems + questions += plan["open_questions"] + uncovered = list(result.get("uncovered") or []) + diagnostics = result.get("diagnostics") or diagnostics + doc_truncated = doc_truncated or bool(result.get("doc_truncated")) + catalog_truncated = catalog_truncated or bool(result.get("catalog_truncated")) + salvaged = salvaged or bool(result.get("salvaged")) + model_id = result.get("model_id") or model_id + progress = {"round": rnd, "covered": len(covered), "uncovered": len(uncovered)} + # 每轮先落一次库：这一轮要等一次模型调用，页面这时就能显示「第几轮、起草了几条、还剩几条」 + draft.payload = state() + await session.commit() + if not fresh or not uncovered or len(covered) >= MAX_METRICS: + break + + if not blocks: error = _problems_text(problems) or "模型没有给出可用的取数 SQL 和指标" - if result.get("diagnostics"): - error = f"{error}（{result['diagnostics']}）" - if plan["open_questions"]: + if diagnostics: + error = f"{error}（{diagnostics}）" + if questions: # 模型把缺什么说清楚了（目录里没有这些列、业务上还没定）：这是一份要补信息的草稿，不是失败 - draft.payload = {**base, "trial": {"run_id": "", "status": "", "error": ""}} - draft.model_id = result["model_id"] or None + draft.payload = {**state(), "trial": {"run_id": "", "status": "", "error": ""}} + draft.model_id = 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.payload = {**state(), "trial": {"run_id": "", "status": "failed", "error": error}} + draft.model_id = model_id or None draft.error = error draft.status = "failed" await session.commit() return + + try: + graph = draftplan.build_graph(blocks, name=draft.title, question=guidance or draft.title, + datasource=source.name) + except ValueError as e: + # 各组声明的入口对不上（同名不同类型 / 必填）：合不成一个入口，整份草稿建不出图，照实记为失败 + draft.payload = {**state(), "dropped": [*problems, {"name": "入口字段", "reason": str(e)}], + "trial": {"run_id": "", "status": "failed", "error": str(e)}} + draft.model_id = model_id or None + draft.error = str(e) + draft.status = "failed" + await session.commit() + return + # 对齐要扫所有已发布口径：某一份读不出来不该让整份草稿失败，照实不做对齐（alignment 三个空列表） try: existing = await _published_metrics(session) except Exception: # noqa: BLE001 - 对齐是辅助信息，不是草稿本身 logger.exception("读已发布口径失败，草稿 %s 不做对齐", draft_id) existing = [] - alignment = docdraft.alignment(plan["metrics"], existing) + metrics = [m for block in blocks for m in block["metrics"]] + alignment = docdraft.alignment(metrics, existing) - plan["datasource"] = source.name - graph = draftplan.build_graph(plan, name=draft.title, - question=guidance or draft.title) - questions = list(plan["open_questions"]) + [ + asked = list(dict.fromkeys(questions + [ f"{p['name']}：{p['reason']}" for p in problems - if p.get("name") not in ("sql", "metrics")] + if p.get("name") not in ("sql", "metrics")])) workflow = Workflow( name=(draft.title.strip() or "文档口径草稿")[:200], - description=(f"从业务文档起草的口径：{len(plan['metrics'])} 个指标，取数 SQL 由模型对照数据目录生成，" - f"后台自动只读试运行；人工审改后按受管级别发布"), + description=(f"从业务文档起草的口径：{len(metrics)} 个指标、{len(blocks)} 组取数，取数 SQL 由模型对照" + f"数据目录生成，后台自动只读试运行；人工审改后按受管级别发布"), graph=graph, status="draft", tags=["doc_draft"], origin={"kind": "doc_draft", "draft_id": draft.id, "datasource_id": source.id, "documents": [doc.id for doc in docs], "guidance": guidance}) session.add(workflow) await session.flush() session.add(WorkflowVersion(workflow_id=workflow.id, version=1, graph=graph, - graph_hash=graph_hash(graph), note="从业务文档一次生成")) + graph_hash=graph_hash(graph), note="从业务文档分组生成")) payload = { - **base, "workflow_id": workflow.id, "open_questions": questions, + **state(), "workflow_id": workflow.id, "open_questions": asked, "alignment": alignment, } except docdraft.DraftUnavailable as e: @@ -764,11 +881,11 @@ async def _extract_in_background(draft_id: str) -> None: return draft.payload = payload draft.workflow_id = workflow.id - draft.model_id = result["model_id"] or None + draft.model_id = model_id or None draft.error = None draft.status = _status_for(payload) await session.commit() - values = {field["name"]: field.get("default") for field in plan["entry_fields"] + values = {field["name"]: field.get("default") for field in _merged_fields(blocks) if field.get("default") not in (None, "")} try: diff --git a/backend/tests/test_ledger_docdraft_api.py b/backend/tests/test_ledger_docdraft_api.py index 0c7eb05e..7e3e7df1 100644 --- a/backend/tests/test_ledger_docdraft_api.py +++ b/backend/tests/test_ledger_docdraft_api.py @@ -3,6 +3,7 @@ - 抽取先建一行 drafting、在后台跑模型；结果写成草稿：payload 里有 metrics / sql / entry_fields / trial，工作流和 v1 已建； - 没有数据源、或数据源没有表结构时 422，一个草稿都不建；SQL 对照不上目录时草稿 failed； - 有未决问题 → questions_open（图和试跑照样有），没有 → ready； +- 一轮覆盖不完就带着 covered 续跑：第二轮收成第二个取数节点，两组指标挂同一张口径卡；某一轮没交新指标就停； - PATCH 换指标 / 消问题并重算状态；dismiss 置 rejected 并写审计；列表给计数、按改过的先后排； - GET /drafts/{id}/trial 交回契约里的形状。 @@ -509,6 +510,154 @@ async def test_a_real_trial_run_feeds_the_trial_endpoint(client, documents, tmp_ await session.commit() +async def test_a_big_document_is_drafted_in_several_rounds_onto_one_multi_fetch_graph( + client, documents, datasource, monkeypatch): + """第一轮交一组、uncovered 还剩一条：续跑第二轮，收成一张两个取数节点、卡上两条指标的图。""" + doc = await documents("dd-rounds") + source_id = await datasource() + first = json.loads(plan_reply(f"{doc}-2")) + first["uncovered"] = ["已支付订单数"] + second = json.loads(plan_reply(f"{doc}-2")) + second["sql"] = ("SELECT COUNT(*) AS orders FROM orders WHERE status = 'paid' " + "AND store = '{{ input.store }}'") + second["metrics"] = [{**second["metrics"][0], "id": "m1", "name": "已支付订单数", + "expression": "cell(nodes.fetch, 0, 'orders')"}] + second["uncovered"] = [] + model = fake_model(monkeypatch, json.dumps(first, ensure_ascii=False), json.dumps(second, 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"] == "ready" + assert [m["name"] for m in made["metrics"]] == ["已支付退款率", "已支付订单数"] + assert made["blocks"] == 2 + assert made["progress"] == {"round": 2, "covered": 2, "uncovered": 0} + assert made["truncated"] is False and made["doc_truncated"] is False and made["catalog_truncated"] is False + # 第二轮提示词写明上一轮起草过什么：模型据此不再重复起草 + assert len(model.calls) == 2 and "已支付退款率" in str(model.calls[1][-1].content) + + async with SessionLocal() as session: + row = await session.get(CaliberDraft, body["draft_id"]) + workflow = await session.get(Workflow, row.workflow_id) + graph = workflow.graph + assert [n["id"] for n in graph["nodes"]] == ["start", "fetch_1", "fetch_2", "card", "write", "out"] + card = next(n for n in graph["nodes"] if n["id"] == "card")["data"]["config"]["metrics"] + # 两组都交 m1：跨组重名按 m1…mN 重排，表达式改写到各自的取数节点 + assert [m["id"] for m in card] == ["m1", "m2"] + assert [m["expression"] for m in card] == ["cell(nodes.fetch_1, 0, 'refund_rate')", + "cell(nodes.fetch_2, 0, 'orders')"] + assert graph["nodes"][-1]["data"]["config"]["contract"]["required"] == ["m1", "m2"] + assert validate_graph(GraphSpec.model_validate(graph)).ok + assert made["trial"]["status"] == "succeeded" and made["trial"]["run_id"] + + +async def test_a_round_that_brings_nothing_new_stops_the_loop_without_a_half_built_group( + client, documents, datasource, monkeypatch): + """第二轮什么也没交：停在那一轮，不拿空 SQL 拼出第三组取数，也不留一条半成品指标。""" + doc = await documents("dd-rounds-stop") + source_id = await datasource() + first = json.loads(plan_reply(f"{doc}-2")) + first["uncovered"] = ["已支付订单数"] + fake_model(monkeypatch, json.dumps(first, ensure_ascii=False), json.dumps({ + "entry_fields": [], "sql": "", "metrics": [], "open_questions": [], "uncovered": ["已支付订单数"], + }, 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"] == "ready" and made["blocks"] == 1 and made["metric_count"] == 1 + assert made["progress"] == {"round": 2, "covered": 1, "uncovered": 1} + assert made["uncovered"] == ["已支付订单数"] + async with SessionLocal() as session: + row = await session.get(CaliberDraft, body["draft_id"]) + workflow = await session.get(Workflow, row.workflow_id) + assert [n["id"] for n in workflow.graph["nodes"]] == ["start", "fetch", "card", "write", "out"] + card = next(n for n in workflow.graph["nodes"] if n["id"] == "card") + assert [m["id"] for m in card["data"]["config"]["metrics"]] == ["m1"] + + +async def test_conflicting_entry_fields_across_groups_fail_the_draft_with_a_reason( + client, documents, datasource, monkeypatch): + """两组都声明 region，但类型对不上：合不成一个入口，build_graph 抛 ValueError，草稿照实记为失败。""" + doc = await documents("dd-rounds-clash") + source_id = await datasource() + first = json.loads(plan_reply(f"{doc}-2")) + first["uncovered"] = ["已支付订单数"] + second = json.loads(plan_reply(f"{doc}-2")) + second["entry_fields"] = [{**second["entry_fields"][0], "type": "integer"}] + second["metrics"] = [{**second["metrics"][0], "id": "m2", "name": "已支付订单数", + "expression": "cell(nodes.fetch, 0, 'orders')"}] + second["uncovered"] = [] + fake_model(monkeypatch, json.dumps(first, ensure_ascii=False), json.dumps(second, 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"] == "failed" and made["workflow_id"] is None + assert "入口字段" in made["error"] and "store" in made["error"] + assert made["metric_count"] == 2 and made["progress"] == {"round": 2, "covered": 2, "uncovered": 0} + assert any(d["name"] == "入口字段" for d in made["dropped"]) + + +async def test_a_real_trial_run_reads_values_from_both_fetch_nodes(client, documents, tmp_path, monkeypatch): + """两个取数节点的草稿图真跑一次探针：两条 SQL 各出一个值，卡上两条指标都读得出来（不 mock 试跑）。""" + from app.engine.runner import run_manager + from tests import ledger_samples as S + + await run_manager.setup() + run_id = None + try: + name = await S.fresh_source(tmp_path / "shop-two.db") + async with SessionLocal() as session: + row = (await session.execute(select(DataSource).where(DataSource.name == name))).scalar_one() + row.schema_cache = {"tables": {"orders": {"columns": [ + {"name": "status", "type": "TEXT"}, {"name": "region", "type": "TEXT"}, + {"name": "amount", "type": "REAL"}, {"name": "paid_at", "type": "TEXT"}]}}, "total": 1} + await session.commit() + source_id = row.id + doc = await documents("dd-real-two") + fields = [{"name": "region", "type": "string", "required": False, "default": "east", + "description": "区域"}] + gmv = json.loads(plan_reply(f"{doc}-2")) + gmv.update({"entry_fields": fields, "uncovered": ["已支付订单数"]}) + gmv["sql"] = ("SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' " + "AND region = '{{ input.region }}'") + gmv["metrics"][0].update({"expression": "cell(nodes.fetch, 0, 'gmv')", "name": "已支付销售额", + "unit": "元", "decimals": 2}) + count = json.loads(plan_reply(f"{doc}-2")) + count.update({"entry_fields": fields, "uncovered": []}) + count["sql"] = ("SELECT COUNT(*) AS orders FROM orders WHERE status = 'paid' " + "AND region = '{{ input.region }}'") + count["metrics"] = [{**count["metrics"][0], "id": "m1", "name": "已支付订单数", "unit": "单", + "decimals": 0, "expression": "cell(nodes.fetch, 0, 'orders')"}] + fake_model(monkeypatch, json.dumps(gmv, ensure_ascii=False), json.dumps(count, ensure_ascii=False)) + S.script(monkeypatch, "已支付销售额 [[m:m1]]，已支付订单数 [[m:m2]]。") + monkeypatch.setattr(api_ledger, "_run_trial", REAL_RUN_TRIAL) + + draft_id = (await extract(client, doc, source_id))["draft_id"] + made = (await client.get(f"/api/ledger/drafts/{draft_id}")).json() + assert made["status"] == "ready" and made["trial"]["status"] == "succeeded", made.get("error") + run_id = made["trial"]["run_id"] + + trial = (await client.get(f"/api/ledger/drafts/{draft_id}/trial")).json() + assert trial["status"] == "succeeded" and trial["error"] == "" + assert {m["id"]: float(m["value"]) for m in trial["metrics"]} == {"m1": 156.5, "m2": 4.0} + async with SessionLocal() as session: + run = await session.get(Run, run_id) + tools = [n["id"] for n in (run.graph or {}).get("nodes") or [] if n.get("type") == "tool"] + assert tools == ["fetch_1", "fetch_2"], "试跑要把两组取数都带上" + finally: + await run_manager.shutdown() + async with SessionLocal() as session: + if run_id: + row = await session.get(Run, run_id) + if row is not None: + await session.delete(row) + source = (await session.execute(select(DataSource).where( + DataSource.name == S.CURRENT["source"]))).scalars().first() + 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")