better:优化hfc逻辑

This commit is contained in:
SengokuCola
2025-04-19 18:48:59 +08:00
parent b1dc34f7b1
commit c9ab9d4935
11 changed files with 227 additions and 475 deletions

View File

@@ -32,11 +32,10 @@ class ChatBot:
self.mood_manager.start_mood_update() # 启动情绪更新
self.think_flow_chat = ThinkFlowChat()
self.reasoning_chat = ReasoningChat()
self.heartFC_chat = HeartFC_Chat()
self.heartFC_processor = HeartFC_Processor(self.heartFC_chat)
self.only_process_chat = MessageProcessor()
self.heartFC_processor = HeartFC_Processor() # 新增
# 创建初始化PFC管理器的任务会在_ensure_started时执行
self.only_process_chat = MessageProcessor()
self.pfc_manager = PFCManager.get_instance()
async def _ensure_started(self):
@@ -120,7 +119,7 @@ class ChatBot:
else:
if groupinfo.group_id in global_config.talk_allowed_groups:
# logger.debug(f"开始群聊模式{str(message_data)[:50]}...")
if global_config.response_mode == "heart_flow":
if global_config.response_mode == "heart_FC":
# logger.info(f"启动最新最好的思维流FC模式{str(message_data)[:50]}...")
await self.heartFC_processor.process_message(message_data)

View File

