M4:db_schema 认整个写成一个字符串的 JSON 数组的表名

验收时盲算、反方审稿的模型把 table 写成 '["orders", "stores"]'(一个字符串),工具只按逗号拆,回「中没有 ["orders"」
「中没有 "stores"]」,两张表明明都在,模型重试几轮才改成不传 table。表名参数的解析抽成 _schema_tables:元素都是
字符串的 JSON 数组(整个作为一个字符串传进来,或者数组里的一项)按数组认,其余照旧按逗号拆;不是 JSON 的
([orders]、单引号)、元素不是字符串的照旧。只动 db_schema 这一段,查询工具不动。
This commit is contained in:
ThearchyHelios
2026-10-07 00:02:15 +08:00
parent df6b0dd3ae
commit 76cb4d2845
3 changed files with 155 additions and 4 deletions
+22 -4
View File
@@ -644,6 +644,27 @@ async def _store_schema(source: Any, ctx: ToolContext, *,
return None
def _schema_tables(table: str | list[str] | None) -> list[str]:
"""db_schema 的 table 参数 → 要看的表名。"user, roles"、["user","roles"]、整个写成一个字符串的 JSON 数组
'["user", "roles"]' 是同一个意思,模型三种都会写,为此拒绝一次调用纯属浪费一步。最后那种只按逗号拆就成了
「没有 ["user"」「没有 "roles"]」,两张表明明都在(2026-10 验收:盲算、反方审稿的核查 agent 因此白白多调了几轮
模型)。只认元素都是字符串的 JSON 数组,别的写法照旧按逗号拆。"""
wanted = [table] if isinstance(table, str) else list(table or [])
names: list[str] = []
for one in wanted:
text = str(one).strip()
if text.startswith("[") and text.endswith("]"):
try:
parsed = json.loads(text)
except ValueError:
parsed = None
if isinstance(parsed, list) and all(isinstance(t, str) for t in parsed):
names += parsed
continue
names.append(str(one))
return [t.strip() for one in names for t in one.split(",") if t.strip()]
def _make_schema_tool(source: Any, *, fixed: bool = False,
catalog: Mapping[str, data_catalog.CatalogEntry] | None = None) -> StructuredTool:
"""source 同 _make_query_tool:上传源的表结构取绑定快照里冻结的那份。
@@ -655,10 +676,7 @@ def _make_schema_tool(source: Any, *, fixed: bool = False,
async def _run(table: str | list[str] | None = None) -> str:
if gone := _vanished(source, fixed):
return f"{DATA_UNAVAILABLE}{gone}"
wanted = [table] if isinstance(table, str) else list(table or [])
# 逗号分隔也认:"user, roles" 和 ["user","roles"] 是同一个意思,
# 而模型两种都会写。为此拒绝一次调用,纯属浪费一步
wanted = [t.strip() for one in wanted for t in str(one).split(",") if t.strip()]
wanted = _schema_tables(table)
if wanted:
return "\n\n".join(introspect.describe_table(source, t, catalog=notes) for t in wanted)
tables = introspect.table_names(source)
+46
View File
@@ -334,6 +334,52 @@ async def test_schema_tool_still_lists_everything_when_asked_for_nothing(source)
assert "张表" in out and "orders" in out
@pytest.fixture
async def depots(tmp_path):
"""两张表的库(虚构的仓储):parcels、depots,结构已探查。"""
from app.data import introspect
path = tmp_path / "depots.db"
db = sqlite3.connect(path)
db.executescript("CREATE TABLE parcels (id INTEGER PRIMARY KEY, depot_id INTEGER, fee REAL);"
"CREATE TABLE depots (id INTEGER PRIMARY KEY, zone TEXT);")
db.commit()
db.close()
src = SimpleNamespace(id="tool-args-depots-src", name="depots", kind="sqlite", database=str(path), host=None,
port=None, username=None, password=None, options={}, readonly=True, description="测试库",
schema_cache={})
src.schema_cache = await introspect.introspect(src)
return src
@pytest.mark.parametrize("written", ['["parcels", "depots"]', '["parcels","depots"]', ' [ "parcels" , "depots" ] ',
['["parcels", "depots"]'], ['["parcels"]', "depots"]])
async def test_schema_tool_reads_a_json_array_written_as_one_string(depots, written) -> None:
"""2026-10 验收:模型把 table 写成一整个字符串 '["orders", "stores"]'(参数说明里的 ["user","roles"] 引得它这么写),
只按逗号拆就成了「没有 ["orders"」「没有 "stores"]」——两张都在,模型却白白多调了几轮。写成字符串的 JSON 数组
和逗号分隔一样认。"""
out = await _make_schema_tool(depots).coroutine(table=written)
assert out.count("表 parcels") == 1 and out.count("表 depots") == 1, out
assert "没有" not in out
async def test_schema_tool_reads_a_one_table_json_array(depots) -> None:
out = await _make_schema_tool(depots).coroutine(table='["parcels"]')
assert out.startswith("表 parcels") and "depots" not in out and "没有" not in out
# 空数组和不传一样:列出所有表
assert "共 2 张表" in await _make_schema_tool(depots).coroutine(table="[]")
@pytest.mark.parametrize("written,missing", [
("[parcels]", "[parcels]"), # 不是 JSON(名字没加引号):照旧整个当表名
("['parcels']", "['parcels']"), # 单引号也不是 JSON
("[1, 2]", "[1"), # 数组里不是表名:照旧按逗号拆
])
async def test_schema_tool_leaves_other_bracketed_text_as_before(depots, written, missing) -> None:
out = await _make_schema_tool(depots).coroutine(table=written)
assert f"中没有 {missing}。" in out
def test_copilot_is_told_not_to_pin_max_steps() -> None:
"""Copilot 把 max_steps: 8 写进了生成的图,后端默认值根本轮不上。"""
from app.api.copilot import NODE_REFERENCE
@@ -0,0 +1,87 @@
"""核查 agent(engine/verify_agent.py)的 db_schema 认「整个写成一个字符串的 JSON 数组」的表名。
2026-10 验收:盲算、反方审稿的模型把 table 写成 '["orders", "stores"]'(一个字符串),工具只按逗号拆,回「中没有
["orders"」「中没有 "stores"]」,两张表明明都在;模型照原样重试、再失败,到第三轮才改成不传 table,白白多调了几轮
模型。现在第一次调用就拿到两张表的结构(tools/datasource.py 的 _schema_tables),模型看到的工具结果、时间线上的
tool.end 都是表结构。
模型是 MockChatModel 剧本,数据源是现造的 SQLite 文件,表名、数字都是虚构的。
"""
from __future__ import annotations
import json
import sqlite3
import uuid
import pytest
from langchain_core.messages import AIMessage, ToolMessage
from app.core.events import EventType
from app.data import introspect
from app.data.engine import engines
from app.db.base import SessionLocal
from app.db.models import DataSource
from app.engine import verify_agent
from app.engine.verify_agent import run_verify_agent
from app.providers.factory import ModelSpec
from app.providers.mock_model import MockChatModel
from app.tools.registry import Versions
SCHEMA = {"type": "object", "properties": {"value": {"type": "number", "description": "包裹运费合计"}},
"required": ["value"]}
@pytest.fixture
async def depots(tmp_path):
"""两张表:包裹(运费 12、8、5)和仓库。返回源名。"""
path = tmp_path / f"{uuid.uuid4().hex[:8]}.db"
db = sqlite3.connect(path)
db.executescript("CREATE TABLE parcels (id INTEGER PRIMARY KEY, depot_id INTEGER, fee INTEGER);"
"CREATE TABLE depots (id INTEGER PRIMARY KEY, zone TEXT);"
"INSERT INTO parcels VALUES (1, 1, 12), (2, 1, 8), (3, 2, 5);"
"INSERT INTO depots VALUES (1, '东区'), (2, '西区');")
db.commit()
db.close()
name = f"depots_{uuid.uuid4().hex[:8]}"
async with SessionLocal() as session:
row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True)
session.add(row)
await session.flush()
row.schema_cache = await introspect.introspect(row)
await session.commit()
source_id = row.id
yield name
await engines.invalidate(source_id)
async def test_a_json_array_written_as_one_string_gets_both_tables_at_once(monkeypatch, depots):
seen: list[list] = []
steps = [[(f"db_schema__{depots}", {"table": '["parcels", "depots"]'}, "s1")],
[(f"db_query__{depots}", {"sql": "SELECT SUM(fee) AS total FROM parcels"}, "c1")]]
def decide(model, messages):
seen.append(list(messages))
if model.response_format:
return AIMessage(content=json.dumps({"value": {"alias": "Q1", "row": 0, "column": "total"},
"rationale": "运费合计见 [[see:Q1]]。"}, ensure_ascii=False))
if not steps:
return AIMessage(content="查完了。")
return AIMessage(content="查一下。", tool_calls=[{"name": n, "args": a, "id": i} for n, a, i in steps.pop(0)])
async def fake_model(session, spec):
return MockChatModel(model_name="mock-fast"), "mock-fast"
monkeypatch.setattr(MockChatModel, "_decide", decide)
monkeypatch.setattr(verify_agent, "get_chat_model", fake_model)
events: list[tuple] = []
out = await run_verify_agent(spec=ModelSpec(model="mock-fast"), system="核对包裹运费。", task="算一遍包裹运费合计。",
sources=[depots], output_schema=SCHEMA, data_versions=Versions.CURRENT,
emit=lambda kind, **data: events.append((kind, data)), run_id="run-m4", node_id="dual")
assert out.error is None and out.result["value"] == 25
# 模型在循环里看到的那次工具结果:两张表的结构,没有「中没有」
[answer] = [str(m.content) for messages in seen for m in messages
if isinstance(m, ToolMessage) and m.tool_call_id == "s1"][:1]
assert "表 parcels" in answer and "表 depots" in answer and "没有" not in answer
# 时间线上那一条 tool.end 也是表结构
[end] = [d for kind, d in events if kind == EventType.TOOL_END and d.get("call_id") == "s1"]
assert "表 parcels" in end["preview"] and "没有" not in end["preview"]