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

feat: 完全分离回复 兴趣和 消息阅读；添加概率回复机制，优化兴趣监控逻辑，重构相关功能以支持更灵活的回复触发条件

Browse Source
This commit is contained in:
SengokuCola
2025-04-17 16:51:35 +08:00
parent cfdaf00559
commit a2333f9f82
7 changed files with 730 additions and 376 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
src/plugins/chat_module/heartFC_chat/heartFC__generator.py
View File

@@ -48,45 +48,21 @@ class ResponseGenerator:
arousal_multiplier = MoodManager.get_instance().get_arousal_multiplier()
with Timer() as t_generate_response:
checked = False
if random.random() > 0:
checked = False
current_model = self.model_normal
current_model.temperature = (
global_config.llm_normal["temp"] * arousal_multiplier
) # 激活度越高，温度越高
model_response = await self._generate_response_with_model(
message, current_model, thinking_id, mode="normal"
)
model_checked_response = model_response
else:
checked = True
current_model = self.model_normal
current_model.temperature = (
global_config.llm_normal["temp"] * arousal_multiplier
) # 激活度越高，温度越高
print(f"生成{message.processed_plain_text}回复温度是：{current_model.temperature}")
model_response = await self._generate_response_with_model(
message, current_model, thinking_id, mode="simple"
)
current_model = self.model_normal
current_model.temperature = (
global_config.llm_normal["temp"] * arousal_multiplier
) # 激活度越高，温度越高
model_response = await self._generate_response_with_model(
message, current_model, thinking_id, mode="normal"
)
current_model.temperature = global_config.llm_normal["temp"]
model_checked_response = await self._check_response_with_model(
message, model_response, current_model, thinking_id
)
if model_response:
if checked:
logger.info(
f"{global_config.BOT_NICKNAME}的回复是：{model_response}，思忖后，回复是：{model_checked_response},生成回复时间: {t_generate_response.human_readable}"
)
else:
logger.info(
f"{global_config.BOT_NICKNAME}的回复是：{model_response},生成回复时间: {t_generate_response.human_readable}"
)
model_processed_response = await self._process_response(model_checked_response)
logger.info(
f"{global_config.BOT_NICKNAME}的回复是：{model_response},生成回复时间: {t_generate_response.human_readable}"
)
model_processed_response = await self._process_response(model_response)
return model_processed_response
else:

413
src/plugins/chat_module/heartFC_chat/heartFC_chat.py
View File