@@ -30,7 +30,7 @@ chat_config = LogConfig(
logger = get_module_logger("heartFC_chat", config=chat_config)
# 新增常量
# 检测群聊兴趣的间隔时间
INTEREST_MONITOR_INTERVAL_SECONDS = 1
@@ -42,7 +42,6 @@ class HeartFC_Chat:
if HeartFC_Chat._instance is not None:
# Prevent re-initialization if used as a singleton
return
self.logger = logger # Make logger accessible via self
self.gpt = ResponseGenerator()
self.mood_manager = MoodManager.get_instance()
self.mood_manager.start_mood_update()
@@ -64,9 +63,8 @@ class HeartFC_Chat:
# --- End Added Class Method ---
async def start(self):
"""启动异步任务,如兴趣监控"""
logger.info("HeartFC_Chat 正在启动异步任务...")
await self.interest_manager.start_background_tasks()
"""启动异步任务,如回复启动"""
logger.debug("HeartFC_Chat 正在启动异步任务...")
self._initialize_monitor_task()
logger.info("HeartFC_Chat 异步任务启动完成")
@@ -76,7 +74,6 @@ class HeartFC_Chat:
try:
loop = asyncio.get_running_loop()
self._interest_monitor_task = loop.create_task(self._interest_monitor_loop())
logger.info(f"兴趣监控任务已创建。监控间隔: {INTEREST_MONITOR_INTERVAL_SECONDS}秒。")
except RuntimeError:
logger.error("创建兴趣监控任务失败:没有运行中的事件循环。")
raise
@@ -88,12 +85,12 @@ class HeartFC_Chat:
"""获取现有PFChatting实例或创建新实例。"""
async with self._pf_chatting_lock:
if stream_id not in self.pf_chatting_instances:
self.logger.info(f"为流 {stream_id} 创建新的PFChatting实例")
logger.info(f"为流 {stream_id} 创建新的PFChatting实例")
# 传递 self (HeartFC_Chat 实例) 进行依赖注入
instance = PFChatting(stream_id, self)
# 执行异步初始化
if not await instance._initialize():
self.logger.error(f"为流 {stream_id} 初始化PFChatting失败")
logger.error(f"为流 {stream_id} 初始化PFChatting失败")
return None
self.pf_chatting_instances[stream_id] = instance
return self.pf_chatting_instances[stream_id]
@@ -106,9 +103,8 @@ class HeartFC_Chat:
while True:
await asyncio.sleep(INTEREST_MONITOR_INTERVAL_SECONDS)
try:
# 从心流中获取活跃流
active_stream_ids = list(heartflow.get_all_subheartflows_streams_ids())
# logger.trace(f"检查 {len(active_stream_ids)} 个活跃流是否足以开启心流对话...") # 调试日志
for stream_id in active_stream_ids:
stream_name = chat_manager.get_stream_name(stream_id) or stream_id # 获取流名称
sub_hf = heartflow.get_subheartflow(stream_id)
@@ -121,8 +117,6 @@ class HeartFC_Chat:
interest_chatting = self.interest_manager.get_interest_chatting(stream_id)
if interest_chatting:
should_trigger = interest_chatting.should_evaluate_reply()
# if should_trigger:
# logger.info(f"[{stream_name}] 基于兴趣概率决定启动交流模式 (概率: {interest_chatting.current_reply_probability:.4f})。")
else:
logger.trace(
f"[{stream_name}] 没有找到对应的 InterestChatting 实例,跳过基于兴趣的触发检查。"
@@ -132,9 +126,9 @@ class HeartFC_Chat:
logger.error(traceback.format_exc())
if should_trigger:
# 启动一次麦麦聊天
pf_instance = await self._get_or_create_pf_chatting(stream_id)
if pf_instance:
# logger.info(f"[{stream_name}] 触发条件满足, 委托给PFChatting.")
asyncio.create_task(pf_instance.add_time())
else:
logger.error(f"[{stream_name}] 无法获取或创建PFChatting实例。跳过触发。")
@@ -282,6 +276,9 @@ class HeartFC_Chat:
)
self.mood_manager.update_mood_from_emotion(emotion, global_config.mood_intensity_factor)
#暂不使用
async def trigger_reply_generation(self, stream_id: str, observed_messages: List[dict]):
"""根据 SubHeartflow 的触发信号生成回复 (基于观察)"""
stream_name = chat_manager.get_stream_name(stream_id) or stream_id # <--- 在开始时获取名称
@@ -428,10 +425,7 @@ class HeartFC_Chat:
text = msg_dict.get("detailed_plain_text", "")
if text:
context_texts.append(text)
observation_context_text = "\n".join(context_texts)
logger.debug(
f"[{stream_name}] Context for tools:\n{observation_context_text[-200:]}..."
) # 打印部分上下文
observation_context_text = " ".join(context_texts)
else:
logger.warning(f"[{stream_name}] observed_messages 列表为空,无法为工具提供上下文。")
@@ -541,9 +535,3 @@ class HeartFC_Chat:
# 可以在这里添加清理逻辑,如果有的话
pass
# --- 结束重构 ---
# _create_thinking_message, _send_response_messages, _handle_emoji, _update_relationship
# 这几个辅助方法目前仍然依赖 MessageRecv 对象。
# 如果无法可靠地从 Observation 获取并重建最后一条消息的 MessageRecv
# 或者希望回复不锚定具体消息,那么这些方法也需要进一步重构。

View File

@@ -26,15 +26,37 @@ logger = get_module_logger("heartFC_processor", config=processor_config)
class HeartFC_Processor:
def __init__(self, chat_instance: HeartFC_Chat):
def __init__(self):
self.storage = MessageStorage()
self.interest_manager = (
InterestManager()
) # TODO: 可能需要传递 chat_instance 给 InterestManager 或修改其方法签名
self.chat_instance = chat_instance # 持有 HeartFC_Chat 实例
)
# self.chat_instance = chat_instance # 持有 HeartFC_Chat 实例
async def process_message(self, message_data: str) -> None:
"""处理接收到的消息,更新状态,并将回复决策委托给 InterestManager"""
"""处理接收到的原始消息数据,完成消息解析、缓冲、过滤、存储、兴趣度计算与更新等核心流程。
此函数是消息处理的核心入口,负责接收原始字符串格式的消息数据,并将其转化为结构化的 `MessageRecv` 对象。
主要执行步骤包括:
1. 解析 `message_data` 为 `MessageRecv` 对象,提取用户信息、群组信息等。
2. 将消息加入 `message_buffer` 进行缓冲处理,以应对消息轰炸或者某些人一条消息分几次发等情况。
3. 获取或创建对应的 `chat_stream` 和 `subheartflow` 实例,用于管理会话状态和心流。
4. 对消息内容进行初步处理(如提取纯文本)。
5. 应用全局配置中的过滤词和正则表达式,过滤不符合规则的消息。
6. 查询消息缓冲结果,如果消息被缓冲器拦截(例如,判断为消息轰炸的一部分),则中止后续处理。
7. 对于通过缓冲的消息,将其存储到 `MessageStorage` 中。
8. 调用海马体(`HippocampusManager`)计算消息内容的记忆激活率。(这部分算法后续会进行优化)
9. 根据是否被提及(@)和记忆激活率,计算最终的兴趣度增量。(提及的额外兴趣增幅)
10. 使用计算出的增量更新 `InterestManager` 中对应会话的兴趣度。
11. 记录处理后的消息信息及当前的兴趣度到日志。
注意:此函数本身不负责生成和发送回复。回复的决策和生成逻辑被移至 `HeartFC_Chat` 类中的监控任务,
该任务会根据 `InterestManager` 中的兴趣度变化来决定何时触发回复。
Args:
message_data: str: 从消息源接收到的原始消息字符串。
"""
timing_results = {} # 初始化 timing_results
message = None
try:
@@ -60,7 +82,6 @@ class HeartFC_Processor:
message.update_chat_stream(chat)
# 创建心流与chat的观察 (在接收消息时创建,以便后续观察和思考)
heartflow.create_subheartflow(chat.stream_id)
await message.process()

