fix(agent): keep thinking status separate

This commit is contained in:
jxxghp
2026-10-01 10:26:22 +08:00
parent cd11e1abbf
commit 842ba547ec
3 changed files with 59 additions and 13 deletions

View File

@@ -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:

View File

@@ -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],

View File

@@ -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 结束,再发布新一轮上下文。"""