@@ -18,6 +18,8 @@ from src.plugins.respon_info_catcher.info_catcher import info_catcher_manager
from ...utils.timer_calculater import Timer
from src.do_tool.tool_use import ToolUser
from .interest import InterestManager, InterestChatting
from src.plugins.chat.chat_stream import chat_manager
from src.plugins.chat.message import MessageInfo
# 定义日志配置
chat_config = LogConfig(
@@ -28,7 +30,6 @@ chat_config = LogConfig(
logger = get_module_logger("heartFC_chat", config=chat_config)
# 新增常量
INTEREST_LEVEL_REPLY_THRESHOLD = 4.0
INTEREST_MONITOR_INTERVAL_SECONDS = 1
class HeartFC_Chat:
@@ -41,87 +42,105 @@ class HeartFC_Chat:
self._interest_monitor_task: Optional[asyncio.Task] = None
async def start(self):
"""Starts asynchronous tasks like the interest monitor."""
logger.info("HeartFC_Chat starting asynchronous tasks...")
"""启动异步任务,如兴趣监控器"""
logger.info("HeartFC_Chat 正在启动异步任务...")
await self.interest_manager.start_background_tasks()
self._initialize_monitor_task()
logger.info("HeartFC_Chat asynchronous tasks started.")
logger.info("HeartFC_Chat 异步任务启动完成")
def _initialize_monitor_task(self):
"""启动后台兴趣监控任务"""
"""启动后台兴趣监控任务，可以检查兴趣是否足以开启心流对话"""
if self._interest_monitor_task is None or self._interest_monitor_task.done():
try:
loop = asyncio.get_running_loop()
self._interest_monitor_task = loop.create_task(self._interest_monitor_loop())
logger.info(f"Interest monitor task created. Interval: {INTEREST_MONITOR_INTERVAL_SECONDS}s, Level Threshold: {INTEREST_LEVEL_REPLY_THRESHOLD}")
logger.info(f"兴趣监控任务已创建。监控间隔: {INTEREST_MONITOR_INTERVAL_SECONDS}秒。")
except RuntimeError:
logger.error("Failed to create interest monitor task: No running event loop.")
logger.error("创建兴趣监控任务失败：没有运行中的事件循环。")
raise
else:
logger.warning("Interest monitor task creation skipped: already running or exists.")
logger.warning("跳过兴趣监控任务创建：任务已存在或正在运行。")
async def _interest_monitor_loop(self):
"""后台任务，定期检查兴趣度变化并触发回复"""
logger.info("Interest monitor loop starting...")
await asyncio.sleep(0.3)
logger.info("兴趣监控循环开始...")
while True:
await asyncio.sleep(INTEREST_MONITOR_INTERVAL_SECONDS)
try:
interest_items_snapshot: List[tuple[str, InterestChatting]] = []
stream_ids = list(self.interest_manager.interest_dict.keys())
for stream_id in stream_ids:
chatting_instance = self.interest_manager.get_interest_chatting(stream_id)
if chatting_instance:
interest_items_snapshot.append((stream_id, chatting_instance))
# --- 修改：遍历 SubHeartflow 并检查触发器 ---
active_stream_ids = list(heartflow.get_all_subheartflows_streams_ids()) # 需要 heartflow 提供此方法
logger.trace(f"检查 {len(active_stream_ids)} 个活跃流是否足以开启心流对话...")
for stream_id, chatting_instance in interest_items_snapshot:
triggering_message = chatting_instance.last_triggering_message
current_interest = chatting_instance.get_interest()
for stream_id in active_stream_ids:
sub_hf = heartflow.get_subheartflow(stream_id)
if not sub_hf:
logger.warning(f"监控循环: 无法获取活跃流 {stream_id} 的 sub_hf")
continue
# 添加调试日志，检查触发条件
# logger.debug(f"[兴趣监控][{stream_id}] 当前兴趣: {current_interest:.2f}, 阈值: {INTEREST_LEVEL_REPLY_THRESHOLD}, 触发消息存在: {triggering_message is not None}")
# --- 获取 Observation 和消息列表 --- #
observation = sub_hf._get_primary_observation()
if not observation:
logger.warning(f"[{stream_id}] SubHeartflow 没有在观察，无法检查触发器。")
continue
observed_messages = observation.talking_message # 获取消息字典列表
# --- 结束获取 --- #
if current_interest > INTEREST_LEVEL_REPLY_THRESHOLD and triggering_message is not None:
logger.info(f"[{stream_id}] 检测到高兴趣度 ({current_interest:.2f} > {INTEREST_LEVEL_REPLY_THRESHOLD}). 基于消息 ID: {triggering_message.message_info.message_id} 的上下文触发回复") # 更新日志信息使其更清晰
should_trigger = False
try:
# check_reply_trigger 可以选择性地接收 observed_messages 作为参数
should_trigger = await sub_hf.check_reply_trigger() # 目前 check_reply_trigger 还不处理这个
except Exception as e:
logger.error(f"错误调用 check_reply_trigger 流 {stream_id}: {e}")
logger.error(traceback.format_exc())
chatting_instance.reset_trigger_info()
logger.debug(f"[{stream_id}] Trigger info reset before starting reply task.")
if should_trigger:
logger.info(f"[{stream_id}] SubHeartflow 决定开启心流对话。")
# 调用修改后的处理函数，传递 stream_id 和 observed_messages
asyncio.create_task(self._process_triggered_reply(stream_id, observed_messages))
asyncio.create_task(self._process_triggered_reply(stream_id, triggering_message))
except asyncio.CancelledError:
logger.info("Interest monitor loop cancelled.")
logger.info("兴趣监控循环已取消。")
break
except Exception as e:
logger.error(f"Error in interest monitor loop: {e}")
logger.error(f"兴趣监控循环错误: {e}")
logger.error(traceback.format_exc())
await asyncio.sleep(5)
await asyncio.sleep(5) # 发生错误时等待
async def _process_triggered_reply(self, stream_id: str, triggering_message: MessageRecv):
"""Helper coroutine to handle the processing of a triggered reply based on interest level."""
async def _process_triggered_reply(self, stream_id: str, observed_messages: List[dict]):
"""Helper coroutine to handle the processing of a triggered reply based on SubHeartflow trigger."""
try:
logger.info(f"[{stream_id}] Starting level-triggered reply generation for message ID: {triggering_message.message_info.message_id}...")
await self.trigger_reply_generation(triggering_message)
logger.info(f"[{stream_id}] SubHeartflow 触发回复...")
# 调用修改后的 trigger_reply_generation
await self.trigger_reply_generation(stream_id, observed_messages)
# 在回复处理后降低兴趣度，降低固定值：新阈值的一半
decrease_value = INTEREST_LEVEL_REPLY_THRESHOLD / 2
self.interest_manager.decrease_interest(stream_id, value=decrease_value)
post_trigger_interest = self.interest_manager.get_interest(stream_id)
# 更新日志以反映降低的是基于新阈值的固定值
logger.info(f"[{stream_id}] Interest decreased by fixed value {decrease_value:.2f} (LevelThreshold/2) after processing level-triggered reply. Current interest: {post_trigger_interest:.2f}")
# --- 调整兴趣降低逻辑 ---
# 这里的兴趣降低可能不再适用，或者需要基于不同的逻辑
# 例如，回复后可以将 SubHeartflow 的某种"回复意愿"状态重置
# 暂时注释掉，或根据需要调整
# chatting_instance = self.interest_manager.get_interest_chatting(stream_id)
# if chatting_instance:
# decrease_value = chatting_instance.trigger_threshold / 2 # 使用实例的阈值
# self.interest_manager.decrease_interest(stream_id, value=decrease_value)
# post_trigger_interest = self.interest_manager.get_interest(stream_id) # 获取更新后的兴趣
# logger.info(f"[{stream_id}] Interest decreased by {decrease_value:.2f} (InstanceThreshold/2) after processing triggered reply. Current interest: {post_trigger_interest:.2f}")
# else:
# logger.warning(f"[{stream_id}] Could not find InterestChatting instance after reply processing to decrease interest.")
logger.debug(f"[{stream_id}] Reply processing finished. (Interest decrease logic needs review).")
except Exception as e:
logger.error(f"Error processing level-triggered reply for stream_id {stream_id}, context message_id {triggering_message.message_info.message_id}: {e}")
logger.error(f"Error processing SubHeartflow-triggered reply for stream_id {stream_id}: {e}") # 更新日志信息
logger.error(traceback.format_exc())
# --- 结束修改 ---
async def _create_thinking_message(self, message: MessageRecv):
"""创建思考消息 (从 message 获取信息)"""
chat = message.chat_stream
if not chat:
logger.error(f"Cannot create thinking message, chat_stream is None for message ID: {message.message_info.message_id}")
return None
userinfo = message.message_info.user_info # 发起思考的用户（即原始消息发送者）
messageinfo = message.message_info # 原始消息信息
async def _create_thinking_message(self, anchor_message: Optional[MessageRecv]):
"""创建思考消息 (尝试锚定到 anchor_message)"""
if not anchor_message or not anchor_message.chat_stream:
logger.error("无法创建思考消息，缺少有效的锚点消息或聊天流。")
return None
chat = anchor_message.chat_stream
messageinfo = anchor_message.message_info
bot_user_info = UserInfo(
user_id=global_config.BOT_QQ,
user_nickname=global_config.BOT_NICKNAME,
@@ -133,17 +152,21 @@ class HeartFC_Chat:
thinking_message = MessageThinking(
message_id=thinking_id,
chat_stream=chat,
bot_user_info=bot_user_info, # 思考消息的发出者是 bot
reply=message, # 回复的是原始消息
bot_user_info=bot_user_info,
reply=anchor_message, # 回复的是锚点消息
thinking_start_time=thinking_time_point,
)
MessageManager().add_message(thinking_message)
return thinking_id
async def _send_response_messages(self, message: MessageRecv, response_set: List[str], thinking_id) -> MessageSending:
chat = message.chat_stream
async def _send_response_messages(self, anchor_message: Optional[MessageRecv], response_set: List[str], thinking_id) -> Optional[MessageSending]:
"""发送回复消息 (尝试锚定到 anchor_message)"""
if not anchor_message or not anchor_message.chat_stream:
logger.error("无法发送回复，缺少有效的锚点消息或聊天流。")
return None
chat = anchor_message.chat_stream
container = MessageManager().get_container(chat.stream_id)
thinking_message = None
for msg in container.messages:
@@ -152,26 +175,26 @@ class HeartFC_Chat:
container.messages.remove(msg)
break
if not thinking_message:
logger.warning("未找到对应的思考消息，可能已超时被移除")
logger.warning(f"[{chat.stream_id}] 未找到对应的思考消息 {thinking_id}，可能已超时被移除")
return None
thinking_start_time = thinking_message.thinking_start_time
message_set = MessageSet(chat, thinking_id)
mark_head = False
first_bot_msg = None
for msg in response_set:
message_segment = Seg(type="text", data=msg)
for msg_text in response_set:
message_segment = Seg(type="text", data=msg_text)
bot_message = MessageSending(
message_id=thinking_id,
message_id=thinking_id, # 使用 thinking_id 作为批次标识
chat_stream=chat,
bot_user_info=UserInfo(
user_id=global_config.BOT_QQ,
user_nickname=global_config.BOT_NICKNAME,
platform=message.message_info.platform, # 从传入的 message 获取 platform
platform=anchor_message.message_info.platform,
),
sender_info=message.message_info.user_info, # 发送给谁
sender_info=anchor_message.message_info.user_info, # 发送给锚点消息的用户
message_segment=message_segment,
reply=message, # 回复原始消息
reply=anchor_message, # 回复锚点消息
is_head=not mark_head,
is_emoji=False,
thinking_start_time=thinking_start_time,
@@ -180,185 +203,277 @@ class HeartFC_Chat:
mark_head = True
first_bot_msg = bot_message
message_set.add_message(bot_message)
MessageManager().add_message(message_set)
return first_bot_msg
async def _handle_emoji(self, message: MessageRecv, response_set, send_emoji=""):
"""处理表情包 (从 message 获取信息)"""
chat = message.chat_stream
if message_set.messages: # 确保有消息才添加
MessageManager().add_message(message_set)
return first_bot_msg
else:
logger.warning(f"[{chat.stream_id}] 没有生成有效的回复消息集，无法发送。")
return None
async def _handle_emoji(self, anchor_message: Optional[MessageRecv], response_set, send_emoji=""):
"""处理表情包 (尝试锚定到 anchor_message)"""
if not anchor_message or not anchor_message.chat_stream:
logger.error("无法处理表情包，缺少有效的锚点消息或聊天流。")
return
chat = anchor_message.chat_stream
if send_emoji:
emoji_raw = await emoji_manager.get_emoji_for_text(send_emoji)
else:
emoji_text_source = "".join(response_set) if response_set else ""
emoji_raw = await emoji_manager.get_emoji_for_text(emoji_text_source)
if emoji_raw:
emoji_path, description = emoji_raw
emoji_cq = image_path_to_base64(emoji_path)
thinking_time_point = round(message.message_info.time, 2)
# 使用当前时间戳，因为没有原始消息的时间戳
thinking_time_point = round(time.time(), 2)
message_segment = Seg(type="emoji", data=emoji_cq)
bot_message = MessageSending(
message_id="mt" + str(thinking_time_point),
message_id="me" + str(thinking_time_point), # 使用不同的 ID 前缀?
chat_stream=chat,
bot_user_info=UserInfo(
user_id=global_config.BOT_QQ,
user_nickname=global_config.BOT_NICKNAME,
platform=message.message_info.platform,
platform=anchor_message.message_info.platform,
),
sender_info=message.message_info.user_info, # 发送给谁
sender_info=anchor_message.message_info.user_info,
message_segment=message_segment,
reply=message, # 回复原始消息
reply=anchor_message, # 回复锚点消息
is_head=False,
is_emoji=True,
)
MessageManager().add_message(bot_message)
async def _update_relationship(self, message: MessageRecv, response_set):
"""更新关系情绪"""
async def _update_relationship(self, anchor_message: Optional[MessageRecv], response_set):
"""更新关系情绪 (尝试基于 anchor_message)"""
if not anchor_message or not anchor_message.chat_stream:
logger.error("无法更新关系情绪，缺少有效的锚点消息或聊天流。")
return
# 关系更新依赖于理解回复是针对谁的，以及原始消息的上下文
# 这里的实现可能需要调整，取决于关系管理器如何工作
ori_response = ",".join(response_set)
stance, emotion = await self.gpt._get_emotion_tags(ori_response, message.processed_plain_text)
# 注意：anchor_message.processed_plain_text 是锚点消息的文本，不一定是思考的全部上下文
stance, emotion = await self.gpt._get_emotion_tags(ori_response, anchor_message.processed_plain_text)
await relationship_manager.calculate_update_relationship_value(
chat_stream=message.chat_stream, label=emotion, stance=stance
chat_stream=anchor_message.chat_stream, # 使用锚点消息的流
label=emotion,
stance=stance
)
self.mood_manager.update_mood_from_emotion(emotion, global_config.mood_intensity_factor)
async def trigger_reply_generation(self, message: MessageRecv):
"""根据意愿阈值触发的实际回复生成和发送逻辑 (V3 - 简化参数)"""
chat = message.chat_stream
userinfo = message.message_info.user_info
messageinfo = message.message_info
async def trigger_reply_generation(self, stream_id: str, observed_messages: List[dict]):
"""根据 SubHeartflow 的触发信号生成回复 (基于观察)"""
chat = None
sub_hf = None
anchor_message: Optional[MessageRecv] = None # <--- 重命名，用于锚定回复的消息对象
userinfo: Optional[UserInfo] = None
messageinfo: Optional[MessageInfo] = None
timing_results = {}
current_mind = None
response_set = None
thinking_id = None
info_catcher = None
try:
# --- 1. 获取核心对象：ChatStream 和 SubHeartflow ---
try:
with Timer("观察", timing_results):
sub_hf = heartflow.get_subheartflow(chat.stream_id)
if not sub_hf:
logger.warning(f"尝试观察时未找到 stream_id {chat.stream_id} 的 subheartflow")
with Timer("获取聊天流和子心流", timing_results):
chat = chat_manager.get_stream(stream_id)
if not chat:
logger.error(f"[{stream_id}] 无法找到聊天流对象，无法生成回复。")
return
sub_hf = heartflow.get_subheartflow(stream_id)
if not sub_hf:
logger.error(f"[{stream_id}] 无法找到子心流对象，无法生成回复。")
return
await sub_hf.do_observe()
except Exception as e:
logger.error(f"心流观察失败: {e}")
logger.error(traceback.format_exc())
logger.error(f"[{stream_id}] 获取 ChatStream 或 SubHeartflow 时出错: {e}")
logger.error(traceback.format_exc())
return
container = MessageManager().get_container(chat.stream_id)
thinking_count = container.count_thinking_messages()
max_thinking_messages = getattr(global_config, 'max_concurrent_thinking_messages', 3)
if thinking_count >= max_thinking_messages:
logger.warning(f"聊天流 {chat.stream_id} 已有 {thinking_count} 条思考消息，取消回复。触发消息: {message.processed_plain_text[:30]}...")
# --- 2. 尝试从 observed_messages 重建最后一条消息作为锚点 --- #
try:
with Timer("获取最后消息锚点", timing_results):
if observed_messages:
last_msg_dict = observed_messages[-1] # 直接从传入列表获取最后一条
# 尝试从字典重建 MessageRecv 对象（可能需要调整 MessageRecv 的构造方式或创建一个辅助函数）
# 这是一个简化示例，假设 MessageRecv 可以从字典初始化
# 你可能需要根据 MessageRecv 的实际 __init__ 来调整
try:
anchor_message = MessageRecv(last_msg_dict) # 假设 MessageRecv 支持从字典创建
userinfo = anchor_message.message_info.user_info
messageinfo = anchor_message.message_info
logger.debug(f"[{stream_id}] 获取到最后消息作为锚点: ID={messageinfo.message_id}, Sender={userinfo.user_nickname}")
except Exception as e_msg:
logger.error(f"[{stream_id}] 从字典重建最后消息 MessageRecv 失败: {e_msg}. 字典: {last_msg_dict}")
anchor_message = None # 重置以表示失败
else:
logger.warning(f"[{stream_id}] 无法从 Observation 获取最后消息锚点。")
except Exception as e:
logger.error(f"[{stream_id}] 获取最后消息锚点时出错: {e}")
logger.error(traceback.format_exc())
# 即使没有锚点，也可能继续尝试生成非回复性消息，取决于后续逻辑
# --- 3. 检查是否能继续 (需要思考消息锚点) ---
if not anchor_message:
logger.warning(f"[{stream_id}] 没有有效的消息锚点，无法创建思考消息和发送回复。取消回复生成。")
return
# --- 4. 检查并发思考限制 (使用 anchor_message 简化获取) ---
try:
container = MessageManager().get_container(chat.stream_id)
thinking_count = container.count_thinking_messages()
max_thinking_messages = getattr(global_config, 'max_concurrent_thinking_messages', 3)
if thinking_count >= max_thinking_messages:
logger.warning(f"聊天流 {chat.stream_id} 已有 {thinking_count} 条思考消息，取消回复。")
return
except Exception as e:
logger.error(f"[{stream_id}] 检查并发思考限制时出错: {e}")
return
# --- 5. 创建思考消息 (使用 anchor_message) ---
try:
with Timer("创建思考消息", timing_results):
thinking_id = await self._create_thinking_message(message)
# 注意：这里传递 anchor_message 给 _create_thinking_message
thinking_id = await self._create_thinking_message(anchor_message)
except Exception as e:
logger.error(f"心流创建思考消息失败: {e}")
logger.error(f"[{stream_id}] 创建思考消息失败: {e}")
return
if not thinking_id:
logger.error("未能成功创建思考消息 ID，无法继续回复流程。")
logger.error(f"[{stream_id}] 未能成功创建思考消息 ID，无法继续回复流程。")
return
logger.trace(f"创建捕捉器，thinking_id:{thinking_id}")
# --- 6. 信息捕捉器 (使用 anchor_message) ---
logger.trace(f"[{stream_id}] 创建捕捉器，thinking_id:{thinking_id}")
info_catcher = info_catcher_manager.get_info_catcher(thinking_id)
info_catcher.catch_decide_to_response(message)
info_catcher.catch_decide_to_response(anchor_message)
# --- 7. 思考前使用工具 --- #
get_mid_memory_id = []
tool_result_info = {}
send_emoji = ""
observation_context_text = "" # 从 observation 获取上下文文本
try:
with Timer("思考前使用工具", timing_results):
tool_result = await self.tool_user.use_tool(
message.processed_plain_text,
userinfo.user_nickname,
chat,
heartflow.get_subheartflow(chat.stream_id),
)
if tool_result.get("used_tools", False):
if "structured_info" in tool_result:
tool_result_info = tool_result["structured_info"]
get_mid_memory_id = []
for tool_name, tool_data in tool_result_info.items():
if tool_name == "mid_chat_mem":
for mid_memory in tool_data:
get_mid_memory_id.append(mid_memory["content"])
if tool_name == "send_emoji":
send_emoji = tool_data[0]["content"]
except Exception as e:
logger.error(f"思考前工具调用失败: {e}")
logger.error(traceback.format_exc())
# --- 使用传入的 observed_messages 构建上下文文本 --- #
if observed_messages:
# 可以选择转换全部消息，或只转换最后几条
# 这里示例转换全部消息
context_texts = []
for msg_dict in observed_messages:
# 假设 detailed_plain_text 字段包含所需文本
# 你可能需要更复杂的逻辑来格式化，例如添加发送者和时间
text = msg_dict.get('detailed_plain_text', '')
if text: context_texts.append(text)
observation_context_text = "\n".join(context_texts)
logger.debug(f"[{stream_id}] Context for tools:\n{observation_context_text[-200:]}...") # 打印部分上下文
else:
logger.warning(f"[{stream_id}] observed_messages 列表为空，无法为工具提供上下文。")
current_mind, past_mind = "", ""
try:
with Timer("思考前脑内状态", timing_results):
sub_hf = heartflow.get_subheartflow(chat.stream_id)
if sub_hf:
current_mind, past_mind = await sub_hf.do_thinking_before_reply(
message_txt=message.processed_plain_text,
sender_info=userinfo,
if observation_context_text:
with Timer("思考前使用工具", timing_results):
tool_result = await self.tool_user.use_tool(
message_txt=observation_context_text, # <--- 使用观察上下文
chat_stream=chat,
obs_id=get_mid_memory_id,
extra_info=tool_result_info,
sub_heartflow=sub_hf
)
else:
logger.warning(f"尝试思考前状态时未找到 stream_id {chat.stream_id} 的 subheartflow")
if tool_result.get("used_tools", False):
if "structured_info" in tool_result:
tool_result_info = tool_result["structured_info"]
get_mid_memory_id = []
for tool_name, tool_data in tool_result_info.items():
if tool_name == "mid_chat_mem":
for mid_memory in tool_data:
get_mid_memory_id.append(mid_memory["content"])
if tool_name == "send_emoji":
send_emoji = tool_data[0]["content"]
except Exception as e:
logger.error(f"心流思考前脑内状态失败: {e}")
logger.error(f"[{stream_id}] 思考前工具调用失败: {e}")
logger.error(traceback.format_exc())
if info_catcher:
info_catcher.catch_afer_shf_step(timing_results.get("思考前脑内状态"), past_mind, current_mind)
# --- 8. 调用 SubHeartflow 进行思考 (不传递具体消息文本和发送者) ---
try:
with Timer("生成回复", timing_results):
response_set = await self.gpt.generate_response(message, thinking_id)
with Timer("生成内心想法(SubHF)", timing_results):
# 不再传递 message_txt 和 sender_info, SubHeartflow 应基于其内部观察
current_mind, past_mind = await sub_hf.do_thinking_before_reply(
chat_stream=chat,
extra_info=tool_result_info,
obs_id=get_mid_memory_id,
)
logger.info(f"[{stream_id}] SubHeartflow 思考完成: {current_mind}")
except Exception as e:
logger.error(f"GPT 生成回复失败: {e}")
logger.error(f"[{stream_id}] SubHeartflow 思考失败: {e}")
logger.error(traceback.format_exc())
if info_catcher: info_catcher.done_catch()
return # 思考失败则不继续
if info_catcher:
info_catcher.catch_afer_shf_step(timing_results.get("生成内心想法(SubHF)"), past_mind, current_mind)
# --- 9. 调用 ResponseGenerator 生成回复 (使用 anchor_message 和 current_mind) ---
try:
with Timer("生成最终回复(GPT)", timing_results):
response_set = await self.gpt.generate_response(anchor_message, thinking_id, current_mind=current_mind)
except Exception as e:
logger.error(f"[{stream_id}] GPT 生成回复失败: {e}")
logger.error(traceback.format_exc())
if info_catcher: info_catcher.done_catch()
return
if info_catcher:
info_catcher.catch_after_generate_response(timing_results.get("生成回复"))
info_catcher.catch_after_generate_response(timing_results.get("生成最终回复(GPT)"))
if not response_set:
logger.info("回复生成失败，返回为空")
logger.info(f"[{stream_id}] 回复生成失败或为空。")
if info_catcher: info_catcher.done_catch()
return
# --- 10. 发送消息 (使用 anchor_message) ---
first_bot_msg = None
try:
with Timer("发送消息", timing_results):
first_bot_msg = await self._send_response_messages(message, response_set, thinking_id)
first_bot_msg = await self._send_response_messages(anchor_message, response_set, thinking_id)
except Exception as e:
logger.error(f"心流发送消息失败: {e}")
logger.error(f"[{stream_id}] 发送消息失败: {e}")
logger.error(traceback.format_exc())
if info_catcher:
info_catcher.catch_after_response(timing_results.get("发送消息"), response_set, first_bot_msg)
info_catcher.done_catch()
info_catcher.done_catch() # 完成捕捉
# --- 11. 处理表情包 (使用 anchor_message) ---
try:
with Timer("处理表情包", timing_results):
if send_emoji:
logger.info(f"麦麦决定发送表情包{send_emoji}")
await self._handle_emoji(message, response_set, send_emoji)
logger.info(f"[{stream_id}] 决定发送表情包 {send_emoji}")
await self._handle_emoji(anchor_message, response_set, send_emoji)
except Exception as e:
logger.error(f"心流处理表情包失败: {e}")
logger.error(f"[{stream_id}] 处理表情包失败: {e}")
logger.error(traceback.format_exc())
# --- 12. 记录性能日志 --- #
timing_str = " | ".join([f"{step}: {duration:.2f}秒" for step, duration in timing_results.items()])
trigger_msg = message.processed_plain_text
response_msg = " ".join(response_set) if response_set else "无回复"
logger.info(f"回复任务完成: 触发消息: {trigger_msg[:20]}... | 思维消息: {response_msg[:20]}... | 性能计时: {timing_str}")
logger.info(f"[{stream_id}] 回复任务完成 (Observation Triggered): | 思维消息: {response_msg[:30]}... | 性能计时: {timing_str}")
if first_bot_msg:
# --- 13. 更新关系情绪 (使用 anchor_message) ---
if first_bot_msg: # 仅在成功发送消息后
try:
with Timer("更新关系情绪", timing_results):
await self._update_relationship(message, response_set)
await self._update_relationship(anchor_message, response_set)
except Exception as e:
logger.error(f"更新关系情绪失败: {e}")
logger.error(f"[{stream_id}] 更新关系情绪失败: {e}")
logger.error(traceback.format_exc())
except Exception as e:
logger.error(f"回复生成任务失败 (trigger_reply_generation V3): {e}")
logger.error(f"回复生成任务失败 (trigger_reply_generation V4 - Observation Triggered): {e}")
logger.error(traceback.format_exc())
finally:
pass
# 可以在这里添加清理逻辑，如果有的话
pass
# --- 结束重构 ---
# _create_thinking_message, _send_response_messages, _handle_emoji, _update_relationship
# 这几个辅助方法目前仍然依赖 MessageRecv 对象。
# 如果无法可靠地从 Observation 获取并重建最后一条消息的 MessageRecv，
# 或者希望回复不锚定具体消息，那么这些方法也需要进一步重构。

2
src/plugins/chat_module/heartFC_chat/heartFC_processor.py
View File

@@ -120,7 +120,7 @@ class HeartFC_Processor:
# 更新兴趣度
try:
self.interest_manager.increase_interest(chat.stream_id, value=interested_rate, message=message)
self.interest_manager.increase_interest(chat.stream_id, value=interested_rate)
current_interest = self.interest_manager.get_interest(chat.stream_id) # 获取更新后的值用于日志
logger.trace(f"使用激活率 {interested_rate:.2f} 更新后 (通过缓冲后)，当前兴趣度: {current_interest:.2f}")

226
src/plugins/chat_module/heartFC_chat/interest.py
View File

@@ -6,6 +6,7 @@ import json # 引入 json
import os # 引入 os
import traceback # <--- 添加导入
from typing import Optional # <--- 添加导入
import random # <--- 添加导入 random
from src.common.logger import get_module_logger, LogConfig, DEFAULT_CONFIG # 引入 DEFAULT_CONFIG
from src.plugins.chat.chat_stream import chat_manager # *** Import ChatManager ***
from ...chat.message import MessageRecv # 导入 MessageRecv
@@ -20,7 +21,6 @@ logger = get_module_logger("InterestManager", config=interest_log_config)
# 定义常量
DEFAULT_DECAY_RATE_PER_SECOND = 0.95 # 每秒衰减率 (兴趣保留 99%)
# DEFAULT_INCREASE_AMOUNT = 10.0 # 不再需要固定增加值
MAX_INTEREST = 10.0 # 最大兴趣值
MIN_INTEREST_THRESHOLD = 0.1 # 低于此值可能被清理 (可选)
CLEANUP_INTERVAL_SECONDS = 3600 # 清理任务运行间隔 (例如：1小时)
@@ -32,16 +32,39 @@ HISTORY_LOG_FILENAME = "interest_history.log" # 新的历史日志文件名
# 移除阈值，将移至 HeartFC_Chat
# INTEREST_INCREASE_THRESHOLD = 0.5
# --- 新增：概率回复相关常量 ---
REPLY_TRIGGER_THRESHOLD = 5.0 # 触发概率回复的兴趣阈值 (示例值)
BASE_REPLY_PROBABILITY = 0.05 # 首次超过阈值时的基础回复概率 (示例值)
PROBABILITY_INCREASE_RATE_PER_SECOND = 0.02 # 高于阈值时，每秒概率增加量 (线性增长, 示例值)
PROBABILITY_DECAY_FACTOR_PER_SECOND = 0.3 # 低于阈值时，每秒概率衰减因子 (指数衰减, 示例值)
MAX_REPLY_PROBABILITY = 0.95 # 回复概率上限 (示例值)
# --- 结束：概率回复相关常量 ---
class InterestChatting:
def __init__(self, decay_rate=DEFAULT_DECAY_RATE_PER_SECOND, max_interest=MAX_INTEREST):
def __init__(self,
decay_rate=DEFAULT_DECAY_RATE_PER_SECOND,
max_interest=MAX_INTEREST,
trigger_threshold=REPLY_TRIGGER_THRESHOLD,
base_reply_probability=BASE_REPLY_PROBABILITY,
increase_rate=PROBABILITY_INCREASE_RATE_PER_SECOND,
decay_factor=PROBABILITY_DECAY_FACTOR_PER_SECOND,
max_probability=MAX_REPLY_PROBABILITY):
self.interest_level: float = 0.0
self.last_update_time: float = time.time()
self.last_update_time: float = time.time() # 同时作为兴趣和概率的更新时间基准
self.decay_rate_per_second: float = decay_rate
# self.increase_amount: float = increase_amount # 移除固定的 increase_amount
self.max_interest: float = max_interest
# 新增：用于追踪最后一次显著增加的信息，供外部监控任务使用
self.last_increase_amount: float = 0.0
self.last_triggering_message: MessageRecv | None = None
self.last_interaction_time: float = self.last_update_time # 新增：最后交互时间
# --- 新增：概率回复相关属性 ---
self.trigger_threshold: float = trigger_threshold
self.base_reply_probability: float = base_reply_probability
self.probability_increase_rate: float = increase_rate
self.probability_decay_factor: float = decay_factor
self.max_reply_probability: float = max_probability
self.current_reply_probability: float = 0.0
self.is_above_threshold: bool = False # 标记兴趣值是否高于阈值
# --- 结束：概率回复相关属性 ---
def _calculate_decay(self, current_time: float):
"""计算从上次更新到现在的衰减"""
@@ -49,6 +72,7 @@ class InterestChatting:
if time_delta > 0:
# 指数衰减: interest = interest * (decay_rate ^ time_delta)
# 添加处理极小兴趣值避免 math domain error
old_interest = self.interest_level
if self.interest_level < 1e-9:
self.interest_level = 0.0
else:
@@ -71,46 +95,141 @@ class InterestChatting:
# 防止低于阈值 (如果需要)
# self.interest_level = max(self.interest_level, MIN_INTEREST_THRESHOLD)
self.last_update_time = current_time
# 只有在兴趣值发生变化时才更新时间戳
if old_interest != self.interest_level:
self.last_update_time = current_time
def increase_interest(self, current_time: float, value: float, message: Optional[MessageRecv]):
"""根据传入的值增加兴趣值，并记录增加量和关联消息"""
self._calculate_decay(current_time) # 先计算衰减
# 记录这次增加的具体数值和消息，供外部判断是否触发
def _update_reply_probability(self, current_time: float):
"""根据当前兴趣是否超过阈值及时间差，更新回复概率"""
time_delta = current_time - self.last_update_time
if time_delta <= 0:
return # 时间未前进，无需更新
currently_above = self.interest_level >= self.trigger_threshold
if currently_above:
if not self.is_above_threshold:
# 刚跨过阈值，重置为基础概率
self.current_reply_probability = self.base_reply_probability
logger.debug(f"兴趣跨过阈值 ({self.trigger_threshold}). 概率重置为基础值: {self.base_reply_probability:.4f}")
else:
# 持续高于阈值，线性增加概率
increase_amount = self.probability_increase_rate * time_delta
self.current_reply_probability += increase_amount
logger.debug(f"兴趣高于阈值 ({self.trigger_threshold}) 持续 {time_delta:.2f}秒. 概率增加 {increase_amount:.4f} 到 {self.current_reply_probability:.4f}")
# 限制概率不超过最大值
self.current_reply_probability = min(self.current_reply_probability, self.max_reply_probability)
else: # 低于阈值
if self.is_above_threshold:
# 刚低于阈值，开始衰减
logger.debug(f"兴趣低于阈值 ({self.trigger_threshold}). 概率衰减开始于 {self.current_reply_probability:.4f}")
# else: # 持续低于阈值，继续衰减
# pass # 不需要特殊处理
# 指数衰减概率
# 检查 decay_factor 是否有效
if 0 < self.probability_decay_factor < 1:
decay_multiplier = math.pow(self.probability_decay_factor, time_delta)
old_prob = self.current_reply_probability
self.current_reply_probability *= decay_multiplier
# 避免因浮点数精度问题导致概率略微大于0，直接设为0
if self.current_reply_probability < 1e-6:
self.current_reply_probability = 0.0
logger.debug(f"兴趣低于阈值 ({self.trigger_threshold}) 持续 {time_delta:.2f}秒. 概率从 {old_prob:.4f} 衰减到 {self.current_reply_probability:.4f} (因子: {self.probability_decay_factor})")
elif self.probability_decay_factor <= 0:
# 如果衰减因子无效或为0，直接清零
if self.current_reply_probability > 0:
logger.warning(f"无效的衰减因子 ({self.probability_decay_factor}). 设置概率为0.")
self.current_reply_probability = 0.0
# else: decay_factor >= 1, probability will not decay or increase, which might be intended in some cases.
# 确保概率不低于0
self.current_reply_probability = max(self.current_reply_probability, 0.0)
# 更新状态标记
self.is_above_threshold = currently_above
# 更新时间戳放在调用者处，确保 interest 和 probability 基于同一点更新
def increase_interest(self, current_time: float, value: float):
"""根据传入的值增加兴趣值，并记录增加量"""
# 先更新概率和计算衰减（基于上次更新时间）
self._update_reply_probability(current_time)
self._calculate_decay(current_time)
# 记录这次增加的具体数值，供外部判断是否触发
self.last_increase_amount = value
self.last_triggering_message = message
# 应用增加
self.interest_level += value
self.interest_level = min(self.interest_level, self.max_interest) # 不超过最大值
self.last_update_time = current_time # 更新时间戳
self.last_interaction_time = current_time # 更新最后交互时间
def decrease_interest(self, current_time: float, value: float):
"""降低兴趣值并更新时间 (确保不低于0)"""
# 先更新概率（基于上次更新时间）
self._update_reply_probability(current_time)
# 注意：降低兴趣度是否需要先衰减？取决于具体逻辑，这里假设不衰减直接减
self.interest_level -= value
self.interest_level = max(self.interest_level, 0.0) # 确保不低于0
self.last_update_time = current_time # 降低也更新时间戳
self.last_interaction_time = current_time # 更新最后交互时间
def reset_trigger_info(self):
"""重置触发相关信息，在外部任务处理后调用"""
self.last_increase_amount = 0.0
self.last_triggering_message = None
def get_interest(self) -> float:
"""获取当前兴趣值 (由后台任务更新)"""
"""获取当前兴趣值 (计算衰减后)"""
# 注意：这个方法现在会触发概率和兴趣的更新
current_time = time.time()
self._update_reply_probability(current_time)
self._calculate_decay(current_time)
self.last_update_time = current_time # 更新时间戳
return self.interest_level
def get_state(self) -> dict:
"""获取当前状态字典"""
# 不再需要传入 current_time 来计算，直接获取
interest = self.get_interest() # 使用修改后的 get_interest
# 调用 get_interest 来确保状态已更新
interest = self.get_interest()
return {
"interest_level": round(interest, 2),
"last_update_time": self.last_update_time,
"current_reply_probability": round(self.current_reply_probability, 4), # 添加概率到状态
"is_above_threshold": self.is_above_threshold, # 添加阈值状态
"last_interaction_time": self.last_interaction_time # 新增：添加最后交互时间到状态
# 可以选择性地暴露 last_increase_amount 给状态，方便调试
# "last_increase_amount": round(self.last_increase_amount, 2)
}
def should_evaluate_reply(self) -> bool:
"""
判断是否应该触发一次回复评估。
首先更新概率状态，然后根据当前概率进行随机判断。
"""
current_time = time.time()
# 确保概率是基于最新兴趣值计算的
self._update_reply_probability(current_time)
# 更新兴趣衰减（如果需要，取决于逻辑，这里保持和 get_interest 一致）
self._calculate_decay(current_time)
self.last_update_time = current_time # 更新时间戳
if self.is_above_threshold and self.current_reply_probability > 0:
# 只有在阈值之上且概率大于0时才有可能触发
trigger = random.random() < self.current_reply_probability
if trigger:
logger.info(f"Reply evaluation triggered! Probability: {self.current_reply_probability:.4f}, Threshold: {self.trigger_threshold}, Interest: {self.interest_level:.2f}")
# 可选：触发后是否重置/降低概率？根据需要决定
# self.current_reply_probability = self.base_reply_probability # 例如，触发后降回基础概率
# self.current_reply_probability *= 0.5 # 例如，触发后概率减半
else:
logger.debug(f"Reply evaluation NOT triggered. Probability: {self.current_reply_probability:.4f}, Random value: {trigger + 1e-9:.4f}") # 打印随机值用于调试
return trigger
else:
# logger.debug(f"Reply evaluation check: Below threshold or zero probability. Probability: {self.current_reply_probability:.4f}")
return False
class InterestManager:
_instance = None
@@ -156,14 +275,14 @@ class InterestManager:
"""后台清理任务的异步函数"""
while True:
await asyncio.sleep(interval_seconds)
logger.info(f"Running periodic cleanup (interval: {interval_seconds}s)...")
logger.info(f"运行定期清理 (间隔: {interval_seconds}秒)...")
self.cleanup_inactive_chats(threshold=threshold, max_age_seconds=max_age_seconds)
async def _periodic_log_task(self, interval_seconds: int):
"""后台日志记录任务的异步函数 (记录历史数据，包含 group_name)"""
while True:
await asyncio.sleep(interval_seconds)
logger.debug(f"Running periodic history logging (interval: {interval_seconds}s)...")
logger.debug(f"运行定期历史记录 (间隔: {interval_seconds}秒)...")
try:
current_timestamp = time.time()
all_states = self.get_all_interest_states() # 获取当前所有状态
@@ -190,7 +309,11 @@ class InterestManager:
"timestamp": round(current_timestamp, 2),
"stream_id": stream_id,
"interest_level": state.get("interest_level", 0.0), # 确保有默认值
"group_name": group_name # *** Add group_name ***
"group_name": group_name, # *** Add group_name ***
# --- 新增：记录概率相关信息 ---
"reply_probability": state.get("current_reply_probability", 0.0),
"is_above_threshold": state.get("is_above_threshold", False)
# --- 结束新增 ---
}
# 将每个条目作为单独的 JSON 行写入
f.write(json.dumps(log_entry, ensure_ascii=False) + '\n')
@@ -230,7 +353,7 @@ class InterestManager:
# logger.debug(f"Applied decay to {count} streams.")
async def start_background_tasks(self):
"""Starts the background cleanup, logging, and decay tasks."""
"""启动清理，启动衰减，启动记录，启动启动启动启动启动"""
if self._cleanup_task is None or self._cleanup_task.done():
self._cleanup_task = asyncio.create_task(
self._periodic_cleanup_task(
@@ -239,26 +362,26 @@ class InterestManager:
max_age_seconds=INACTIVE_THRESHOLD_SECONDS
)
)
logger.info(f"Periodic cleanup task created. Interval: {CLEANUP_INTERVAL_SECONDS}s, Inactive Threshold: {INACTIVE_THRESHOLD_SECONDS}s")
logger.info(f"已创建定期清理任务。间隔时间: {CLEANUP_INTERVAL_SECONDS}秒, 不活跃阈值: {INACTIVE_THRESHOLD_SECONDS}秒")
else:
logger.warning("Cleanup task creation skipped: already running or exists.")
logger.warning("跳过创建清理任务:任务已在运行或存在。")
if self._logging_task is None or self._logging_task.done():
self._logging_task = asyncio.create_task(
self._periodic_log_task(interval_seconds=LOG_INTERVAL_SECONDS)
)
logger.info(f"Periodic logging task created. Interval: {LOG_INTERVAL_SECONDS}s")
logger.info(f"已创建定期日志任务。间隔时间: {LOG_INTERVAL_SECONDS}秒")
else:
logger.warning("Logging task creation skipped: already running or exists.")
logger.warning("跳过创建日志任务:任务已在运行或存在。")
# 启动新的衰减任务
if self._decay_task is None or self._decay_task.done():
self._decay_task = asyncio.create_task(
self._periodic_decay_task()
)
logger.info("Periodic decay task created. Interval: 1s")
logger.info("已创建定期衰减任务。间隔时间: 1秒")
else:
logger.warning("Decay task creation skipped: already running or exists.")
logger.warning("跳过创建衰减任务:任务已在运行或存在。")
def get_all_interest_states(self) -> dict[str, dict]:
"""获取所有聊天流的当前兴趣状态"""
@@ -287,7 +410,16 @@ class InterestManager:
# with self._lock:
if stream_id not in self.interest_dict:
logger.debug(f"Creating new InterestChatting for stream_id: {stream_id}")
self.interest_dict[stream_id] = InterestChatting()
# --- 修改：创建时传入概率相关参数 (如果需要定制化，否则使用默认值) ---
self.interest_dict[stream_id] = InterestChatting(
# decay_rate=..., max_interest=..., # 可以从配置读取
trigger_threshold=REPLY_TRIGGER_THRESHOLD, # 使用全局常量
base_reply_probability=BASE_REPLY_PROBABILITY,
increase_rate=PROBABILITY_INCREASE_RATE_PER_SECOND,
decay_factor=PROBABILITY_DECAY_FACTOR_PER_SECOND,
max_probability=MAX_REPLY_PROBABILITY
)
# --- 结束修改 ---
# 首次创建时兴趣为 0，由第一次消息的 activate rate 决定初始值
return self.interest_dict[stream_id]
@@ -298,13 +430,13 @@ class InterestManager:
# 直接调用修改后的 get_interest，不传入时间
return interest_chatting.get_interest()
def increase_interest(self, stream_id: str, value: float, message: MessageRecv):
"""当收到消息时，增加指定聊天流的兴趣度，并传递关联消息"""
def increase_interest(self, stream_id: str, value: float):
"""当收到消息时，增加指定聊天流的兴趣度"""
current_time = time.time()
interest_chatting = self._get_or_create_interest_chatting(stream_id)
# 调用修改后的 increase_interest，传入 message
interest_chatting.increase_interest(current_time, value, message)
logger.debug(f"Increased interest for stream_id: {stream_id} by {value:.2f} to {interest_chatting.interest_level:.2f}") # 更新日志
# 调用修改后的 increase_interest，不再传入 message
interest_chatting.increase_interest(current_time, value)
logger.debug(f"增加了聊天流 {stream_id} 的兴趣度 {value:.2f}，当前值为 {interest_chatting.interest_level:.2f}") # 更新日志
def decrease_interest(self, stream_id: str, value: float):
"""降低指定聊天流的兴趣度"""
@@ -313,13 +445,13 @@ class InterestManager:
interest_chatting = self.get_interest_chatting(stream_id)
if interest_chatting:
interest_chatting.decrease_interest(current_time, value)
logger.debug(f"Decreased interest for stream_id: {stream_id} by {value:.2f} to {interest_chatting.interest_level:.2f}")
logger.debug(f"降低了聊天流 {stream_id} 的兴趣度 {value:.2f}，当前值为 {interest_chatting.interest_level:.2f}")
else:
logger.warning(f"Attempted to decrease interest for non-existent stream_id: {stream_id}")
logger.warning(f"尝试降低不存在的聊天流 {stream_id} 的兴趣度")
def cleanup_inactive_chats(self, threshold=MIN_INTEREST_THRESHOLD, max_age_seconds=INACTIVE_THRESHOLD_SECONDS):
"""
清理长时间不活跃或兴趣度过低的聊天流记录
清理长时间不活跃的聊天流记录
threshold: 低于此兴趣度的将被清理
max_age_seconds: 超过此时间未更新的将被清理
"""
@@ -334,37 +466,27 @@ class InterestManager:
# 先计算当前兴趣，确保是最新的
# 加锁保护 chatting 对象状态的读取和可能的修改
# with self._lock: # 如果 InterestChatting 内部操作不是原子的
interest = chatting.get_interest()
last_update = chatting.last_update_time
last_interaction = chatting.last_interaction_time # 使用最后交互时间
should_remove = False
reason = ""
if interest < threshold:
should_remove = True
reason = f"interest ({interest:.2f}) < threshold ({threshold})"
# 只有设置了 max_age_seconds 才检查时间
if max_age_seconds is not None and (current_time - last_update) > max_age_seconds:
if max_age_seconds is not None and (current_time - last_interaction) > max_age_seconds: # 使用 last_interaction
should_remove = True
reason = f"inactive time ({current_time - last_update:.0f}s) > max age ({max_age_seconds}s)" + (f", {reason}" if reason else "") # 附加之前的理由
reason = f"inactive time ({current_time - last_interaction:.0f}s) > max age ({max_age_seconds}s)" # 更新日志信息
if should_remove:
keys_to_remove.append(stream_id)
logger.debug(f"Marking stream_id {stream_id} for removal. Reason: {reason}")
if keys_to_remove:
logger.info(f"Cleanup identified {len(keys_to_remove)} inactive/low-interest streams.")
logger.info(f"清理识别到 {len(keys_to_remove)} 个不活跃/低兴趣的流。")
# with self._lock: # 确保删除操作的原子性
for key in keys_to_remove:
# 再次检查 key 是否存在，以防万一在迭代和删除之间状态改变
if key in self.interest_dict:
del self.interest_dict[key]
logger.debug(f"Removed stream_id: {key}")
logger.debug(f"移除了流_id: {key}")
final_count = initial_count - len(keys_to_remove)
logger.info(f"Cleanup finished. Removed {len(keys_to_remove)} streams. Current count: {final_count}")
logger.info(f"清理完成。移除了 {len(keys_to_remove)} 个流。当前数量: {final_count}")
else:
logger.info(f"Cleanup finished. No streams met removal criteria. Current count: {initial_count}")
# 不再需要手动创建实例和任务
# manager = InterestManager()
# asyncio.create_task(periodic_cleanup(manager, 3600))
logger.info(f"清理完成。没有流符合移除条件。当前数量: {initial_count}")