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

refactor(chat): 异步化聊天系统并重构兴趣值计算机制

Browse Source 
将同步调用改为异步调用以提升性能，重构兴趣值计算流程以支持更灵活的组件化架构。主要改进包括：

- 异步化ChatManager相关方法，避免阻塞主线程
- 重构兴趣值计算系统，从插件内部计算改为通过兴趣管理器统一处理
- 新增should_act字段支持更细粒度的动作决策
- 优化初始化逻辑，避免构造函数中的异步操作
- 扩展插件系统支持兴趣计算器组件注册
- 更新数据库模型以支持新的兴趣值相关字段

这些改进提升了系统的响应性能和可扩展性，同时保持了API的向后兼容性。
This commit is contained in:
Windpicker-owo
2025-10-05 01:25:52 +08:00
parent 80c3ee8524
commit 481252d660
38 changed files with 1493 additions and 262 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
src/chat/message_manager/context_manager.py
View File

@@ -55,8 +55,8 @@ class SingleStreamContextManager:
"""
try:
self.context.add_message(message)
# 推迟兴趣度计算到分发阶段
message.interest_value = getattr(message, "interest_value", None)
# 在上下文管理器中计算兴趣值
await self._calculate_message_interest(message)
self.total_messages += 1
self.last_access_time = time.time()
# 启动流的循环任务（如果还未启动）
@@ -228,51 +228,44 @@ class SingleStreamContextManager:
async def _calculate_message_interest(self, message: DatabaseMessages) -> float:
"""
异步计算消息的兴趣度。
此方法通过检查当前是否存在正在运行的 asyncio 事件循环来兼容同步和异步调用。
在上下文管理器中计算消息的兴趣度
"""
# 内部异步函数，封装实际的计算逻辑
async def _get_score():
try:
from src.plugins.built_in.affinity_flow_chatter.interest_scoring import (
chatter_interest_scoring_system,
)
interest_score = await chatter_interest_scoring_system._calculate_single_message_score(
message=message, bot_nickname=global_config.bot.nickname
)
interest_value = interest_score.total_score
logger.debug(f"使用插件内部系统计算兴趣度: {interest_value:.3f}")
return interest_value
except ImportError as e:
logger.debug(f"兴趣度计算插件加载失败，可能未启用: {e}")
return 0.5
except Exception as e:
# 在某些情况下（例如机器人自己的消息），没有兴趣度是正常的
logger.info(f"插件内部兴趣度计算失败，使用默认值: {e}")
return 0.5
# 检查并获取当前事件循环
try:
loop = asyncio.get_running_loop()
except RuntimeError: # 'RuntimeError: There is no current event loop...'
loop = None
from src.chat.interest_system.interest_manager import get_interest_manager
if loop and loop.is_running():
# 如果事件循环正在运行，直接 await
return await _get_score()
else:
# 否则，使用 asyncio.run() 来安全执行
return asyncio.run(_get_score())
interest_manager = get_interest_manager()
if interest_manager.has_calculator():
# 使用兴趣值计算组件计算
result = await interest_manager.calculate_interest(message)
if result.success:
# 更新消息对象的兴趣值相关字段
message.interest_value = result.interest_value
message.should_reply = result.should_reply
message.should_act = result.should_act
logger.debug(f"消息 {message.message_id} 兴趣值已更新: {result.interest_value:.3f}, "
f"should_reply: {result.should_reply}, should_act: {result.should_act}")
return result.interest_value
else:
logger.warning(f"消息 {message.message_id} 兴趣值计算失败: {result.error_message}")
return 0.5
else:
logger.debug("未找到兴趣值计算器，使用默认兴趣值")
return 0.5
except Exception as e:
logger.error(f"计算消息兴趣度时发生错误: {e}", exc_info=True)
return 0.5
async def add_message_async(self, message: DatabaseMessages, skip_energy_update: bool = False) -> bool:
"""异步实现的 add_message：将消息添加到 context，并 await 能量更新与分发。"""
try:
self.context.add_message(message)
# 推迟兴趣度计算到分发阶段
message.interest_value = getattr(message, "interest_value", None)
# 在上下文管理器中计算兴趣值
await self._calculate_message_interest(message)
self.total_messages += 1
self.last_access_time = time.time()
@@ -280,7 +273,7 @@ class SingleStreamContextManager:
# 启动流的循环任务（如果还未启动）
asyncio.create_task(stream_loop_manager.start_stream_loop(self.stream_id))
logger.debug(f"添加消息到单流上下文(异步): {self.stream_id} (兴趣度待计算)")
logger.debug(f"添加消息到单流上下文(异步): {self.stream_id}")
return True
except Exception as e:
logger.error(f"添加消息到单流上下文失败 (async) {self.stream_id}: {e}", exc_info=True)

14
src/chat/message_manager/distribution_manager.py
View File

@@ -181,7 +181,7 @@ class StreamLoopManager:
async def _fallback_acquire_slot(self, stream_id: str, force: bool) -> bool:
"""回退方案：获取槽位（原始方法）"""
# 判断是否需要强制分发
should_force = force or self._should_force_dispatch_for_stream(stream_id)
should_force = force or await self._should_force_dispatch_for_stream(stream_id)
# 检查是否超过最大并发限制
current_streams = len(self.stream_loops)
@@ -410,7 +410,7 @@ class StreamLoopManager:
"""
try:
chat_manager = get_chat_manager()
chat_stream = chat_manager.get_stream(stream_id)
chat_stream = await chat_manager.get_stream(stream_id)
if chat_stream:
return chat_stream.context_manager.context
return None
@@ -538,13 +538,13 @@ class StreamLoopManager:
self.chatter_manager = chatter_manager
logger.info(f"设置chatter管理器: {chatter_manager.__class__.__name__}")
def _should_force_dispatch_for_stream(self, stream_id: str) -> bool:
async def _should_force_dispatch_for_stream(self, stream_id: str) -> bool:
if not self.force_dispatch_unread_threshold or self.force_dispatch_unread_threshold <= 0:
return False
try:
chat_manager = get_chat_manager()
chat_stream = chat_manager.get_stream(stream_id)
chat_stream = await chat_manager.get_stream(stream_id)
if not chat_stream:
return False
@@ -595,7 +595,7 @@ class StreamLoopManager:
"""分发完成后基于历史消息刷新能量值"""
try:
chat_manager = get_chat_manager()
chat_stream = chat_manager.get_stream(stream_id)
chat_stream = await chat_manager.get_stream(stream_id)
if not chat_stream:
logger.debug(f"刷新能量时未找到聊天流: {stream_id}")
return
@@ -622,7 +622,7 @@ class StreamLoopManager:
except Exception as e:
logger.error(f"等待流循环任务结束时出错: {stream_id} - {e}")
def _force_dispatch_stream(self, stream_id: str) -> None:
async def _force_dispatch_stream(self, stream_id: str) -> None:
"""强制分发流处理
当流的未读消息超过阈值时，强制触发分发处理
@@ -657,7 +657,7 @@ class StreamLoopManager:
# 获取聊天管理器和流
chat_manager = get_chat_manager()
chat_stream = chat_manager.get_stream(stream_id)
chat_stream = await chat_manager.get_stream(stream_id)
if not chat_stream:
logger.warning(f"强制分发时未找到流: {stream_id}")
return

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

@@ -132,7 +132,7 @@ class MessageManager:
"""添加消息到指定聊天流"""
try:
chat_manager = get_chat_manager()
chat_stream = chat_manager.get_stream(stream_id)
chat_stream = await chat_manager.get_stream(stream_id)
if not chat_stream:
logger.warning(f"MessageManager.add_message: 聊天流 {stream_id} 不存在")
return
@@ -153,7 +153,7 @@ class MessageManager:
"""更新消息信息"""
try:
chat_manager = get_chat_manager()
chat_stream = chat_manager.get_stream(stream_id)
chat_stream = await chat_manager.get_stream(stream_id)
if not chat_stream:
logger.warning(f"MessageManager.update_message: 聊天流 {stream_id} 不存在")
return
@@ -180,7 +180,7 @@ class MessageManager:
try:
chat_manager = get_chat_manager()
chat_stream = chat_manager.get_stream(stream_id)
chat_stream = await chat_manager.get_stream(stream_id)
if not chat_stream:
logger.warning(f"MessageManager.bulk_update_messages: 聊天流 {stream_id} 不存在")
return 0
@@ -211,7 +211,7 @@ class MessageManager:
"""添加动作到消息"""
try:
chat_manager = get_chat_manager()
chat_stream = chat_manager.get_stream(stream_id)
chat_stream = await chat_manager.get_stream(stream_id)
if not chat_stream:
logger.warning(f"MessageManager.add_action: 聊天流 {stream_id} 不存在")
return
@@ -223,12 +223,12 @@ class MessageManager:
except Exception as e:
logger.error(f"为消息 {message_id} 添加动作时发生错误: {e}")
def deactivate_stream(self, stream_id: str):
async def deactivate_stream(self, stream_id: str):
"""停用聊天流"""
try:
# 通过 ChatManager 获取 ChatStream
chat_manager = get_chat_manager()
chat_stream = chat_manager.get_stream(stream_id)
chat_stream = await chat_manager.get_stream(stream_id)
if not chat_stream:
logger.warning(f"停用流失败: 聊天流 {stream_id} 不存在")
return
@@ -245,12 +245,12 @@ class MessageManager:
except Exception as e:
logger.error(f"停用聊天流 {stream_id} 时发生错误: {e}")
def activate_stream(self, stream_id: str):
async def activate_stream(self, stream_id: str):
"""激活聊天流"""
try:
# 通过 ChatManager 获取 ChatStream
chat_manager = get_chat_manager()
chat_stream = chat_manager.get_stream(stream_id)
chat_stream = await chat_manager.get_stream(stream_id)
if not chat_stream:
logger.warning(f"激活流失败: 聊天流 {stream_id} 不存在")
return
@@ -262,12 +262,12 @@ class MessageManager:
except Exception as e:
logger.error(f"激活聊天流 {stream_id} 时发生错误: {e}")
def get_stream_stats(self, stream_id: str) -> StreamStats | None:
async def get_stream_stats(self, stream_id: str) -> StreamStats | None:
"""获取聊天流统计"""
try:
# 通过 ChatManager 获取 ChatStream
chat_manager = get_chat_manager()
chat_stream = chat_manager.get_stream(stream_id)
chat_stream = await chat_manager.get_stream(stream_id)
if not chat_stream:
return None
@@ -360,7 +360,7 @@ class MessageManager:
pass
# 增加打断计数并应用afc阈值降低
chat_stream.context_manager.context.increment_interruption_count()
await chat_stream.context_manager.context.increment_interruption_count()
chat_stream.context_manager.context.apply_interruption_afc_reduction(
global_config.chat.interruption_afc_reduction
)
@@ -382,7 +382,7 @@ class MessageManager:
try:
# 通过 ChatManager 获取 ChatStream
chat_manager = get_chat_manager()
chat_stream = chat_manager.get_stream(stream_id)
chat_stream = await chat_manager.get_stream(stream_id)
if not chat_stream:
logger.warning(f"清除消息失败: 聊天流 {stream_id} 不存在")
return
@@ -411,7 +411,7 @@ class MessageManager:
"""清除指定聊天流的所有未读消息"""
try:
chat_manager = get_chat_manager()
chat_stream = chat_manager.get_stream(stream_id)
chat_stream = await chat_manager.get_stream(stream_id)
if not chat_stream:
logger.warning(f"clear_stream_unread_messages: 聊天流 {stream_id} 不存在")
return