fz-mirrors/AstrBot
1
0
Fork 0
mirror of https://github.com/AstrBotDevs/AstrBot synced 2026-07-15 17:30:13 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(pipeline): improve type safety and code quality

Browse Source 
stage.py:
- Add StageProcessResult type alias for better type checking
- Change process method signature for better compatibility

respond/stage.py & result_decorate/stage.py:
- Improve type annotations and code structure

content_safety_check/stage.py:
- Add better type handling

agent_sub_stages (internal.py, third_party.py):
- Improve type annotations for better code quality

Other pipeline stages:
- Minor improvements to type annotations
This commit is contained in:
LIghtJUNction
2026-03-31 20:16:59 +08:00
parent a306c63a92
commit 585ffb3982
16 changed files with 217 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
astrbot/core/pipeline/content_safety_check/stage.py
View File

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

44
astrbot/core/pipeline/content_safety_check/strategies/baidu_aip.py
View File

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

3
astrbot/core/pipeline/preprocess_stage/stage.py
View File

@@ -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.<platform>.pre_ack_emoji
supported = {"telegram", "lark", "discord"}

9
astrbot/core/pipeline/process_stage/method/agent_request.py
View File

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

42
astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py
View File

@@ -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"}

34
astrbot/core/pipeline/process_stage/method/agent_sub_stages/third_party.py
View File

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

4
astrbot/core/pipeline/process_stage/method/star_request.py
View File

@@ -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()

14
astrbot/core/pipeline/process_stage/stage.py
View File

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

6
astrbot/core/pipeline/rate_limit_check/stage.py
View File

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

84
astrbot/core/pipeline/respond/stage.py
View File

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

78
astrbot/core/pipeline/result_decorate/stage.py
View File

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

3
astrbot/core/pipeline/scheduler.py
View File

@@ -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:
"""初始化管道调度器时, 初始化所有阶段"""

4
astrbot/core/pipeline/session_status_check/stage.py
View File

@@ -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} 已被关闭,已终止事件传播｡")

10
astrbot/core/pipeline/stage.py
View File

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

4
astrbot/core/pipeline/waking_check/stage.py
View File

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

4
astrbot/core/pipeline/whitelist_check/stage.py
View File

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