diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index e02fc3ae..4f008d48 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -130,7 +130,7 @@ async def _datasource_context( ) -> copilot_context.DatasourceContext: """这一轮的数据源上下文:读目录、大库按需求挑表、组装(copilot_context)。挑表失败退回只列表名,不抛异常。 - 非流式生成、发布前修复、升级都走这里;流式生成把三步拆开,挑表放进流里(期间发心跳)。 + 非流式生成、发布前修复、升级都走这里;流式生成把三步拆开,挑表放进流里(期间发心跳、挑完发 context 操作)。 graph 是现有工作流:它的 SQL 查过的表一律给全字段。 """ plan = await copilot_context.plan_context(sources, graph=graph) @@ -666,7 +666,7 @@ async def generate( 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") @@ -1704,6 +1704,8 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g # 这一份 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") diff --git a/backend/app/api/copilot_context.py b/backend/app/api/copilot_context.py index e7bcf9f6..3a5e3aa2 100644 --- a/backend/app/api/copilot_context.py +++ b/backend/app/api/copilot_context.py @@ -22,7 +22,7 @@ **用法。** plan_context 读目录、按大小定每个源怎么给(不调模型,接口函数在开流之前调,因为请求的会话只在 那时可用);pick_model 拿挑表用的模型(同一个助手模型,关掉思考、额度取小);ContextPlan.resolve 挑表并组装 (流式生成在流里调,期间照常发心跳);DatasourceContext 给出提示词片段(section)、工具清单里列出的可查对象 -(listed)。 +(listed)和告诉前端这一轮参考了哪些表的 context 操作(op)。 """ from __future__ import annotations @@ -376,7 +376,7 @@ class ContextPlan: return out async def resolve(self, model: Any | None, *, need: str, unavailable: str | None = None) -> DatasourceContext: - """挑表并组装。不抛异常:挑表的任何失败都退回只列表名(fallback),原因写进日志。 + """挑表并组装。不抛异常:挑表的任何失败都退回只列表名(fallback),原因写进 context 操作和日志。 model 是 pick_model 拿到的模型;拿不到时为 None,unavailable 是原因。没有要挑表的源时不调模型。 """ @@ -529,6 +529,28 @@ class DatasourceContext: 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: diff --git a/backend/tests/test_copilot_table_pick.py b/backend/tests/test_copilot_table_pick.py index c08f7693..7e249404 100644 --- a/backend/tests/test_copilot_table_pick.py +++ b/backend/tests/test_copilot_table_pick.py @@ -8,7 +8,7 @@ - 挑表、补路径、组装的结果(合成景区库:入园记录和渠道之间要经关联表 channel_visits) - 挑表失败(报错、超时、返回不存在的表、返回为空)一律退回只列表名,生成照常进行 - 小库保持详细模式,额外带上目录 -- 自查修正沿用同一份上下文,不再重新挑表 +- 流式输出里有 context 操作;自查修正沿用同一份上下文,不再重新挑表 - 非流式生成、发布前修复、升级共用同一套组装 """ from __future__ import annotations @@ -176,6 +176,12 @@ 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] + + # ---------------------------------------------------------------- 组装 @@ -289,7 +295,9 @@ async def test_several_sources_pick_with_source_names(scenic_db, tmp_path): 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"] - assert all(s.selected_by == "model" for s in ctx.sources) + 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) @@ -313,7 +321,9 @@ async def test_small_source_keeps_detail_and_carries_the_catalog(small_source): # 额外带上目录和连接条件 assert f"中文名:订单{INFERRED_MARK}" in text and "单位:元" in text assert f"orders.customer_id = customers.id(多对一){INFERRED_MARK}" in text - assert ctx.sources[0].selected_by == "all" and ctx.sources[0].tables == ["customers", "orders"] + op = ctx.op() + assert op["sources"] == [{"source": small_source.name, "tables": ["customers", "orders"], "selected_by": "all", + "total": 2}] # ---------------------------------------------------------------- 回退:绝不阻断生成 @@ -335,6 +345,10 @@ async def test_pick_failures_fall_back_to_names_only(client, monkeypatch, scenic 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 @@ -344,10 +358,10 @@ async def test_pick_failures_fall_back_to_names_only(client, monkeypatch, scenic -# ---------------------------------------------------------------- 流式:自查沿用同一份上下文 +# ---------------------------------------------------------------- 流式:context 操作、自查沿用同一份上下文 -async def test_stream_prompt_carries_selected_tables(client, monkeypatch, scenic_source): +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) @@ -356,7 +370,13 @@ async def test_stream_prompt_carries_selected_tables(client, monkeypatch, scenic events = await _stream(client, monkeypatch, model, { "instruction": NEED, "intent": "answer", "datasource_ids": [scenic_source.id]}) - assert events[-1]["op"] == "final", events[-1] + 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 @@ -409,7 +429,8 @@ async def test_small_sources_do_not_call_the_picker(client, monkeypatch, small_s model = Scripted(_graph_using(f"db_query__{small_source.name}")) events = await _stream(client, monkeypatch, model, { "instruction": "订单总额", "intent": "answer", "datasource_ids": [small_source.id]}) - assert events[-1]["op"] == "final", events[-1] + 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] @@ -450,7 +471,7 @@ def _node(nid, ntype, label, **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"}, @@ -477,6 +498,7 @@ async def test_publish_fix_shares_the_context(monkeypatch, scenic_source): 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):