feat：进一步合并normal和focus模式，移除interest_dict（附带其他合理性修改）
This commit is contained in:
SengokuCola
@@ -1,161 +0,0 @@
|
||||
import json
|
||||
from datetime import datetime
|
||||
from typing import Dict, Any
|
||||
from pathlib import Path
|
||||
from src.common.logger import get_logger
|
||||
|
||||
logger = get_logger("hfc_performance")
|
||||
|
||||
|
||||
class HFCPerformanceLogger:
|
||||
"""HFC性能记录管理器"""
|
||||
|
||||
# 版本号常量，可在启动时修改
|
||||
INTERNAL_VERSION = "v7.0.0"
|
||||
|
||||
def __init__(self, chat_id: str):
|
||||
self.chat_id = chat_id
|
||||
self.version = self.INTERNAL_VERSION
|
||||
self.log_dir = Path("log/hfc_loop")
|
||||
self.session_start_time = datetime.now()
|
||||
|
||||
# 确保目录存在
|
||||
self.log_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# 当前会话的日志文件，包含版本号
|
||||
version_suffix = self.version.replace(".", "_")
|
||||
self.session_file = (
|
||||
self.log_dir / f"{chat_id}_{version_suffix}_{self.session_start_time.strftime('%Y%m%d_%H%M%S')}.json"
|
||||
)
|
||||
self.current_session_data = []
|
||||
|
||||
def record_cycle(self, cycle_data: Dict[str, Any]):
|
||||
"""记录单次循环数据"""
|
||||
try:
|
||||
# 构建记录数据
|
||||
record = {
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"version": self.version,
|
||||
"cycle_id": cycle_data.get("cycle_id"),
|
||||
"chat_id": self.chat_id,
|
||||
"action_type": cycle_data.get("action_type", "unknown"),
|
||||
"total_time": cycle_data.get("total_time", 0),
|
||||
"step_times": cycle_data.get("step_times", {}),
|
||||
"reasoning": cycle_data.get("reasoning", ""),
|
||||
"success": cycle_data.get("success", False),
|
||||
}
|
||||
|
||||
# 添加到当前会话数据
|
||||
self.current_session_data.append(record)
|
||||
|
||||
# 立即写入文件（防止数据丢失）
|
||||
self._write_session_data()
|
||||
|
||||
# 构建详细的日志信息
|
||||
log_parts = [
|
||||
f"cycle_id={record['cycle_id']}",
|
||||
f"action={record['action_type']}",
|
||||
f"time={record['total_time']:.2f}s",
|
||||
]
|
||||
|
||||
logger.debug(f"记录HFC循环数据: {', '.join(log_parts)}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"记录HFC循环数据失败: {e}")
|
||||
|
||||
def _write_session_data(self):
|
||||
"""写入当前会话数据到文件"""
|
||||
try:
|
||||
with open(self.session_file, "w", encoding="utf-8") as f:
|
||||
json.dump(self.current_session_data, f, ensure_ascii=False, indent=2)
|
||||
except Exception as e:
|
||||
logger.error(f"写入会话数据失败: {e}")
|
||||
|
||||
def get_current_session_stats(self) -> Dict[str, Any]:
|
||||
"""获取当前会话的基本信息"""
|
||||
if not self.current_session_data:
|
||||
return {}
|
||||
|
||||
return {
|
||||
"chat_id": self.chat_id,
|
||||
"version": self.version,
|
||||
"session_file": str(self.session_file),
|
||||
"record_count": len(self.current_session_data),
|
||||
"start_time": self.session_start_time.isoformat(),
|
||||
}
|
||||
|
||||
def finalize_session(self):
|
||||
"""结束会话"""
|
||||
try:
|
||||
if self.current_session_data:
|
||||
logger.info(f"完成会话，当前会话 {len(self.current_session_data)} 条记录")
|
||||
except Exception as e:
|
||||
logger.error(f"结束会话失败: {e}")
|
||||
|
||||
@classmethod
|
||||
def cleanup_old_logs(cls, max_size_mb: float = 50.0):
|
||||
"""
|
||||
清理旧的HFC日志文件，保持目录大小在指定限制内
|
||||
|
||||
Args:
|
||||
max_size_mb: 最大目录大小限制（MB）
|
||||
"""
|
||||
log_dir = Path("log/hfc_loop")
|
||||
if not log_dir.exists():
|
||||
logger.info("HFC日志目录不存在，跳过日志清理")
|
||||
return
|
||||
|
||||
# 获取所有日志文件及其信息
|
||||
log_files = []
|
||||
total_size = 0
|
||||
|
||||
for log_file in log_dir.glob("*.json"):
|
||||
try:
|
||||
file_stat = log_file.stat()
|
||||
log_files.append({"path": log_file, "size": file_stat.st_size, "mtime": file_stat.st_mtime})
|
||||
total_size += file_stat.st_size
|
||||
except Exception as e:
|
||||
logger.warning(f"无法获取文件信息 {log_file}: {e}")
|
||||
|
||||
if not log_files:
|
||||
logger.info("没有找到HFC日志文件")
|
||||
return
|
||||
|
||||
max_size_bytes = max_size_mb * 1024 * 1024
|
||||
current_size_mb = total_size / (1024 * 1024)
|
||||
|
||||
logger.info(f"HFC日志目录当前大小: {current_size_mb:.2f}MB，限制: {max_size_mb}MB")
|
||||
|
||||
if total_size <= max_size_bytes:
|
||||
logger.info("HFC日志目录大小在限制范围内，无需清理")
|
||||
return
|
||||
|
||||
# 按修改时间排序（最早的在前面）
|
||||
log_files.sort(key=lambda x: x["mtime"])
|
||||
|
||||
deleted_count = 0
|
||||
deleted_size = 0
|
||||
|
||||
for file_info in log_files:
|
||||
if total_size <= max_size_bytes:
|
||||
break
|
||||
|
||||
try:
|
||||
file_size = file_info["size"]
|
||||
file_path = file_info["path"]
|
||||
|
||||
file_path.unlink()
|
||||
total_size -= file_size
|
||||
deleted_size += file_size
|
||||
deleted_count += 1
|
||||
|
||||
logger.info(f"删除旧日志文件: {file_path.name} ({file_size / 1024:.1f}KB)")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"删除日志文件失败 {file_info['path']}: {e}")
|
||||
|
||||
final_size_mb = total_size / (1024 * 1024)
|
||||
deleted_size_mb = deleted_size / (1024 * 1024)
|
||||
|
||||
logger.info(f"HFC日志清理完成: 删除了{deleted_count}个文件，释放{deleted_size_mb:.2f}MB空间")
|
||||
logger.info(f"清理后目录大小: {final_size_mb:.2f}MB")
|
||||
@@ -9,8 +9,6 @@ from typing import Dict, Any
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
log_dir = "log/log_cycle_debug/"
|
||||
|
||||
|
||||
class CycleDetail:
|
||||
"""循环信息记录类"""
|
||||
@@ -104,34 +102,6 @@ class CycleDetail:
|
||||
self.loop_action_info = loop_info["loop_action_info"]
|
||||
|
||||
|
||||
async def create_empty_anchor_message(
|
||||
platform: str, group_info: dict, chat_stream: ChatStream
|
||||
) -> Optional[MessageRecv]:
|
||||
"""
|
||||
重构观察到的最后一条消息作为回复的锚点，
|
||||
如果重构失败或观察为空，则创建一个占位符。
|
||||
"""
|
||||
|
||||
placeholder_id = f"mid_pf_{int(time.time() * 1000)}"
|
||||
placeholder_user = UserInfo(user_id="system_trigger", user_nickname="System Trigger", platform=platform)
|
||||
placeholder_msg_info = BaseMessageInfo(
|
||||
message_id=placeholder_id,
|
||||
platform=platform,
|
||||
group_info=group_info,
|
||||
user_info=placeholder_user,
|
||||
time=time.time(),
|
||||
)
|
||||
placeholder_msg_dict = {
|
||||
"message_info": placeholder_msg_info.to_dict(),
|
||||
"processed_plain_text": "[System Trigger Context]",
|
||||
"raw_message": "",
|
||||
"time": placeholder_msg_info.time,
|
||||
}
|
||||
anchor_message = MessageRecv(placeholder_msg_dict)
|
||||
anchor_message.update_chat_stream(chat_stream)
|
||||
|
||||
return anchor_message
|
||||
|
||||
|
||||
def parse_thinking_id_to_timestamp(thinking_id: str) -> float:
|
||||
"""
|
||||
@@ -143,21 +113,3 @@ def parse_thinking_id_to_timestamp(thinking_id: str) -> float:
|
||||
ts_str = thinking_id[3:]
|
||||
return float(ts_str)
|
||||
|
||||
|
||||
def get_keywords_from_json(json_str: str) -> list[str]:
|
||||
# 提取JSON内容
|
||||
start = json_str.find("{")
|
||||
end = json_str.rfind("}") + 1
|
||||
if start == -1 or end == 0:
|
||||
logger.error("未找到有效的JSON内容")
|
||||
return []
|
||||
|
||||
json_content = json_str[start:end]
|
||||
|
||||
# 解析JSON
|
||||
try:
|
||||
json_data = json.loads(json_content)
|
||||
return json_data.get("keywords", [])
|
||||
except json.JSONDecodeError as e:
|
||||
logger.error(f"JSON解析失败: {e}")
|
||||
return []
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import traceback
|
||||
from src.chat.heart_flow.sub_heartflow import SubHeartflow, ChatState
|
||||
from src.common.logger import get_logger
|
||||
from typing import Any, Optional
|
||||
@@ -30,11 +31,12 @@ class Heartflow:
|
||||
# 注册子心流
|
||||
self.subheartflows[subheartflow_id] = new_subflow
|
||||
heartflow_name = get_chat_manager().get_stream_name(subheartflow_id) or subheartflow_id
|
||||
logger.debug(f"[{heartflow_name}] 开始接收消息")
|
||||
logger.info(f"[{heartflow_name}] 开始接收消息")
|
||||
|
||||
return new_subflow
|
||||
except Exception as e:
|
||||
logger.error(f"创建子心流 {subheartflow_id} 失败: {e}", exc_info=True)
|
||||
traceback.print_exc()
|
||||
return None
|
||||
|
||||
async def force_change_subheartflow_status(self, subheartflow_id: str, status: ChatState) -> None:
|
||||
|
||||
@@ -108,13 +108,14 @@ class HeartFCMessageReceiver:
|
||||
|
||||
interested_rate, is_mentioned = await _calculate_interest(message)
|
||||
message.interest_value = interested_rate
|
||||
message.is_mentioned = is_mentioned
|
||||
|
||||
await self.storage.store_message(message, chat)
|
||||
|
||||
subheartflow = await heartflow.get_or_create_subheartflow(chat.stream_id)
|
||||
message.update_chat_stream(chat)
|
||||
|
||||
subheartflow.add_message_to_normal_chat_cache(message, interested_rate, is_mentioned)
|
||||
# subheartflow.add_message_to_normal_chat_cache(message, interested_rate, is_mentioned)
|
||||
|
||||
chat_mood = mood_manager.get_mood_by_chat_id(subheartflow.chat_id)
|
||||
asyncio.create_task(chat_mood.update_mood_by_message(message, interested_rate))
|
||||
|
||||
@@ -39,16 +39,21 @@ class SubHeartflow:
|
||||
|
||||
self.is_group_chat, self.chat_target_info = get_chat_type_and_target_info(self.chat_id)
|
||||
self.log_prefix = get_chat_manager().get_stream_name(self.subheartflow_id) or self.subheartflow_id
|
||||
# 兴趣消息集合
|
||||
self.interest_dict: Dict[str, tuple[MessageRecv, float, bool]] = {}
|
||||
|
||||
|
||||
# focus模式退出冷却时间管理
|
||||
self.last_focus_exit_time: float = 0 # 上次退出focus模式的时间
|
||||
|
||||
# 随便水群 normal_chat 和 认真水群 focus_chat 实例
|
||||
# CHAT模式激活 随便水群 FOCUS模式激活 认真水群
|
||||
self.heart_fc_instance: Optional[HeartFChatting] = None # 该sub_heartflow的HeartFChatting实例
|
||||
self.normal_chat_instance: Optional[NormalChat] = None # 该sub_heartflow的NormalChat实例
|
||||
self.heart_fc_instance: Optional[HeartFChatting] = HeartFChatting(
|
||||
chat_id=self.subheartflow_id,
|
||||
on_stop_focus_chat=self._handle_stop_focus_chat_request,
|
||||
) # 该sub_heartflow的HeartFChatting实例
|
||||
self.normal_chat_instance: Optional[NormalChat] = NormalChat(
|
||||
chat_stream=get_chat_manager().get_stream(self.chat_id),
|
||||
on_switch_to_focus_callback=self._handle_switch_to_focus_request,
|
||||
get_cooldown_progress_callback=self.get_cooldown_progress,
|
||||
) # 该sub_heartflow的NormalChat实例
|
||||
|
||||
async def initialize(self):
|
||||
"""异步初始化方法，创建兴趣流并确定聊天类型"""
|
||||
@@ -79,10 +84,6 @@ class SubHeartflow:
|
||||
# 使用更短的超时时间，强制快速停止
|
||||
await asyncio.wait_for(self.normal_chat_instance.stop_chat(), timeout=3.0)
|
||||
logger.debug(f"{self.log_prefix} stop_chat() 调用完成")
|
||||
except asyncio.TimeoutError:
|
||||
logger.warning(f"{self.log_prefix} 停止 NormalChat 超时，强制清理")
|
||||
# 超时时强制清理实例
|
||||
self.normal_chat_instance = None
|
||||
except Exception as e:
|
||||
logger.error(f"{self.log_prefix} 停止 NormalChat 监控任务时出错: {e}")
|
||||
# 出错时也要清理实例，避免状态不一致
|
||||
@@ -93,8 +94,10 @@ class SubHeartflow:
|
||||
logger.warning(f"{self.log_prefix} 强制清理 NormalChat 实例")
|
||||
self.normal_chat_instance = None
|
||||
logger.debug(f"{self.log_prefix} _stop_normal_chat 完成")
|
||||
else:
|
||||
logger.info(f"{self.log_prefix} 没有normal聊天实例，无需停止normal聊天")
|
||||
|
||||
async def _start_normal_chat(self, rewind=False) -> bool:
|
||||
async def _start_normal_chat(self) -> bool:
|
||||
"""
|
||||
启动 NormalChat 实例，并进行异步初始化。
|
||||
进入 CHAT 状态时使用。
|
||||
@@ -102,30 +105,23 @@ class SubHeartflow:
|
||||
"""
|
||||
await self._stop_heart_fc_chat() # 确保 专注聊天已停止
|
||||
|
||||
self.interest_dict.clear()
|
||||
|
||||
log_prefix = self.log_prefix
|
||||
try:
|
||||
# 获取聊天流并创建 NormalChat 实例 (同步部分)
|
||||
chat_stream = get_chat_manager().get_stream(self.chat_id)
|
||||
if not chat_stream:
|
||||
logger.error(f"{log_prefix} 无法获取 chat_stream，无法启动 NormalChat。")
|
||||
return False
|
||||
# 在 rewind 为 True 或 NormalChat 实例尚未创建时，创建新实例
|
||||
if rewind or not self.normal_chat_instance:
|
||||
# 在 NormalChat 实例尚未创建时，创建新实例
|
||||
if not self.normal_chat_instance:
|
||||
# 提供回调函数，用于接收需要切换到focus模式的通知
|
||||
self.normal_chat_instance = NormalChat(
|
||||
chat_stream=chat_stream,
|
||||
interest_dict=self.interest_dict,
|
||||
on_switch_to_focus_callback=self._handle_switch_to_focus_request,
|
||||
get_cooldown_progress_callback=self.get_cooldown_progress,
|
||||
)
|
||||
|
||||
logger.info(f"{log_prefix} 开始普通聊天，随便水群...")
|
||||
logger.info(f"[{self.log_prefix}] 开始普通聊天")
|
||||
await self.normal_chat_instance.start_chat() # start_chat now ensures init is called again if needed
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"{log_prefix} 启动 NormalChat 或其初始化时出错: {e}")
|
||||
logger.error(f"[{self.log_prefix}] 启动 NormalChat 或其初始化时出错: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
self.normal_chat_instance = None # 启动/初始化失败，清理实例
|
||||
return False
|
||||
@@ -173,68 +169,36 @@ class SubHeartflow:
|
||||
|
||||
async def _stop_heart_fc_chat(self):
|
||||
"""停止并清理 HeartFChatting 实例"""
|
||||
if self.heart_fc_instance:
|
||||
logger.debug(f"{self.log_prefix} 结束专注聊天...")
|
||||
if self.heart_fc_instance.running:
|
||||
logger.info(f"{self.log_prefix} 结束专注聊天...")
|
||||
try:
|
||||
await self.heart_fc_instance.shutdown()
|
||||
except Exception as e:
|
||||
logger.error(f"{self.log_prefix} 关闭 HeartFChatting 实例时出错: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
finally:
|
||||
# 无论是否成功关闭，都清理引用
|
||||
self.heart_fc_instance = None
|
||||
else:
|
||||
logger.info(f"{self.log_prefix} 没有专注聊天实例，无需停止专注聊天")
|
||||
|
||||
async def _start_heart_fc_chat(self) -> bool:
|
||||
"""启动 HeartFChatting 实例，确保 NormalChat 已停止"""
|
||||
logger.debug(f"{self.log_prefix} 开始启动 HeartFChatting")
|
||||
|
||||
try:
|
||||
# 确保普通聊天监控已停止
|
||||
await self._stop_normal_chat()
|
||||
self.interest_dict.clear()
|
||||
|
||||
log_prefix = self.log_prefix
|
||||
# 如果实例已存在，检查其循环任务状态
|
||||
if self.heart_fc_instance:
|
||||
logger.debug(f"{log_prefix} HeartFChatting 实例已存在，检查状态")
|
||||
# 如果任务已完成或不存在，则尝试重新启动
|
||||
if self.heart_fc_instance._loop_task is None or self.heart_fc_instance._loop_task.done():
|
||||
logger.info(f"{log_prefix} HeartFChatting 实例存在但循环未运行，尝试启动...")
|
||||
try:
|
||||
# 添加超时保护
|
||||
await asyncio.wait_for(self.heart_fc_instance.start(), timeout=15.0)
|
||||
logger.info(f"{log_prefix} HeartFChatting 循环已启动。")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"{log_prefix} 尝试启动现有 HeartFChatting 循环时出错: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
# 出错时清理实例，准备重新创建
|
||||
self.heart_fc_instance = None
|
||||
else:
|
||||
# 任务正在运行
|
||||
logger.debug(f"{log_prefix} HeartFChatting 已在运行中。")
|
||||
return True # 已经在运行
|
||||
|
||||
# 如果实例不存在，则创建并启动
|
||||
logger.info(f"{log_prefix} 麦麦准备开始专注聊天...")
|
||||
try:
|
||||
logger.debug(f"{log_prefix} 创建新的 HeartFChatting 实例")
|
||||
self.heart_fc_instance = HeartFChatting(
|
||||
chat_id=self.subheartflow_id,
|
||||
on_stop_focus_chat=self._handle_stop_focus_chat_request,
|
||||
)
|
||||
|
||||
logger.debug(f"{log_prefix} 启动 HeartFChatting 实例")
|
||||
# 添加超时保护
|
||||
await asyncio.wait_for(self.heart_fc_instance.start(), timeout=15.0)
|
||||
logger.debug(f"{log_prefix} 麦麦已成功进入专注聊天模式 (新实例已启动)。")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"{log_prefix} 创建或启动 HeartFChatting 实例时出错: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
self.heart_fc_instance = None # 创建或初始化异常，清理实例
|
||||
return False
|
||||
# 如果任务已完成或不存在，则尝试重新启动
|
||||
if self.heart_fc_instance._loop_task is None or self.heart_fc_instance._loop_task.done():
|
||||
logger.info(f"{self.log_prefix} HeartFChatting 实例存在但循环未运行，尝试启动...")
|
||||
try:
|
||||
# 添加超时保护
|
||||
await asyncio.wait_for(self.heart_fc_instance.start(), timeout=15.0)
|
||||
logger.info(f"{self.log_prefix} HeartFChatting 循环已启动。")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"{self.log_prefix} 尝试启动现有 HeartFChatting 循环时出错: {e}")
|
||||
logger.error(traceback.format_exc())
|
||||
# 出错时清理实例，准备重新创建
|
||||
self.heart_fc_instance = None
|
||||
else:
|
||||
# 任务正在运行
|
||||
logger.debug(f"{self.log_prefix} HeartFChatting 已在运行中。")
|
||||
return True # 已经在运行
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"{self.log_prefix} _start_heart_fc_chat 执行时出错: {e}")
|
||||
@@ -248,39 +212,36 @@ class SubHeartflow:
|
||||
"""
|
||||
current_state = self.chat_state.chat_status
|
||||
state_changed = False
|
||||
log_prefix = f"[{self.log_prefix}]"
|
||||
|
||||
|
||||
if new_state == ChatState.NORMAL:
|
||||
logger.debug(f"{log_prefix} 准备进入 normal聊天 状态")
|
||||
if await self._start_normal_chat():
|
||||
logger.debug(f"{log_prefix} 成功进入或保持 NormalChat 状态。")
|
||||
state_changed = True
|
||||
else:
|
||||
logger.error(f"{log_prefix} 启动 NormalChat 失败，无法进入 CHAT 状态。")
|
||||
# 启动失败时，保持当前状态
|
||||
if self.normal_chat_instance.running:
|
||||
logger.info(f"{self.log_prefix} 当前状态已经为normal")
|
||||
return
|
||||
else:
|
||||
if await self._start_normal_chat():
|
||||
logger.debug(f"{self.log_prefix} 成功进入或保持 NormalChat 状态。")
|
||||
state_changed = True
|
||||
else:
|
||||
logger.error(f"{self.log_prefix} 启动 NormalChat 失败，无法进入 CHAT 状态。")
|
||||
return
|
||||
|
||||
elif new_state == ChatState.FOCUSED:
|
||||
logger.debug(f"{log_prefix} 准备进入 focus聊天 状态")
|
||||
if self.heart_fc_instance.running:
|
||||
logger.info(f"{self.log_prefix} 当前状态已经为focused")
|
||||
return
|
||||
if await self._start_heart_fc_chat():
|
||||
logger.debug(f"{log_prefix} 成功进入或保持 HeartFChatting 状态。")
|
||||
logger.debug(f"{self.log_prefix} 成功进入或保持 HeartFChatting 状态。")
|
||||
state_changed = True
|
||||
else:
|
||||
logger.error(f"{log_prefix} 启动 HeartFChatting 失败，无法进入 FOCUSED 状态。")
|
||||
logger.error(f"{self.log_prefix} 启动 HeartFChatting 失败，无法进入 FOCUSED 状态。")
|
||||
# 启动失败时，保持当前状态
|
||||
return
|
||||
|
||||
elif new_state == ChatState.ABSENT:
|
||||
logger.info(f"{log_prefix} 进入 ABSENT 状态，停止所有聊天活动...")
|
||||
self.interest_dict.clear()
|
||||
await self._stop_normal_chat()
|
||||
await self._stop_heart_fc_chat()
|
||||
state_changed = True
|
||||
|
||||
# --- 记录focus模式退出时间 ---
|
||||
if state_changed and current_state == ChatState.FOCUSED and new_state != ChatState.FOCUSED:
|
||||
self.last_focus_exit_time = time.time()
|
||||
logger.debug(f"{log_prefix} 记录focus模式退出时间: {self.last_focus_exit_time}")
|
||||
logger.debug(f"{self.log_prefix} 记录focus模式退出时间: {self.last_focus_exit_time}")
|
||||
|
||||
# --- 更新状态和最后活动时间 ---
|
||||
if state_changed:
|
||||
@@ -292,7 +253,7 @@ class SubHeartflow:
|
||||
self.chat_state_changed_time = time.time()
|
||||
else:
|
||||
logger.debug(
|
||||
f"{log_prefix} 尝试将状态从 {current_state.value} 变为 {new_state.value}，但未成功或未执行更改。"
|
||||
f"{self.log_prefix} 尝试将状态从 {current_state.value} 变为 {new_state.value}，但未成功或未执行更改。"
|
||||
)
|
||||
|
||||
def add_message_to_normal_chat_cache(self, message: MessageRecv, interest_value: float, is_mentioned: bool):
|
||||
|
||||
@@ -11,7 +11,7 @@ from src.chat.message_receive.chat_stream import ChatStream, get_chat_manager
|
||||
from src.chat.utils.timer_calculator import Timer
|
||||
from src.common.message_repository import count_messages
|
||||
from src.chat.utils.prompt_builder import global_prompt_manager
|
||||
from ..message_receive.message import MessageSending, MessageRecv, MessageThinking, MessageSet
|
||||
from ..message_receive.message import MessageSending, MessageThinking, MessageSet, MessageRecv,message_from_db_dict
|
||||
from src.chat.message_receive.normal_message_sender import message_manager
|
||||
from src.chat.normal_chat.willing.willing_manager import get_willing_manager
|
||||
from src.chat.planner_actions.action_manager import ActionManager
|
||||
@@ -20,6 +20,7 @@ from .priority_manager import PriorityManager
|
||||
import traceback
|
||||
from src.chat.planner_actions.planner import ActionPlanner
|
||||
from src.chat.planner_actions.action_modifier import ActionModifier
|
||||
from src.chat.utils.chat_message_builder import get_raw_msg_by_timestamp_with_chat_inclusive
|
||||
|
||||
from src.chat.utils.utils import get_chat_type_and_target_info
|
||||
from src.mood.mood_manager import mood_manager
|
||||
@@ -28,6 +29,7 @@ willing_manager = get_willing_manager()
|
||||
|
||||
logger = get_logger("normal_chat")
|
||||
|
||||
LOOP_INTERVAL = 0.3
|
||||
|
||||
class NormalChat:
|
||||
"""
|
||||
@@ -38,7 +40,6 @@ class NormalChat:
|
||||
def __init__(
|
||||
self,
|
||||
chat_stream: ChatStream,
|
||||
interest_dict: dict = None,
|
||||
on_switch_to_focus_callback=None,
|
||||
get_cooldown_progress_callback=None,
|
||||
):
|
||||
@@ -50,14 +51,12 @@ class NormalChat:
|
||||
"""
|
||||
self.chat_stream = chat_stream
|
||||
self.stream_id = chat_stream.stream_id
|
||||
self.last_read_time = time.time()-1
|
||||
|
||||
self.stream_name = get_chat_manager().get_stream_name(self.stream_id) or self.stream_id
|
||||
|
||||
self.relationship_builder = relationship_builder_manager.get_or_create_builder(self.stream_id)
|
||||
|
||||
# Interest dict
|
||||
self.interest_dict = interest_dict
|
||||
|
||||
self.is_group_chat, self.chat_target_info = get_chat_type_and_target_info(self.stream_id)
|
||||
|
||||
self.willing_amplifier = 1
|
||||
@@ -65,6 +64,8 @@ class NormalChat:
|
||||
|
||||
self.mood_manager = mood_manager
|
||||
self.start_time = time.time()
|
||||
|
||||
self.running = False
|
||||
|
||||
self._initialized = False # Track initialization status
|
||||
|
||||
@@ -93,14 +94,13 @@ class NormalChat:
|
||||
|
||||
# 任务管理
|
||||
self._chat_task: Optional[asyncio.Task] = None
|
||||
self._priority_chat_task: Optional[asyncio.Task] = None # for priority mode consumer
|
||||
self._disabled = False # 停用标志
|
||||
|
||||
# 新增：回复模式和优先级管理器
|
||||
self.reply_mode = self.chat_stream.context.get_priority_mode()
|
||||
if self.reply_mode == "priority":
|
||||
interest_dict = interest_dict or {}
|
||||
self.priority_manager = PriorityManager(
|
||||
interest_dict=interest_dict,
|
||||
normal_queue_max_size=5,
|
||||
)
|
||||
else:
|
||||
@@ -114,34 +114,90 @@ class NormalChat:
|
||||
if self.reply_mode == "priority" and self._priority_chat_task and not self._priority_chat_task.done():
|
||||
self._priority_chat_task.cancel()
|
||||
logger.info(f"[{self.stream_name}] NormalChat 已停用。")
|
||||
|
||||
async def _interest_mode_loopbody(self):
|
||||
try:
|
||||
await asyncio.sleep(LOOP_INTERVAL)
|
||||
|
||||
if self._disabled:
|
||||
return False
|
||||
|
||||
async def _priority_chat_loop_add_message(self):
|
||||
while not self._disabled:
|
||||
now = time.time()
|
||||
new_messages_data = get_raw_msg_by_timestamp_with_chat_inclusive(
|
||||
chat_id=self.stream_id, timestamp_start=self.last_read_time, timestamp_end=now, limit_mode="earliest"
|
||||
)
|
||||
|
||||
if new_messages_data:
|
||||
self.last_read_time = now
|
||||
|
||||
for msg_data in new_messages_data:
|
||||
try:
|
||||
self.adjust_reply_frequency()
|
||||
await self.normal_response(
|
||||
message_data=msg_data,
|
||||
is_mentioned=msg_data.get("is_mentioned", False),
|
||||
interested_rate=msg_data.get("interest_rate", 0.0) * self.willing_amplifier,
|
||||
)
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"[{self.stream_name}] 处理消息时出错: {e} {traceback.format_exc()}")
|
||||
|
||||
|
||||
except asyncio.CancelledError:
|
||||
logger.info(f"[{self.stream_name}] 兴趣模式轮询任务被取消")
|
||||
return False
|
||||
except Exception:
|
||||
logger.error(f"[{self.stream_name}] 兴趣模式轮询循环出现错误: {traceback.format_exc()}", exc_info=True)
|
||||
await asyncio.sleep(10)
|
||||
|
||||
async def _priority_mode_loopbody(self):
|
||||
try:
|
||||
# 创建字典条目的副本以避免在迭代时发生修改
|
||||
items_to_process = list(self.interest_dict.items())
|
||||
for msg_id, value in items_to_process:
|
||||
# 尝试从原始字典中弹出条目，如果它已被其他任务处理，则跳过
|
||||
if self.interest_dict.pop(msg_id, None) is None:
|
||||
continue # 条目已被其他任务处理
|
||||
await asyncio.sleep(LOOP_INTERVAL)
|
||||
|
||||
message, interest_value, _ = value
|
||||
if not self._disabled:
|
||||
# 更新消息段信息
|
||||
# self._update_user_message_segments(message)
|
||||
if self._disabled:
|
||||
return False
|
||||
|
||||
# 添加消息到优先级管理器
|
||||
if self.priority_manager:
|
||||
self.priority_manager.add_message(message, interest_value)
|
||||
|
||||
except Exception:
|
||||
logger.error(
|
||||
f"[{self.stream_name}] 优先级聊天循环添加消息时出现错误: {traceback.format_exc()}", exc_info=True
|
||||
now = time.time()
|
||||
new_messages_data = get_raw_msg_by_timestamp_with_chat_inclusive(
|
||||
chat_id=self.stream_id, timestamp_start=self.last_read_time, timestamp_end=now, limit_mode="earliest"
|
||||
)
|
||||
print(traceback.format_exc())
|
||||
# 出现错误时，等待一段时间再重试
|
||||
raise
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
if new_messages_data:
|
||||
self.last_read_time = now
|
||||
|
||||
for msg_data in new_messages_data:
|
||||
try:
|
||||
if self.priority_manager:
|
||||
self.priority_manager.add_message(msg_data, msg_data.get("interest_rate", 0.0))
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"[{self.stream_name}] 添加消息到优先级队列时出错: {e} {traceback.format_exc()}")
|
||||
|
||||
|
||||
except asyncio.CancelledError:
|
||||
logger.info(f"[{self.stream_name}] 优先级消息生产者任务被取消")
|
||||
return False
|
||||
except Exception:
|
||||
logger.error(f"[{self.stream_name}] 优先级消息生产者循环出现错误: {traceback.format_exc()}", exc_info=True)
|
||||
await asyncio.sleep(10)
|
||||
|
||||
async def _interest_message_polling_loop(self):
|
||||
"""
|
||||
[Interest Mode] 通过轮询数据库获取新消息并直接处理。
|
||||
"""
|
||||
logger.info(f"[{self.stream_name}] 兴趣模式消息轮询任务开始")
|
||||
try:
|
||||
while not self._disabled:
|
||||
success = await self._interest_mode_loopbody()
|
||||
|
||||
if not success:
|
||||
break
|
||||
|
||||
except asyncio.CancelledError:
|
||||
logger.info(f"[{self.stream_name}] 兴趣模式消息轮询任务被优雅地取消了")
|
||||
|
||||
|
||||
|
||||
|
||||
async def _priority_chat_loop(self):
|
||||
"""
|
||||
@@ -149,16 +205,16 @@ class NormalChat:
|
||||
"""
|
||||
while not self._disabled:
|
||||
try:
|
||||
if not self.priority_manager.is_empty():
|
||||
# 获取最高优先级的消息
|
||||
message = self.priority_manager.get_highest_priority_message()
|
||||
if self.priority_manager and not self.priority_manager.is_empty():
|
||||
# 获取最高优先级的消息,现在是字典
|
||||
message_data = self.priority_manager.get_highest_priority_message()
|
||||
|
||||
if message:
|
||||
if message_data:
|
||||
logger.info(
|
||||
f"[{self.stream_name}] 从队列中取出消息进行处理: User {message.message_info.user_info.user_id}, Time: {time.strftime('%H:%M:%S', time.localtime(message.message_info.time))}"
|
||||
f"[{self.stream_name}] 从队列中取出消息进行处理: User {message_data.get('user_id')}, Time: {time.strftime('%H:%M:%S', time.localtime(message_data.get('time')))}"
|
||||
)
|
||||
|
||||
do_reply = await self.reply_one_message(message)
|
||||
do_reply = await self.reply_one_message(message_data)
|
||||
response_set = do_reply if do_reply else []
|
||||
factor = 0.5
|
||||
cnt = sum([len(r) for r in response_set])
|
||||
@@ -176,14 +232,12 @@ class NormalChat:
|
||||
await asyncio.sleep(10)
|
||||
|
||||
# 改为实例方法
|
||||
async def _create_thinking_message(self, message: MessageRecv, timestamp: Optional[float] = None) -> str:
|
||||
async def _create_thinking_message(self, message_data: dict, timestamp: Optional[float] = None) -> str:
|
||||
"""创建思考消息"""
|
||||
messageinfo = message.message_info
|
||||
|
||||
bot_user_info = UserInfo(
|
||||
user_id=global_config.bot.qq_account,
|
||||
user_nickname=global_config.bot.nickname,
|
||||
platform=messageinfo.platform,
|
||||
platform=message_data.get("chat_info_platform"),
|
||||
)
|
||||
|
||||
thinking_time_point = round(time.time(), 2)
|
||||
@@ -192,7 +246,7 @@ class NormalChat:
|
||||
message_id=thinking_id,
|
||||
chat_stream=self.chat_stream,
|
||||
bot_user_info=bot_user_info,
|
||||
reply=message,
|
||||
reply=None,
|
||||
thinking_start_time=thinking_time_point,
|
||||
timestamp=timestamp if timestamp is not None else None,
|
||||
)
|
||||
@@ -202,7 +256,7 @@ class NormalChat:
|
||||
|
||||
# 改为实例方法
|
||||
async def _add_messages_to_manager(
|
||||
self, message: MessageRecv, response_set: List[str], thinking_id
|
||||
self, message_data: dict, response_set: List[str], thinking_id
|
||||
) -> Optional[MessageSending]:
|
||||
"""发送回复消息"""
|
||||
container = await message_manager.get_container(self.stream_id) # 使用 self.stream_id
|
||||
@@ -221,6 +275,15 @@ class NormalChat:
|
||||
thinking_start_time = thinking_message.thinking_start_time
|
||||
message_set = MessageSet(self.chat_stream, thinking_id) # 使用 self.chat_stream
|
||||
|
||||
sender_info = UserInfo(
|
||||
user_id=message_data.get("user_id"),
|
||||
user_nickname=message_data.get("user_nickname"),
|
||||
platform=message_data.get("chat_info_platform"),
|
||||
)
|
||||
|
||||
reply = message_from_db_dict(message_data)
|
||||
|
||||
|
||||
mark_head = False
|
||||
first_bot_msg = None
|
||||
for msg in response_set:
|
||||
@@ -233,11 +296,11 @@ class NormalChat:
|
||||
bot_user_info=UserInfo(
|
||||
user_id=global_config.bot.qq_account,
|
||||
user_nickname=global_config.bot.nickname,
|
||||
platform=message.message_info.platform,
|
||||
platform=message_data.get("chat_info_platform"),
|
||||
),
|
||||
sender_info=message.message_info.user_info,
|
||||
sender_info=sender_info,
|
||||
message_segment=message_segment,
|
||||
reply=message,
|
||||
reply=reply,
|
||||
is_head=not mark_head,
|
||||
is_emoji=False,
|
||||
thinking_start_time=thinking_start_time,
|
||||
@@ -252,122 +315,8 @@ class NormalChat:
|
||||
|
||||
return first_bot_msg
|
||||
|
||||
async def _reply_interested_message(self) -> None:
|
||||
"""
|
||||
后台任务方法，轮询当前实例关联chat的兴趣消息
|
||||
通常由start_monitoring_interest()启动
|
||||
"""
|
||||
logger.debug(f"[{self.stream_name}] 兴趣监控任务开始")
|
||||
|
||||
try:
|
||||
while True:
|
||||
# 第一层检查：立即检查取消和停用状态
|
||||
if self._disabled:
|
||||
logger.info(f"[{self.stream_name}] 检测到停用标志，退出兴趣监控")
|
||||
break
|
||||
|
||||
# 检查当前任务是否已被取消
|
||||
current_task = asyncio.current_task()
|
||||
if current_task and current_task.cancelled():
|
||||
logger.info(f"[{self.stream_name}] 当前任务已被取消，退出")
|
||||
break
|
||||
|
||||
try:
|
||||
# 短暂等待，让出控制权
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
# 第二层检查：睡眠后再次检查状态
|
||||
if self._disabled:
|
||||
logger.info(f"[{self.stream_name}] 睡眠后检测到停用标志，退出")
|
||||
break
|
||||
|
||||
# 获取待处理消息
|
||||
items_to_process = list(self.interest_dict.items())
|
||||
if not items_to_process:
|
||||
# 没有消息时继续下一轮循环
|
||||
continue
|
||||
|
||||
# 第三层检查：在处理消息前最后检查一次
|
||||
if self._disabled:
|
||||
logger.info(f"[{self.stream_name}] 处理消息前检测到停用标志，退出")
|
||||
break
|
||||
|
||||
# 使用异步上下文管理器处理消息
|
||||
try:
|
||||
async with global_prompt_manager.async_message_scope(
|
||||
self.chat_stream.context.get_template_name()
|
||||
):
|
||||
# 在上下文内部再次检查取消状态
|
||||
if self._disabled:
|
||||
logger.info(f"[{self.stream_name}] 在处理上下文中检测到停止信号，退出")
|
||||
break
|
||||
|
||||
semaphore = asyncio.Semaphore(5)
|
||||
|
||||
async def process_and_acquire(
|
||||
msg_id, message, interest_value, is_mentioned, semaphore=semaphore
|
||||
):
|
||||
"""处理单个兴趣消息并管理信号量"""
|
||||
async with semaphore:
|
||||
try:
|
||||
# 在处理每个消息前检查停止状态
|
||||
if self._disabled:
|
||||
logger.debug(
|
||||
f"[{self.stream_name}] 处理消息时检测到停用，跳过消息 {msg_id}"
|
||||
)
|
||||
return
|
||||
|
||||
# 处理消息
|
||||
self.adjust_reply_frequency()
|
||||
|
||||
await self.normal_response(
|
||||
message=message,
|
||||
is_mentioned=is_mentioned,
|
||||
interested_rate=interest_value * self.willing_amplifier,
|
||||
)
|
||||
except asyncio.CancelledError:
|
||||
logger.debug(f"[{self.stream_name}] 处理消息 {msg_id} 时被取消")
|
||||
raise # 重新抛出取消异常
|
||||
except Exception as e:
|
||||
logger.error(f"[{self.stream_name}] 处理兴趣消息{msg_id}时出错: {e}")
|
||||
# 不打印完整traceback，避免日志污染
|
||||
finally:
|
||||
# 无论如何都要清理消息
|
||||
self.interest_dict.pop(msg_id, None)
|
||||
|
||||
tasks = [
|
||||
process_and_acquire(msg_id, message, interest_value, is_mentioned)
|
||||
for msg_id, (message, interest_value, is_mentioned) in items_to_process
|
||||
]
|
||||
|
||||
if tasks:
|
||||
await asyncio.gather(*tasks, return_exceptions=True)
|
||||
|
||||
except asyncio.CancelledError:
|
||||
logger.info(f"[{self.stream_name}] 处理上下文时任务被取消")
|
||||
break
|
||||
except Exception as e:
|
||||
logger.error(f"[{self.stream_name}] 处理上下文时出错: {e}")
|
||||
# 出错后短暂等待，避免快速重试
|
||||
await asyncio.sleep(0.5)
|
||||
|
||||
except asyncio.CancelledError:
|
||||
logger.info(f"[{self.stream_name}] 主循环中任务被取消")
|
||||
break
|
||||
except Exception as e:
|
||||
logger.error(f"[{self.stream_name}] 主循环出错: {e}")
|
||||
# 出错后等待一秒再继续
|
||||
await asyncio.sleep(1.0)
|
||||
|
||||
except asyncio.CancelledError:
|
||||
logger.info(f"[{self.stream_name}] 兴趣监控任务被取消")
|
||||
except Exception as e:
|
||||
logger.error(f"[{self.stream_name}] 兴趣监控任务严重错误: {e}")
|
||||
finally:
|
||||
logger.debug(f"[{self.stream_name}] 兴趣监控任务结束")
|
||||
|
||||
# 改为实例方法, 移除 chat 参数
|
||||
async def normal_response(self, message: MessageRecv, is_mentioned: bool, interested_rate: float) -> None:
|
||||
async def normal_response(self, message_data: dict, is_mentioned: bool, interested_rate: float) -> None:
|
||||
"""
|
||||
处理接收到的消息。
|
||||
在"兴趣"模式下，判断是否回复并生成内容。
|
||||
@@ -396,22 +345,23 @@ class NormalChat:
|
||||
) # 如果被提及，且开启了提及必回复，则基础概率为1，否则需要意愿判断
|
||||
|
||||
# 意愿管理器：设置当前message信息
|
||||
willing_manager.setup(message, self.chat_stream, is_mentioned, interested_rate)
|
||||
willing_manager.setup(message_data, self.chat_stream)
|
||||
# TODO: willing_manager 也需要修改以接收字典
|
||||
|
||||
# 获取回复概率
|
||||
# is_willing = False
|
||||
# 仅在未被提及或基础概率不为1时查询意愿概率
|
||||
if reply_probability < 1: # 简化逻辑，如果未提及 (reply_probability 为 0)，则获取意愿概率
|
||||
# is_willing = True
|
||||
reply_probability = await willing_manager.get_reply_probability(message.message_info.message_id)
|
||||
reply_probability = await willing_manager.get_reply_probability(message_data.get("message_id"))
|
||||
|
||||
if message.message_info.additional_config:
|
||||
if "maimcore_reply_probability_gain" in message.message_info.additional_config.keys():
|
||||
reply_probability += message.message_info.additional_config["maimcore_reply_probability_gain"]
|
||||
reply_probability = min(max(reply_probability, 0), 1) # 确保概率在 0-1 之间
|
||||
additional_config = message_data.get("additional_config", {})
|
||||
if additional_config and "maimcore_reply_probability_gain" in additional_config:
|
||||
reply_probability += additional_config["maimcore_reply_probability_gain"]
|
||||
reply_probability = min(max(reply_probability, 0), 1) # 确保概率在 0-1 之间
|
||||
|
||||
# 处理表情包
|
||||
if message.is_emoji or message.is_picid:
|
||||
if message_data.get("is_emoji") or message_data.get("is_picid"):
|
||||
reply_probability = 0
|
||||
|
||||
# 应用疲劳期回复频率调整
|
||||
@@ -427,53 +377,50 @@ class NormalChat:
|
||||
|
||||
# 打印消息信息
|
||||
mes_name = self.chat_stream.group_info.group_name if self.chat_stream.group_info else "私聊"
|
||||
# current_time = time.strftime("%H:%M:%S", time.localtime(message.message_info.time))
|
||||
# 使用 self.stream_id
|
||||
# willing_log = f"[激活值:{await willing_manager.get_willing(self.stream_id):.2f}]" if is_willing else ""
|
||||
if reply_probability > 0.1:
|
||||
logger.info(
|
||||
f"[{mes_name}]"
|
||||
f"{message.message_info.user_info.user_nickname}:" # 使用 self.chat_stream
|
||||
f"{message.processed_plain_text}[兴趣:{interested_rate:.2f}][回复概率:{reply_probability * 100:.1f}%]"
|
||||
f"{message_data.get('user_nickname')}:"
|
||||
f"{message_data.get('processed_plain_text')}[兴趣:{interested_rate:.2f}][回复概率:{reply_probability * 100:.1f}%]"
|
||||
)
|
||||
do_reply = False
|
||||
response_set = None # 初始化 response_set
|
||||
if random() < reply_probability:
|
||||
with Timer("获取回复", timing_results):
|
||||
await willing_manager.before_generate_reply_handle(message.message_info.message_id)
|
||||
do_reply = await self.reply_one_message(message)
|
||||
await willing_manager.before_generate_reply_handle(message_data.get("message_id"))
|
||||
do_reply = await self.reply_one_message(message_data)
|
||||
response_set = do_reply if do_reply else None
|
||||
|
||||
# 输出性能计时结果
|
||||
if do_reply and response_set: # 确保 response_set 不是 None
|
||||
timing_str = " | ".join([f"{step}: {duration:.2f}秒" for step, duration in timing_results.items()])
|
||||
trigger_msg = message.processed_plain_text
|
||||
trigger_msg = message_data.get("processed_plain_text")
|
||||
response_msg = " ".join([item[1] for item in response_set if item[0] == "text"])
|
||||
logger.info(
|
||||
f"[{self.stream_name}]回复消息: {trigger_msg[:30]}... | 回复内容: {response_msg[:30]}... | 计时: {timing_str}"
|
||||
)
|
||||
await willing_manager.after_generate_reply_handle(message.message_info.message_id)
|
||||
await willing_manager.after_generate_reply_handle(message_data.get("message_id"))
|
||||
elif not do_reply:
|
||||
# 不回复处理
|
||||
await willing_manager.not_reply_handle(message.message_info.message_id)
|
||||
await willing_manager.not_reply_handle(message_data.get("message_id"))
|
||||
|
||||
# 意愿管理器：注销当前message信息 (无论是否回复，只要处理过就删除)
|
||||
willing_manager.delete(message.message_info.message_id)
|
||||
willing_manager.delete(message_data.get("message_id"))
|
||||
|
||||
async def _generate_normal_response(
|
||||
self, message: MessageRecv, available_actions: Optional[list]
|
||||
self, message_data: dict, available_actions: Optional[list]
|
||||
) -> Optional[list]:
|
||||
"""生成普通回复"""
|
||||
try:
|
||||
person_info_manager = get_person_info_manager()
|
||||
person_id = person_info_manager.get_person_id(
|
||||
message.chat_stream.user_info.platform, message.chat_stream.user_info.user_id
|
||||
message_data.get("chat_info_platform"), message_data.get("user_id")
|
||||
)
|
||||
person_name = await person_info_manager.get_value(person_id, "person_name")
|
||||
reply_to_str = f"{person_name}:{message.processed_plain_text}"
|
||||
reply_to_str = f"{person_name}:{message_data.get('processed_plain_text')}"
|
||||
|
||||
success, reply_set = await generator_api.generate_reply(
|
||||
chat_stream=message.chat_stream,
|
||||
chat_stream=self.chat_stream,
|
||||
reply_to=reply_to_str,
|
||||
available_actions=available_actions,
|
||||
enable_tool=global_config.tool.enable_in_normal_chat,
|
||||
@@ -481,7 +428,7 @@ class NormalChat:
|
||||
)
|
||||
|
||||
if not success or not reply_set:
|
||||
logger.info(f"对 {message.processed_plain_text} 的回复生成失败")
|
||||
logger.info(f"对 {message_data.get('processed_plain_text')} 的回复生成失败")
|
||||
return None
|
||||
|
||||
return reply_set
|
||||
@@ -490,7 +437,7 @@ class NormalChat:
|
||||
logger.error(f"[{self.stream_name}] 回复生成出现错误：{str(e)} {traceback.format_exc()}")
|
||||
return None
|
||||
|
||||
async def _plan_and_execute_actions(self, message: MessageRecv, thinking_id: str) -> Optional[dict]:
|
||||
async def _plan_and_execute_actions(self, message_data: dict, thinking_id: str) -> Optional[dict]:
|
||||
"""规划和执行额外动作"""
|
||||
no_action = {
|
||||
"action_result": {
|
||||
@@ -539,7 +486,7 @@ class NormalChat:
|
||||
return no_action
|
||||
|
||||
# 执行额外的动作（不影响回复生成）
|
||||
action_result = await self._execute_action(action_type, action_data, message, thinking_id)
|
||||
action_result = await self._execute_action(action_type, action_data, message_data, thinking_id)
|
||||
if action_result is not None:
|
||||
logger.info(f"[{self.stream_name}] 额外动作 {action_type} 执行完成")
|
||||
else:
|
||||
@@ -556,17 +503,17 @@ class NormalChat:
|
||||
logger.error(f"[{self.stream_name}] Planner执行失败: {e}")
|
||||
return no_action
|
||||
|
||||
async def reply_one_message(self, message: MessageRecv) -> None:
|
||||
async def reply_one_message(self, message_data: dict) -> None:
|
||||
# 回复前处理
|
||||
await self.relationship_builder.build_relation()
|
||||
|
||||
thinking_id = await self._create_thinking_message(message)
|
||||
thinking_id = await self._create_thinking_message(message_data)
|
||||
|
||||
# 如果启用planner，预先修改可用actions（避免在并行任务中重复调用）
|
||||
available_actions = None
|
||||
if self.enable_planner:
|
||||
try:
|
||||
await self.action_modifier.modify_actions(mode="normal", message_content=message.processed_plain_text)
|
||||
await self.action_modifier.modify_actions(mode="normal", message_content=message_data.get("processed_plain_text"))
|
||||
available_actions = self.action_manager.get_using_actions_for_mode("normal")
|
||||
except Exception as e:
|
||||
logger.warning(f"[{self.stream_name}] 获取available_actions失败: {e}")
|
||||
@@ -576,8 +523,8 @@ class NormalChat:
|
||||
self.action_type = None # 初始化动作类型
|
||||
self.is_parallel_action = False # 初始化并行动作标志
|
||||
|
||||
gen_task = asyncio.create_task(self._generate_normal_response(message, available_actions))
|
||||
plan_task = asyncio.create_task(self._plan_and_execute_actions(message, thinking_id))
|
||||
gen_task = asyncio.create_task(self._generate_normal_response(message_data, available_actions))
|
||||
plan_task = asyncio.create_task(self._plan_and_execute_actions(message_data, thinking_id))
|
||||
|
||||
try:
|
||||
gather_timeout = global_config.chat.thinking_timeout
|
||||
@@ -661,7 +608,7 @@ class NormalChat:
|
||||
return False
|
||||
|
||||
# 发送回复 (不再需要传入 chat)
|
||||
first_bot_msg = await self._add_messages_to_manager(message, reply_texts, thinking_id)
|
||||
first_bot_msg = await self._add_messages_to_manager(message_data, reply_texts, thinking_id)
|
||||
|
||||
# 检查 first_bot_msg 是否为 None (例如思考消息已被移除的情况)
|
||||
if first_bot_msg:
|
||||
@@ -670,13 +617,13 @@ class NormalChat:
|
||||
# 记录回复信息到最近回复列表中
|
||||
reply_info = {
|
||||
"time": time.time(),
|
||||
"user_message": message.processed_plain_text,
|
||||
"user_message": message_data.get("processed_plain_text"),
|
||||
"user_info": {
|
||||
"user_id": message.message_info.user_info.user_id,
|
||||
"user_nickname": message.message_info.user_info.user_nickname,
|
||||
"user_id": message_data.get("user_id"),
|
||||
"user_nickname": message_data.get("user_nickname"),
|
||||
},
|
||||
"response": response_set,
|
||||
"is_reference_reply": message.reply is not None, # 判断是否为引用回复
|
||||
"is_reference_reply": message_data.get("reply") is not None, # 判断是否为引用回复
|
||||
}
|
||||
self.recent_replies.append(reply_info)
|
||||
# 保持最近回复历史在限定数量内
|
||||
@@ -688,8 +635,6 @@ class NormalChat:
|
||||
|
||||
async def start_chat(self):
|
||||
"""启动聊天任务。"""
|
||||
logger.debug(f"[{self.stream_name}] 开始启动聊天任务")
|
||||
|
||||
# 重置停用标志
|
||||
self._disabled = False
|
||||
|
||||
@@ -701,104 +646,90 @@ class NormalChat:
|
||||
# 清理可能存在的已完成任务引用
|
||||
if self._chat_task and self._chat_task.done():
|
||||
self._chat_task = None
|
||||
if self._priority_chat_task and self._priority_chat_task.done():
|
||||
self._priority_chat_task = None
|
||||
|
||||
try:
|
||||
logger.info(f"[{self.stream_name}] 创建新的聊天轮询任务，模式: {self.reply_mode}")
|
||||
|
||||
if self.reply_mode == "priority":
|
||||
polling_task_send = asyncio.create_task(self._priority_chat_loop())
|
||||
polling_task_recv = asyncio.create_task(self._priority_chat_loop_add_message())
|
||||
print("555")
|
||||
polling_task = asyncio.gather(polling_task_send, polling_task_recv)
|
||||
print("666")
|
||||
# Start producer loop
|
||||
producer_task = asyncio.create_task(self._priority_message_producer_loop())
|
||||
self._chat_task = producer_task
|
||||
self._chat_task.add_done_callback(lambda t: self._handle_task_completion(t, "priority_producer"))
|
||||
|
||||
else: # 默认或 "interest" 模式
|
||||
polling_task = asyncio.create_task(self._reply_interested_message())
|
||||
# Start consumer loop
|
||||
consumer_task = asyncio.create_task(self._priority_chat_loop())
|
||||
self._priority_chat_task = consumer_task
|
||||
self._priority_chat_task.add_done_callback(lambda t: self._handle_task_completion(t, "priority_consumer"))
|
||||
else: # Interest mode
|
||||
polling_task = asyncio.create_task(self._interest_message_polling_loop())
|
||||
self._chat_task = polling_task
|
||||
self._chat_task.add_done_callback(lambda t: self._handle_task_completion(t, "interest_polling"))
|
||||
|
||||
# 设置回调
|
||||
polling_task.add_done_callback(lambda t: self._handle_task_completion(t))
|
||||
|
||||
# 保存任务引用
|
||||
self._chat_task = polling_task
|
||||
self.running = True
|
||||
|
||||
logger.debug(f"[{self.stream_name}] 聊天任务启动完成")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[{self.stream_name}] 启动聊天任务失败: {e}")
|
||||
self._chat_task = None
|
||||
self._priority_chat_task = None
|
||||
raise
|
||||
|
||||
def _handle_task_completion(self, task: asyncio.Task):
|
||||
def _handle_task_completion(self, task: asyncio.Task, task_name: str = "unknown"):
|
||||
"""任务完成回调处理"""
|
||||
try:
|
||||
# 简化回调逻辑，避免复杂的异常处理
|
||||
logger.debug(f"[{self.stream_name}] 任务完成回调被调用")
|
||||
logger.debug(f"[{self.stream_name}] 任务 '{task_name}' 完成回调被调用")
|
||||
|
||||
# 检查是否是我们管理的任务
|
||||
if task is not self._chat_task:
|
||||
# 如果已经不是当前任务（可能在stop_chat中已被清空），直接返回
|
||||
logger.debug(f"[{self.stream_name}] 回调的任务不是当前管理的任务")
|
||||
if task is self._chat_task:
|
||||
self._chat_task = None
|
||||
elif task is self._priority_chat_task:
|
||||
self._priority_chat_task = None
|
||||
else:
|
||||
logger.debug(f"[{self.stream_name}] 回调的任务 '{task_name}' 不是当前管理的任务")
|
||||
return
|
||||
|
||||
# 清理任务引用
|
||||
self._chat_task = None
|
||||
logger.debug(f"[{self.stream_name}] 任务引用已清理")
|
||||
logger.debug(f"[{self.stream_name}] 任务 '{task_name}' 引用已清理")
|
||||
|
||||
# 简单记录任务状态，不进行复杂处理
|
||||
if task.cancelled():
|
||||
logger.debug(f"[{self.stream_name}] 任务已取消")
|
||||
logger.debug(f"[{self.stream_name}] 任务 '{task_name}' 已取消")
|
||||
elif task.done():
|
||||
try:
|
||||
# 尝试获取异常，但不抛出
|
||||
exc = task.exception()
|
||||
if exc:
|
||||
logger.error(f"[{self.stream_name}] 任务异常: {type(exc).__name__}: {exc}", exc_info=exc)
|
||||
else:
|
||||
logger.debug(f"[{self.stream_name}] 任务正常完成")
|
||||
except Exception as e:
|
||||
# 获取异常时也可能出错，静默处理
|
||||
logger.debug(f"[{self.stream_name}] 获取任务异常时出错: {e}")
|
||||
exc = task.exception()
|
||||
if exc:
|
||||
logger.error(f"[{self.stream_name}] 任务 '{task_name}' 异常: {type(exc).__name__}: {exc}", exc_info=exc)
|
||||
else:
|
||||
logger.debug(f"[{self.stream_name}] 任务 '{task_name}' 正常完成")
|
||||
|
||||
except Exception as e:
|
||||
# 回调函数中的任何异常都要捕获，避免影响系统
|
||||
logger.error(f"[{self.stream_name}] 任务完成回调处理出错: {e}")
|
||||
# 确保任务引用被清理
|
||||
self._chat_task = None
|
||||
self._priority_chat_task = None
|
||||
|
||||
# 改为实例方法, 移除 stream_id 参数
|
||||
async def stop_chat(self):
|
||||
"""停止当前实例的兴趣监控任务。"""
|
||||
logger.debug(f"[{self.stream_name}] 开始停止聊天任务")
|
||||
|
||||
# 立即设置停用标志，防止新任务启动
|
||||
self._disabled = True
|
||||
|
||||
# 如果没有运行中的任务，直接返回
|
||||
if not self._chat_task or self._chat_task.done():
|
||||
logger.debug(f"[{self.stream_name}] 没有运行中的任务，直接完成停止")
|
||||
self._chat_task = None
|
||||
return
|
||||
if self._chat_task and not self._chat_task.done():
|
||||
self._chat_task.cancel()
|
||||
if self._priority_chat_task and not self._priority_chat_task.done():
|
||||
self._priority_chat_task.cancel()
|
||||
|
||||
# 保存任务引用并立即清空，避免回调中的循环引用
|
||||
task_to_cancel = self._chat_task
|
||||
self._chat_task = None
|
||||
self._priority_chat_task = None
|
||||
|
||||
logger.debug(f"[{self.stream_name}] 取消聊天任务")
|
||||
|
||||
# 尝试优雅取消任务
|
||||
task_to_cancel.cancel()
|
||||
|
||||
# 异步清理思考消息，不阻塞当前流程
|
||||
asyncio.create_task(self._cleanup_thinking_messages_async())
|
||||
|
||||
async def _cleanup_thinking_messages_async(self):
|
||||
"""异步清理思考消息，避免阻塞主流程"""
|
||||
try:
|
||||
# 添加短暂延迟，让任务有时间响应取消
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
container = await message_manager.get_container(self.stream_id)
|
||||
if container:
|
||||
# 查找并移除所有 MessageThinking 类型的消息
|
||||
thinking_messages = [msg for msg in container.messages[:] if isinstance(msg, MessageThinking)]
|
||||
if thinking_messages:
|
||||
for msg in thinking_messages:
|
||||
@@ -806,7 +737,6 @@ class NormalChat:
|
||||
logger.info(f"[{self.stream_name}] 清理了 {len(thinking_messages)} 条未处理的思考消息。")
|
||||
except Exception as e:
|
||||
logger.error(f"[{self.stream_name}] 异步清理思考消息时出错: {e}")
|
||||
# 不打印完整栈跟踪，避免日志污染
|
||||
|
||||
def adjust_reply_frequency(self):
|
||||
"""
|
||||
@@ -879,7 +809,7 @@ class NormalChat:
|
||||
)
|
||||
|
||||
async def _execute_action(
|
||||
self, action_type: str, action_data: dict, message: MessageRecv, thinking_id: str
|
||||
self, action_type: str, action_data: dict, message_data: dict, thinking_id: str
|
||||
) -> Optional[bool]:
|
||||
"""执行具体的动作，只返回执行成功与否"""
|
||||
try:
|
||||
|
||||
@@ -23,9 +23,6 @@ from rich.traceback import install
|
||||
# 导入新的插件管理器
|
||||
from src.plugin_system.core.plugin_manager import plugin_manager
|
||||
|
||||
# 导入HFC性能记录器用于日志清理
|
||||
from src.chat.focus_chat.hfc_performance_logger import HFCPerformanceLogger
|
||||
|
||||
# 导入消息API和traceback模块
|
||||
from src.common.message import get_global_api
|
||||
|
||||
@@ -69,11 +66,6 @@ class MainSystem:
|
||||
"""初始化其他组件"""
|
||||
init_start_time = time.time()
|
||||
|
||||
# 清理HFC旧日志文件（保持目录大小在50MB以内）
|
||||
logger.info("开始清理HFC旧日志文件...")
|
||||
HFCPerformanceLogger.cleanup_old_logs(max_size_mb=50.0)
|
||||
logger.info("HFC日志清理完成")
|
||||
|
||||
# 添加在线时间统计任务
|
||||
await async_task_manager.add_task(OnlineTimeRecordTask())
|
||||
|
||||
|
||||
@@ -38,7 +38,7 @@ def init_prompt():
|
||||
现在，发送了消息，引起了你的注意，你对其进行了阅读和思考，请你输出一句话描述你新的情绪状态，不要输出任何其他内容
|
||||
请只输出情绪状态，不要输出其他内容：
|
||||
""",
|
||||
"change_mood_prompt",
|
||||
"change_mood_prompt_vtb",
|
||||
)
|
||||
Prompt(
|
||||
"""
|
||||
@@ -51,7 +51,7 @@ def init_prompt():
|
||||
距离你上次关注直播间消息已经过去了一段时间，你冷静了下来，请你输出一句话描述你现在的情绪状态
|
||||
请只输出情绪状态，不要输出其他内容：
|
||||
""",
|
||||
"regress_mood_prompt",
|
||||
"regress_mood_prompt_vtb",
|
||||
)
|
||||
Prompt(
|
||||
"""
|
||||
@@ -183,7 +183,7 @@ class ChatMood:
|
||||
|
||||
async def _update_text_mood():
|
||||
prompt = await global_prompt_manager.format_prompt(
|
||||
"change_mood_prompt",
|
||||
"change_mood_prompt_vtb",
|
||||
chat_talking_prompt=chat_talking_prompt,
|
||||
indentify_block=indentify_block,
|
||||
mood_state=self.mood_state,
|
||||
@@ -257,7 +257,7 @@ class ChatMood:
|
||||
|
||||
async def _regress_text_mood():
|
||||
prompt = await global_prompt_manager.format_prompt(
|
||||
"regress_mood_prompt",
|
||||
"regress_mood_prompt_vtb",
|
||||
chat_talking_prompt=chat_talking_prompt,
|
||||
indentify_block=indentify_block,
|
||||
mood_state=self.mood_state,
|
||||
|
||||
@@ -33,7 +33,6 @@ class BaseAction(ABC):
|
||||
thinking_id: str,
|
||||
chat_stream=None,
|
||||
log_prefix: str = "",
|
||||
shutting_down: bool = False,
|
||||
plugin_config: dict = None,
|
||||
**kwargs,
|
||||
):
|
||||
@@ -59,7 +58,6 @@ class BaseAction(ABC):
|
||||
self.cycle_timers = cycle_timers
|
||||
self.thinking_id = thinking_id
|
||||
self.log_prefix = log_prefix
|
||||
self.shutting_down = shutting_down
|
||||
|
||||
# 保存插件配置
|
||||
self.plugin_config = plugin_config or {}
|
||||
|
||||
Reference in New Issue
Block a user