合入数据目录阶段 1B:助手建图按任务挑表、沿关系补连接路径、只给相关表全字段,流里告知参考了哪些表,挑不出时回退到只列表名

# Conflicts:
#	backend/app/api/copilot.py
#	frontend/src/dev/preview.tsx
#	scripts/check-stream.mjs
This commit is contained in:
ThearchyHelios
2026-10-03 11:02:08 +08:00
12 changed files with 1632 additions and 80 deletions
+142 -76
View File
@@ -13,7 +13,11 @@ from pydantic import BaseModel, Field
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.api import copilot_context
from app.api.coded import DATASOURCE_SCOPE_EMPTY, CodedHTTPException
# 摘要的预算和截断阈值随数据源上下文搬到了 copilot_context,这里留着老名字
from app.api.copilot_context import DATASOURCE_BUDGET_CHARS as _DATASOURCE_BUDGET_CHARS # noqa: F401
from app.api.copilot_context import DETAIL_TABLE_LIMIT as _DETAIL_TABLE_LIMIT # noqa: F401
# 起别名:本文件底下有个叫 explain 的接口,generate 里又有个局部变量叫 raw,同名会互相盖掉
from app.core.errors import explain as explain_error, graph_error, not_configured, raw as raw_error
from app.data.sqlcheck import CHECK_CODES
@@ -62,12 +66,15 @@ def _scope_note(rows: list[Any], scoped: bool) -> str:
"db_schema__ 工具;前几轮用过别的库,这一轮也不要沿用")
def _tool_catalog(rows: list[Any]) -> str:
def _tool_catalog(rows: list[Any], context: copilot_context.DatasourceContext | None = None) -> str:
"""给 Copilot 看的工具清单。
all_specs() 只有静态注册的内置工具,数据源工具是运行时按库动态生成的
(db_query__<源>),不补进来 Copilot 就不知道它们存在,只会退回让用户
自己写 SQL 的代码节点——接了数据库等于白接。
「可查」后面列前 12 个对象。给了这一轮的数据源上下文时按它排:大库里按需求挑中的表排在前面
(DatasourceContext.listed),不然列出来的是表结构里排在最前的那几张,和需求未必相干。
"""
lines = [
f"- {name}({spec.category}):{spec.description}"
@@ -78,7 +85,7 @@ def _tool_catalog(rows: list[Any]) -> str:
from app.tools.datasource import QUERY_PREFIX, SCHEMA_PREFIX
for row in rows:
tables = _introspect.table_names(row)
tables = context.listed(row) if context is not None else _introspect.table_names(row)
listed = "、".join(tables[:12]) + (f" 等 {len(tables)} 个对象" if len(tables) > 12 else "")
lines.append(
f"- {QUERY_PREFIX}{row.name}(数据库):在「{row.name}」上执行 SQL。"
@@ -91,53 +98,45 @@ def _tool_catalog(rows: list[Any]) -> str:
return "\n".join(lines)
# 数据源摘要的字符预算。超了就退成"只列对象名",字段让 Copilot 用
# db_schema 工具按需查。实测一个 53 对象的库带字段约 4000 字符,
# 接三五个库就会把真正的需求淹掉——system prompt 里塞满 schema 而挤掉
# 用户想要什么,是本末倒置。
_DATASOURCE_BUDGET_CHARS = 6000
# detail 模式下每个库最多列这么多对象(与 introspect.summary 的默认值一致)。
# 超过就说明会截断,那还不如切紧凑模式把对象名列全。
_DETAIL_TABLE_LIMIT = 40
#: 带进挑表需求里的前几轮问题:追问常常只说「那上上个月呢」,光看这一句挑不出表
_PICK_HISTORY_TURNS = 3
def _datasource_section(rows: list[Any]) -> str:
"""数据源与结构摘要。没有数据源时返回空串,不占 prompt。"""
from app.data import introspect as _introspect
async def _pick_need(session: AsyncSession, payload: GenerateIn) -> str:
"""挑表用的需求:这一轮的原话,问数据追问时再带上这次对话前几轮的问题。"""
from app.api.conversations import recent_turns
if not rows:
return ""
if not payload.conversation_id:
return payload.instruction
turns = await recent_turns(session, payload.conversation_id)
asked = [t.question for t in turns if t.question][-_PICK_HISTORY_TURNS:]
return payload.instruction + (PICK_HISTORY_PROMPT.format(questions=";".join(asked)) if asked else "")
# detail 模式每个库最多列 40 个对象,多出来的直接看不见。对象一多,
# "40 张表带字段"反而不如"全部表只给名字"——Copilot 找不到那张表,
# 字段写得再全也没用。所以超出截断阈值就整体切到紧凑模式。
detail = all(len(_introspect.table_names(row)) <= _DETAIL_TABLE_LIMIT for row in rows)
blocks = [_introspect.summary(row, detail=detail) for row in rows]
if detail and sum(len(b) for b in blocks) > _DATASOURCE_BUDGET_CHARS:
detail = False
blocks = [_introspect.summary(row, detail=False) for row in rows]
if not detail:
# Copilot 仍然知道有哪些对象可查,要字段时在图里放一个 db_schema 节点,
# 或者交给运行期的 agent 自己探
blocks.append(
" (对象较多,上面只列了名字。写 SQL 前用 db_schema 工具确认字段,别猜)"
)
return (
"\n\n已接入的数据源(涉及取数时优先用它们,而不是让用户自己写 SQL 的代码节点):\n"
+ "\n".join(blocks)
+ "\n\n写 SQL 的硬性要求:\n"
# 这三条都是真实踩过的:Oracle 上漏 schema 前缀直接 ORA-00942,
# 写惯 MySQL 的模型会顺手写 LIMIT,而一条没有行数限制的查询能拖垮生产库
"- 表名照抄上面的全名(含 schema 前缀),漏掉前缀在 Oracle 上会直接报 ORA-00942\n"
"- 认准方言:Oracle 用 FETCH FIRST n ROWS ONLY,不是 LIMIT\n"
"- 一次只写一条语句;分号拼接会被拒\n"
"- 字段拿不准就在图里先放一个 db_schema 工具节点,别猜字段名\n"
# 真实踩过:列名是照着业务说法猜的,跑到查库那一步才「no such column」
"- 列名只照上面结构里列出来的写;结构里只列了表名时先用 db_schema 查字段。搭完自查会拿 SQL 里的列名"
"去对结构,对不上会打回来让你改\n"
"- 比率、增幅、占比这类派生计算进口径卡(metrics 节点);查询结果里原样的数,报告撰写节点(report)"
"可以直接用 [[v:Q1.r0.列名]] 引用。别让 llm 节点直接对数字做算术\n"
)
PICK_HISTORY_PROMPT = "\n(这次对话之前问过:{questions})"
async def _picker(session: AsyncSession, payload: GenerateIn,
plan: copilot_context.ContextPlan) -> tuple[Any | None, str | None]:
"""挑表用的模型:同一个助手模型(copilot_model_spec),思考、额度在 pick_model 里调。不用挑表时不拿。"""
if not plan.needs_pick:
return None, None
spec = await copilot_model_spec(session, payload, max_tokens=copilot_context.PICK_MAX_TOKENS)
return await copilot_context.pick_model(session, spec)
async def _datasource_context(
session: AsyncSession, sources: list[Any], payload: GenerateIn, *, need: str,
graph: dict[str, Any] | None,
) -> copilot_context.DatasourceContext:
"""这一轮的数据源上下文:读目录、大库按需求挑表、组装(copilot_context)。挑表失败退回只列表名,不抛异常。
非流式生成、发布前修复、升级都走这里;流式生成把三步拆开,挑表放进流里(期间发心跳、挑完发 context 操作)。
graph 是现有工作流:它的 SQL 查过的表一律给全字段。
"""
plan = await copilot_context.plan_context(sources, graph=graph)
picker, unavailable = await _picker(session, payload, plan)
return await plan.resolve(picker, need=need, unavailable=unavailable)
NODE_REFERENCE = """\
@@ -677,8 +676,18 @@ async def generate(
"""
sources = await _sources(session, payload.datasource_ids)
scope = {r.name for r in sources} if payload.datasource_ids else None
tool_list = _tool_catalog(sources)
datasources = _datasource_section(sources) + _scope_note(sources, scope is not None)
# 先拿模型:没配好就直接 400,不白花一次挑表
try:
model, _ = await get_chat_model(session, await copilot_model_spec(session, payload))
except ProviderNotConfigured as e:
raise HTTPException(400, _unconfigured(e)) from e
# 和流式生成同一套数据源上下文,只是不发 context 操作(这条接口一次返回整张图)
context = await _datasource_context(
session, sources, payload, need=await _pick_need(session, payload),
graph=payload.base_graph if (payload.base_graph or {}).get("nodes")
else await _previous_graph(session, payload.conversation_id))
tool_list = _tool_catalog(sources, context)
datasources = context.section() + _scope_note(sources, scope is not None)
system = (
"你是一个 agent 工作流编排专家。根据用户需求产出一张可执行的工作流图。\n\n"
@@ -701,11 +710,6 @@ async def generate(
else:
user = _user_message(payload, patch=False)
try:
model, _ = await get_chat_model(session, await copilot_model_spec(session, payload))
except ProviderNotConfigured as e:
raise HTTPException(400, _unconfigured(e)) from e
messages = [("system", system),
("human", _with_history(user, await _history_section(session, payload.conversation_id)))]
try:
@@ -1049,13 +1053,16 @@ async def _iter_ops(model: Any, messages: list[Any]):
_HEARTBEAT_SECONDS = 3.0
async def _with_heartbeat(source: Any, phase: str = "planning"):
async def _with_heartbeat(source: Any, phase: str = "planning", *, started: float | None = None):
"""模型沉默时按拍补心跳,让前端能显示阶段和已用时长。
必须在这一层做而不是在 _iter_ops 里判断时间差:模型不吐 chunk 时那个
async for 的循环体根本不执行,压根轮不到检查。
started 给了就从那一刻算已用时长:流式生成先挑表、再起草,两段的心跳要接着算,
不能起草一开始就跳回 0 秒。
"""
started = time.monotonic()
started = time.monotonic() if started is None else started
iterator = source.__aiter__()
while True:
pending = asyncio.ensure_future(iterator.__anext__())
@@ -1083,6 +1090,18 @@ async def _with_heartbeat(source: Any, phase: str = "planning"):
}
async def _heartbeat_until(task: asyncio.Future, *, started: float, phase: str = "planning"):
"""等 task 做完,期间按拍出心跳(同 _with_heartbeat)。结果由调用方取 task.result()。
流式生成在流里挑表时用:挑表最长要等 copilot_context.PICK_TIMEOUT_S,这段时间不出声,
前端就只能干挂着「正在连接模型」。阶段算「理解需求」:挑表正是在读需求。
"""
while not task.done():
done, _ = await asyncio.wait({task}, timeout=_HEARTBEAT_SECONDS)
if not done:
yield {"op": "heartbeat", "phase": phase, "elapsed_ms": int((time.monotonic() - started) * 1000)}
#: 自查最多交回去改几轮。一轮只花一次"只输出改动"的调用;两轮还改不好,多半是
#: 需求本身有歧义,该让人看了,而不是继续替他烧钱
_SELF_CHECK_ROUNDS = 2
@@ -1628,6 +1647,12 @@ def dropped_tool_warnings(changes: list[dict[str, Any]], instruction: str) -> li
def _repair_request(
nodes: dict[str, dict[str, Any]], edges: list[dict[str, Any]], errors: list[dict[str, Any]],
) -> str:
"""自查交回修正的请求:只写图和问题。
数据源上下文(挑中的表、字段、目录、连接条件)在 system 里,修正轮和起草用同一条 system,
不重新挑表:两次挑出来的表不一样,模型前后看到的结构就对不上,改着改着换了一张表。
猜错的列名由 sql_unknown_column 的 for_model 带回那几张表的真实列。
"""
graph = {"nodes": list(nodes.values()), "edges": edges}
return (
"这是你刚搭好的工作流:\n"
@@ -1653,21 +1678,22 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g
sources = await _sources(session, payload.datasource_ids)
scope = {r.name for r in sources} if payload.datasource_ids else None
tool_list = _tool_catalog(sources)
datasources = _datasource_section(sources) + _scope_note(sources, scope is not None)
# 基于数据目录的 SQL 检查器:在开流之前读好目录(流里不再碰这个会话),自查和收尾共用
checkers = await load_checkers(session, sources)
system = (
"你是一个 agent 工作流编排专家。根据用户需求,以操作流的方式逐步搭出一张可执行的工作流图。\n\n"
f"{NODE_REFERENCE}\n\n可用工具:\n{tool_list}{datasources}\n\n"
"结构要求:\n"
"1. 必须有且只有一个 input 节点和至少一个 output 节点\n"
"2. 节点 id 用简短英文小写下划线;label 用中文,一眼看懂\n"
"3. 用 assign_to 传结果,下游 {{ vars.变量名 }} 引用\n"
"4. 只能用上面列出的工具名\n"
"5. 能用 3 个节点解决就别堆 8 个\n\n"
f"{_STREAM_PROTOCOL}"
)
def system_for(context: copilot_context.DatasourceContext) -> str:
return (
"你是一个 agent 工作流编排专家。根据用户需求,以操作流的方式逐步搭出一张可执行的工作流图。\n\n"
f"{NODE_REFERENCE}\n\n可用工具:\n{_tool_catalog(sources, context)}"
f"{context.section()}{_scope_note(sources, scope is not None)}\n\n"
"结构要求:\n"
"1. 必须有且只有一个 input 节点和至少一个 output 节点\n"
"2. 节点 id 用简短英文小写下划线;label 用中文,一眼看懂\n"
"3. 用 assign_to 传结果,下游 {{ vars.变量名 }} 引用\n"
"4. 只能用上面列出的工具名\n"
"5. 能用 3 个节点解决就别堆 8 个\n\n"
f"{_STREAM_PROTOCOL}"
)
# 现有图状态:修改场景从 base_graph 起步,新建场景从空图起步
nodes: dict[str, dict[str, Any]] = {}
@@ -1695,13 +1721,18 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g
except ProviderNotConfigured as e:
raise HTTPException(400, _unconfigured(e)) from e
messages = [("system", system),
("human", _with_history(user, await _history_section(session, payload.conversation_id)))]
human = _with_history(user, await _history_section(session, payload.conversation_id))
# 兜底用:模型只发了改动操作时,把它们重放到这张图上
prev_graph = await _previous_graph(session, payload.conversation_id)
# 比对工具绑定的基准:画布上是改之前的图;问数据页是上一轮的图(沿用它的节点 id)
baseline = copy.deepcopy(
(payload.base_graph or {}).get("nodes") or (prev_graph or {}).get("nodes") or [])
# 数据源上下文:读目录、拿挑表模型都要用请求的会话,开流之前做完;挑表本身放进流里,期间照常发心跳。
# 现有工作流(画布上的原图、问数据追问时上一轮的图)查过的表一律给全字段
plan = await copilot_context.plan_context(
sources, graph=payload.base_graph if (payload.base_graph or {}).get("nodes") else prev_graph)
picker, unavailable = await _picker(session, payload, plan)
need = await _pick_need(session, payload)
async def event_stream():
explanation = ""
@@ -1710,8 +1741,27 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g
# 建完要不要跑,由模型在 done.run 里表态
autorun = True
yield f"data: {json.dumps({'op': 'model', 'model': model_id}, ensure_ascii=False)}\n\n"
# 大库先按需求挑表。resolve 自己兜住了挑表的所有失败(退回只列表名),这里的 except 只防组装出意外:
# 挑表是让模型少猜,不能因为它让这一轮生成失败
started = time.monotonic()
pending = asyncio.ensure_future(plan.resolve(picker, need=need, unavailable=unavailable))
try:
async for op in _with_heartbeat(_iter_ops(model, messages)):
async for beat in _heartbeat_until(pending, started=started):
yield _sse(beat)
context = pending.result()
except Exception as e: # noqa: BLE001
context = plan.fallback(f"组装数据源上下文时出错:{explain_error(e)[0]}")
finally:
# 用户中途断开时流在上面的 yield 处被关掉,挑表的调用不能留在后台接着等
if not pending.done():
pending.cancel()
# 这一份 system 一直用到最后:自查交回修正时沿用同一份上下文,不重新挑表
system = system_for(context)
messages = [("system", system), ("human", human)]
if (shown := context.op()) is not None:
yield _sse(shown)
try:
async for op in _with_heartbeat(_iter_ops(model, messages), started=started):
kind = op.get("op")
# 思考和心跳不是图操作,不进 _apply_op,直接转给前端
if kind in ("thinking", "heartbeat"):
@@ -1903,6 +1953,12 @@ def _publish_fix_request(graph: dict[str, Any], errors: list[dict[str, Any]], le
)
def _publish_fix_need(errors: list[dict[str, Any]], level: str) -> str:
"""发布前修复挑表用的需求:要过哪一级的发布前检查、挡着的是哪些问题。"""
return (f"让这张工作流通过「{_LEVEL_WORD.get(level, level)}」级别的发布前检查,挡住发布的问题:\n"
+ "\n".join(f"- {_issue_line(i)}" for i in errors))
async def assist_publish_fix(
session: AsyncSession, graph: dict[str, Any], *, level: str,
provider: str | None = None, model: str | None = None,
@@ -1934,16 +1990,18 @@ async def assist_publish_fix(
if not errors:
out["summary"] = "没有阻止发布的错误,无需交给助手"
return out
request = GenerateIn(instruction="发布前自动修复", provider=provider, model=model)
try:
chat, _ = await _op_model(
session, GenerateIn(instruction="发布前自动修复", provider=provider, model=model))
chat, _ = await _op_model(session, request)
except ProviderNotConfigured as e:
out["summary"] = _unconfigured(e)
return out
# 和画布上改图同一套数据源上下文:挑表的需求是挡住发布的这些问题,图里查过的表一律给全字段
context = await _datasource_context(session, sources, request, need=_publish_fix_need(errors, level), graph=base)
system = (
"你是一个 agent 工作流编排专家。现在要做的是:让一张工作流过发布前检查,以操作流的方式输出修改。\n\n"
f"{NODE_REFERENCE}\n\n可用工具:\n{_tool_catalog(sources)}{_datasource_section(sources)}\n\n"
f"{NODE_REFERENCE}\n\n可用工具:\n{_tool_catalog(sources, context)}{context.section()}\n\n"
f"{_STREAM_PROTOCOL}\n\n另外可以输出 question 操作:{{\"op\":\"question\",\"node_id\":\"…\",\"text\":\"…\"}}"
)
nodes = copy.deepcopy(before_nodes)
@@ -2066,6 +2124,12 @@ def _upgrade_assist_request(graph: dict[str, Any], spec: GraphSpec, feeders: lis
)
def _upgrade_need(feeders: list[Any]) -> str:
"""升级挑表用的需求:哪几段沙箱代码要改写成口径卡表达式(改写要用到上游查库节点查出的列)。"""
return ("把喂给口径卡的纯算术沙箱代码改写成口径卡表达式,表达式直接读上游查库节点查出的列。涉及的代码节点:"
+ "、".join(f"「{code.title}」" for code, _cards in feeders))
def _kept_code(spec: GraphSpec) -> list[str]:
"""assist 之后仍然喂着口径卡的计算角色沙箱代码:不是纯算术、或者 Copilot 拿不准的,保留原样并警告。"""
from app.engine.upgrade import compute_feeders
@@ -2108,17 +2172,19 @@ async def assist_upgrade(
defaults=defaults) or []
known = {_sig(e) for e in base_errors}
out["warnings"] = _kept_code(spec)
request = GenerateIn(instruction="升级为可追溯结构", provider=provider, model=model)
try:
chat, _ = await _op_model(
session, GenerateIn(instruction="升级为可追溯结构", provider=provider, model=model))
chat, _ = await _op_model(session, request)
except ProviderNotConfigured as e:
out.update(ok=False, summary=_unconfigured(e))
return out
# 同一套数据源上下文:改写时要用到查库节点查出的列,图里查过的表一律给全字段
context = await _datasource_context(session, sources, request, need=_upgrade_need(feeders), graph=base)
system = (
"你是一个 agent 工作流编排专家。现在要做的是:把一张工作流里喂给口径卡的纯算术沙箱代码改写成口径卡表达式,"
"以操作流的方式输出修改。\n\n"
f"{NODE_REFERENCE}\n\n可用工具:\n{_tool_catalog(sources)}{_datasource_section(sources)}\n\n"
f"{NODE_REFERENCE}\n\n可用工具:\n{_tool_catalog(sources, context)}{context.section()}\n\n"
f"{_STREAM_PROTOCOL}\n\n另外可以输出 question 操作:{{\"op\":\"question\",\"node_id\":\"…\",\"text\":\"…\"}}"
)
nodes = copy.deepcopy(before_nodes)
+713
View File
@@ -0,0 +1,713 @@
"""助手建图时给模型看的数据源上下文:小库给详细摘要,大库先按需求挑表,再把挑中的表讲透(数据目录阶段 1B)。
**为什么要挑表。** 接进来的业务库常有上百张表。以前超过详细摘要的阈值(一个库超过 40 张表,或者几个库的摘要
加起来超过 6000 字)就整体退成「只列表名」,工具清单里也只列前 12 个表名:助手一个字段都看不到,只能照着表名
编字段、猜怎么连表,跑到查库那一步才「no such column」。现在大库先用一次轻量的模型调用,按这一轮的需求从
紧凑的表目录(每张表一行:表名|中文名|粒度)里挑出相关的表,再沿关系图补上连接要经过的中间表;挑中的表给
全部字段、数据目录和可以照抄进 SQL 的连接条件,其余表只列名字,让模型知道还有别的表可查。
**按数据源分三种(selected_by)。**
- all:放得下详细摘要的库,照旧每张表一行带字段,额外带上数据目录和表之间的连接条件;
- model:大库,挑表成功:挑中的表讲透,其余只列名字;
- fallback:大库,挑表失败(模型不可用、报错、超时、回复无法识别、返回为空或返回的表都不存在),退回以前的
「只列表名」。挑表只是让模型少猜,**绝不能挡住生成**:没有它,模型照旧可以先用 db_schema 查字段。
现有工作流(画布上改图时的原图、问数据追问时上一轮的图)的 SQL 里查过哪些表是确定的事实:大库里这些表一律
给全字段,挑表失败时也一样。
**一轮只挑一次。** 流式生成的自查修正、升级的修正轮都沿用同一份上下文(同一段 system 提示词),不重新挑表:
两次挑出来的表不一样,模型前后看到的结构就对不上。
**用法。** plan_context 读目录、按大小定每个源怎么给(不调模型,接口函数在开流之前调,因为请求的会话只在
那时可用);pick_model 拿挑表用的模型(同一个助手模型,关掉思考、额度取小);ContextPlan.resolve 挑表并组装
(流式生成在流里调,期间照常发心跳);DatasourceContext 给出提示词片段(section)、工具清单里列出的可查对象
(listed)和告诉前端这一轮参考了哪些表的 context 操作(op)。
"""
from __future__ import annotations
import asyncio
import json
import logging
import math
import time
from collections.abc import Iterable, Mapping
from dataclasses import dataclass, field
from typing import Any, Literal
from app.data import catalog
from app.data import introspect
from app.data.catalog import CARDINALITY_LABEL, INFERRED_MARK, JoinEdge
from app.providers.factory import ModelSpec, get_chat_model
logger = logging.getLogger(__name__)
# ==========================================================================
# 参数
# ==========================================================================
#: 详细摘要的字符预算。超了的库改为按需求挑表,字段让挑中的表带、其余的让模型用 db_schema 工具按需查。
#: 实测一个 53 对象的库带字段约 4000 字符,接三五个库就会把真正的需求淹掉——system prompt 里塞满 schema
#: 而挤掉用户想要什么,是本末倒置
DATASOURCE_BUDGET_CHARS = 6000
#: 详细摘要每个库最多列这么多对象(与 introspect.summary 的默认值一致)。超过就说明会截断:多出来的表模型
#: 根本看不见,那还不如挑表
DETAIL_TABLE_LIMIT = 40
#: 挑表最多挑几张(所有数据源合计,含现有工作流用到的表)
PICK_LIMIT = 12
#: 补上连接用的中间表之后,给全字段的表最多几张(所有数据源合计)
CONTEXT_TABLE_LIMIT = 16
#: 挑表调用的时限(秒)。挑表在生成之前、用户在等,宁可放弃挑表也不能让生成干等
PICK_TIMEOUT_S = 20.0
#: 挑表调用的输出额度:十几个表名的 JSON 用不了几百 token
PICK_MAX_TOKENS = 1024
#: 详细模式下附带的连接条件和数据目录的字符预算(所有数据源合计)。超了的不附,模型用 db_schema 查时照样看得到
CATALOG_BUDGET_CHARS = 6000
#: 一个源最多列多少条连接条件、多少条连接路径
JOIN_LINES_LIMIT = 40
PATH_LINES_LIMIT = 12
SelectedBy = Literal["all", "model", "fallback"]
# ==========================================================================
# 写给模型的文字(不上界面)
# ==========================================================================
SECTION_HEAD_PROMPT = "\n\n已接入的数据源(涉及取数时优先用它们,而不是让用户自己写 SQL 的代码节点):\n"
#: 有源只列了名字时补的一句(以前紧凑模式的原话)
NAMES_ONLY_PROMPT = " (对象较多,上面只列了名字。写 SQL 前用 db_schema 工具确认字段,别猜)"
INFERRED_PROMPT = (" 标「推断,未确认」的说明和连接条件是按列名、数据库注释或模型推断出来的,没有人确认过:"
"照抄前对照字段核实,拿不准就先用 db_schema 工具查")
SQL_RULES_PROMPT = (
"\n\n写 SQL 的硬性要求:\n"
# 这三条都是真实踩过的:Oracle 上漏 schema 前缀直接 ORA-00942,
# 写惯 MySQL 的模型会顺手写 LIMIT,而一条没有行数限制的查询能拖垮生产库
"- 表名照抄上面的全名(含 schema 前缀),漏掉前缀在 Oracle 上会直接报 ORA-00942\n"
"- 认准方言:Oracle 用 FETCH FIRST n ROWS ONLY,不是 LIMIT\n"
"- 一次只写一条语句;分号拼接会被拒\n"
"- 字段拿不准就在图里先放一个 db_schema 工具节点,别猜字段名\n"
# 真实踩过:列名是照着业务说法猜的,跑到查库那一步才「no such column」
"- 列名只照上面结构里列出来的写;结构里只列了表名时先用 db_schema 查字段。搭完自查会拿 SQL 里的列名"
"去对结构,对不上会打回来让你改\n"
"- 比率、增幅、占比这类派生计算进口径卡(metrics 节点);查询结果里原样的数,报告撰写节点(report)"
"可以直接用 [[v:Q1.r0.列名]] 引用。别让 llm 节点直接对数字做算术\n"
)
PICK_SYSTEM = (
"你在帮工作流助手挑选数据表。用户这一轮提了一个需求,助手接下来要据此搭工作流、写 SQL;数据库的表太多,"
"没法把每张表的字段都给它看,它只会看到你挑出的表的全部字段。请从表目录里挑出完成这个需求要用到的表。\n\n"
"挑选规则:\n"
f"- 最多挑 {PICK_LIMIT} 张,按和需求的相关程度从高到低排\n"
"- 事实表和维度表都要:要按渠道、门店、日期之类分组或筛选时,把对应的维度表也挑上\n"
"- 两张表要靠一张关联表才能连起来时,关联表也挑上\n"
"- 「当前工作流已用到」的表,需求还用得着就挑上\n"
"- 拿不准要不要的,宁可挑上\n"
"- table 照抄目录里的表名(含 schema 前缀),不要改写、不要编造;source 写表所在的数据源名\n"
"- 需求和数据库无关(闲聊、只改提示词或流程结构)时返回空列表\n\n"
'只输出 JSON:{"tables": [{"source": "数据源名", "table": "表名"}]}'
)
PICK_PROMPT = "需求:{need}\n\n{used}表目录(每行一张表:表名|中文名|粒度;行尾标「推断,未确认」的是推断出来的):\n\n{tables}"
PICK_USED_PROMPT = "当前工作流已用到:{tables}\n\n"
PICK_PROMPT_SCHEMA = {
"title": "picked_tables",
"description": "完成需求要用到的表,按相关程度从高到低排",
"type": "object",
"properties": {
"tables": {
"type": "array",
"items": {
"type": "object",
"properties": {
"source": {"type": "string", "description": "表所在的数据源名"},
"table": {"type": "string", "description": "照抄表目录里的表名"},
},
"required": ["table"],
},
},
},
"required": ["tables"],
}
# ==========================================================================
# 拿模型
# ==========================================================================
async def pick_model(session: Any, spec: ModelSpec) -> tuple[Any | None, str | None]:
"""挑表用的模型:(模型, None),拿不到时 (None, 原因)。
spec 是助手的模型规格(copilot_model_spec),这里关掉思考、额度取小、温度 0、请求时限对齐 PICK_TIMEOUT_S:
挑表是一次分类,不需要思考,思考只会拖慢生成开始的时间。没配接入时 get_chat_model 会静默退回演示模型,
演示模型挑不了表,直接按挑表失败处理。
get_chat_model 用本模块的名字:建图模型和挑表模型分得开(测试里各换各的假模型,互不串)。
"""
from app.core.errors import explain
from app.providers.mock_model import MockChatModel
spec = spec.model_copy(update={"thinking": "off", "max_tokens": PICK_MAX_TOKENS, "temperature": 0,
"timeout": math.ceil(PICK_TIMEOUT_S)})
try:
model, _ = await get_chat_model(session, spec)
except Exception as e: # noqa: BLE001 - 拿不到模型只是不挑表,生成照常
return None, f"助手使用的模型不可用:{explain(e)[0]}"
if isinstance(model, MockChatModel):
return None, "演示模型不能按需求挑选数据表"
return model, None
# ==========================================================================
# 规划:读目录、定模式
# ==========================================================================
@dataclass
class SourcePlan:
"""一个数据源怎么给:all 给详细摘要,pick 先挑表。"""
source: Any
mode: Literal["all", "pick"]
#: 目录原样(含驳回项):关系图要靠驳回项压住同一条关系的现推结果
raw_notes: dict[str, dict[str, Any]]
#: 现有工作流 SQL 里查过的表(schema_cache 的键)
seeds: list[str]
@property
def tables(self) -> dict[str, Any]:
return (getattr(self.source, "schema_cache", None) or {}).get("tables") or {}
@property
def notes(self) -> dict[str, dict[str, Any]]:
"""给模型看的目录:去掉驳回项。"""
return {t: catalog.visible_notes(n) for t, n in self.raw_notes.items()}
def _modes(sources: list[Any]) -> list[Literal["all", "pick"]]:
"""每个源给详细摘要还是先挑表。
沿用以前的判断:表超过 DETAIL_TABLE_LIMIT 的库一定挑表;其余的按详细摘要从短到长放进 DATASOURCE_BUDGET_CHARS,
放得下的给详细摘要,放不下的挑表。全都放得下时和以前一样全是详细模式;以前超预算是所有库一起退成只列表名,
现在是放不下的那几个库去挑表。没探查过结构的库没有表可挑,照旧给一行说明。
"""
modes: list[Literal["all", "pick"]] = []
sized: list[tuple[int, int]] = []
for i, src in enumerate(sources):
names = introspect.table_names(src)
if names and len(names) > DETAIL_TABLE_LIMIT:
modes.append("pick")
continue
modes.append("all")
if names:
sized.append((len(introspect.summary(src, detail=True)), i))
used = 0
for size, i in sorted(sized):
if used + size <= DATASOURCE_BUDGET_CHARS:
used += size
else:
modes[i] = "pick"
return modes
def graph_tables(graph: Mapping[str, Any] | None, sources: Iterable[Any]) -> dict[str, list[str]]:
"""现有工作流里查库节点的 SQL 用到的表:{源名: [schema_cache 的键]},按出现顺序。
只认调用工具节点上写死的 SQL(db_query__<源>);Agent 运行时自己写的 SQL 图里看不到。节点认两种写法:
画布的 {data: {config}} 和 GraphSpec 的 {config}。SQL 解析不了的跳过。
"""
from app.engine.evidence import sql_tables
from app.tools.datasource import QUERY_PREFIX
by_name = {getattr(s, "name", None): s for s in sources}
out: dict[str, list[str]] = {}
for node in (graph or {}).get("nodes") or []:
if not isinstance(node, dict):
continue
config = node.get("config")
if not isinstance(config, dict):
config = (node.get("data") or {}).get("config") if isinstance(node.get("data"), dict) else None
if not isinstance(config, dict):
continue
tool = str(config.get("tool") or "")
source = by_name.get(tool[len(QUERY_PREFIX):]) if tool.startswith(QUERY_PREFIX) else None
args = config.get("args")
sql = args.get("sql") if isinstance(args, dict) else None
if source is None or not isinstance(sql, str):
continue
try:
names = sql_tables(sql)
except Exception: # noqa: BLE001 - 认不出表就当没用到,只少一个提示
continue
found = out.setdefault(source.name, [])
for name in names:
key = catalog.resolve_table_name(source.schema_cache, name)
if key and key not in found:
found.append(key)
return {k: v for k, v in out.items() if v}
async def plan_context(sources: list[Any], *, graph: Mapping[str, Any] | None = None) -> ContextPlan:
"""读各源的数据目录、定每个源怎么给。不调模型。
graph 是现有工作流(改图时的原图、追问时上一轮的图),它的 SQL 里查过的表在大库里一律给全字段。目录读不到
就当没有(tools.datasource._catalog_of:用单独的会话读,出错不连累请求的会话)。
"""
from app.tools.datasource import _catalog_of
seeds = graph_tables(graph, sources)
plans = []
for src, mode in zip(sources, _modes(sources)):
entries = await _catalog_of(src) if introspect.table_names(src) else {}
plans.append(SourcePlan(source=src, mode=mode, raw_notes={t: e.notes for t, e in entries.items()},
seeds=seeds.get(src.name, [])))
return ContextPlan(plans)
# ==========================================================================
# 挑表
# ==========================================================================
class _Unreadable(ValueError):
"""模型的回复里识别不出表名。"""
def _picked(raw: Any) -> list[tuple[str | None, str]] | None:
"""模型的回复 → [(源名或 None, 表名)]。结构化输出的对象、正文里的 JSON、字符串列表都认;对不上格式返回 None。"""
if hasattr(raw, "model_dump"):
raw = raw.model_dump()
if isinstance(raw, str):
text = raw.strip()
start = min((i for i in (text.find("{"), text.find("[")) if i >= 0), default=-1)
if start < 0:
return None
end = text.rfind("}" if text[start] == "{" else "]")
try:
raw = json.loads(text[start:end + 1])
except ValueError:
return None
if isinstance(raw, dict):
raw = raw.get("tables")
if not isinstance(raw, list):
return None
out: list[tuple[str | None, str]] = []
for item in raw:
if isinstance(item, str) and item.strip():
out.append((None, item.strip()))
elif isinstance(item, dict) and isinstance(item.get("table"), str) and item["table"].strip():
src = item.get("source")
out.append((src.strip() if isinstance(src, str) and src.strip() else None, item["table"].strip()))
return out
async def _ask(model: Any, messages: list[tuple[str, str]]) -> list[tuple[str | None, str]]:
"""问一次。先走结构化输出;不支持、报错、或给回来的不是约定的结构,再退回纯文本、从正文里取 JSON。
结构化输出报错也退回(同 catalog._ask_model):有的网关不支持工具调用,一调就报错,纯文本却能用。两次加起来
仍受 PICK_TIMEOUT_S 约束。
"""
from app.engine.state import message_text
try:
parsed = _picked(await model.with_structured_output(PICK_PROMPT_SCHEMA).ainvoke(messages))
except Exception: # noqa: BLE001 - 退回纯文本
parsed = None
if parsed is not None:
return parsed
parsed = _picked(message_text(await model.ainvoke(messages)))
if parsed is None:
raise _Unreadable("模型的回复不是约定的格式,无法识别要用的表")
return parsed
@dataclass
class ContextPlan:
"""plan_context 的结果:各源怎么给,等着挑表、组装。"""
plans: list[SourcePlan]
@property
def needs_pick(self) -> bool:
return any(p.mode == "pick" for p in self.plans)
def _pick_messages(self, need: str) -> list[tuple[str, str]]:
blocks, used = [], []
for p in self.plans:
if p.mode != "pick":
continue
qualified = _qualifier(p.tables)
index = catalog.render_table_index(p.source.schema_cache, p.notes)
blocks.append(f"数据源「{p.source.name}」(共 {len(p.tables)} 张表):\n{index}")
if p.seeds:
used.append(f"数据源「{p.source.name}」的 " + "、".join(qualified(t) for t in p.seeds))
human = PICK_PROMPT.format(need=need.strip(), used=PICK_USED_PROMPT.format(tables=";".join(used)) if used
else "", tables="\n\n".join(blocks))
return [("system", PICK_SYSTEM), ("human", human)]
def _resolve(self, picks: list[tuple[str | None, str]]) -> list[tuple[int, str]]:
"""模型挑的表对到表结构上:[(第几个源, schema_cache 的键)]。对不上的丢掉。
只有一个源要挑时不看模型写的源名(写错了也不要紧);有几个时按源名认,表名前面带着源名(「源名.表名」)
也认,写了一个不存在的源名的丢掉;没写源名的按顺序在各源里找。
"""
pick = [i for i, p in enumerate(self.plans) if p.mode == "pick"]
by_name = {self.plans[i].source.name: i for i in pick}
out: list[tuple[int, str]] = []
for source, table in picks:
name = table
if len(pick) == 1:
candidates = pick
elif source in by_name:
candidates = [by_name[source]]
else:
prefixed = next(((by_name[n], table[len(n) + 1:]) for n in by_name
if table.startswith(n) and table[len(n):len(n) + 1] in (".", ":", ":")), None)
if prefixed is not None:
candidates, name = [prefixed[0]], prefixed[1]
elif source is not None:
continue
else:
candidates = pick
for i in candidates:
key = catalog.resolve_table_name(self.plans[i].source.schema_cache, name)
if key:
out.append((i, key))
break
return out
async def resolve(self, model: Any | None, *, need: str, unavailable: str | None = None) -> DatasourceContext:
"""挑表并组装。不抛异常:挑表的任何失败都退回只列表名(fallback),原因写进 context 操作和日志。
model 是 pick_model 拿到的模型;拿不到时为 None,unavailable 是原因。没有要挑表的源时不调模型。
"""
if not self.needs_pick:
return self._assemble([], reason=None, elapsed_ms=None)
started = time.monotonic()
reason: str | None = None
chosen: list[tuple[int, str]] = []
if model is None:
reason = unavailable or "未配置可用的模型"
else:
try:
picks = await asyncio.wait_for(_ask(model, self._pick_messages(need)), timeout=PICK_TIMEOUT_S)
chosen = self._resolve(picks)
if not picks:
reason = "模型未挑出任何表"
elif not chosen:
reason = "模型挑出的表在表结构中都不存在"
except asyncio.TimeoutError:
reason = f"挑选数据表超过 {PICK_TIMEOUT_S:g} 秒未完成"
except _Unreadable as e:
reason = str(e)
except Exception as e: # noqa: BLE001 - 挑表失败不挡生成
from app.core.errors import explain
reason = f"挑选数据表的模型调用失败:{explain(e)[0]}"
elapsed = int((time.monotonic() - started) * 1000)
ctx = self._assemble(chosen if reason is None else [], reason=reason, elapsed_ms=elapsed)
if reason is None:
logger.info("助手挑表用时 %d ms:%s", elapsed, ";".join(
f"{s.source.name} {len(s.tables)} 张" for s in ctx.sources if s.selected_by == "model"))
else:
logger.warning("助手挑表未成功(%s),用时 %d ms,退回只列表名", reason, elapsed)
return ctx
def fallback(self, reason: str) -> DatasourceContext:
"""不挑表、直接按挑表失败组装:resolve 本身出了意外时用。"""
return self._assemble([], reason=reason, elapsed_ms=None)
def _assemble(self, chosen: list[tuple[int, str]], *, reason: str | None,
elapsed_ms: int | None) -> DatasourceContext:
"""定下每个源给全字段的表。
先放现有工作流用到的表,再按模型给的顺序放挑中的表,合计不超过 PICK_LIMIT;然后每个源在挑中的表两两之间
找不超过 2 跳的连接路径,把中间表补进来,所有源合计不超过 CONTEXT_TABLE_LIMIT。已经直接相连的一对不补;
有一条路径的中间表已经在里面的也不补。
"""
picked: dict[int, list[str]] = {}
room = PICK_LIMIT
for i, key in [(i, t) for i, p in enumerate(self.plans) if p.mode == "pick" for t in p.seeds] + chosen:
have = picked.setdefault(i, [])
if room > 0 and key not in have:
have.append(key)
room -= 1
extra = CONTEXT_TABLE_LIMIT - sum(len(v) for v in picked.values())
out: list[SourceContext] = []
for i, p in enumerate(self.plans):
if p.mode == "all":
out.append(SourceContext(p, "all", list(p.tables)))
continue
first = picked.get(i, [])
graph: dict[str, list[JoinEdge]] = {}
tables, routes = list(first), []
if len(first) > 1:
graph = _relation_graph(p)
tables, routes = _complete(first, graph, extra)
extra -= len(tables) - len(first)
out.append(SourceContext(p, "fallback" if reason is not None else "model", tables, picked=first,
routes=routes, graph=graph))
return DatasourceContext(out, reason=reason, elapsed_ms=elapsed_ms)
def _relation_graph(plan: SourcePlan) -> dict[str, list[JoinEdge]]:
"""这个源的关系图:目录里的关系(驳回的不进)加上按结构现推的(外键约束、命名推断)。推不出来就当没有。"""
try:
return catalog.relation_graph(plan.raw_notes, schema_cache=plan.source.schema_cache)
except Exception: # noqa: BLE001 - 少了连接条件只是少一个提示
logger.exception("数据源 %s 的关系图推不出来,本轮不给连接条件", plan.source.name)
return {}
def _complete(picked: list[str], graph: Mapping[str, list[JoinEdge]],
room: int) -> tuple[list[str], list[tuple[str, str, tuple[JoinEdge, ...]]]]:
"""补连接用的中间表:(给全字段的表, [(a, b, 经中间表的两跳路径)])。room 是还能补几张。"""
tables = list(picked)
routes: list[tuple[str, str, tuple[JoinEdge, ...]]] = []
for i, a in enumerate(picked):
for b in picked[i + 1:]:
if any(e.to_table == b for e in graph.get(a, ())):
continue
paths = [p for p in catalog.join_paths(graph, a, b, max_hops=2) if len(p) == 2]
via = next((p for p in paths if p[0].to_table in tables), None)
if via is None and paths and len(tables) - len(picked) < room:
via = paths[0]
tables.append(via[0].to_table)
if via is not None:
routes.append((a, b, via))
return tables, routes
# ==========================================================================
# 组装好的上下文
# ==========================================================================
def _qualifier(tables: Mapping[str, Any]):
"""表名(schema_cache 的键)→ 全名:模型照着全名写 SQL,漏了 schema 前缀在 Oracle 上直接报错。"""
return lambda key: str((tables.get(key) or {}).get("qualified") or key)
@dataclass
class SourceContext:
"""一个源这一轮怎么给的:selected_by、给全字段的表(all 时是全部表)。"""
plan: SourcePlan
selected_by: SelectedBy
#: 给全字段的表(schema_cache 的键):挑中的表在前,补进来的中间表在后
tables: list[str]
#: 挑中的表(含现有工作流用到的),不含补进来的中间表
picked: list[str] = field(default_factory=list)
routes: list[tuple[str, str, tuple[JoinEdge, ...]]] = field(default_factory=list)
graph: dict[str, list[JoinEdge]] = field(default_factory=dict)
@property
def source(self) -> Any:
return self.plan.source
@property
def total(self) -> int:
return len(self.plan.tables)
def qualified(self, key: str) -> str:
return _qualifier(self.plan.tables)(key)
@dataclass
class DatasourceContext:
sources: list[SourceContext]
#: 挑表失败的原因(没有失败时为 None)
reason: str | None = None
#: 挑表用了多久(毫秒);没有挑表时为 None
elapsed_ms: int | None = None
def listed(self, source: Any) -> list[str]:
"""工具清单里「可查」后面列的对象:挑中的表在前,其余按表结构的顺序。"""
names = introspect.table_names(source)
ctx = next((s for s in self.sources if s.source is source), None)
if ctx is None or ctx.selected_by == "all" or not ctx.tables:
return names
first = [ctx.qualified(t) for t in ctx.tables]
return first + [n for n in names if n not in first]
def op(self) -> dict[str, Any] | None:
"""告诉前端这一轮参考了哪些表:{"op": "context", "sources": [...], "elapsed_ms"?}。没有可查的表时为 None。
每个源一项:source 源名,tables 给了全字段的表(全名),selected_by,total 这个源一共几张表;挑表失败的
带上 reason。只列了名字的表不进 tables:界面上说的「参考了 N 张表」是模型看得到字段的那些。
"""
rows = []
for s in self.sources:
if not s.total:
continue
row: dict[str, Any] = {"source": s.source.name, "tables": [s.qualified(t) for t in s.tables],
"selected_by": s.selected_by, "total": s.total}
if s.selected_by == "fallback" and self.reason:
row["reason"] = self.reason
rows.append(row)
if not rows:
return None
out: dict[str, Any] = {"op": "context", "sources": rows}
if self.elapsed_ms is not None:
out["elapsed_ms"] = self.elapsed_ms
return out
def section(self) -> str:
"""system 提示词里的数据源一节。没有数据源时返回空串,不占提示词。"""
if not self.sources:
return ""
budget = [CATALOG_BUDGET_CHARS]
blocks: list[str] = []
names_only = False
for s in self.sources:
try:
if s.selected_by == "all":
blocks.append(_detail_block(s, budget))
elif s.tables:
blocks.append(_selected_block(s))
else:
blocks.append(introspect.summary(s.source, detail=False))
names_only = True
except Exception: # noqa: BLE001 - 结构缓存里有意外的写法:这个源退回只列表名,不挡生成
logger.exception("数据源 %s 的上下文组装失败,本轮只列表名", s.source.name)
blocks.append(introspect.summary(s.source, detail=False))
names_only = True
text = "\n".join(blocks)
tail = []
if names_only:
tail.append(NAMES_ONLY_PROMPT)
if INFERRED_MARK in text:
tail.append(INFERRED_PROMPT)
return SECTION_HEAD_PROMPT + "\n".join([text, *tail]) + SQL_RULES_PROMPT
# ==========================================================================
# 渲染
# ==========================================================================
def _notes_body(s: SourceContext, key: str) -> list[str]:
"""一张表的目录正文(render_table_notes 去掉标题行:「推断,未确认」的说明在整节末尾说一次)。"""
block = catalog.render_table_notes(s.plan.notes.get(key), meta=s.plan.tables.get(key),
system_notes=catalog.system_notes_source(s.source))
return block.split("\n")[1:] if block else []
def _condition(edge: JoinEdge, qualified) -> str:
"""连接条件,表名用全名:channel_visits.visit_id = visits.id(多列用 AND 连接)。"""
left, right = qualified(edge.from_table), qualified(edge.to_table)
return " AND ".join(f"{left}.{a} = {right}.{b}" for a, b in zip(edge.from_columns, edge.to_columns))
def _edge_line(edge: JoinEdge, qualified) -> str:
card = f"({CARDINALITY_LABEL[edge.cardinality]})" if edge.cardinality in CARDINALITY_LABEL else ""
return _condition(edge, qualified) + card + (INFERRED_MARK if edge.status == "proposed" else "")
def _edges_within(graph: Mapping[str, list[JoinEdge]], tables: list[str]) -> list[JoinEdge]:
"""两端都在 tables 里的关系,每条一次;一对多的翻过来,从「多」的一端写起(外键所在的表在左)。"""
wanted = set(tables)
seen: set[str] = set()
out = []
for table in tables:
for edge in graph.get(table, ()):
if edge.to_table not in wanted or edge.relation_id in seen:
continue
seen.add(edge.relation_id)
out.append(edge.reversed() if edge.cardinality == "one_to_many" else edge)
return out
def _join_lines(s: SourceContext, graph: Mapping[str, list[JoinEdge]], tables: list[str]) -> list[str]:
edges = _edges_within(graph, tables)
if not edges:
return []
lines = [" 表之间的连接条件(写 JOIN … ON … 时照抄):"]
lines += [f" {_edge_line(e, s.qualified)}" for e in edges[:JOIN_LINES_LIMIT]]
if len(edges) > JOIN_LINES_LIMIT:
lines.append(f" …另有 {len(edges) - JOIN_LINES_LIMIT} 条,用 db_schema 工具查看各表的关联关系")
return lines
def _detail_block(s: SourceContext, budget: list[int]) -> str:
"""放得下详细摘要的源:照旧每张表一行带字段,后面附表之间的连接条件和数据目录。
连接条件和目录共用一份预算(budget,所有详细模式的源合计 CATALOG_BUDGET_CHARS):详细摘要本身已经按
DATASOURCE_BUDGET_CHARS 放进来了,附带的东西不能再把提示词撑大一倍。连接条件先放(最短、最能防止猜着连表),
目录按表放,放不下的表让模型用 db_schema 查(db_schema 的输出里带目录)。
"""
lines = [introspect.summary(s.source, detail=True)]
if not s.total:
return lines[0]
joins = _join_lines(s, _relation_graph(s.plan), list(s.plan.tables))
size = sum(len(line) + 1 for line in joins)
if joins and size <= budget[0]:
budget[0] -= size
lines += joins
elif joins:
lines.append(" (表之间的连接条件用 db_schema 工具查看各表的关联关系)")
notes: list[str] = []
skipped = False
for key in s.plan.tables:
body = _notes_body(s, key)
if not body:
continue
piece = [f" · {s.qualified(key)}:", *(f" {line}" for line in body)]
size = sum(len(line) + 1 for line in piece)
if size > budget[0]:
skipped = True
continue
budget[0] -= size
notes += piece
if notes:
lines.append(" 数据目录:")
lines += notes
if skipped:
lines.append(" (其余表的数据目录用 db_schema 工具查看)")
return "\n".join(lines)
def _source_head(source: Any) -> str:
"""一个源的标题行,和 introspect 的详细摘要同一个写法。"""
schema = (getattr(source, "schema_cache", None) or {}).get("schema")
return (f"- {source.name}({source.kind},{'只读' if source.readonly else '可写'}):"
f"{source.description or '未填说明'}" + (f"|schema:{schema}" if schema else ""))
def _route_line(a: str, b: str, path: tuple[JoinEdge, ...], qualified) -> str:
first, second = path
qa, mid, qb = qualified(a), qualified(first.to_table), qualified(b)
return (f" {qa} 与 {qb} 没有直接关联,经 {mid} 连接:FROM {qa} JOIN {mid} ON {_condition(first, qualified)} "
f"JOIN {qb} ON {_condition(second, qualified)}" + (INFERRED_MARK if any(e.status == "proposed" for e in path)
else ""))
def _selected_block(s: SourceContext) -> str:
"""挑了表的源:挑中的表给全部字段和目录,然后是连接条件、连接路径,其余表只列名字。"""
lead = "按这一轮的需求挑出" if s.selected_by == "model" else "现有工作流用到"
blocks, noted = [], False
for key in s.tables:
blocks += [f" {line}" for line in introspect.describe_table(s.source, key).split("\n")]
body = _notes_body(s, key)
if body:
noted = True
blocks.append(" 数据目录:")
blocks += [f" {line}" for line in body]
lines = [_source_head(s.source),
f" {lead}的 {len(s.tables)} 张表(全部字段{',有数据目录的附在表后' if noted else ''}):", *blocks]
lines += _join_lines(s, s.graph, s.tables)
if s.routes:
lines.append(" 连接路径:")
lines += [_route_line(a, b, path, s.qualified) for a, b, path in s.routes[:PATH_LINES_LIMIT]]
shown = set(s.tables)
rest = [s.qualified(t) for t in s.plan.tables if t not in shown]
if rest:
lines.append(f" 其余 {len(rest)} 个对象只列名字(要用时先用 db_schema__{s.source.name} 查字段,别猜):"
+ "、".join(rest))
return "\n".join(lines)
__all__ = [
"CATALOG_BUDGET_CHARS", "CONTEXT_TABLE_LIMIT", "ContextPlan", "DATASOURCE_BUDGET_CHARS", "DETAIL_TABLE_LIMIT",
"DatasourceContext", "PICK_LIMIT", "PICK_MAX_TOKENS", "PICK_TIMEOUT_S", "SourceContext", "SourcePlan",
"graph_tables", "pick_model", "plan_context",
]
+531
View File
@@ -0,0 +1,531 @@
"""助手建图时按需求挑表、组装数据源上下文(数据目录阶段 1B)。
背景:大库(几十上百张表)超出详细摘要的阈值时,以前只给模型看表名,助手一个字段都看不到,只能照着
名字编字段、猜怎么连表。现在先用一次轻量的模型调用按需求挑出相关的表,再沿关系图补上连接用的中间表,
挑中的表给全部字段、数据目录和可以照抄的连接条件,其余表只列名字。
这组守的是:
- 挑表、补路径、组装的结果(合成景区库:入园记录和渠道之间要经关联表 channel_visits)
- 挑表失败(报错、超时、返回不存在的表、返回为空)一律退回只列表名,生成照常进行
- 小库保持详细模式,额外带上目录
- 流式输出里有 context 操作;自查修正沿用同一份上下文,不再重新挑表
- 非流式生成、发布前修复、升级共用同一套组装
"""
from __future__ import annotations
import asyncio
import json
import sqlite3
import uuid
import pytest
from httpx import ASGITransport, AsyncClient
from app.data import catalog
from app.data.catalog import INFERRED_MARK, make_item
from app.data.engine import engines
from app.data.introspect import introspect
from app.db.base import SessionLocal
from app.db.models import DataSource
from app.main import app
from tests.fixtures.catalog import scenic
@pytest.fixture
async def client():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
yield c
async def _add_source(database: str, *, prefix: str = "p1b") -> DataSource:
async with SessionLocal() as session:
row = DataSource(name=f"{prefix}_{uuid.uuid4().hex[:8]}", kind="sqlite", database=database,
description="景区业务库")
session.add(row)
await session.commit()
row.schema_cache = await introspect(row)
await session.commit()
await session.refresh(row)
session.expunge(row)
return row
async def _drop_source(row: DataSource) -> None:
await engines.invalidate(row.id)
async with SessionLocal() as session:
live = await session.get(DataSource, row.id)
if live is not None:
await session.delete(live)
await session.commit()
@pytest.fixture
async def scenic_source(scenic_db):
"""接上合成景区库(50 张表 + 1 个视图),用完删掉:留着会让同一进程里别的测试也进入大库模式。"""
row = await _add_source(scenic_db)
yield row
await _drop_source(row)
@pytest.fixture
async def small_source(tmp_path):
"""两张表的小库:orders.customer_id 只能靠命名推断连到 customers。"""
path = tmp_path / "small.db"
db = sqlite3.connect(path)
db.executescript(
"CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT NOT NULL);"
"CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER NOT NULL, amount REAL NOT NULL,"
" ordered_at TEXT NOT NULL);")
db.commit()
db.close()
row = await _add_source(str(path), prefix="p1b_small")
yield row
await _drop_source(row)
class Picker:
"""挑表用的假模型:结构化输出返回自己,ainvoke 记下收到的提示词,按设定回复、报错或拖延。"""
def __init__(self, reply=None, *, error: Exception | None = None, delay: float = 0.0) -> None:
self.reply, self.error, self.delay = reply, error, delay
self.calls: list[str] = []
def with_structured_output(self, schema, **_):
assert schema["title"]
return self
async def ainvoke(self, messages):
self.calls.append("\n".join(text for _role, text in messages))
if self.delay:
await asyncio.sleep(self.delay)
if self.error is not None:
raise self.error
return self.reply
def _use_picker(monkeypatch, picker: Picker) -> list:
"""挑表模型换成假的,返回每次拿模型时的规格(核对思考关、额度小)。"""
import app.api.copilot_context as copilot_context
specs: list = []
async def _model(_session, spec):
specs.append(spec)
return picker, "fake-pick"
monkeypatch.setattr(copilot_context, "get_chat_model", _model)
return specs
class Scripted:
"""按剧本吐操作流的建图模型:第一轮吐 first,收到自查请求时吐 fix。记下每次的 system 和 human。"""
def __init__(self, first: list[dict], fix: list[dict] | None = None) -> None:
self.first = [json.dumps(o, ensure_ascii=False) for o in first]
self.fix = [json.dumps(o, ensure_ascii=False) for o in (fix or [{"op": "done"}])]
self.systems: list[str] = []
self.humans: list[str] = []
async def astream(self, messages):
from langchain_core.messages import AIMessageChunk
self.systems.append(next(t for r, t in messages if r == "system"))
human = next(t for r, t in messages if r == "human")
self.humans.append(human)
for line in (self.fix if "上线前自查" in human else self.first):
yield AIMessageChunk(content=line + "\n")
def _graph_using(tool: str) -> list[dict]:
return [
{"op": "add_node", "node": {"id": "start", "type": "input", "label": "输入", "config": {}}},
{"op": "add_node", "node": {"id": "ask", "type": "agent", "label": "查库", "config": {
"prompt": "{{ input.question }}", "tools": [tool], "assign_to": "answer"}}},
{"op": "add_edge", "edge": {"source": "start", "target": "ask"}},
{"op": "add_node", "node": {"id": "out", "type": "output", "label": "成果", "config": {
"fields": [{"name": "answer", "value": "{{ vars.answer }}"}]}}},
{"op": "add_edge", "edge": {"source": "ask", "target": "out"}},
{"op": "done", "explanation": "查一下", "run": True},
]
async def _stream(client, monkeypatch, model, body: dict) -> list[dict]:
import app.api.copilot as copilot
async def _model(*_a, **_k):
return model, "mock-fast"
monkeypatch.setattr(copilot, "get_chat_model", _model)
events: list[dict] = []
async with client.stream("POST", "/api/copilot/generate-stream", json=body) as r:
assert r.status_code == 200, await r.aread()
async for line in r.aiter_lines():
if line.startswith("data: "):
events.append(json.loads(line[6:]))
return events
#: 需求和假模型挑出的表:入园记录、渠道。两者之间没有直接关联,要经关联表 channel_visits
NEED = "各渠道上月的入园人次"
PICKED = {"tables": [{"table": "visits"}, {"table": "channels"}]}
#: 三张表各自独有的字段(describe_table 的写法:列名两个空格类型)
SELECTED_COLUMNS = ("visitor_count INTEGER", "visit_time TEXT", "ticket_no TEXT", "channel_type TEXT",
"channel_code TEXT", "attributed_share REAL")
#: 没挑中的表独有的字段:不该出现在提示词里
OTHER_COLUMNS = ("store_code", "sku", "on_hand", "refund_amount")
JOINS = ("channel_visits.visit_id = visits.id", "channel_visits.channel_id = channels.id")
def _context_op(events: list[dict]) -> dict:
found = [e for e in events if e.get("op") == "context"]
assert len(found) == 1, [e.get("op") for e in events]
return found[0]
# ---------------------------------------------------------------- 组装
async def test_pick_brings_in_the_bridge_table_and_spells_out_fields_and_joins(scenic_source):
from app.api import copilot_context
picker = Picker(PICKED)
plan = await copilot_context.plan_context([scenic_source])
assert plan.needs_pick
ctx = await plan.resolve(picker, need=NEED)
text = ctx.section()
assert len(picker.calls) == 1
# 挑表的输入是需求加紧凑表目录(每张表一行),不是全部字段
assert NEED in picker.calls[0] and "channel_visits" in picker.calls[0]
assert "attributed_share" not in picker.calls[0]
(source,) = ctx.sources
assert source.selected_by == "model"
assert source.tables == ["visits", "channels", "channel_visits"]
for col in SELECTED_COLUMNS:
assert col in text, col
for col in OTHER_COLUMNS:
assert col not in text, col
for cond in JOINS:
assert cond in text, cond
# 其余表只给名字,让模型知道还有别的表
assert "stores" in text and "parks" in text and "v_daily_visits" in text
# 写 SQL 的硬性要求保留
assert "写 SQL 的硬性要求" in text and "FETCH FIRST n ROWS ONLY" in text
async def test_inferred_relations_are_marked_and_catalog_is_attached(scenic_source):
"""visits.gate_id → gates 没有外键约束,是按列名推断的,连接条件要标出来;目录里推断的项同样要标。"""
from app.api import copilot_context
async with SessionLocal() as session:
await catalog.write_entry(session, scenic_source.id, "visits", {
"label": make_item("入园记录", "llm"), "grain": make_item("每张门票每次检票一行", "human")},
if_version=0, actor="王敏")
plan = await copilot_context.plan_context([scenic_source])
ctx = await plan.resolve(Picker({"tables": [{"table": "visits"}, {"table": "gates"}]}), need="各入口闸机的入园人次")
text = ctx.section()
assert f"visits.gate_id = gates.id(多对一){INFERRED_MARK}" in text
assert f"中文名:入园记录{INFERRED_MARK}" in text
assert "粒度:每张门票每次检票一行" in text and f"每张门票每次检票一行{INFERRED_MARK}" not in text
async def test_rejected_relations_are_not_used_for_join_paths(scenic_source):
"""人工驳回的关系不能拿来补路径:驳回 channel_visits → visits 后,关联表不会因为补路径被带进来。"""
from app.api import copilot_context
rels = [dict(r, status="rejected") if r["to_table"] == "visits" else r
for r in catalog.fk_relations(scenic_source.schema_cache, "channel_visits")]
async with SessionLocal() as session:
await catalog.write_entry(session, scenic_source.id, "channel_visits", {"relations": rels},
if_version=0, actor="王敏")
plan = await copilot_context.plan_context([scenic_source])
ctx = await plan.resolve(Picker(PICKED), need=NEED)
assert ctx.sources[0].tables == ["visits", "channels"]
assert "channel_visits.visit_id = visits.id" not in ctx.section()
async def test_picks_are_capped_and_paths_respect_the_total_limit(scenic_source):
from app.api import copilot_context
names = list(scenic_source.schema_cache["tables"])
plan = await copilot_context.plan_context([scenic_source])
ctx = await plan.resolve(Picker({"tables": [{"table": n} for n in reversed(names)]}), need="全部都要")
# 挑中的按模型给的顺序截到上限;补中间表之后合计也不超过总上限
assert ctx.sources[0].picked == list(reversed(names))[:copilot_context.PICK_LIMIT]
assert ctx.sources[0].tables[:copilot_context.PICK_LIMIT] == ctx.sources[0].picked
assert len(ctx.sources[0].tables) <= copilot_context.CONTEXT_TABLE_LIMIT
# 两两都要补中间表的一组:总数不超过上限
spread = ["reviews", "channels", "stores", "tour_groups", "coupons", "products"]
ctx = await plan.resolve(Picker({"tables": [{"table": n} for n in spread]}), need="x")
assert set(spread) <= set(ctx.sources[0].tables)
assert len(ctx.sources[0].tables) <= copilot_context.CONTEXT_TABLE_LIMIT
async def test_tables_used_by_the_current_graph_are_always_included(scenic_source):
"""改图、追问时现有工作流已经查了哪些表是确定的事实:挑表成功时并进来,挑表失败时也给它们的字段。"""
from app.api import copilot_context
graph = {"nodes": [{"id": "q", "type": "tool", "data": {"label": "查门店", "config": {
"tool": f"db_query__{scenic_source.name}", "args": {"sql": "SELECT s.name FROM stores s"}}}}], "edges": []}
plan = await copilot_context.plan_context([scenic_source], graph=graph)
picker = Picker({"tables": [{"table": "visits"}]})
ctx = await plan.resolve(picker, need="再按入园人次排一下")
assert ctx.sources[0].tables[:2] == ["stores", "visits"]
assert "当前工作流已用到" in picker.calls[0] and "stores" in picker.calls[0]
failed = await plan.resolve(Picker(error=RuntimeError("网关超时")), need="再按入园人次排一下")
assert failed.sources[0].selected_by == "fallback"
assert failed.sources[0].tables == ["stores"]
assert "store_code TEXT" in failed.section()
async def test_several_sources_pick_with_source_names(scenic_db, tmp_path):
from app.api import copilot_context
a = await _add_source(scenic_db)
b = await _add_source(str(scenic.build(tmp_path / "b.db")))
try:
plan = await copilot_context.plan_context([a, b])
picker = Picker({"tables": [{"source": a.name, "table": "visits"}, {"source": b.name, "table": "orders"},
f"{b.name}.payments", {"source": "没有这个库", "table": "refunds"}]})
ctx = await plan.resolve(picker, need="订单和入园")
assert f"数据源「{a.name}」" in picker.calls[0] and f"数据源「{b.name}」" in picker.calls[0]
by_name = {s.source.name: s for s in ctx.sources}
assert by_name[a.name].tables == ["visits"]
assert by_name[b.name].tables == ["orders", "payments"]
op = ctx.op()
assert [s["source"] for s in op["sources"]] == [a.name, b.name]
assert all(s["selected_by"] == "model" for s in op["sources"])
finally:
await _drop_source(a)
await _drop_source(b)
async def test_small_source_keeps_detail_and_carries_the_catalog(small_source):
from app.api import copilot_context
async with SessionLocal() as session:
await catalog.write_entry(session, small_source.id, "orders", {
"label": make_item("订单", "llm"),
"columns": {"amount": {"unit": make_item("元", "human")}}}, if_version=0, actor="王敏")
picker = Picker(error=AssertionError("小库不该挑表"))
plan = await copilot_context.plan_context([small_source])
assert not plan.needs_pick
ctx = await plan.resolve(picker, need="订单总额")
text = ctx.section()
assert not picker.calls
# 详细模式的写法照旧:每张表一行带字段
assert "· orders[表]" in text and "· customers[表]" in text
# 额外带上目录和连接条件
assert f"中文名:订单{INFERRED_MARK}" in text and "单位:元" in text
assert f"orders.customer_id = customers.id(多对一){INFERRED_MARK}" in text
op = ctx.op()
assert op["sources"] == [{"source": small_source.name, "tables": ["customers", "orders"], "selected_by": "all",
"total": 2}]
# ---------------------------------------------------------------- 回退:绝不阻断生成
@pytest.mark.parametrize("picker", [
Picker(error=RuntimeError("网关超时")),
Picker(PICKED, delay=5),
Picker({"tables": [{"table": "no_such_table"}, {"table": "也没有这张"}]}),
Picker({"tables": []}),
Picker("我觉得都需要"),
], ids=["error", "timeout", "unknown", "empty", "garbage"])
async def test_pick_failures_fall_back_to_names_only(client, monkeypatch, scenic_source, picker):
import app.api.copilot_context as copilot_context
monkeypatch.setattr(copilot_context, "PICK_TIMEOUT_S", 0.2)
_use_picker(monkeypatch, picker)
model = Scripted(_graph_using(f"db_query__{scenic_source.name}"))
events = await _stream(client, monkeypatch, model, {
"instruction": NEED, "intent": "answer", "datasource_ids": [scenic_source.id]})
op = _context_op(events)
assert op["sources"] == [{"source": scenic_source.name, "tables": [], "selected_by": "fallback",
"total": scenic.TABLE_COUNT + len(scenic.VIEWS), "reason": op["sources"][0]["reason"]}]
assert op["sources"][0]["reason"]
system = model.systems[0]
# 和以前一样只列表名、提醒先查字段
assert "channel_visits" in system and "attributed_share" not in system
assert "写 SQL 前用 db_schema 工具确认字段" in system
# 生成照常进行
assert events[-1]["op"] == "final", events[-1]
# ---------------------------------------------------------------- 流式:context 操作、自查沿用同一份上下文
async def test_stream_reports_the_context_and_prompt_carries_selected_tables(client, monkeypatch, scenic_source):
from app.api import copilot_context
picker = Picker(PICKED)
specs = _use_picker(monkeypatch, picker)
model = Scripted(_graph_using(f"db_query__{scenic_source.name}"))
events = await _stream(client, monkeypatch, model, {
"instruction": NEED, "intent": "answer", "datasource_ids": [scenic_source.id]})
op = _context_op(events)
assert op["sources"] == [{"source": scenic_source.name, "tables": ["visits", "channels", "channel_visits"],
"selected_by": "model", "total": scenic.TABLE_COUNT + len(scenic.VIEWS)}]
assert isinstance(op["elapsed_ms"], int) and op["elapsed_ms"] >= 0
# context 在模型首帧之后、第一条图操作之前
kinds = [e["op"] for e in events]
assert kinds.index("model") < kinds.index("context") < kinds.index("add_node")
# 挑表用同一个助手模型,关掉思考、额度取小
(spec,) = specs
assert spec.thinking == "off" and spec.max_tokens == copilot_context.PICK_MAX_TOKENS
assert spec.max_tokens <= 2048
system = model.systems[0]
for col in SELECTED_COLUMNS:
assert col in system, col
for col in OTHER_COLUMNS:
assert col not in system, col
for cond in JOINS:
assert cond in system, cond
# 工具清单里列出的可查对象,先列挑中的表
tool_line = next(line for line in system.splitlines() if line.startswith(f"- db_query__{scenic_source.name}"))
assert "可查:visits、channels、channel_visits" in tool_line
async def test_self_check_repair_reuses_the_same_context(client, monkeypatch, scenic_source):
picker = Picker(PICKED)
_use_picker(monkeypatch, picker)
tool = f"db_query__{scenic_source.name}"
first = [
{"op": "add_node", "node": {"id": "start", "type": "input", "label": "输入", "config": {}}},
{"op": "add_node", "node": {"id": "q", "type": "tool", "label": "查入园", "config": {
"tool": tool, "args": {"sql": "SELECT v.no_such_col FROM visits v"}, "assign_to": "rows"}}},
{"op": "add_edge", "edge": {"source": "start", "target": "q"}},
{"op": "add_node", "node": {"id": "out", "type": "output", "label": "成果", "config": {
"fields": [{"name": "rows", "value": "{{ vars.rows }}"}]}}},
{"op": "add_edge", "edge": {"source": "q", "target": "out"}},
{"op": "done", "explanation": "查一下", "run": False},
]
fix = [{"op": "update_node", "id": "q", "config": {"args": {"sql": "SELECT v.visitor_count FROM visits v"}}},
{"op": "done"}]
model = Scripted(first, fix)
events = await _stream(client, monkeypatch, model, {
"instruction": NEED, "datasource_ids": [scenic_source.id]})
assert any(e.get("op") == "check" and e.get("status") == "repairing" for e in events)
assert len(model.systems) == 2 and "上线前自查" in model.humans[1]
# 修正轮带的是同一份上下文,没有重新挑表
assert model.systems[1] == model.systems[0]
assert "attributed_share REAL" in model.systems[1]
assert len(picker.calls) == 1
async def test_small_sources_do_not_call_the_picker(client, monkeypatch, small_source):
picker = Picker(error=AssertionError("小库不该挑表"))
_use_picker(monkeypatch, picker)
model = Scripted(_graph_using(f"db_query__{small_source.name}"))
events = await _stream(client, monkeypatch, model, {
"instruction": "订单总额", "intent": "answer", "datasource_ids": [small_source.id]})
op = _context_op(events)
assert op["sources"][0]["selected_by"] == "all" and "elapsed_ms" not in op
assert not picker.calls
assert "· orders[表]" in model.systems[0]
# ---------------------------------------------------------------- 非流式和两个兜底入口:同一套组装,不发 context
async def test_one_shot_generate_uses_the_same_context(client, monkeypatch, scenic_source):
import app.api.copilot as copilot
picker = Picker(PICKED)
_use_picker(monkeypatch, picker)
seen: list = []
class _OneShot:
def with_structured_output(self, *_a, **_k):
return self
async def ainvoke(self, messages):
seen.append(messages)
return {"nodes": [], "edges": []}
async def _model(*_a, **_k):
return _OneShot(), "mock-fast"
monkeypatch.setattr(copilot, "get_chat_model", _model)
r = await client.post("/api/copilot/generate", json={"instruction": NEED, "datasource_ids": [scenic_source.id]})
assert r.status_code == 200, r.text
assert len(picker.calls) == 1 and NEED in picker.calls[0]
system = next(t for role, t in seen[0] if role == "system")
for cond in JOINS:
assert cond in system
assert "attributed_share REAL" in system
def _node(nid, ntype, label, **config):
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label, "config": config}}
async def test_publish_fix_shares_the_context(monkeypatch, scenic_source):
"""发布前修复:现有图查的表一定带上,需求写的是挡住发布的问题;不发 context 操作(返回值里没有)。"""
import app.api.copilot as copilot
picker = Picker({"tables": [{"source": scenic_source.name, "table": "visits"},
{"source": scenic_source.name, "table": "channels"}]})
_use_picker(monkeypatch, picker)
model = Scripted([{"op": "plan", "summary": "看一下"}, {"op": "done", "explanation": "无法修改"}])
async def _model(*_a, **_k):
return model, "mock-fast"
monkeypatch.setattr(copilot, "get_chat_model", _model)
graph = {"nodes": [
_node("start", "input", "周期"),
_node("fetch", "tool", "取数", tool=f"db_query__{scenic_source.name}",
args={"sql": "SELECT SUM(amount) AS gmv FROM store_sales"}),
_node("write", "report", "报告撰写", instructions="写周报", numbers="strict", on_violation="fail"),
_node("done", "output", "出具", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]),
_node("raw", "output", "原始数", fields=[{"name": "数", "value": "{{ nodes.fetch }}"}]),
], "edges": [{"id": f"e{i}", "source": a, "target": b} for i, (a, b) in enumerate(
[("start", "fetch"), ("fetch", "write"), ("write", "done"), ("fetch", "raw")])]}
async with SessionLocal() as session:
out = await copilot.assist_publish_fix(session, graph, level="governed")
assert model.systems, out
assert len(picker.calls) == 1 and "发布前检查" in picker.calls[0]
system = model.systems[0]
assert "attributed_share REAL" in system and "sold_at TEXT" in system
assert "context" not in json.dumps(out, ensure_ascii=False)
async def test_upgrade_shares_the_context(monkeypatch, scenic_source):
import app.api.copilot as copilot
picker = Picker({"tables": [{"source": scenic_source.name, "table": "no_such_table"}]})
_use_picker(monkeypatch, picker)
model = Scripted([{"op": "plan", "summary": "看一下"}, {"op": "done", "explanation": "不改"}])
async def _model(*_a, **_k):
return model, "mock-fast"
monkeypatch.setattr(copilot, "get_chat_model", _model)
graph = {"nodes": [
_node("start", "input", "输入"),
_node("fetch", "tool", "查库", tool=f"db_query__{scenic_source.name}",
args={"sql": "SELECT SUM(amount) AS a, COUNT(*) AS b FROM store_sales"}),
_node("calc", "code", "算比率", language="python", assign_to="calc",
code="import json\nprint(json.dumps({'ratio': 3 / 4}))"),
_node("card", "metrics", "口径卡", metrics=[{"id": "ratio", "name": "客单价", "expression": "vars.calc.ratio"}]),
_node("write", "report", "报告撰写", instructions="写一句话"),
_node("out", "output", "成果", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]),
], "edges": [{"id": f"e{i}", "source": a, "target": b} for i, (a, b) in enumerate(
[("start", "fetch"), ("fetch", "calc"), ("calc", "card"), ("card", "write"), ("write", "out")])]}
async with SessionLocal() as session:
await copilot.assist_upgrade(session, graph, level="published")
assert model.systems and len(picker.calls) == 1
# 挑表失败也不挡:现有图查的表照样给全字段,其余表只列名字
system = model.systems[0]
assert "sold_at TEXT" in system and "attributed_share" not in system
+10 -1
View File
@@ -13,7 +13,7 @@ import mdSamples from '../run/__tests__/markdown-samples.json'
import evidenceFixture from '../run/__tests__/evidence-doc.json'
import evidenceJudge from '../run/__tests__/evidence-judge.json'
import {
CALIBER_ONE_OUTPUT, CALIBERS_OUTPUT, COPILOT_STUCK, COPILOT_TOOLS_DROPPED, MIXED_OUTPUT, abandonedRun, cancelledRun, exhaustedTeam, longLoop,
CALIBER_ONE_OUTPUT, CALIBERS_OUTPUT, COPILOT_CONTEXT, COPILOT_CONTEXT_FALLBACK, COPILOT_STUCK, COPILOT_TOOLS_DROPPED, MIXED_OUTPUT, abandonedRun, cancelledRun, exhaustedTeam, longLoop,
markupRun, mergeRun, mixedRun, pipelineRun, repairRun, teamRun, timeoutRun,
} from '../run/__tests__/synthetic'
import '../index.css'
@@ -323,6 +323,8 @@ function synthetic(name: string): StreamTurn[] {
// 问数据的一轮:先建图、再执行。执行开始后「规划」收成一行
const plan = decodeCopilot([
{ op: 'heartbeat', phase: 'planning', elapsed_ms: 6400 },
{ op: 'context', elapsed_ms: 2100, sources: [
{ source: 'warehouse', tables: ['v_device_kpi', 'dim_device'], selected_by: 'model', total: 53 }] },
{ op: 'thinking', delta: '要先查表结构,再按设备汇总 KPI。' },
{ op: 'plan', summary: '取数 → 汇总 → 出结论' },
{ op: 'add_node', node: { id: 'q', type: 'tool', label: '查询设备 KPI' } },
@@ -337,6 +339,13 @@ function synthetic(name: string): StreamTurn[] {
graph: { nodes: [{ id: 'q', type: 'tool', data: { label: '查询设备 KPI' } },
{ id: 'o', type: 'output', data: { label: '成果' } }] } }]
}
case 'copilot-context':
// 大库按需求挑表:过程里一行「参考了 N 张表」,展开按数据源分组列表名
return [{ id: 'copilot-context', question: '各渠道上月的入园人次', phase: 'done', status: '已应用到画布',
steps: decodeCopilot(COPILOT_CONTEXT, { context: 'canvas' }) }]
case 'copilot-fallback':
return [{ id: 'copilot-fallback', question: '各渠道上月的入园人次', phase: 'done', status: '已应用到画布',
steps: decodeCopilot(COPILOT_CONTEXT_FALLBACK, { context: 'canvas' }) }]
case 'copilot':
return [{ id: 'copilot', question: '做一个每天的出勤日报', phase: 'done',
status: '已应用到画布,仍有 2 处问题待处理',
+35
View File
@@ -434,6 +434,41 @@ export const COPILOT_TOOLS_DROPPED = [
] },
]
/**
* 助手按需求挑表(context 操作,copilot_context.DatasourceContext.op):大库挑了 3 张(含补进来的关联表),
* 小库全部表都带字段。挑表用了 3 秒多,先到一拍心跳,再是 context,然后照常思考、起草、自查
*/
export const COPILOT_CONTEXT = [
{ op: 'model', model: 'claude-sonnet-4' },
{ op: 'heartbeat', phase: 'planning', elapsed_ms: 3000 },
{ op: 'context', elapsed_ms: 3420, sources: [
{ source: 'scenic', tables: ['visits', 'channels', 'channel_visits'], selected_by: 'model', total: 51 },
{ source: 'shop', tables: ['orders', 'customers'], selected_by: 'all', total: 2 },
] },
{ op: 'heartbeat', phase: 'planning', elapsed_ms: 6000 },
{ op: 'thinking', delta: '入园记录不带渠道,要经 channel_visits 连到渠道表,再按渠道汇总上月的入园人次。' },
{ op: 'plan', summary: '取数 → 出结论' },
{ op: 'add_node', node: { id: 'q', type: 'tool', label: '查询各渠道入园人次' } },
{ op: 'add_node', node: { id: 'o', type: 'output', label: '成果' } },
{ op: 'done', explanation: '两步' },
{ op: 'check', status: 'passed', repaired: 0 },
{ op: 'final', graph: { nodes: [{ id: 'q' }, { id: 'o' }] } },
]
/** 挑表失败(超时):只给了表名,原因随 context 带来;生成照常进行 */
export const COPILOT_CONTEXT_FALLBACK = [
{ op: 'model', model: 'claude-sonnet-4' },
{ op: 'heartbeat', phase: 'planning', elapsed_ms: 18000 },
{ op: 'context', elapsed_ms: 20004, sources: [
{ source: 'scenic', tables: [], selected_by: 'fallback', total: 51, reason: '挑选数据表超过 20 秒未完成' },
] },
{ op: 'plan', summary: '先查表结构,再取数' },
{ op: 'add_node', node: { id: 's', type: 'tool', label: '查看表结构' } },
{ op: 'add_node', node: { id: 'o', type: 'output', label: '成果' } },
{ op: 'done', explanation: '两步' },
{ op: 'final', graph: { nodes: [{ id: 's' }, { id: 'o' }] } },
]
/** Copilot 操作流:思考被心跳打断、自查修一轮没修好、模型写了不存在的节点类型 */
export const COPILOT_STUCK = [
{ op: 'heartbeat', phase: 'planning', elapsed_ms: 3000 },
+87
View File
@@ -2487,6 +2487,83 @@ function labelsOf(ops: CopilotOp[], fallback?: (id: string) => string | undefine
return (id) => map.get(id) ?? fallback?.(id)
}
/** context 操作里一个数据源的那一项(服务端 copilot_context.DatasourceContext.op) */
export interface CopilotContextSource {
source: string
/** 模型看得到全部字段的表(全名)。只给了表名的不在这里 */
tables: string[]
/** model:按需求挑的;all:库不大,全部表都带字段;fallback:没能挑出来,只给了表名 */
selectedBy: 'model' | 'all' | 'fallback'
/** 这个库一共几张表 */
total?: number
/** 没能挑出来的原因 */
reason?: string
}
/** 挑表失败时的说法:模型这一轮只看到表名,字段要它自己去查 */
export const CONTEXT_FALLBACK_TEXT = '未能按需求挑选,已提供全部表名'
function contextSources(op: CopilotOp): CopilotContextSource[] {
const list: unknown[] = Array.isArray(op.sources) ? op.sources : []
return list
.filter((x): x is Record<string, any> => !!x && typeof x === 'object' && typeof (x as any).source === 'string')
.map((x) => {
const total = num(x.total)
return {
source: x.source,
tables: Array.isArray(x.tables) ? x.tables.map(String) : [],
selectedBy: x.selected_by === 'model' || x.selected_by === 'fallback' ? x.selected_by : 'all',
...(total != null ? { total } : {}),
...(typeof x.reason === 'string' && x.reason ? { reason: x.reason } : {}),
}
})
}
/** 一个库在展开区里的那一行:怎么来的 + 表名 */
function contextLine(g: CopilotContextSource): string {
const names = g.tables.join('、')
const of = g.total != null ? `共 ${g.total} 张表` : ''
switch (g.selectedBy) {
case 'all':
return `「${g.source}」全部 ${g.tables.length} 张表:${names}`
case 'model':
return g.tables.length
? `「${g.source}」按需求${g.total != null ? `从 ${g.total} 张表中` : ''}挑出 ${g.tables.length} 张:${names}`
: `「${g.source}」${of ? `${of},` : ''}未挑中与需求相关的表,已提供全部表名`
default:
return `「${g.source}」${of ? `${of},` : ''}${CONTEXT_FALLBACK_TEXT}`
+ (g.tables.length ? `;现有工作流用到的 ${g.tables.length} 张表附带全部字段:${names}` : '')
}
}
/**
* 助手这一轮参考了哪些表(context 操作)→ 过程里的一行。
*
* 大库放不下全部字段,服务端先按需求挑表,挑中的表给模型全部字段。这件事要看得见:生成的 SQL 用错了表,
* 先得知道模型当时看到的是哪几张。标题只说几张(「参考了 8 张表」),展开按数据源分组列表名;挑表失败时
* 标题直说只给了表名,原因放在展开区。认不出的形状返回 null:不出这一行,也不报错。
*/
export function copilotContext(op: CopilotOp): { title: string; sub?: string; detail: string; ms?: number } | null {
const groups = contextSources(op)
if (!groups.length) return null
const count = groups.reduce((n, g) => n + g.tables.length, 0)
const failed = groups.filter((g) => g.selectedBy === 'fallback')
// 有表带着字段(小库,或者现有工作流用到的表)时标题仍说几张,没挑成的事放副标题
const sub = count && failed.length
? failed.length === groups.length ? CONTEXT_FALLBACK_TEXT
: `「${failed.map((g) => g.source).join('」「')}」${CONTEXT_FALLBACK_TEXT}`
: undefined
const reasons = [...new Set(failed.map((g) => g.reason).filter(Boolean))]
// 挑过表才有 elapsed_ms:挑表是生成之前多出来的一次模型调用,没挑成(比如超时)也要让人看到等了多久
const ms = num(op.elapsed_ms)
return {
title: count ? `参考了 ${count} 张表` : CONTEXT_FALLBACK_TEXT,
...(sub ? { sub } : {}),
detail: [...groups.map(contextLine), ...reasons.map((r) => `原因:${r}`)].join('\n'),
...(ms != null && ms >= 10 ? { ms } : {}),
}
}
/** 用了限定范围之外的数据源:自查交回模型改也改不掉时,人要知道是范围的事,不是图写错了 */
const OUT_OF_SCOPE_NEXT = '本轮限定了查询的数据源,但工作流使用了范围之外的数据库。请取消限定后重新提问,或将所需数据库加入查询范围'
@@ -2591,6 +2668,16 @@ export function decodeCopilot(ops: CopilotOp[], opts?: {
...(elapsed ? { ms: elapsed, meta: formatDuration(elapsed) } : {}) })
break
}
case 'context': {
// 这一轮参考了哪些表:标题说几张,展开看按数据源分组的表名
const ctx = copilotContext(op)
if (!ctx) break
add({ id: `cx-${i}`, seq: i, kind: 'schema', status: 'done', code: 'copilot_context',
title: ctx.title, detail: ctx.detail,
...(ctx.sub ? { sub: ctx.sub } : {}),
...(ctx.ms != null ? { ms: ctx.ms, meta: formatDuration(ctx.ms) } : {}) })
break
}
case 'plan':
add({ id: `cp-${i}`, seq: i, kind: 'note', status: 'done',
title: clip(String(op.summary ?? '已完成规划'), 120) })
+7 -1
View File
@@ -12,7 +12,7 @@ import { humanizeError } from '../lib/errors'
import { formatShortcut } from '../lib/keys'
import { UPGRADE_TEXT } from '../lib/terms'
import type { CopilotOp } from '../run/decode'
import { endingOf, reduceTeam, settleTeam } from '../run/decode'
import { copilotContext, endingOf, reduceTeam, settleTeam } from '../run/decode'
import { activeEdgesOf, applyDerived } from '../run/derive'
import {
emptyTrace, finalizeTrace, foldEvent, isActivePhase, isSettled, isTerminal, runStatusOf,
@@ -1888,6 +1888,12 @@ export const useStudio = create<StudioState>((set, get) => ({
},
})
break
case 'context': {
// 服务端按需求挑了表:浮条和轮次卡上先说参考了几张表,展开的过程里有表名(decodeCopilot)
const ctx = copilotContext(op)
if (ctx) set({ copilot: { ...s.copilot, lastOp: ctx.title } })
break
}
case 'plan':
set({ copilot: { ...s.copilot, phase: 'planning', lastOp: op.summary ?? '规划中' } })
break
+13
View File
@@ -472,6 +472,9 @@ async function fakeApi(route) {
}
const ops = [
{ op: 'heartbeat', phase: 'planning', elapsed_ms: 800 },
// 服务端按需求挑了表(copilot_context):过程里多一行「参考了 N 张表」
{ op: 'context', elapsed_ms: 1200, sources: [
{ source: 'shop', tables: ['orders', 'order_items'], selected_by: 'model', total: 48 }] },
{ op: 'plan', summary: '查一下再回答' },
...GRAPH.nodes.map((n) => ({ op: 'add_node', node: n })),
{ op: 'done', explanation: '一张三步的图' },
@@ -1024,6 +1027,16 @@ for (const theme of THEMES) {
check('完整答案取到了,meta 里没有「没取全」', !meta?.clipped, JSON.stringify(meta?.clipped))
await page.getByText('展开全部').first().click().catch(() => {})
check('完整结尾在页面上', await shows(page, '【完整结尾】'))
// 建图时服务端说了这一轮参考了哪些表:收在「规划」里,点开是一行,再点开看表名
check('建图的 context 操作原样记进这一轮', t.ops.some((o) => o.op === 'context'))
const turnCard = page.locator('[data-turn]').last()
await turnCard.locator('button', { hasText: '规划' }).first().click()
const ctxRow = turnCard.locator('[data-step-code="copilot_context"]')
check('点开「规划」看到参考了几张表', await ctxRow.waitFor({ timeout: 3000 }).then(() => true, () => false)
&& (await ctxRow.innerText()).includes('参考了 2 张表'))
await ctxRow.locator('button').first().click()
check('再点开看到按数据源分组的表名', await ctxRow.getByText('「shop」按需求从 48 张表中挑出 2 张:orders、order_items')
.waitFor({ timeout: 3000 }).then(() => true, () => false))
check('没有运行时报错', errors.length === 0, errors[0] ?? '')
await ctx.close()
})
+5 -2
View File
@@ -232,9 +232,12 @@ const MODEL_ONLY = {
'_STREAM_PROTOCOL', '_RULES_HEAD', '_NO_LOWERING', '_ASK_HUMAN', '_UPGRADE_TASK', 'CELL_RULES', 'ENTITY_RULES',
'EXTRACT_SYSTEM', 'GRAPH_SCHEMA', 'REVIEW_SCHEMA', '_CN_N',
]),
// 拼提示词的函数:生成 / 修正 / 升级请求、抽取消息、交给抽取模型的输出结构
// 拼提示词的函数:生成 / 修正 / 升级请求、抽取消息、交给抽取模型的输出结构;
// 助手的数据源上下文(api/copilot_context.py 的各段渲染、挑表请求,copilot.py 里挑表用的需求)
funcs: new Set(['_user_message', '_repair_request', '_publish_fix_request', '_upgrade_assist_request',
'_extract_messages', '_from_rows', 'catalog_prompt']),
'_extract_messages', '_from_rows', 'catalog_prompt',
'_pick_messages', '_pick_need', '_publish_fix_need', '_upgrade_need',
'_detail_block', '_selected_block', '_join_lines', '_route_line', '_source_head']),
// 第二个参数是交回模型改写的指令(第一个是给人看的那句):engine/evidence.py 的 _Reason(text, fix)、fail(text, fix)
fixArg: new Set(['_Reason', 'fail']),
names_re: /(?:^|_)(?:SYSTEM|PROMPT|NUDGE)(?:_|$)/,
+43
View File
@@ -648,6 +648,49 @@ await section('从 SQL 认表名', async () => {
check('不是查询就不硬认', g('这不是 SQL') === null)
})
await section('Copilot:这一轮参考了哪些表(context 操作)', async () => {
// 大库先按需求挑表,挑中的表给模型全部字段。生成的 SQL 用错了表时,先得知道模型当时看到的是哪几张
const steps = mod.decodeCopilot(synthetic.COPILOT_CONTEXT, { context: 'canvas' })
const row = steps.find((s) => s.code === 'copilot_context')
check('一行说清参考了几张表', row?.title === '参考了 5 张表', row?.title)
check('展开按数据源分组列出表名', !!row?.detail?.includes('「scenic」按需求从 51 张表中挑出 3 张:visits、channels、channel_visits')
&& row.detail.includes('「shop」全部 2 张表:orders、customers'), row?.detail)
check('挑表的耗时挂在行尾', row?.ms === 3420 && !!row?.meta, `${row?.ms} ${row?.meta}`)
check('是看表结构那一类、标着规划阶段', row?.kind === 'schema' && row?.stage === 'plan' && row?.status === 'done')
check('排在思考之前,心跳仍然只更新那一条阶段行',
steps.indexOf(row) < steps.findIndex((s) => s.kind === 'think')
&& steps.filter((s) => s.kind === 'lifecycle' && s.id.startsWith('hb-')).length === 1,
steps.map((s) => `${s.kind}:${s.title}`).join(' | '))
check('收尾后没有还在转的行', !steps.some((s) => s.status === 'running'),
steps.filter((s) => s.status === 'running').map((s) => s.title).join(','))
check('结局不受影响', mod.copilotOutcome(synthetic.COPILOT_CONTEXT).kind === 'built')
const fb = mod.decodeCopilot(synthetic.COPILOT_CONTEXT_FALLBACK, { context: 'chat' }).find((s) => s.code === 'copilot_context')
check('挑表失败:标题直说只给了表名', fb?.title === '未能按需求挑选,已提供全部表名', fb?.title)
check('挑表失败:原因和库的规模放在展开区', !!fb?.detail?.includes('原因:挑选数据表超过 20 秒未完成')
&& fb.detail.includes('「scenic」共 51 张表'), fb?.detail)
check('挑表失败也说等了多久', fb?.ms === 20004, String(fb?.ms))
const seeded = mod.copilotContext({ op: 'context', sources: [
{ source: 'scenic', tables: ['stores'], selected_by: 'fallback', total: 51, reason: '模型未挑出任何表' }] })
check('没挑成但现有工作流的表带着字段:标题说几张、副标题说没挑成',
seeded?.title === '参考了 1 张表' && seeded?.sub === '未能按需求挑选,已提供全部表名'
&& seeded.detail.includes('现有工作流用到的 1 张表附带全部字段:stores'), JSON.stringify(seeded))
const mixed = mod.copilotContext({ op: 'context', sources: [
{ source: 'a', tables: ['x'], selected_by: 'all', total: 1 },
{ source: 'b', tables: [], selected_by: 'fallback', total: 80, reason: 'r' }] })
check('只有一个库没挑成:副标题点名是哪个库', mixed?.sub === '「b」未能按需求挑选,已提供全部表名', mixed?.sub)
const none = mod.copilotContext({ op: 'context', sources: [
{ source: 'a', tables: ['x'], selected_by: 'model', total: 60 },
{ source: 'b', tables: [], selected_by: 'model', total: 70 }] })
check('挑表成功但某个库一张没挑中:说清只给了表名', !!none?.detail.includes('「b」共 70 张表,未挑中与需求相关的表,已提供全部表名'),
none?.detail)
check('形状不对不出行、不报错', mod.decodeCopilot([
{ op: 'context' }, { op: 'context', sources: 'x' }, { op: 'context', sources: [null, 3, { tables: ['t'] }] },
]).length === 0)
check('认不出的操作照旧忽略', mod.decodeCopilot([{ op: 'something_new', x: 1 }]).length === 0)
})
await section('Copilot:心跳穿插、回话、少了一步、报错', async () => {
const steps = mod.decodeCopilot(synthetic.COPILOT_STUCK, { context: 'canvas' })
check('心跳穿插的思考仍然并成一条', steps.filter((s) => s.kind === 'think').length === 1,
+37
View File
@@ -731,6 +731,43 @@ await section('合并查询:输入、合并 SQL、结果预览,警告各占
await narrow.page.close()
})
await section('参考了哪些表:画布右栏和问数据页(context 操作)', async () => {
const c = await open('syn=copilot-context&dense=1', { w: 380, h: 800, name: 'copilot-context' })
const row = c.page.locator('[data-step-code="copilot_context"]')
check('画布右栏:一行说参考了几张表', await row.count() === 1 && (await row.innerText()).includes('参考了 5 张表'),
await row.count() ? await row.innerText() : '没有这一行')
check('表名收在展开区,不铺满窄栏', !(await row.innerText()).includes('channel_visits'))
await row.locator('button').first().click()
await c.page.waitForTimeout(150)
const expanded = await row.innerText()
check('展开看到按数据源分组的表名', expanded.includes('「scenic」按需求从 51 张表中挑出 3 张')
&& expanded.includes('channel_visits') && expanded.includes('「shop」全部 2 张表'), expanded.slice(0, 160))
const overflow = await c.page.evaluate(() => document.documentElement.scrollWidth - document.documentElement.clientWidth)
check('展开后不撑破 380px 窄栏', overflow <= 0, `${overflow}px`)
check('没有运行时报错', c.errors.length === 0, c.errors.join(' | '))
await c.page.close()
const f = await open('syn=copilot-fallback&dense=1', { w: 380, h: 800, name: 'copilot-fallback' })
const fb = f.page.locator('[data-step-code="copilot_context"]')
check('挑表失败:写明只给了表名', await fb.count() === 1 && (await fb.innerText()).includes('未能按需求挑选,已提供全部表名'))
await fb.locator('button').first().click()
await f.page.waitForTimeout(150)
check('挑表失败:展开看到原因', (await fb.innerText()).includes('原因:挑选数据表超过 20 秒未完成'))
await f.page.close()
const q = await open('syn=chat', { w: 1100, h: 800, name: 'chat-context' })
check('问数据页:执行开始后这一行随「规划」收起', !(await q.page.locator('body').innerText()).includes('参考了 2 张表'))
await q.page.locator('button', { hasText: '规划' }).first().click()
await q.page.waitForTimeout(200)
const qrow = q.page.locator('[data-step-code="copilot_context"]')
check('问数据页:点开「规划」看到参考了几张表', await qrow.count() === 1 && (await qrow.innerText()).includes('参考了 2 张表'))
await qrow.locator('button').first().click()
await q.page.waitForTimeout(150)
check('问数据页:展开看到表名', (await qrow.innerText()).includes('v_device_kpi、dim_device'))
check('问数据页:没有运行时报错', q.errors.length === 0, q.errors.join(' | '))
await q.page.close()
})
await section('步骤行和画布联动', async () => {
const { page } = await open('syn=mixed&link=1', { w: 1100, h: 800 })
const row = page.locator('[data-node-id="agent"]').first()
+9
View File
@@ -1043,6 +1043,9 @@ await section('助手改图:一步撤销、回执、只回话、失败退回
}
state.stream = [
{ op: 'model', model: 'fake-model' },
// 服务端按需求挑了表(copilot_context):过程里多一行「参考了 N 张表」
{ op: 'context', elapsed_ms: 1500, sources: [
{ source: 'shop', tables: ['orders', 'order_items', 'customers'], selected_by: 'model', total: 48 }] },
{ op: 'plan', summary: '加一步润色' },
{ op: 'update_node', id: 'answer', config: finalGraph.nodes[3].data.config },
{ op: 'add_node', node: { id: 'polish', type: 'llm', label: '润色', config: { prompt: '{{ nodes.answer.text }}' } } },
@@ -1059,6 +1062,12 @@ await section('助手改图:一步撤销、回执、只回话、失败退回
const t1 = await S(page, () => window.__studio.getState().copilotTurns.at(-1))
check('开始了就告诉输入框(runCopilot 返回 true)', began === true, String(began))
check('收到 final:这一轮是「已应用」', t1.outcome === 'applied', t1.outcome)
const ctxRow = page.locator('[data-step-code="copilot_context"]').last()
check('助手栏的过程里一行说参考了几张表', await ctxRow.waitFor({ timeout: 3000 }).then(() => true, () => false)
&& (await ctxRow.innerText()).includes('参考了 3 张表'))
await ctxRow.locator('button').first().click()
check('点开看到按数据源分组的表名', await ctxRow.getByText('「shop」按需求从 48 张表中挑出 3 张:orders、order_items、customers')
.waitFor({ timeout: 3000 }).then(() => true, () => false))
check('diff 记下新增和改动', t1.diff?.added.includes('polish') && t1.diff?.changed.includes('answer'), JSON.stringify(t1.diff))
check('final.issues 进了这一轮(少了一步)', t1.issues?.some((i) => i.code === 'unknown_node_type'))
const pos1 = await S(page, () => Object.fromEntries(window.__studio.getState().nodes.map((n) => [n.id, n.position])))