diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index 66673160..87782cea 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -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) diff --git a/backend/app/api/copilot_context.py b/backend/app/api/copilot_context.py new file mode 100644 index 00000000..d2893c1f --- /dev/null +++ b/backend/app/api/copilot_context.py @@ -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", +] diff --git a/backend/tests/test_copilot_table_pick.py b/backend/tests/test_copilot_table_pick.py new file mode 100644 index 00000000..7e249404 --- /dev/null +++ b/backend/tests/test_copilot_table_pick.py @@ -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 diff --git a/frontend/src/dev/preview.tsx b/frontend/src/dev/preview.tsx index 5743eb8c..a1354c35 100644 --- a/frontend/src/dev/preview.tsx +++ b/frontend/src/dev/preview.tsx @@ -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 处问题待处理', diff --git a/frontend/src/run/__tests__/synthetic.ts b/frontend/src/run/__tests__/synthetic.ts index ed777ba8..4ecc8fe8 100644 --- a/frontend/src/run/__tests__/synthetic.ts +++ b/frontend/src/run/__tests__/synthetic.ts @@ -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 }, diff --git a/frontend/src/run/decode.ts b/frontend/src/run/decode.ts index 5348c930..f9ecb1ff 100644 --- a/frontend/src/run/decode.ts +++ b/frontend/src/run/decode.ts @@ -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 => !!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) }) diff --git a/frontend/src/store/studio.ts b/frontend/src/store/studio.ts index e8478261..3ea731c1 100644 --- a/frontend/src/store/studio.ts +++ b/frontend/src/store/studio.ts @@ -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((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 diff --git a/scripts/check-chat.mjs b/scripts/check-chat.mjs index ee36a40e..40fd88e5 100644 --- a/scripts/check-chat.mjs +++ b/scripts/check-chat.mjs @@ -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() }) diff --git a/scripts/check-copy.mjs b/scripts/check-copy.mjs index df5bc5f2..2376d6f1 100644 --- a/scripts/check-copy.mjs +++ b/scripts/check-copy.mjs @@ -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)(?:_|$)/, diff --git a/scripts/check-decode.mjs b/scripts/check-decode.mjs index e8f4e9ee..cec98cab 100644 --- a/scripts/check-decode.mjs +++ b/scripts/check-decode.mjs @@ -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, diff --git a/scripts/check-stream.mjs b/scripts/check-stream.mjs index b7938867..82427557 100644 --- a/scripts/check-stream.mjs +++ b/scripts/check-stream.mjs @@ -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() diff --git a/scripts/check-studio.mjs b/scripts/check-studio.mjs index 8177f7a6..a66c5282 100644 --- a/scripts/check-studio.mjs +++ b/scripts/check-studio.mjs @@ -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])))