diff --git a/astrbot/core/pipeline/content_safety_check/stage.py b/astrbot/core/pipeline/content_safety_check/stage.py index e336050ce..e35b96ec1 100644 --- a/astrbot/core/pipeline/content_safety_check/stage.py +++ b/astrbot/core/pipeline/content_safety_check/stage.py @@ -20,14 +20,20 @@ class ContentSafetyCheckStage(Stage): config = ctx.astrbot_config["content_safety"] self.strategy_selector = StrategySelector(config) - async def process( # type: ignore[invalid-method-override] + async def process( self, event: AstrMessageEvent, - check_text: str | None = None, + ) -> AsyncGenerator[None, None]: + async for item in self.process_text(event, event.get_message_str()): + yield item + + async def process_text( + self, + event: AstrMessageEvent, + check_text: str, ) -> AsyncGenerator[None, None]: """检查内容安全""" - text = check_text if check_text else event.get_message_str() - ok, info = self.strategy_selector.check(text) + ok, info = self.strategy_selector.check(check_text) if not ok: if event.is_at_or_wake_command: event.set_result( @@ -35,7 +41,7 @@ class ContentSafetyCheckStage(Stage): "你的消息或者大模型的响应中包含不适当的内容,已被屏蔽｡", ), ) - yield + yield None event.stop_event() logger.info(f"内容安全检查不通过,原因:{info}") return diff --git a/astrbot/core/pipeline/content_safety_check/strategies/baidu_aip.py b/astrbot/core/pipeline/content_safety_check/strategies/baidu_aip.py index ca8a9dc8f..b8db3aea8 100644 --- a/astrbot/core/pipeline/content_safety_check/strategies/baidu_aip.py +++ b/astrbot/core/pipeline/content_safety_check/strategies/baidu_aip.py @@ -1,14 +1,30 @@ """使用此功能应该先 pip install baidu-aip""" -from typing import Any, cast - -from aip import AipContentCensor +from typing import TypedDict, TypeGuard from . import ContentSafetyStrategy +class BaiduAipViolation(TypedDict, total=False): + msg: str + + +def _is_violation_list(value: object) -> TypeGuard[list[BaiduAipViolation]]: + if not isinstance(value, list): + return False + for item in value: + if not isinstance(item, dict): + return False + message = item.get("msg") + if message is not None and not isinstance(message, str): + return False + return True + + class BaiduAipStrategy(ContentSafetyStrategy): def __init__(self, appid: str, ak: str, sk: str) -> None: + from aip import AipContentCensor # type: ignore[unresolved-import] + self.app_id = appid self.api_key = ak self.secret_key = sk @@ -16,17 +32,23 @@ class BaiduAipStrategy(ContentSafetyStrategy): def check(self, content: str) -> tuple[bool, str]: res = self.client.textCensorUserDefined(content) - if "conclusionType" not in res: + conclusion_type = res.get("conclusionType") + if not isinstance(conclusion_type, int): return False, "" - if res["conclusionType"] == 1: + if conclusion_type == 1: return True, "" - if "data" not in res: + + data = res.get("data") + conclusion = res.get("conclusion") + if not _is_violation_list(data) or not isinstance(conclusion, str): return False, "" - count = len(res["data"]) + + count = len(data) parts = [f"百度审核服务发现 {count} 处违规:\n"] - for i in res["data"]: - # 百度 AIP 返回结构是动态 dict;类型检查时 i 可能被推断为序列,转成 dict 后用 get 取字段 - parts.append(f"{cast(dict[str, Any], i).get('msg', '')};\n") - parts.append("\n判断结果:" + res["conclusion"]) + for item in data: + message = item.get("msg") + if message: + parts.append(f"{message};\n") + parts.append("\n判断结果:" + conclusion) info = "".join(parts) return False, info diff --git a/astrbot/core/pipeline/preprocess_stage/stage.py b/astrbot/core/pipeline/preprocess_stage/stage.py index 56171a211..ce4aff15f 100644 --- a/astrbot/core/pipeline/preprocess_stage/stage.py +++ b/astrbot/core/pipeline/preprocess_stage/stage.py @@ -1,7 +1,6 @@ import asyncio import random import traceback -from collections.abc import AsyncGenerator from astrbot.core import logger from astrbot.core.message.components import Image, Plain, Record @@ -23,7 +22,7 @@ class PreProcessStage(Stage): async def process( self, event: AstrMessageEvent, - ) -> None | AsyncGenerator[None, None]: + ) -> None: """在处理事件之前的预处理""" # 平台特异配置:platform_specific..pre_ack_emoji supported = {"telegram", "lark", "discord"} diff --git a/astrbot/core/pipeline/process_stage/method/agent_request.py b/astrbot/core/pipeline/process_stage/method/agent_request.py index e7efc2cb9..1eec0884f 100644 --- a/astrbot/core/pipeline/process_stage/method/agent_request.py +++ b/astrbot/core/pipeline/process_stage/method/agent_request.py @@ -25,15 +25,14 @@ class AgentRequestSubStage(Stage): self.prov_wake_prefix = self.prov_wake_prefix[len(bwp) :] agent_runner_type = self.config["provider_settings"]["agent_runner_type"] + self.agent_sub_stage: InternalAgentSubStage | ThirdPartyAgentSubStage if agent_runner_type == "local": self.agent_sub_stage = InternalAgentSubStage() else: self.agent_sub_stage = ThirdPartyAgentSubStage() await self.agent_sub_stage.initialize(ctx) - async def process( # type: ignore[invalid-method-override] - self, event: AstrMessageEvent - ) -> AsyncGenerator[None, None]: + async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None, None]: if not self.ctx.astrbot_config["provider_settings"]["enable"]: logger.debug( "This pipeline does not enable AI capability, skip processing." @@ -46,5 +45,5 @@ class AgentRequestSubStage(Stage): ) return - async for resp in self.agent_sub_stage.process(event): - yield resp + async for _ in self.agent_sub_stage.process(event): + yield None diff --git a/astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py b/astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py index 4df0aad02..6c0731871 100644 --- a/astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py +++ b/astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py @@ -151,10 +151,10 @@ class InternalAgentSubStage(Stage): max_quoted_fallback_images=settings.get("max_quoted_fallback_images", 20), ) - async def process( # type: ignore[invalid-method-override] + async def process( self, event: AstrMessageEvent, - ) -> None | AsyncGenerator[None, None]: + ) -> AsyncGenerator[None, None]: follow_up_capture: FollowUpCapture | None = None follow_up_consumed_marked = False follow_up_activated = False @@ -325,7 +325,7 @@ class InternalAgentSubStage(Stage): ), ), ) - yield + yield None # 保存历史记录 if agent_runner.done() and ( @@ -355,7 +355,7 @@ class InternalAgentSubStage(Stage): ), ), ) - yield + yield None if agent_runner.done(): if final_llm_resp := agent_runner.get_final_llm_resp(): if final_llm_resp.completion_text: @@ -383,7 +383,7 @@ class InternalAgentSubStage(Stage): stream_to_general, show_reasoning=self.show_reasoning, ): - yield + yield None final_resp = agent_runner.get_final_llm_resp() @@ -450,7 +450,7 @@ class InternalAgentSubStage(Stage): consumed_marked=follow_up_consumed_marked, ) - async def _save_to_history( # type: ignore[invalid-method-override] + async def _save_to_history( self, event: AstrMessageEvent, req: ProviderRequest, @@ -514,6 +514,36 @@ class InternalAgentSubStage(Stage): ) +async def _record_internal_agent_stats( + event: AstrMessageEvent, + req: ProviderRequest, + agent_runner: AgentRunner, + llm_response: LLMResponse | None, +) -> None: + from astrbot.core import db_helper + + status = "aborted" if agent_runner.was_aborted() else "completed" + if llm_response is None and not agent_runner.was_aborted(): + status = "error" + + provider_id = str(agent_runner.provider.provider_config.get("id", "") or "unknown") + provider_model = agent_runner.provider.get_model() or None + conversation_id = req.conversation.cid if req.conversation else None + + try: + await db_helper.insert_provider_stat( + agent_type="internal", + status=status, + umo=event.unified_msg_origin, + conversation_id=conversation_id, + provider_id=provider_id, + provider_model=provider_model, + stats=agent_runner.stats.to_dict(), + ) + except Exception: + logger.warning("record internal agent stats failed", exc_info=True) + + # we prevent astrbot from connecting to known malicious hosts # these hosts are base64 encoded BLOCKED = {"dGZid2h2d3IuY2xvdWQuc2VhbG9zLmlv", "a291cmljaGF0"} diff --git a/astrbot/core/pipeline/process_stage/method/agent_sub_stages/third_party.py b/astrbot/core/pipeline/process_stage/method/agent_sub_stages/third_party.py index 37c56d906..ad37d44a6 100644 --- a/astrbot/core/pipeline/process_stage/method/agent_sub_stages/third_party.py +++ b/astrbot/core/pipeline/process_stage/method/agent_sub_stages/third_party.py @@ -181,7 +181,7 @@ class ThirdPartyAgentSubStage(Stage): source="Third-party runner config", ) - async def _resolve_persona_custom_error_message( # type: ignore[invalid-method-override] + async def _resolve_persona_custom_error_message( self, event: AstrMessageEvent ) -> str | None: try: @@ -199,7 +199,7 @@ class ThirdPartyAgentSubStage(Stage): logger.debug("Failed to resolve persona custom error message: %s", e) return None - async def _handle_streaming_response( # type: ignore[invalid-method-override] + async def _handle_streaming_response( self, *, runner: "BaseAgentRunner", @@ -232,7 +232,7 @@ class ThirdPartyAgentSubStage(Stage): .set_result_content_type(ResultContentType.STREAMING_RESULT) .set_async_stream(_stream_runner_chain()), ) - yield + yield None if runner.done(): final_chain, is_runner_error = aggregator.finalize( @@ -246,7 +246,7 @@ class ThirdPartyAgentSubStage(Stage): ), ) - async def _handle_non_streaming_response( # type: ignore[invalid-method-override] + async def _handle_non_streaming_response( self, *, runner: "BaseAgentRunner", @@ -279,12 +279,12 @@ class ThirdPartyAgentSubStage(Stage): ), ) # Second yield keeps scheduler progress consistent after final result update. - yield + yield None - async def process( # type: ignore[invalid-method-override] + async def process( self, event: AstrMessageEvent, - ) -> None | AsyncGenerator[None, None]: + ) -> AsyncGenerator[None, None]: req: ProviderRequest | None = None if self.provider_wake_prefix and not event.message_str.startswith( @@ -345,7 +345,9 @@ class ThirdPartyAgentSubStage(Stage): DifyAgentRunner, ) - runner = DifyAgentRunner[AstrAgentContext]() + runner: BaseAgentRunner[AstrAgentContext] = DifyAgentRunner[ + AstrAgentContext + ]() elif self.runner_type == "coze": from astrbot.core.agent.runners.coze.coze_agent_runner import ( CozeAgentRunner, @@ -402,13 +404,25 @@ class ThirdPartyAgentSubStage(Stage): stream_watchdog_task.cancel() try: + from astrbot.core.astr_agent_tool_exec import FunctionToolExecutor + + provider = self.ctx.plugin_manager.context.get_using_provider( + umo=event.unified_msg_origin, + ) + if provider is None: + raise ValueError( + "No active provider is available for third-party runner" + ) + await runner.reset( + provider=provider, request=req, run_context=AgentContextWrapper( context=astr_agent_ctx, tool_call_timeout=120, session_manager=ToolSessionManager(), ), + tool_executor=FunctionToolExecutor(), agent_hooks=MAIN_AGENT_HOOKS, provider_config=self.prov_cfg, streaming=streaming_response, @@ -427,7 +441,7 @@ class ThirdPartyAgentSubStage(Stage): close_runner_once=close_runner_once, mark_stream_consumed=mark_stream_consumed, ): - yield + yield None else: async for _ in self._handle_non_streaming_response( runner=runner, @@ -435,7 +449,7 @@ class ThirdPartyAgentSubStage(Stage): stream_to_general=stream_to_general, custom_error_message=custom_error_message, ): - yield + yield None finally: if ( stream_watchdog_task diff --git a/astrbot/core/pipeline/process_stage/method/star_request.py b/astrbot/core/pipeline/process_stage/method/star_request.py index 03a021d35..dbada63c3 100644 --- a/astrbot/core/pipeline/process_stage/method/star_request.py +++ b/astrbot/core/pipeline/process_stage/method/star_request.py @@ -19,7 +19,7 @@ class StarRequestSubStage(Stage): self.identifier = ctx.astrbot_config["provider_settings"]["identifier"] self.ctx = ctx - async def process( # type: ignore[invalid-method-override] + async def process( self, event: AstrMessageEvent, ) -> AsyncGenerator[Any, None]: @@ -80,7 +80,7 @@ class StarRequestSubStage(Stage): if not event.is_stopped() and event.is_at_or_wake_command: ret = f":(\n\n在调用插件 {md.name} 的处理函数 {handler.handler_name} 时出现异常:{e}" event.set_result(MessageEventResult().message(ret)) - yield + yield None event.clear_result() event.stop_event() diff --git a/astrbot/core/pipeline/process_stage/stage.py b/astrbot/core/pipeline/process_stage/stage.py index f8051c31b..2714aface 100644 --- a/astrbot/core/pipeline/process_stage/stage.py +++ b/astrbot/core/pipeline/process_stage/stage.py @@ -28,10 +28,10 @@ class ProcessStage(Stage): self.star_request_sub_stage = StarRequestSubStage() await self.star_request_sub_stage.initialize(ctx) - async def process( # type: ignore[invalid-method-override] + async def process( self, event: AstrMessageEvent, - ) -> None | AsyncGenerator[None, None]: + ) -> AsyncGenerator[None, None]: """处理事件""" activated_handlers: list[StarHandlerMetadata] = event.get_extra( "activated_handlers", @@ -46,16 +46,16 @@ class ProcessStage(Stage): _t = False async for _ in self.agent_sub_stage.process(event): _t = True - yield + yield None if not _t: - yield + yield None else: - yield + yield None if self.sdk_plugin_bridge is not None and not event.is_stopped(): sdk_result = await self.sdk_plugin_bridge.dispatch_message(event) if sdk_result.sent_message or sdk_result.stopped: - yield + yield None # 调用 LLM 相关请求 if not self.ctx.astrbot_config["provider_settings"].get("enable", True): @@ -77,4 +77,4 @@ class ProcessStage(Stage): # 是否有过发送操作 and 是否是被 @ 或者通过唤醒前缀 if (effective_result and not event.is_stopped()) or not effective_result: async for _ in self.agent_sub_stage.process(event): - yield + yield None diff --git a/astrbot/core/pipeline/rate_limit_check/stage.py b/astrbot/core/pipeline/rate_limit_check/stage.py index 4668e9e9b..49ad2f56e 100644 --- a/astrbot/core/pipeline/rate_limit_check/stage.py +++ b/astrbot/core/pipeline/rate_limit_check/stage.py @@ -1,6 +1,5 @@ import asyncio from collections import defaultdict, deque -from collections.abc import AsyncGenerator from datetime import datetime, timedelta from astrbot.core import logger @@ -42,7 +41,7 @@ class RateLimitStage(Stage): async def process( self, event: AstrMessageEvent, - ) -> None | AsyncGenerator[None, None]: + ) -> None: """检查并处理限流逻辑｡如果触发限流,流水线会 stall 并在窗口期后自动恢复｡ Args: @@ -79,7 +78,8 @@ class RateLimitStage(Stage): logger.info( f"会话 {session_id} 被限流｡根据限流策略,此请求已被丢弃,直到限额于 {stall_duration:.2f} 秒后重置｡", ) - return event.stop_event() + event.stop_event() + return def _remove_expired_timestamps( self, diff --git a/astrbot/core/pipeline/respond/stage.py b/astrbot/core/pipeline/respond/stage.py index 15da2711a..df24a913a 100644 --- a/astrbot/core/pipeline/respond/stage.py +++ b/astrbot/core/pipeline/respond/stage.py @@ -1,8 +1,6 @@ import asyncio import math import random -from collections.abc import AsyncGenerator -from typing import ClassVar import astrbot.core.message.components as Comp from astrbot.core import logger @@ -17,38 +15,46 @@ from astrbot.core.utils.path_util import path_Mapping @register_stage class RespondStage(Stage): - # 组件类型到其非空判断函数的映射 - _component_validators: ClassVar[dict[type, lambda comp: bool]] = { - Comp.Plain: lambda comp: bool( - comp.text and comp.text.strip(), - ), # 纯文本消息需要strip - Comp.Face: lambda comp: comp.id is not None, # QQ表情 - Comp.Record: lambda comp: bool(comp.file), # 语音 - Comp.Video: lambda comp: bool(comp.file), # 视频 - Comp.At: lambda comp: bool(comp.qq) or bool(comp.name), # @ - Comp.Image: lambda comp: bool(comp.file), # 图片 - Comp.Reply: lambda comp: bool(comp.id) and comp.sender_id is not None, # 回复 - Comp.Poke: lambda comp: comp.target_id() is not None, # 戳一戳 - Comp.Node: lambda comp: bool(comp.content), # 转发节点 - Comp.Nodes: lambda comp: bool(comp.nodes), # 多个转发节点 - Comp.File: lambda comp: bool(comp.file_ or comp.url), - Comp.WechatEmoji: lambda comp: comp.md5 is not None, # 微信表情 - Comp.Json: lambda comp: bool(comp.data), # Json 卡片 - Comp.Share: lambda comp: bool(comp.url) or bool(comp.title), - Comp.Music: lambda comp: ( - (comp.id and comp._type and comp._type != "custom") - or (comp._type == "custom" and comp.url and comp.audio and comp.title) - ), # 音乐分享 - Comp.Forward: lambda comp: bool(comp.id), # 合并转发 - Comp.Location: lambda comp: bool( - comp.lat is not None and comp.lon is not None - ), # 位置 - Comp.Contact: lambda comp: bool(comp._type and comp.id), # 推荐好友 or 群 - Comp.Shake: lambda _: True, # 窗口抖动(戳一戳) - Comp.Dice: lambda _: True, # 掷骰子魔法表情 - Comp.RPS: lambda _: True, # 猜拳魔法表情 - Comp.Unknown: lambda comp: bool(comp.text and comp.text.strip()), - } + @staticmethod + def _has_meaningful_content(comp: BaseMessageComponent) -> bool: + if isinstance(comp, Comp.Plain | Comp.Unknown): + return bool(comp.text.strip()) + if isinstance(comp, Comp.Face): + return comp.id is not None + if isinstance(comp, Comp.Record | Comp.Video | Comp.Image): + return bool(comp.file) + if isinstance(comp, Comp.At): + return bool(comp.qq) or bool(comp.name) + if isinstance(comp, Comp.Reply): + return bool(comp.id) and comp.sender_id is not None + if isinstance(comp, Comp.Poke): + return comp.target_id() is not None + if isinstance(comp, Comp.Node): + return bool(comp.content) + if isinstance(comp, Comp.Nodes): + return bool(comp.nodes) + if isinstance(comp, Comp.File): + return bool(comp.file_ or comp.url) + if isinstance(comp, Comp.WechatEmoji): + return comp.md5 is not None + if isinstance(comp, Comp.Json): + return bool(comp.data) + if isinstance(comp, Comp.Share): + return bool(comp.url) or bool(comp.title) + if isinstance(comp, Comp.Music): + return bool( + (comp.id and comp._type and comp._type != "custom") + or (comp._type == "custom" and comp.url and comp.audio and comp.title) + ) + if isinstance(comp, Comp.Forward): + return bool(comp.id) + if isinstance(comp, Comp.Location): + return comp.lat is not None and comp.lon is not None + if isinstance(comp, Comp.Contact): + return bool(comp._type and comp.id) + if isinstance(comp, Comp.Shake | Comp.Dice | Comp.RPS): + return True + return False async def initialize(self, ctx: PipelineContext) -> None: self.ctx = ctx @@ -118,12 +124,8 @@ class RespondStage(Stage): return True for comp in chain: - comp_type = type(comp) - - # 检查组件类型是否在字典中 - if comp_type in self._component_validators: - if self._component_validators[comp_type](comp): - return False + if self._has_meaningful_content(comp): + return False # 如果所有组件都为空 return True @@ -170,7 +172,7 @@ class RespondStage(Stage): async def process( self, event: AstrMessageEvent, - ) -> None | AsyncGenerator[None, None]: + ) -> None: result = event.get_result() if result is None: return diff --git a/astrbot/core/pipeline/result_decorate/stage.py b/astrbot/core/pipeline/result_decorate/stage.py index 3c567288a..a3108dcd0 100644 --- a/astrbot/core/pipeline/result_decorate/stage.py +++ b/astrbot/core/pipeline/result_decorate/stage.py @@ -5,7 +5,16 @@ import traceback from collections.abc import AsyncGenerator from astrbot.core import file_token_service, html_renderer, logger -from astrbot.core.message.components import At, Image, Json, Node, Plain, Record, Reply +from astrbot.core.message.components import ( + At, + BaseMessageComponent, + Image, + Json, + Node, + Plain, + Record, + Reply, +) from astrbot.core.message.message_event_result import ResultContentType from astrbot.core.pipeline.content_safety_check.stage import ContentSafetyCheckStage from astrbot.core.pipeline.context import PipelineContext @@ -73,6 +82,7 @@ class ResultDecorateStage(Stage): self.split_words = ctx.astrbot_config["platform_settings"][ "segmented_reply" ].get("split_words", ["。", "？", "！", "~", "…"]) + self.split_words_pattern: re.Pattern[str] | None if self.split_words: escaped_words = sorted( [re.escape(word) for word in self.split_words], key=len, reverse=True @@ -90,12 +100,15 @@ class ResultDecorateStage(Stage): self.content_safe_check_reply = ctx.astrbot_config["content_safety"][ "also_use_in_response" ] - self.content_safe_check_stage = None + self.content_safe_check_stage: ContentSafetyCheckStage | None = None if self.content_safe_check_reply: for stage_cls in registered_stages: if stage_cls.__name__ == "ContentSafetyCheckStage": - self.content_safe_check_stage = stage_cls() - await self.content_safe_check_stage.initialize(ctx) + stage = stage_cls() + if isinstance(stage, ContentSafetyCheckStage): + self.content_safe_check_stage = stage + await stage.initialize(ctx) + break provider_cfg = ctx.astrbot_config.get("provider_settings", {}) self.show_reasoning = provider_cfg.get("display_reasoning_text", False) @@ -106,26 +119,20 @@ class ResultDecorateStage(Stage): return [text] segments = self.split_words_pattern.findall(text) - result = [] - for seg in segments: - if isinstance(seg, tuple): - content = seg[0] - if not isinstance(content, str): - continue - for word in self.split_words: - if content.endswith(word): - content = content[: -len(word)] - break - if content.strip(): - result.append(content) - elif seg and seg.strip(): - result.append(seg) + result: list[str] = [] + for content, _ in segments: + for word in self.split_words: + if content.endswith(word): + content = content[: -len(word)] + break + if content.strip(): + result.append(content) return result if result else [text] - async def process( # type: ignore[invalid-method-override] + async def process( self, event: AstrMessageEvent, - ) -> None | AsyncGenerator[None, None]: + ) -> AsyncGenerator[None, None]: result = event.get_result() if result is None or not result.chain: return @@ -148,11 +155,8 @@ class ResultDecorateStage(Stage): text += comp.text if isinstance(self.content_safe_check_stage, ContentSafetyCheckStage): - async for _ in self.content_safe_check_stage.process( - event, - check_text=text, - ): - yield + async for _ in self.content_safe_check_stage.process_text(event, text): + yield None # 发送消息前事件钩子 handlers = star_handlers_registry.get_handlers_by_event_type( @@ -210,7 +214,7 @@ class ResultDecorateStage(Stage): if ( self.only_llm_result and result.is_model_result() ) or not self.only_llm_result: - new_chain = [] + new_chain: list[BaseMessageComponent] = [] for comp in result.chain: if isinstance(comp, Plain): if len(comp.text) > self.words_count_threshold: @@ -256,17 +260,17 @@ class ResultDecorateStage(Stage): event.unified_msg_origin, ) - should_tts = ( + tts_requested = ( bool(self.ctx.astrbot_config["provider_tts_settings"]["enable"]) and result.is_llm_result() and await SessionServiceManager.should_process_tts_request(event) and random.random() <= self.tts_trigger_probability - and tts_provider ) - if should_tts and not tts_provider: + if tts_requested and tts_provider is None: logger.warning( f"会话 {event.unified_msg_origin} 未配置文本转语音模型。", ) + should_tts = tts_requested and tts_provider is not None if ( not should_tts @@ -291,7 +295,7 @@ class ResultDecorateStage(Stage): result.chain.insert(0, Plain(f"🤔 思考: {reasoning_content}\n")) if should_tts and tts_provider: - new_chain = [] + tts_chain: list[BaseMessageComponent] = [] for comp in result.chain: if isinstance(comp, Plain) and len(comp.text) > 1: try: @@ -302,7 +306,7 @@ class ResultDecorateStage(Stage): logger.error( f"由于 TTS 音频文件未找到，消息段转语音失败: {comp.text}", ) - new_chain.append(comp) + tts_chain.append(comp) continue use_file_service = self.ctx.astrbot_config[ @@ -315,7 +319,7 @@ class ResultDecorateStage(Stage): "provider_tts_settings" ]["dual_output"] - url = None + url: str | None = None if use_file_service and callback_api_base: token = await file_token_service.register_file( audio_path, @@ -323,7 +327,7 @@ class ResultDecorateStage(Stage): url = f"{callback_api_base}/api/file/{token}" logger.debug(f"已注册：{url}") - new_chain.append( + tts_chain.append( Record( file=url or audio_path, url=url or audio_path, @@ -331,14 +335,14 @@ class ResultDecorateStage(Stage): ), ) if dual_output: - new_chain.append(comp) + tts_chain.append(comp) except Exception: logger.error(traceback.format_exc()) logger.error("TTS 失败，使用文本发送。") - new_chain.append(comp) + tts_chain.append(comp) else: - new_chain.append(comp) - result.chain = new_chain + tts_chain.append(comp) + result.chain = tts_chain # 文本转图片 elif ( diff --git a/astrbot/core/pipeline/scheduler.py b/astrbot/core/pipeline/scheduler.py index 02efb8d24..e49dbd14b 100644 --- a/astrbot/core/pipeline/scheduler.py +++ b/astrbot/core/pipeline/scheduler.py @@ -1,4 +1,5 @@ from collections.abc import AsyncGenerator +from typing import Any from astrbot.core import logger from astrbot.core.platform import AstrMessageEvent @@ -23,7 +24,7 @@ class PipelineScheduler: key=lambda x: STAGES_ORDER.index(x.__name__), ) # 按照顺序排序 self.ctx = context # 上下文对象 - self.stages = [] # 存储阶段实例 + self.stages: list[Any] = [] # 存储阶段实例 async def initialize(self) -> None: """初始化管道调度器时, 初始化所有阶段""" diff --git a/astrbot/core/pipeline/session_status_check/stage.py b/astrbot/core/pipeline/session_status_check/stage.py index 9682a08ec..c7636089d 100644 --- a/astrbot/core/pipeline/session_status_check/stage.py +++ b/astrbot/core/pipeline/session_status_check/stage.py @@ -1,5 +1,3 @@ -from collections.abc import AsyncGenerator - from astrbot.core import logger from astrbot.core.pipeline.context import PipelineContext from astrbot.core.pipeline.stage import Stage, register_stage @@ -18,7 +16,7 @@ class SessionStatusCheckStage(Stage): async def process( self, event: AstrMessageEvent, - ) -> None | AsyncGenerator[None, None]: + ) -> None: # 检查会话是否整体启用 if not await SessionServiceManager.is_session_enabled(event.unified_msg_origin): logger.debug(f"会话 {event.unified_msg_origin} 已被关闭,已终止事件传播｡") diff --git a/astrbot/core/pipeline/stage.py b/astrbot/core/pipeline/stage.py index 03fd61540..b063213b9 100644 --- a/astrbot/core/pipeline/stage.py +++ b/astrbot/core/pipeline/stage.py @@ -1,13 +1,15 @@ from __future__ import annotations import abc -from collections.abc import AsyncGenerator +from collections.abc import AsyncGenerator, Awaitable +from typing import Any, TypeAlias from astrbot.core.platform.astr_message_event import AstrMessageEvent from .context import PipelineContext registered_stages: list[type[Stage]] = [] # 维护了所有已注册的 Stage 实现类类型 +StageProcessResult: TypeAlias = AsyncGenerator[Any, None] | Awaitable[None] def register_stage(cls): @@ -30,16 +32,16 @@ class Stage(abc.ABC): raise NotImplementedError @abc.abstractmethod - async def process( + def process( self, event: AstrMessageEvent, - ) -> None | AsyncGenerator[None, None]: + ) -> StageProcessResult: """处理事件 Args: event (AstrMessageEvent): 事件对象,包含事件的相关信息 Returns: - Union[None, AsyncGenerator[None, None]]: 处理结果,可能是 None 或者异步生成器, 如果为 None 则表示不需要继续处理, 如果为异步生成器则表示需要继续处理(进入下一个阶段) + StageProcessResult: 处理结果,可能是普通 awaitable 或异步生成器｡ """ raise NotImplementedError diff --git a/astrbot/core/pipeline/waking_check/stage.py b/astrbot/core/pipeline/waking_check/stage.py index 2048793bc..54ec0bf8b 100644 --- a/astrbot/core/pipeline/waking_check/stage.py +++ b/astrbot/core/pipeline/waking_check/stage.py @@ -1,4 +1,4 @@ -from collections.abc import AsyncGenerator, Callable +from collections.abc import Callable from astrbot import logger from astrbot.core.message.components import At, AtAll, Reply @@ -75,7 +75,7 @@ class WakingCheckStage(Stage): async def process( self, event: AstrMessageEvent, - ) -> None | AsyncGenerator[None, None]: + ) -> None: # apply unique session if self.unique_session and event.message_obj.type == MessageType.GROUP_MESSAGE: sid = build_unique_session_id(event) diff --git a/astrbot/core/pipeline/whitelist_check/stage.py b/astrbot/core/pipeline/whitelist_check/stage.py index 917e42959..879e35441 100644 --- a/astrbot/core/pipeline/whitelist_check/stage.py +++ b/astrbot/core/pipeline/whitelist_check/stage.py @@ -1,5 +1,3 @@ -from collections.abc import AsyncGenerator - from astrbot.core import logger from astrbot.core.pipeline.context import PipelineContext from astrbot.core.pipeline.stage import Stage, register_stage @@ -30,7 +28,7 @@ class WhitelistCheckStage(Stage): async def process( self, event: AstrMessageEvent, - ) -> None | AsyncGenerator[None, None]: + ) -> None: if not self.enable_whitelist_check: # 白名单检查未启用 return