From 6ac43c600e773c7ee627920febb354226fc54f06 Mon Sep 17 00:00:00 2001 From: Dt8333 <25431943+Dt8333@users.noreply.github.com> Date: Wed, 12 Nov 2025 18:01:20 +0800 Subject: [PATCH] perf: improve streaming fallback strategy for streaming-unsupported platform (#3547) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: 修改tool_loop_agent_runner，新增stream_to_general属性。 Co-authored-by: aider (openai/gemini-2.5-flash-preview) * refactor: 优化text_chat_stream，直接yield完整信息 Co-authored-by: aider (openai/gemini-2.5-flash-preview) * feat(core): :sparkles: 添加streaming_fallback选项，允许进行流式请求和非流式输出 添加了streaming_fallback配置，默认为false。在PlatformMetadata中新增字段用于标识是否支持真流式输出。在LLMRequest中添加判断是否启用Fallback。 #3431 #2793 #3014 * refactor(core): 将stream_to_general移出toolLoopAgentRunner * refactor(core.platform): 修改metadata中的属性名称 * fix: update streaming provider settings descriptions and add conditions * fix: update streaming configuration to use unsupported_streaming_strategy and adjust related logic * fix: remove support_streaming_message flag from WecomAIBotAdapter registration * fix: update hint for non-streaming platform handling in configuration * fix(core.pipeline): Update astrbot/core/pipeline/process_stage/method/llm_request.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * fix(core.pipeline): Update astrbot/core/pipeline/process_stage/method/llm_request.py --------- Co-authored-by: aider (openai/gemini-2.5-flash-preview) Co-authored-by: Soulter <37870767+Soulter@users.noreply.github.com> Co-authored-by: Soulter <905617992@qq.com> Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- astrbot/core/config/default.py | 18 ++++++++++++------ .../process_stage/method/llm_request.py | 19 ++++++++++++++++--- astrbot/core/pipeline/respond/stage.py | 11 +++++++---- astrbot/core/platform/platform_metadata.py | 3 +++ astrbot/core/platform/register.py | 2 ++ .../aiocqhttp/aiocqhttp_platform_adapter.py | 2 ++ .../sources/dingtalk/dingtalk_adapter.py | 5 ++++- .../discord/discord_platform_adapter.py | 5 ++++- .../platform/sources/lark/lark_adapter.py | 5 ++++- .../sources/misskey/misskey_adapter.py | 5 ++++- .../platform/sources/satori/satori_adapter.py | 4 ++-- .../platform/sources/slack/slack_adapter.py | 2 ++ .../wechatpadpro/wechatpadpro_adapter.py | 5 ++++- .../platform/sources/wecom/wecom_adapter.py | 3 ++- .../weixin_offacc_adapter.py | 5 ++++- .../src/components/shared/AstrBotConfigV4.vue | 3 ++- 16 files changed, 74 insertions(+), 23 deletions(-) diff --git a/astrbot/core/config/default.py b/astrbot/core/config/default.py index 1021d81b5..7cd024e70 100644 --- a/astrbot/core/config/default.py +++ b/astrbot/core/config/default.py @@ -68,7 +68,7 @@ DEFAULT_CONFIG = { "dequeue_context_length": 1, "streaming_response": False, "show_tool_use_status": False, - "streaming_segmented": False, + "unsupported_streaming_strategy": "realtime_segmenting", "max_agent_step": 30, "tool_call_timeout": 60, }, @@ -1993,8 +1993,8 @@ CONFIG_METADATA_2 = { "show_tool_use_status": { "type": "bool", }, - "streaming_segmented": { - "type": "bool", + "unsupported_streaming_strategy": { + "type": "string", }, "max_agent_step": { "description": "工具调用轮数上限", @@ -2299,9 +2299,15 @@ CONFIG_METADATA_3 = { "description": "流式回复", "type": "bool", }, - "provider_settings.streaming_segmented": { - "description": "不支持流式回复的平台采取分段输出", - "type": "bool", + "provider_settings.unsupported_streaming_strategy": { + "description": "不支持流式回复的平台", + "type": "string", + "options": ["realtime_segmenting", "turn_off"], + "hint": "选择在不支持流式回复的平台上的处理方式。实时分段回复会在系统接收流式响应检测到诸如标点符号等分段点时，立即发送当前已接收的内容", + "labels": ["实时分段回复", "关闭流式回复"], + "condition": { + "provider_settings.streaming_response": True, + }, }, "provider_settings.max_context_length": { "description": "最多携带对话轮数", diff --git a/astrbot/core/pipeline/process_stage/method/llm_request.py b/astrbot/core/pipeline/process_stage/method/llm_request.py index 5974cd519..f7677d373 100644 --- a/astrbot/core/pipeline/process_stage/method/llm_request.py +++ b/astrbot/core/pipeline/process_stage/method/llm_request.py @@ -294,6 +294,7 @@ async def run_agent( agent_runner: AgentRunner, max_step: int = 30, show_tool_use: bool = True, + stream_to_general: bool = False, ) -> AsyncGenerator[MessageChain, None]: step_idx = 0 astr_event = agent_runner.run_context.context.event @@ -321,7 +322,10 @@ async def run_agent( await astr_event.send(resp.data["chain"]) continue - if not agent_runner.streaming: + if stream_to_general and resp.type == "streaming_delta": + continue + + if stream_to_general or not agent_runner.streaming: content_typ = ( ResultContentType.LLM_RESULT if resp.type == "llm_result" @@ -363,6 +367,9 @@ class LLMRequestSubStage(Stage): self.max_context_length - 1, ) self.streaming_response: bool = settings["streaming_response"] + self.unsupported_streaming_strategy: str = settings[ + "unsupported_streaming_strategy" + ] self.max_step: int = settings.get("max_agent_step", 30) self.tool_call_timeout: int = settings.get("tool_call_timeout", 60) if isinstance(self.max_step, bool): # workaround: #2622 @@ -540,6 +547,10 @@ class LLMRequestSubStage(Stage): new_tool_set.add_tool(tool) req.func_tool = new_tool_set + stream_to_general = ( + self.unsupported_streaming_strategy == "turn_off" + and not event.platform_meta.support_streaming_message + ) # 备份 req.contexts backup_contexts = copy.deepcopy(req.contexts) @@ -567,7 +578,7 @@ class LLMRequestSubStage(Stage): streaming=streaming_response, ) - if streaming_response: + if streaming_response and not stream_to_general: # 流式响应 event.set_result( MessageEventResult() @@ -594,7 +605,9 @@ class LLMRequestSubStage(Stage): ), ) else: - async for _ in run_agent(agent_runner, self.max_step, self.show_tool_use): + async for _ in run_agent( + agent_runner, self.max_step, self.show_tool_use, stream_to_general + ): yield # 恢复备份的 contexts diff --git a/astrbot/core/pipeline/respond/stage.py b/astrbot/core/pipeline/respond/stage.py index f20445594..86946d023 100644 --- a/astrbot/core/pipeline/respond/stage.py +++ b/astrbot/core/pipeline/respond/stage.py @@ -169,12 +169,15 @@ class RespondStage(Stage): logger.warning("async_stream 为空，跳过发送。") return # 流式结果直接交付平台适配器处理 - use_fallback = self.config.get("provider_settings", {}).get( - "streaming_segmented", - False, + realtime_segmenting = ( + self.config.get("provider_settings", {}).get( + "unsupported_streaming_strategy", + "realtime_segmenting", + ) + == "realtime_segmenting" ) logger.info(f"应用流式输出({event.get_platform_id()})") - await event.send_streaming(result.async_stream, use_fallback) + await event.send_streaming(result.async_stream, realtime_segmenting) return if len(result.chain) > 0: # 检查路径映射 diff --git a/astrbot/core/platform/platform_metadata.py b/astrbot/core/platform/platform_metadata.py index d75811245..c63bd82b1 100644 --- a/astrbot/core/platform/platform_metadata.py +++ b/astrbot/core/platform/platform_metadata.py @@ -16,3 +16,6 @@ class PlatformMetadata: """显示在 WebUI 配置页中的平台名称，如空则是 name""" logo_path: str | None = None """平台适配器的 logo 文件路径（相对于插件目录）""" + + support_streaming_message: bool = True + """平台是否支持真实流式传输""" diff --git a/astrbot/core/platform/register.py b/astrbot/core/platform/register.py index 0c6267492..c1721c5c5 100644 --- a/astrbot/core/platform/register.py +++ b/astrbot/core/platform/register.py @@ -14,6 +14,7 @@ def register_platform_adapter( default_config_tmpl: dict | None = None, adapter_display_name: str | None = None, logo_path: str | None = None, + support_streaming_message: bool = True, ): """用于注册平台适配器的带参装饰器。 @@ -42,6 +43,7 @@ def register_platform_adapter( default_config_tmpl=default_config_tmpl, adapter_display_name=adapter_display_name, logo_path=logo_path, + support_streaming_message=support_streaming_message, ) platform_registry.append(pm) platform_cls_map[adapter_name] = cls diff --git a/astrbot/core/platform/sources/aiocqhttp/aiocqhttp_platform_adapter.py b/astrbot/core/platform/sources/aiocqhttp/aiocqhttp_platform_adapter.py index 81deead13..91a16c697 100644 --- a/astrbot/core/platform/sources/aiocqhttp/aiocqhttp_platform_adapter.py +++ b/astrbot/core/platform/sources/aiocqhttp/aiocqhttp_platform_adapter.py @@ -29,6 +29,7 @@ from .aiocqhttp_message_event import AiocqhttpMessageEvent @register_platform_adapter( "aiocqhttp", "适用于 OneBot V11 标准的消息平台适配器，支持反向 WebSockets。", + support_streaming_message=False, ) class AiocqhttpAdapter(Platform): def __init__( @@ -49,6 +50,7 @@ class AiocqhttpAdapter(Platform): name="aiocqhttp", description="适用于 OneBot 标准的消息平台适配器，支持反向 WebSockets。", id=self.config.get("id"), + support_streaming_message=False, ) self.bot = CQHttp( diff --git a/astrbot/core/platform/sources/dingtalk/dingtalk_adapter.py b/astrbot/core/platform/sources/dingtalk/dingtalk_adapter.py index 43d231771..3ffb71493 100644 --- a/astrbot/core/platform/sources/dingtalk/dingtalk_adapter.py +++ b/astrbot/core/platform/sources/dingtalk/dingtalk_adapter.py @@ -37,7 +37,9 @@ class MyEventHandler(dingtalk_stream.EventHandler): return AckMessage.STATUS_OK, "OK" -@register_platform_adapter("dingtalk", "钉钉机器人官方 API 适配器") +@register_platform_adapter( + "dingtalk", "钉钉机器人官方 API 适配器", support_streaming_message=False +) class DingtalkPlatformAdapter(Platform): def __init__( self, @@ -86,6 +88,7 @@ class DingtalkPlatformAdapter(Platform): name="dingtalk", description="钉钉机器人官方 API 适配器", id=self.config.get("id"), + support_streaming_message=False, ) async def convert_msg( diff --git a/astrbot/core/platform/sources/discord/discord_platform_adapter.py b/astrbot/core/platform/sources/discord/discord_platform_adapter.py index 2752f3a9b..49b886dea 100644 --- a/astrbot/core/platform/sources/discord/discord_platform_adapter.py +++ b/astrbot/core/platform/sources/discord/discord_platform_adapter.py @@ -34,7 +34,9 @@ else: # 注册平台适配器 -@register_platform_adapter("discord", "Discord 适配器 (基于 Pycord)") +@register_platform_adapter( + "discord", "Discord 适配器 (基于 Pycord)", support_streaming_message=False +) class DiscordPlatformAdapter(Platform): def __init__( self, @@ -111,6 +113,7 @@ class DiscordPlatformAdapter(Platform): "Discord 适配器", id=self.config.get("id"), default_config_tmpl=self.config, + support_streaming_message=False, ) @override diff --git a/astrbot/core/platform/sources/lark/lark_adapter.py b/astrbot/core/platform/sources/lark/lark_adapter.py index b59dbaca4..e6e6d4d2b 100644 --- a/astrbot/core/platform/sources/lark/lark_adapter.py +++ b/astrbot/core/platform/sources/lark/lark_adapter.py @@ -23,7 +23,9 @@ from ...register import register_platform_adapter from .lark_event import LarkMessageEvent -@register_platform_adapter("lark", "飞书机器人官方 API 适配器") +@register_platform_adapter( + "lark", "飞书机器人官方 API 适配器", support_streaming_message=False +) class LarkPlatformAdapter(Platform): def __init__( self, @@ -115,6 +117,7 @@ class LarkPlatformAdapter(Platform): name="lark", description="飞书机器人官方 API 适配器", id=self.config.get("id"), + support_streaming_message=False, ) async def convert_msg(self, event: lark.im.v1.P2ImMessageReceiveV1): diff --git a/astrbot/core/platform/sources/misskey/misskey_adapter.py b/astrbot/core/platform/sources/misskey/misskey_adapter.py index 0a553dc6f..ddeec93bc 100644 --- a/astrbot/core/platform/sources/misskey/misskey_adapter.py +++ b/astrbot/core/platform/sources/misskey/misskey_adapter.py @@ -45,7 +45,9 @@ MAX_FILE_UPLOAD_COUNT = 16 DEFAULT_UPLOAD_CONCURRENCY = 3 -@register_platform_adapter("misskey", "Misskey 平台适配器") +@register_platform_adapter( + "misskey", "Misskey 平台适配器", support_streaming_message=False +) class MisskeyPlatformAdapter(Platform): def __init__( self, @@ -120,6 +122,7 @@ class MisskeyPlatformAdapter(Platform): description="Misskey 平台适配器", id=self.config.get("id", "misskey"), default_config_tmpl=default_config, + support_streaming_message=False, ) async def run(self): diff --git a/astrbot/core/platform/sources/satori/satori_adapter.py b/astrbot/core/platform/sources/satori/satori_adapter.py index b5751ebd2..fd90804f0 100644 --- a/astrbot/core/platform/sources/satori/satori_adapter.py +++ b/astrbot/core/platform/sources/satori/satori_adapter.py @@ -29,8 +29,7 @@ from astrbot.core.platform.astr_message_event import MessageSession @register_platform_adapter( - "satori", - "Satori 协议适配器", + "satori", "Satori 协议适配器", support_streaming_message=False ) class SatoriPlatformAdapter(Platform): def __init__( @@ -60,6 +59,7 @@ class SatoriPlatformAdapter(Platform): name="satori", description="Satori 通用协议适配器", id=self.config["id"], + support_streaming_message=False, ) self.ws: ClientConnection | None = None diff --git a/astrbot/core/platform/sources/slack/slack_adapter.py b/astrbot/core/platform/sources/slack/slack_adapter.py index 6bb5a505e..d5427deb7 100644 --- a/astrbot/core/platform/sources/slack/slack_adapter.py +++ b/astrbot/core/platform/sources/slack/slack_adapter.py @@ -30,6 +30,7 @@ from .slack_event import SlackMessageEvent @register_platform_adapter( "slack", "适用于 Slack 的消息平台适配器，支持 Socket Mode 和 Webhook Mode。", + support_streaming_message=False, ) class SlackAdapter(Platform): def __init__( @@ -68,6 +69,7 @@ class SlackAdapter(Platform): name="slack", description="适用于 Slack 的消息平台适配器，支持 Socket Mode 和 Webhook Mode。", id=self.config.get("id"), + support_streaming_message=False, ) # 初始化 Slack Web Client diff --git a/astrbot/core/platform/sources/wechatpadpro/wechatpadpro_adapter.py b/astrbot/core/platform/sources/wechatpadpro/wechatpadpro_adapter.py index 165375cd5..e8629ec11 100644 --- a/astrbot/core/platform/sources/wechatpadpro/wechatpadpro_adapter.py +++ b/astrbot/core/platform/sources/wechatpadpro/wechatpadpro_adapter.py @@ -32,7 +32,9 @@ except ImportError as e: ) -@register_platform_adapter("wechatpadpro", "WeChatPadPro 消息平台适配器") +@register_platform_adapter( + "wechatpadpro", "WeChatPadPro 消息平台适配器", support_streaming_message=False +) class WeChatPadProAdapter(Platform): def __init__( self, @@ -51,6 +53,7 @@ class WeChatPadProAdapter(Platform): name="wechatpadpro", description="WeChatPadPro 消息平台适配器", id=self.config.get("id", "wechatpadpro"), + support_streaming_message=False, ) # 保存配置信息 diff --git a/astrbot/core/platform/sources/wecom/wecom_adapter.py b/astrbot/core/platform/sources/wecom/wecom_adapter.py index ffd5ec8ee..1ea4c8e20 100644 --- a/astrbot/core/platform/sources/wecom/wecom_adapter.py +++ b/astrbot/core/platform/sources/wecom/wecom_adapter.py @@ -110,7 +110,7 @@ class WecomServer: await self.shutdown_event.wait() -@register_platform_adapter("wecom", "wecom 适配器") +@register_platform_adapter("wecom", "wecom 适配器", support_streaming_message=False) class WecomPlatformAdapter(Platform): def __init__( self, @@ -196,6 +196,7 @@ class WecomPlatformAdapter(Platform): "wecom", "wecom 适配器", id=self.config.get("id", "wecom"), + support_streaming_message=False, ) @override diff --git a/astrbot/core/platform/sources/weixin_official_account/weixin_offacc_adapter.py b/astrbot/core/platform/sources/weixin_official_account/weixin_offacc_adapter.py index f44b06e90..d1309374f 100644 --- a/astrbot/core/platform/sources/weixin_official_account/weixin_offacc_adapter.py +++ b/astrbot/core/platform/sources/weixin_official_account/weixin_offacc_adapter.py @@ -113,7 +113,9 @@ class WecomServer: await self.shutdown_event.wait() -@register_platform_adapter("weixin_official_account", "微信公众平台 适配器") +@register_platform_adapter( + "weixin_official_account", "微信公众平台 适配器", support_streaming_message=False +) class WeixinOfficialAccountPlatformAdapter(Platform): def __init__( self, @@ -195,6 +197,7 @@ class WeixinOfficialAccountPlatformAdapter(Platform): "weixin_official_account", "微信公众平台 适配器", id=self.config.get("id", "weixin_official_account"), + support_streaming_message=False, ) @override diff --git a/dashboard/src/components/shared/AstrBotConfigV4.vue b/dashboard/src/components/shared/AstrBotConfigV4.vue index 6ae758dfb..bfe99667b 100644 --- a/dashboard/src/components/shared/AstrBotConfigV4.vue +++ b/dashboard/src/components/shared/AstrBotConfigV4.vue @@ -154,7 +154,8 @@ function hasVisibleItemsAfter(items, currentIndex) {