diff --git a/app/agent/callback/__init__.py b/app/agent/callback/__init__.py index 50d725679..0f3d0a1e5 100644 --- a/app/agent/callback/__init__.py +++ b/app/agent/callback/__init__.py @@ -78,6 +78,8 @@ class StreamingHandler: self._allow_dispatch_without_context = False # 非啰嗦模式下的待输出工具统计，等下一段文本到来时再统一补一句摘要 self._pending_tool_stats: dict[str, dict[str, Any]] = {} + # 本轮已写入缓冲区的工具摘要行，供 Telegram 富文本渲染时做区分样式 + self._tool_summaries: set[str] = set() def set_dispatch_policy( self, allow_dispatch_without_context: bool = False @@ -139,6 +141,7 @@ class StreamingHandler: self._message_response = None self._msg_start_offset = 0 self._pending_tool_stats = {} + self._tool_summaries = set() def reset(self): """ @@ -153,6 +156,7 @@ class StreamingHandler: self._sent_text = "" self._msg_start_offset = 0 self._pending_tool_stats = {} + self._tool_summaries = set() async def start_streaming( self, @@ -215,6 +219,7 @@ class StreamingHandler: self._message_response = None self._msg_start_offset = 0 self._pending_tool_stats = {} + self._tool_summaries = set() # 检查渠道是否支持消息编辑，不支持则仅收集 token 到 buffer，不实时推送 if not self._can_stream(): @@ -284,6 +289,7 @@ class StreamingHandler: self._message_response = None self._msg_start_offset = 0 self._pending_tool_stats = {} + self._tool_summaries = set() if all_sent: # 所有内容已通过流式发送，清空缓冲区 self._buffer = "" @@ -461,11 +467,14 @@ class StreamingHandler: return "" summary = f"（{'，'.join(parts)}）" + self._tool_summaries.add(summary) + # 摘要前始终保证一个空行，让工具执行信息与正文分属不同段落， + # 避免 Markdown 富文本把单个换行折叠成同一段落内的软换行 visible_buffer = self._buffer.rstrip(" \t") - last_char = visible_buffer[-1:] if visible_buffer.strip() else "" + trailing_newlines = len(visible_buffer) - len(visible_buffer.rstrip("\n")) prefix = "" - if self._buffer and last_char != "\n": - prefix = "\n\n" + if visible_buffer.strip(): + prefix = "\n" * max(2 - trailing_newlines, 0) return f"{prefix}{summary}\n\n" @staticmethod @@ -517,9 +526,23 @@ class StreamingHandler: """ 为 Telegram 流式消息返回 Rich Markdown，其他渠道继续使用原有格式。 """ - if self._channel == NotificationChannel.Telegram.value: + if self._channel != NotificationChannel.Telegram.value: + return None + return self._quote_tool_summary_lines(text) + + def _quote_tool_summary_lines(self, text: str) -> str: + """ + 将缓冲区中的工具摘要整行转换为 Markdown 引用块。 + + 富文本会把普通段落间的空行折叠成紧凑排版，引用块作为独立 block 类型 + 渲染，保证工具执行信息在 Telegram 上始终与正文有可辨识的视觉分隔。 + """ + if not self._tool_summaries or not text: return text - return None + return "\n".join( + f"> {line}" if line in self._tool_summaries else line + for line in text.split("\n") + ) async def _flush_loop(self): """ diff --git a/tests/test_agent_tool_streaming.py b/tests/test_agent_tool_streaming.py index 89dcfc2b9..01be1be78 100644 --- a/tests/test_agent_tool_streaming.py +++ b/tests/test_agent_tool_streaming.py @@ -179,12 +179,19 @@ class TestAgentToolStreaming: assert buffered_message == "好的，我来帮您执行\n抱歉，您没有执行此工具的权限" - def test_non_verbose_tool_call_reuses_existing_newline_before_summary(self): - """校验非详细模式复用已有换行追加工具摘要。""" + def test_non_verbose_tool_call_completes_blank_line_before_summary(self): + """校验非详细模式在摘要前补足空行，保证工具摘要独立成段。""" result, buffered_message = asyncio.run(self._run_tool("prefix\n")) assert result == "ok" - assert buffered_message == "prefix\n（调用了 1 次工具）\n\n" + assert buffered_message == "prefix\n\n（调用了 1 次工具）\n\n" + + def test_non_verbose_tool_call_keeps_existing_blank_line_before_summary(self): + """校验缓冲区已有空行时摘要不再追加多余换行。""" + result, buffered_message = asyncio.run(self._run_tool("prefix\n\n")) + + assert result == "ok" + assert buffered_message == "prefix\n\n（调用了 1 次工具）\n\n" def test_non_verbose_tool_call_emits_summary_even_when_buffer_was_empty(self): """校验空缓冲区仍会输出工具调用摘要。""" @@ -396,6 +403,55 @@ class TestAgentToolStreaming: assert notification.rich_message == "hello" assert handler.has_sent_message + def test_rich_message_quotes_tool_summary_lines_for_telegram(self): + """校验 Telegram 富文本将工具摘要行转换为引用块，与正文视觉分隔。""" + handler = StreamingHandler() + handler._channel = NotificationChannel.Telegram.value + handler._source = "telegram" + handler.record_tool_call( + tool_name="list_directory", + tool_message="查看目录", + tool_kwargs={"path": "/tmp"}, + ) + handler.flush_pending_tool_summary() + text = handler._buffer + + rich_message = handler._get_rich_message(text) + + assert text == "（查看了 1 个目录）\n\n" + assert rich_message == "> （查看了 1 个目录）\n\n" + + def test_rich_message_keeps_body_text_unquoted_for_telegram(self): + """校验 Telegram 富文本只转换工具摘要行，正文保持原样。""" + handler = StreamingHandler() + handler._channel = NotificationChannel.Telegram.value + handler.emit("正文内容\n\n") + handler.record_tool_call( + tool_name="execute_command", + tool_message="执行命令", + tool_kwargs={}, + ) + handler.emit("后续结论") + + rich_message = handler._get_rich_message(handler._buffer) + + assert rich_message == "正文内容\n\n> （执行了 1 条命令）\n\n后续结论" + + def test_rich_message_returns_none_for_non_telegram_channels(self): + """校验非 Telegram 渠道不启用富文本，摘要保持原有纯文本格式。""" + handler = StreamingHandler() + handler._channel = NotificationChannel.Feishu.value + handler._source = "feishu-main" + handler.record_tool_call( + tool_name="execute_command", + tool_message="执行命令", + tool_kwargs={}, + ) + handler.flush_pending_tool_summary() + + assert handler._get_rich_message(handler._buffer) is None + assert handler._buffer == "（执行了 1 条命令）\n\n" + def test_flush_edits_message_via_threadpool(self): """校验刷新时通过线程池编辑已有消息。""" handler = StreamingHandler()