View File

@@ -21,11 +21,11 @@ logger = get_module_logger("InterestManager", config=interest_log_config)
DEFAULT_DECAY_RATE_PER_SECOND = 0.98 # 每秒衰减率 (兴趣保留 99%)
MAX_INTEREST = 15.0 # 最大兴趣值
# MIN_INTEREST_THRESHOLD = 0.1 # 低于此值可能被清理 (可选)
CLEANUP_INTERVAL_SECONDS = 3600 # 清理任务运行间隔 (例如:1小时)
INACTIVE_THRESHOLD_SECONDS = 3600 # 不活跃时间阈值 (例如:1小时)
CLEANUP_INTERVAL_SECONDS = 1200 # 清理任务运行间隔 (例如:20分钟)
INACTIVE_THRESHOLD_SECONDS = 1200 # 不活跃时间阈值 (例如:20分钟)
LOG_INTERVAL_SECONDS = 3 # 日志记录间隔 (例如30秒)
LOG_DIRECTORY = "logs/interest" # 日志目录
LOG_FILENAME = "interest_log.json" # 快照日志文件名 (保留,以防其他地方用到)
# LOG_FILENAME = "interest_log.json" # 快照日志文件名 (保留,以防其他地方用到)
HISTORY_LOG_FILENAME = "interest_history.log" # 新的历史日志文件名
# 移除阈值,将移至 HeartFC_Chat
# INTEREST_INCREASE_THRESHOLD = 0.5
@@ -54,7 +54,6 @@ class InterestChatting:
self.last_update_time: float = time.time() # 同时作为兴趣和概率的更新时间基准
self.decay_rate_per_second: float = decay_rate
self.max_interest: float = max_interest
self.last_increase_amount: float = 0.0
self.last_interaction_time: float = self.last_update_time # 新增:最后交互时间
# --- 新增:概率回复相关属性 ---
@@ -131,15 +130,7 @@ class InterestChatting:
# 限制概率不超过最大值
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 是否有效
else:
if 0 < self.probability_decay_factor < 1:
decay_multiplier = math.pow(self.probability_decay_factor, time_delta)
# old_prob = self.current_reply_probability
@@ -167,8 +158,6 @@ class InterestChatting:
# 先更新概率和计算衰减(基于上次更新时间)
self._update_reply_probability(current_time)
self._calculate_decay(current_time)
# 记录这次增加的具体数值,供外部判断是否触发
self.last_increase_amount = value
# 应用增加
self.interest_level += value
self.interest_level = min(self.interest_level, self.max_interest) # 不超过最大值
@@ -185,10 +174,6 @@ class InterestChatting:
self.last_update_time = current_time # 降低也更新时间戳
self.last_interaction_time = current_time # 更新最后交互时间
def reset_trigger_info(self):
"""重置触发相关信息,在外部任务处理后调用"""
self.last_increase_amount = 0.0
def get_interest(self) -> float:
"""获取当前兴趣值 (计算衰减后)"""
# 注意:这个方法现在会触发概率和兴趣的更新
@@ -262,7 +247,7 @@ class InterestManager:
# key: stream_id (str), value: InterestChatting instance
self.interest_dict: dict[str, InterestChatting] = {}
# 保留旧的快照文件路径变量,尽管此任务不再写入
self._snapshot_log_file_path = os.path.join(LOG_DIRECTORY, LOG_FILENAME)
# self._snapshot_log_file_path = os.path.join(LOG_DIRECTORY, LOG_FILENAME)
# 定义新的历史日志文件路径
self._history_log_file_path = os.path.join(LOG_DIRECTORY, HISTORY_LOG_FILENAME)
self._ensure_log_directory()
@@ -412,13 +397,8 @@ class InterestManager:
def _get_or_create_interest_chatting(self, stream_id: str) -> InterestChatting:
"""获取或创建指定流的 InterestChatting 实例 (线程安全)"""
# 由于字典操作本身在 CPython 中大部分是原子的,
# 且主要写入发生在 __init__ 和 cleanup (由单任务执行)
# 读取和 get_or_create 主要在事件循环线程,简单场景下可能不需要锁。
# 但为保险起见或跨线程使用考虑,可加锁。
# with self._lock:
if stream_id not in self.interest_dict:
logger.debug(f"Creating new InterestChatting for stream_id: {stream_id}")
logger.debug(f"创建兴趣流: {stream_id}")
# --- 修改:创建时传入概率相关参数 (如果需要定制化,否则使用默认值) ---
self.interest_dict[stream_id] = InterestChatting(
# decay_rate=..., max_interest=..., # 可以从配置读取

View File

@@ -13,6 +13,8 @@ from src.plugins.chat.chat_stream import chat_manager
from .messagesender import MessageManager
from src.common.logger import get_module_logger, LogConfig, DEFAULT_CONFIG # 引入 DEFAULT_CONFIG
from src.plugins.models.utils_model import LLMRequest
from src.plugins.chat.utils import parse_text_timestamps
from src.plugins.person_info.relationship_manager import relationship_manager
# 定义日志配置 (使用 loguru 格式)
interest_log_config = LogConfig(
@@ -102,8 +104,8 @@ class PFChatting:
async def _initialize(self) -> bool:
"""
Lazy initialization to resolve chat_stream and sub_hf using the provided identifier.
Ensures the instance is ready to handle triggers.
懒初始化以使用提供的标识符解析chat_streamsub_hf
确保实例已准备好处理触发器。
"""
async with self._init_lock:
if self._initialized:
@@ -171,7 +173,7 @@ class PFChatting:
# Start the loop if it wasn't active and timer is positive
if not self._loop_active and self._loop_timer > 0:
logger.info(f"{log_prefix} 麦麦有兴趣!开始聊天")
# logger.info(f"{log_prefix} 麦麦有兴趣!开始聊天")
self._loop_active = True
if self._loop_task and not self._loop_task.done():
logger.warning(f"{log_prefix} 发现意外的循环任务正在进行。取消它。")
@@ -363,9 +365,17 @@ class PFChatting:
async def _planner(self) -> Dict[str, Any]:
"""
规划器 (Planner): 使用LLM根据上下文决定是否和如何回复。
Returns a dictionary containing the decision and context.
{'action': str, 'reasoning': str, 'emoji_query': str, 'current_mind': str,
'send_emoji_from_tools': str, 'observed_messages': List[dict]}
返回:
dict: 包含决策和上下文的字典,结构如下:
{
'action': str, # 执行动作 (不回复/文字回复/表情包)
'reasoning': str, # 决策理由
'emoji_query': str, # 表情包查询词
'current_mind': str, # 当前心理状态
'send_emoji_from_tools': str, # 工具推荐的表情包
'observed_messages': List[dict] # 观察到的消息列表
}
"""
log_prefix = self._get_log_prefix()
observed_messages: List[dict] = []
@@ -376,14 +386,15 @@ class PFChatting:
# --- 获取最新的观察信息 ---
try:
if self.sub_hf and self.sub_hf._get_primary_observation():
observation = self.sub_hf._get_primary_observation()
logger.debug(f"{log_prefix}[Planner] 调用 observation.observe()...")
observation = self.sub_hf._get_primary_observation() # Call only once
if observation: # Now check if the result is truthy
# logger.debug(f"{log_prefix}[Planner] 调用 observation.observe()...")
await observation.observe() # 主动观察以获取最新消息
observed_messages = observation.talking_message # 获取更新后的消息列表
logger.debug(f"{log_prefix}[Planner] 获取到 {len(observed_messages)}观察消息。")
logger.debug(f"{log_prefix}[Planner] 观察获取到 {len(observed_messages)} 条消息。")
else:
logger.warning(f"{log_prefix}[Planner] 无法获取 SubHeartflow 或 Observation 来获取消息")
logger.warning(f"{log_prefix}[Planner] 无法获取 Observation。")
except Exception as e:
logger.error(f"{log_prefix}[Planner] 获取观察信息时出错: {e}")
logger.error(traceback.format_exc())
@@ -396,49 +407,30 @@ class PFChatting:
context_texts = [
msg.get("detailed_plain_text", "") for msg in observed_messages if msg.get("detailed_plain_text")
]
observation_context_text = "\n".join(context_texts)
logger.debug(f"{log_prefix}[Planner] Context for tools: {observation_context_text[:100]}...")
if observation_context_text and self.sub_hf:
# Ensure SubHeartflow exists for tool use context
tool_result = await self.heartfc_chat.tool_user.use_tool(
message_txt=observation_context_text, chat_stream=self.chat_stream, sub_heartflow=self.sub_hf
)
if tool_result.get("used_tools", False):
tool_result_info = tool_result.get("structured_info", {})
logger.debug(f"{log_prefix}[Planner] Tool results: {tool_result_info}")
if "mid_chat_mem" in tool_result_info:
get_mid_memory_id = [
mem["content"] for mem in tool_result_info["mid_chat_mem"] if "content" in mem
]
if "send_emoji" in tool_result_info and tool_result_info["send_emoji"]:
send_emoji_from_tools = tool_result_info["send_emoji"][0].get("content", "") # Use renamed var
elif not self.sub_hf:
logger.warning(f"{log_prefix}[Planner] Skipping tool use because SubHeartflow is not available.")
observation_context_text = " ".join(context_texts)
# logger.debug(f"{log_prefix}[Planner] Context for tools: {observation_context_text[:100]}...")
tool_result = await self.heartfc_chat.tool_user.use_tool(
message_txt=observation_context_text, chat_stream=self.chat_stream, sub_heartflow=self.sub_hf
)
if tool_result.get("used_tools", False):
tool_result_info = tool_result.get("structured_info", {})
logger.debug(f"{log_prefix}[Planner] 规划前工具结果: {tool_result_info}")
if "mid_chat_mem" in tool_result_info:
get_mid_memory_id = [
mem["content"] for mem in tool_result_info["mid_chat_mem"] if "content" in mem
]
except Exception as e_tool:
logger.error(f"{log_prefix}[Planner] Tool use failed: {e_tool}")
# Continue even if tool use fails
logger.error(f"{log_prefix}[Planner] 规划前工具使用失败: {e_tool}")
# --- 结束工具使用 ---
# 心流思考然后plan
try:
if self.sub_hf:
# Ensure arguments match the current do_thinking_before_reply signature
current_mind, past_mind = await self.sub_hf.do_thinking_before_reply(
chat_stream=self.chat_stream,
extra_info=tool_result_info,
obs_id=get_mid_memory_id,
)
logger.info(f"{log_prefix}[Planner] SubHeartflow thought: {current_mind}")
else:
logger.warning(f"{log_prefix}[Planner] Skipping SubHeartflow thinking because it is not available.")
current_mind = "[心流思考不可用]" # Set a default/indicator value
except Exception as e_shf:
logger.error(f"{log_prefix}[Planner] SubHeartflow thinking failed: {e_shf}")
logger.error(traceback.format_exc())
current_mind = "[心流思考出错]"
current_mind, _past_mind = await self.sub_hf.do_thinking_before_reply(
chat_stream=self.chat_stream,
extra_info=tool_result_info,
obs_id=get_mid_memory_id,
)
# --- 使用 LLM 进行决策 ---
action = "no_reply" # Default action
@@ -448,8 +440,8 @@ class PFChatting:
try:
# 构建提示 (Now includes current_mind)
prompt = self._build_planner_prompt(observed_messages, current_mind)
logger.debug(f"{log_prefix}[Planner] Prompt: {prompt}")
prompt = await self._build_planner_prompt(observed_messages, current_mind)
logger.debug(f"{log_prefix}[Planner] 规划器 Prompt: {prompt}")
# 准备 LLM 请求 Payload
payload = {
@@ -459,7 +451,6 @@ class PFChatting:
"tool_choice": {"type": "function", "function": {"name": "decide_reply_action"}}, # 强制调用此工具
}
logger.debug(f"{log_prefix}[Planner] 发送 Planner LLM 请求...")
# 调用 LLM
response = await self.planner_llm._execute_request(
endpoint="/chat/completions", payload=payload, prompt=prompt
@@ -717,7 +708,7 @@ class PFChatting:
logger.info(f"{self._get_log_prefix()} PFChatting shutdown complete.")
def _build_planner_prompt(self, observed_messages: List[dict], current_mind: Optional[str]) -> str:
async def _build_planner_prompt(self, observed_messages: List[dict], current_mind: Optional[str]) -> str:
"""构建 Planner LLM 的提示词 (现在包含 current_mind)"""
prompt = f"你的名字是 {global_config.BOT_NICKNAME}。是一个女大二学生正在QQ聊天正在决定是否以及如何回应当前的聊天。\n"
@@ -749,6 +740,10 @@ class PFChatting:
prompt += "4. 如果你已经回复过消息,也没有人又回复你,选择'no_reply'"
prompt += "必须调用 'decide_reply_action' 工具并提供 'action''reasoning'"
prompt = await relationship_manager.convert_all_person_sign_to_person_name(prompt)
prompt = parse_text_timestamps(prompt, mode="lite")
return prompt
# --- 回复器 (Replier) 的定义 --- #
@@ -771,7 +766,7 @@ class PFChatting:
# --- Tool Use and SubHF Thinking are now in _planner ---
# --- Generate Response with LLM ---
logger.debug(f"{log_prefix}[Replier-{thinking_id}] Calling LLM to generate response...")
# logger.debug(f"{log_prefix}[Replier-{thinking_id}] Calling LLM to generate response...")
# 注意:实际的生成调用是在 self.heartfc_chat.gpt.generate_response 中
response_set = await self.heartfc_chat.gpt.generate_response(
anchor_message,
@@ -785,7 +780,7 @@ class PFChatting:
return None # Indicate failure
# --- 准备并返回结果 ---
logger.info(f"{log_prefix}[Replier-{thinking_id}] 成功生成了回复集: {' '.join(response_set)[:50]}...")
logger.info(f"{log_prefix}[Replier-{thinking_id}] 成功生成了回复集: {' '.join(response_set)[:100]}...")
return {
"response_set": response_set,
"send_emoji": send_emoji, # Pass through the emoji determined earlier (usually by tools)