gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'dev' of https://github.com/MoFox-Studio/MoFox_Bot into dev

Browse Source
This commit is contained in:
BuildTools
2025-09-12 21:39:11 +08:00
parent 675bd4b75e fc5980a282
commit 34d0309d1f
11 changed files with 424 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/chat/knowledge/knowledge_lib.py
View File

@@ -30,6 +30,11 @@ DATA_PATH = os.path.join(ROOT_PATH, "data")
qa_manager = None
inspire_manager = None
def initialize_lpmm_knowledge():
"""初始化LPMM知识库"""
global qa_manager, inspire_manager
# 检查LPMM知识库是否启用
if global_config.lpmm_knowledge.enable:
logger.info("正在初始化Mai-LPMM")

6
src/chat/utils/prompt.py
View File

@@ -71,6 +71,7 @@ class PromptParameters:
identity_block: str = ""
schedule_block: str = ""
moderation_prompt_block: str = ""
safety_guidelines_block: str = ""
reply_target_block: str = ""
mood_prompt: str = ""
action_descriptions: str = ""
@@ -641,7 +642,7 @@ class Prompt:
)
# 创建工具执行器
tool_executor = ToolExecutor()
tool_executor = ToolExecutor(chat_id=self.parameters.chat_id)
# 执行工具获取信息
tool_results, _, _ = await tool_executor.execute_from_chat_message(
@@ -768,6 +769,7 @@ class Prompt:
"reply_style": global_config.personality.reply_style,
"keywords_reaction_prompt": self.parameters.keywords_reaction_prompt or context_data.get("keywords_reaction_prompt", ""),
"moderation_prompt": self.parameters.moderation_prompt_block or context_data.get("moderation_prompt", ""),
"safety_guidelines_block": self.parameters.safety_guidelines_block or context_data.get("safety_guidelines_block", ""),
}
def _prepare_normal_params(self, context_data: Dict[str, Any]) -> Dict[str, Any]:
@@ -791,6 +793,7 @@ class Prompt:
"mood_state": self.parameters.mood_prompt or context_data.get("mood_state", ""),
"keywords_reaction_prompt": self.parameters.keywords_reaction_prompt or context_data.get("keywords_reaction_prompt", ""),
"moderation_prompt": self.parameters.moderation_prompt_block or context_data.get("moderation_prompt", ""),
"safety_guidelines_block": self.parameters.safety_guidelines_block or context_data.get("safety_guidelines_block", ""),
}
def _prepare_default_params(self, context_data: Dict[str, Any]) -> Dict[str, Any]:
@@ -810,6 +813,7 @@ class Prompt:
"reply_style": global_config.personality.reply_style,
"keywords_reaction_prompt": self.parameters.keywords_reaction_prompt or context_data.get("keywords_reaction_prompt", ""),
"moderation_prompt": self.parameters.moderation_prompt_block or context_data.get("moderation_prompt", ""),
"safety_guidelines_block": self.parameters.safety_guidelines_block or context_data.get("safety_guidelines_block", ""),
}
def format(self, *args, **kwargs) -> str:

53
src/common/data_models/__init__.py Normal file
View File

@@ -0,0 +1,53 @@
import copy
from typing import Any
class BaseDataModel:
def deepcopy(self):
return copy.deepcopy(self)
def temporarily_transform_class_to_dict(obj: Any) -> Any:
# sourcery skip: assign-if-exp, reintroduce-else
"""
将对象或容器中的 BaseDataModel 子类（类对象）或 BaseDataModel 实例
递归转换为普通 dict，不修改原对象。
- 对于类对象（isinstance(value, type) 且 issubclass(..., BaseDataModel)），
读取类的 __dict__ 中非 dunder 项并递归转换。
- 对于实例（isinstance(value, BaseDataModel)），读取 vars(instance) 并递归转换。
"""
def _transform(value: Any) -> Any:
# 值是类对象且为 BaseDataModel 的子类
if isinstance(value, type) and issubclass(value, BaseDataModel):
return {k: _transform(v) for k, v in value.__dict__.items() if not k.startswith("__") and not callable(v)}
# 值是 BaseDataModel 的实例
if isinstance(value, BaseDataModel):
return {k: _transform(v) for k, v in vars(value).items()}
# 常见容器类型，递归处理
if isinstance(value, dict):
return {k: _transform(v) for k, v in value.items()}
if isinstance(value, list):
return [_transform(v) for v in value]
if isinstance(value, tuple):
return tuple(_transform(v) for v in value)
if isinstance(value, set):
return {_transform(v) for v in value}
# 基本类型，直接返回
return value
result = _transform(obj)
def flatten(target_dict: dict):
flat_dict = {}
for k, v in target_dict.items():
if isinstance(v, dict):
# 递归扁平化子字典
sub_flat = flatten(v)
flat_dict.update(sub_flat)
else:
flat_dict[k] = v
return flat_dict
return flatten(result) if isinstance(result, dict) else result

