问数据页:问题终于传进图里了,而且不再被误报卡住

两条日志(e3c2f743 / 61655cfe,相隔 30 秒)是同一件事的两半,指向三个问题。

**一、用户的问题根本没进图。** store/chat.ts 的 launch 里写死 `input: {}`。
Copilot 只要给入口声明了一个必填字段(它有时候会,比如 topic),这次运行就
在第一个节点上立刻挂掉:"缺少必填输入:topic"——用户刚问完就看到报错,而且
错误里那个字段名他从来没见过,那是生成的图里的东西。e3c2f743 就是这么死的。
之前"能跑"纯属运气:那几次 Copilot 恰好没声明字段。

**二、把提问当成了流程描述。** 两个入口共用一句 "请设计一个工作流:<原话>"。
画布上这是对的(用户在描述流程),但在问数据页用户是在**提问**。于是
"shop 里都有哪些数据?"被设计成了一个"给这段文字生成摘要"的流程,跑完把
用户的问题原样复述一遍还回去(61655cfe 的成果就是这个)。加 intent 区分:
answer 意图下明说"流程是手段,用户要的是答案",并把注入契约写死成
{{ input.question }}——不说死键名,模型会自己编一个(见过 input.text),
下游取到空字符串,而模板取不到值不报错。

**三、我上一个功能把整条对话路径卡住了。** 变量分析器把 `{{ input }}` 报成
"没有任何节点产出"——那是完全合法的写法(template_context 里 input 就是顶层
键),而 runs 启动前要过校验,一条假错误意味着什么都跑不了。这正是我自己在
上个 commit 里写的"误报是 linter 最坏的失败模式",隔天就踩了。

修的时候连着踩了三步,每步都是测试当场抓的:
- 注册裸根之后,input.quesiton 回退到根被当成合法,拼写检查整个失效
- 收紧后 {{ vars.rows[0].name }} 被误报——那是在往值内部钻,结构枚举不了
- 再收紧后 {{ input.question }} 被误报——问数据页的入口**故意不声明字段**

最终规则:裸根合法;input/vars/nodes 只校验到第二段(再往下是值内部);
**入口声明了字段才是闭集,没声明就是开放的**(运行时传什么键都行);
根本不认识的根降成 warning——能穷举才配报 error,假错误比漏报贵得多。

