refactor: 统一类型注解风格并优化代码结构

- 将裸 except 改为显式 Exception 捕获
- 用列表推导式替换冗余 for 循环
- 为类属性添加 ClassVar 注解
- 统一 Union/Optional 写法为 |
- 移除未使用的导入
- 修复 SQLAlchemy 空值比较语法
- 优化字符串拼接与字典更新逻辑
- 补充缺失的 noqa 注释与异常链

BREAKING CHANGE: 所有插件基类的类级字段现要求显式 ClassVar 注解,自定义插件需同步更新
This commit is contained in:
明天好像没什么
2025-10-31 22:42:39 +08:00
parent 5080cfccfc
commit 0e129d385e
105 changed files with 592 additions and 561 deletions

View File

@@ -3,7 +3,7 @@ import asyncio
import random
import time
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, ClassVar
from src.chat.message_receive.chat_stream import ChatStream
from src.common.data_models.database_data_model import DatabaseMessages
@@ -26,30 +26,30 @@ class BaseAction(ABC):
新的激活机制 (推荐使用)
==================================================================================
推荐通过重写 go_activate() 方法来自定义激活逻辑:
示例 1 - 关键词激活:
async def go_activate(self, llm_judge_model=None) -> bool:
return await self._keyword_match(["你好", "hello"])
示例 2 - LLM 判断激活:
async def go_activate(self, llm_judge_model=None) -> bool:
return await self._llm_judge_activation(
"当用户询问天气信息时激活",
llm_judge_model
)
示例 3 - 组合多种条件:
async def go_activate(self, llm_judge_model=None) -> bool:
# 30% 随机概率,或者匹配关键词
if await self._random_activation(0.3):
return True
return await self._keyword_match(["表情", "emoji"])
提供的工具函数:
- _random_activation(probability): 随机激活
- _keyword_match(keywords, case_sensitive): 关键词匹配(自动获取聊天内容)
- _llm_judge_activation(judge_prompt, llm_judge_model): LLM 判断(自动获取聊天内容)
注意:聊天内容会自动从实例属性中获取,无需手动传入。
==================================================================================
@@ -68,7 +68,7 @@ class BaseAction(ABC):
==================================================================================
- mode_enable: 启用的聊天模式
- parallel_action: 是否允许并行执行
二步Action相关属性
- is_two_step_action: 是否为二步Action
- step_one_description: 第一步的描述
@@ -80,7 +80,7 @@ class BaseAction(ABC):
"""是否为二步Action。如果为TrueAction将分两步执行第一步选择操作第二步执行具体操作"""
step_one_description: str = ""
"""第一步的描述用于向LLM展示Action的基本功能"""
sub_actions: list[tuple[str, str, dict[str, str]]] = []
sub_actions: ClassVar[list[tuple[str, str, dict[str, str]]] ] = []
"""子Action列表格式为[(子Action名, 子Action描述, 子Action参数)]。仅在二步Action中使用"""
def __init__(
@@ -110,7 +110,7 @@ class BaseAction(ABC):
**kwargs: 其他参数
"""
if plugin_config is None:
plugin_config = {}
plugin_config: ClassVar = {}
self.action_data = action_data
self.reasoning = reasoning
self.cycle_timers = cycle_timers
@@ -489,7 +489,7 @@ class BaseAction(ABC):
plugin_config = component_registry.get_plugin_config(component_info.plugin_name)
# 3. 实例化被调用的Action
action_params = {
action_params: ClassVar = {
"action_data": called_action_data,
"reasoning": f"Called by {self.action_name}",
"cycle_timers": self.cycle_timers,
@@ -615,9 +615,9 @@ class BaseAction(ABC):
def _get_chat_content(self) -> str:
"""获取聊天内容用于激活判断
从实例属性中获取聊天内容。子类可以重写此方法来自定义获取逻辑。
Returns:
str: 聊天内容
"""
@@ -645,7 +645,7 @@ class BaseAction(ABC):
也可以使用提供的工具函数来简化常见的激活判断。
默认实现会检查类属性中的激活类型配置,提供向后兼容支持。
聊天内容会自动从实例属性中获取,不需要手动传入。
Args:
@@ -721,7 +721,7 @@ class BaseAction(ABC):
case_sensitive: bool = False,
) -> bool:
"""关键词匹配工具函数
聊天内容会自动从实例属性中获取。
Args:
@@ -742,7 +742,7 @@ class BaseAction(ABC):
if not case_sensitive:
search_text = search_text.lower()
matched_keywords = []
matched_keywords: ClassVar = []
for keyword in keywords:
check_keyword = keyword if case_sensitive else keyword.lower()
if check_keyword in search_text:
@@ -766,7 +766,7 @@ class BaseAction(ABC):
使用 LLM 来判断是否应该激活此 Action。
会自动构建完整的判断提示词,只需要提供核心判断逻辑即可。
聊天内容会自动从实例属性中获取。
Args:

View File

@@ -1,5 +1,5 @@
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, ClassVar
from src.common.data_models.message_manager_data_model import StreamContext
from src.plugin_system.base.component_types import ChatterInfo, ComponentType
@@ -15,7 +15,7 @@ class BaseChatter(ABC):
"""Chatter组件的名称"""
chatter_description: str = ""
"""Chatter组件的描述"""
chat_types: list[ChatType] = [ChatType.PRIVATE, ChatType.GROUP]
chat_types: ClassVar[list[ChatType]] = [ChatType.PRIVATE, ChatType.GROUP]
def __init__(self, stream_id: str, action_manager: "ChatterActionManager"):
"""

View File

@@ -1,4 +1,5 @@
from abc import ABC, abstractmethod
from typing import ClassVar
from src.common.logger import get_logger
@@ -21,7 +22,7 @@ class BaseEventHandler(ABC):
"""处理器权重,越大权重越高"""
intercept_message: bool = False
"""是否拦截消息,默认为否"""
init_subscribe: list[EventType | str] = [EventType.UNKNOWN]
init_subscribe: ClassVar[list[EventType | str]] = [EventType.UNKNOWN]
"""初始化时订阅的事件名称"""
plugin_name = None

View File

@@ -1,5 +1,5 @@
from abc import ABC, abstractmethod
from typing import Any
from typing import Any, ClassVar
from src.chat.utils.prompt_params import PromptParameters
from src.common.logger import get_logger
@@ -27,7 +27,7 @@ class BasePrompt(ABC):
# 定义此组件希望如何注入到核心Prompt中
# 这是一个 InjectionRule 对象的列表,可以实现复杂的注入逻辑
# 例如: [InjectionRule(target_prompt="planner_prompt", injection_type=InjectionType.APPEND, priority=50)]
injection_rules: list[InjectionRule] = []
injection_rules: ClassVar[list[InjectionRule] ] = []
"""定义注入规则的列表"""
# 旧的注入点定义,用于向后兼容。如果定义了这个,它将被自动转换为 injection_rules。

View File

@@ -1,5 +1,5 @@
from abc import ABC, abstractmethod
from typing import Any
from typing import Any, ClassVar
from rich.traceback import install
@@ -18,7 +18,7 @@ class BaseTool(ABC):
"""工具的名称"""
description: str = ""
"""工具的描述"""
parameters: list[tuple[str, ToolParamType, str, bool, list[str] | None]] = []
parameters: ClassVar[list[tuple[str, ToolParamType, str, bool, list[str] | None]] ] = []
"""工具的参数定义,为[("param_name", param_type, "description", required, enum_values)]格式
param_name: 参数名称
param_type: 参数类型
@@ -44,7 +44,7 @@ class BaseTool(ABC):
"""是否为二步工具。如果为True工具将分两步调用第一步展示工具信息第二步执行具体操作"""
step_one_description: str = ""
"""第一步的描述用于向LLM展示工具的基本功能"""
sub_tools: list[tuple[str, str, list[tuple[str, ToolParamType, str, bool, list[str] | None]]]] = []
sub_tools: ClassVar[list[tuple[str, str, list[tuple[str, ToolParamType, str, bool, list[str] | None]]]] ] = []
"""子工具列表,格式为[(子工具名, 子工具描述, 子工具参数)]。仅在二步工具中使用"""
def __init__(self, plugin_config: dict | None = None, chat_stream: Any = None):
@@ -112,7 +112,7 @@ class BaseTool(ABC):
if not cls.is_two_step_tool:
return []
definitions = []
definitions: ClassVar = []
for sub_name, sub_desc, sub_params in cls.sub_tools:
definitions.append({"name": f"{cls.name}_{sub_name}", "description": sub_desc, "parameters": sub_params})
return definitions

View File

@@ -3,7 +3,7 @@ import os
import shutil
from abc import ABC, abstractmethod
from pathlib import Path
from typing import Any
from typing import Any, ClassVar
import toml
@@ -30,11 +30,11 @@ class PluginBase(ABC):
config_file_name: str
enable_plugin: bool = True
config_schema: dict[str, dict[str, ConfigField] | str] = {}
config_schema: ClassVar[dict[str, dict[str, ConfigField] | str] ] = {}
permission_nodes: list["PermissionNodeField"] = []
permission_nodes: ClassVar[list["PermissionNodeField"] ] = []
config_section_descriptions: dict[str, str] = {}
config_section_descriptions: ClassVar[dict[str, str] ] = {}
def __init__(self, plugin_dir: str, metadata: PluginMetadata):
"""初始化插件
@@ -206,12 +206,12 @@ class PluginBase(ABC):
if not self.config_schema:
return {}
config_data = {}
config_data: ClassVar = {}
# 遍历每个配置节
for section, fields in self.config_schema.items():
if isinstance(fields, dict):
section_data = {}
section_data: ClassVar = {}
# 遍历节内的字段
for field_name, field in fields.items():
@@ -331,7 +331,7 @@ class PluginBase(ABC):
try:
with open(user_config_path, encoding="utf-8") as f:
user_config = toml.load(f) or {}
user_config: ClassVar = toml.load(f) or {}
except Exception as e:
logger.error(f"{self.log_prefix} 加载用户配置文件 {user_config_path} 失败: {e}", exc_info=True)
self.config = self._generate_config_from_schema() # 加载失败时使用默认 schema

View File

@@ -5,7 +5,7 @@
import re
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, ClassVar
from src.common.data_models.database_data_model import DatabaseMessages
from src.common.logger import get_logger
@@ -42,7 +42,7 @@ class PlusCommand(ABC):
command_description: str = ""
"""命令描述"""
command_aliases: list[str] = []
command_aliases: ClassVar[list[str] ] = []
"""命令别名列表,如 ['say', 'repeat']"""
priority: int = 0
@@ -435,7 +435,3 @@ def create_plus_command_adapter(plus_command_class):
return AdapterClass
# 兼容旧的命名
PlusCommandAdapter = create_plus_command_adapter