From 842ba547ec0e514e541b5a17ba0ac77c129f8f92 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Thu, 1 Oct 2026 10:26:22 +0800 Subject: [PATCH] fix(agent): keep thinking status separate --- app/agent/callback/__init__.py | 34 +++++++++++++++---------- app/agent/web.py | 4 +++ tests/test_agent_streaming_lifecycle.py | 34 +++++++++++++++++++++++++ 3 files changed, 59 insertions(+), 13 deletions(-) diff --git a/app/agent/callback/__init__.py b/app/agent/callback/__init__.py index 4cf840d97..74f7ee725 100644 --- a/app/agent/callback/__init__.py +++ b/app/agent/callback/__init__.py @@ -98,18 +98,27 @@ class StreamingHandler: self._thinking_started_at = time.monotonic() self._thinking_active = True self._thinking_status_text = self._format_thinking_status(0) - if self._can_stream(): - self._buffer = self._thinking_status_text + if self._should_buffer_thinking_status(): + self._buffer = f"{self._thinking_status_text}\n\n" def thinking_finished(self) -> None: """结束模型思考状态，移除尚未被正文替换的状态占位文本。""" with self._lock: if not self._thinking_active: return - if self._buffer == self._thinking_status_text: - self._buffer = "" - self._thinking_active = False - self._thinking_status_text = "" + self._remove_thinking_status_locked() + + def _should_buffer_thinking_status(self) -> bool: + """判断是否把思考状态写入消息渠道的可编辑正文。""" + return self._can_stream() + + def _remove_thinking_status_locked(self) -> None: + """从正文缓冲区移除思考状态块，同时结束本轮计时。""" + status_text = self._thinking_status_text + if status_text and self._buffer.startswith(status_text): + self._buffer = self._buffer[len(status_text) :].lstrip("\n") + self._thinking_active = False + self._thinking_status_text = "" def _format_thinking_status(self, elapsed_seconds: int) -> str: """生成消息渠道可直接展示的思考状态文本。""" @@ -118,11 +127,13 @@ class StreamingHandler: def _refresh_thinking_status(self) -> None: """更新可编辑消息中的思考计时文本，保持正文尚未生成时也有反馈。""" with self._lock: - if not self._thinking_active or not self._can_stream() or self._buffer != self._thinking_status_text: + if not self._thinking_active or not self._should_buffer_thinking_status(): return elapsed = int(max(0.0, time.monotonic() - (self._thinking_started_at or time.monotonic()))) - self._thinking_status_text = self._format_thinking_status(elapsed) - self._buffer = self._thinking_status_text + next_status_text = self._format_thinking_status(elapsed) + if self._buffer.startswith(self._thinking_status_text): + self._buffer = next_status_text + self._buffer[len(self._thinking_status_text) :] + self._thinking_status_text = next_status_text def set_dispatch_policy(self, allow_dispatch_without_context: bool = False) -> None: """ @@ -139,10 +150,7 @@ class StreamingHandler: with self._lock: emitted = token or "" if emitted and self._thinking_active: - if self._buffer == self._thinking_status_text: - self._buffer = "" - self._thinking_active = False - self._thinking_status_text = "" + self._remove_thinking_status_locked() if self._pending_tool_stats: if self._live_tool_summary: diff --git a/app/agent/web.py b/app/agent/web.py index f7dfcf036..019fe77b6 100644 --- a/app/agent/web.py +++ b/app/agent/web.py @@ -19,6 +19,10 @@ class _WebAgentStreamingHandlerMixin: _lock: Lock _pending_tool_stats: dict[str, int] + def _should_buffer_thinking_status(self) -> bool: + """WebAgent 通过结构化 SSE 展示思考状态，不把它写入正文。""" + return False + def __init__( self, on_emit: Callable[[str], None], diff --git a/tests/test_agent_streaming_lifecycle.py b/tests/test_agent_streaming_lifecycle.py index 87d253566..898278138 100644 --- a/tests/test_agent_streaming_lifecycle.py +++ b/tests/test_agent_streaming_lifecycle.py @@ -5,6 +5,7 @@ import asyncio import pytest from app.agent.callback import StreamingHandler +from app.agent.web import _get_web_agent_streaming_handler_type from app.schemas.types import NotificationChannel @@ -16,6 +17,7 @@ def test_thinking_status_is_replaced_by_first_answer_token() -> None: handler.thinking_started() assert handler._thinking_active is True assert "思考中" in handler._buffer + assert handler._buffer.endswith("\n\n") handler.emit("最终答案") @@ -23,6 +25,38 @@ def test_thinking_status_is_replaced_by_first_answer_token() -> None: assert handler._buffer == "最终答案" +def test_thinking_status_stays_separate_from_message_channel_tool_summary() -> None: + """消息渠道的思考状态应与工具统计保持独立段落，并在正文到达时移除。""" + handler = StreamingHandler() + handler._can_stream = lambda: True + handler._streaming_enabled = True + + handler.thinking_started() + handler.record_tool_call("read_file", tool_kwargs={"file_path": "README.md"}) + + assert handler._buffer.startswith("🤔 思考中 · 已用时 0 秒\n\n") + assert "\n\n（读取了 1 个文件）\n\n" in handler._buffer + + handler.emit("最终答案") + + assert "思考中" not in handler._buffer + assert handler._buffer.endswith("\n\n最终答案") + + +def test_web_agent_thinking_status_is_only_a_structured_event() -> None: + """WebAgent 的思考状态不能进入正文，只能通过 SSE 结构化事件展示。""" + events: list[dict[str, object]] = [] + handler = _get_web_agent_streaming_handler_type()(lambda _text: None, events.append) + + handler.thinking_started() + handler.thinking_finished() + + assert handler._buffer == "" + assert events[0]["type"] == "thinking" + assert events[0]["status"] == "running" + assert events[1] == {"type": "thinking", "status": "done"} + + @pytest.mark.asyncio async def test_repeated_streaming_start_retains_previous_flush_owner() -> None: """重复启动必须先等待旧 flush owner 结束，再发布新一轮上下文。"""