remove：冗余的sbhf代码和focus代码
This commit is contained in:
SengokuCola
@@ -1,173 +0,0 @@
|
||||
import asyncio
|
||||
import traceback
|
||||
from typing import Optional, Coroutine, Callable, Any, List
|
||||
from src.common.logger import get_logger
|
||||
from src.chat.heart_flow.subheartflow_manager import SubHeartflowManager
|
||||
from src.config.config import global_config
|
||||
|
||||
logger = get_logger("background_tasks")
|
||||
|
||||
|
||||
# 新增私聊激活检查间隔
|
||||
PRIVATE_CHAT_ACTIVATION_CHECK_INTERVAL_SECONDS = 5 # 与兴趣评估类似，设为5秒
|
||||
|
||||
CLEANUP_INTERVAL_SECONDS = 1200
|
||||
|
||||
|
||||
async def _run_periodic_loop(
|
||||
task_name: str, interval: int, task_func: Callable[..., Coroutine[Any, Any, None]], **kwargs
|
||||
):
|
||||
"""周期性任务主循环"""
|
||||
while True:
|
||||
start_time = asyncio.get_event_loop().time()
|
||||
# logger.debug(f"开始执行后台任务: {task_name}")
|
||||
|
||||
try:
|
||||
await task_func(**kwargs) # 执行实际任务
|
||||
except asyncio.CancelledError:
|
||||
logger.info(f"任务 {task_name} 已取消")
|
||||
break
|
||||
except Exception as e:
|
||||
logger.error(f"任务 {task_name} 执行出错: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
|
||||
# 计算并执行间隔等待
|
||||
elapsed = asyncio.get_event_loop().time() - start_time
|
||||
sleep_time = max(0, interval - elapsed)
|
||||
# if sleep_time < 0.1: # 任务超时处理, DEBUG 时可能干扰断点
|
||||
# logger.warning(f"任务 {task_name} 超时执行 ({elapsed:.2f}s > {interval}s)")
|
||||
await asyncio.sleep(sleep_time)
|
||||
|
||||
logger.debug(f"任务循环结束: {task_name}") # 调整日志信息
|
||||
|
||||
|
||||
class BackgroundTaskManager:
|
||||
"""管理 Heartflow 的后台周期性任务。"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
subheartflow_manager: SubHeartflowManager,
|
||||
):
|
||||
self.subheartflow_manager = subheartflow_manager
|
||||
|
||||
# Task references
|
||||
self._cleanup_task: Optional[asyncio.Task] = None
|
||||
self._hf_judge_state_update_task: Optional[asyncio.Task] = None
|
||||
self._private_chat_activation_task: Optional[asyncio.Task] = None # 新增私聊激活任务引用
|
||||
self._tasks: List[Optional[asyncio.Task]] = [] # Keep track of all tasks
|
||||
|
||||
async def start_tasks(self):
|
||||
"""启动所有后台任务
|
||||
|
||||
功能说明:
|
||||
- 启动核心后台任务: 状态更新、清理、日志记录、兴趣评估和随机停用
|
||||
- 每个任务启动前检查是否已在运行
|
||||
- 将任务引用保存到任务列表
|
||||
"""
|
||||
|
||||
task_configs = []
|
||||
|
||||
# 根据 chat_mode 条件添加其他任务
|
||||
if not (global_config.chat.chat_mode == "normal"):
|
||||
task_configs.extend(
|
||||
[
|
||||
(
|
||||
self._run_cleanup_cycle,
|
||||
"info",
|
||||
f"清理任务已启动 间隔:{CLEANUP_INTERVAL_SECONDS}s",
|
||||
"_cleanup_task",
|
||||
),
|
||||
# 新增私聊激活任务配置
|
||||
(
|
||||
# Use lambda to pass the interval to the runner function
|
||||
lambda: self._run_private_chat_activation_cycle(PRIVATE_CHAT_ACTIVATION_CHECK_INTERVAL_SECONDS),
|
||||
"debug",
|
||||
f"私聊激活检查任务已启动 间隔:{PRIVATE_CHAT_ACTIVATION_CHECK_INTERVAL_SECONDS}s",
|
||||
"_private_chat_activation_task",
|
||||
),
|
||||
]
|
||||
)
|
||||
|
||||
# 统一启动所有任务
|
||||
for task_func, log_level, log_msg, task_attr_name in task_configs:
|
||||
# 检查任务变量是否存在且未完成
|
||||
current_task_var = getattr(self, task_attr_name)
|
||||
if current_task_var is None or current_task_var.done():
|
||||
new_task = asyncio.create_task(task_func())
|
||||
setattr(self, task_attr_name, new_task) # 更新任务变量
|
||||
if new_task not in self._tasks: # 避免重复添加
|
||||
self._tasks.append(new_task)
|
||||
|
||||
# 根据配置记录不同级别的日志
|
||||
getattr(logger, log_level)(log_msg)
|
||||
else:
|
||||
logger.warning(f"{task_attr_name}任务已在运行")
|
||||
|
||||
async def stop_tasks(self):
|
||||
"""停止所有后台任务。
|
||||
|
||||
该方法会:
|
||||
1. 遍历所有后台任务并取消未完成的任务
|
||||
2. 等待所有取消操作完成
|
||||
3. 清空任务列表
|
||||
"""
|
||||
logger.info("正在停止所有后台任务...")
|
||||
cancelled_count = 0
|
||||
|
||||
# 第一步：取消所有运行中的任务
|
||||
for task in self._tasks:
|
||||
if task and not task.done():
|
||||
task.cancel() # 发送取消请求
|
||||
cancelled_count += 1
|
||||
|
||||
# 第二步：处理取消结果
|
||||
if cancelled_count > 0:
|
||||
logger.debug(f"正在等待{cancelled_count}个任务完成取消...")
|
||||
# 使用gather等待所有取消操作完成，忽略异常
|
||||
await asyncio.gather(*[t for t in self._tasks if t and t.cancelled()], return_exceptions=True)
|
||||
logger.info(f"成功取消{cancelled_count}个后台任务")
|
||||
else:
|
||||
logger.info("没有需要取消的后台任务")
|
||||
|
||||
# 第三步：清空任务列表
|
||||
self._tasks = [] # 重置任务列表
|
||||
|
||||
# 状态转换处理
|
||||
|
||||
async def _perform_cleanup_work(self):
|
||||
"""执行子心流清理任务
|
||||
1. 获取需要清理的不活跃子心流列表
|
||||
2. 逐个停止这些子心流
|
||||
3. 记录清理结果
|
||||
"""
|
||||
# 获取需要清理的子心流列表(包含ID和原因)
|
||||
flows_to_stop = self.subheartflow_manager.get_inactive_subheartflows()
|
||||
|
||||
if not flows_to_stop:
|
||||
return # 没有需要清理的子心流直接返回
|
||||
|
||||
logger.info(f"准备删除 {len(flows_to_stop)} 个不活跃(1h)子心流")
|
||||
stopped_count = 0
|
||||
|
||||
# 逐个停止子心流
|
||||
for flow_id in flows_to_stop:
|
||||
success = await self.subheartflow_manager.delete_subflow(flow_id)
|
||||
if success:
|
||||
stopped_count += 1
|
||||
logger.debug(f"[清理任务] 已停止子心流 {flow_id}")
|
||||
|
||||
# 记录最终清理结果
|
||||
logger.info(f"[清理任务] 清理完成, 共停止 {stopped_count}/{len(flows_to_stop)} 个子心流")
|
||||
|
||||
async def _run_cleanup_cycle(self):
|
||||
await _run_periodic_loop(
|
||||
task_name="Subflow Cleanup", interval=CLEANUP_INTERVAL_SECONDS, task_func=self._perform_cleanup_work
|
||||
)
|
||||
|
||||
# 新增私聊激活任务运行器
|
||||
async def _run_private_chat_activation_cycle(self, interval: int):
|
||||
await _run_periodic_loop(
|
||||
task_name="Private Chat Activation Check",
|
||||
interval=interval,
|
||||
task_func=self.subheartflow_manager.sbhf_absent_private_into_focus,
|
||||
)
|
||||
@@ -1,84 +1,56 @@
|
||||
from src.chat.heart_flow.sub_heartflow import SubHeartflow, ChatState
|
||||
from src.common.logger import get_logger
|
||||
from typing import Any, Optional, List
|
||||
from src.chat.heart_flow.subheartflow_manager import SubHeartflowManager
|
||||
from src.chat.heart_flow.background_tasks import BackgroundTaskManager # Import BackgroundTaskManager
|
||||
|
||||
from typing import Any, Optional
|
||||
from typing import Dict
|
||||
from src.chat.message_receive.chat_stream import get_chat_manager
|
||||
logger = get_logger("heartflow")
|
||||
|
||||
|
||||
class Heartflow:
|
||||
"""主心流协调器，负责初始化并协调各个子系统:
|
||||
- 状态管理 (MaiState)
|
||||
- 子心流管理 (SubHeartflow)
|
||||
- 后台任务 (BackgroundTaskManager)
|
||||
"""
|
||||
"""主心流协调器，负责初始化并协调聊天"""
|
||||
|
||||
def __init__(self):
|
||||
# 子心流管理 (在初始化时传入 current_state)
|
||||
self.subheartflow_manager: SubHeartflowManager = SubHeartflowManager()
|
||||
|
||||
# 后台任务管理器 (整合所有定时任务)
|
||||
self.background_task_manager: BackgroundTaskManager = BackgroundTaskManager(
|
||||
subheartflow_manager=self.subheartflow_manager,
|
||||
)
|
||||
self.subheartflows: Dict[Any, "SubHeartflow"] = {}
|
||||
|
||||
async def get_or_create_subheartflow(self, subheartflow_id: Any) -> Optional["SubHeartflow"]:
|
||||
"""获取或创建一个新的SubHeartflow实例 - 委托给 SubHeartflowManager"""
|
||||
# 不再需要传入 self.current_state
|
||||
return await self.subheartflow_manager.get_or_create_subheartflow(subheartflow_id)
|
||||
"""获取或创建一个新的SubHeartflow实例"""
|
||||
if subheartflow_id in self.subheartflows:
|
||||
subflow = self.subheartflows.get(subheartflow_id)
|
||||
if subflow:
|
||||
return subflow
|
||||
|
||||
try:
|
||||
new_subflow = SubHeartflow(
|
||||
subheartflow_id,
|
||||
)
|
||||
|
||||
await new_subflow.initialize()
|
||||
|
||||
# 注册子心流
|
||||
self.subheartflows[subheartflow_id] = new_subflow
|
||||
heartflow_name = get_chat_manager().get_stream_name(subheartflow_id) or subheartflow_id
|
||||
logger.info(f"[{heartflow_name}] 开始接收消息")
|
||||
|
||||
return new_subflow
|
||||
except Exception as e:
|
||||
logger.error(f"创建子心流 {subheartflow_id} 失败: {e}", exc_info=True)
|
||||
return None
|
||||
|
||||
|
||||
async def force_change_subheartflow_status(self, subheartflow_id: str, status: ChatState) -> None:
|
||||
"""强制改变子心流的状态"""
|
||||
# 这里的 message 是可选的，可能是一个消息对象，也可能是其他类型的数据
|
||||
return await self.subheartflow_manager.force_change_state(subheartflow_id, status)
|
||||
|
||||
async def api_get_all_states(self):
|
||||
"""获取所有状态"""
|
||||
return await self.interest_logger.api_get_all_states()
|
||||
|
||||
async def api_get_subheartflow_cycle_info(self, subheartflow_id: str, history_len: int) -> Optional[dict]:
|
||||
"""获取子心流的循环信息"""
|
||||
subheartflow = await self.subheartflow_manager.get_or_create_subheartflow(subheartflow_id)
|
||||
if not subheartflow:
|
||||
logger.warning(f"尝试获取不存在的子心流 {subheartflow_id} 的周期信息")
|
||||
return None
|
||||
heartfc_instance = subheartflow.heart_fc_instance
|
||||
if not heartfc_instance:
|
||||
logger.warning(f"子心流 {subheartflow_id} 没有心流实例，无法获取周期信息")
|
||||
return None
|
||||
|
||||
return heartfc_instance.get_cycle_history(last_n=history_len)
|
||||
|
||||
async def api_get_normal_chat_replies(self, subheartflow_id: str, limit: int = 10) -> Optional[List[dict]]:
|
||||
"""获取子心流的NormalChat回复记录
|
||||
|
||||
Args:
|
||||
subheartflow_id: 子心流ID
|
||||
limit: 最大返回数量，默认10条
|
||||
|
||||
Returns:
|
||||
Optional[List[dict]]: 回复记录列表，如果子心流不存在则返回None
|
||||
"""
|
||||
subheartflow = await self.subheartflow_manager.get_or_create_subheartflow(subheartflow_id)
|
||||
if not subheartflow:
|
||||
logger.warning(f"尝试获取不存在的子心流 {subheartflow_id} 的NormalChat回复记录")
|
||||
return None
|
||||
|
||||
return subheartflow.get_normal_chat_recent_replies(limit)
|
||||
|
||||
async def heartflow_start_working(self):
|
||||
"""启动后台任务"""
|
||||
await self.background_task_manager.start_tasks()
|
||||
logger.info("[Heartflow] 后台任务已启动")
|
||||
|
||||
# 根本不会用到这个函数吧，那样麦麦直接死了
|
||||
async def stop_working(self):
|
||||
"""停止所有任务和子心流"""
|
||||
logger.info("[Heartflow] 正在停止任务和子心流...")
|
||||
await self.background_task_manager.stop_tasks()
|
||||
await self.subheartflow_manager.deactivate_all_subflows()
|
||||
logger.info("[Heartflow] 所有任务和子心流已停止")
|
||||
return await self.force_change_state(subheartflow_id, status)
|
||||
|
||||
async def force_change_state(self, subflow_id: Any, target_state: ChatState) -> bool:
|
||||
"""强制改变指定子心流的状态"""
|
||||
subflow = self.subheartflows.get(subflow_id)
|
||||
if not subflow:
|
||||
logger.warning(f"[强制状态转换]尝试转换不存在的子心流{subflow_id} 到 {target_state.value}")
|
||||
return False
|
||||
await subflow.change_chat_state(target_state)
|
||||
logger.info(f"[强制状态转换]子心流 {subflow_id} 已转换到 {target_state.value}")
|
||||
return True
|
||||
|
||||
|
||||
heartflow = Heartflow()
|
||||
|
||||
@@ -10,29 +10,13 @@ from src.common.logger import get_logger
|
||||
import re
|
||||
import math
|
||||
import traceback
|
||||
from typing import Optional, Tuple
|
||||
from typing import Tuple
|
||||
|
||||
from src.person_info.relationship_manager import get_relationship_manager
|
||||
|
||||
# from ..message_receive.message_buffer import message_buffer
|
||||
|
||||
logger = get_logger("chat")
|
||||
|
||||
|
||||
async def _handle_error(error: Exception, context: str, message: Optional[MessageRecv] = None) -> None:
|
||||
"""统一的错误处理函数
|
||||
|
||||
Args:
|
||||
error: 捕获到的异常
|
||||
context: 错误发生的上下文描述
|
||||
message: 可选的消息对象，用于记录相关消息内容
|
||||
"""
|
||||
logger.error(f"{context}: {error}")
|
||||
logger.error(traceback.format_exc())
|
||||
if message and hasattr(message, "raw_message"):
|
||||
logger.error(f"相关消息原始内容: {message.raw_message}")
|
||||
|
||||
|
||||
async def _process_relationship(message: MessageRecv) -> None:
|
||||
"""处理用户关系逻辑
|
||||
|
||||
@@ -149,4 +133,5 @@ class HeartFCMessageReceiver:
|
||||
await _process_relationship(message)
|
||||
|
||||
except Exception as e:
|
||||
await _handle_error(e, "消息处理失败", message)
|
||||
logger.error(f"消息处理失败: {e}")
|
||||
print(traceback.format_exc())
|
||||
|
||||
@@ -44,10 +44,6 @@ class SubHeartflow:
|
||||
# 兴趣消息集合
|
||||
self.interest_dict: Dict[str, tuple[MessageRecv, float, bool]] = {}
|
||||
|
||||
# 活动状态管理
|
||||
self.should_stop = False # 停止标志
|
||||
self.task: Optional[asyncio.Task] = None # 后台任务
|
||||
|
||||
# focus模式退出冷却时间管理
|
||||
self.last_focus_exit_time: float = 0 # 上次退出focus模式的时间
|
||||
|
||||
@@ -211,10 +207,6 @@ class SubHeartflow:
|
||||
await asyncio.wait_for(self.heart_fc_instance.start(), timeout=15.0)
|
||||
logger.info(f"{log_prefix} HeartFChatting 循环已启动。")
|
||||
return True
|
||||
except asyncio.TimeoutError:
|
||||
logger.error(f"{log_prefix} 启动现有 HeartFChatting 循环超时")
|
||||
# 超时时清理实例，准备重新创建
|
||||
self.heart_fc_instance = None
|
||||
except Exception as e:
|
||||
logger.error(f"{log_prefix} 尝试启动现有 HeartFChatting 循环时出错: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
@@ -231,7 +223,6 @@ class SubHeartflow:
|
||||
logger.debug(f"{log_prefix} 创建新的 HeartFChatting 实例")
|
||||
self.heart_fc_instance = HeartFChatting(
|
||||
chat_id=self.subheartflow_id,
|
||||
# observations=self.observations,
|
||||
on_stop_focus_chat=self._handle_stop_focus_chat_request,
|
||||
)
|
||||
|
||||
@@ -241,10 +232,6 @@ class SubHeartflow:
|
||||
logger.debug(f"{log_prefix} 麦麦已成功进入专注聊天模式 (新实例已启动)。")
|
||||
return True
|
||||
|
||||
except asyncio.TimeoutError:
|
||||
logger.error(f"{log_prefix} 创建或启动新 HeartFChatting 实例超时")
|
||||
self.heart_fc_instance = None # 超时时清理实例
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.error(f"{log_prefix} 创建或启动 HeartFChatting 实例时出错: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
@@ -255,8 +242,6 @@ class SubHeartflow:
|
||||
logger.error(f"{self.log_prefix} _start_heart_fc_chat 执行时出错: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
return False
|
||||
finally:
|
||||
logger.debug(f"{self.log_prefix} _start_heart_fc_chat 完成")
|
||||
|
||||
async def change_chat_state(self, new_state: ChatState) -> None:
|
||||
"""
|
||||
@@ -312,25 +297,6 @@ class SubHeartflow:
|
||||
f"{log_prefix} 尝试将状态从 {current_state.value} 变为 {new_state.value}，但未成功或未执行更改。"
|
||||
)
|
||||
|
||||
|
||||
def get_normal_chat_last_speak_time(self) -> float:
|
||||
if self.normal_chat_instance:
|
||||
return self.normal_chat_instance.last_speak_time
|
||||
return 0
|
||||
|
||||
def get_normal_chat_recent_replies(self, limit: int = 10) -> List[dict]:
|
||||
"""获取NormalChat实例的最近回复记录
|
||||
|
||||
Args:
|
||||
limit: 最大返回数量，默认10条
|
||||
|
||||
Returns:
|
||||
List[dict]: 最近的回复记录列表，如果没有NormalChat实例则返回空列表
|
||||
"""
|
||||
if self.normal_chat_instance:
|
||||
return self.normal_chat_instance.get_recent_replies(limit)
|
||||
return []
|
||||
|
||||
def add_message_to_normal_chat_cache(self, message: MessageRecv, interest_value: float, is_mentioned: bool):
|
||||
self.interest_dict[message.message_info.message_id] = (message, interest_value, is_mentioned)
|
||||
# 如果字典长度超过10，删除最旧的消息
|
||||
@@ -338,55 +304,6 @@ class SubHeartflow:
|
||||
oldest_key = next(iter(self.interest_dict))
|
||||
self.interest_dict.pop(oldest_key)
|
||||
|
||||
def get_normal_chat_action_manager(self):
|
||||
"""获取NormalChat的ActionManager实例
|
||||
|
||||
Returns:
|
||||
ActionManager: NormalChat的ActionManager实例，如果不存在则返回None
|
||||
"""
|
||||
if self.normal_chat_instance:
|
||||
return self.normal_chat_instance.get_action_manager()
|
||||
return None
|
||||
|
||||
async def get_full_state(self) -> dict:
|
||||
"""获取子心流的完整状态，包括兴趣、思维和聊天状态。"""
|
||||
return {
|
||||
"interest_state": "interest_state",
|
||||
"chat_state": self.chat_state.chat_status.value,
|
||||
"chat_state_changed_time": self.chat_state_changed_time,
|
||||
}
|
||||
|
||||
async def shutdown(self):
|
||||
"""安全地关闭子心流及其管理的任务"""
|
||||
if self.should_stop:
|
||||
logger.info(f"{self.log_prefix} 子心流已在关闭过程中。")
|
||||
return
|
||||
|
||||
logger.info(f"{self.log_prefix} 开始关闭子心流...")
|
||||
self.should_stop = True # 标记为停止，让后台任务退出
|
||||
|
||||
# 使用新的停止方法
|
||||
await self._stop_normal_chat()
|
||||
await self._stop_heart_fc_chat()
|
||||
|
||||
# 取消可能存在的旧后台任务 (self.task)
|
||||
if self.task and not self.task.done():
|
||||
logger.debug(f"{self.log_prefix} 取消子心流主任务 (Shutdown)...")
|
||||
self.task.cancel()
|
||||
try:
|
||||
await asyncio.wait_for(self.task, timeout=1.0) # 给点时间响应取消
|
||||
except asyncio.CancelledError:
|
||||
logger.debug(f"{self.log_prefix} 子心流主任务已取消 (Shutdown)。")
|
||||
except asyncio.TimeoutError:
|
||||
logger.warning(f"{self.log_prefix} 等待子心流主任务取消超时 (Shutdown)。")
|
||||
except Exception as e:
|
||||
logger.error(f"{self.log_prefix} 等待子心流主任务取消时发生错误 (Shutdown): {e}")
|
||||
|
||||
self.task = None # 清理任务引用
|
||||
self.chat_state.chat_status = ChatState.ABSENT # 状态重置为不参与
|
||||
|
||||
logger.info(f"{self.log_prefix} 子心流关闭完成。")
|
||||
|
||||
def is_in_focus_cooldown(self) -> bool:
|
||||
"""检查是否在focus模式的冷却期内
|
||||
|
||||
|
||||
@@ -1,337 +0,0 @@
|
||||
import asyncio
|
||||
import time
|
||||
from typing import Dict, Any, Optional, List
|
||||
from src.common.logger import get_logger
|
||||
from src.chat.message_receive.chat_stream import get_chat_manager
|
||||
from src.chat.heart_flow.sub_heartflow import SubHeartflow, ChatState
|
||||
|
||||
|
||||
# 初始化日志记录器
|
||||
|
||||
logger = get_logger("subheartflow_manager")
|
||||
|
||||
# 子心流管理相关常量
|
||||
INACTIVE_THRESHOLD_SECONDS = 3600 # 子心流不活跃超时时间(秒)
|
||||
NORMAL_CHAT_TIMEOUT_SECONDS = 30 * 60 # 30分钟
|
||||
|
||||
|
||||
async def _try_set_subflow_absent_internal(subflow: "SubHeartflow", log_prefix: str) -> bool:
|
||||
"""
|
||||
尝试将给定的子心流对象状态设置为 ABSENT (内部方法，不处理锁)。
|
||||
|
||||
Args:
|
||||
subflow: 子心流对象。
|
||||
log_prefix: 用于日志记录的前缀 (例如 "[子心流管理]" 或 "[停用]")。
|
||||
|
||||
Returns:
|
||||
bool: 如果状态成功变为 ABSENT 或原本就是 ABSENT，返回 True；否则返回 False。
|
||||
"""
|
||||
flow_id = subflow.subheartflow_id
|
||||
stream_name = get_chat_manager().get_stream_name(flow_id) or flow_id
|
||||
|
||||
if subflow.chat_state.chat_status != ChatState.ABSENT:
|
||||
logger.debug(f"{log_prefix} 设置 {stream_name} 状态为 ABSENT")
|
||||
try:
|
||||
await subflow.change_chat_state(ChatState.ABSENT)
|
||||
# 再次检查以确认状态已更改 (change_chat_state 内部应确保)
|
||||
if subflow.chat_state.chat_status == ChatState.ABSENT:
|
||||
return True
|
||||
else:
|
||||
logger.warning(
|
||||
f"{log_prefix} 调用 change_chat_state 后，{stream_name} 状态仍为 {subflow.chat_state.chat_status.value}"
|
||||
)
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.error(f"{log_prefix} 设置 {stream_name} 状态为 ABSENT 时失败: {e}", exc_info=True)
|
||||
return False
|
||||
else:
|
||||
logger.debug(f"{log_prefix} {stream_name} 已是 ABSENT 状态")
|
||||
return True # 已经是目标状态，视为成功
|
||||
|
||||
|
||||
class SubHeartflowManager:
|
||||
"""管理所有活跃的 SubHeartflow 实例。"""
|
||||
|
||||
def __init__(self):
|
||||
self.subheartflows: Dict[Any, "SubHeartflow"] = {}
|
||||
self._lock = asyncio.Lock() # 用于保护 self.subheartflows 的访问
|
||||
|
||||
async def force_change_state(self, subflow_id: Any, target_state: ChatState) -> bool:
|
||||
"""强制改变指定子心流的状态"""
|
||||
async with self._lock:
|
||||
subflow = self.subheartflows.get(subflow_id)
|
||||
if not subflow:
|
||||
logger.warning(f"[强制状态转换]尝试转换不存在的子心流{subflow_id} 到 {target_state.value}")
|
||||
return False
|
||||
await subflow.change_chat_state(target_state)
|
||||
logger.info(f"[强制状态转换]子心流 {subflow_id} 已转换到 {target_state.value}")
|
||||
return True
|
||||
|
||||
def get_all_subheartflows(self) -> List["SubHeartflow"]:
|
||||
"""获取所有当前管理的 SubHeartflow 实例列表 (快照)。"""
|
||||
return list(self.subheartflows.values())
|
||||
|
||||
async def get_or_create_subheartflow(self, subheartflow_id: Any) -> Optional["SubHeartflow"]:
|
||||
"""获取或创建指定ID的子心流实例
|
||||
|
||||
Args:
|
||||
subheartflow_id: 子心流唯一标识符
|
||||
mai_states 参数已被移除，使用 self.mai_state_info
|
||||
|
||||
Returns:
|
||||
成功返回SubHeartflow实例，失败返回None
|
||||
"""
|
||||
async with self._lock:
|
||||
# 检查是否已存在该子心流
|
||||
if subheartflow_id in self.subheartflows:
|
||||
subflow = self.subheartflows[subheartflow_id]
|
||||
if subflow.should_stop:
|
||||
logger.warning(f"尝试获取已停止的子心流 {subheartflow_id}，正在重新激活")
|
||||
subflow.should_stop = False # 重置停止标志
|
||||
return subflow
|
||||
|
||||
try:
|
||||
new_subflow = SubHeartflow(
|
||||
subheartflow_id,
|
||||
)
|
||||
|
||||
# 然后再进行异步初始化，此时 SubHeartflow 内部若需启动 HeartFChatting，就能拿到 observation
|
||||
await new_subflow.initialize()
|
||||
|
||||
# 注册子心流
|
||||
self.subheartflows[subheartflow_id] = new_subflow
|
||||
heartflow_name = get_chat_manager().get_stream_name(subheartflow_id) or subheartflow_id
|
||||
logger.info(f"[{heartflow_name}] 开始接收消息")
|
||||
|
||||
return new_subflow
|
||||
except Exception as e:
|
||||
logger.error(f"创建子心流 {subheartflow_id} 失败: {e}", exc_info=True)
|
||||
return None
|
||||
|
||||
async def sleep_subheartflow(self, subheartflow_id: Any, reason: str) -> bool:
|
||||
"""停止指定的子心流并将其状态设置为 ABSENT"""
|
||||
log_prefix = "[子心流管理]"
|
||||
async with self._lock: # 加锁以安全访问字典
|
||||
subheartflow = self.subheartflows.get(subheartflow_id)
|
||||
|
||||
stream_name = get_chat_manager().get_stream_name(subheartflow_id) or subheartflow_id
|
||||
logger.info(f"{log_prefix} 正在停止 {stream_name}, 原因: {reason}")
|
||||
|
||||
# 调用内部方法处理状态变更
|
||||
success = await _try_set_subflow_absent_internal(subheartflow, log_prefix)
|
||||
|
||||
return success
|
||||
# 锁在此处自动释放
|
||||
|
||||
def get_inactive_subheartflows(self, max_age_seconds=INACTIVE_THRESHOLD_SECONDS):
|
||||
"""识别并返回需要清理的不活跃(处于ABSENT状态超过一小时)子心流(id, 原因)"""
|
||||
_current_time = time.time()
|
||||
flows_to_stop = []
|
||||
|
||||
for subheartflow_id, subheartflow in list(self.subheartflows.items()):
|
||||
state = subheartflow.chat_state.chat_status
|
||||
if state != ChatState.ABSENT:
|
||||
continue
|
||||
subheartflow.update_last_chat_state_time()
|
||||
_absent_last_time = subheartflow.chat_state_last_time
|
||||
flows_to_stop.append(subheartflow_id)
|
||||
|
||||
return flows_to_stop
|
||||
|
||||
async def deactivate_all_subflows(self):
|
||||
"""将所有子心流的状态更改为 ABSENT (例如主状态变为OFFLINE时调用)"""
|
||||
log_prefix = "[停用]"
|
||||
changed_count = 0
|
||||
processed_count = 0
|
||||
|
||||
async with self._lock: # 获取锁以安全迭代
|
||||
# 使用 list() 创建一个当前值的快照，防止在迭代时修改字典
|
||||
flows_to_update = list(self.subheartflows.values())
|
||||
processed_count = len(flows_to_update)
|
||||
if not flows_to_update:
|
||||
logger.debug(f"{log_prefix} 无活跃子心流，无需操作")
|
||||
return
|
||||
|
||||
for subflow in flows_to_update:
|
||||
# 记录原始状态，以便统计实际改变的数量
|
||||
original_state_was_absent = subflow.chat_state.chat_status == ChatState.ABSENT
|
||||
|
||||
success = await _try_set_subflow_absent_internal(subflow, log_prefix)
|
||||
|
||||
# 如果成功设置为 ABSENT 且原始状态不是 ABSENT，则计数
|
||||
if success and not original_state_was_absent:
|
||||
if subflow.chat_state.chat_status == ChatState.ABSENT:
|
||||
changed_count += 1
|
||||
else:
|
||||
# 这种情况理论上不应发生，如果内部方法返回 True 的话
|
||||
stream_name = (
|
||||
get_chat_manager().get_stream_name(subflow.subheartflow_id) or subflow.subheartflow_id
|
||||
)
|
||||
logger.warning(f"{log_prefix} 内部方法声称成功但 {stream_name} 状态未变为 ABSENT。")
|
||||
# 锁在此处自动释放
|
||||
|
||||
logger.info(
|
||||
f"{log_prefix} 完成，共处理 {processed_count} 个子心流，成功将 {changed_count} 个非 ABSENT 子心流的状态更改为 ABSENT。"
|
||||
)
|
||||
|
||||
# async def sbhf_normal_into_focus(self):
|
||||
# """评估子心流兴趣度，满足条件则提升到FOCUSED状态（基于start_hfc_probability）"""
|
||||
# try:
|
||||
# for sub_hf in list(self.subheartflows.values()):
|
||||
# flow_id = sub_hf.subheartflow_id
|
||||
# stream_name = get_chat_manager().get_stream_name(flow_id) or flow_id
|
||||
|
||||
# # 跳过已经是FOCUSED状态的子心流
|
||||
# if sub_hf.chat_state.chat_status == ChatState.FOCUSED:
|
||||
# continue
|
||||
|
||||
# if sub_hf.interest_chatting.start_hfc_probability == 0:
|
||||
# continue
|
||||
# else:
|
||||
# logger.debug(
|
||||
# f"{stream_name}，现在状态: {sub_hf.chat_state.chat_status.value}，进入专注概率: {sub_hf.interest_chatting.start_hfc_probability}"
|
||||
# )
|
||||
|
||||
# if random.random() >= sub_hf.interest_chatting.start_hfc_probability:
|
||||
# continue
|
||||
|
||||
# # 获取最新状态并执行提升
|
||||
# current_subflow = self.subheartflows.get(flow_id)
|
||||
# if not current_subflow:
|
||||
# continue
|
||||
|
||||
# logger.info(
|
||||
# f"{stream_name} 触发 认真水群 (概率={current_subflow.interest_chatting.start_hfc_probability:.2f})"
|
||||
# )
|
||||
|
||||
# # 执行状态提升
|
||||
# await current_subflow.change_chat_state(ChatState.FOCUSED)
|
||||
|
||||
# except Exception as e:
|
||||
# logger.error(f"启动HFC 兴趣评估失败: {e}", exc_info=True)
|
||||
|
||||
async def sbhf_focus_into_normal(self, subflow_id: Any):
|
||||
"""
|
||||
接收来自 HeartFChatting 的请求，将特定子心流的状态转换为 NORMAL。
|
||||
通常在连续多次 "no_reply" 后被调用。
|
||||
对于私聊和群聊，都转换为 NORMAL。
|
||||
|
||||
Args:
|
||||
subflow_id: 需要转换状态的子心流 ID。
|
||||
"""
|
||||
async with self._lock:
|
||||
subflow = self.subheartflows.get(subflow_id)
|
||||
if not subflow:
|
||||
logger.warning(f"[状态转换请求] 尝试转换不存在的子心流 {subflow_id} 到 NORMAL")
|
||||
return
|
||||
|
||||
stream_name = get_chat_manager().get_stream_name(subflow_id) or subflow_id
|
||||
current_state = subflow.chat_state.chat_status
|
||||
|
||||
if current_state == ChatState.FOCUSED:
|
||||
target_state = ChatState.NORMAL
|
||||
log_reason = "转为NORMAL"
|
||||
|
||||
logger.info(
|
||||
f"[状态转换请求] 接收到请求，将 {stream_name} (当前: {current_state.value}) 尝试转换为 {target_state.value} ({log_reason})"
|
||||
)
|
||||
try:
|
||||
# 从HFC到CHAT时，清空兴趣字典
|
||||
subflow.interest_dict.clear()
|
||||
await subflow.change_chat_state(target_state)
|
||||
final_state = subflow.chat_state.chat_status
|
||||
if final_state == target_state:
|
||||
logger.debug(f"[状态转换请求] {stream_name} 状态已成功转换为 {final_state.value}")
|
||||
else:
|
||||
logger.warning(
|
||||
f"[状态转换请求] 尝试将 {stream_name} 转换为 {target_state.value} 后，状态实际为 {final_state.value}"
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"[状态转换请求] 转换 {stream_name} 到 {target_state.value} 时出错: {e}", exc_info=True
|
||||
)
|
||||
elif current_state == ChatState.ABSENT:
|
||||
logger.debug(f"[状态转换请求] {stream_name} 处于 ABSENT 状态，尝试转为 NORMAL")
|
||||
await subflow.change_chat_state(ChatState.NORMAL)
|
||||
else:
|
||||
logger.debug(f"[状态转换请求] {stream_name} 当前状态为 {current_state.value}，无需转换")
|
||||
|
||||
async def delete_subflow(self, subheartflow_id: Any):
|
||||
"""删除指定的子心流。"""
|
||||
async with self._lock:
|
||||
subflow = self.subheartflows.pop(subheartflow_id, None)
|
||||
if subflow:
|
||||
logger.info(f"正在删除 SubHeartflow: {subheartflow_id}...")
|
||||
try:
|
||||
# 调用 shutdown 方法确保资源释放
|
||||
await subflow.shutdown()
|
||||
logger.info(f"SubHeartflow {subheartflow_id} 已成功删除。")
|
||||
except Exception as e:
|
||||
logger.error(f"删除 SubHeartflow {subheartflow_id} 时出错: {e}", exc_info=True)
|
||||
else:
|
||||
logger.warning(f"尝试删除不存在的 SubHeartflow: {subheartflow_id}")
|
||||
|
||||
# --- 新增：处理私聊从 ABSENT 直接到 FOCUSED 的逻辑 --- #
|
||||
async def sbhf_absent_private_into_focus(self):
|
||||
"""检查 ABSENT 状态的私聊子心流是否有新活动，若有则直接转换为 FOCUSED。"""
|
||||
log_prefix_task = "[私聊激活检查]"
|
||||
transitioned_count = 0
|
||||
checked_count = 0
|
||||
|
||||
async with self._lock:
|
||||
# --- 筛选出所有 ABSENT 状态的私聊子心流 --- #
|
||||
eligible_subflows = [
|
||||
hf
|
||||
for hf in self.subheartflows.values()
|
||||
if hf.chat_state.chat_status == ChatState.ABSENT and not hf.is_group_chat
|
||||
]
|
||||
checked_count = len(eligible_subflows)
|
||||
|
||||
if not eligible_subflows:
|
||||
# logger.debug(f"{log_prefix_task} 没有 ABSENT 状态的私聊子心流可以评估。")
|
||||
return
|
||||
|
||||
# --- 遍历评估每个符合条件的私聊 --- #
|
||||
for sub_hf in eligible_subflows:
|
||||
flow_id = sub_hf.subheartflow_id
|
||||
stream_name = get_chat_manager().get_stream_name(flow_id) or flow_id
|
||||
log_prefix = f"[{stream_name}]({log_prefix_task})"
|
||||
|
||||
try:
|
||||
# --- 检查是否有新活动 --- #
|
||||
observation = sub_hf._get_primary_observation() # 获取主要观察者
|
||||
is_active = False
|
||||
if observation:
|
||||
# 检查自上次状态变为 ABSENT 后是否有新消息
|
||||
# 使用 chat_state_changed_time 可能更精确
|
||||
# 加一点点缓冲时间（例如 1 秒）以防时间戳完全相等
|
||||
timestamp_to_check = sub_hf.chat_state_changed_time - 1
|
||||
has_new = await observation.has_new_messages_since(timestamp_to_check)
|
||||
if has_new:
|
||||
is_active = True
|
||||
logger.debug(f"{log_prefix} 检测到新消息，标记为活跃。")
|
||||
else:
|
||||
logger.warning(f"{log_prefix} 无法获取主要观察者来检查活动状态。")
|
||||
|
||||
# --- 如果活跃，则尝试转换 --- #
|
||||
if is_active:
|
||||
await sub_hf.change_chat_state(ChatState.FOCUSED)
|
||||
# 确认转换成功
|
||||
if sub_hf.chat_state.chat_status == ChatState.FOCUSED:
|
||||
transitioned_count += 1
|
||||
logger.info(f"{log_prefix} 成功进入 FOCUSED 状态。")
|
||||
else:
|
||||
logger.warning(
|
||||
f"{log_prefix} 尝试进入 FOCUSED 状态失败。当前状态: {sub_hf.chat_state.chat_status.value}"
|
||||
)
|
||||
# else: # 不活跃，无需操作
|
||||
# logger.debug(f"{log_prefix} 未检测到新活动，保持 ABSENT。")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"{log_prefix} 检查私聊活动或转换状态时出错: {e}", exc_info=True)
|
||||
|
||||
# --- 循环结束后记录总结日志 --- #
|
||||
if transitioned_count > 0:
|
||||
logger.debug(
|
||||
f"{log_prefix_task} 完成，共检查 {checked_count} 个私聊，{transitioned_count} 个转换为 FOCUSED。"
|
||||
)
|
||||
Reference in New Issue
Block a user