From 1b595f699cc20150564c2627b7452faae8237c90 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 14 Sep 2026 10:54:31 +0800 Subject: [PATCH] =?UTF-8?q?Copilot=20=E6=B5=81=E5=BC=8F=E7=94=9F=E6=88=90?= =?UTF-8?q?=EF=BC=9A=E5=9B=BE=E5=9C=A8=E7=94=BB=E5=B8=83=E4=B8=8A=E5=AE=9E?= =?UTF-8?q?=E6=97=B6=E9=95=BF=E5=87=BA=E6=9D=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 此前点击生成后是 20-30 秒黑盒等待，模态框关着，用户看不到任何过程。 改成操作流协议：模型按 NDJSON 逐行输出 add_node / add_edge / update_node / remove_node / done，后端边流边经 SSE 转发，前端每收 一个操作就应用到画布——节点带紫色描边逐个出现（临时摆位跟着入边源）， 左下角浮条显示当前动作并可随时取消；done 后后端统一排版+校验， 以 final 事件整体落位，高亮几秒后淡出。修改现有图的场景只输出改动 操作，用户能看到 Copilot 具体动了哪里。 行解析对模型犯规免疫：markdown 围栏、夹杂的解释文字、done 之后的 多余输出都被跳过或截断（单测覆盖跨块行缓冲）。 顺手修结构化输出必失败的 bug：langchain 靠 title 字段识别裸 JSON Schema dict，四处 with_structured_output 的 schema 都缺它，抛 ValueError: Unsupported function 后靠 fallback 文本解析兜底—— 每次都白调一次模型。补上 title（llm 节点 output_schema、branch 语义分类、supervisor 路由、旧版 copilot generate）。 实测：首节点 7.2s 上画布（原先干等 27s），浏览器采样节点数 0→2→3→4→5→6 逐步生长，6 个新节点全部高亮，零页面错误。 --- backend/app/api/copilot.py | 199 ++++++++++++++++++++++++++++ backend/app/engine/nodes/control.py | 1 + backend/app/engine/nodes/llm.py | 3 + backend/app/engine/nodes/multi.py | 1 + frontend/src/api/client.ts | 50 +++++++ frontend/src/canvas/NodeCard.tsx | 2 + frontend/src/index.css | 12 ++ frontend/src/pages/StudioPage.tsx | 104 +++++++++------ frontend/src/store/studio.ts | 139 ++++++++++++++++++- 9 files changed, 466 insertions(+), 45 deletions(-) diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index aa36cdc2..2573d7b6 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -102,6 +102,7 @@ class GenerateOut(BaseModel): GRAPH_SCHEMA: dict[str, Any] = { + "title": "workflow_graph", # langchain 靠 title 识别 JSON Schema dict "type": "object", "properties": { "explanation": {"type": "string", "description": "一两句话说明这张图怎么跑"}, @@ -223,6 +224,204 @@ async def generate( ) +# -------------------------------------------------------------------------- +# 流式生成：模型输出逐行操作流（NDJSON），画布看着图长出来 +# -------------------------------------------------------------------------- + +_STREAM_PROTOCOL = """\ +输出格式（严格遵守）： +- 每行一个完整的 JSON 对象，除此之外不输出任何东西——不要 markdown 围栏、不要解释文字、不要空行注释 +- 可用操作： + {"op":"plan","summary":"一句话说明打算怎么搭"} ← 第一行 + {"op":"add_node","node":{"id":"...","type":"...","label":"中文标签","config":{...}}} + {"op":"update_node","id":"...","label":"...","config":{...}} ← 修改现有节点（config 整体替换） + {"op":"remove_node","id":"..."} + {"op":"add_edge","edge":{"source":"...","target":"...","sourceHandle":"..."}} + {"op":"remove_edge","source":"...","target":"...","sourceHandle":"..."} + {"op":"done","explanation":"两三句话说明这张图怎么跑"} ← 最后一行 +- 按执行顺序添加节点（先入口后出口）；每加一个节点，立刻把连向它的边（两端都已存在的）输出出来，让图连贯地生长 +- 修改现有图时只输出改动，没提到的节点不要动""" + + +def _parse_op_line(line: str) -> dict[str, Any] | None: + """解析一行操作。围栏行、空行、解析失败、缺 op 的一律返回 None—— + 模型偶尔犯规不该毁掉整条流，跳过就是。""" + text = line.strip() + if not text or text.startswith("```"): + return None + try: + obj = json.loads(text) + except json.JSONDecodeError: + return None + if not isinstance(obj, dict) or "op" not in obj: + return None + return obj + + +def _apply_op(nodes: dict[str, dict[str, Any]], edges: list[dict[str, Any]], op: dict[str, Any]) -> bool: + """把一个操作应用到服务端维护的图状态。返回是否真的改了图。""" + kind = op.get("op") + if kind == "add_node": + node = op.get("node") or {} + if not node.get("id") or not node.get("type"): + return False + nodes[node["id"]] = { + "id": node["id"], + "type": node["type"], + "position": {"x": 0, "y": 0}, + "data": {"label": node.get("label") or "", "config": node.get("config") or {}}, + } + return True + if kind == "update_node": + node = nodes.get(op.get("id") or "") + if not node: + return False + if op.get("label") is not None: + node["data"]["label"] = op["label"] + if op.get("config") is not None: + node["data"]["config"] = op["config"] + return True + if kind == "remove_node": + node_id = op.get("id") + if node_id not in nodes: + return False + nodes.pop(node_id) + edges[:] = [e for e in edges if e["source"] != node_id and e["target"] != node_id] + return True + if kind == "add_edge": + edge = op.get("edge") or {} + if not edge.get("source") or not edge.get("target"): + return False + edges.append( + { + "source": edge["source"], + "target": edge["target"], + "sourceHandle": edge.get("sourceHandle"), + } + ) + return True + if kind == "remove_edge": + before = len(edges) + edges[:] = [ + e + for e in edges + if not ( + e["source"] == op.get("source") + and e["target"] == op.get("target") + and (op.get("sourceHandle") is None or e.get("sourceHandle") == op.get("sourceHandle")) + ) + ] + return len(edges) != before + return False + + +async def _iter_ops(model: Any, messages: list[Any]): + """流式读模型输出，按行切出操作。done 之后即停——后面就算有闲话也不要了。""" + from app.engine.state import message_text as _text + + buffer = "" + async for chunk in model.astream(messages): + buffer += _text(chunk) + while "\n" in buffer: + line, buffer = buffer.split("\n", 1) + op = _parse_op_line(line) + if op: + yield op + if op.get("op") == "done": + return + op = _parse_op_line(buffer) + if op: + yield op + + +@router.post("/generate-stream") +async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(get_session)): + """流式版生成：SSE 逐操作推送，前端边收边把节点摆上画布。 + + 结束时（done 或流断）发一条 final：服务端把累积的图排版、校验后整体给出—— + 过程可见性来自操作流，最终质量仍由和手工编排相同的 layout + validate 保证。 + """ + from fastapi.responses import StreamingResponse + + tool_list = "\n".join( + f"- {name}（{spec.category}）：{spec.description}" + for name, spec in sorted(all_specs().items()) + ) + system = ( + "你是一个 agent 工作流编排专家。根据用户需求，以操作流的方式逐步搭出一张可执行的工作流图。\n\n" + f"{NODE_REFERENCE}\n\n可用工具：\n{tool_list}\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]] = {} + edges: list[dict[str, Any]] = [] + if payload.base_graph and payload.base_graph.get("nodes"): + for n in payload.base_graph["nodes"]: + nodes[n["id"]] = n + edges = [ + {k: v for k, v in e.items() if k in ("source", "target", "sourceHandle")} + for e in payload.base_graph.get("edges") or [] + ] + user = ( + "这是当前的工作流：\n" + f"{json.dumps(_slim(payload.base_graph), ensure_ascii=False, indent=2)}\n\n" + f"请按下面的要求修改它（只输出改动操作）：\n{payload.instruction}" + ) + else: + user = f"请设计一个工作流：{payload.instruction}" + + try: + model, _ = await get_chat_model( + session, ModelSpec(provider=payload.provider, model=payload.model, max_tokens=8192) + ) + except ProviderNotConfigured as e: + raise HTTPException(400, str(e)) from e + + messages = [("system", system), ("human", user)] + + async def event_stream(): + explanation = "" + try: + async for op in _iter_ops(model, messages): + if op.get("op") == "done": + explanation = str(op.get("explanation", "")) + changed = _apply_op(nodes, edges, op) + if changed or op.get("op") in ("plan", "done"): + yield f"data: {json.dumps(op, ensure_ascii=False)}\n\n" + except Exception as e: # noqa: BLE001 + yield f"data: {json.dumps({'op': 'error', 'message': f'{type(e).__name__}: {e}'}, ensure_ascii=False)}\n\n" + return + + # 收尾：排版 + 校验，把最终图整体交付 + try: + spec = auto_layout( + GraphSpec.model_validate({"nodes": list(nodes.values()), "edges": edges}) + ) + issues = [i.model_dump() for i in validate_graph(spec).issues] + final = { + "op": "final", + "graph": spec.model_dump(mode="json"), + "issues": issues, + "explanation": explanation, + } + except Exception as e: # noqa: BLE001 + final = {"op": "error", "message": f"生成的图结构非法：{e}"} + yield f"data: {json.dumps(final, ensure_ascii=False)}\n\n" + + return StreamingResponse( + event_stream(), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, + ) + + class FromRunIn(BaseModel): run_id: str name: str = "" diff --git a/backend/app/engine/nodes/control.py b/backend/app/engine/nodes/control.py index 7bf2cc0a..338b26dc 100644 --- a/backend/app/engine/nodes/control.py +++ b/backend/app/engine/nodes/control.py @@ -81,6 +81,7 @@ async def _classify_with_llm( ) schema = { + "title": "classify", # langchain 靠 title 识别 JSON Schema dict "type": "object", "properties": { "key": {"type": "string", "enum": keys}, diff --git a/backend/app/engine/nodes/llm.py b/backend/app/engine/nodes/llm.py index acf73e44..9a212818 100644 --- a/backend/app/engine/nodes/llm.py +++ b/backend/app/engine/nodes/llm.py @@ -176,6 +176,9 @@ async def run_llm(state: GraphState, ctx: NodeContext) -> dict[str, Any]: # 要结构化结果就走原生 structured output，这条路径拿不到 token 流 ctx.emit(EventType.LLM_START, structured=True, message_count=len(messages)) try: + # langchain 靠 "title" 识别裸 JSON Schema dict，缺了会抛 Unsupported function + if isinstance(schema, dict) and "title" not in schema: + schema = {"title": "output", **schema} structured = model.with_structured_output(schema) value = await structured.ainvoke(messages) except Exception as e: # noqa: BLE001 diff --git a/backend/app/engine/nodes/multi.py b/backend/app/engine/nodes/multi.py index 56c999db..18390e35 100644 --- a/backend/app/engine/nodes/multi.py +++ b/backend/app/engine/nodes/multi.py @@ -59,6 +59,7 @@ async def run_supervisor(state: GraphState, ctx: NodeContext) -> dict[str, Any]: for n in names ) route_schema = { + "title": "route", # langchain 靠 title 识别 JSON Schema dict "type": "object", "properties": { "next": {"type": "string", "enum": names + ["FINISH"]}, diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index 17bd9e78..e50d0b62 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -202,6 +202,56 @@ export const api = { }, } +/** + * 流式生成工作流：SSE 逐操作回调，返回取消函数。 + * 每个操作（add_node / add_edge / …）到达即回调，画布边收边长。 + */ +export function streamCopilot( + body: { instruction: string; base_graph?: GraphSpec | null }, + onOp: (op: any) => void, + onEnd: (error?: string) => void, +): () => void { + const controller = new AbortController() + void (async () => { + try { + const res = await fetch(`${BASE}/copilot/generate-stream`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', ...actorHeader() }, + body: JSON.stringify(body), + signal: controller.signal, + }) + if (!res.ok || !res.body) { + let detail = `${res.status} ${res.statusText}` + try { detail = (await res.json()).detail ?? detail } catch { /* keep */ } + onEnd(String(detail)) + return + } + const reader = res.body.getReader() + const decoder = new TextDecoder() + let buffer = '' + for (;;) { + const { done, value } = await reader.read() + if (done) break + buffer += decoder.decode(value, { stream: true }) + let idx: number + while ((idx = buffer.indexOf('\n\n')) >= 0) { + const frame = buffer.slice(0, idx) + buffer = buffer.slice(idx + 2) + if (!frame.startsWith('data: ')) continue + try { + onOp(JSON.parse(frame.slice(6))) + } catch { /* 半截帧，忽略 */ } + } + } + onEnd() + } catch (e: any) { + if (e?.name !== 'AbortError') onEnd(e?.message ?? '连接中断') + else onEnd() + } + })() + return () => controller.abort() +} + /** * 订阅一次运行的事件流。 * diff --git a/frontend/src/canvas/NodeCard.tsx b/frontend/src/canvas/NodeCard.tsx index b3d5ba96..e236efb2 100644 --- a/frontend/src/canvas/NodeCard.tsx +++ b/frontend/src/canvas/NodeCard.tsx @@ -58,6 +58,7 @@ function NodeCardImpl({ id, data, selected }: NodeProps) { const runtime = useStudio((s) => s.runtime[id]) // selector 必须返回稳定引用：filter() 每次都造新数组，会触发无限重渲染 const hasError = useStudio((s) => s.issues.some((i) => i.node_id === id && i.level === 'error')) + const copilotNew = useStudio((s) => s.copilotNew.includes(id)) const Icon = def?.icon ?? Sparkles const handles = sourceHandles(data.nodeType, data.config) const status = runtime?.status ?? 'idle' @@ -73,6 +74,7 @@ function NodeCardImpl({ id, data, selected }: NodeProps) { status === 'done' && 'node-done', status === 'failed' && 'node-failed', status === 'waiting' && 'node-waiting', + copilotNew && 'node-copilot-new', selected && 'ring-2 ring-[var(--accent)] ring-offset-1 ring-offset-[var(--bg)]', )} style={{ width: 238, borderColor: selected ? 'var(--accent)' : undefined }} diff --git a/frontend/src/index.css b/frontend/src/index.css index 350deb73..37581db1 100644 --- a/frontend/src/index.css +++ b/frontend/src/index.css @@ -201,6 +201,18 @@ pre, border-color: var(--warn) !important; } +/* Copilot 刚放上来的节点：紫色描边 + 出现动画，final 后几秒淡出 */ +@keyframes copilot-pop { + from { opacity: 0; transform: scale(0.85); } + to { opacity: 1; transform: scale(1); } +} +.node-copilot-new { + animation: copilot-pop 0.25s ease-out; + border-color: #bc8cff !important; + box-shadow: 0 0 0 3px color-mix(in srgb, #bc8cff 25%, transparent); + transition: box-shadow 0.8s, border-color 0.8s; +} + @keyframes fade-up { from { opacity: 0; transform: translateY(4px); } to { opacity: 1; transform: none; } diff --git a/frontend/src/pages/StudioPage.tsx b/frontend/src/pages/StudioPage.tsx index 0a1895c0..f5bde685 100644 --- a/frontend/src/pages/StudioPage.tsx +++ b/frontend/src/pages/StudioPage.tsx @@ -142,8 +142,9 @@ export function StudioPage() { -
+
+