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

feat(chatter_manager): 清理 processing_message_id 以防止重复回复检测失效

Browse Source 
feat(distribution_manager): 添加子任务跟踪和取消逻辑，优化流处理
feat(default_generator): 优化 respond 和 reply 模式下的消息处理逻辑
feat(affinity_chatter): 处理取消异常时清理 processing_message_id
feat(planner): 确保在规划流程取消时清理 processing_message_id
This commit is contained in:
Windpicker-owo
2025-11-10 17:12:20 +08:00
parent 861cc18e7d
commit aab3f19f10
5 changed files with 124 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/chat/chatter_manager.py
View File

@@ -145,13 +145,17 @@ class ChatterManager:
return result
except asyncio.CancelledError:
self.stats["failed_executions"] += 1
logger.info(f"流 {stream_id} 处理被取消，不清空未读消息")
logger.info(f"流 {stream_id} 处理被取消")
context.triggering_user_id = None # 清除触发用户ID
# 确保清理 processing_message_id 以防止重复回复检测失效
context.processing_message_id = None
raise
except Exception as e:
self.stats["failed_executions"] += 1
logger.error(f"处理流 {stream_id} 时发生错误: {e}")
context.triggering_user_id = None # 清除触发用户ID
# 确保清理 processing_message_id
context.processing_message_id = None
raise
finally:
# 清除触发用户ID（所有情况下都需要）

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

@@ -332,8 +332,9 @@ class StreamLoopManager:
# 设置处理状态为正在处理
self._set_stream_processing_status(stream_id, True)
# 子任务跟踪
# 子任务跟踪 - 存储到 context 中以便打断时可以访问
child_tasks = set()
context._active_process_tasks = child_tasks # 临时存储子任务集合
try:
start_time = time.time()
@@ -362,8 +363,16 @@ class StreamLoopManager:
context.is_chatter_processing = True
logger.debug(f"设置 Chatter 处理标志: {stream_id}")
# 直接调用chatter_manager处理流上下文
results = await self.chatter_manager.process_stream_context(stream_id, context)
# 创建 chatter 处理任务，以便可以在打断时取消
chatter_task = asyncio.create_task(
self.chatter_manager.process_stream_context(stream_id, context),
name=f"chatter_process_{stream_id}"
)
child_tasks.add(chatter_task)
context._chatter_task = chatter_task # 保存主 chatter 任务引用
# 等待 chatter 任务完成
results = await chatter_task
success = results.get("success", False)
if success:
@@ -380,11 +389,25 @@ class StreamLoopManager:
return success
except asyncio.CancelledError:
logger.debug(f"流处理被取消: {stream_id}")
# 取消所有子任务
for child_task in child_tasks:
logger.info(f"流处理被取消，正在清理所有子任务: {stream_id}")
# 取消所有子任务（包括 chatter 任务和其后台任务）
cancel_count = 0
for child_task in list(child_tasks): # 使用 list() 创建副本避免迭代时修改
if not child_task.done():
child_task.cancel()
cancel_count += 1
if cancel_count > 0:
logger.info(f"已取消 {cancel_count} 个子任务: {stream_id}")
# 等待所有子任务真正结束（设置短超时）
try:
await asyncio.wait_for(
asyncio.gather(*child_tasks, return_exceptions=True),
timeout=1.0
)
except asyncio.TimeoutError:
logger.warning(f"等待子任务取消超时: {stream_id}")
raise
except Exception as e:
logger.error(f"流处理异常: {stream_id} - {e}", exc_info=True)
@@ -398,6 +421,12 @@ class StreamLoopManager:
context.is_chatter_processing = False
logger.debug(f"清除 Chatter 处理标志: {stream_id}")
# 清理临时存储的任务引用
if hasattr(context, '_active_process_tasks'):
delattr(context, '_active_process_tasks')
if hasattr(context, '_chatter_task'):
delattr(context, '_chatter_task')
# 无论成功或失败，都要设置处理状态为未处理
self._set_stream_processing_status(stream_id, False)

67
src/chat/replyer/default_generator.py
View File

