合入两处修正(部署上一段问数据会话里查出):数据源的探查范围(introspect_scope)、定时维护(catalog_maintenance)不再被拼进连接串——以前设过它们(运行时按需加入范围也会自动写)的 MySQL / PostgreSQL / Oracle 数据源,驱动 connect() 收到不认识的参数,每一条查询都失败,加守卫测试扫 app/data 下所有 *_OPTION 常量;上游查库全失败时报告照实说没查到,不再照着上游文字写出引用不存在的 Q1… 的标记。验收:后端全量 14568 通过;文案检查;泄露扫描干净

This commit is contained in:
ThearchyHelios
2026-10-19 08:25:44 +08:00
4 changed files with 181 additions and 7 deletions
+6 -1
View File
@@ -76,8 +76,13 @@ CATALOG_DRAFT_GUIDANCE_OPTION = "catalog_draft_guidance"
# mask_columns:证据面板遮罩的列,驱动不认识它,拼进连接串会被当成未知参数拒掉
# catalog_profile:数据剖析设置,同上;它还是个对象,拼进去连 URL 都不成形
# catalog_draft_guidance:数据目录的起草要求,一段中文,拼进去连接直接失败
# introspect_scope:探查范围的规则(data/source_objects.SCOPE_OPTION),运行时按需把范围外的表加进来时也会自动写;
# catalog_maintenance:定时维护的步骤(data/source_maintenance.MAINTENANCE_OPTION)。这两项都是对象,以前漏在名单外,
# 设过之后驱动 connect() 收到不认识的关键字参数,这个数据源上每一条查询都失败。
# 在别的模块声明的两项这里写字面量(那两个模块反过来要用数据层,导入会成环);tests/test_datasource_option_keys.py 把
# app/data 下所有 *_OPTION 常量扫出来核对,新加的配置项忘了进这份名单,那条测试先报
_NON_DRIVER_OPTIONS = frozenset({"schema", QUERY_TIMEOUT_OPTION, MASK_COLUMNS_OPTION, CATALOG_PROFILE_OPTION,
CATALOG_DRAFT_GUIDANCE_OPTION})
CATALOG_DRAFT_GUIDANCE_OPTION, "introspect_scope", "catalog_maintenance"})
_MASK_SPLIT = re.compile(r"[,,、;;\n]+")
+74 -6
View File
@@ -208,6 +208,63 @@ def report_fields(state: GraphState, spec: GraphSpec, node: GraphNode) -> list[d
return found
#: 写作提示里「没有查到的数据」一节最多列几次、每次的原因留多长:只是让写作者知道哪部分没查到、为什么,不是排错日志
FAILED_SHOWN = 8
FAILED_REASON_CHARS = 160
def failed_queries(state: GraphState, spec: GraphSpec, node: GraphNode) -> list[dict[str, Any]]:
"""上游 Agent 查库失败的那几次:调用记录(产出的 tool_calls)里平台查询工具交回「查询失败…」的。
失败的查询不进证据台账,证据目录里没有它;写作者只在上游的文字里看到 SQL 和列名,会照着写出引用不存在的 Q1… 的
标记,重写一次也常常还是那样,最后把解析不了的标记原样交给看的人(2026-10 部署上撞过:数据源连不上,六次查询全失败,
答案里全是没填上的 ⟦?v:Q1…⟧)。照实告诉写作者哪几次没查到、为什么,让它照实说没查到。范围和 report_fields 一致:
祖先节点,写了 evidence_from 就只收那几个。原因只取报错的第一行:工具交回时已经按这次运行的遮罩规则处理过(datasource
的 mask_error_text),和上游 Agent 自己看到的是同一句。"""
from app.capabilities.registry import is_platform_query
sources = _setting(spec, node, "evidence_from", "ancestors")
if isinstance(sources, str) and sources != "ancestors":
sources = [sources]
keep = {str(s) for s in sources} if isinstance(sources, list) else None
allowed = _ancestors(spec, node.id)
outputs = state.get("nodes") or {}
found: list[dict[str, Any]] = []
for other in spec.nodes:
result = outputs.get(other.id)
if other.id not in allowed or (keep is not None and other.id not in keep) or not isinstance(result, dict):
continue
for call in result.get("tool_calls") or []:
if not isinstance(call, dict) or call.get("repeat") or call.get("skipped") or call.get("denied"):
continue
name, said = str(call.get("tool") or ""), str(call.get("result") or "").strip()
if not is_platform_query(name) or not (call.get("ok") is False or said.startswith("查询失败")):
continue
args = call.get("args") if isinstance(call.get("args"), dict) else {}
reason = said.splitlines()[0] if said else "没有交回结果"
found.append({"node_id": other.id, "title": other.title or other.id, "tool": name,
"sql": " ".join(str(args.get("sql") or "").split())[:200],
"reason": reason[:FAILED_REASON_CHARS]})
return found
def _failed_section(failed: list[dict[str, Any]], *, any_query: bool) -> str:
"""写作提示里「没有查到的数据」一节,和照实说没查到的要求。一次成功的查询都没有时,不许写单元格引用。"""
lines = [f"- 「{f['title']}」{f['tool']}" + (f":{f['sql']}" if f["sql"] else "") + f"({f['reason']})"
for f in failed[:FAILED_SHOWN]]
if len(failed) > FAILED_SHOWN:
lines.append(f"- 另有 {len(failed) - FAILED_SHOWN} 次,同样没有结果")
rule = (FAILED_NONE_RULE if not any_query else FAILED_SOME_RULE)
return f"## 没有查到的数据\n上游有 {len(failed)} 次数据查询失败了,证据目录里没有它们的结果:\n" + "\n".join(lines) + f"\n\n{rule}"
#: 一次成功的查询都没有时交给写作者的要求(写作提示和重写请求里同一句)
FAILED_NONE_RULE = ("这次一次成功的查询都没有:不要写 [[v:…]] 单元格引用,也不要写任何数字;照实写没有查到数据、失败的原因,"
"以及怎么排查(比如检查数据源连接)。")
#: 有成功的查询、也有失败的时
FAILED_SOME_RULE = "上面这几次没有结果可以引用:不要为它们写引用、也不要估数,照实说这部分没有查到。"
async def model_window(session: Any, spec: ModelSpec, model_id: str) -> int | None:
"""写作模型的上下文窗口(token):模型接入里给这个模型登记的 context 优先,其次模型目录
(providers/catalog.py);都没有返回 None,由 catalog_budget 取保守值。"""
@@ -531,16 +588,18 @@ def _named(violations: list[dict[str, Any]], limit: int = 3) -> str:
return "、".join(f"「{n}」" for n in names)
def _repair_request(violations: list[dict[str, Any]]) -> str:
def _repair_request(violations: list[dict[str, Any]], failed_rule: str | None = None) -> str:
return (
"你写的报告没有通过系统核对,问题如下:\n"
f"{describe_violations(violations)}\n\n"
"请重写整篇报告:所有数字都用引用标记写,只引用证据目录里有的指标和输入;"
"不要解释改了什么,只输出新的报告正文。"
+ (f"{failed_rule}" if failed_rule else "")
+ "不要解释改了什么,只输出新的报告正文。"
)
def _messages(ctx: NodeContext, state: GraphState, writer: WriterCatalog, claims: str = "off") -> list[BaseMessage]:
def _messages(ctx: NodeContext, state: GraphState, writer: WriterCatalog, claims: str = "off",
failed: str | None = None) -> list[BaseMessage]:
system = ctx.render_str(ctx.cfg("system", ""), state)
instructions = ctx.render_str(ctx.cfg("instructions", ""), state).strip() \
or "根据下面的证据写一份简洁的报告,先总后分。"
@@ -554,7 +613,7 @@ def _messages(ctx: NodeContext, state: GraphState, writer: WriterCatalog, claims
return [
SystemMessage(content="\n\n".join(p for p in (system, ROLE) if p)),
HumanMessage(content=(
f"{instructions}\n\n{writer.text}\n\n{rules}\n\n"
f"{instructions}\n\n{writer.text}\n\n" + (f"{failed}\n\n" if failed else "") + f"{rules}\n\n"
"只输出报告正文(Markdown),不要写撰写说明。"
)),
]
@@ -678,7 +737,16 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
budget=writer.budget, window=window, queries=writer.queries, omitted=len(writer.omitted),
compact=writer.compact, compact_rows=writer.compact_rows,
nodes=[{k: r[k] for k in ("node_id", "title", "total", "omitted", "compact")} for r in writer.nodes])
messages = _messages(ctx, state, writer, claims)
failed = failed_queries(upstream, spec, ctx.node) # type: ignore[arg-type]
any_query = any(entry.get("kind") == "query" for entry in catalog.values())
failed_rule = (FAILED_SOME_RULE if any_query else FAILED_NONE_RULE) if failed else None
if failed:
ctx.emit(EventType.LOG, level="warn", code="report_queries_failed",
message=f"上游有 {len(failed)} 次数据查询失败" + (",一次成功的都没有" if not any_query else "")
+ ":已告诉写作模型照实说没有查到,不为它们写引用",
failed=len(failed), any_query=any_query)
messages = _messages(ctx, state, writer, claims,
_failed_section(failed, any_query=any_query) if failed else None)
started = time.perf_counter()
spent: list[dict[str, Any]] = []
nudged = False
@@ -726,7 +794,7 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
ctx.emit(EventType.LOG, level="warn", code="report_repair",
message=f"报告有 {len(blocking)} 处未通过核对({_named(blocking)}),"
f"已要求模型重写(第 {repairs} 次)")
messages = [*messages, response, HumanMessage(content=_repair_request(blocking))]
messages = [*messages, response, HumanMessage(content=_repair_request(blocking, failed_rule))]
response, raw = await draft(messages)
doc = await compose(raw)
@@ -0,0 +1,47 @@
"""数据源 options 里 AgentLab 自己的配置项一律不交给驱动(data/engine._NON_DRIVER_OPTIONS)。
2026-10 部署上撞过:探查范围(introspect_scope,大库那一波加的,运行时按需加入表时也会自动写)、定时维护的步骤
(catalog_maintenance)没进摘除名单,被当成驱动参数拼进连接串,MySQL 驱动 connect() 收到不认识的关键字参数,这个数据源上
每一条查询都失败(「connect() got an unexpected keyword argument 'introspect_scope'」)。名单是手写的,以后再加配置项照样会
漏,所以这里把 app/data 下所有 *_OPTION 常量(数据源 options 的键都这样声明)扫出来逐个核对:新加一项忘了进名单,这条测试先报。
"""
from __future__ import annotations
import re
from pathlib import Path
from types import SimpleNamespace
import pytest
from app.data.engine import _NON_DRIVER_OPTIONS, build_url
DATA_DIR = Path(__file__).resolve().parents[1] / "app" / "data"
_DECL = re.compile(r'^([A-Z_]*_OPTION)\s*(?::\s*\w+\s*)?=\s*"([^"]+)"', re.M)
def _declared() -> dict[str, str]:
out: dict[str, str] = {}
for path in sorted(DATA_DIR.glob("*.py")):
for name, key in _DECL.findall(path.read_text(encoding="utf-8")):
out[f"{path.stem}.{name}"] = key
return out
def test_every_declared_option_key_is_kept_away_from_the_driver():
declared = _declared()
assert {"source_objects.SCOPE_OPTION", "source_maintenance.MAINTENANCE_OPTION"} <= set(declared), declared
missing = {where: key for where, key in declared.items() if key not in _NON_DRIVER_OPTIONS}
assert not missing, f"这些数据源配置项会被拼进连接串交给驱动:{missing}"
@pytest.mark.parametrize("kind", ["mysql", "postgresql", "oracle"])
def test_app_options_never_reach_the_connection_url(kind):
keys = set(_declared().values()) | set(_NON_DRIVER_OPTIONS)
app_options = {key: {"mode": "rules"} if key in ("introspect_scope", "catalog_maintenance", "catalog_profile") else "x"
for key in keys}
src = SimpleNamespace(kind=kind, host="db.example", port=None, database="shop", username="reader", password=None,
options={"charset": "utf8mb4", **app_options})
url = build_url(src)
assert "charset=utf8mb4" in url, "驱动参数照旧拼进去"
for key in keys:
assert f"{key}=" not in url, (key, url)
@@ -0,0 +1,54 @@
"""上游查库全失败时,报告照实说没查到(engine/nodes/report.failed_queries)。
2026-10 部署上撞过:数据源连不上,Agent 的六次查询全部「查询失败」,证据目录里一次查询都没有;写作模型照着上游文字里的
SQL 和列名写出 [[v:Q1.r0.…]],重写一次还是那样,答案里全是没填上的 ⟦?v:Q1…⟧。现在写作提示里多一节「没有查到的数据」,
一次成功的都没有时不许写单元格引用、照实写没查到和原因;重写请求里同一句。上游节点的调用记录照 Agent 产出的形状写
(tool_calls:tool、args、ok、result),数据源和 SQL 都是虚构的。
"""
from __future__ import annotations
from tests.test_report_node import engine_up, events, finish, node, script # noqa: F401
FAILED_CALLS = ("[{'tool': 'db_query__shop', 'args': {'sql': 'SELECT COUNT(*) AS n FROM orders'}, 'ok': True, "
"'result': '查询失败:connect() got an unexpected keyword argument'}, "
"{'tool': 'db_query__shop', 'args': {'sql': 'SELECT COUNT(*) AS n FROM orders'}, 'repeat': True}, "
"{'tool': 'db_schema__shop', 'args': {}, 'ok': True, 'result': '表 orders …'}]")
def graph(calls: str = FAILED_CALLS) -> dict:
nodes = [
node("start", "input", fields=[{"name": "q", "default": "订单有多少"}]),
node("agent1", "transform", mode="expression",
expression="{'text': '三条查询都失败了:连不上数据源', 'tool_calls': " + calls + "}"),
node("write", "report", instructions="回答:{{ input.q }}"),
node("out", "output", fields=[{"name": "答案", "value": "{{ nodes.write.text }}"}]),
]
chain = ["start", "agent1", "write", "out"]
return {"nodes": nodes, "edges": [{"source": a, "target": b} for a, b in zip(chain, chain[1:])]}
def _prompt(messages) -> str:
return "\n".join(str(m.content) for m in messages)
async def test_the_writer_is_told_the_queries_failed_and_not_to_cite(monkeypatch):
seen = script(monkeypatch, "订单有 [[v:Q1.r0.n]] 单。", "没有查到订单数:数据源连接失败,请检查数据源连接。")
run = await finish(graph())
first = _prompt(seen[0])
assert "## 没有查到的数据" in first and "上游有 1 次数据查询失败了" in first, first
assert "db_query__shop:SELECT COUNT(*) AS n FROM orders(查询失败:connect() got an unexpected keyword argument)" in first
assert "这次一次成功的查询都没有:不要写 [[v:…]] 单元格引用" in first
assert "db_schema__shop" not in first.split("## 没有查到的数据")[1].split("\n\n")[0], "看表结构不算查询失败"
# 第一稿照旧引用了不存在的 Q1:重写请求里同一句要求
assert len(seen) == 2 and "这次一次成功的查询都没有" in str(seen[1][-1].content)
assert run.output["答案"] == "没有查到订单数:数据源连接失败,请检查数据源连接。"
[warned] = [e for e in await events(run.id, "log") if (e.data or {}).get("code") == "report_queries_failed"]
assert warned.data["failed"] == 1 and warned.data["any_query"] is False
async def test_without_failed_queries_the_prompt_is_unchanged(monkeypatch):
seen = script(monkeypatch, "没有数据。")
calls = "[{'tool': 'db_schema__shop', 'args': {}, 'ok': True, 'result': '表 orders …'}]"
run = await finish(graph(calls))
assert "没有查到的数据" not in _prompt(seen[0])
assert not [e for e in await events(run.id, "log") if (e.data or {}).get("code") == "report_queries_failed"]