diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index f5fc5935..5b8803d2 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -1,6 +1,8 @@ from __future__ import annotations +import asyncio import json +import time from typing import Any from fastapi import APIRouter, Depends, HTTPException @@ -378,11 +380,25 @@ def _apply_op(nodes: dict[str, dict[str, Any]], edges: list[dict[str, Any]], op: async def _iter_ops(model: Any, messages: list[Any]): - """流式读模型输出，按行切出操作。done 之后即停——后面就算有闲话也不要了。""" + """流式读模型输出，按行切出操作。done 之后即停——后面就算有闲话也不要了。 + + 除了操作，还把模型的思考原样转出去。原先这里只取正文、丢掉其余一切， + 于是从请求发出到第一个操作到达之间（实测 5~30 秒）前端一个字都收不到， + 只能干挂着"正在起草…"——用户没法判断是在想、还是已经卡死了。 + + 思考走 thinking_text()（Claude 4.6+ 的 thinking 块），不是"把非 JSON 行 + 当成思考"：后者会把模型的 markdown 注释、协议跑偏时的乱输出一并当作思考 + 展示出来，反而误导。不支持 thinking 的模型就只有心跳，不硬凑。 + """ from app.engine.state import message_text as _text + from app.engine.state import thinking_text buffer = "" async for chunk in model.astream(messages): + reasoning = thinking_text(chunk) + if reasoning: + yield {"op": "thinking", "delta": reasoning} + buffer += _text(chunk) while "\n" in buffer: line, buffer = buffer.split("\n", 1) @@ -396,6 +412,45 @@ async def _iter_ops(model: Any, messages: list[Any]): yield op +# 多久没动静就补一拍心跳。3 秒是"用户开始怀疑是不是卡了"的量级。 +_HEARTBEAT_SECONDS = 3.0 + + +async def _with_heartbeat(source: Any): + """模型沉默时按拍补心跳，让前端能显示阶段和已用时长。 + + 必须在这一层做而不是在 _iter_ops 里判断时间差：模型不吐 chunk 时那个 + async for 的循环体根本不执行，压根轮不到检查。 + """ + started = time.monotonic() + phase = "planning" + iterator = source.__aiter__() + while True: + pending = asyncio.ensure_future(iterator.__anext__()) + while True: + done, _ = await asyncio.wait({pending}, timeout=_HEARTBEAT_SECONDS) + if pending in done: + try: + op = pending.result() + except StopAsyncIteration: + return + kind = op.get("op") + # 阶段跟着实际操作走，前端据此换文案 + if kind == "add_node": + phase = "building" + elif kind in ("add_edge", "connect"): + phase = "wiring" + elif kind == "done": + phase = "finalizing" + yield op + break + yield { + "op": "heartbeat", + "phase": phase, + "elapsed_ms": int((time.monotonic() - started) * 1000), + } + + @router.post("/generate-stream") async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(get_session)): """流式版生成：SSE 逐操作推送，前端边收边把节点摆上画布。 @@ -451,11 +506,16 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g explanation = "" yield f"data: {json.dumps({'op': 'model', 'model': model_id}, ensure_ascii=False)}\n\n" try: - async for op in _iter_ops(model, messages): - if op.get("op") == "done": + async for op in _with_heartbeat(_iter_ops(model, messages)): + kind = op.get("op") + # 思考和心跳不是图操作，不进 _apply_op，直接转给前端 + if kind in ("thinking", "heartbeat"): + yield f"data: {json.dumps(op, ensure_ascii=False)}\n\n" + continue + if kind == "done": explanation = str(op.get("explanation", "")) changed = _apply_op(nodes, edges, op) - if changed or op.get("op") in ("plan", "done"): + if changed or kind 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" diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 3d5aab87..fc5b21a5 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -31,6 +31,8 @@ dependencies = [ # 再加一次 OCI 镜像拉取（实测约 55s），所以不进默认安装。 # environment.yml 里已经带上了；单独装用 `pip install 'agentlab-backend[microvm]'`。 microvm = ["microsandbox>=0.6"] +# 测试：conda 环境默认不装，跑测试前 pip install 'agentlab-backend[dev]' +dev = ["pytest>=8.0", "pytest-asyncio>=0.24"] [build-system] requires = ["hatchling"] diff --git a/backend/pytest.ini b/backend/pytest.ini new file mode 100644 index 00000000..b30d93f3 --- /dev/null +++ b/backend/pytest.ini @@ -0,0 +1,5 @@ +[pytest] +# asyncio_mode=auto 省掉每个协程测试都写一遍 @pytest.mark.asyncio +asyncio_mode = auto +testpaths = tests +addopts = -q --tb=short diff --git a/backend/tests/test_copilot_stream.py b/backend/tests/test_copilot_stream.py new file mode 100644 index 00000000..79a43bf5 --- /dev/null +++ b/backend/tests/test_copilot_stream.py @@ -0,0 +1,115 @@ +"""Copilot 流式生成：等待期间必须有反馈。 + +这组测试守的是一个体验契约，不是实现细节：从请求发出到第一个节点落到画布上， +本机实测有 31.7 秒（DeepSeek V4 pro，中等复杂度的需求）。这段时间里前端如果 +一个事件都收不到，用户没法判断是在想还是已经卡死——所以 thinking 和 heartbeat +这两路反馈只要断了，就是回归。 +""" +from __future__ import annotations + +import asyncio + +import pytest + +from app.api import copilot as cp +from app.api.copilot import _iter_ops, _with_heartbeat + + +class Chunk: + """模拟流式 chunk。content 是块列表时对应 Claude 4.6+ 的 thinking 混排。""" + + def __init__(self, content): + self.content = content + + +class FakeModel: + def __init__(self, chunks, delay: float = 0.0): + self._chunks = chunks + self._delay = delay + + async def astream(self, messages): + for chunk in self._chunks: + if self._delay: + await asyncio.sleep(self._delay) + yield chunk + + +async def _collect(gen): + return [op async for op in gen] + + +@pytest.mark.asyncio +async def test_thinking_blocks_become_events(): + """支持 thinking 的模型，思考要原样转给前端。""" + model = FakeModel([ + Chunk([{"type": "thinking", "thinking": "用户要一个调研流程。"}]), + Chunk([{"type": "thinking", "thinking": "先搜索，再抓取。"}]), + Chunk([{"type": "text", "text": '{"op":"plan","summary":"三步"}\n'}]), + Chunk([{"type": "text", "text": '{"op":"done","explanation":"完成"}\n'}]), + ]) + ops = await _collect(_iter_ops(model, [])) + kinds = [o["op"] for o in ops] + + assert kinds.count("thinking") == 2 + assert ops[0]["delta"] == "用户要一个调研流程。" + assert kinds[-1] == "done", "done 之后就该停，后面的闲话不要" + + +@pytest.mark.asyncio +async def test_plain_text_is_not_passed_off_as_thinking(): + """不支持 thinking 的模型不该被凭空造出思考。 + + 有一种取巧写法是"把非 JSON 行当成思考"，那会把模型的 markdown 注释、 + 协议跑偏时的乱输出一并当作思考展示——宁可只有心跳，也不要假的思考。 + """ + model = FakeModel([ + Chunk("模型的随口注释，不是 JSON\n"), + Chunk('{"op":"plan","summary":"x"}\n'), + Chunk('{"op":"done","explanation":"y"}\n'), + ]) + kinds = [o["op"] for o in await _collect(_iter_ops(model, []))] + + assert "thinking" not in kinds + assert kinds == ["plan", "done"] + + +@pytest.mark.asyncio +async def test_heartbeat_fires_while_model_is_silent(monkeypatch): + """模型长时间不吐字时要补心跳，否则前端一片死寂。 + + 心跳必须在 _with_heartbeat 这一层做：模型不产出 chunk 时，_iter_ops 里那个 + async for 的循环体根本不执行，在里面判断时间差是轮不到的。 + """ + monkeypatch.setattr(cp, "_HEARTBEAT_SECONDS", 0.2) + + class SlowModel: + async def astream(self, messages): + await asyncio.sleep(0.9) # 模拟"正在想" + yield Chunk('{"op":"plan","summary":"终于"}\n') + yield Chunk('{"op":"done","explanation":"z"}\n') + + ops = await _collect(_with_heartbeat(_iter_ops(SlowModel(), []))) + beats = [o for o in ops if o["op"] == "heartbeat"] + + assert len(beats) >= 2, "0.9 秒的沉默至少该有两拍" + assert beats[0]["phase"] == "planning" + assert beats[0]["elapsed_ms"] > 0 + assert [o["op"] for o in ops][-1] == "done" + + +@pytest.mark.asyncio +async def test_heartbeat_phase_follows_progress(monkeypatch): + """阶段要跟着实际操作走——阶段会变本身就是"它还活着"的信号。""" + monkeypatch.setattr(cp, "_HEARTBEAT_SECONDS", 0.2) + + class Staged: + async def astream(self, messages): + yield Chunk('{"op":"add_node","node":{"id":"a","type":"input"}}\n') + await asyncio.sleep(0.5) # 放完节点后卡一会儿 + yield Chunk('{"op":"done","explanation":"d"}\n') + + ops = await _collect(_with_heartbeat(_iter_ops(Staged(), []))) + beats = [o for o in ops if o["op"] == "heartbeat"] + + assert beats, "add_node 之后的停顿也该有心跳" + assert beats[0]["phase"] == "building", "放过节点了，阶段不该还停在 planning" diff --git a/environment.yml b/environment.yml index abe98f70..b89b8ae8 100644 --- a/environment.yml +++ b/environment.yml @@ -15,4 +15,4 @@ dependencies: # [microvm] 带上独立内核的沙箱后端（约 35MB wheel，首次运行还要下载 # 约 50MB 运行时，外加一次 OCI 镜像拉取）。不需要就把方括号去掉， # 沙箱会退回 Seatbelt —— 代价是内存限额在 macOS 上限不住。 - - -e ./backend[microvm] + - -e ./backend[microvm,dev] diff --git a/frontend/src/pages/StudioPage.tsx b/frontend/src/pages/StudioPage.tsx index 75d9ddfa..0b697bbd 100644 --- a/frontend/src/pages/StudioPage.tsx +++ b/frontend/src/pages/StudioPage.tsx @@ -1,6 +1,7 @@ import { useCallback, useEffect, useRef, useState } from 'react' import { - AlertTriangle, Check, ChevronDown, Copy, LayoutGrid, Plus, Save, ShieldCheck, Sparkles, Trash2, Wand2, + AlertTriangle, Check, ChevronDown, ChevronRight, Copy, LayoutGrid, Plus, Save, ShieldCheck, + Sparkles, Trash2, Wand2, } from 'lucide-react' import clsx from 'clsx' import { api } from '../api/client' @@ -475,36 +476,101 @@ function CopilotModal({ open, onClose }: { open: boolean; onClose: () => void }) } /** Copilot 工作时的画布浮条：显示当前操作，可随时取消；结束后展示说明。 */ +// 阶段文案。从提交到第一个节点落地中间有 5~30 秒，这段时间界面上只有 +// "正在起草…"三个字一动不动，用户分不清是在想还是已经卡死——阶段会变的 +// 本身就是"它还活着"的信号。 +const COPILOT_PHASES: Record = { + connecting: '正在连接模型…', + planning: '正在理解需求、规划结构…', + building: '正在放置节点…', + wiring: '正在连接数据流…', + finalizing: '正在排版和校验…', +} + function CopilotStatusBar() { const copilot = useStudio((s) => s.copilot) const stopCopilot = useStudio((s) => s.stopCopilot) + const retryCopilot = useStudio((s) => s.retryCopilot) const [dismissed, setDismissed] = useState(false) + const [showThinking, setShowThinking] = useState(true) + const thinkingRef = useRef(null) useEffect(() => { // 新一轮生成开始时重新显示 if (copilot.active) setDismissed(false) }, [copilot.active]) + // 思考是流式追加的，跟着滚到底，否则用户只能看到最早那几行 + useEffect(() => { + const el = thinkingRef.current + if (el) el.scrollTop = el.scrollHeight + }, [copilot.thinking]) + if (dismissed) return null if (!copilot.active && !copilot.explanation && !copilot.error) return null + const seconds = copilot.elapsedMs ? Math.round(copilot.elapsedMs / 1000) : 0 + // 有具体操作就显示操作，还没开始动就显示阶段——不再是恒定的"正在起草…" + const headline = copilot.lastOp || COPILOT_PHASES[copilot.phase] || '正在起草…' + return (
{copilot.active ? ( -
- - - Copilot - {copilot.model && {copilot.model}} - {copilot.lastOp} - - +
+
+ + + Copilot + {copilot.model && {copilot.model}} + {headline} + + {seconds > 0 && {seconds}s} + +
+ + {/* 支持 thinking 的模型（Claude 4.6+）才有内容；没有就只显示阶段 */} + {copilot.thinking && ( +
+ + {showThinking && ( +
+ {copilot.thinking} +
+ )} +
+ )}
) : copilot.error ? ( -
- {copilot.error} - +
+
+ {copilot.error} + +
+ {/* 失败多半跟需求本身无关（模型抽风、协议跑偏、网断了）， + 不该逼用户重新打开 Modal 把需求再敲一遍 */} + {copilot.lastInstruction && ( +
+ + {copilot.lastInstruction} + + +
+ )}
) : (
diff --git a/frontend/src/store/studio.ts b/frontend/src/store/studio.ts index 7a042fa8..31bd3e03 100644 --- a/frontend/src/store/studio.ts +++ b/frontend/src/store/studio.ts @@ -78,7 +78,19 @@ interface StudioState { streaming: boolean unsubscribe: (() => void) | null - copilot: { active: boolean; lastOp: string; explanation: string; error: string; model: string } + copilot: { + active: boolean + lastOp: string + explanation: string + error: string + model: string + // 从提交到第一个操作之间有 5~30 秒，得让用户看见这段在发生什么 + phase: 'connecting' | 'planning' | 'building' | 'wiring' | 'finalizing' | '' + thinking: string // 模型的思考原文（只有支持 thinking 的模型有） + elapsedMs: number + lastInstruction: string // 失败后"用同一需求重试"要用 + lastUseBase: boolean + } copilotNew: string[] cancelCopilot: (() => void) | null @@ -100,6 +112,7 @@ interface StudioState { startFormalRun: (input: Record) => Promise runCopilot: (instruction: string, useBase: boolean, model?: string | null) => void stopCopilot: () => void + retryCopilot: () => void attachRun: (runId: string) => Promise stopRun: () => Promise clearRun: () => void @@ -121,7 +134,8 @@ export const useStudio = create((set, get) => ({ activeEdges: [], streaming: false, unsubscribe: null, - copilot: { active: false, lastOp: '', explanation: '', error: '', model: '' }, + copilot: { active: false, lastOp: '', explanation: '', error: '', model: '', + phase: '', thinking: '', elapsedMs: 0, lastInstruction: '', lastUseBase: false }, copilotNew: [], cancelCopilot: null, @@ -297,7 +311,11 @@ export const useStudio = create((set, get) => ({ const state = get() state.cancelCopilot?.() set({ - copilot: { active: true, lastOp: '正在起草…', explanation: '', error: '', model: model ?? '' }, + copilot: { + active: true, lastOp: '', explanation: '', error: '', model: model ?? '', + phase: 'connecting', thinking: '', elapsedMs: 0, + lastInstruction: instruction, lastUseBase: useBase, + }, copilotNew: [], }) if (!useBase) set({ nodes: [], edges: [], dirty: true }) @@ -327,8 +345,28 @@ export const useStudio = create((set, get) => ({ // 后端首帧告知实际用的模型，浮条上直接显示，不用猜 set({ copilot: { ...s.copilot, model: op.model ?? '' } }) break + case 'thinking': + // 模型正在想什么，原样接上去。看着它想比盯着"正在起草…"强得多 + set({ + copilot: { + ...s.copilot, + phase: s.copilot.phase === 'connecting' ? 'planning' : s.copilot.phase, + thinking: (s.copilot.thinking + (op.delta ?? '')).slice(-4000), + }, + }) + break + case 'heartbeat': + // 模型沉默期间的补拍，用来刷新阶段和已用时长 + set({ + copilot: { + ...s.copilot, + phase: (op.phase as any) ?? s.copilot.phase, + elapsedMs: op.elapsed_ms ?? s.copilot.elapsedMs, + }, + }) + break case 'plan': - set({ copilot: { ...s.copilot, lastOp: op.summary ?? '规划中' } }) + set({ copilot: { ...s.copilot, phase: 'planning', lastOp: op.summary ?? '规划中' } }) break case 'add_node': { const n = op.node @@ -393,21 +431,24 @@ export const useStudio = create((set, get) => ({ }) break case 'done': - set({ copilot: { ...s.copilot, lastOp: '排版整理中…', explanation: op.explanation ?? '' } }) + set({ copilot: { ...s.copilot, phase: 'finalizing', lastOp: '排版整理中…', + explanation: op.explanation ?? '' } }) break case 'final': { // 后端排版+校验后的最终图整体落位；高亮集合保留几秒供辨认 const { nodes, edges } = toFlow(op.graph) set({ nodes, edges, dirty: true, - copilot: { active: false, lastOp: '', model: get().copilot.model, - explanation: op.explanation ?? get().copilot.explanation, error: '' } }) + copilot: { ...get().copilot, active: false, lastOp: '', phase: '', + thinking: '', elapsedMs: 0, error: '', + explanation: op.explanation ?? get().copilot.explanation } }) void get().validate() setTimeout(() => set({ copilotNew: [] }), 6000) break } case 'error': - set({ copilot: { active: false, lastOp: '', explanation: '', - model: get().copilot.model, error: op.message ?? '生成失败' } }) + // 保留 lastInstruction：失败浮条上的"重试"要用它，不能让用户重填 + set({ copilot: { ...get().copilot, active: false, lastOp: '', phase: '', + thinking: '', explanation: '', error: op.message ?? '生成失败' } }) break } }, @@ -425,11 +466,19 @@ export const useStudio = create((set, get) => ({ stopCopilot: () => { get().cancelCopilot?.() set({ - copilot: { ...get().copilot, active: false }, + copilot: { ...get().copilot, active: false, phase: '', thinking: '' }, cancelCopilot: null, }) }, + retryCopilot: () => { + // 失败后用同一条需求重来。以前只能重新打开 Modal 把需求再敲一遍， + // 而失败往往跟需求本身无关（模型抽风、协议跑偏、网络断了） + const { lastInstruction, lastUseBase, model } = get().copilot + if (!lastInstruction) return + get().runCopilot(lastInstruction, lastUseBase, model || null) + }, + startFormalRun: async (input) => { const { workflow } = get() if (!workflow) return null