235
src/common/data_models/database_data_model.py Normal file
View File

@@ -0,0 +1,235 @@
import json
from typing import Optional, Any, Dict
from dataclasses import dataclass, field
from . import BaseDataModel
@dataclass
class DatabaseUserInfo(BaseDataModel):
platform: str = field(default_factory=str)
user_id: str = field(default_factory=str)
user_nickname: str = field(default_factory=str)
user_cardname: Optional[str] = None
# def __post_init__(self):
# assert isinstance(self.platform, str), "platform must be a string"
# assert isinstance(self.user_id, str), "user_id must be a string"
# assert isinstance(self.user_nickname, str), "user_nickname must be a string"
# assert isinstance(self.user_cardname, str) or self.user_cardname is None, (
# "user_cardname must be a string or None"
# )
@dataclass
class DatabaseGroupInfo(BaseDataModel):
group_id: str = field(default_factory=str)
group_name: str = field(default_factory=str)
group_platform: Optional[str] = None
# def __post_init__(self):
# assert isinstance(self.group_id, str), "group_id must be a string"
# assert isinstance(self.group_name, str), "group_name must be a string"
# assert isinstance(self.group_platform, str) or self.group_platform is None, (
# "group_platform must be a string or None"
# )
@dataclass
class DatabaseChatInfo(BaseDataModel):
stream_id: str = field(default_factory=str)
platform: str = field(default_factory=str)
create_time: float = field(default_factory=float)
last_active_time: float = field(default_factory=float)
user_info: DatabaseUserInfo = field(default_factory=DatabaseUserInfo)
group_info: Optional[DatabaseGroupInfo] = None
# def __post_init__(self):
# assert isinstance(self.stream_id, str), "stream_id must be a string"
# assert isinstance(self.platform, str), "platform must be a string"
# assert isinstance(self.create_time, float), "create_time must be a float"
# assert isinstance(self.last_active_time, float), "last_active_time must be a float"
# assert isinstance(self.user_info, DatabaseUserInfo), "user_info must be a DatabaseUserInfo instance"
# assert isinstance(self.group_info, DatabaseGroupInfo) or self.group_info is None, (
# "group_info must be a DatabaseGroupInfo instance or None"
# )
@dataclass(init=False)
class DatabaseMessages(BaseDataModel):
def __init__(
self,
message_id: str = "",
time: float = 0.0,
chat_id: str = "",
reply_to: Optional[str] = None,
interest_value: Optional[float] = None,
key_words: Optional[str] = None,
key_words_lite: Optional[str] = None,
is_mentioned: Optional[bool] = None,
is_at: Optional[bool] = None,
reply_probability_boost: Optional[float] = None,
processed_plain_text: Optional[str] = None,
display_message: Optional[str] = None,
priority_mode: Optional[str] = None,
priority_info: Optional[str] = None,
additional_config: Optional[str] = None,
is_emoji: bool = False,
is_picid: bool = False,
is_command: bool = False,
is_notify: bool = False,
selected_expressions: Optional[str] = None,
user_id: str = "",
user_nickname: str = "",
user_cardname: Optional[str] = None,
user_platform: str = "",
chat_info_group_id: Optional[str] = None,
chat_info_group_name: Optional[str] = None,
chat_info_group_platform: Optional[str] = None,
chat_info_user_id: str = "",
chat_info_user_nickname: str = "",
chat_info_user_cardname: Optional[str] = None,
chat_info_user_platform: str = "",
chat_info_stream_id: str = "",
chat_info_platform: str = "",
chat_info_create_time: float = 0.0,
chat_info_last_active_time: float = 0.0,
**kwargs: Any,
):
self.message_id = message_id
self.time = time
self.chat_id = chat_id
self.reply_to = reply_to
self.interest_value = interest_value
self.key_words = key_words
self.key_words_lite = key_words_lite
self.is_mentioned = is_mentioned
self.is_at = is_at
self.reply_probability_boost = reply_probability_boost
self.processed_plain_text = processed_plain_text
self.display_message = display_message
self.priority_mode = priority_mode
self.priority_info = priority_info
self.additional_config = additional_config
self.is_emoji = is_emoji
self.is_picid = is_picid
self.is_command = is_command
self.is_notify = is_notify
self.selected_expressions = selected_expressions
self.group_info: Optional[DatabaseGroupInfo] = None
self.user_info = DatabaseUserInfo(
user_id=user_id,
user_nickname=user_nickname,
user_cardname=user_cardname,
platform=user_platform,
)
if chat_info_group_id and chat_info_group_name:
self.group_info = DatabaseGroupInfo(
group_id=chat_info_group_id,
group_name=chat_info_group_name,
group_platform=chat_info_group_platform,
)
self.chat_info = DatabaseChatInfo(
stream_id=chat_info_stream_id,
platform=chat_info_platform,
create_time=chat_info_create_time,
last_active_time=chat_info_last_active_time,
user_info=DatabaseUserInfo(
user_id=chat_info_user_id,
user_nickname=chat_info_user_nickname,
user_cardname=chat_info_user_cardname,
platform=chat_info_user_platform,
),
group_info=self.group_info,
)
if kwargs:
for key, value in kwargs.items():
setattr(self, key, value)
# def __post_init__(self):
# assert isinstance(self.message_id, str), "message_id must be a string"
# assert isinstance(self.time, float), "time must be a float"
# assert isinstance(self.chat_id, str), "chat_id must be a string"
# assert isinstance(self.reply_to, str) or self.reply_to is None, "reply_to must be a string or None"
# assert isinstance(self.interest_value, float) or self.interest_value is None, (
# "interest_value must be a float or None"
# )
def flatten(self) -> Dict[str, Any]:
"""
将消息数据模型转换为字典格式，便于存储或传输
"""
return {
"message_id": self.message_id,
"time": self.time,
"chat_id": self.chat_id,
"reply_to": self.reply_to,
"interest_value": self.interest_value,
"key_words": self.key_words,
"key_words_lite": self.key_words_lite,
"is_mentioned": self.is_mentioned,
"is_at": self.is_at,
"reply_probability_boost": self.reply_probability_boost,
"processed_plain_text": self.processed_plain_text,
"display_message": self.display_message,
"priority_mode": self.priority_mode,
"priority_info": self.priority_info,
"additional_config": self.additional_config,
"is_emoji": self.is_emoji,
"is_picid": self.is_picid,
"is_command": self.is_command,
"is_notify": self.is_notify,
"selected_expressions": self.selected_expressions,
"user_id": self.user_info.user_id,
"user_nickname": self.user_info.user_nickname,
"user_cardname": self.user_info.user_cardname,
"user_platform": self.user_info.platform,
"chat_info_group_id": self.group_info.group_id if self.group_info else None,
"chat_info_group_name": self.group_info.group_name if self.group_info else None,
"chat_info_group_platform": self.group_info.group_platform if self.group_info else None,
"chat_info_stream_id": self.chat_info.stream_id,
"chat_info_platform": self.chat_info.platform,
"chat_info_create_time": self.chat_info.create_time,
"chat_info_last_active_time": self.chat_info.last_active_time,
"chat_info_user_platform": self.chat_info.user_info.platform,
"chat_info_user_id": self.chat_info.user_info.user_id,
"chat_info_user_nickname": self.chat_info.user_info.user_nickname,
"chat_info_user_cardname": self.chat_info.user_info.user_cardname,
}
@dataclass(init=False)
class DatabaseActionRecords(BaseDataModel):
def __init__(
self,
action_id: str,
time: float,
action_name: str,
action_data: str,
action_done: bool,
action_build_into_prompt: bool,
action_prompt_display: str,
chat_id: str,
chat_info_stream_id: str,
chat_info_platform: str,
):
self.action_id = action_id
self.time = time
self.action_name = action_name
if isinstance(action_data, str):
self.action_data = json.loads(action_data)
else:
raise ValueError("action_data must be a JSON string")
self.action_done = action_done
self.action_build_into_prompt = action_build_into_prompt
self.action_prompt_display = action_prompt_display
self.chat_id = chat_id
self.chat_info_stream_id = chat_info_stream_id
self.chat_info_platform = chat_info_platform

