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

feat: 在插件系统中添加 Chatter 组件

Browse Source 
- 在 component_types.py 中新增了用于 CHATTER 的 ComponentType。
- 实现了 ChatterInfo 类，用于存储 Chatter 组件的相关信息。
- 增强了 ComponentRegistry，以支持 Chatter 组件的注册与管理。
- 创建了 ChatterManager，用于管理 Chatter 实例并处理聊天流。
- 开发了 BaseChatter 抽象类，用于定义 Chatter 的行为规范。
- 实现了 AffinityChatter，作为具备兴趣评分与关系构建功能的具体 Chatter 组件。
- 添加了一个内置的 Chatter 插件，并附带完整文档与使用示例。
- 更新了 PluginManager，在插件概览中加入 Chatter 组件的统计信息。
This commit is contained in:
Windpicker-owo
2025-09-23 00:17:32 +08:00
parent e382b2ffd9
commit e1683ee9e6
15 changed files with 806 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/chat/affinity_flow/__init__.py
View File

@@ -3,6 +3,8 @@
提供全局的AFC管理器实例
"""
from src.chat.affinity_flow.afc_manager import afc_manager
# Avoid importing submodules at package import time to prevent circular imports.
# Consumers should import specific submodules directly, for example:
# from src.chat.affinity_flow.afc_manager import afc_manager
__all__ = ["afc_manager", "AFCManager", "AffinityFlowChatter"]

5
src/chat/affinity_flow/chatter.py
View File

@@ -11,6 +11,7 @@ from typing import Dict
from src.chat.planner_actions.action_manager import ActionManager
from src.chat.planner_actions.planner import ActionPlanner
from src.common.data_models.message_manager_data_model import StreamContext
from src.plugin_system.base.base_chatter import BaseChatter
from src.plugin_system.base.component_types import ChatMode
from src.common.logger import get_logger
@@ -18,7 +19,7 @@ from src.common.logger import get_logger
logger = get_logger("affinity_chatter")
class AffinityFlowChatter:
class AffinityFlowChatter(BaseChatter):
"""单个亲和力聊天处理器"""
def __init__(self, stream_id: str, planner: ActionPlanner, action_manager: ActionManager):
@@ -44,7 +45,7 @@ class AffinityFlowChatter:
}
self.last_activity_time = time.time()
async def process_stream_context(self, context: StreamContext) -> Dict[str, any]:
async def execute(self, context: StreamContext) -> dict:
"""
处理StreamContext对象

136
src/chat/chatter_manager.py Normal file
View File

