diff --git a/CHANGELOG.md b/CHANGELOG.md index f5e8a4c..e3332f2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,8 @@ ## Unreleased +- DeepSeek Flash 的思考过程恢复实时显示，不再等到正文开始后才集中出现。 + - 国内模型成本统计跟随最新公开价格更新，DeepSeek 会按实际峰谷时段和缓存命中量计算，历史调价后的记录也可审计矫正。 - GLM-5.3 Flash 替换旧版 GLM 并向默认档位开放；上传图片会直接交给主模型理解，长工具任务可延续既有分析状态。 diff --git a/DESIGN.md b/DESIGN.md index 9b24cc8..74bed9a 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -208,7 +208,7 @@ Admin GET /v1/admin/*(require_admin;overview + usage/models|users + storage/ Export GET /v1/tasks/{id}/export(docx) ``` -**SSE 事件**:`run_start / llm_start / text{delta} / reasoning{delta}(thinking 模型推理流,前端灰色折叠卡)/ progress_snapshot{run_id,steps,waiting}(当前 user message 即 run 边界,从 messages 投影恢复)/ tool_call / tool_result(预览,完整走 DB)/ llm_end / model_switch / warn{msg}(熔断·重复拦截·折叠失败等运行时提醒)/ context_fold{phase,...}(§8.8 Phase 2 折叠 start/done)/ cancelled / error / done`。`task_progress` 每次提交完整步骤快照,前端整体替换；旧 `set_plan/update_step` 仅在历史投影时兼容。进度 dock 是运行状态而非历史消息:活跃时紧凑显示,正常完成后隐藏,`ask_user` 等待确认/取消/异常时折叠保留。fan-out:每订阅独立 queue;迟到订阅先从 PG 恢复当前 run 最新进度,终态迟到订阅立收 done。普通直播事件不持久化(messages 走 PG)。 +**SSE 事件**:`run_start / llm_start / text{delta} / reasoning{delta}(thinking 模型推理流,前端灰色折叠卡)/ reasoning_reset(传输层放弃当前响应并重试时清除已直播推理)/ progress_snapshot{run_id,steps,waiting}(当前 user message 即 run 边界,从 messages 投影恢复)/ tool_call / tool_result(预览,完整走 DB)/ llm_end / model_switch / warn{msg}(熔断·重复拦截·折叠失败等运行时提醒)/ context_fold{phase,...}(§8.8 Phase 2 折叠 start/done)/ cancelled / error / done`。`task_progress` 每次提交完整步骤快照,前端整体替换；旧 `set_plan/update_step` 仅在历史投影时兼容。进度 dock 是运行状态而非历史消息:活跃时紧凑显示,正常完成后隐藏,`ask_user` 等待确认/取消/异常时折叠保留。fan-out:每订阅独立 queue;迟到订阅先从 PG 恢复当前 run 最新进度,终态迟到订阅立收 done。普通直播事件不持久化(messages 走 PG)。 **版本化**:`/v1` minor 半年兼容,major 6 个月 deprecation。**CORS**:本地 `*`,部署收紧。 ### 7.3 认证 diff --git a/core/loop.py b/core/loop.py index 7ec3656..88ef9cc 100644 --- a/core/loop.py +++ b/core/loop.py @@ -639,9 +639,8 @@ class AgentLoop: cancel_check=self._is_cancelled, ) cancelled = False - pending_events: List[dict] = [] may_reroute = self.caps.family == "deepseek_v4" - output_route_known = not may_reroute + reasoning_emitted = False try: for chunk in stream: if self._is_cancelled(): @@ -653,7 +652,10 @@ class AgentLoop: may_reroute and any(name in self._DEEPSEEK_NONSTREAM_TOOLS for name in tool_names) ): - # 推理 delta 先暂存，避免切换后非流式完整 reasoning 再发一次造成重复。 + # 当前流会被放弃并以非流式完整重发。已实时展示的推理属于被放弃 + # 的响应，先通知前端清掉，避免随后展示最终响应时重复或与历史不一致。 + if reasoning_emitted: + self._emit({"type": "reasoning_reset"}) raise PreferNonstreamToolCall(next( name for name in tool_names if name in self._DEEPSEEK_NONSTREAM_TOOLS @@ -663,20 +665,13 @@ class AgentLoop: # _execute_tool_call 时机发更直观)。 delta_text = extract_delta_content(chunk) if delta_text: - pending_events.append({"type": "text", "delta": delta_text}) + self._emit({"type": "text", "delta": delta_text}) # thinking 模型的推理 delta 也实时流出(reasoning 事件):深度推理可达 # 分钟级,不发的话前端全程静止"思考中",用户以为卡死。 delta_reasoning = extract_delta_reasoning(chunk) if delta_reasoning: - pending_events.append({"type": "reasoning", "delta": delta_reasoning}) - # 首个正文或任意工具名已确定本轮不会再切换；释放之前暂存的推理片段， - # 后续 chunk 也继续即时释放。只有尚未看到输出类型时才短暂缓冲。 - if delta_text or tool_names: - output_route_known = True - if output_route_known and pending_events: - for event in pending_events: - self._emit(event) - pending_events.clear() + self._emit({"type": "reasoning", "delta": delta_reasoning}) + reasoning_emitted = True # interruptible stream 会在无新 chunk 的等待期直接因 cancel 结束迭代； # 循环体没有机会执行上面的检查，故在正常耗尽处再判一次。 if self._is_cancelled(): @@ -690,10 +685,6 @@ class AgentLoop: if cancelled: return None, True - # 极少数 provider 只有 reasoning/空收尾、始终没有正文或工具名，不能吞掉已收内容。 - for event in pending_events: - self._emit(event) - # 用 litellm 官方 helper 拼回完整 response(包括 tool_calls 拼接 + usage)。 # messages 参数仅用于失败时回填 prompt token 估算,正常路径 stream_options.include_usage # 已让最后一个 chunk 带准确 usage。 diff --git a/tests/test_loop_malformed_retry.py b/tests/test_loop_malformed_retry.py index 960ac6f..734a69f 100644 --- a/tests/test_loop_malformed_retry.py +++ b/tests/test_loop_malformed_retry.py @@ -35,6 +35,7 @@ def _make_loop(stream_results, nonstream_results): loop.caps = SimpleNamespace(reliable_context=64_000, family="test", variant="t") loop.session = SimpleNamespace(messages=[], task_id="test-task") loop.user_id = "test-user" # 无 DB 环境:_log_malformed_args 的落库路径应静默跳过 + loop.user_root = None # salvage(0.58.24)接在畸形重试前:空 schemas → 未知工具无白名单 → _try_salvage 返 False, # 这些用例本就测「salvage 救不了 → 落非流式重试」,给个空 executor 即可走到该分支。 loop.executor = SimpleNamespace(schemas=lambda: []) @@ -56,8 +57,8 @@ def _make_loop(stream_results, nonstream_results): class TestMalformedRetry(unittest.TestCase): - def test_deepseek_write_name_aborts_before_buffered_reasoning_is_emitted(self): - """真实 collect 路径见到 write 名即关流，之前的 reasoning 不重复上屏。""" + def test_deepseek_write_name_resets_streamed_reasoning_before_reroute(self): + """真实 collect 路径实时发 reasoning；见到 write 后重置再改走非流式。""" reasoning = SimpleNamespace( choices=[SimpleNamespace(delta=SimpleNamespace( reasoning_content="thinking", content=None, tool_calls=None, @@ -86,7 +87,44 @@ class TestMalformedRetry(unittest.TestCase): with self.assertRaises(PreferNonstreamToolCall): loop._collect_stream_once([]) - self.assertEqual(loop.events, []) + self.assertEqual(loop.events, [ + {"type": "reasoning", "delta": "thinking"}, + {"type": "reasoning_reset"}, + ]) + + def test_deepseek_reasoning_is_emitted_before_output_route_is_known(self): + """普通 DeepSeek 回答不能等首个正文 token 才释放已经到达的推理。""" + reasoning = SimpleNamespace( + choices=[SimpleNamespace(delta=SimpleNamespace( + reasoning_content="thinking", content=None, tool_calls=None, + ))] + ) + text = SimpleNamespace( + choices=[SimpleNamespace(delta=SimpleNamespace( + reasoning_content=None, content="answer", tool_calls=None, + ))] + ) + loop = object.__new__(AgentLoop) + loop.caps = SimpleNamespace( + family="deepseek_v4", default_reasoning_effort=None, + ) + loop.executor = SimpleNamespace(schemas=lambda: []) + loop.cancel_check = None + loop.events = [] + loop._emit = loop.events.append + loop.llm = SimpleNamespace( + chat_stream=lambda **_kwargs: iter([reasoning, text]), + ) + + with unittest.mock.patch( + "core.loop.litellm.stream_chunk_builder", return_value=object() + ): + loop._collect_stream_once([]) + + self.assertEqual(loop.events, [ + {"type": "reasoning", "delta": "thinking"}, + {"type": "text", "delta": "answer"}, + ]) def test_preferred_tool_reroutes_before_arguments_stream(self): """首包识别 write/edit 后直接非流式，不把它算作一次畸形失败。""" diff --git a/tests/test_static_vendor.py b/tests/test_static_vendor.py index afe7899..8c16fd0 100644 --- a/tests/test_static_vendor.py +++ b/tests/test_static_vendor.py @@ -261,6 +261,13 @@ class StaticVendorTests(unittest.TestCase): self.assertIn("await yieldToStreamPaint()", chat_js) self.assertIn('document.visibilityState === "hidden"', chat_js) + def test_sse_consumer_resets_abandoned_reasoning_stream(self) -> None: + chat_js = (JS_DIR / "chat.js").read_text(encoding="utf-8") + + self.assertIn("function resetReasonSeg(run)", chat_js) + self.assertIn('t === "reasoning_reset"', chat_js) + self.assertIn("resetReasonSeg(ctx)", chat_js) + def test_custom_task_name_is_optional_with_explicit_directory(self) -> None: html = DEV_HTML.read_text(encoding="utf-8") newtask_js = (JS_DIR / "newtask.js").read_text(encoding="utf-8") diff --git a/web/routers/messages.py b/web/routers/messages.py index 9febbb0..b5631d0 100644 --- a/web/routers/messages.py +++ b/web/routers/messages.py @@ -636,7 +636,7 @@ def register_message_routes(app, *, require_user) -> None: user_id: UUID = Depends(require_user), ): """SSE 流。订阅当前 task 的活动 event(单活 run 形态下无歧义)。 - 事件类型:run_start / llm_start / text / reasoning / tool_call / + 事件类型:run_start / llm_start / text / reasoning / reasoning_reset / tool_call / tool_result / llm_end / cancelled / error / done。data 是 JSON dict (已剔除 `type` 字段,移到 event 名)。 """ diff --git a/web/static/js/chat.js b/web/static/js/chat.js index 990b328..e43702d 100644 --- a/web/static/js/chat.js +++ b/web/static/js/chat.js @@ -1403,6 +1403,15 @@ function closeReasonSeg(run) { run.curReason = null; } +// provider 传输层放弃当前流并重试时，移除这次响应已经直播出来的推理。 +// 重试返回的最终 reasoning 会重新建立一段，确保直播态与最终持久化消息一致。 +function resetReasonSeg(run) { + const rs = run.curReason; + if (!rs) return; + rs.el.remove(); + run.curReason = null; +} + // ───── run 活跃状态指示(阶段 + 已耗时) ───── // 长耗时阶段(TTFT / 深度推理 / 工具执行)原本页面全静止,和卡死无法区分。 // 每秒把「阶段 · Ns」写进空占位段的 data-status(CSS :empty::before 渲染), @@ -3433,6 +3442,8 @@ function handleSseEvent(ev, asstCard, ctx) { if (nearBottom) stream.scrollTop = stream.scrollHeight; }); } + } else if (t === "reasoning_reset") { + resetReasonSeg(ctx); } else if (t === "text" && ev.data && ev.data.delta) { closeReasonSeg(ctx); // 正文开始 → 思考过程定稿折叠 ctx.acc += ev.data.delta;