25
src/common/data_models/info_data_model.py Normal file
View File

@@ -0,0 +1,25 @@
from dataclasses import dataclass, field
from typing import Optional, Dict, TYPE_CHECKING
from . import BaseDataModel
if TYPE_CHECKING:
from .database_data_model import DatabaseMessages
from src.plugin_system.base.component_types import ActionInfo
@dataclass
class TargetPersonInfo(BaseDataModel):
platform: str = field(default_factory=str)
user_id: str = field(default_factory=str)
user_nickname: str = field(default_factory=str)
person_id: Optional[str] = None
person_name: Optional[str] = None
@dataclass
class ActionPlannerInfo(BaseDataModel):
action_type: str = field(default_factory=str)
reasoning: Optional[str] = None
action_data: Optional[Dict] = None
action_message: Optional["DatabaseMessages"] = None
available_actions: Optional[Dict[str, "ActionInfo"]] = None

16
src/common/data_models/llm_data_model.py Normal file
View File

@@ -0,0 +1,16 @@
from dataclasses import dataclass
from typing import Optional, List, Tuple, TYPE_CHECKING, Any
from . import BaseDataModel
if TYPE_CHECKING:
from src.llm_models.payload_content.tool_option import ToolCall
@dataclass
class LLMGenerationDataModel(BaseDataModel):
content: Optional[str] = None
reasoning: Optional[str] = None
model: Optional[str] = None
tool_calls: Optional[List["ToolCall"]] = None
prompt: Optional[str] = None
selected_expressions: Optional[List[int]] = None
reply_set: Optional[List[Tuple[str, Any]]] = None