@@ -0,0 +1,136 @@
from typing import Dict, List, Optional, Any
import time
from src.plugin_system.base.base_chatter import BaseChatter
from src.common.data_models.message_manager_data_model import StreamContext
from src.chat.planner_actions.planner import ActionPlanner
from src.chat.planner_actions.action_manager import ActionManager
from src.plugin_system.base.component_types import ChatType, ComponentType
from src.common.logger import get_logger
logger = get_logger("chatter_manager")
class ChatterManager:
def __init__(self, action_manager: ActionManager):
self.action_manager = action_manager
self.chatter_classes: Dict[ChatType, List[type]] = {}
self.instances: Dict[str, BaseChatter] = {}
# 管理器统计
self.stats = {
"chatters_registered": 0,
"streams_processed": 0,
"successful_executions": 0,
"failed_executions": 0,
}
def _auto_register_from_component_registry(self):
"""从组件注册表自动注册已注册的chatter组件"""
try:
from src.plugin_system.core.component_registry import component_registry
# 获取所有CHATTER类型的组件
chatter_components = component_registry.get_enabled_chatter_registry()
for chatter_name, chatter_class in chatter_components.items():
self.register_chatter(chatter_class)
logger.info(f"自动注册chatter组件: {chatter_name}")
except Exception as e:
logger.warning(f"自动注册chatter组件时发生错误: {e}")
def register_chatter(self, chatter_class: type):
"""注册聊天处理器类"""
for chat_type in chatter_class.chat_types:
if chat_type not in self.chatter_classes:
self.chatter_classes[chat_type] = []
self.chatter_classes[chat_type].append(chatter_class)
logger.info(f"注册聊天处理器 {chatter_class.__name__} 支持 {chat_type.value} 聊天类型")
self.stats["chatters_registered"] += 1
def get_chatter_class(self, chat_type: ChatType) -> Optional[type]:
"""获取指定聊天类型的聊天处理器类"""
if chat_type in self.chatter_classes:
return self.chatter_classes[chat_type][0]
return None
def get_supported_chat_types(self) -> List[ChatType]:
"""获取支持的聊天类型列表"""
return list(self.chatter_classes.keys())
def get_registered_chatters(self) -> Dict[ChatType, List[type]]:
"""获取已注册的聊天处理器"""
return self.chatter_classes.copy()
def get_stream_instance(self, stream_id: str) -> Optional[BaseChatter]:
"""获取指定流的聊天处理器实例"""
return self.instances.get(stream_id)
def cleanup_inactive_instances(self, max_inactive_minutes: int = 60):
"""清理不活跃的实例"""
current_time = time.time()
max_inactive_seconds = max_inactive_minutes * 60
inactive_streams = []
for stream_id, instance in self.instances.items():
if hasattr(instance, 'get_activity_time'):
activity_time = instance.get_activity_time()
if (current_time - activity_time) > max_inactive_seconds:
inactive_streams.append(stream_id)
for stream_id in inactive_streams:
del self.instances[stream_id]
logger.info(f"清理不活跃聊天流实例: {stream_id}")
async def process_stream_context(self, stream_id: str, context: StreamContext) -> dict:
"""处理流上下文"""
chat_type = context.chat_type
logger.debug(f"处理流 {stream_id}，聊天类型: {chat_type.value}")
if not self.chatter_classes:
self._auto_register_from_component_registry()
# 获取适合该聊天类型的chatter
chatter_class = self.get_chatter_class(chat_type)
if not chatter_class:
# 如果没有找到精确匹配，尝试查找支持ALL类型的chatter
from src.plugin_system.base.component_types import ChatType
all_chatter_class = self.get_chatter_class(ChatType.ALL)
if all_chatter_class:
chatter_class = all_chatter_class
logger.info(f"流 {stream_id} 使用通用chatter (类型: {chat_type.value})")
else:
raise ValueError(f"No chatter registered for chat type {chat_type}")
if stream_id not in self.instances:
planner = ActionPlanner(stream_id, self.action_manager)
self.instances[stream_id] = chatter_class(stream_id=stream_id, planner=planner, action_manager=self.action_manager)
logger.info(f"创建新的聊天流实例: {stream_id} 使用 {chatter_class.__name__} (类型: {chat_type.value})")
self.stats["streams_processed"] += 1
try:
result = await self.instances[stream_id].execute(context)
self.stats["successful_executions"] += 1
# 记录处理结果
success = result.get("success", False)
actions_count = result.get("actions_count", 0)
logger.debug(f"流 {stream_id} 处理完成: 成功={success}, 动作数={actions_count}")
return result
except Exception as e:
self.stats["failed_executions"] += 1
logger.error(f"处理流 {stream_id} 时发生错误: {e}")
raise
def get_stats(self) -> Dict[str, Any]:
"""获取管理器统计信息"""
stats = self.stats.copy()
stats["active_instances"] = len(self.instances)
stats["registered_chatter_types"] = len(self.chatter_classes)
return stats
def reset_stats(self):
"""重置统计信息"""
self.stats = {
"chatters_registered": 0,
"streams_processed": 0,
"successful_executions": 0,
"failed_executions": 0,
}

19
src/chat/message_manager/message_manager.py
View File

