From 6c30becd94237e4d65787676412234a9ccdbfd07 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sun, 4 Oct 2026 01:09:12 +0800 Subject: [PATCH] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=20F3=20/?= =?UTF-8?q?=20B8=EF=BC=9A=E4=B8=80=E9=94=AE=E5=8D=87=E7=BA=A7=E7=9A=84?= =?UTF-8?q?=E8=AF=AD=E4=B9=89=E5=B1=82=EF=BC=88assist=5Fupgrade=EF=BC=89?= =?UTF-8?q?=E8=87=AA=E6=9F=A5=E3=80=81=E4=BF=AE=E6=AD=A3=E8=BD=AE=E5=92=8C?= =?UTF-8?q?=E6=9C=80=E7=BB=88=E5=88=A4=E6=96=AD=E9=83=BD=E5=B8=A6=E4=B8=8A?= =?UTF-8?q?=E5=9F=BA=E4=BA=8E=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=E7=9A=84?= =?UTF-8?q?=20SQL=20=E6=A3=80=E6=9F=A5=E5=99=A8=EF=BC=8C=E5=92=8C=E5=8F=91?= =?UTF-8?q?=E5=B8=83=E9=97=A8=E7=A6=81=E4=B8=80=E4=B8=AA=E5=8F=A3=E5=BE=84?= =?UTF-8?q?=E2=80=94=E2=80=94=E5=8A=A9=E6=89=8B=E6=94=B9=E5=87=BA=E4=B8=80?= =?UTF-8?q?=E5=AF=B9=E5=A4=9A=E5=85=B3=E8=81=94=E5=90=8E=E6=B1=82=E5=92=8C?= =?UTF-8?q?=E8=BF=99=E7=B1=BB=20SQL=20=E4=BC=9A=E4=BA=A4=E5=9B=9E=E5=8E=BB?= =?UTF-8?q?=E6=94=B9=EF=BC=8C=E6=94=B9=E4=B8=8D=E5=A5=BD=E5=B0=B1=E4=BD=9C?= =?UTF-8?q?=E5=BA=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/copilot.py | 12 ++++-- backend/tests/test_copilot_sqlcheck.py | 51 ++++++++++++++++++++++++++ 2 files changed, 60 insertions(+), 3 deletions(-) diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index 7ef959d0..f253d126 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -2251,13 +2251,18 @@ async def assist_upgrade( if not feeders: out["summary"] = "口径卡的输入均不来自负责计算的沙箱代码，无需交给助手改写" return out + from app.data.sqlcheck import load_checkers + sources = await _sources(session, None) + # 基于数据目录的 SQL 检查器：自查和最终判断都带上，和发布门禁一个口径（受管级别下一对多关联后求和这类 error + # 会挡发布）。不带的话，助手改出这种 SQL 照样被采纳，交出去的图要到发布时才被拦 + checkers = await load_checkers(session, sources) before_nodes = {str(n["id"]): n for n in copy.deepcopy(base.get("nodes") or []) if isinstance(n, dict)} before_edges = [e for e in copy.deepcopy(base.get("edges") or []) if isinstance(e, dict)] baseline = list(copy.deepcopy(before_nodes).values()) defaults = base.get("defaults") base_errors = _blocking_issues(before_nodes, before_edges, sources=sources, baseline=baseline, level=level, - defaults=defaults) or [] + defaults=defaults, checkers=checkers) or [] known = {_sig(e) for e in base_errors} out["warnings"] = _kept_code(spec) request = GenerateIn(instruction="升级为可追溯结构", provider=provider, model=model) @@ -2309,7 +2314,8 @@ async def assist_upgrade( await collect(_upgrade_assist_request(base, spec, feeders), first=True) # 自查：和搭图同一套规则，只把这一轮冒出来的 error 交回去改（原来就有的不归这一轮管） for _ in range(_SELF_CHECK_ROUNDS): - now = _blocking_issues(nodes, edges, sources=sources, baseline=baseline, level=level, defaults=defaults) + now = _blocking_issues(nodes, edges, sources=sources, baseline=baseline, level=level, defaults=defaults, + checkers=checkers) fresh = [e for e in now or [] if _sig(e) not in known] if not effective or not fresh: break @@ -2326,7 +2332,7 @@ async def assist_upgrade( proposed = _proposed_graph(base, nodes, edges) reasons = forbidden_changes(base, proposed, tried) after = _blocking_issues(_nodes_by_id(proposed), proposed.get("edges") or [], sources=sources, - baseline=baseline, level=level, defaults=proposed.get("defaults")) + baseline=baseline, level=level, defaults=proposed.get("defaults"), checkers=checkers) if reasons: out["reason"] = "助手的修改降低了要求，已作废：" + "；".join(reasons) elif after is None: diff --git a/backend/tests/test_copilot_sqlcheck.py b/backend/tests/test_copilot_sqlcheck.py index efdd24df..2a6cb072 100644 --- a/backend/tests/test_copilot_sqlcheck.py +++ b/backend/tests/test_copilot_sqlcheck.py @@ -241,3 +241,54 @@ async def test_clean_sql_costs_no_extra_call(scenic_source, monkeypatch): final = events[-1] assert final["autorun"] is True assert not [i for i in final["issues"] if i.get("code") in ("fanout_sum", "missing_valid_filter")] + + +# -------------------------------------------------------------------------- +# 一键升级的语义层（assist_upgrade）：自查和最终判断同样带 SQL 检查器，和发布门禁一个口径 +# -------------------------------------------------------------------------- + + +def _upgrade_graph(source: str) -> dict: + """查库 → 沙箱代码算比率 → 口径卡读代码的产出 → 报告撰写 → 出口。查库的 SQL 本身没问题。""" + def n(nid, ntype, x, **config): + return {"id": nid, "type": ntype, "position": {"x": x, "y": 80}, "data": {"label": nid, "config": config}} + + nodes = [ + n("start", "input", 0), + n("fetch", "tool", 280, tool=f"db_query__{source}", args={"sql": CLEAN_SQL}), + n("calc", "code", 560, language="python", assign_to="calc", code="import json\nprint(json.dumps({'ratio': 3 / 4}))"), + n("card", "metrics", 840, metrics=[{"id": "ratio", "name": "客单价", "expression": "vars.calc.ratio"}]), + n("write", "report", 1120, instructions="写一句话"), + n("out", "output", 1400, fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]), + ] + return {"nodes": nodes, "edges": [{"id": f"e{i}", "source": a["id"], "target": b["id"]} + for i, (a, b) in enumerate(zip(nodes, nodes[1:]))]} + + +async def test_upgrade_assist_checks_sql_like_the_publish_gate(scenic_source, monkeypatch): + """助手改写口径卡时顺手把查询改成了一对多关联后求和：受管级别的发布门禁会挡，升级的自查也得交回去改， + 改不好就作废——以前自查和最终判断都没带 SQL 检查器，这样的改写会被采纳，交出去的图发布时才被拦。""" + import app.api.copilot as copilot + from app.db.base import SessionLocal + + first = [{"op": "plan", "summary": "把纯算术的代码挪进口径卡"}, + {"op": "update_node", "id": "card", "config": {"metrics": [ + {"id": "ratio", "name": "客单价", "expression": "cell(nodes.fetch, 0, 'gmv') / 4"}]}}, + {"op": "update_node", "id": "fetch", "config": {"args": {"sql": FANOUT_SQL}}}, + {"op": "done", "explanation": "客单价改由口径卡直接算"}] + fix = [{"op": "update_node", "id": "fetch", "config": {"args": {"sql": FANOUT_SQL + " "}}}, + {"op": "done", "explanation": "改了"}] + model = _Scripted(first, fix) + + async def _model(*_a, **_k): + return model, "mock-fast" + + monkeypatch.setattr(copilot, "get_chat_model", _model) + async with SessionLocal() as session: + out = await copilot.assist_upgrade(session, _upgrade_graph(scenic_source), level="governed") + # 自查把一对多求和交回模型，带着给模型的改法 + assert len(model.calls) >= 2 and "上线前自查" in model.calls[1] + assert "orders 关联 order_items 是一对多" in model.calls[1] + # 改不好：整个作废，原因写的是这条 SQL 检查 + assert out["accepted"] is False and out["ok"] is False + assert "一对多" in (out["reason"] or ""), out["reason"]