@@ -1162,24 +1162,59 @@ class DefaultReplyer:
# 兼容旧的reply_to
sender, target = self._parse_reply_target(reply_to)
else:
# 获取 platform，如果不存在则从 chat_stream 获取，如果还是 None 则使用默认值
# 对于 respond 动作，reply_message 可能为 None（统一回应未读消息）
# 对于 reply 动作，reply_message 必须存在（针对特定消息回复）
if reply_message is None:
logger.warning("reply_message 为 None，无法构建prompt")
return ""
# 统一处理 DatabaseMessages 对象和字典
if isinstance(reply_message, DatabaseMessages):
platform = reply_message.chat_info.platform
user_id = reply_message.user_info.user_id
user_nickname = reply_message.user_info.user_nickname
user_cardname = reply_message.user_info.user_cardname
processed_plain_text = reply_message.processed_plain_text
# respond 模式：没有特定目标消息，使用通用的 sender 和 target
if prompt_mode == "normal":
# 从未读消息中获取最新的消息作为参考
from src.plugin_system.apis.chat_api import get_chat_manager
chat_manager = get_chat_manager()
chat_stream_obj = await chat_manager.get_stream(chat_id)
if chat_stream_obj:
unread_messages = chat_stream_obj.context_manager.get_unread_messages()
if unread_messages:
# 使用最后一条未读消息作为参考
last_msg = unread_messages[-1]
platform = last_msg.chat_info.platform if hasattr(last_msg, 'chat_info') else chat_stream.platform
user_id = last_msg.user_info.user_id if hasattr(last_msg, 'user_info') else ""
user_nickname = last_msg.user_info.user_nickname if hasattr(last_msg, 'user_info') else ""
user_cardname = last_msg.user_info.user_cardname if hasattr(last_msg, 'user_info') else ""
processed_plain_text = last_msg.processed_plain_text or ""
else:
# 没有未读消息，使用默认值
platform = chat_stream.platform
user_id = ""
user_nickname = ""
user_cardname = ""
processed_plain_text = ""
else:
# 无法获取 chat_stream，使用默认值
platform = chat_stream.platform
user_id = ""
user_nickname = ""
user_cardname = ""
processed_plain_text = ""
else:
# reply 模式下 reply_message 为 None 是错误的
logger.warning("reply_message 为 None，但处于 reply 模式，无法构建prompt")
return ""
else:
platform = reply_message.get("chat_info_platform")
user_id = reply_message.get("user_id")
user_nickname = reply_message.get("user_nickname")
user_cardname = reply_message.get("user_cardname")
processed_plain_text = reply_message.get("processed_plain_text")
# 有 reply_message，正常处理
# 统一处理 DatabaseMessages 对象和字典
if isinstance(reply_message, DatabaseMessages):
platform = reply_message.chat_info.platform
user_id = reply_message.user_info.user_id
user_nickname = reply_message.user_info.user_nickname
user_cardname = reply_message.user_info.user_cardname
processed_plain_text = reply_message.processed_plain_text
else:
platform = reply_message.get("chat_info_platform")
user_id = reply_message.get("user_id")
user_nickname = reply_message.get("user_nickname")
user_cardname = reply_message.get("user_cardname")
processed_plain_text = reply_message.get("processed_plain_text")
person_id = person_info_manager.get_person_id(
platform, # type: ignore

9
src/plugins/built_in/affinity_flow_chatter/core/affinity_chatter.py
View File

@@ -103,10 +103,19 @@ class AffinityChatter(BaseChatter):
return result
except asyncio.CancelledError:
logger.info(f"亲和力聊天处理器 {self.stream_id} 处理被取消")
self.stats["failed_executions"] += 1
self.last_activity_time = time.time()
# 清理 processing_message_id
context.processing_message_id = None
raise
except Exception as e:
logger.error(f"亲和力聊天处理器 {self.stream_id} 处理StreamContext时出错: {e}\n{traceback.format_exc()}")
self.stats["failed_executions"] += 1
self.last_activity_time = time.time()
# 清理 processing_message_id
context.processing_message_id = None
return {
"success": False,

24
src/plugins/built_in/affinity_flow_chatter/planner/planner.py
View File

@@ -78,9 +78,19 @@ class ChatterActionPlanner:
return await self._enhanced_plan_flow(context)
except asyncio.CancelledError:
logger.info(f"规划流程被取消: {self.chat_id}")
self.planner_stats["failed_plans"] += 1
# 确保清理 processing_message_id
if context:
context.processing_message_id = None
raise
except Exception as e:
logger.error(f"规划流程出错: {e}")
self.planner_stats["failed_plans"] += 1
# 确保清理 processing_message_id
if context:
context.processing_message_id = None
return [], None
async def _enhanced_plan_flow(self, context: "StreamContext | None") -> tuple[list[dict[str, Any]], Any | None]:
@@ -257,6 +267,13 @@ class ChatterActionPlanner:
# 14. 返回结果
return self._build_return_result(filtered_plan)
except asyncio.CancelledError:
logger.info(f"Focus模式流程被取消: {self.chat_id}")
self.planner_stats["failed_plans"] += 1
# 清理处理标记
if context:
context.processing_message_id = None
raise
except Exception as e:
logger.error(f"Focus模式流程出错: {e}")
self.planner_stats["failed_plans"] += 1
@@ -394,6 +411,13 @@ class ChatterActionPlanner:
return [asdict(no_action)], None
except asyncio.CancelledError:
logger.info(f"Normal模式流程被取消: {self.chat_id}")
self.planner_stats["failed_plans"] += 1
# 清理处理标记
if context:
context.processing_message_id = None
raise
except Exception as e:
logger.error(f"Normal模式 - 流程出错: {e}")
self.planner_stats["failed_plans"] += 1