@@ -11,7 +11,8 @@ from typing import Dict, Optional, Any, TYPE_CHECKING
from src.common.logger import get_logger
from src.common.data_models.database_data_model import DatabaseMessages
from src.common.data_models.message_manager_data_model import StreamContext, MessageManagerStats, StreamStats
from src.chat.affinity_flow.afc_manager import afc_manager
from src.chat.chatter_manager import ChatterManager
from src.chat.planner_actions.action_manager import ActionManager
if TYPE_CHECKING:
from src.common.data_models.message_manager_data_model import StreamContext
@@ -31,6 +32,10 @@ class MessageManager:
# 统计信息
self.stats = MessageManagerStats()
# 初始化chatter manager
self.action_manager = ActionManager()
self.chatter_manager = ChatterManager(self.action_manager)
async def start(self):
"""启动消息管理器"""
if self.is_running:
@@ -125,15 +130,23 @@ class MessageManager:
# 直接使用StreamContext对象进行处理
if unread_messages:
try:
# 发送到AFC处理器，传递StreamContext对象
results = await afc_manager.process_stream_context(stream_id, context)
# 记录当前chat type用于调试
logger.debug(f"聊天流 {stream_id} 检测到的chat type: {context.chat_type.value}")
# 发送到chatter manager，传递StreamContext对象
results = await self.chatter_manager.process_stream_context(stream_id, context)
# 处理结果，标记消息为已读
if results.get("success", False):
self._clear_all_unread_messages(context)
logger.debug(f"聊天流 {stream_id} 处理成功，清除了 {len(unread_messages)} 条未读消息")
else:
logger.warning(f"聊天流 {stream_id} 处理失败: {results.get('error_message', '未知错误')}")
except Exception as e:
logger.error(f"处理聊天流 {stream_id} 时发生异常，将清除所有未读消息: {e}")
# 出现异常时也清除未读消息，避免重复处理
self._clear_all_unread_messages(context)
raise
logger.debug(f"聊天流 {stream_id} 消息处理完成")

22
src/chat/planner_actions/planner.py
View File

@@ -4,19 +4,23 @@
"""
from dataclasses import asdict
from typing import Dict, List, Optional, Tuple
from typing import TYPE_CHECKING, Dict, List, Optional, Tuple
from src.chat.planner_actions.action_manager import ActionManager
from src.plugin_system.base.component_types import ChatMode
from src.chat.planner_actions.plan_executor import PlanExecutor
from src.chat.planner_actions.plan_filter import PlanFilter
from src.chat.planner_actions.plan_generator import PlanGenerator
from src.chat.affinity_flow.interest_scoring import InterestScoringSystem
from src.chat.affinity_flow.relationship_tracker import UserRelationshipTracker
from src.common.data_models.info_data_model import Plan
from src.common.data_models.message_manager_data_model import StreamContext
from src.common.logger import get_logger
from src.config.config import global_config
from src.plugin_system.base.component_types import ChatMode
if TYPE_CHECKING:
from src.chat.planner_actions.action_manager import ActionManager
from src.common.data_models.message_manager_data_model import StreamContext
from src.common.data_models.info_data_model import Plan
# 导入提示词模块以确保其被初始化
from src.chat.planner_actions import planner_prompts # noqa
@@ -35,7 +39,7 @@ class ActionPlanner:
4. 完整的规划流程：生成→筛选→执行的完整三阶段流程
"""
def __init__(self, chat_id: str, action_manager: ActionManager):
def __init__(self, chat_id: str, action_manager: "ActionManager"):
"""
初始化增强版ActionPlanner。
@@ -85,7 +89,7 @@ class ActionPlanner:
}
async def plan(
self, mode: ChatMode = ChatMode.FOCUS, context: StreamContext = None
self, mode: ChatMode = ChatMode.FOCUS, context: "StreamContext" = None
) -> Tuple[List[Dict], Optional[Dict]]:
"""
执行完整的增强版规划流程。
@@ -109,7 +113,7 @@ class ActionPlanner:
self.planner_stats["failed_plans"] += 1
return [], None
async def _enhanced_plan_flow(self, mode: ChatMode, context: StreamContext) -> Tuple[List[Dict], Optional[Dict]]:
async def _enhanced_plan_flow(self, mode: ChatMode, context: "StreamContext") -> Tuple[List[Dict], Optional[Dict]]:
"""执行增强版规划流程"""
try:
# 1. 生成初始 Plan
@@ -204,7 +208,7 @@ class ActionPlanner:
self.planner_stats["replies_generated"] += reply_count
self.planner_stats["other_actions_executed"] += other_count
def _build_return_result(self, plan: Plan) -> Tuple[List[Dict], Optional[Dict]]:
def _build_return_result(self, plan: "Plan") -> Tuple[List[Dict], Optional[Dict]]:
"""构建返回结果"""
final_actions = plan.decided_actions or []
final_target_message = next((act.action_message for act in final_actions if act.action_message), None)