From adbf4471cf3c77c1613f31604c45d345e6b312f8 Mon Sep 17 00:00:00 2001 From: Liu Date: Fri, 1 May 2026 11:37:25 +0800 Subject: [PATCH] =?UTF-8?q?=E8=B0=83=E6=95=B4=E6=8F=92=E4=BB=B6=E6=89=A7?= =?UTF-8?q?=E8=A1=8C=E6=A8=A1=E5=BC=8F=E5=B9=B6=E4=BF=AE=E5=A4=8D=E5=85=A8?= =?UTF-8?q?=E7=90=83=E6=96=B0=E9=97=BB=E5=90=8E=E5=8F=B0=E7=BA=BF=E7=A8=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 为消息插件新增按消息动态超时能力，并让机器人侧按当前命令读取超时策略。 2. 将斗鱼日报、身价关系图/重算、百科问答出题判题切到后台执行。 3. 将系统更新、黑丝视频、猛男视频、成员锐评默认配置为后台模式并放宽超时。 4. 修复全球新闻插件在线程中直接挂协程导致任务不真正执行的问题。 --- .../plugin_common/message_plugin_interface.py | 10 +++++ plugins/douyu/main.py | 44 +++++++++++++++++++ plugins/game_task/main.py | 21 +++++++++ plugins/global_news/main.py | 38 ++++++++++------ plugins/member_roast/config.toml | 8 ++++ plugins/system_updater/config.toml | 10 ++++- plugins/value_rank/main.py | 25 +++++++++++ plugins/video/config.toml | 10 ++++- plugins/video_man/config.toml | 10 ++++- robot.py | 26 +++++++++-- 10 files changed, 183 insertions(+), 19 deletions(-) diff --git a/base/plugin_common/message_plugin_interface.py b/base/plugin_common/message_plugin_interface.py index 102bc63..bab2996 100644 --- a/base/plugin_common/message_plugin_interface.py +++ b/base/plugin_common/message_plugin_interface.py @@ -101,6 +101,16 @@ class MessagePluginInterface(PluginInterface): raw_mode = runtime_config.get("message_dispatch_mode") or runtime_config.get("dispatch_mode") or "sync" return self.normalize_message_dispatch_mode(raw_mode) + def get_message_process_timeout_seconds(self, message: Dict[str, Any]) -> Optional[int]: + """返回当前消息建议使用的插件总超时秒数。 + + 默认行为： + 1. 返回 `None`，表示继续沿用插件配置或机器人侧的自动推断逻辑； + 2. 适合“同一个插件里既有轻命令，也有重命令”的场景，避免所有命令共用同一个超时； + 3. 子类若需要按命令动态放宽超时，可覆盖本方法并返回正整数秒数。 + """ + return None + # ---------------- 插件定时调度能力（可选实现） ---------------- def get_schedule_actions(self) -> List[Dict[str, Any]]: """返回插件支持的可调度动作定义列表。 diff --git a/plugins/douyu/main.py b/plugins/douyu/main.py index bb6f71a..c1ea004 100644 --- a/plugins/douyu/main.py +++ b/plugins/douyu/main.py @@ -591,6 +591,20 @@ class DouyuPlugin(MessagePluginInterface): self._status_check_retry_count = 3 self._status_check_retry_delay_seconds = 1 self._daily_report_llm_client: Optional[UnifiedLLMClient] = None + # 斗鱼插件是典型“快命令 + 慢命令”混合体： + # 1. 订阅/列表类命令基本都是 Redis 读写，应该继续走前台同步，保证即时反馈； + # 2. 日报类命令会拉历史弹幕、调 LLM、渲染图片，天然属于长任务； + # 3. 因此这里把“慢命令名单”集中收口，供分发模式与超时策略共同复用。 + self._background_report_commands = { + "#斗鱼弹幕日报", + "斗鱼弹幕日报", + "#强制斗鱼弹幕日报", + "强制斗鱼弹幕日报", + "#斗鱼粉丝日报", + "斗鱼粉丝日报", + "#强制斗鱼粉丝日报", + "强制斗鱼粉丝日报", + } # 直播间语义画像： # 1. 允许按房间号补充“主播职业生涯、圈内关系、常见梗来源”等背景； # 2. 这些信息不会直接替代真实弹幕，只用于帮助 LLM 更准确理解圈内黑话； @@ -624,6 +638,36 @@ class DouyuPlugin(MessagePluginInterface): except Exception: return False, day_text + @staticmethod + def _extract_command_token(message: Dict[str, Any]) -> str: + """从消息里提取首个命令词。""" + content = str(message.get("content", "") or "").strip() + return content.split()[0] if content else "" + + def get_message_dispatch_mode(self, message: Dict[str, Any]) -> str: + """按命令决定斗鱼插件走前台还是后台。 + + 设计说明： + 1. 订阅、取消订阅、列表查询都很轻，继续前台执行能保证手感； + 2. 日报命令一旦命中，后面会进入“查素材 -> 调模型 -> 渲染图片”的长链路； + 3. 因此只有日报相关命令切后台，避免它们把前台 20 个消息槽位长期占住。 + """ + command = self._extract_command_token(message) + if command in self._background_report_commands: + return self.normalize_message_dispatch_mode("background") + return super().get_message_dispatch_mode(message) + + def get_message_process_timeout_seconds(self, message: Dict[str, Any]) -> Optional[int]: + """只为日报命令放宽总超时，普通命令继续走默认保护值。""" + command = self._extract_command_token(message) + if command in self._background_report_commands: + # 用户已经明确存在 200 秒级长任务： + # 1. 斗鱼日报除了 LLM，还包含素材整理与图片渲染； + # 2. 这里放宽到 15 分钟，足够覆盖补发历史日报或高峰期模型排队； + # 3. 非日报命令不受影响，仍保持原有更紧的保护策略。 + return 900 + return None + @staticmethod def _normalize_text_list(values: Any) -> List[str]: """ diff --git a/plugins/game_task/main.py b/plugins/game_task/main.py index cb8e997..f836964 100644 --- a/plugins/game_task/main.py +++ b/plugins/game_task/main.py @@ -59,6 +59,11 @@ class GameTaskPlugin(MessagePluginInterface): self.LOG = logger # 注册功能权限 self.feature = self.register_feature() + # 百科问答里只有“出题 / 判题”会真正走 LLM： + # 1. `/t` 需要模型随机生成题目； + # 2. `/a` 需要模型判分并给理由； + # 3. `/s /r /l /h` 主要是本地 DB 读写，保持前台即可。 + self._background_commands = {"/t", "/a"} def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件""" @@ -118,6 +123,22 @@ class GameTaskPlugin(MessagePluginInterface): return command in self._commands + def get_message_dispatch_mode(self, message: Dict[str, Any]) -> str: + """只把 LLM 型命令切到后台，避免百科插件拖慢前台消息链。""" + content = str(message.get("content", "") or "").strip() + command = content.split()[0] if content else "" + if command in self._background_commands: + return self.normalize_message_dispatch_mode("background") + return super().get_message_dispatch_mode(message) + + def get_message_process_timeout_seconds(self, message: Dict[str, Any]) -> Optional[int]: + """为出题/判题命令单独放宽超时，兼容慢模型或排队场景。""" + content = str(message.get("content", "") or "").strip() + command = content.split()[0] if content else "" + if command in self._background_commands: + return 120 + return None + def calculate_game_points(self, message: Dict[str, Any], success: bool, response: str) -> int: """计算游戏积分""" if not success: diff --git a/plugins/global_news/main.py b/plugins/global_news/main.py index 90a1c51..03dc85a 100644 --- a/plugins/global_news/main.py +++ b/plugins/global_news/main.py @@ -1,6 +1,6 @@ import asyncio import threading -import time # 添加这一行 +import time from typing import Dict, Any, List, Optional, Tuple from base.plugin_common.message_plugin_interface import MessagePluginInterface @@ -146,31 +146,43 @@ class GlobalNewsPlugin(MessagePluginInterface): self._news_tasks[task_id] = thread self.LOG.info(f"启动新闻获取任务: {task_id}") - async def _fetch_news_thread(self, task_id: str, sender: str, roomid: str): - """在单独的线程中运行异步新闻获取任务""" + def _fetch_news_thread(self, task_id: str, sender: str, roomid: str): + """在单独线程里执行新闻抓取主流程。 + + 这里必须保持为同步函数： + 1. `threading.Thread(target=...)` 只能直接执行普通可调用对象； + 2. 之前把协程函数直接塞给 `target`，线程里只会得到一个未执行的 coroutine，任务实际上不会跑； + 3. 现在在线程内部显式创建事件循环，再把异步抓取和发消息协程跑完，才能真正脱离主链路执行。 + """ + loop = asyncio.new_event_loop() try: - loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) news_result = loop.run_until_complete(self._fetch_news_async()) - loop.close() # 处理结果 + receiver = roomid if roomid else sender if news_result: - # 发送新闻图片 - receiver = roomid if roomid else sender - await self.bot.send_image_message(receiver, news_result) - await self.bot.send_text_message("🌍全球新闻获取完成！", receiver, sender) + # 在线程自有事件循环里把图片和完成提示真正发出去， + # 避免这里只拿到 coroutine 对象却没有执行。 + loop.run_until_complete(self.bot.send_image_message(receiver, news_result)) + loop.run_until_complete(self.bot.send_text_message(receiver, "🌍全球新闻获取完成！", sender)) else: - await self.bot.send_text_message( - (roomid if roomid else sender), "❌获取新闻失败，请稍后再试", sender) + loop.run_until_complete(self.bot.send_text_message(receiver, "❌获取新闻失败，请稍后再试", sender)) except Exception as e: self.LOG.error(f"新闻获取任务出错: {e}") - await self.bot.send_text_message((roomid if roomid else sender), f"❌获取新闻出错: {str(e)}", - sender) + try: + receiver = roomid if roomid else sender + loop.run_until_complete(self.bot.send_text_message(receiver, f"❌获取新闻出错: {str(e)}", sender)) + except Exception as send_error: + self.LOG.error(f"新闻获取失败后的通知发送异常: {send_error}") finally: # 清理任务 if task_id in self._news_tasks: del self._news_tasks[task_id] + try: + loop.close() + except Exception: + pass async def _fetch_news_async(self) -> str: """异步获取所有新闻源的新闻""" diff --git a/plugins/member_roast/config.toml b/plugins/member_roast/config.toml index 039ae98..4c34e40 100644 --- a/plugins/member_roast/config.toml +++ b/plugins/member_roast/config.toml @@ -48,3 +48,11 @@ history_group_summary_limit = 10 max_output_chars = 320 min_output_chars = 140 sharpness_level = "high" + +[runtime] +# 成员锐评命令需要同时查画像、拉最近消息，再走一次 LLM 生成： +# 1. 这条链路比普通查询明显更重，而且用户已经接受“先提示处理中，再稍后出结果”的交互； +# 2. 默认切到后台后，就不会因为某次模型慢响应把前台消息槽位卡住； +# 3. 总超时放宽到 4 分钟，兼容群画像较大或模型排队的情况。 +message_dispatch_mode = "background" +plugin_process_timeout_seconds = 240 diff --git a/plugins/system_updater/config.toml b/plugins/system_updater/config.toml index 4fcbf3e..21d0bb4 100644 --- a/plugins/system_updater/config.toml +++ b/plugins/system_updater/config.toml @@ -3,4 +3,12 @@ enable = true commands = ["更新系统", "系统更新", "重启系统", "更新重启"] wait_time = 5 # 设置管理员微信ID，只有这些ID可以执行更新操作 -shell_path= "/home/liuwei/abot/restart.sh" \ No newline at end of file +shell_path= "/home/liuwei/abot/restart.sh" + +[runtime] +# 系统更新属于典型后台维护任务： +# 1. 命令命中后会执行重启脚本，整个过程可能持续几十秒到数分钟； +# 2. 这类任务不应该长期占住前台消息并发槽位，否则会影响其他插件收消息； +# 3. 因此默认切到后台执行，并把总超时放宽到 10 分钟。 +message_dispatch_mode = "background" +plugin_process_timeout_seconds = 600 diff --git a/plugins/value_rank/main.py b/plugins/value_rank/main.py index 9ce3e8b..438e420 100644 --- a/plugins/value_rank/main.py +++ b/plugins/value_rank/main.py @@ -571,6 +571,11 @@ class ValueRankPlugin(MessagePluginInterface): self.mention_batch_size = 200 self.mention_window_start_minutes = 20 self.mention_window_end_minutes = 10 + # 身价排行里只有少数命令是真正的长任务： + # 1. `社交关系图` 需要拼 HTML 再截图渲染； + # 2. `重算身价` 会扫描整群候选成员并重写快照； + # 3. 其他榜单/说明类命令基本是读库拼文本，不值得全部切到后台。 + self._background_commands = {"社交关系图", "重算身价"} def initialize(self, context: Dict[str, Any]) -> bool: """初始化插件与配置。""" @@ -639,6 +644,26 @@ class ValueRankPlugin(MessagePluginInterface): command = content.split(" ")[0] return command in self._commands + def get_message_dispatch_mode(self, message: Dict[str, Any]) -> str: + """按命令决定是否切入后台任务池。""" + content = str(message.get("content", "") or "").strip() + command = content.split()[0] if content else "" + if command in self._background_commands: + # 这两个命令明显比普通查询重很多： + # 1. `社交关系图` 的瓶颈主要在模板渲染与截图； + # 2. `重算身价` 会遍历群成员并批量回写快照； + # 3. 改成后台后，轻量榜单查询就不会再被这类维护型命令拖住。 + return self.normalize_message_dispatch_mode("background") + return super().get_message_dispatch_mode(message) + + def get_message_process_timeout_seconds(self, message: Dict[str, Any]) -> Optional[int]: + """为重渲染/重算命令单独放宽总超时。""" + content = str(message.get("content", "") or "").strip() + command = content.split()[0] if content else "" + if command in self._background_commands: + return 240 + return None + @plugin_stats_decorator(plugin_name="身价排行") async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: """处理用户命令入口。""" diff --git a/plugins/video/config.toml b/plugins/video/config.toml index 631a3d2..14f8a10 100644 --- a/plugins/video/config.toml +++ b/plugins/video/config.toml @@ -4,4 +4,12 @@ command = ["黑丝视频", "黑丝", "来个黑丝", "搞个黑丝"] command-format = """ 🎬视频指令： 黑丝 -""" \ No newline at end of file +""" + +[runtime] +# 视频插件会经历“拉取接口 -> 下载文件 -> 抽首帧 -> 发送视频”整条链路： +# 1. 任一环节抖动都可能让处理时间明显长于普通文本命令； +# 2. 切到后台后，慢下载不会再卡住前台消息处理； +# 3. 总超时放宽到 4 分钟，兼容网络波动和大一点的视频文件。 +message_dispatch_mode = "background" +plugin_process_timeout_seconds = 240 diff --git a/plugins/video_man/config.toml b/plugins/video_man/config.toml index 8d55de2..e76b84b 100644 --- a/plugins/video_man/config.toml +++ b/plugins/video_man/config.toml @@ -6,4 +6,12 @@ command-format = """ 猛男 肌肉 帅哥 -""" \ No newline at end of file +""" + +[runtime] +# 猛男视频和普通视频插件的耗时结构基本一致： +# 1. 需要先拉接口，再下载视频文件，并额外抽首帧做封面； +# 2. 这些 IO/编解码步骤不适合长期占住前台并发槽位； +# 3. 因此同样默认走后台模式，并保留 4 分钟总超时。 +message_dispatch_mode = "background" +plugin_process_timeout_seconds = 240 diff --git a/robot.py b/robot.py index e5b007e..cc9fa49 100644 --- a/robot.py +++ b/robot.py @@ -674,7 +674,7 @@ class Robot: # 检查插件是否可以处理该消息 if plugin.can_process(plugin_msg): - protection_policy = self._build_message_plugin_protection_policy(plugin) + protection_policy = self._build_message_plugin_protection_policy(plugin, plugin_msg) acquire_result = self.plugin_manager.try_acquire_plugin_execution( plugin, recovery_seconds=protection_policy["circuit_recovery_seconds"], @@ -819,7 +819,7 @@ class Robot: except (TypeError, ValueError): return default - def _build_message_plugin_protection_policy(self, plugin) -> dict: + def _build_message_plugin_protection_policy(self, plugin, plugin_msg: dict = None) -> dict: """构建消息插件执行保护策略。""" plugin_config = getattr(plugin, "_config", {}) or {} runtime_config = plugin_config.get("runtime", {}) if isinstance(plugin_config, dict) else {} @@ -827,12 +827,32 @@ class Robot: breaker_config = runtime_config.get("circuit_breaker", {}) if isinstance(runtime_config, dict) else {} breaker_config = breaker_config if isinstance(breaker_config, dict) else {} + dynamic_timeout = 0 + if plugin_msg and hasattr(plugin, "get_message_process_timeout_seconds"): + try: + # 允许插件按“当前消息内容”给出更精细的超时建议： + # 1. 同一个插件里，日报/渲染/重算命令往往比普通查询慢很多； + # 2. 以前只能给整个插件统一配一个超时，容易出现“轻命令超时过大、重命令超时不够”的两难； + # 3. 这里把粒度放到单条消息，便于插件只给真正的长任务放宽保护时间。 + dynamic_timeout = self._safe_positive_int( + plugin.get_message_process_timeout_seconds(plugin_msg), + 0, + ) + except Exception as e: + self.LOG.warning( + self._trace_message( + plugin_msg.get("full_wx_msg"), + f"读取插件动态超时失败，已回退默认策略: plugin={plugin.name}, error={e}" + ) + ) + # 超时策略尽量遵循“显式配置优先，已有内部超时参数兜底”的思路： # 1. 新插件如果有特殊需求，只需要在 runtime / circuit_breaker 下声明自己的超时； # 2. 老插件不改代码也能自动复用现有的 request / llm / render 超时字段； # 3. 最终统一加一个缓冲区，避免外层 wait_for 比插件内部自己的超时还更早打断。 explicit_timeout = ( - runtime_config.get("plugin_process_timeout_seconds") + dynamic_timeout + or runtime_config.get("plugin_process_timeout_seconds") or runtime_config.get("message_timeout_seconds") or breaker_config.get("timeout_seconds") or getattr(plugin, "plugin_process_timeout_seconds", 0)