数据目录阶段 1B 流式输出:挑表后发 context 操作,说明这一轮参考了哪些表、怎么挑的、挑表用时,挑表失败时带上原因

This commit is contained in:
ThearchyHelios
2026-10-03 10:19:41 +08:00
parent e74097a103
commit 9e809bafe6
3 changed files with 58 additions and 12 deletions
+4 -2
View File
@@ -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")
+24 -2
View File
@@ -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:
+30 -8
View File
@@ -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):