Losita/mai-bot
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'r-dev' of https://github.com/A-Dawn/MaiBot into r-dev

Browse Source
This commit is contained in:
DawnARC
2026-04-13 13:10:02 +08:00
parent b3fb662f3d 412166ed7e
commit 056a3af642
83 changed files with 3789 additions and 3693 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
src/config/config.py
View File

@@ -23,7 +23,6 @@ from .official_configs import (
EmojiConfig,
ExpressionConfig,
KeywordReactionConfig,
LPMMKnowledgeConfig,
MaiSakaConfig,
MaimMessageConfig,
MCPConfig,
@@ -54,9 +53,10 @@ PROJECT_ROOT: Path = Path(__file__).parent.parent.parent.absolute().resolve()
CONFIG_DIR: Path = PROJECT_ROOT / "config"
BOT_CONFIG_PATH: Path = (CONFIG_DIR / "bot_config.toml").resolve().absolute()
MODEL_CONFIG_PATH: Path = (CONFIG_DIR / "model_config.toml").resolve().absolute()
LEGACY_ENV_PATH: Path = (PROJECT_ROOT / ".env").resolve().absolute()
MMC_VERSION: str = "1.0.0"
CONFIG_VERSION: str = "8.5.2"
MODEL_CONFIG_VERSION: str = "1.13.1"
CONFIG_VERSION: str = "8.7.1"
MODEL_CONFIG_VERSION: str = "1.14.0"
logger = get_logger("config")
@@ -454,6 +454,20 @@ def generate_new_config_file(config_class: type[T], config_path: Path, inner_con
write_config_to_file(config, config_path, inner_config_version)
def remove_legacy_env_file(env_path: Path) -> None:
"""删除已完成迁移的旧版 `.env` 文件。"""
if not env_path.exists():
return
try:
env_path.unlink()
except OSError as exc:
logger.warning(f"旧版 .env 配置文件删除失败，请手动删除: {env_path}，原因: {exc}")
else:
logger.warning(f"检测到旧版环境变量绑定配置迁移成功，已删除旧版 .env 文件: {env_path}")
def load_config_from_file(
config_class: type[T], config_path: Path, new_ver: str, override_repr: bool = False
) -> tuple[T, bool]:
@@ -467,10 +481,12 @@ def load_config_from_file(
if not isinstance(inner_version, str):
raise TypeError(t("config.invalid_inner_version"))
old_ver: str = inner_version
env_migration_applied: bool = False
config_data.remove("inner") # 移除 inner 部分，避免干扰后续处理
config_data = config_data.unwrap() # 转换为普通字典，方便后续处理
if config_path.name == "bot_config.toml" and config_class.__name__ == "Config":
env_migration = migrate_legacy_bind_env_to_bot_config_dict(config_data)
env_migration_applied = env_migration.migrated
if env_migration.migrated:
logger.warning(f"检测到旧版环境变量绑定配置，已迁移到主配置: {env_migration.reason}")
config_data = env_migration.data
@@ -497,9 +513,11 @@ def load_config_from_file(
raise e
else:
raise e
if compare_versions(old_ver, new_ver):
if compare_versions(old_ver, new_ver) or env_migration_applied:
output_config_changes(attribute_data, logger, old_ver, new_ver, config_path.name)
write_config_to_file(target_config, config_path, new_ver, override_repr)
if env_migration_applied:
remove_legacy_env_file(LEGACY_ENV_PATH)
updated = True
return target_config, updated
except Exception as e:

329
src/config/legacy_migration.py
View File

@@ -1,12 +1,8 @@
"""
legacy_migration.py
一个“可随时拔掉”的旧配置兼容层：
- 仅在配置解析失败时尝试修复旧格式数据（7.x -> 8.x 这一类结构性变更）
- 不依赖 Pydantic / ConfigBase，仅对 dict 做最小转换
- 成功则返回(修复后的 dict, True)，失败则返回(原 dict, False)
设计目标：与现有 config 加载逻辑的接触点尽可能小，未来不需要时可一键移除。
旧配置兼容层。
仅保留当前仍需要的“解析前结构修复”，避免老配置在 `from_dict` 前直接失败。
"""
from __future__ import annotations
@@ -16,12 +12,7 @@ from typing import Any, Optional
import os
from src.common.logger import get_logger
logger = get_logger("legacy_migration")
# 方便未来快速关闭/移除
ENABLE_LEGACY_MIGRATION: bool = True
@@ -43,6 +34,7 @@ def _as_list(x: Any) -> Optional[list[Any]]:
def _parse_host_env(value: Any) -> Optional[str]:
if not isinstance(value, str):
return None
normalized_value = value.strip()
return normalized_value or None
@@ -75,116 +67,73 @@ def _migrate_env_value(section: dict[str, Any], key: str, parsed_env_value: Any,
return True
def _move_section_key(source: dict[str, Any], target: dict[str, Any], key: str) -> bool:
"""将配置项从旧分组移动到新分组，若新分组已有值则保留新值。"""
if key not in source:
return False
if key not in target:
target[key] = source[key]
source.pop(key, None)
return True
def _parse_triplet_target(s: str) -> Optional[dict[str, str]]:
"""
解析 "platform:id:type" -> {platform,item_id,rule_type}
返回 None 表示无法解析。
解析 "platform:id:type" -> {platform, item_id, rule_type}
"""
if not isinstance(s, str):
return None
parts = s.split(":", 2)
if len(parts) != 3:
return None
platform, item_id, rule_type = parts
if rule_type not in ("group", "private"):
return None
return {"platform": platform, "item_id": item_id, "rule_type": rule_type}
def _parse_quad_prompt(s: str) -> Optional[dict[str, str]]:
"""
解析 "platform:id:type:prompt" -> {platform,item_id,rule_type,prompt}
prompt 允许包含冒号，因此只切前三个冒号。
"""
if not isinstance(s, str):
return None
parts = s.split(":", 3)
if len(parts) != 4:
return None
platform, item_id, rule_type, prompt = parts
if rule_type not in ("group", "private"):
return None
if not prompt:
return None
return {"platform": platform, "item_id": item_id, "rule_type": rule_type, "prompt": prompt}
def _parse_enable_disable(v: Any) -> Optional[bool]:
"""
兼容旧值 "enable"/"disable" 以及 bool。
"""
if isinstance(v, bool):
return v
if isinstance(v, str):
vv = v.strip().lower()
if vv == "enable":
normalized_value = v.strip().lower()
if normalized_value == "enable":
return True
if vv == "disable":
if normalized_value == "disable":
return False
return None
def _migrate_expression_learning_list(expr: dict[str, Any]) -> bool:
"""
旧：
learning_list = [
["", "enable", "enable", "enable"],
["qq:1919810:group", "enable", "enable", "enable"],
]
兼容旧旧格式：
learning_list = [
["qq:1919810:group", "enable", "enable", "0.5"],
["", "disable", "disable", "0.1"],
]
新：
[[expression.learning_list]]
platform="", item_id="", rule_type="group", use_expression=true, enable_learning=true, enable_jargon_learning=true
将旧版 expression.learning_list 转成当前结构。
"""
ll = _as_list(expr.get("learning_list"))
if ll is None:
learning_list = _as_list(expr.get("learning_list"))
if learning_list is None:
return False
# 如果已经是新格式（列表里是 dict），跳过
if ll and all(isinstance(i, dict) for i in ll):
if learning_list and all(isinstance(item, dict) for item in learning_list):
return False
migrated_items: list[dict[str, Any]] = []
for row in ll:
r = _as_list(row)
if r is None or len(r) < 4:
# 行结构不对，无法安全迁移
for row in learning_list:
row_items = _as_list(row)
if row_items is None or len(row_items) < 4:
return False
target_raw = r[0]
use_expression = _parse_enable_disable(r[1])
enable_learning = _parse_enable_disable(r[2])
enable_jargon_learning = _parse_enable_disable(r[3])
target_raw = row_items[0]
use_expression = _parse_enable_disable(row_items[1])
enable_learning = _parse_enable_disable(row_items[2])
enable_jargon_learning = _parse_enable_disable(row_items[3])
if enable_jargon_learning is None:
# 更早期的配置在第 4 列记录的是一个已废弃的数值权重/阈值，
# 当前 schema 已没有对应字段。这里按保守策略兼容迁移：
# 丢弃旧数值，并将 enable_jargon_learning 置为 False。
# 更早期版本第 4 列是已废弃的数值阈值，这里仅做保守兼容。
try:
float(str(r[3]))
float(str(row_items[3]))
except (TypeError, ValueError):
pass
else:
enable_jargon_learning = False
if use_expression is None or enable_learning is None or enable_jargon_learning is None:
return False
# 旧格式中 target 允许为空字符串：表示全局；新结构必须有三元组字段
if target_raw == "" or target_raw is None:
target = {"platform": "", "item_id": "", "rule_type": "group"}
else:
@@ -209,99 +158,56 @@ def _migrate_expression_learning_list(expr: dict[str, Any]) -> bool:
def _migrate_expression_groups(expr: dict[str, Any]) -> bool:
"""
旧：
expression_groups = [
["qq:1:group","qq:2:group"],
["qq:3:group"],
]
新：
expression_groups = [
{ expression_groups = [ {platform="qq", item_id="1", rule_type="group"}, ... ] },
{ expression_groups = [ ... ] },
]
将旧版 expression.expression_groups 转成当前结构。
"""
eg = _as_list(expr.get("expression_groups"))
if eg is None:
expression_groups = _as_list(expr.get("expression_groups"))
if expression_groups is None:
return False
if expression_groups and all(isinstance(item, dict) for item in expression_groups):
return False
# 已经是新格式（列表里是 dict 且包含 expression_groups），跳过
if eg and all(isinstance(i, dict) for i in eg):
return False
migrated: list[dict[str, Any]] = []
for group in eg:
g = _as_list(group)
if g is None:
migrated_groups: list[dict[str, Any]] = []
for group in expression_groups:
group_items = _as_list(group)
if group_items is None:
return False
targets: list[dict[str, str]] = []
for item in g:
for item in group_items:
parsed = _parse_triplet_target(str(item))
if parsed is None:
return False
targets.append(parsed)
migrated.append({"expression_groups": targets})
expr["expression_groups"] = migrated
migrated_groups.append({"expression_groups": targets})
expr["expression_groups"] = migrated_groups
return True
def _migrate_target_item_list(parent: dict[str, Any], key: str) -> bool:
"""
将 list[str] 的 "platform:id:type" 迁移为 list[{platform,item_id,rule_type}]
用于：memory.global_memory_blacklist 等。
将 list[str] 的 "platform:id:type" 迁移为 list[TargetItem]。
"""
raw = _as_list(parent.get(key))
if raw is None:
if raw is None or not raw:
return False
if raw and all(isinstance(i, dict) for i in raw):
if all(isinstance(item, dict) for item in raw):
return False
targets: list[dict[str, str]] = []
for item in raw:
parsed = _parse_triplet_target(str(item))
if parsed is None:
return False
targets.append(parsed)
parent[key] = targets
return True
def _migrate_extra_prompt_list(exp: dict[str, Any], key: str) -> bool:
"""
将 list[str] 的 "platform:id:type:prompt" 迁移为 list[{platform,item_id,rule_type,prompt}]
用于：experimental.chat_prompts
"""
raw = _as_list(exp.get(key))
if raw is None:
return False
if raw and all(isinstance(i, dict) for i in raw):
return False
items: list[dict[str, str]] = []
for item in raw:
parsed = _parse_quad_prompt(str(item))
if parsed is None:
return False
items.append(parsed)
exp[key] = items
return True
def _parse_multimodal_replyer(v: Any) -> Optional[bool]:
"""兼容旧 replyer_generator_type 到布尔开关的迁移。"""
if isinstance(v, bool):
return v
if not isinstance(v, str):
return None
normalized_value = v.strip().lower()
if normalized_value == "multimodal":
return True
if normalized_value == "legacy":
return False
return None
def migrate_legacy_bind_env_to_bot_config_dict(data: dict[str, Any]) -> MigrationResult:
"""将旧版环境变量中的绑定地址迁移到主配置结构。"""
"""将旧版 `.env` 中的绑定地址迁移到主配置结构。"""
migrated_any = False
reasons: list[str] = []
@@ -339,8 +245,7 @@ def migrate_legacy_bind_env_to_bot_config_dict(data: dict[str, Any]) -> Migratio
def try_migrate_legacy_bot_config_dict(data: dict[str, Any]) -> MigrationResult:
"""
尝试对“总配置 bot_config.toml”的 dict（已 unwrap）进行旧格式修复。
仅做我们明确知道的结构性变更；其它字段不动。
尝试修复 `bot_config.toml` 的少量旧结构，仅保留当前仍需要的兼容逻辑。
"""
if not ENABLE_LEGACY_MIGRATION:
return MigrationResult(data=data, migrated=False, reason="disabled")
@@ -353,41 +258,30 @@ def try_migrate_legacy_bot_config_dict(data: dict[str, Any]) -> MigrationResult:
if _migrate_expression_learning_list(expr):
migrated_any = True
reasons.append("expression.learning_list")
if _migrate_expression_groups(expr):
migrated_any = True
reasons.append("expression.expression_groups")
# allow_reflect: 旧 list[str] -> 新 list[TargetItem]
if _migrate_target_item_list(expr, "allow_reflect"):
migrated_any = True
reasons.append("expression.allow_reflect")
# manual_reflect_operator_id: 旧 str -> 新 Optional[TargetItem]
mroi = expr.get("manual_reflect_operator_id")
if isinstance(mroi, str) and mroi.strip():
parsed = _parse_triplet_target(mroi.strip())
manual_reflect_operator_id = expr.get("manual_reflect_operator_id")
if isinstance(manual_reflect_operator_id, str) and manual_reflect_operator_id.strip():
parsed = _parse_triplet_target(manual_reflect_operator_id.strip())
if parsed is not None:
expr["manual_reflect_operator_id"] = parsed
migrated_any = True
reasons.append("expression.manual_reflect_operator_id")
chat = _as_dict(data.get("chat"))
if chat is None:
chat = {}
data["chat"] = chat
elif "private_plan_style" in chat:
chat.pop("private_plan_style", None)
migrated_any = True
reasons.append("chat.private_plan_style_removed")
if isinstance(manual_reflect_operator_id, str) and not manual_reflect_operator_id.strip():
expr.pop("manual_reflect_operator_id", None)
migrated_any = True
reasons.append("expression.manual_reflect_operator_id_empty")
personality = _as_dict(data.get("personality"))
visual = _as_dict(data.get("visual"))
if visual is None and (
(personality is not None and "visual_style" in personality)
or "multimodal_planner" in chat
or "replyer_generator_type" in chat
):
visual = {}
data["visual"] = visual
if visual is not None and personality is not None and "visual_style" in personality:
if "visual_style" not in visual:
visual["visual_style"] = personality["visual_style"]
@@ -395,108 +289,19 @@ def try_migrate_legacy_bot_config_dict(data: dict[str, Any]) -> MigrationResult:
migrated_any = True
reasons.append("personality.visual_style_moved_to_visual.visual_style")
if visual is not None and "multimodal_planner" in chat:
if "multimodal_planner" not in visual and isinstance(chat["multimodal_planner"], bool):
visual["multimodal_planner"] = chat["multimodal_planner"]
if "multimodal_planner" in visual:
chat.pop("multimodal_planner", None)
if visual is not None and "multimodal_planner" in visual and "planner_mode" not in visual:
multimodal_planner = visual.pop("multimodal_planner")
if isinstance(multimodal_planner, bool):
visual["planner_mode"] = "multimodal" if multimodal_planner else "text"
migrated_any = True
reasons.append("chat.multimodal_planner_moved_to_visual.multimodal_planner")
reasons.append("visual.multimodal_planner_moved_to_visual.planner_mode")
else:
visual["multimodal_planner"] = multimodal_planner
if visual is not None and "replyer_generator_type" in chat:
multimodal_replyer = _parse_multimodal_replyer(chat["replyer_generator_type"])
if "multimodal_replyer" not in visual and multimodal_replyer is not None:
visual["multimodal_replyer"] = multimodal_replyer
if "multimodal_replyer" in visual:
chat.pop("replyer_generator_type", None)
migrated_any = True
reasons.append("chat.replyer_generator_type_moved_to_visual.multimodal_replyer")
maisaka = _as_dict(data.get("maisaka"))
mem = _as_dict(data.get("memory"))
debug = _as_dict(data.get("debug"))
if maisaka is not None:
moved_memory_keys = ("enable_memory_query_tool", "memory_query_default_limit")
if any(key in maisaka for key in moved_memory_keys) and mem is None:
mem = {}
data["memory"] = mem
if mem is not None:
for moved_key in moved_memory_keys:
if _move_section_key(maisaka, mem, moved_key):
migrated_any = True
reasons.append(f"maisaka.{moved_key}_moved_to_memory")
if mem is not None and "show_memory_prompt" in mem and debug is None:
debug = {}
data["debug"] = debug
if mem is not None:
if _migrate_target_item_list(mem, "global_memory_blacklist"):
migrated_any = True
reasons.append("memory.global_memory_blacklist")
if debug is not None and _move_section_key(mem, debug, "show_memory_prompt"):
migrated_any = True
reasons.append("memory.show_memory_prompt_moved_to_debug")
for removed_key in (
"agent_timeout_seconds",
"max_agent_iterations",
):
if removed_key in mem:
mem.pop(removed_key, None)
migrated_any = True
reasons.append(f"memory.{removed_key}_removed")
relationship = _as_dict(data.get("relationship"))
if relationship is not None:
data.pop("relationship", None)
memory = _as_dict(data.get("memory"))
if memory is not None and _migrate_target_item_list(memory, "global_memory_blacklist"):
migrated_any = True
reasons.append("relationship_removed")
exp = _as_dict(data.get("experimental"))
if exp is not None:
if _migrate_extra_prompt_list(exp, "chat_prompts"):
migrated_any = True
reasons.append("experimental.chat_prompts")
if "private_plan_style" in exp:
exp.pop("private_plan_style", None)
migrated_any = True
reasons.append("experimental.private_plan_style_removed")
for key in ("group_chat_prompt", "private_chat_prompts", "chat_prompts"):
if key in exp and key not in chat:
chat[key] = exp[key]
migrated_any = True
reasons.append(f"experimental.{key}_moved_to_chat")
data.pop("experimental", None)
migrated_any = True
reasons.append("experimental_removed")
if chat is not None and "think_mode" in chat:
chat.pop("think_mode", None)
migrated_any = True
reasons.append("chat.think_mode_removed")
tool = _as_dict(data.get("tool"))
if tool is not None:
data.pop("tool", None)
migrated_any = True
reasons.append("tool_section_removed")
# ExpressionConfig 中的 manual_reflect_operator_id:
# 旧版本可能是 ""（字符串），新版本期望 Optional[TargetItem]。
# 空字符串视为未配置，转换为 None/删除键以避免校验错误。
expr = _as_dict(data.get("expression"))
if expr is not None:
mroi = expr.get("manual_reflect_operator_id")
if isinstance(mroi, str) and not mroi.strip():
expr.pop("manual_reflect_operator_id", None)
migrated_any = True
reasons.append("expression.manual_reflect_operator_id_empty")
reasons.append("memory.global_memory_blacklist")
reason = ",".join(reasons)
return MigrationResult(data=data, migrated=migrated_any, reason=reason)

11
src/config/model_configs.py
View File

@@ -307,6 +307,15 @@ class ModelInfo(ConfigBase):
)
"""强制流式输出模式 (若模型不支持非流式输出, 请设置为true启用强制流式输出, 默认值为false)"""
visual: bool = Field(
default=False,
json_schema_extra={
"x-widget": "switch",
"x-icon": "image",
},
)
"""是否为多模态模型。开启后表示该模型支持视觉输入。"""
extra_params: dict[str, Any] = Field(
default_factory=dict,
json_schema_extra={
@@ -437,4 +446,4 @@ class ModelTaskConfig(ConfigBase):
"x-icon": "database",
},
)
"""嵌入模型配置"""
"""嵌入模型配置"""

85
src/config/official_configs.py
View File

@@ -145,23 +145,23 @@ class VisualConfig(ConfigBase):
__ui_label__ = "视觉"
__ui_icon__ = "image"
multimodal_planner: bool = Field(
default=True,
planner_mode: Literal["text", "multimodal", "auto"] = Field(
default="auto",
json_schema_extra={
"x-widget": "switch",
"x-icon": "image",
},
)
"""是否直接输入图片"""
multimodal_replyer: bool = Field(
default=False,
json_schema_extra={
"x-widget": "switch",
"x-widget": "select",
"x-icon": "git-branch",
},
)
"""是否启用 Maisaka 多模态 replyer 生成器"""
"""规划器模式，auto根据模型信息自动选择，text为纯文本模式，multimodal为多模态模式"""
replyer_mode: Literal["text", "multimodal", "auto"] = Field(
default="auto",
json_schema_extra={
"x-widget": "select",
"x-icon": "git-branch",
},
)
"""回复器模式，auto根据模型信息自动选择，text为纯文本模式，multimodal为多模态模式"""
visual_style: str = Field(
default="请用中文描述这张图片的内容。如果有文字，请把文字描述概括出来，请留意其主题，直观感受，输出为一段平文本，最多30字，请注意不要分点，就输出一段文本",
@@ -239,16 +239,12 @@ class ChatConfig(ConfigBase):
)
"""Planner 连续被新消息打断的最大次数，0 表示不启用打断"""
plan_reply_log_max_per_chat: int = Field(
default=1024,
json_schema_extra={
"x-widget": "input",
"x-icon": "file-text",
},
)
"""每个聊天流最大保存的Plan/Reply日志数量，超过此数量时会自动删除最老的日志"""
group_chat_prompt: str = Field(
default="你需要控制自己发言的频率，如果是一对一聊天，可以以较均匀的频率发言；如果用户较多，不要每句都回复，控制回复频率，不要回复的太频繁！控制回复的频率，不要每个人的消息都回复。",
default="""
你正在qq群里聊天，下面是群里正在聊的内容，其中包含聊天记录和聊天中的图片。
回复尽量简短一些。最好一次对一个话题进行回复，免得啰嗦或者回复内容太乱。请注意把握聊天内容。
不要回复的太频繁！控制回复的频率，不要每个人的消息都回复，只回复你感兴趣的或者主动提及你的。
""",
json_schema_extra={
"x-widget": "textarea",
"x-icon": "users",
@@ -257,7 +253,11 @@ class ChatConfig(ConfigBase):
"""_wrap_群聊通用注意事项"""
private_chat_prompts: str = Field(
default="你需要控制自己发言的频率，可以以较均匀的频率发言。",
default="""
你正在聊天，下面是正在聊的内容，其中包含聊天记录和聊天中的图片。
回复尽量简短一些。请注意把握聊天内容。
请考虑对方的发言频率，想法，思考自己何时回复以及回复内容。
""",
json_schema_extra={
"x-widget": "textarea",
"x-icon": "user",
@@ -740,6 +740,15 @@ class LearningItem(ConfigBase):
)
"""是否启用jargon学习"""
advanced_chosen: bool = Field(
default=False,
json_schema_extra={
"x-widget": "switch",
"x-icon": "sparkles",
},
)
"""是否启用基于子代理的二次表达方式选择"""
class ExpressionGroup(ConfigBase):
"""表达互通组配置类，若列表为空代表全局共享"""
@@ -769,6 +778,7 @@ class ExpressionConfig(ConfigBase):
use_expression=True,
enable_learning=True,
enable_jargon_learning=True,
advanced_chosen=False,
)
],
json_schema_extra={
@@ -1640,35 +1650,6 @@ class MaiSakaConfig(ConfigBase):
)
"""MaiSaka 使用的用户名称"""
tool_filter_task_name: str = Field(
default="utils",
json_schema_extra={
"x-widget": "input",
"x-icon": "sparkles",
},
)
"""工具筛选预判使用的模型任务名"""
tool_filter_threshold: int = Field(
default=20,
ge=1,
json_schema_extra={
"x-widget": "input",
"x-icon": "filter",
},
)
"""当可用工具总数超过该阈值时，先进行一轮工具筛选"""
tool_filter_max_keep: int = Field(
default=5,
ge=1,
json_schema_extra={
"x-widget": "input",
"x-icon": "list-filter",
},
)
"""工具筛选阶段最多保留的非内置工具数量"""
show_image_path: bool = Field(
default=True,
json_schema_extra={