feat(platform): 补 db_query 只读数据库查询工具——根治内部助手拿 run_command 手搓连库脚本(2026-09-07 记账失败排查实录: 沙箱/tmp tmpfs不持久+平台目录只读+输出双截断+框架API试错,十余轮查一条数据)。SELECT白名单+强制参数化(禁裸%)/tables列定义捷径(sqlor列名转小写用AS别名)/行数200字符8000显式截断/SELECT后COMMIT防MDL锁/剔除ns的page键防sqlor分页变形;PLATFORM_PROMPT 记账failed处置改为db_query核实+禁手搓
This commit is contained in:
parent
2e5ba512d2
commit
31834c812e
@ -10,6 +10,11 @@
|
||||
- extract_llm_api_spec — LLM 通读文档提取配置规格（端点/协议/定价）
|
||||
- apply_llm_config — 写入配置（幂等：供应商按名复用、模型按 (vendor, vendor_model_id) 只更新定价）
|
||||
- platform_modules — 平台模块清单
|
||||
- db_query — 只读数据库查询（2026-09-07）：SELECT 白名单 + ${k}$ 强制参数化（禁裸 %，
|
||||
aiomysql 会误当格式占位符）+ tables 参数直查列定义（sqlor 返回列名转小写，用 AS 别名）
|
||||
+ 行数 200/字符 8000 显式截断 + SELECT 后 COMMIT 防 MDL 锁 + ns 剔除 'page' 键
|
||||
（sqlor.sqlExe 见 page 键走分页路径返回变形）。根治 agent 拿 run_command 手搓连库
|
||||
脚本撞沙箱四重摩擦（/tmp tmpfs 不持久、平台目录只读、输出双截断、框架 API 试错）
|
||||
- **前端入口** `wwwroot/agent_platform/index.ui`：主菜单「平台内部助手」（仅管理员可见）
|
||||
- **模型治理页弹窗入口**（2026-09-07）：pipeline-llm `wwwroot/index.ui` 的模型注册卡片弹窗内
|
||||
「🤖 自动配置」「✅ 上线检查」按钮分别弹出 `auto_config_popup.dspy` / `online_check_popup.dspy`——
|
||||
|
||||
@ -2156,6 +2156,144 @@ async def _h_platform_modules(sor, params, ctx):
|
||||
return json.dumps(mods, ensure_ascii=False)
|
||||
|
||||
|
||||
# ────────────────────── 工具 9：只读数据库查询 ──────────────────────
|
||||
# 根因（2026-09-07 记账失败排查实录）：agent 想核实库内数据（产品映射/记账流水）时，
|
||||
# 工具集里没有 DB 查询动作，只能拿 run_command 手搓连库脚本——撞上沙箱四重摩擦：
|
||||
# ① bwrap 的 /tmp 是会话私有 tmpfs，脚本不跨命令持久（写完下一条就 No such file）
|
||||
# ② /d/pipeline 只读 bind，应用目录写不进
|
||||
# ③ run_command 输出双重截断（stdout[:2000] + tool_result[:500]），无截断提示
|
||||
# ④ 框架 API 逐个踩坑（initEnv 签名/sqlor % 格式化/SHOW 语句返回 int）
|
||||
# 十余轮试错查一条数据。补只读查询工具根治：SELECT 白名单 + ${k}$ 强制参数化 +
|
||||
# 行数/字符上限显式截断 + SELECT 后 COMMIT（防 MDL 锁，已知坑）。
|
||||
|
||||
_DB_QUERY_MAX_ROWS = 200
|
||||
_DB_QUERY_MAX_CHARS = 8000
|
||||
_DB_QUERY_MAX_TABLES = 10
|
||||
|
||||
|
||||
def _validate_readonly_sql(sql: str) -> str:
|
||||
"""只读校验：仅单条 SELECT。返回 ''=通过，否则可行动的错误信息。"""
|
||||
s = re.sub(r"--[^\n]*", " ", sql or "") # 去行注释（防注释夹带）
|
||||
s = re.sub(r"/\*.*?\*/", " ", s, flags=re.S) # 去块注释
|
||||
s = " ".join(s.split()).strip()
|
||||
if not s:
|
||||
return "SQL 为空"
|
||||
if ";" in s.rstrip(";"):
|
||||
return "只允许单条语句（禁止分号分隔多条）"
|
||||
head = s.lstrip("(").lstrip().split(" ", 1)[0].lower()
|
||||
if head != "select":
|
||||
return ("只允许 SELECT 查询（收到 %s）。SHOW/DESC 不支持——查表结构用 tables 参数"
|
||||
"（返回列定义），或 SELECT information_schema.COLUMNS" % head)
|
||||
low = s.lower()
|
||||
for bad in ("into outfile", "into dumpfile", "load_file(", "for update",
|
||||
"lock in share mode"):
|
||||
if bad in low:
|
||||
return "禁止 %s 子句（只读工具）" % bad
|
||||
# 裸 % 会被 aiomysql 当格式占位符 → "not enough arguments for format string"
|
||||
# （内部助手实录踩坑）。LIKE 通配符必须走 ${k}$ 参数化由 params 传值。
|
||||
if "%" in s:
|
||||
return ("SQL 里禁止直接写 %（会被数据库驱动误当格式占位符报 "
|
||||
"not enough arguments）。用 ${key}$ 占位符 + params 传值："
|
||||
"如 LIKE ${k}$ 配 params {\"k\": \"%xxx%\"}")
|
||||
return ""
|
||||
|
||||
|
||||
async def _h_db_query(sor, params, ctx):
|
||||
"""只读数据库查询（SELECT 白名单 / tables 表结构捷径）。
|
||||
|
||||
- sql + params：单条 SELECT，占位符 ${k}$ 参数化，禁止裸 %
|
||||
- tables：逗号分隔表名，直接返回各表列定义（免手写 information_schema SQL）
|
||||
- 行数上限 200、输出上限 8000 字符，超限显式告知（禁止静默截断）
|
||||
- SELECT 后 COMMIT 释放 MDL 锁（已知坑）
|
||||
"""
|
||||
err = await _require_owner(sor, ctx)
|
||||
if err:
|
||||
return err
|
||||
sql = (params.get("sql") or "").strip()
|
||||
tables = (params.get("tables") or "").strip()
|
||||
ns = params.get("params") or {}
|
||||
if isinstance(ns, str):
|
||||
try:
|
||||
ns = json.loads(ns)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
return "params 必须是 JSON 对象，如 {\"k\": \"%xxx%\"}"
|
||||
if not isinstance(ns, dict):
|
||||
return "params 必须是 JSON 对象（键对应 SQL 里的 ${键}$ 占位符）"
|
||||
# 防御：sqlor.sqlExe 见 ns 含 'page' 键会走分页路径返回 DictObject(total,rows)
|
||||
# 而非行列表——占位符命名撞 'page' 时静默变形，这里剔除。
|
||||
ns.pop("page", None)
|
||||
|
||||
# ── 分支 1：tables 表结构捷径 ──
|
||||
if tables and not sql:
|
||||
tlist = [t.strip() for t in tables.split(",") if t.strip()][:_DB_QUERY_MAX_TABLES]
|
||||
out = {}
|
||||
try:
|
||||
for t in tlist:
|
||||
# 注意：sqlor 把返回列名统一转小写（sor.py _get_data），
|
||||
# 故用 AS 别名取小写键，不能 getattr(r, "COLUMN_NAME")。
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT COLUMN_NAME AS col_name, COLUMN_TYPE AS col_type, "
|
||||
"IS_NULLABLE AS col_null, COLUMN_KEY AS col_key, "
|
||||
"COLUMN_DEFAULT AS col_default FROM information_schema.COLUMNS "
|
||||
"WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=${t}$ "
|
||||
"ORDER BY ORDINAL_POSITION", {"t": t})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
out[t] = [{"name": getattr(r, "col_name", ""),
|
||||
"type": getattr(r, "col_type", ""),
|
||||
"null": getattr(r, "col_null", ""),
|
||||
"key": getattr(r, "col_key", ""),
|
||||
"default": getattr(r, "col_default", None)}
|
||||
for r in (recs or [])]
|
||||
except Exception as e:
|
||||
return "查表结构失败: %s" % str(e)[:300]
|
||||
empty = [t for t in tlist if not out.get(t)]
|
||||
note = ("表不存在或无列: %s" % ",".join(empty)) if empty else ""
|
||||
text = json.dumps({"tables": out}, ensure_ascii=False, default=str)
|
||||
if note:
|
||||
text += "\n[⚠️ %s]" % note
|
||||
return text
|
||||
|
||||
if not sql:
|
||||
return "需要 sql（SELECT 语句）或 tables（逗号分隔表名，查列定义）之一"
|
||||
|
||||
# ── 分支 2：SELECT 查询 ──
|
||||
verr = _validate_readonly_sql(sql)
|
||||
if verr:
|
||||
return verr
|
||||
try:
|
||||
recs = await sor.sqlExe(sql, ns)
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
except Exception as e:
|
||||
msg = str(e)[:300]
|
||||
hint = ""
|
||||
if "Unknown column" in msg or "1054" in msg:
|
||||
hint = "（列名不存在：先用 tables 参数查该表的列定义再改 SQL）"
|
||||
elif "Unknown table" in msg or "1146" in msg:
|
||||
hint = "（表不存在：核对表名，如 llm_usage/product/llm_model）"
|
||||
elif "not enough arguments" in msg:
|
||||
hint = "（SQL 含裸 % 或占位符缺值：LIKE 等用 ${k}$ + params 传值）"
|
||||
return "SQL 执行失败: %s%s" % (msg, hint)
|
||||
|
||||
# sqlor 返回 DictObject（dict 子类，__setattr__ 写入键而非 __dict__）——
|
||||
# 必须 dict(r) 转换，不能用 r.__dict__（永远为空）。同 check_model_accounting 先例。
|
||||
rows = [dict(r) for r in (recs or [])]
|
||||
total = len(rows)
|
||||
note = ""
|
||||
if total > _DB_QUERY_MAX_ROWS:
|
||||
rows = rows[:_DB_QUERY_MAX_ROWS]
|
||||
note = ("⚠️ 结果共 %d 行，只返回前 %d 行——请加 WHERE/LIMIT 收窄"
|
||||
% (total, _DB_QUERY_MAX_ROWS))
|
||||
text = json.dumps({"total": total, "rows": rows}, ensure_ascii=False, default=str)
|
||||
while rows and len(text) > _DB_QUERY_MAX_CHARS:
|
||||
rows.pop()
|
||||
text = json.dumps({"total": total, "rows": rows}, ensure_ascii=False, default=str)
|
||||
note = ("⚠️ 输出超 %d 字符上限，已截断到前 %d 行（共 %d 行）——请加 WHERE/LIMIT 收窄"
|
||||
% (_DB_QUERY_MAX_CHARS, len(rows), total))
|
||||
if note:
|
||||
text += "\n[%s]" % note
|
||||
return text
|
||||
|
||||
|
||||
# ────────────────────── 工具定义 ──────────────────────
|
||||
|
||||
PLATFORM_TOOLS = [
|
||||
@ -2227,6 +2365,20 @@ PLATFORM_TOOLS = [
|
||||
parameters={},
|
||||
category="platform",
|
||||
),
|
||||
ToolDefinition(
|
||||
name="db_query",
|
||||
description="只读数据库查询（排查/核实库内数据的唯一正道——禁止用 run_command 手搓连库脚本）。"
|
||||
"两种用法：①sql+params 执行单条 SELECT（占位符 ${k}$ 参数化，SQL 内禁写裸 %，"
|
||||
"LIKE 通配符放 params 传值如 {\"k\":\"%xxx%\"}）；②tables=逗号分隔表名，直接返回列定义"
|
||||
"（免查 information_schema，SQL 报列名/表名错误时先用它核对结构）。"
|
||||
"行数上限200、输出上限8000字符，超限会显式提示收窄。仅owner组织角色可用。",
|
||||
parameters={
|
||||
"sql": "单条 SELECT 语句（占位符 ${k}$；禁裸 %、禁多语句、禁 INTO OUTFILE）",
|
||||
"params": "占位符取值 JSON 对象，如 {\"k\": \"%happyhorse%\", \"id\": \"abc123\"}",
|
||||
"tables": "（与 sql 二选一）逗号分隔表名，返回各表列定义，如 llm_usage,product",
|
||||
},
|
||||
category="platform",
|
||||
),
|
||||
]
|
||||
|
||||
PLATFORM_PROMPT = """
|
||||
@ -2254,7 +2406,10 @@ PLATFORM_PROMPT = """
|
||||
8. `check_model_accounting` 检查记账（传上一步的 task_ref）：
|
||||
- accounted 且 amount_match=true → 全链通过，汇报金额
|
||||
- created → 等 60~120 秒再查一次（出账循环 60 秒一轮）
|
||||
- failed → 按 reason/hint 处置（未映射产品→告知用户在产品管理导入产线模型）
|
||||
- failed → 按 reason/hint 处置；需要核实库内数据（产品映射/定价挂接/流水明细）
|
||||
一律用 `db_query` 只读查询，**禁止用 run_command 手搓连库脚本**（沙箱 /tmp 不跨
|
||||
命令持久、平台目录只读、输出双截断——手搓必陷多轮试错）。
|
||||
未映射产品→告知用户在产品管理导入产线模型（人工步骤，如实转告即止，不要自行深挖）
|
||||
9. 汇报：配置/定价/测试/记账四段结果 + 出处 URL + 遗留事项（如模板需人工核对项）。
|
||||
用户发现问题后按其说明修改，再重新走对应环节（全链幂等，重跑安全）。
|
||||
|
||||
@ -2264,7 +2419,9 @@ PLATFORM_PROMPT = """
|
||||
禁止编造参数或跳过测试；t2t 用默认提示词即可。
|
||||
2. `check_model_accounting` 传上一步 task_ref 检查记账：
|
||||
accounted 且 amount_match → 记账通过；created → 等 60~120 秒复查一次再下结论；
|
||||
failed → 按 reason/hint 报告原因与处置建议。
|
||||
failed → 按 reason/hint 报告原因与处置建议，需核实库内数据用 `db_query`
|
||||
（如 `SELECT ... FROM product WHERE resource_ref_id=${id}$` 查产品映射是否存在），
|
||||
禁止 run_command 手搓连库脚本。
|
||||
3. `platform_llm_status` 复核该模型/定价/供应商当前状态（active/ppid 挂接等）。
|
||||
4. **最后明确告诉用户检查结果：通过 / 不通过** + 各环节明细（调用/记账/状态）
|
||||
+ 遗留事项。结论必须如实——任何一环失败都是不通过，禁止粉饰。
|
||||
@ -2315,6 +2472,7 @@ PLATFORM_HANDLERS = {
|
||||
"test_model_call": _h_test_model_call,
|
||||
"check_model_accounting": _h_check_model_accounting,
|
||||
"platform_modules": _h_platform_modules,
|
||||
"db_query": _h_db_query,
|
||||
}
|
||||
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user