36
src/common/data_models/message_data_model.py Normal file
View File

@@ -0,0 +1,36 @@
from typing import Optional, TYPE_CHECKING
from dataclasses import dataclass, field
from . import BaseDataModel
if TYPE_CHECKING:
from .database_data_model import DatabaseMessages
@dataclass
class MessageAndActionModel(BaseDataModel):
chat_id: str = field(default_factory=str)
time: float = field(default_factory=float)
user_id: str = field(default_factory=str)
user_platform: str = field(default_factory=str)
user_nickname: str = field(default_factory=str)
user_cardname: Optional[str] = None
processed_plain_text: Optional[str] = None
display_message: Optional[str] = None
chat_info_platform: str = field(default_factory=str)
is_action_record: bool = field(default=False)
action_name: Optional[str] = None
@classmethod
def from_DatabaseMessages(cls, message: "DatabaseMessages"):
return cls(
chat_id=message.chat_id,
time=message.time,
user_id=message.user_info.user_id,
user_platform=message.user_info.platform,
user_nickname=message.user_info.user_nickname,
user_cardname=message.user_info.user_cardname,
processed_plain_text=message.processed_plain_text,
display_message=message.display_message,
chat_info_platform=message.chat_info.platform,
)

3
src/config/official_configs.py
View File

@@ -662,9 +662,6 @@ class SleepSystemConfig(ValidatedConfigBase):
)
max_sleep_delay_minutes: int = Field(default=60, description="单日最大延迟入睡分钟数")
enable_pre_sleep_notification: bool = Field(default=True, description="是否启用睡前消息")
pre_sleep_notification_groups: List[str] = Field(
default_factory=list, description='接收睡前消息的群号列表, 格式: ["platform:group_id1", "platform:group_id2"]'
)
pre_sleep_prompt: str = Field(
default="我准备睡觉了，请生成一句简短自然的晚安问候。", description="用于生成睡前消息的提示"
)

7
src/main.py
View File

@@ -116,9 +116,9 @@ class MainSystem:
# 停止消息重组器
from src.plugin_system.core.event_manager import event_manager
from src.plugin_system import EventType
import asyncio
asyncio.run(event_manager.trigger_event(EventType.ON_STOP,permission_group="SYSTEM"))
from src.utils.message_chunker import reassembler
import asyncio
loop = asyncio.get_event_loop()
if loop.is_running():
@@ -250,6 +250,11 @@ MoFox_Bot(第三方修改版)
self.hippocampus_manager.initialize()
logger.info("记忆系统初始化成功")
# 初始化LPMM知识库
from src.chat.knowledge.knowledge_lib import initialize_lpmm_knowledge
initialize_lpmm_knowledge()
logger.info("LPMM知识库初始化成功")
# 初始化异步记忆管理器
try:
from src.chat.memory_system.async_memory_optimizer import async_memory_manager

2
src/plugin_system/base/base_action.py
View File

@@ -8,7 +8,6 @@ from src.common.logger import get_logger
from src.chat.message_receive.chat_stream import ChatStream
from src.plugin_system.base.component_types import ActionActivationType, ChatMode, ActionInfo, ComponentType, ChatType
from src.plugin_system.apis import send_api, database_api, message_api
from src.plugin_system.core.component_registry import component_registry
logger = get_logger("base_action")
@@ -398,6 +397,7 @@ class BaseAction(ABC):
try:
# 1. 从注册中心获取Action类
from src.plugin_system.core.component_registry import component_registry
action_class = component_registry.get_component_class(action_name, ComponentType.ACTION)
if not action_class:
logger.error(f"{log_prefix} 未找到Action: {action_name}")

2
template/bot_config_template.toml
View File

@@ -484,8 +484,6 @@ max_sleep_delay_minutes = 60
# 是否在进入“准备入睡”状态时发送一条消息通知。
enable_pre_sleep_notification = false
# 接收睡前消息的群组列表。格式为: ["platform:group_id1", "platform:group_id2"]，例如 ["qq:12345678"]
pre_sleep_notification_groups = []
# 用于生成睡前消息的提示。AI会根据这个提示生成一句晚安问候。
pre_sleep_prompt = "我准备睡觉了，请生成一句简短自然的晚安问候。"
insomnia_duration_minutes = [30, 60] # 单次失眠状态的持续时间范围（分钟）