diff --git a/backend/app/engine/nodes/dual_check.py b/backend/app/engine/nodes/dual_check.py index 83fa9cb1..3717c372 100644 --- a/backend/app/engine/nodes/dual_check.py +++ b/backend/app/engine/nodes/dual_check.py @@ -68,9 +68,10 @@ DIFF_TIMEOUT_S = 15.0 #: 交给盲算 agent 的数据目录最多这么多字(表清单、各表的说明分开算),再多就请它自己用 db_schema 查 _INDEX_CHARS = 3_000 _NOTES_CHARS = 6_000 -#: 交给盲算 agent 的同一行其他列最多几列、每个值最多几个字 +#: 交给盲算 agent 的同一行其他列、运行输入最多几项、每个值最多几个字 _DIMS = 6 _DIM_CHARS = 40 +_INPUTS = 12 #: 卡上给容器输入记的大概(metrics._brief):「[12 项]」「{3 个字段}」,写进计算式时不当成字面量 _BRIEF = re.compile(r"^(?:\[\d+ 项\]|\{\d+ 个字段\})$") @@ -102,13 +103,15 @@ BLIND_TASK_PROMPT = ( "口径说明:{definition}\n" "计算式:{formula}\n" "你要算的是计算式里的 {label}{whole}:原流程里它是一次查询结果中「{column}」列的值{dims}。只交这一个数。\n" - "数据源:{source}\n\n" + "数据源:{source}\n" + "{inputs}\n" "{catalog}" ) BLIND_UNIT_PROMPT = "(单位:{unit})" BLIND_NO_DEFINITION_PROMPT = "(未提供,按指标名称和数据目录理解)" BLIND_WHOLE_PROMPT = "(这个指标就是它本身)" BLIND_DIMS_PROMPT = ",所在行的其他列:{dims}" +BLIND_INPUTS_PROMPT = "本次运行的输入(统计期之类的参数在这里):{inputs}\n" BLIND_INDEX_PROMPT = "数据源里的表(表名|中文名|粒度):\n{index}" BLIND_NOTES_PROMPT = "数据目录(标「推断,未确认」的项是推断,使用前请核实):\n{notes}" BLIND_NO_CATALOG_PROMPT = "这个数据源还没有数据目录,请用 db_schema 工具查看表结构。" @@ -351,10 +354,13 @@ async def _check(state: GraphState, ctx: NodeContext, targets: list[dict[str, An sources = _Sources() load, outputs = _loader(state) aliases = _original_aliases(state) + inputs = _run_inputs(state) rows: list[dict[str, Any]] = [] jobs: dict[str, dict[str, Any]] = {} for metric in targets: row, metric_jobs = await _plan_metric(state, metric, options, sources, aliases, load, outputs) + for job in metric_jobs.values(): + job["inputs"] = inputs rows.append(row) for key, job in metric_jobs.items(): # 两个指标用到同一格只盲算一次,按先出现的那个指标说清要算什么;先出现的没写口径说明、后一个写了,用后一个 @@ -373,6 +379,19 @@ async def _check(state: GraphState, ctx: NodeContext, targets: list[dict[str, An return {"rows": rows, "results": results, "model": model, "same_model": same, "usage": usage} +def _run_inputs(state: GraphState) -> list[str]: + """这次运行的输入里的标量(统计期、地区这类参数):原流程的 SQL 常常是拿它们拼出来的,盲算 agent 不知道统计期 + 就算不出同一个数。只给参数,不给 SQL。""" + out: list[str] = [] + for key, value in (state.get("input") or {}).items(): + if isinstance(key, str) and key and (value is None or isinstance(value, (str, int, float, bool))): + text = str(value).replace("\n", " ") + out.append(f"{key} = {text[:_DIM_CHARS]}") + if len(out) >= _INPUTS: + break + return out + + def _original_aliases(state: GraphState) -> dict[str, str]: """原流程的查询快照在这次运行的证据台账里的编号(Q1、Q2…,和报告取全部上游查询时的编号一致)。""" ledger = [e for e in state.get("evidence") or [] if isinstance(e, dict)] @@ -576,7 +595,8 @@ def _task_text(job: dict[str, Any], catalog: str) -> str: definition=job["definition"] or BLIND_NO_DEFINITION_PROMPT, formula=job["formula"], label=job["label"], whole=BLIND_WHOLE_PROMPT if job["formula"].strip() == job["label"] else "", column=job["column"], dims=BLIND_DIMS_PROMPT.format(dims="、".join(job["dims"])) if job["dims"] else "", - source=job["source"], catalog=catalog) + source=job["source"], catalog=catalog, + inputs=BLIND_INPUTS_PROMPT.format(inputs=";".join(job["inputs"])) if job.get("inputs") else "") async def _blind(ctx: NodeContext, jobs: dict[str, dict[str, Any]], options: _Options, settings: Mapping[str, Any], diff --git a/backend/tests/test_dual_check_node.py b/backend/tests/test_dual_check_node.py index 7f3bc783..09392a85 100644 --- a/backend/tests/test_dual_check_node.py +++ b/backend/tests/test_dual_check_node.py @@ -106,8 +106,8 @@ async def wait(run_id: str, statuses=("succeeded", "failed", "interrupted")) -> raise AssertionError(f"运行 {run_id} 停在 {row.status}:{row.error}") -async def start(spec: dict, **kwargs) -> Run: - run = await run_manager.start(graph=spec, input_payload={}, **kwargs) +async def start(spec: dict, inputs: dict | None = None, **kwargs) -> Run: + run = await run_manager.start(graph=spec, input_payload=inputs or {}, **kwargs) return await wait(run.id) @@ -185,14 +185,17 @@ async def test_agreeing_check_passes_without_asking(monkeypatch, shop): async def test_grouped_rows_tell_the_blind_agent_which_group(monkeypatch, shop): - """结果分了组:盲算 agent 要知道算的是哪一组(同一行的维度列),但看不到同一行别的数。""" + """结果分了组:盲算 agent 要知道算的是哪一组(同一行的维度列),但看不到同一行别的数;统计期这类参数在运行输入里。""" name, _ = shop metric = {**GMV, "name": "华东销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"} blind = Blind(monkeypatch, name, {"华东销售额": (BY_REGION, "gmv")}) - run = await start(graph(name, BY_REGION, metrics=[metric], dual={"definitions": {}})) + spec = graph(name, BY_REGION, metrics=[metric], dual={"definitions": {}}) + spec["nodes"][0]["data"]["config"]["fields"] = [{"name": "week_start"}, {"name": "week_end"}] + run = await start(spec, {"week_start": "2026-09-01", "week_end": "2026-09-07"}) assert run.status == "succeeded", run.error [task] = blind.tasks - assert "region 为「华东」" in task and "(未提供" in task + assert "region 为「华东」" in task and "(未提供" in task and "300" not in task + assert "week_start = 2026-09-01" in task and "week_end = 2026-09-07" in task out = dual_output(run, await events(run.id)) assert out["status"] == "agreed" and out["metrics"][0]["status"] == "agreed"