端到端验证:问 "shop 里都有哪些数据?挑一个最大的表说说",现在生成
input→tool→tool→llm→output(带 db_query__shop),input 带着问题,
真查出 backend_log 481,242 行 / 528.64 MB 并列出全部字段。
This commit is contained in:
ThearchyHelios
2026-09-18 13:52:43 +08:00
parent 105120f22f
commit fa6a8a2ffe
5 changed files with 222 additions and 15 deletions
+47 -2
View File
@@ -137,6 +137,14 @@ NODE_REFERENCE = """\
- {{ last_message }} 最近一条消息文本
- 过滤器:{{ vars.x | json }}
**引用必须有来源**(这条会被校验,不满足整张图跑不起来):
- 写 {{ vars.X }} 之前,必须有某个节点的 config.assign_to 正好是 "X"
- 写 {{ input.X }} 之前,X 必须在入口节点的 fields 里声明过
- 写 {{ nodes.Y.* }} 之前,Y 必须是真实存在的节点 id
- 引用的那个节点还必须排在**前面**——后面的节点产出的值,前面取不到
模板取不到值时渲染成空字符串、不报错,所以这类错误在运行期完全静默:
下游拿到一段空文本,然后一本正经地基于空内容编一段回答出来。
连线规则:
- edges 里每条边 {source, target, sourceHandle?}
- branch 节点的出边必须带 sourceHandle,取值是 cases 里的 key,另外要有一条 sourceHandle="default" 兜底
@@ -188,6 +196,43 @@ class GenerateIn(BaseModel):
base_graph: dict[str, Any] | None = Field(default=None, description="在现有图上修改时传入")
provider: str | None = None
model: str | None = None
#: build = 用户在描述一个流程(画布);answer = 用户在问一个问题(问数据页)
intent: str = Field(default="build")
def _user_message(payload: GenerateIn, *, patch: bool) -> str:
"""把用户说的话包成给模型的请求。
两个入口的语义完全不同,包法必须跟着变:
- 画布上,用户**在描述一个流程**("读用户的问题,先查知识库…"),
"请设计一个工作流:<原话>" 是对的。
- 问数据页,用户**在问一个问题**("shop 里都有哪些数据?")。同样
包成"请设计一个工作流:shop 里都有哪些数据?",模型会老老实实设计
一个**关于这句话**的流程——实际见过它生成"给这段文字生成摘要",
跑完把用户的问题原样复述一遍还回去。流程是手段,不是用户要的东西。
"""
if patch:
return (
"这是当前的工作流:\n"
"请按下面的要求修改它(只输出改动操作):\n" + payload.instruction
)
if payload.intent == "answer":
return (
"用户问了下面这个问题。请设计一个**能回答它**的工作流——"
"流程是手段,用户要的是答案。\n\n"
f"问题:{payload.instruction}\n\n"
"这条路径的硬性约定:\n"
# 问题由 ChatPage 注入到 input.question。不把键名说死,模型会自己
# 编一个(见过 input.text、input.topic),下游取到的就是空字符串——
# 而模板取不到值不报错,于是模型一本正经地基于空内容编一段回答
"- 问题已经由系统注入成 `{{ input.question }}`,要用就写这一个。"
"入口节点**不要声明任何字段**,也不会有人再填一次表单\n"
"- 涉及数据的问题必须用 db_query__* / db_schema__* 去真查,"
"不要只对问题本身做文字加工(改写、摘要、分类都不是回答)\n"
"- 出口节点给出的应当是这个问题的答案本身"
)
return f"请设计一个工作流:{payload.instruction}"
COPILOT_SETTING_KEY = "copilot"
@@ -332,7 +377,7 @@ async def generate(
f"请按下面的要求修改它,返回完整的新图(不是补丁):\n{payload.instruction}"
)
else:
user = f"请设计一个工作流:{payload.instruction}"
user = _user_message(payload, patch=False)
try:
model, _ = await get_chat_model(session, await copilot_model_spec(session, payload))
@@ -587,7 +632,7 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g
f"请按下面的要求修改它(只输出改动操作):\n{payload.instruction}"
)
else:
user = f"请设计一个工作流:{payload.instruction}"
user = _user_message(payload, patch=False)
try:
spec_ = await copilot_model_spec(session, payload)
+59 -8
View File
@@ -132,6 +132,43 @@ def _label_of(node: GraphNode) -> str:
return node.data.label or node.id
def _resolve_fallback(
path: str, index: dict[str, "Variable"], *, input_is_closed: bool = True
) -> "Variable | None":
"""精确路径没命中时,看看能不能退一层认下来。
分寸全在这里:退得太松,`{{ input.quesiton }}` 会退到 `input` 被当成合法,
拼写检查整个失效;退得太紧,`{{ loops.item }}` 会被误报——而 loops 的成员
是运行期才有的,我根本枚举不了。
所以按命名空间分开对待:**成员可枚举的不许退,不可枚举的退到根**。
"""
parts = [p for p in path.split(".") if p]
if not parts:
return None
root = parts[0]
# 裸根本身就是合法引用:{{ input }} 取整个输入 dict
if len(parts) == 1:
return index.get(root)
# 入口不声明字段时,input.* 是开放的:运行时传什么键都行(问数据页就是
# 这样——问题由系统注入成 input.question,入口节点里一个字段都没有)。
# 声明了字段才是闭集,那时候拼错才叫拼错。
if root == "input" and not input_is_closed:
return index.get("input")
# 这三个的第二段可枚举(入口字段名、assign_to、节点 id),必须存在。
# 再往下就是在往那个值**内部**钻了(vars.rows[0].name、nodes.a.text),
# 值的结构是运行期才有的,枚举不了,所以只校验到第二段
if root in ("input", "vars", "nodes"):
return index.get(f"{root}.{parts[1]}")
# loops.item / usage.total_tokens / messages[0].content 这类,
# 成员是运行期的状态,枚举不了,根存在就算数
return index.get(root)
def analyze(spec: GraphSpec) -> VariableReport:
"""静态分析一张图的变量。不需要跑,也不需要有运行记录。"""
depth = _depths(spec)
@@ -183,11 +220,19 @@ def analyze(spec: GraphSpec) -> VariableReport:
produced_by=node.id, produced_by_label=label, order=order,
)
# 每个根命名空间本身也是合法引用:{{ input }} 取整个输入 dict,
# {{ vars }} 取整个变量池——template_context 里它们就是顶层的键。
# 上一版把 input/nodes/vars 跳过了(想着"具体成员已经列过"),结果
# {{ input }} 被报成"没有任何节点产出",而那是完全正确的写法。
# 误报会把整张图卡住:runs 启动前要过校验,一条假错误就意味着什么都跑不了。
for root, desc in _BUILTIN_ROOTS.items():
if root in ("input", "nodes", "vars"):
continue # 这三个的具体成员上面已经列过了
ensure(root, kind="builtin", label=desc)
# 入口声明了字段,input.* 才是闭集;一个字段都没声明时运行时传什么键都行
input_is_closed = any(
n.type == "input" and (n.data.config or {}).get("fields") for n in spec.nodes
)
# ---- 引用侧 ----
for node in spec.nodes:
cfg = node.data.config or {}
@@ -203,12 +248,9 @@ def analyze(spec: GraphSpec) -> VariableReport:
continue
ref = VarRef(node_id=node.id, node_label=label, field=field, expr=expr)
# 精确命中(input.x / vars.x / nodes.x)或者根命中(last_message)
target = index.get(path)
if target is None and root in _BUILTIN_ROOTS:
# nodes.foo.text 这种,退到 nodes.foo 这一层
parent = ".".join(path.split(".")[:2])
target = index.get(parent) or index.get(root)
if target is None:
target = _resolve_fallback(path, index, input_is_closed=input_is_closed)
if target is not None:
target.refs.append(ref)
# 变量存在,但产出它的节点排在引用者之后——跑到这儿时它还是空的
@@ -226,9 +268,18 @@ def analyze(spec: GraphSpec) -> VariableReport:
# 谁都不认识它。模板取不到值会渲染成空字符串,不报错——
# 所以这条必须在跑之前说出来
# 严重度看"我有多确信"。这条会拦住整次运行(runs 启动前要过
# 校验),所以只在真的能穷举时才报 error:
#
# - vars.X / input.X / nodes.X —— 这三个命名空间的成员全是
# 我从图里扫出来的,没扫到就是真没有,多半是拼错
# - 根本不认识的根 —— 可能只是我没建模的东西,降成 warning。
# 假错误比漏报贵得多:漏报只是少提醒一次,假错误让整张图跑不了
known_root = root in _BUILTIN_ROOTS
hint = _did_you_mean(path, index)
report.issues.append(VarIssue(
level="error", node_id=node.id, path=path,
level="error" if known_root else "warning",
node_id=node.id, path=path,
message=(
f"{{{{ {path} }}}} 没有任何节点产出,"
f"是不是想写 {{{{ {hint} }}}}?取不到值会渲染成空字符串,不报错"
+74
View File
@@ -51,6 +51,60 @@ def test_input_fields_are_also_available_as_vars() -> None:
assert [i for i in analyze(g).issues if i.level != "info"] == []
def test_bare_namespace_roots_are_valid() -> None:
"""{{ input }} 取整个输入 dict,{{ vars }} 取整个变量池——都是合法写法。
上一版把这三个根跳过了没注册,于是 {{ input }} 被报成"没有任何节点产出"。
这不是一条无害的多余提示:runs 启动前要过校验,一条假错误意味着整张图
什么都跑不了。实际把问数据页整个卡住过。
"""
g = _graph([
{"id": "start", "type": "input", "config": {"fields": [{"name": "q"}]}},
{"id": "a", "type": "llm", "config": {
"prompt": "{{ input }} {{ vars }} {{ nodes }} {{ messages }} {{ last_message }}",
}},
], [("start", "a")])
bad = [i for i in analyze(g).issues if i.level in ("error", "warning")]
assert bad == [], [i.message for i in bad]
def test_unknown_root_is_only_a_warning() -> None:
"""认不出的根可能只是我没建模的东西,不该拦住运行。
能穷举的(vars/input/nodes 的成员)才报 error——那几个命名空间的成员
全是从图里扫出来的,没扫到就是真没有。
"""
g = _graph([
{"id": "a", "type": "llm", "config": {"prompt": "{{ 某个我不认识的东西.x }}"}},
])
issues = [i for i in analyze(g).issues if i.level != "info"]
assert len(issues) == 1 and issues[0].level == "warning", [i.model_dump() for i in issues]
def test_input_is_open_when_no_fields_are_declared() -> None:
"""入口不声明字段时,input.* 是开放的——运行时传什么键都行。
问数据页就是这样:问题由系统注入成 input.question,而入口节点里一个
字段都没有(声明了反而会要求用户再填一次表单)。把它报成"未定义"会让
整条对话路径跑不起来——实际发生过。
"""
g = _graph([
{"id": "start", "type": "input", "config": {}},
{"id": "a", "type": "llm", "config": {"prompt": "{{ input.question }}"}},
], [("start", "a")])
assert [i for i in analyze(g).issues if i.level != "info"] == []
def test_input_is_closed_once_fields_are_declared() -> None:
"""声明了字段就是闭集,那时候拼错才叫拼错。"""
g = _graph([
{"id": "start", "type": "input", "config": {"fields": [{"name": "question"}]}},
{"id": "a", "type": "llm", "config": {"prompt": "{{ input.quesiton }}"}},
], [("start", "a")])
errs = [i for i in analyze(g).issues if i.level == "error"]
assert len(errs) == 1 and "input.question" in errs[0].message
def test_typo_is_reported_with_a_suggestion() -> None:
"""拼错是这里最常见的错误。只说"不存在"没用,能猜就直说。"""
g = _graph([
@@ -105,6 +159,26 @@ def test_unused_variable_is_only_info() -> None:
assert "vars.unused" in issues[0].message
def test_drilling_into_a_value_is_fine_but_the_variable_must_exist() -> None:
"""往值内部钻不校验(结构是运行期才有的),但第二段必须存在。
分寸就在这里:退得太松,{{ input.quesiton }} 会退到 input 被当成合法,
拼写检查整个失效;退得太紧,{{ vars.rows[0].name }} 会被误报。
"""
g = _graph([
{"id": "a", "type": "llm", "config": {
"assign_to": "rows",
"prompt": "{{ vars.rows[0].name }} {{ nodes.a.text }}", # 都该通过
}},
{"id": "b", "type": "llm", "config": {
"prompt": "{{ vars.rowz[0].name }}", # 第二段拼错了
}},
], [("a", "b")])
errs = [i for i in analyze(g).issues if i.level == "error"]
assert len(errs) == 1, [i.message for i in errs]
assert "vars.rowz" in errs[0].message and errs[0].node_id == "b"
def test_builtins_and_filters_and_indexes_resolve() -> None:
g = _graph([
{"id": "a", "type": "llm", "config": {
+8 -1
View File
@@ -229,7 +229,14 @@ export const api = {
* 每个操作(add_node / add_edge / …)到达即回调,画布边收边长。
*/
export function streamCopilot(
body: { instruction: string; base_graph?: GraphSpec | null; provider?: string | null; model?: string | null },
body: {
instruction: string
base_graph?: GraphSpec | null
provider?: string | null
model?: string | null
/** answer = 用户在问问题(问数据页),build = 用户在描述流程(画布)*/
intent?: 'build' | 'answer'
},
onOp: (op: any) => void,
onEnd: (error?: string) => void,
): () => void {
+34 -4
View File
@@ -108,7 +108,7 @@ export const useChat = create<ChatState>((set, get) => ({
}))
const cancelCopilot = streamCopilot(
{ instruction: question, base_graph: null },
{ instruction: question, base_graph: null, intent: 'answer' },
(op) => {
// 先原样收下,翻译交给 decodeCopilot。thinking 合并在写入时做:
// 一次生成几百上千条 delta,逐条存下来光数组就比图大一个量级
@@ -150,7 +150,7 @@ export const useChat = create<ChatState>((set, get) => ({
graph, phase: 'running', status: PHASE_TEXT.running,
explanation: op.explanation ?? '',
}))
void launch(id, graph, patch, set)
void launch(id, graph, question, patch, set)
break
}
case 'error':
@@ -204,10 +204,40 @@ export const useChat = create<ChatState>((set, get) => ({
},
}))
/**
* 用户的问题填进这张图的入口字段。
*
* 在此之前这里是硬编码的 `input: {}`——问题根本没进图。Copilot 只要给入口
* 声明了一个必填字段(它有时候会,比如 `topic`),这次运行就在第一个节点
* 上立刻挂掉:"缺少必填输入:topic"。用户看到的是自己刚问完就报错,而且
* 错误里提的那个字段名他从来没见过——那是生成的图里的东西。
*
* 之前"能跑"纯属运气:Copilot 那一次恰好没声明字段。
*/
function fillInput(graph: GraphSpec, question: string): Record<string, any> {
const entry = (graph.nodes ?? []).find((n: any) => n.type === 'input')
const fields: any[] = (entry?.data?.config?.fields ?? []) as any[]
if (!fields.length) return { question } // 没声明字段也带上,下游可能用 input.question
const payload: Record<string, any> = {}
for (const f of fields) {
if (f?.name) payload[f.name] = f.default ?? ''
}
// 问题往哪个字段放:名字像"问题"的优先,其次第一个必填的,再不然第一个。
// 只有一个字段时不用挑——那必然就是它
const named = fields.find((f) =>
/question|query|topic|input|问题|需求|主题/i.test(String(f?.name ?? '')))
const target = named ?? fields.find((f) => f?.required) ?? fields[0]
if (target?.name) payload[target.name] = question
return payload
}
/** 图建好了就直接跑——用户要的是答案,不是一张图。 */
async function launch(turnId: string, graph: GraphSpec, patch: Patch, set: any) {
async function launch(
turnId: string, graph: GraphSpec, question: string, patch: Patch, set: any,
) {
try {
const run = await api.runs.start({ graph, input: {} })
const run = await api.runs.start({ graph, input: fillInput(graph, question) })
patch(() => ({ run }))
watch(run.id, turnId, patch, set)
} catch (e: any) {