diff --git a/interest_monitor_gui.py b/interest_monitor_gui.py index 2185912a7..5b19d4808 100644 --- a/interest_monitor_gui.py +++ b/interest_monitor_gui.py @@ -5,32 +5,33 @@ import os from datetime import datetime, timedelta import random from collections import deque -import json # 引入 json +import json # 引入 json # --- 引入 Matplotlib --- from matplotlib.figure import Figure from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg -import matplotlib.dates as mdates # 用于处理日期格式 -import matplotlib # 导入 matplotlib +import matplotlib.dates as mdates # 用于处理日期格式 +import matplotlib # 导入 matplotlib # --- 配置 --- -LOG_FILE_PATH = os.path.join("logs", "interest", "interest_history.log") # 指向历史日志文件 -REFRESH_INTERVAL_MS = 200 # 刷新间隔 (毫秒) - 可以适当调长，因为读取文件可能耗时 +LOG_FILE_PATH = os.path.join("logs", "interest", "interest_history.log") # 指向历史日志文件 +REFRESH_INTERVAL_MS = 200 # 刷新间隔 (毫秒) - 可以适当调长，因为读取文件可能耗时 WINDOW_TITLE = "Interest Monitor (Live History)" -MAX_HISTORY_POINTS = 1000 # 图表上显示的最大历史点数 (可以增加) -MAX_STREAMS_TO_DISPLAY = 15 # 最多显示多少个聊天流的折线图 (可以增加) +MAX_HISTORY_POINTS = 1000 # 图表上显示的最大历史点数 (可以增加) +MAX_STREAMS_TO_DISPLAY = 15 # 最多显示多少个聊天流的折线图 (可以增加) # *** 添加 Matplotlib 中文字体配置 *** # 尝试使用 'SimHei' 或 'Microsoft YaHei'，如果找不到，matplotlib 会回退到默认字体 # 确保你的系统上安装了这些字体 -matplotlib.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei'] -matplotlib.rcParams['axes.unicode_minus'] = False # 解决负号'-'显示为方块的问题 +matplotlib.rcParams["font.sans-serif"] = ["SimHei", "Microsoft YaHei"] +matplotlib.rcParams["axes.unicode_minus"] = False # 解决负号'-'显示为方块的问题 + class InterestMonitorApp: def __init__(self, root): self.root = root self.root.title(WINDOW_TITLE) - self.root.geometry("1800x800") # 调整窗口大小以适应图表 + self.root.geometry("1800x800") # 调整窗口大小以适应图表 # --- 数据存储 --- # 使用 deque 来存储有限的历史数据点 @@ -38,9 +39,9 @@ class InterestMonitorApp: self.stream_history = {} # key: stream_id, value: deque([(timestamp, reply_probability), ...]) # <--- 新增：存储概率历史 self.probability_history = {} - self.stream_colors = {} # 为每个 stream 分配颜色 - self.stream_display_names = {} # *** New: Store display names (group_name) *** - self.selected_stream_id = tk.StringVar() # 用于 Combobox 绑定 + self.stream_colors = {} # 为每个 stream 分配颜色 + self.stream_display_names = {} # *** New: Store display names (group_name) *** + self.selected_stream_id = tk.StringVar() # 用于 Combobox 绑定 # --- UI 元素 --- # 创建 Notebook (选项卡控件) @@ -49,7 +50,7 @@ class InterestMonitorApp: # --- 第一个选项卡：所有流 --- self.frame_all = ttk.Frame(self.notebook, padding="5 5 5 5") - self.notebook.add(self.frame_all, text='所有聊天流') + self.notebook.add(self.frame_all, text="所有聊天流") # 状态标签 self.status_label = tk.Label(root, text="Initializing...", anchor="w", fg="grey") @@ -61,36 +62,40 @@ class InterestMonitorApp: # 配置在 update_plot 中进行，避免重复 # 创建 Tkinter 画布嵌入 Matplotlib 图表 (用于第一个选项卡) - self.canvas = FigureCanvasTkAgg(self.fig, master=self.frame_all) # <--- 放入 frame_all + self.canvas = FigureCanvasTkAgg(self.fig, master=self.frame_all) # <--- 放入 frame_all self.canvas_widget = self.canvas.get_tk_widget() self.canvas_widget.pack(side=tk.TOP, fill=tk.BOTH, expand=1) # --- 第二个选项卡：单个流 --- self.frame_single = ttk.Frame(self.notebook, padding="5 5 5 5") - self.notebook.add(self.frame_single, text='单个聊天流详情') + self.notebook.add(self.frame_single, text="单个聊天流详情") # 单个流选项卡的上部控制区域 self.control_frame_single = ttk.Frame(self.frame_single) self.control_frame_single.pack(side=tk.TOP, fill=tk.X, pady=5) ttk.Label(self.control_frame_single, text="选择聊天流:").pack(side=tk.LEFT, padx=(0, 5)) - self.stream_selector = ttk.Combobox(self.control_frame_single, textvariable=self.selected_stream_id, state="readonly", width=50) + self.stream_selector = ttk.Combobox( + self.control_frame_single, textvariable=self.selected_stream_id, state="readonly", width=50 + ) self.stream_selector.pack(side=tk.LEFT, fill=tk.X, expand=True) self.stream_selector.bind("<>", self.on_stream_selected) # Matplotlib 图表设置 (用于第二个选项卡) self.fig_single = Figure(figsize=(5, 4), dpi=100) # 修改：创建两个子图，一个显示兴趣度，一个显示概率 - self.ax_single_interest = self.fig_single.add_subplot(211) # 2行1列的第1个 - self.ax_single_probability = self.fig_single.add_subplot(212, sharex=self.ax_single_interest) # 2行1列的第2个，共享X轴 + self.ax_single_interest = self.fig_single.add_subplot(211) # 2行1列的第1个 + self.ax_single_probability = self.fig_single.add_subplot( + 212, sharex=self.ax_single_interest + ) # 2行1列的第2个，共享X轴 # 创建 Tkinter 画布嵌入 Matplotlib 图表 (用于第二个选项卡) - self.canvas_single = FigureCanvasTkAgg(self.fig_single, master=self.frame_single) # <--- 放入 frame_single + self.canvas_single = FigureCanvasTkAgg(self.fig_single, master=self.frame_single) # <--- 放入 frame_single self.canvas_widget_single = self.canvas_single.get_tk_widget() self.canvas_widget_single.pack(side=tk.TOP, fill=tk.BOTH, expand=1) # --- 初始化和启动刷新 --- - self.update_display() # 首次加载并开始刷新循环 + self.update_display() # 首次加载并开始刷新循环 def on_stream_selected(self, event=None): """当 Combobox 选择改变时调用，更新单个流的图表""" @@ -110,15 +115,15 @@ class InterestMonitorApp: # *** Reset display names each time we reload *** new_stream_history = {} new_stream_display_names = {} - new_probability_history = {} # <--- 重置概率历史 + new_probability_history = {} # <--- 重置概率历史 read_count = 0 error_count = 0 # *** Calculate the timestamp threshold for the last 30 minutes *** current_time = time.time() - time_threshold = current_time - (15 * 60) # 30 minutes in seconds + time_threshold = current_time - (15 * 60) # 30 minutes in seconds try: - with open(LOG_FILE_PATH, 'r', encoding='utf-8') as f: + with open(LOG_FILE_PATH, "r", encoding="utf-8") as f: for line in f: read_count += 1 try: @@ -127,28 +132,30 @@ class InterestMonitorApp: # *** Add time filtering *** if timestamp is None or float(timestamp) < time_threshold: - continue # Skip old or invalid entries + continue # Skip old or invalid entries stream_id = log_entry.get("stream_id") interest_level = log_entry.get("interest_level") - group_name = log_entry.get("group_name", stream_id) # *** Get group_name, fallback to stream_id *** - reply_probability = log_entry.get("reply_probability") # <--- 获取概率值 + group_name = log_entry.get( + "group_name", stream_id + ) # *** Get group_name, fallback to stream_id *** + reply_probability = log_entry.get("reply_probability") # <--- 获取概率值 # *** Check other required fields AFTER time filtering *** if stream_id is None or interest_level is None: - error_count += 1 - continue # 跳过无效行 + error_count += 1 + continue # 跳过无效行 # 如果是第一次读到这个 stream_id，则创建 deque if stream_id not in new_stream_history: new_stream_history[stream_id] = deque(maxlen=MAX_HISTORY_POINTS) - new_probability_history[stream_id] = deque(maxlen=MAX_HISTORY_POINTS) # <--- 创建概率 deque + new_probability_history[stream_id] = deque(maxlen=MAX_HISTORY_POINTS) # <--- 创建概率 deque # 检查是否已有颜色，没有则分配 if stream_id not in self.stream_colors: self.stream_colors[stream_id] = self.get_random_color() - + # *** Store the latest display name found for this stream_id *** - new_stream_display_names[stream_id] = group_name + new_stream_display_names[stream_id] = group_name # 添加数据点 new_stream_history[stream_id].append((float(timestamp), float(interest_level))) @@ -163,22 +170,22 @@ class InterestMonitorApp: except json.JSONDecodeError: error_count += 1 # logger.warning(f"Skipping invalid JSON line: {line.strip()}") - continue # 跳过无法解析的行 + continue # 跳过无法解析的行 except (TypeError, ValueError): - error_count += 1 - # logger.warning(f"Skipping line due to data type error ({e}): {line.strip()}") - continue # 跳过数据类型错误的行 + error_count += 1 + # logger.warning(f"Skipping line due to data type error ({e}): {line.strip()}") + continue # 跳过数据类型错误的行 # 读取完成后，用新数据替换旧数据 self.stream_history = new_stream_history - self.stream_display_names = new_stream_display_names # *** Update display names *** - self.probability_history = new_probability_history # <--- 更新概率历史 + self.stream_display_names = new_stream_display_names # *** Update display names *** + self.probability_history = new_probability_history # <--- 更新概率历史 status_msg = f"Data loaded at {datetime.now().strftime('%H:%M:%S')}. Lines read: {read_count}." if error_count > 0: - status_msg += f" Skipped {error_count} invalid lines." - self.set_status(status_msg, "orange") + status_msg += f" Skipped {error_count} invalid lines." + self.set_status(status_msg, "orange") else: - self.set_status(status_msg, "green") + self.set_status(status_msg, "green") except IOError as e: self.set_status(f"Error reading file {LOG_FILE_PATH}: {e}", "red") @@ -192,12 +199,16 @@ class InterestMonitorApp: """更新单个流选项卡中的 Combobox 列表""" # 创建 (display_name, stream_id) 对的列表，按 display_name 排序 available_streams = sorted( - [(name, sid) for sid, name in self.stream_display_names.items() if sid in self.stream_history and self.stream_history[sid]], - key=lambda item: item[0] # 按显示名称排序 + [ + (name, sid) + for sid, name in self.stream_display_names.items() + if sid in self.stream_history and self.stream_history[sid] + ], + key=lambda item: item[0], # 按显示名称排序 ) # 更新 Combobox 的值 (仅显示 display_name) - self.stream_selector['values'] = [name for name, sid in available_streams] + self.stream_selector["values"] = [name for name, sid in available_streams] # 检查当前选中的 stream_id 是否仍然有效 current_selection_name = self.selected_stream_id.get() @@ -211,28 +222,28 @@ class InterestMonitorApp: elif not available_streams: # 如果没有可选流，清空选择 self.selected_stream_id.set("") - self.update_single_stream_plot() # 清空图表 + self.update_single_stream_plot() # 清空图表 def update_all_streams_plot(self): """更新第一个选项卡的 Matplotlib 图表 (显示所有流)""" - self.ax.clear() # 清除旧图 + self.ax.clear() # 清除旧图 # *** 设置中文标题和标签 *** self.ax.set_title("兴趣度随时间变化图 (所有活跃流)") self.ax.set_xlabel("时间") self.ax.set_ylabel("兴趣度") - self.ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M:%S')) + self.ax.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S")) self.ax.grid(True) - self.ax.set_ylim(0, 10) # 固定 Y 轴范围 0-10 + self.ax.set_ylim(0, 10) # 固定 Y 轴范围 0-10 # 只绘制最新的 N 个 stream (按最后记录的兴趣度排序) # 注意：现在是基于文件读取的快照排序，可能不是实时最新 active_streams = sorted( self.stream_history.items(), - key=lambda item: item[1][-1][1] if item[1] else 0, # 按最后兴趣度排序 - reverse=True + key=lambda item: item[1][-1][1] if item[1] else 0, # 按最后兴趣度排序 + reverse=True, )[:MAX_STREAMS_TO_DISPLAY] - all_times = [] # 用于确定 X 轴范围 + all_times = [] # 用于确定 X 轴范围 for stream_id, history in active_streams: if not history: @@ -242,52 +253,50 @@ class InterestMonitorApp: # 将 time.time() 时间戳转换为 matplotlib 可识别的日期格式 try: mpl_dates = [datetime.fromtimestamp(ts) for ts in timestamps] - all_times.extend(mpl_dates) # 收集所有时间点 - + all_times.extend(mpl_dates) # 收集所有时间点 + # *** Use display name for label *** display_label = self.stream_display_names.get(stream_id, stream_id) self.ax.plot( mpl_dates, interests, - label=display_label, # *** Use display_label *** - color=self.stream_colors.get(stream_id, 'grey'), - marker='.', - markersize=3, - linestyle='-', - linewidth=1 + label=display_label, # *** Use display_label *** + color=self.stream_colors.get(stream_id, "grey"), + marker=".", + markersize=3, + linestyle="-", + linewidth=1, ) except ValueError as e: - print(f"Skipping plot for {stream_id} due to invalid timestamp: {e}") - continue + print(f"Skipping plot for {stream_id} due to invalid timestamp: {e}") + continue if all_times: - # 根据数据动态调整 X 轴范围，留一点边距 - min_time = min(all_times) - max_time = max(all_times) - # delta = max_time - min_time - # self.ax.set_xlim(min_time - delta * 0.05, max_time + delta * 0.05) - self.ax.set_xlim(min_time, max_time) + # 根据数据动态调整 X 轴范围，留一点边距 + min_time = min(all_times) + max_time = max(all_times) + # delta = max_time - min_time + # self.ax.set_xlim(min_time - delta * 0.05, max_time + delta * 0.05) + self.ax.set_xlim(min_time, max_time) - # 自动格式化X轴标签 - self.fig.autofmt_xdate() + # 自动格式化X轴标签 + self.fig.autofmt_xdate() else: # 如果没有数据，设置一个默认的时间范围，例如最近一小时 now = datetime.now() one_hour_ago = now - timedelta(hours=1) self.ax.set_xlim(one_hour_ago, now) - # 添加图例 if active_streams: - # 调整图例位置和大小 - # 字体已通过全局 matplotlib.rcParams 设置 - self.ax.legend(loc='upper left', bbox_to_anchor=(1.02, 1), borderaxespad=0., fontsize='x-small') - # 调整布局，确保图例不被裁剪 - self.fig.tight_layout(rect=[0, 0, 0.85, 1]) # 右侧留出空间给图例 + # 调整图例位置和大小 + # 字体已通过全局 matplotlib.rcParams 设置 + self.ax.legend(loc="upper left", bbox_to_anchor=(1.02, 1), borderaxespad=0.0, fontsize="x-small") + # 调整布局，确保图例不被裁剪 + self.fig.tight_layout(rect=[0, 0, 0.85, 1]) # 右侧留出空间给图例 - - self.canvas.draw() # 重绘画布 + self.canvas.draw() # 重绘画布 def update_single_stream_plot(self): """更新第二个选项卡的 Matplotlib 图表 (显示单个选定的流)""" @@ -298,14 +307,14 @@ class InterestMonitorApp: self.ax_single_interest.set_title("兴趣度") self.ax_single_interest.set_ylabel("兴趣度") self.ax_single_interest.grid(True) - self.ax_single_interest.set_ylim(0, 10) # 固定 Y 轴范围 0-10 + self.ax_single_interest.set_ylim(0, 10) # 固定 Y 轴范围 0-10 self.ax_single_probability.set_title("回复评估概率") self.ax_single_probability.set_xlabel("时间") self.ax_single_probability.set_ylabel("概率") self.ax_single_probability.grid(True) - self.ax_single_probability.set_ylim(0, 1.05) # 固定 Y 轴范围 0-1 - self.ax_single_probability.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M:%S')) + self.ax_single_probability.set_ylim(0, 1.05) # 固定 Y 轴范围 0-1 + self.ax_single_probability.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S")) selected_name = self.selected_stream_id.get() selected_sid = None @@ -317,7 +326,7 @@ class InterestMonitorApp: selected_sid = sid break - all_times = [] # 用于确定 X 轴范围 + all_times = [] # 用于确定 X 轴范围 # --- 新增：绘制兴趣度图 --- if selected_sid and selected_sid in self.stream_history and self.stream_history[selected_sid]: @@ -329,14 +338,14 @@ class InterestMonitorApp: self.ax_single_interest.plot( mpl_dates, interests, - color=self.stream_colors.get(selected_sid, 'blue'), - marker='.', + color=self.stream_colors.get(selected_sid, "blue"), + marker=".", markersize=3, - linestyle='-', - linewidth=1 + linestyle="-", + linewidth=1, ) except ValueError as e: - print(f"Skipping interest plot for {selected_sid} due to invalid timestamp: {e}") + print(f"Skipping interest plot for {selected_sid} due to invalid timestamp: {e}") # --- 新增：绘制概率图 --- if selected_sid and selected_sid in self.probability_history and self.probability_history[selected_sid]: @@ -345,28 +354,28 @@ class InterestMonitorApp: try: prob_mpl_dates = [datetime.fromtimestamp(ts) for ts in prob_timestamps] # 注意：概率图的时间点可能与兴趣度不同，也需要加入 all_times - all_times.extend(prob_mpl_dates) + all_times.extend(prob_mpl_dates) self.ax_single_probability.plot( prob_mpl_dates, probabilities, - color=self.stream_colors.get(selected_sid, 'green'), # 可以用不同颜色 - marker='.', + color=self.stream_colors.get(selected_sid, "green"), # 可以用不同颜色 + marker=".", markersize=3, - linestyle='-', - linewidth=1 + linestyle="-", + linewidth=1, ) except ValueError as e: - print(f"Skipping probability plot for {selected_sid} due to invalid timestamp: {e}") + print(f"Skipping probability plot for {selected_sid} due to invalid timestamp: {e}") # --- 新增：调整 X 轴范围和格式 --- if all_times: - min_time = min(all_times) - max_time = max(all_times) - # 设置共享的 X 轴范围 - self.ax_single_interest.set_xlim(min_time, max_time) - # self.ax_single_probability.set_xlim(min_time, max_time) # sharex 会自动同步 - # 自动格式化X轴标签 (应用到共享轴的最后一个子图上通常即可) - self.fig_single.autofmt_xdate() + min_time = min(all_times) + max_time = max(all_times) + # 设置共享的 X 轴范围 + self.ax_single_interest.set_xlim(min_time, max_time) + # self.ax_single_probability.set_xlim(min_time, max_time) # sharex 会自动同步 + # 自动格式化X轴标签 (应用到共享轴的最后一个子图上通常即可) + self.fig_single.autofmt_xdate() else: # 如果没有数据，设置一个默认的时间范围 now = datetime.now() @@ -380,16 +389,17 @@ class InterestMonitorApp: def update_display(self): """主更新循环""" try: - self.load_and_update_history() # 从文件加载数据并更新内部状态 + self.load_and_update_history() # 从文件加载数据并更新内部状态 # *** 修改：分别调用两个图表的更新方法 *** - self.update_all_streams_plot() # 更新所有流的图表 - self.update_single_stream_plot() # 更新单个流的图表 + self.update_all_streams_plot() # 更新所有流的图表 + self.update_single_stream_plot() # 更新单个流的图表 except Exception as e: # 提供更详细的错误信息 import traceback + error_msg = f"Error during update: {e}\n{traceback.format_exc()}" self.set_status(error_msg, "red") - print(error_msg) # 打印详细错误到控制台 + print(error_msg) # 打印详细错误到控制台 # 安排下一次刷新 self.root.after(REFRESH_INTERVAL_MS, self.update_display) @@ -398,13 +408,14 @@ class InterestMonitorApp: """更新状态栏标签""" # 限制状态栏消息长度 max_len = 150 - display_message = (message[:max_len] + '...') if len(message) > max_len else message + display_message = (message[:max_len] + "...") if len(message) > max_len else message self.status_label.config(text=display_message, fg=color) if __name__ == "__main__": # 导入 timedelta 用于默认时间范围 from datetime import timedelta + root = tk.Tk() app = InterestMonitorApp(root) - root.mainloop() \ No newline at end of file + root.mainloop() diff --git a/src/do_tool/tool_use.py b/src/do_tool/tool_use.py index dd01f3f75..f054ecdd4 100644 --- a/src/do_tool/tool_use.py +++ b/src/do_tool/tool_use.py @@ -22,9 +22,7 @@ class ToolUser: ) @staticmethod - async def _build_tool_prompt( - message_txt: str, chat_stream: ChatStream, subheartflow: SubHeartflow = None - ): + async def _build_tool_prompt(message_txt: str, chat_stream: ChatStream, subheartflow: SubHeartflow = None): """构建工具使用的提示词 Args: @@ -114,9 +112,7 @@ class ToolUser: logger.error(f"执行工具调用时发生错误: {str(e)}") return None - async def use_tool( - self, message_txt: str, chat_stream: ChatStream, sub_heartflow: SubHeartflow = None - ): + async def use_tool(self, message_txt: str, chat_stream: ChatStream, sub_heartflow: SubHeartflow = None): """使用工具辅助思考，判断是否需要额外信息 Args: diff --git a/src/heart_flow/observation.py b/src/heart_flow/observation.py index 89ee7bb90..3bb66efc2 100644 --- a/src/heart_flow/observation.py +++ b/src/heart_flow/observation.py @@ -39,7 +39,7 @@ class ChattingObservation(Observation): self.mid_memory_info = "" self.now_message_info = "" - self._observe_lock = asyncio.Lock() # 添加锁 + self._observe_lock = asyncio.Lock() # 添加锁 self.llm_summary = LLMRequest( model=global_config.llm_observation, temperature=0.7, max_tokens=300, request_type="chat_observation" @@ -73,7 +73,7 @@ class ChattingObservation(Observation): return self.now_message_info async def observe(self): - async with self._observe_lock: # 获取锁 + async with self._observe_lock: # 获取锁 # 查找新消息，最多获取 self.max_now_obs_len 条 new_messages_cursor = ( db.messages.find({"chat_id": self.chat_id, "time": {"$gt": self.last_observe_time}}) @@ -87,30 +87,38 @@ class ChattingObservation(Observation): # 如果没有获取到限制数量内的较新消息，可能仍然有更早的消息，但我们只关注最近的 # 检查是否有任何新消息（即使超出限制），以决定是否更新 last_observe_time # 注意：这里的查询也可能与其他并发 observe 冲突，但锁保护了状态更新 - any_new_message = db.messages.find_one({"chat_id": self.chat_id, "time": {"$gt": self.last_observe_time}}) + any_new_message = db.messages.find_one( + {"chat_id": self.chat_id, "time": {"$gt": self.last_observe_time}} + ) if not any_new_message: - return # 确实没有新消息 + return # 确实没有新消息 # 如果有超过限制的更早的新消息，仍然需要更新时间戳，防止重复获取旧消息 # 但不将它们加入 talking_message - latest_message_time_cursor = db.messages.find({"chat_id": self.chat_id, "time": {"$gt": self.last_observe_time}}).sort("time", -1).limit(1) + latest_message_time_cursor = ( + db.messages.find({"chat_id": self.chat_id, "time": {"$gt": self.last_observe_time}}) + .sort("time", -1) + .limit(1) + ) latest_time_doc = next(latest_message_time_cursor, None) if latest_time_doc: # 确保只在严格大于时更新，避免因并发查询导致时间戳回退 if latest_time_doc["time"] > self.last_observe_time: self.last_observe_time = latest_time_doc["time"] - return # 返回，因为我们只关心限制内的最新消息 + return # 返回，因为我们只关心限制内的最新消息 # 在持有锁的情况下，再次过滤，确保只处理真正新的消息 # 防止处理在等待锁期间已被其他协程处理的消息 truly_new_messages = [msg for msg in new_messages if msg["time"] > self.last_observe_time] if not truly_new_messages: - logger.debug(f"Chat {self.chat_id}: Fetched messages, but already processed by another concurrent observe call.") - return # 所有获取的消息都已被处理 + logger.debug( + f"Chat {self.chat_id}: Fetched messages, but already processed by another concurrent observe call." + ) + return # 所有获取的消息都已被处理 # 如果获取到了 truly_new_messages (在限制内且时间戳大于上次记录) - self.last_observe_time = truly_new_messages[-1]["time"] # 更新时间戳为获取到的最新消息的时间 + self.last_observe_time = truly_new_messages[-1]["time"] # 更新时间戳为获取到的最新消息的时间 self.talking_message.extend(truly_new_messages) @@ -124,21 +132,23 @@ class ChattingObservation(Observation): # 锁保证了这部分逻辑的原子性 if len(self.talking_message) > self.max_now_obs_len: - try: # 使用 try...finally 仅用于可能的LLM调用错误处理 + try: # 使用 try...finally 仅用于可能的LLM调用错误处理 # 计算需要移除的消息数量，保留最新的 max_now_obs_len 条 messages_to_remove_count = len(self.talking_message) - self.max_now_obs_len oldest_messages = self.talking_message[:messages_to_remove_count] - self.talking_message = self.talking_message[messages_to_remove_count:] # 保留后半部分，即最新的 - oldest_messages_str = "\n".join([msg["detailed_plain_text"] for msg in oldest_messages if "detailed_plain_text" in msg]) # 增加检查 + self.talking_message = self.talking_message[messages_to_remove_count:] # 保留后半部分，即最新的 + oldest_messages_str = "\n".join( + [msg["detailed_plain_text"] for msg in oldest_messages if "detailed_plain_text" in msg] + ) # 增加检查 oldest_timestamps = [msg["time"] for msg in oldest_messages] # 调用 LLM 总结主题 prompt = f"请总结以下聊天记录的主题：\n{oldest_messages_str}\n主题,用一句话概括包括人物事件和主要信息，不要分点：" - summary = "无法总结主题" # 默认值 + summary = "无法总结主题" # 默认值 try: summary_result, _ = await self.llm_summary.generate_response_async(prompt) - if summary_result: # 确保结果不为空 - summary = summary_result + if summary_result: # 确保结果不为空 + summary = summary_result except Exception as e: logger.error(f"总结主题失败 for chat {self.chat_id}: {e}") # 保留默认总结 "无法总结主题" @@ -146,7 +156,7 @@ class ChattingObservation(Observation): mid_memory = { "id": str(int(datetime.now().timestamp())), "theme": summary, - "messages": oldest_messages, # 存储原始消息对象 + "messages": oldest_messages, # 存储原始消息对象 "timestamps": oldest_timestamps, "chat_id": self.chat_id, "created_at": datetime.now().timestamp(), @@ -155,16 +165,16 @@ class ChattingObservation(Observation): # 存入内存中的 mid_memorys self.mid_memorys.append(mid_memory) if len(self.mid_memorys) > self.max_mid_memory_len: - self.mid_memorys.pop(0) # 移除最旧的 + self.mid_memorys.pop(0) # 移除最旧的 mid_memory_str = "之前聊天的内容概括是：\n" - for mid_memory_item in self.mid_memorys: # 重命名循环变量以示区分 + for mid_memory_item in self.mid_memorys: # 重命名循环变量以示区分 time_diff = int((datetime.now().timestamp() - mid_memory_item["created_at"]) / 60) mid_memory_str += f"距离现在{time_diff}分钟前(聊天记录id:{mid_memory_item['id']})：{mid_memory_item['theme']}\n" self.mid_memory_info = mid_memory_str - except Exception as e: # 将异常处理移至此处以覆盖整个总结过程 + except Exception as e: # 将异常处理移至此处以覆盖整个总结过程 logger.error(f"处理和总结旧消息时出错 for chat {self.chat_id}: {e}") - traceback.print_exc() # 记录详细堆栈 + traceback.print_exc() # 记录详细堆栈 # print(f"处理后self.talking_message：{self.talking_message}") @@ -173,7 +183,9 @@ class ChattingObservation(Observation): now_message_str += self.translate_message_list_to_str(talking_message=self.talking_message) self.now_message_info = now_message_str - logger.debug(f"Chat {self.chat_id} - 压缩早期记忆：{self.mid_memory_info}\n现在聊天内容：{self.now_message_info}") + logger.debug( + f"Chat {self.chat_id} - 压缩早期记忆：{self.mid_memory_info}\n现在聊天内容：{self.now_message_info}" + ) # 锁在退出 async with 块时自动释放 async def update_talking_summary(self, new_messages_str): diff --git a/src/heart_flow/sub_heartflow.py b/src/heart_flow/sub_heartflow.py index 19f95c010..9704168ad 100644 --- a/src/heart_flow/sub_heartflow.py +++ b/src/heart_flow/sub_heartflow.py @@ -60,8 +60,8 @@ def init_prompt(): prompt += "现在你接下去继续思考，产生新的想法，记得保留你刚刚的想法，不要分点输出，输出连贯的内心独白" prompt += "不要太长，但是记得结合上述的消息，要记得你的人设，关注聊天和新内容，关注你回复的内容，不要思考太多:" Prompt(prompt, "sub_heartflow_prompt_after") - - # prompt += f"你现在正在做的事情是：{schedule_info}\n" + + # prompt += f"你现在正在做的事情是：{schedule_info}\n" prompt += "{extra_info}\n" prompt += "{prompt_personality}\n" prompt += "现在是{time_now}，你正在上网，和qq群里的网友们聊天，群里正在聊的话题是：\n{chat_observe_info}\n" @@ -115,7 +115,7 @@ class SubHeartflow: self.running_knowledges = [] - self._thinking_lock = asyncio.Lock() # 添加思考锁，防止并发思考 + self._thinking_lock = asyncio.Lock() # 添加思考锁，防止并发思考 self.bot_name = global_config.BOT_NICKNAME @@ -165,8 +165,10 @@ class SubHeartflow: # 检查是否超过指定时间没有激活 (例如，没有被调用进行思考) if current_time - self.last_active_time > global_config.sub_heart_flow_stop_time: # 例如 5 分钟 - logger.info(f"子心流 {self.subheartflow_id} 超过 {global_config.sub_heart_flow_stop_time} 秒没有激活，正在销毁..." - f" (Last active: {datetime.fromtimestamp(self.last_active_time).strftime('%Y-%m-%d %H:%M:%S')})") + logger.info( + f"子心流 {self.subheartflow_id} 超过 {global_config.sub_heart_flow_stop_time} 秒没有激活，正在销毁..." + f" (Last active: {datetime.fromtimestamp(self.last_active_time).strftime('%Y-%m-%d %H:%M:%S')})" + ) # 在这里添加实际的销毁逻辑，例如从主 Heartflow 管理器中移除自身 # heartflow.remove_subheartflow(self.subheartflow_id) # 假设有这样的方法 break # 退出循环以停止任务 @@ -176,7 +178,7 @@ class SubHeartflow: # await self.do_a_thinking() # await self.judge_willing() - await asyncio.sleep(global_config.sub_heart_flow_update_interval) # 定期检查销毁条件 + await asyncio.sleep(global_config.sub_heart_flow_update_interval) # 定期检查销毁条件 async def ensure_observed(self): """确保在思考前执行了观察""" @@ -198,20 +200,23 @@ class SubHeartflow: logger.error(f"[{self.subheartflow_id}] do_observe called but no valid observation found.") async def do_thinking_before_reply( - self, chat_stream: ChatStream, extra_info: str, obs_id: list[str] = None # 修改 obs_id 类型为 list[str] + self, + chat_stream: ChatStream, + extra_info: str, + obs_id: list[str] = None, # 修改 obs_id 类型为 list[str] ): - async with self._thinking_lock: # 获取思考锁 + async with self._thinking_lock: # 获取思考锁 # --- 在思考前确保观察已执行 --- # await self.ensure_observed() - self.last_active_time = time.time() # 更新最后激活时间戳 + self.last_active_time = time.time() # 更新最后激活时间戳 current_thinking_info = self.current_mind mood_info = self.current_state.mood observation = self._get_primary_observation() if not observation: logger.error(f"[{self.subheartflow_id}] Cannot perform thinking without observation.") - return "", [] # 返回空结果 + return "", [] # 返回空结果 # --- 获取观察信息 --- # chat_observe_info = "" @@ -220,13 +225,14 @@ class SubHeartflow: chat_observe_info = observation.get_observe_info(obs_id) logger.debug(f"[{self.subheartflow_id}] Using specific observation IDs: {obs_id}") except Exception as e: - logger.error(f"[{self.subheartflow_id}] Error getting observe info with IDs {obs_id}: {e}. Falling back.") - chat_observe_info = observation.get_observe_info() # 出错时回退到默认观察 + logger.error( + f"[{self.subheartflow_id}] Error getting observe info with IDs {obs_id}: {e}. Falling back." + ) + chat_observe_info = observation.get_observe_info() # 出错时回退到默认观察 else: chat_observe_info = observation.get_observe_info() logger.debug(f"[{self.subheartflow_id}] Using default observation info.") - # --- 构建 Prompt (基本逻辑不变) --- # extra_info_prompt = "" if extra_info: @@ -235,19 +241,19 @@ class SubHeartflow: for item in tool_data: extra_info_prompt += f"- {item['name']}: {item['content']}\n" else: - extra_info_prompt = "无工具信息。\n" # 提供默认值 + extra_info_prompt = "无工具信息。\n" # 提供默认值 individuality = Individuality.get_instance() prompt_personality = f"你的名字是{self.bot_name}，你" prompt_personality += individuality.personality.personality_core - + # 添加随机性格侧面 if individuality.personality.personality_sides: random_side = random.choice(individuality.personality.personality_sides) prompt_personality += f"，{random_side}" - + # 添加随机身份细节 - if individuality.identity.identity_detail: + if individuality.identity.identity_detail: random_detail = random.choice(individuality.identity.identity_detail) prompt_personality += f"，{random_detail}" @@ -273,12 +279,12 @@ class SubHeartflow: try: response, reasoning_content = await self.llm_model.generate_response_async(prompt) - if not response: # 如果 LLM 返回空，给一个默认想法 + if not response: # 如果 LLM 返回空，给一个默认想法 response = "(不知道该想些什么...)" logger.warning(f"[{self.subheartflow_id}] LLM returned empty response for thinking.") except Exception as e: logger.error(f"[{self.subheartflow_id}] 内心独白获取失败: {e}") - response = "(思考时发生错误...)" # 错误时的默认想法 + response = "(思考时发生错误...)" # 错误时的默认想法 self.update_current_mind(response) @@ -448,9 +454,9 @@ class SubHeartflow: async def check_reply_trigger(self) -> bool: """根据观察到的信息和内部状态，判断是否应该触发一次回复。 - TODO: 实现具体的判断逻辑。 - 例如：检查 self.observations[0].now_message_info 是否包含提及、问题， - 或者 self.current_mind 中是否包含强烈的回复意图等。 + TODO: 实现具体的判断逻辑。 + 例如：检查 self.observations[0].now_message_info 是否包含提及、问题， + 或者 self.current_mind 中是否包含强烈的回复意图等。 """ # Placeholder: 目前始终返回 False，需要后续实现 logger.trace(f"[{self.subheartflow_id}] check_reply_trigger called. (Logic Pending)") @@ -462,7 +468,7 @@ class SubHeartflow: # logger.info(f"[{self.subheartflow_id}] Triggering reply based on mention.") # return True # ------------------ # - return False # 默认不触发 + return False # 默认不触发 init_prompt() diff --git a/src/main.py b/src/main.py index 898096a9a..9867e06af 100644 --- a/src/main.py +++ b/src/main.py @@ -112,7 +112,7 @@ class MainSystem: logger.success("心流系统启动成功") # 启动 InterestManager 的后台任务 - interest_manager = InterestManager() # 获取单例 + interest_manager = InterestManager() # 获取单例 await interest_manager.start_background_tasks() logger.success("InterestManager 后台任务启动成功") diff --git a/src/plugins/chat/bot.py b/src/plugins/chat/bot.py index a4a843cb9..5c124952b 100644 --- a/src/plugins/chat/bot.py +++ b/src/plugins/chat/bot.py @@ -122,11 +122,9 @@ class ChatBot: # logger.debug(f"开始群聊模式{str(message_data)[:50]}...") if global_config.response_mode == "heart_flow": # logger.info(f"启动最新最好的思维流FC模式{str(message_data)[:50]}...") - + await self.heartFC_processor.process_message(message_data) - - - + elif global_config.response_mode == "reasoning": # logger.debug(f"开始推理模式{str(message_data)[:50]}...") await self.reasoning_chat.process_message(message_data) diff --git a/src/plugins/chat_module/heartFC_chat/heartFC__generator.py b/src/plugins/chat_module/heartFC_chat/heartFC__generator.py index 9f3e577c8..21cdf1ee2 100644 --- a/src/plugins/chat_module/heartFC_chat/heartFC__generator.py +++ b/src/plugins/chat_module/heartFC_chat/heartFC__generator.py @@ -37,7 +37,11 @@ class ResponseGenerator: self.current_model_type = "r1" # 默认使用 R1 self.current_model_name = "unknown model" - async def generate_response(self, message: MessageRecv, thinking_id: str,) -> Optional[List[str]]: + async def generate_response( + self, + message: MessageRecv, + thinking_id: str, + ) -> Optional[List[str]]: """根据当前模型类型选择对应的生成函数""" logger.info( @@ -47,16 +51,12 @@ class ResponseGenerator: arousal_multiplier = MoodManager.get_instance().get_arousal_multiplier() with Timer() as t_generate_response: - current_model = self.model_normal - current_model.temperature = ( - global_config.llm_normal["temp"] * arousal_multiplier - ) # 激活度越高，温度越高 + current_model.temperature = global_config.llm_normal["temp"] * arousal_multiplier # 激活度越高，温度越高 model_response = await self._generate_response_with_model( message, current_model, thinking_id, mode="normal" ) - if model_response: logger.info( f"{global_config.BOT_NICKNAME}的回复是：{model_response},生成回复时间: {t_generate_response.human_readable}" diff --git a/src/plugins/chat_module/heartFC_chat/heartFC_chat.py b/src/plugins/chat_module/heartFC_chat/heartFC_chat.py index e2ce81613..d6c77a864 100644 --- a/src/plugins/chat_module/heartFC_chat/heartFC_chat.py +++ b/src/plugins/chat_module/heartFC_chat/heartFC_chat.py @@ -33,15 +33,16 @@ logger = get_module_logger("heartFC_chat", config=chat_config) # 新增常量 INTEREST_MONITOR_INTERVAL_SECONDS = 1 + class HeartFC_Chat: - _instance = None # For potential singleton access if needed by MessageManager + _instance = None # For potential singleton access if needed by MessageManager def __init__(self): # --- Updated Init --- 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.logger = logger # Make logger accessible via self self.gpt = ResponseGenerator() self.mood_manager = MoodManager.get_instance() self.mood_manager.start_mood_update() @@ -52,13 +53,14 @@ class HeartFC_Chat: self.pf_chatting_instances: Dict[str, PFChatting] = {} self._pf_chatting_lock = Lock() # --- End New PFChatting Management --- - HeartFC_Chat._instance = self # Register instance + HeartFC_Chat._instance = self # Register instance # --- End Updated Init --- # --- Added Class Method for Singleton Access --- @classmethod def get_instance(cls): return cls._instance + # --- End Added Class Method --- async def start(self): @@ -76,8 +78,8 @@ class HeartFC_Chat: self._interest_monitor_task = loop.create_task(self._interest_monitor_loop()) logger.info(f"兴趣监控任务已创建。监控间隔: {INTEREST_MONITOR_INTERVAL_SECONDS}秒。") except RuntimeError: - logger.error("创建兴趣监控任务失败：没有运行中的事件循环。") - raise + logger.error("创建兴趣监控任务失败：没有运行中的事件循环。") + raise else: logger.warning("跳过兴趣监控任务创建：任务已存在或正在运行。") @@ -95,6 +97,7 @@ class HeartFC_Chat: return None self.pf_chatting_instances[stream_id] = instance return self.pf_chatting_instances[stream_id] + # --- End Added PFChatting Instance Manager --- async def _interest_monitor_loop(self): @@ -107,7 +110,7 @@ class HeartFC_Chat: # 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 # 获取流名称 + stream_name = chat_manager.get_stream_name(stream_id) or stream_id # 获取流名称 sub_hf = heartflow.get_subheartflow(stream_id) if not sub_hf: logger.warning(f"监控循环: 无法获取活跃流 {stream_name} 的 sub_hf") @@ -121,7 +124,9 @@ class HeartFC_Chat: # if should_trigger: # logger.info(f"[{stream_name}] 基于兴趣概率决定启动交流模式 (概率: {interest_chatting.current_reply_probability:.4f})。") else: - logger.trace(f"[{stream_name}] 没有找到对应的 InterestChatting 实例，跳过基于兴趣的触发检查。") + logger.trace( + f"[{stream_name}] 没有找到对应的 InterestChatting 实例，跳过基于兴趣的触发检查。" + ) except Exception as e: logger.error(f"检查兴趣触发器时出错 流 {stream_name}: {e}") logger.error(traceback.format_exc()) @@ -140,7 +145,7 @@ class HeartFC_Chat: except Exception as e: logger.error(f"兴趣监控循环错误: {e}") logger.error(traceback.format_exc()) - await asyncio.sleep(5) # 发生错误时等待 + await asyncio.sleep(5) # 发生错误时等待 async def _create_thinking_message(self, anchor_message: Optional[MessageRecv]): """创建思考消息 (尝试锚定到 anchor_message)""" @@ -162,14 +167,16 @@ class HeartFC_Chat: message_id=thinking_id, chat_stream=chat, bot_user_info=bot_user_info, - reply=anchor_message, # 回复的是锚点消息 + reply=anchor_message, # 回复的是锚点消息 thinking_start_time=thinking_time_point, ) MessageManager().add_message(thinking_message) return thinking_id - async def _send_response_messages(self, anchor_message: Optional[MessageRecv], response_set: List[str], thinking_id) -> Optional[MessageSending]: + async def _send_response_messages( + self, anchor_message: Optional[MessageRecv], response_set: List[str], thinking_id + ) -> Optional[MessageSending]: """发送回复消息 (尝试锚定到 anchor_message)""" if not anchor_message or not anchor_message.chat_stream: logger.error("无法发送回复，缺少有效的锚点消息或聊天流。") @@ -184,7 +191,7 @@ class HeartFC_Chat: container.messages.remove(msg) break if not thinking_message: - stream_name = chat_manager.get_stream_name(chat.stream_id) or chat.stream_id # 获取流名称 + stream_name = chat_manager.get_stream_name(chat.stream_id) or chat.stream_id # 获取流名称 logger.warning(f"[{stream_name}] 未找到对应的思考消息 {thinking_id}，可能已超时被移除") return None @@ -195,16 +202,16 @@ class HeartFC_Chat: for msg_text in response_set: message_segment = Seg(type="text", data=msg_text) bot_message = MessageSending( - message_id=thinking_id, # 使用 thinking_id 作为批次标识 + message_id=thinking_id, # 使用 thinking_id 作为批次标识 chat_stream=chat, bot_user_info=UserInfo( user_id=global_config.BOT_QQ, user_nickname=global_config.BOT_NICKNAME, platform=anchor_message.message_info.platform, ), - sender_info=anchor_message.message_info.user_info, # 发送给锚点消息的用户 + sender_info=anchor_message.message_info.user_info, # 发送给锚点消息的用户 message_segment=message_segment, - reply=anchor_message, # 回复锚点消息 + reply=anchor_message, # 回复锚点消息 is_head=not mark_head, is_emoji=False, thinking_start_time=thinking_start_time, @@ -214,19 +221,19 @@ class HeartFC_Chat: first_bot_msg = bot_message message_set.add_message(bot_message) - if message_set.messages: # 确保有消息才添加 + if message_set.messages: # 确保有消息才添加 MessageManager().add_message(message_set) return first_bot_msg else: - stream_name = chat_manager.get_stream_name(chat.stream_id) or chat.stream_id # 获取流名称 + stream_name = chat_manager.get_stream_name(chat.stream_id) or chat.stream_id # 获取流名称 logger.warning(f"[{stream_name}] 没有生成有效的回复消息集，无法发送。") return None async def _handle_emoji(self, anchor_message: Optional[MessageRecv], response_set, send_emoji=""): """处理表情包 (尝试锚定到 anchor_message)""" if not anchor_message or not anchor_message.chat_stream: - logger.error("无法处理表情包，缺少有效的锚点消息或聊天流。") - return + logger.error("无法处理表情包，缺少有效的锚点消息或聊天流。") + return chat = anchor_message.chat_stream if send_emoji: @@ -242,7 +249,7 @@ class HeartFC_Chat: thinking_time_point = round(time.time(), 2) message_segment = Seg(type="emoji", data=emoji_cq) bot_message = MessageSending( - message_id="me" + str(thinking_time_point), # 使用不同的 ID 前缀? + message_id="me" + str(thinking_time_point), # 使用不同的 ID 前缀? chat_stream=chat, bot_user_info=UserInfo( user_id=global_config.BOT_QQ, @@ -251,7 +258,7 @@ class HeartFC_Chat: ), sender_info=anchor_message.message_info.user_info, message_segment=message_segment, - reply=anchor_message, # 回复锚点消息 + reply=anchor_message, # 回复锚点消息 is_head=False, is_emoji=True, ) @@ -260,8 +267,8 @@ class HeartFC_Chat: async def _update_relationship(self, anchor_message: Optional[MessageRecv], response_set): """更新关系情绪 (尝试基于 anchor_message)""" if not anchor_message or not anchor_message.chat_stream: - logger.error("无法更新关系情绪，缺少有效的锚点消息或聊天流。") - return + logger.error("无法更新关系情绪，缺少有效的锚点消息或聊天流。") + return # 关系更新依赖于理解回复是针对谁的，以及原始消息的上下文 # 这里的实现可能需要调整，取决于关系管理器如何工作 @@ -269,18 +276,18 @@ class HeartFC_Chat: # 注意：anchor_message.processed_plain_text 是锚点消息的文本，不一定是思考的全部上下文 stance, emotion = await self.gpt._get_emotion_tags(ori_response, anchor_message.processed_plain_text) await relationship_manager.calculate_update_relationship_value( - chat_stream=anchor_message.chat_stream, # 使用锚点消息的流 + chat_stream=anchor_message.chat_stream, # 使用锚点消息的流 label=emotion, - stance=stance + stance=stance, ) 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 # <--- 在开始时获取名称 + stream_name = chat_manager.get_stream_name(stream_id) or stream_id # <--- 在开始时获取名称 chat = None sub_hf = None - anchor_message: Optional[MessageRecv] = None # <--- 重命名，用于锚定回复的消息对象 + anchor_message: Optional[MessageRecv] = None # <--- 重命名，用于锚定回复的消息对象 userinfo: Optional[UserInfo] = None messageinfo: Optional[BaseMessageInfo] = None @@ -303,9 +310,9 @@ class HeartFC_Chat: logger.error(f"[{stream_name}] 无法找到子心流对象，无法生成回复。") return except Exception as e: - logger.error(f"[{stream_name}] 获取 ChatStream 或 SubHeartflow 时出错: {e}") - logger.error(traceback.format_exc()) - return + logger.error(f"[{stream_name}] 获取 ChatStream 或 SubHeartflow 时出错: {e}") + logger.error(traceback.format_exc()) + return # --- 2. 尝试从 observed_messages 重建最后一条消息作为锚点, 失败则创建占位符 --- # try: @@ -314,36 +321,49 @@ class HeartFC_Chat: if observed_messages: try: last_msg_dict = observed_messages[-1] - logger.debug(f"[{stream_name}] Attempting to reconstruct MessageRecv from last observed message.") + logger.debug( + f"[{stream_name}] Attempting to reconstruct MessageRecv from last observed message." + ) anchor_message = MessageRecv(last_msg_dict, chat_stream=chat) - if not (anchor_message and anchor_message.message_info and anchor_message.message_info.message_id and anchor_message.message_info.user_info): + if not ( + anchor_message + and anchor_message.message_info + and anchor_message.message_info.message_id + and anchor_message.message_info.user_info + ): raise ValueError("Reconstructed MessageRecv missing essential info.") userinfo = anchor_message.message_info.user_info messageinfo = anchor_message.message_info - logger.debug(f"[{stream_name}] Successfully reconstructed anchor message: ID={messageinfo.message_id}, Sender={userinfo.user_nickname}") + logger.debug( + f"[{stream_name}] Successfully reconstructed anchor message: ID={messageinfo.message_id}, Sender={userinfo.user_nickname}" + ) except Exception as e_reconstruct: - logger.warning(f"[{stream_name}] Reconstructing MessageRecv from observed message failed: {e_reconstruct}. Will create placeholder.") + logger.warning( + f"[{stream_name}] Reconstructing MessageRecv from observed message failed: {e_reconstruct}. Will create placeholder." + ) reconstruction_failed = True else: - logger.warning(f"[{stream_name}] observed_messages is empty. Will create placeholder anchor message.") - reconstruction_failed = True # Treat empty observed_messages as a failure to reconstruct + logger.warning( + f"[{stream_name}] observed_messages is empty. Will create placeholder anchor message." + ) + reconstruction_failed = True # Treat empty observed_messages as a failure to reconstruct # 如果重建失败或 observed_messages 为空，创建占位符 if reconstruction_failed: - placeholder_id = f"mid_{int(time.time() * 1000)}" # 使用毫秒时间戳增加唯一性 + placeholder_id = f"mid_{int(time.time() * 1000)}" # 使用毫秒时间戳增加唯一性 placeholder_user = UserInfo(user_id="system_trigger", user_nickname="系统触发") placeholder_msg_info = BaseMessageInfo( message_id=placeholder_id, platform=chat.platform, group_info=chat.group_info, user_info=placeholder_user, - time=time.time() + time=time.time(), # 其他 BaseMessageInfo 可能需要的字段设为默认值或 None ) # 创建 MessageRecv 实例，注意它需要消息字典结构，我们创建一个最小化的 placeholder_msg_dict = { "message_info": placeholder_msg_info.to_dict(), - "processed_plain_text": "", # 提供空文本 + "processed_plain_text": "", # 提供空文本 "raw_message": "", "time": placeholder_msg_info.time, } @@ -353,18 +373,20 @@ class HeartFC_Chat: anchor_message.update_chat_stream(chat) userinfo = anchor_message.message_info.user_info messageinfo = anchor_message.message_info - logger.info(f"[{stream_name}] Created placeholder anchor message: ID={messageinfo.message_id}, Sender={userinfo.user_nickname}") + logger.info( + f"[{stream_name}] Created placeholder anchor message: ID={messageinfo.message_id}, Sender={userinfo.user_nickname}" + ) except Exception as e: logger.error(f"[{stream_name}] 获取或创建锚点消息时出错: {e}") logger.error(traceback.format_exc()) - anchor_message = None # 确保出错时 anchor_message 为 None + anchor_message = None # 确保出错时 anchor_message 为 None # --- 4. 检查并发思考限制 (使用 anchor_message 简化获取) --- try: container = MessageManager().get_container(chat.stream_id) thinking_count = container.count_thinking_messages() - max_thinking_messages = getattr(global_config, 'max_concurrent_thinking_messages', 3) + max_thinking_messages = getattr(global_config, "max_concurrent_thinking_messages", 3) if thinking_count >= max_thinking_messages: logger.warning(f"聊天流 {stream_name} 已有 {thinking_count} 条思考消息，取消回复。") return @@ -393,7 +415,7 @@ class HeartFC_Chat: get_mid_memory_id = [] tool_result_info = {} send_emoji = "" - observation_context_text = "" # 从 observation 获取上下文文本 + observation_context_text = "" # 从 observation 获取上下文文本 try: # --- 使用传入的 observed_messages 构建上下文文本 --- # if observed_messages: @@ -403,20 +425,22 @@ class HeartFC_Chat: for msg_dict in observed_messages: # 假设 detailed_plain_text 字段包含所需文本 # 你可能需要更复杂的逻辑来格式化，例如添加发送者和时间 - text = msg_dict.get('detailed_plain_text', '') - if text: + text = msg_dict.get("detailed_plain_text", "") + if text: context_texts.append(text) observation_context_text = "\n".join(context_texts) - logger.debug(f"[{stream_name}] Context for tools:\n{observation_context_text[-200:]}...") # 打印部分上下文 + logger.debug( + f"[{stream_name}] Context for tools:\n{observation_context_text[-200:]}..." + ) # 打印部分上下文 else: logger.warning(f"[{stream_name}] observed_messages 列表为空，无法为工具提供上下文。") if observation_context_text: with Timer("思考前使用工具", timing_results): tool_result = await self.tool_user.use_tool( - message_txt=observation_context_text, # <--- 使用观察上下文 + message_txt=observation_context_text, # <--- 使用观察上下文 chat_stream=chat, - sub_heartflow=sub_hf + sub_heartflow=sub_hf, ) if tool_result.get("used_tools", False): if "structured_info" in tool_result: @@ -446,9 +470,9 @@ class HeartFC_Chat: except Exception as e: logger.error(f"[{stream_name}] SubHeartflow 思考失败: {e}") logger.error(traceback.format_exc()) - if info_catcher: + if info_catcher: info_catcher.done_catch() - return # 思考失败则不继续 + return # 思考失败则不继续 if info_catcher: info_catcher.catch_afer_shf_step(timing_results.get("生成内心想法(SubHF)"), past_mind, current_mind) @@ -458,16 +482,16 @@ class HeartFC_Chat: # response_set = await self.gpt.generate_response(anchor_message, thinking_id, current_mind=current_mind) response_set = await self.gpt.generate_response(anchor_message, thinking_id) except Exception as e: - logger.error(f"[{stream_name}] GPT 生成回复失败: {e}") - logger.error(traceback.format_exc()) - if info_catcher: - info_catcher.done_catch() - return + logger.error(f"[{stream_name}] GPT 生成回复失败: {e}") + logger.error(traceback.format_exc()) + if info_catcher: + info_catcher.done_catch() + return if info_catcher: info_catcher.catch_after_generate_response(timing_results.get("生成最终回复(GPT)")) if not response_set: logger.info(f"[{stream_name}] 回复生成失败或为空。") - if info_catcher: + if info_catcher: info_catcher.done_catch() return @@ -481,7 +505,7 @@ class HeartFC_Chat: logger.error(traceback.format_exc()) if info_catcher: info_catcher.catch_after_response(timing_results.get("发送消息"), response_set, first_bot_msg) - info_catcher.done_catch() # 完成捕捉 + info_catcher.done_catch() # 完成捕捉 # --- 11. 处理表情包 (使用 anchor_message) --- try: @@ -496,10 +520,12 @@ class HeartFC_Chat: # --- 12. 记录性能日志 --- # timing_str = " | ".join([f"{step}: {duration:.2f}秒" for step, duration in timing_results.items()]) response_msg = " ".join(response_set) if response_set else "无回复" - logger.info(f"[{stream_name}] 回复任务完成 (Observation Triggered): | 思维消息: {response_msg[:30]}... | 性能计时: {timing_str}") + logger.info( + f"[{stream_name}] 回复任务完成 (Observation Triggered): | 思维消息: {response_msg[:30]}... | 性能计时: {timing_str}" + ) # --- 13. 更新关系情绪 (使用 anchor_message) --- - if first_bot_msg: # 仅在成功发送消息后 + if first_bot_msg: # 仅在成功发送消息后 try: with Timer("更新关系情绪", timing_results): await self._update_relationship(anchor_message, response_set) @@ -512,8 +538,9 @@ class HeartFC_Chat: logger.error(traceback.format_exc()) finally: - # 可以在这里添加清理逻辑，如果有的话 - pass + # 可以在这里添加清理逻辑，如果有的话 + pass + # --- 结束重构 --- # _create_thinking_message, _send_response_messages, _handle_emoji, _update_relationship diff --git a/src/plugins/chat_module/heartFC_chat/heartFC_processor.py b/src/plugins/chat_module/heartFC_chat/heartFC_processor.py index 4f37997d9..23bb719b3 100644 --- a/src/plugins/chat_module/heartFC_chat/heartFC_processor.py +++ b/src/plugins/chat_module/heartFC_chat/heartFC_processor.py @@ -12,7 +12,7 @@ from ...chat.chat_stream import chat_manager from ...chat.message_buffer import message_buffer from ...utils.timer_calculater import Timer from .interest import InterestManager -from .heartFC_chat import HeartFC_Chat # 导入 HeartFC_Chat 以调用回复生成 +from .heartFC_chat import HeartFC_Chat # 导入 HeartFC_Chat 以调用回复生成 # 定义日志配置 processor_config = LogConfig( @@ -24,15 +24,18 @@ logger = get_module_logger("heartFC_processor", config=processor_config) # # 定义兴趣度增加触发回复的阈值 (移至 InterestManager) # INTEREST_INCREASE_THRESHOLD = 0.5 + class HeartFC_Processor: def __init__(self, chat_instance: HeartFC_Chat): self.storage = MessageStorage() - self.interest_manager = InterestManager() # TODO: 可能需要传递 chat_instance 给 InterestManager 或修改其方法签名 - self.chat_instance = chat_instance # 持有 HeartFC_Chat 实例 + self.interest_manager = ( + InterestManager() + ) # TODO: 可能需要传递 chat_instance 给 InterestManager 或修改其方法签名 + self.chat_instance = chat_instance # 持有 HeartFC_Chat 实例 async def process_message(self, message_data: str) -> None: """处理接收到的消息，更新状态，并将回复决策委托给 InterestManager""" - timing_results = {} # 初始化 timing_results + timing_results = {} # 初始化 timing_results message = None try: message = MessageRecv(message_data) @@ -50,7 +53,9 @@ class HeartFC_Processor: group_info=groupinfo, ) if not chat: - logger.error(f"无法为消息创建或获取聊天流: user {userinfo.user_id}, group {groupinfo.group_id if groupinfo else 'None'}") + logger.error( + f"无法为消息创建或获取聊天流: user {userinfo.user_id}, group {groupinfo.group_id if groupinfo else 'None'}" + ) return message.update_chat_stream(chat) @@ -77,9 +82,11 @@ class HeartFC_Processor: if message.message_segment.type != "seglist": F_type = message.message_segment.type else: - if (isinstance(message.message_segment.data, list) - and all(isinstance(x, Seg) for x in message.message_segment.data) - and len(message.message_segment.data) == 1): + if ( + isinstance(message.message_segment.data, list) + and all(isinstance(x, Seg) for x in message.message_segment.data) + and len(message.message_segment.data) == 1 + ): F_type = message.message_segment.data[0].type if F_type == "text": logger.debug(f"触发缓冲，消息：{message.processed_plain_text}") @@ -87,7 +94,7 @@ class HeartFC_Processor: logger.debug("触发缓冲，表情包/图片等待中") elif F_type == "seglist": logger.debug("触发缓冲，消息列表等待中") - return # 被缓冲器拦截，不生成回复 + return # 被缓冲器拦截，不生成回复 # ---- 只有通过缓冲的消息才进行存储和后续处理 ---- @@ -103,11 +110,12 @@ class HeartFC_Processor: # 激活度计算 (使用可能被缓冲器更新过的 message.processed_plain_text) is_mentioned, _ = is_mentioned_bot_in_message(message) - interested_rate = 0.0 # 默认值 + interested_rate = 0.0 # 默认值 try: with Timer("记忆激活", timing_results): interested_rate = await HippocampusManager.get_instance().get_activate_from_text( - message.processed_plain_text, fast_retrieval=True # 使用更新后的文本 + message.processed_plain_text, + fast_retrieval=True, # 使用更新后的文本 ) logger.trace(f"记忆激活率 (通过缓冲后): {interested_rate:.2f}") except Exception as e: @@ -120,11 +128,13 @@ class HeartFC_Processor: # 更新兴趣度 try: self.interest_manager.increase_interest(chat.stream_id, value=interested_rate) - current_interest = self.interest_manager.get_interest(chat.stream_id) # 获取更新后的值用于日志 - logger.trace(f"使用激活率 {interested_rate:.2f} 更新后 (通过缓冲后)，当前兴趣度: {current_interest:.2f}") + current_interest = self.interest_manager.get_interest(chat.stream_id) # 获取更新后的值用于日志 + logger.trace( + f"使用激活率 {interested_rate:.2f} 更新后 (通过缓冲后)，当前兴趣度: {current_interest:.2f}" + ) except Exception as e: - logger.error(f"更新兴趣度失败: {e}") # 调整日志消息 + logger.error(f"更新兴趣度失败: {e}") # 调整日志消息 logger.error(traceback.format_exc()) # ---- 兴趣度计算和更新结束 ---- @@ -143,8 +153,8 @@ class HeartFC_Processor: except Exception as e: logger.error(f"消息处理失败 (process_message V3): {e}") logger.error(traceback.format_exc()) - if message: # 记录失败的消息内容 - logger.error(f"失败消息原始内容: {message.raw_message}") + if message: # 记录失败的消息内容 + logger.error(f"失败消息原始内容: {message.raw_message}") def _check_ban_words(self, text: str, chat, userinfo) -> bool: """检查消息中是否包含过滤词""" @@ -166,4 +176,4 @@ class HeartFC_Processor: ) logger.info(f"[正则表达式过滤]消息匹配到{pattern}，filtered") return True - return False \ No newline at end of file + return False diff --git a/src/plugins/chat_module/heartFC_chat/interest.py b/src/plugins/chat_module/heartFC_chat/interest.py index ec996b4d9..376727cb1 100644 --- a/src/plugins/chat_module/heartFC_chat/interest.py +++ b/src/plugins/chat_module/heartFC_chat/interest.py @@ -2,57 +2,60 @@ import time import math import asyncio import threading -import json # 引入 json -import os # 引入 os -from typing import Optional # <--- 添加导入 -import random # <--- 添加导入 random -from src.common.logger import get_module_logger, LogConfig, DEFAULT_CONFIG # 引入 DEFAULT_CONFIG -from src.plugins.chat.chat_stream import chat_manager # *** Import ChatManager *** +import json # 引入 json +import os # 引入 os +from typing import Optional # <--- 添加导入 +import random # <--- 添加导入 random +from src.common.logger import get_module_logger, LogConfig, DEFAULT_CONFIG # 引入 DEFAULT_CONFIG +from src.plugins.chat.chat_stream import chat_manager # *** Import ChatManager *** # 定义日志配置 (使用 loguru 格式) interest_log_config = LogConfig( - console_format=DEFAULT_CONFIG["console_format"], # 使用默认控制台格式 - file_format=DEFAULT_CONFIG["file_format"] # 使用默认文件格式 + console_format=DEFAULT_CONFIG["console_format"], # 使用默认控制台格式 + file_format=DEFAULT_CONFIG["file_format"], # 使用默认文件格式 ) logger = get_module_logger("InterestManager", config=interest_log_config) # 定义常量 DEFAULT_DECAY_RATE_PER_SECOND = 0.98 # 每秒衰减率 (兴趣保留 99%) -MAX_INTEREST = 15.0 # 最大兴趣值 +MAX_INTEREST = 15.0 # 最大兴趣值 # MIN_INTEREST_THRESHOLD = 0.1 # 低于此值可能被清理 (可选) -CLEANUP_INTERVAL_SECONDS = 3600 # 清理任务运行间隔 (例如：1小时) +CLEANUP_INTERVAL_SECONDS = 3600 # 清理任务运行间隔 (例如：1小时) INACTIVE_THRESHOLD_SECONDS = 3600 # 不活跃时间阈值 (例如：1小时) LOG_INTERVAL_SECONDS = 3 # 日志记录间隔 (例如：30秒) -LOG_DIRECTORY = "logs/interest" # 日志目录 -LOG_FILENAME = "interest_log.json" # 快照日志文件名 (保留，以防其他地方用到) -HISTORY_LOG_FILENAME = "interest_history.log" # 新的历史日志文件名 +LOG_DIRECTORY = "logs/interest" # 日志目录 +LOG_FILENAME = "interest_log.json" # 快照日志文件名 (保留，以防其他地方用到) +HISTORY_LOG_FILENAME = "interest_history.log" # 新的历史日志文件名 # 移除阈值，将移至 HeartFC_Chat # INTEREST_INCREASE_THRESHOLD = 0.5 # --- 新增：概率回复相关常量 --- -REPLY_TRIGGER_THRESHOLD = 3.0 # 触发概率回复的兴趣阈值 (示例值) -BASE_REPLY_PROBABILITY = 0.05 # 首次超过阈值时的基础回复概率 (示例值) -PROBABILITY_INCREASE_RATE_PER_SECOND = 0.02 # 高于阈值时，每秒概率增加量 (线性增长, 示例值) +REPLY_TRIGGER_THRESHOLD = 3.0 # 触发概率回复的兴趣阈值 (示例值) +BASE_REPLY_PROBABILITY = 0.05 # 首次超过阈值时的基础回复概率 (示例值) +PROBABILITY_INCREASE_RATE_PER_SECOND = 0.02 # 高于阈值时，每秒概率增加量 (线性增长, 示例值) PROBABILITY_DECAY_FACTOR_PER_SECOND = 0.3 # 低于阈值时，每秒概率衰减因子 (指数衰减, 示例值) -MAX_REPLY_PROBABILITY = 1 # 回复概率上限 (示例值) +MAX_REPLY_PROBABILITY = 1 # 回复概率上限 (示例值) # --- 结束：概率回复相关常量 --- + class InterestChatting: - def __init__(self, - decay_rate=DEFAULT_DECAY_RATE_PER_SECOND, - max_interest=MAX_INTEREST, - trigger_threshold=REPLY_TRIGGER_THRESHOLD, - base_reply_probability=BASE_REPLY_PROBABILITY, - increase_rate=PROBABILITY_INCREASE_RATE_PER_SECOND, - decay_factor=PROBABILITY_DECAY_FACTOR_PER_SECOND, - max_probability=MAX_REPLY_PROBABILITY): + def __init__( + self, + decay_rate=DEFAULT_DECAY_RATE_PER_SECOND, + max_interest=MAX_INTEREST, + trigger_threshold=REPLY_TRIGGER_THRESHOLD, + base_reply_probability=BASE_REPLY_PROBABILITY, + increase_rate=PROBABILITY_INCREASE_RATE_PER_SECOND, + decay_factor=PROBABILITY_DECAY_FACTOR_PER_SECOND, + max_probability=MAX_REPLY_PROBABILITY, + ): self.interest_level: float = 0.0 - self.last_update_time: float = time.time() # 同时作为兴趣和概率的更新时间基准 + self.last_update_time: float = time.time() # 同时作为兴趣和概率的更新时间基准 self.decay_rate_per_second: float = decay_rate self.max_interest: float = max_interest self.last_increase_amount: float = 0.0 - self.last_interaction_time: float = self.last_update_time # 新增：最后交互时间 + self.last_interaction_time: float = self.last_update_time # 新增：最后交互时间 # --- 新增：概率回复相关属性 --- self.trigger_threshold: float = trigger_threshold @@ -61,7 +64,7 @@ class InterestChatting: self.probability_decay_factor: float = decay_factor self.max_reply_probability: float = max_probability self.current_reply_probability: float = 0.0 - self.is_above_threshold: bool = False # 标记兴趣值是否高于阈值 + self.is_above_threshold: bool = False # 标记兴趣值是否高于阈值 # --- 结束：概率回复相关属性 --- def _calculate_decay(self, current_time: float): @@ -76,24 +79,30 @@ class InterestChatting: else: # 检查 decay_rate_per_second 是否为非正数，避免 math domain error if self.decay_rate_per_second <= 0: - logger.warning(f"InterestChatting encountered non-positive decay rate: {self.decay_rate_per_second}. Setting interest to 0.") - self.interest_level = 0.0 + logger.warning( + f"InterestChatting encountered non-positive decay rate: {self.decay_rate_per_second}. Setting interest to 0." + ) + self.interest_level = 0.0 # 检查 interest_level 是否为负数，虽然理论上不应发生，但以防万一 elif self.interest_level < 0: - logger.warning(f"InterestChatting encountered negative interest level: {self.interest_level}. Setting interest to 0.") - self.interest_level = 0.0 + logger.warning( + f"InterestChatting encountered negative interest level: {self.interest_level}. Setting interest to 0." + ) + self.interest_level = 0.0 else: try: decay_factor = math.pow(self.decay_rate_per_second, time_delta) self.interest_level *= decay_factor except ValueError as e: # 捕获潜在的 math domain error，例如对负数开非整数次方（虽然已加保护） - logger.error(f"Math error during decay calculation: {e}. Rate: {self.decay_rate_per_second}, Delta: {time_delta}, Level: {self.interest_level}. Setting interest to 0.") + logger.error( + f"Math error during decay calculation: {e}. Rate: {self.decay_rate_per_second}, Delta: {time_delta}, Level: {self.interest_level}. Setting interest to 0." + ) self.interest_level = 0.0 # 防止低于阈值 (如果需要) # self.interest_level = max(self.interest_level, MIN_INTEREST_THRESHOLD) - + # 只有在兴趣值发生变化时才更新时间戳 if old_interest != self.interest_level: self.last_update_time = current_time @@ -102,7 +111,7 @@ class InterestChatting: """根据当前兴趣是否超过阈值及时间差，更新回复概率""" time_delta = current_time - self.last_update_time if time_delta <= 0: - return # 时间未前进，无需更新 + return # 时间未前进，无需更新 currently_above = self.interest_level >= self.trigger_threshold @@ -110,7 +119,9 @@ class InterestChatting: if not self.is_above_threshold: # 刚跨过阈值，重置为基础概率 self.current_reply_probability = self.base_reply_probability - logger.debug(f"兴趣跨过阈值 ({self.trigger_threshold}). 概率重置为基础值: {self.base_reply_probability:.4f}") + logger.debug( + f"兴趣跨过阈值 ({self.trigger_threshold}). 概率重置为基础值: {self.base_reply_probability:.4f}" + ) else: # 持续高于阈值，线性增加概率 increase_amount = self.probability_increase_rate * time_delta @@ -120,7 +131,7 @@ class InterestChatting: # 限制概率不超过最大值 self.current_reply_probability = min(self.current_reply_probability, self.max_reply_probability) - else: # 低于阈值 + else: # 低于阈值 # if self.is_above_threshold: # # 刚低于阈值，开始衰减 # logger.debug(f"兴趣低于阈值 ({self.trigger_threshold}). 概率衰减开始于 {self.current_reply_probability:.4f}") @@ -140,8 +151,8 @@ class InterestChatting: elif self.probability_decay_factor <= 0: # 如果衰减因子无效或为0，直接清零 if self.current_reply_probability > 0: - logger.warning(f"无效的衰减因子 ({self.probability_decay_factor}). 设置概率为0.") - self.current_reply_probability = 0.0 + logger.warning(f"无效的衰减因子 ({self.probability_decay_factor}). 设置概率为0.") + self.current_reply_probability = 0.0 # else: decay_factor >= 1, probability will not decay or increase, which might be intended in some cases. # 确保概率不低于0 @@ -160,9 +171,9 @@ class InterestChatting: self.last_increase_amount = value # 应用增加 self.interest_level += value - self.interest_level = min(self.interest_level, self.max_interest) # 不超过最大值 - self.last_update_time = current_time # 更新时间戳 - self.last_interaction_time = current_time # 更新最后交互时间 + self.interest_level = min(self.interest_level, self.max_interest) # 不超过最大值 + self.last_update_time = current_time # 更新时间戳 + self.last_interaction_time = current_time # 更新最后交互时间 def decrease_interest(self, current_time: float, value: float): """降低兴趣值并更新时间 (确保不低于0)""" @@ -170,9 +181,9 @@ class InterestChatting: self._update_reply_probability(current_time) # 注意：降低兴趣度是否需要先衰减？取决于具体逻辑，这里假设不衰减直接减 self.interest_level -= value - self.interest_level = max(self.interest_level, 0.0) # 确保不低于0 - self.last_update_time = current_time # 降低也更新时间戳 - self.last_interaction_time = current_time # 更新最后交互时间 + self.interest_level = max(self.interest_level, 0.0) # 确保不低于0 + self.last_update_time = current_time # 降低也更新时间戳 + self.last_interaction_time = current_time # 更新最后交互时间 def reset_trigger_info(self): """重置触发相关信息，在外部任务处理后调用""" @@ -184,7 +195,7 @@ class InterestChatting: current_time = time.time() self._update_reply_probability(current_time) self._calculate_decay(current_time) - self.last_update_time = current_time # 更新时间戳 + self.last_update_time = current_time # 更新时间戳 return self.interest_level def get_state(self) -> dict: @@ -194,9 +205,9 @@ class InterestChatting: return { "interest_level": round(interest, 2), "last_update_time": self.last_update_time, - "current_reply_probability": round(self.current_reply_probability, 4), # 添加概率到状态 - "is_above_threshold": self.is_above_threshold, # 添加阈值状态 - "last_interaction_time": self.last_interaction_time # 新增：添加最后交互时间到状态 + "current_reply_probability": round(self.current_reply_probability, 4), # 添加概率到状态 + "is_above_threshold": self.is_above_threshold, # 添加阈值状态 + "last_interaction_time": self.last_interaction_time, # 新增：添加最后交互时间到状态 # 可以选择性地暴露 last_increase_amount 给状态，方便调试 # "last_increase_amount": round(self.last_increase_amount, 2) } @@ -222,7 +233,7 @@ class InterestChatting: # # self.current_reply_probability = self.base_reply_probability # 例如，触发后降回基础概率 # # self.current_reply_probability *= 0.5 # 例如，触发后概率减半 # else: - # logger.debug(f"回复概率评估未触发。概率: {self.current_reply_probability:.4f}") + # logger.debug(f"回复概率评估未触发。概率: {self.current_reply_probability:.4f}") return trigger else: # logger.debug(f"Reply evaluation check: Below threshold or zero probability. Probability: {self.current_reply_probability:.4f}") @@ -256,10 +267,10 @@ class InterestManager: self._history_log_file_path = os.path.join(LOG_DIRECTORY, HISTORY_LOG_FILENAME) self._ensure_log_directory() self._cleanup_task = None - self._logging_task = None # 添加日志任务变量 + self._logging_task = None # 添加日志任务变量 self._initialized = True - logger.info("InterestManager initialized.") # 修改日志消息 - self._decay_task = None # 新增：衰减任务变量 + logger.info("InterestManager initialized.") # 修改日志消息 + self._decay_task = None # 新增：衰减任务变量 def _ensure_log_directory(self): """确保日志目录存在""" @@ -283,14 +294,14 @@ class InterestManager: # logger.debug(f"运行定期历史记录 (间隔: {interval_seconds}秒)...") try: current_timestamp = time.time() - all_states = self.get_all_interest_states() # 获取当前所有状态 - + all_states = self.get_all_interest_states() # 获取当前所有状态 + # 以追加模式打开历史日志文件 - with open(self._history_log_file_path, 'a', encoding='utf-8') as f: + with open(self._history_log_file_path, "a", encoding="utf-8") as f: count = 0 for stream_id, state in all_states.items(): # *** Get group name from ChatManager *** - group_name = stream_id # Default to stream_id + group_name = stream_id # Default to stream_id try: # Use the imported chat_manager instance chat_stream = chat_manager.get_stream(stream_id) @@ -298,7 +309,11 @@ class InterestManager: group_name = chat_stream.group_info.group_name elif chat_stream and not chat_stream.group_info: # Handle private chats - maybe use user nickname? - group_name = f"私聊_{chat_stream.user_info.user_nickname}" if chat_stream.user_info else stream_id + group_name = ( + f"私聊_{chat_stream.user_info.user_nickname}" + if chat_stream.user_info + else stream_id + ) except Exception as e: logger.warning(f"Could not get group name for stream_id {stream_id}: {e}") # Fallback to stream_id is already handled by default value @@ -306,25 +321,25 @@ class InterestManager: log_entry = { "timestamp": round(current_timestamp, 2), "stream_id": stream_id, - "interest_level": state.get("interest_level", 0.0), # 确保有默认值 - "group_name": group_name, # *** Add group_name *** + "interest_level": state.get("interest_level", 0.0), # 确保有默认值 + "group_name": group_name, # *** Add group_name *** # --- 新增：记录概率相关信息 --- "reply_probability": state.get("current_reply_probability", 0.0), - "is_above_threshold": state.get("is_above_threshold", False) + "is_above_threshold": state.get("is_above_threshold", False), # --- 结束新增 --- } # 将每个条目作为单独的 JSON 行写入 - f.write(json.dumps(log_entry, ensure_ascii=False) + '\n') + f.write(json.dumps(log_entry, ensure_ascii=False) + "\n") count += 1 # logger.debug(f"Successfully appended {count} interest history entries to {self._history_log_file_path}") - + # 注意：不再写入快照文件 interest_log.json # 如果需要快照文件，可以在这里单独写入 self._snapshot_log_file_path # 例如： # with open(self._snapshot_log_file_path, 'w', encoding='utf-8') as snap_f: # json.dump(all_states, snap_f, indent=4, ensure_ascii=False) # logger.debug(f"Successfully wrote snapshot to {self._snapshot_log_file_path}") - + except IOError as e: logger.error(f"Error writing interest history log to {self._history_log_file_path}: {e}") except Exception as e: @@ -333,7 +348,7 @@ class InterestManager: async def _periodic_decay_task(self): """后台衰减任务的异步函数，每秒更新一次所有实例的衰减""" while True: - await asyncio.sleep(1) # 每秒运行一次 + await asyncio.sleep(1) # 每秒运行一次 current_time = time.time() # logger.debug("Running periodic decay calculation...") # 调试日志，可能过于频繁 @@ -355,27 +370,24 @@ class InterestManager: if self._cleanup_task is None or self._cleanup_task.done(): self._cleanup_task = asyncio.create_task( self._periodic_cleanup_task( - interval_seconds=CLEANUP_INTERVAL_SECONDS, - max_age_seconds=INACTIVE_THRESHOLD_SECONDS + interval_seconds=CLEANUP_INTERVAL_SECONDS, max_age_seconds=INACTIVE_THRESHOLD_SECONDS ) ) - logger.info(f"已创建定期清理任务。间隔时间: {CLEANUP_INTERVAL_SECONDS}秒, 不活跃阈值: {INACTIVE_THRESHOLD_SECONDS}秒") + logger.info( + f"已创建定期清理任务。间隔时间: {CLEANUP_INTERVAL_SECONDS}秒, 不活跃阈值: {INACTIVE_THRESHOLD_SECONDS}秒" + ) else: logger.warning("跳过创建清理任务:任务已在运行或存在。") if self._logging_task is None or self._logging_task.done(): - self._logging_task = asyncio.create_task( - self._periodic_log_task(interval_seconds=LOG_INTERVAL_SECONDS) - ) + self._logging_task = asyncio.create_task(self._periodic_log_task(interval_seconds=LOG_INTERVAL_SECONDS)) logger.info(f"已创建定期日志任务。间隔时间: {LOG_INTERVAL_SECONDS}秒") else: logger.warning("跳过创建日志任务:任务已在运行或存在。") # 启动新的衰减任务 if self._decay_task is None or self._decay_task.done(): - self._decay_task = asyncio.create_task( - self._periodic_decay_task() - ) + self._decay_task = asyncio.create_task(self._periodic_decay_task()) logger.info("已创建定期衰减任务。间隔时间: 1秒") else: logger.warning("跳过创建衰减任务:任务已在运行或存在。") @@ -391,7 +403,7 @@ class InterestManager: # 直接调用 get_state，它会使用内部的 get_interest 获取已更新的值 states[stream_id] = chatting.get_state() except Exception as e: - logger.warning(f"Error getting state for stream_id {stream_id}: {e}") + logger.warning(f"Error getting state for stream_id {stream_id}: {e}") return states def get_interest_chatting(self, stream_id: str) -> Optional[InterestChatting]: @@ -410,11 +422,11 @@ class InterestManager: # --- 修改：创建时传入概率相关参数 (如果需要定制化，否则使用默认值) --- self.interest_dict[stream_id] = InterestChatting( # decay_rate=..., max_interest=..., # 可以从配置读取 - trigger_threshold=REPLY_TRIGGER_THRESHOLD, # 使用全局常量 + trigger_threshold=REPLY_TRIGGER_THRESHOLD, # 使用全局常量 base_reply_probability=BASE_REPLY_PROBABILITY, increase_rate=PROBABILITY_INCREASE_RATE_PER_SECOND, decay_factor=PROBABILITY_DECAY_FACTOR_PER_SECOND, - max_probability=MAX_REPLY_PROBABILITY + max_probability=MAX_REPLY_PROBABILITY, ) # --- 结束修改 --- # 首次创建时兴趣为 0，由第一次消息的 activate rate 决定初始值 @@ -433,8 +445,10 @@ class InterestManager: interest_chatting = self._get_or_create_interest_chatting(stream_id) # 调用修改后的 increase_interest，不再传入 message interest_chatting.increase_interest(current_time, value) - stream_name = chat_manager.get_stream_name(stream_id) or stream_id # 获取流名称 - logger.debug(f"增加了聊天流 {stream_name} 的兴趣度 {value:.2f}，当前值为 {interest_chatting.interest_level:.2f}") # 更新日志 + stream_name = chat_manager.get_stream_name(stream_id) or stream_id # 获取流名称 + logger.debug( + f"增加了聊天流 {stream_name} 的兴趣度 {value:.2f}，当前值为 {interest_chatting.interest_level:.2f}" + ) # 更新日志 def decrease_interest(self, stream_id: str, value: float): """降低指定聊天流的兴趣度""" @@ -443,10 +457,12 @@ class InterestManager: interest_chatting = self.get_interest_chatting(stream_id) if interest_chatting: interest_chatting.decrease_interest(current_time, value) - stream_name = chat_manager.get_stream_name(stream_id) or stream_id # 获取流名称 - logger.debug(f"降低了聊天流 {stream_name} 的兴趣度 {value:.2f}，当前值为 {interest_chatting.interest_level:.2f}") + stream_name = chat_manager.get_stream_name(stream_id) or stream_id # 获取流名称 + logger.debug( + f"降低了聊天流 {stream_name} 的兴趣度 {value:.2f}，当前值为 {interest_chatting.interest_level:.2f}" + ) else: - stream_name = chat_manager.get_stream_name(stream_id) or stream_id # 获取流名称 + stream_name = chat_manager.get_stream_name(stream_id) or stream_id # 获取流名称 logger.warning(f"尝试降低不存在的聊天流 {stream_name} 的兴趣度") def cleanup_inactive_chats(self, max_age_seconds=INACTIVE_THRESHOLD_SECONDS): @@ -465,29 +481,31 @@ class InterestManager: # 先计算当前兴趣，确保是最新的 # 加锁保护 chatting 对象状态的读取和可能的修改 # with self._lock: # 如果 InterestChatting 内部操作不是原子的 - last_interaction = chatting.last_interaction_time # 使用最后交互时间 + last_interaction = chatting.last_interaction_time # 使用最后交互时间 should_remove = False reason = "" # 只有设置了 max_age_seconds 才检查时间 - if max_age_seconds is not None and (current_time - last_interaction) > max_age_seconds: # 使用 last_interaction + if ( + max_age_seconds is not None and (current_time - last_interaction) > max_age_seconds + ): # 使用 last_interaction should_remove = True - reason = f"inactive time ({current_time - last_interaction:.0f}s) > max age ({max_age_seconds}s)" # 更新日志信息 + reason = f"inactive time ({current_time - last_interaction:.0f}s) > max age ({max_age_seconds}s)" # 更新日志信息 if should_remove: keys_to_remove.append(stream_id) - stream_name = chat_manager.get_stream_name(stream_id) or stream_id # 获取流名称 + stream_name = chat_manager.get_stream_name(stream_id) or stream_id # 获取流名称 logger.debug(f"Marking stream {stream_name} for removal. Reason: {reason}") if keys_to_remove: logger.info(f"清理识别到 {len(keys_to_remove)} 个不活跃/低兴趣的流。") # with self._lock: # 确保删除操作的原子性 for key in keys_to_remove: - # 再次检查 key 是否存在，以防万一在迭代和删除之间状态改变 + # 再次检查 key 是否存在，以防万一在迭代和删除之间状态改变 if key in self.interest_dict: del self.interest_dict[key] - stream_name = chat_manager.get_stream_name(key) or key # 获取流名称 + stream_name = chat_manager.get_stream_name(key) or key # 获取流名称 logger.debug(f"移除了流: {stream_name}") final_count = initial_count - len(keys_to_remove) logger.info(f"清理完成。移除了 {len(keys_to_remove)} 个流。当前数量: {final_count}") else: - logger.info(f"清理完成。没有流符合移除条件。当前数量: {initial_count}") \ No newline at end of file + logger.info(f"清理完成。没有流符合移除条件。当前数量: {initial_count}") diff --git a/src/plugins/chat_module/heartFC_chat/messagesender.py b/src/plugins/chat_module/heartFC_chat/messagesender.py index ed7221963..f9bcbc7b6 100644 --- a/src/plugins/chat_module/heartFC_chat/messagesender.py +++ b/src/plugins/chat_module/heartFC_chat/messagesender.py @@ -23,6 +23,7 @@ logger = get_module_logger("msg_sender", config=sender_config) class MessageSender: """发送器""" + _instance = None def __new__(cls, *args, **kwargs): @@ -32,7 +33,7 @@ class MessageSender: def __init__(self): # 确保 __init__ 只被调用一次 - if not hasattr(self, '_initialized'): + if not hasattr(self, "_initialized"): self.message_interval = (0.5, 1) # 消息间隔时间范围(秒) self.last_send_time = 0 self._current_bot = None @@ -42,7 +43,6 @@ class MessageSender: """设置当前bot实例""" pass - async def send_via_ws(self, message: MessageSending) -> None: try: await global_api.send_message(message) @@ -56,7 +56,6 @@ class MessageSender: """发送消息""" if isinstance(message, MessageSending): - typing_time = calculate_typing_time( input_string=message.processed_plain_text, thinking_start_time=message.thinking_start_time, @@ -143,6 +142,7 @@ class MessageContainer: class MessageManager: """管理所有聊天流的消息容器""" + _instance = None def __new__(cls, *args, **kwargs): @@ -152,7 +152,7 @@ class MessageManager: def __init__(self): # 确保 __init__ 只被调用一次 - if not hasattr(self, '_initialized'): + if not hasattr(self, "_initialized"): self.containers: Dict[str, MessageContainer] = {} # chat_id -> MessageContainer self.storage = MessageStorage() self._running = True @@ -232,10 +232,10 @@ class MessageManager: while self._running: await asyncio.sleep(1) tasks = [] - for chat_id in list(self.containers.keys()): # 使用 list 复制 key，防止在迭代时修改字典 + for chat_id in list(self.containers.keys()): # 使用 list 复制 key，防止在迭代时修改字典 tasks.append(self.process_chat_messages(chat_id)) - if tasks: # 仅在有任务时执行 gather + if tasks: # 仅在有任务时执行 gather await asyncio.gather(*tasks) diff --git a/src/plugins/chat_module/heartFC_chat/pf_chatting.py b/src/plugins/chat_module/heartFC_chat/pf_chatting.py index b77d351d0..681d0570c 100644 --- a/src/plugins/chat_module/heartFC_chat/pf_chatting.py +++ b/src/plugins/chat_module/heartFC_chat/pf_chatting.py @@ -11,15 +11,15 @@ from ...message import UserInfo from src.heart_flow.heartflow import heartflow, SubHeartflow 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.common.logger import get_module_logger, LogConfig, DEFAULT_CONFIG # 引入 DEFAULT_CONFIG from src.plugins.models.utils_model import LLMRequest # 定义日志配置 (使用 loguru 格式) interest_log_config = LogConfig( - console_format=DEFAULT_CONFIG["console_format"], # 使用默认控制台格式 - file_format=DEFAULT_CONFIG["file_format"] # 使用默认文件格式 + console_format=DEFAULT_CONFIG["console_format"], # 使用默认控制台格式 + file_format=DEFAULT_CONFIG["file_format"], # 使用默认文件格式 ) -logger = get_module_logger("PFChattingLoop", config=interest_log_config) # Logger Name Changed +logger = get_module_logger("PFChattingLoop", config=interest_log_config) # Logger Name Changed # Forward declaration for type hinting @@ -38,30 +38,29 @@ PLANNER_TOOL_DEFINITION = [ "action": { "type": "string", "enum": ["no_reply", "text_reply", "emoji_reply"], - "description": "决定采取的行动：'no_reply'(不回复), 'text_reply'(文本回复) 或 'emoji_reply'(表情回复)。" + "description": "决定采取的行动：'no_reply'(不回复), 'text_reply'(文本回复) 或 'emoji_reply'(表情回复)。", }, - "reasoning": { - "type": "string", - "description": "做出此决定的简要理由。" - }, + "reasoning": {"type": "string", "description": "做出此决定的简要理由。"}, "emoji_query": { "type": "string", - "description": '如果行动是\'emoji_reply\',则指定表情的主题或概念（例如，"开心"、"困惑"）。仅在需要表情回复时提供。' - } + "description": '如果行动是\'emoji_reply\',则指定表情的主题或概念（例如，"开心"、"困惑"）。仅在需要表情回复时提供。', + }, }, - "required": ["action", "reasoning"] # 强制要求提供行动和理由 - } - } + "required": ["action", "reasoning"], # 强制要求提供行动和理由 + }, + }, } ] + class PFChatting: """ Manages a continuous Plan-Filter-Check (now Plan-Replier-Sender) loop for generating replies within a specific chat stream, controlled by a timer. The loop runs as long as the timer > 0. """ - def __init__(self, chat_id: str, heartfc_chat_instance: 'HeartFC_Chat'): + + def __init__(self, chat_id: str, heartfc_chat_instance: "HeartFC_Chat"): """ 初始化PFChatting实例。 @@ -69,34 +68,33 @@ class PFChatting: chat_id: The identifier for the chat stream (e.g., stream_id). heartfc_chat_instance: 访问共享资源和方法的主HeartFC_Chat实例。 """ - self.heartfc_chat = heartfc_chat_instance # 访问logger, gpt, tool_user, _send_response_messages等。 + self.heartfc_chat = heartfc_chat_instance # 访问logger, gpt, tool_user, _send_response_messages等。 self.stream_id: str = chat_id self.chat_stream: Optional[ChatStream] = None self.sub_hf: Optional[SubHeartflow] = None self._initialized = False - self._init_lock = asyncio.Lock() # Ensure initialization happens only once - self._processing_lock = asyncio.Lock() # 确保只有一个 Plan-Replier-Sender 周期在运行 - self._timer_lock = asyncio.Lock() # 用于安全更新计时器 + self._init_lock = asyncio.Lock() # Ensure initialization happens only once + self._processing_lock = asyncio.Lock() # 确保只有一个 Plan-Replier-Sender 周期在运行 + self._timer_lock = asyncio.Lock() # 用于安全更新计时器 self.planner_llm = LLMRequest( model=global_config.llm_normal, temperature=global_config.llm_normal["temp"], max_tokens=1000, - request_type="action_planning" + request_type="action_planning", ) # Internal state for loop control - self._loop_timer: float = 0.0 # Remaining time for the loop in seconds - self._loop_active: bool = False # Is the loop currently running? - self._loop_task: Optional[asyncio.Task] = None # Stores the main loop task - self._trigger_count_this_activation: int = 0 # Counts triggers within an active period - self._initial_duration: float = 30.0 # 首次触发增加的时间 - self._last_added_duration: float = self._initial_duration # <--- 新增：存储上次增加的时间 + self._loop_timer: float = 0.0 # Remaining time for the loop in seconds + self._loop_active: bool = False # Is the loop currently running? + self._loop_task: Optional[asyncio.Task] = None # Stores the main loop task + self._trigger_count_this_activation: int = 0 # Counts triggers within an active period + self._initial_duration: float = 30.0 # 首次触发增加的时间 + self._last_added_duration: float = self._initial_duration # <--- 新增：存储上次增加的时间 # Removed pending_replies as processing is now serial within the loop # self.pending_replies: Dict[str, PendingReply] = {} - def _get_log_prefix(self) -> str: """获取日志前缀，包含可读的流名称""" stream_name = chat_manager.get_stream_name(self.stream_id) or self.stream_id @@ -110,7 +108,7 @@ class PFChatting: async with self._init_lock: if self._initialized: return True - log_prefix = self._get_log_prefix() # 获取前缀 + log_prefix = self._get_log_prefix() # 获取前缀 try: self.chat_stream = chat_manager.get_stream(self.stream_id) @@ -140,23 +138,25 @@ class PFChatting: """ log_prefix = self._get_log_prefix() if not self._initialized: - if not await self._initialize(): - logger.error(f"{log_prefix} 无法添加时间: 未初始化。") - return + if not await self._initialize(): + logger.error(f"{log_prefix} 无法添加时间: 未初始化。") + return async with self._timer_lock: duration_to_add: float = 0.0 - if not self._loop_active: # First trigger for this activation cycle - duration_to_add = self._initial_duration # 使用初始值 - self._last_added_duration = duration_to_add # 更新上次增加的值 - self._trigger_count_this_activation = 1 # Start counting + if not self._loop_active: # First trigger for this activation cycle + duration_to_add = self._initial_duration # 使用初始值 + self._last_added_duration = duration_to_add # 更新上次增加的值 + self._trigger_count_this_activation = 1 # Start counting logger.info(f"{log_prefix} First trigger in activation. Adding {duration_to_add:.2f}s.") - else: # Loop is already active, apply 50% reduction + else: # Loop is already active, apply 50% reduction self._trigger_count_this_activation += 1 duration_to_add = self._last_added_duration * 0.5 - self._last_added_duration = duration_to_add # 更新上次增加的值 - logger.info(f"{log_prefix} Trigger #{self._trigger_count_this_activation}. Adding {duration_to_add:.2f}s (50% of previous). Timer was {self._loop_timer:.1f}s.") + self._last_added_duration = duration_to_add # 更新上次增加的值 + logger.info( + f"{log_prefix} Trigger #{self._trigger_count_this_activation}. Adding {duration_to_add:.2f}s (50% of previous). Timer was {self._loop_timer:.1f}s." + ) # 添加计算出的时间 new_timer_value = self._loop_timer + duration_to_add @@ -174,8 +174,7 @@ class PFChatting: self._loop_task = asyncio.create_task(self._run_pf_loop()) self._loop_task.add_done_callback(self._handle_loop_completion) elif self._loop_active: - logger.debug(f"{log_prefix} Loop already active. Timer extended.") - + logger.debug(f"{log_prefix} Loop already active. Timer extended.") def _handle_loop_completion(self, task: asyncio.Task): """当 _run_pf_loop 任务完成时执行的回调。""" @@ -194,14 +193,13 @@ class PFChatting: # Reset state regardless of how the task finished self._loop_active = False self._loop_task = None - self._last_added_duration = self._initial_duration # <--- 重置下次首次触发的增加时间 - self._trigger_count_this_activation = 0 # 重置计数器 + self._last_added_duration = self._initial_duration # <--- 重置下次首次触发的增加时间 + self._trigger_count_this_activation = 0 # 重置计数器 # Ensure lock is released if the loop somehow exited while holding it if self._processing_lock.locked(): logger.warning(f"{log_prefix} PFChatting: 锁没有正常释放") self._processing_lock.release() - async def _run_pf_loop(self): """ 主循环，当计时器>0时持续进行计划并可能回复消息 @@ -214,13 +212,15 @@ class PFChatting: async with self._timer_lock: current_timer = self._loop_timer if current_timer <= 0: - logger.info(f"{self._get_log_prefix()} PFChatting: 聊太久了，麦麦打算休息一下(已经聊了{current_timer:.1f}秒)，退出PFChatting") + logger.info( + f"{self._get_log_prefix()} PFChatting: 聊太久了，麦麦打算休息一下(已经聊了{current_timer:.1f}秒)，退出PFChatting" + ) break # 退出条件：计时器到期 # 记录循环开始时间 loop_cycle_start_time = time.monotonic() # 标记本周期是否执行了操作 - action_taken_this_cycle = False + action_taken_this_cycle = False # 获取处理锁，确保每个计划-回复-发送周期独占执行 acquired_lock = False @@ -231,13 +231,13 @@ class PFChatting: # --- Planner --- # Planner decides action, reasoning, emoji_query, etc. - planner_result = await self._planner() # Modify planner to return decision dict + planner_result = await self._planner() # Modify planner to return decision dict action = planner_result.get("action", "error") reasoning = planner_result.get("reasoning", "Planner did not provide reasoning.") emoji_query = planner_result.get("emoji_query", "") current_mind = planner_result.get("current_mind", "[Mind unavailable]") send_emoji_from_tools = planner_result.get("send_emoji_from_tools", "") - observed_messages = planner_result.get("observed_messages", []) # Planner needs to return this + observed_messages = planner_result.get("observed_messages", []) # Planner needs to return this if action == "text_reply": logger.info(f"{self._get_log_prefix()} PFChatting: 麦麦决定回复文本.") @@ -245,7 +245,7 @@ class PFChatting: # --- 回复器 --- anchor_message = await self._get_anchor_message(observed_messages) if not anchor_message: - logger.error(f"{self._get_log_prefix()} 循环: 无法获取锚点消息用于回复. 跳过周期.") + logger.error(f"{self._get_log_prefix()} 循环: 无法获取锚点消息用于回复. 跳过周期.") else: thinking_id = await self.heartfc_chat._create_thinking_message(anchor_message) if not thinking_id: @@ -259,12 +259,12 @@ class PFChatting: anchor_message=anchor_message, thinking_id=thinking_id, current_mind=current_mind, - send_emoji=send_emoji_from_tools + send_emoji=send_emoji_from_tools, ) except Exception as e_replier: - logger.error(f"{self._get_log_prefix()} 循环: 回复器工作失败: {e_replier}") - self._cleanup_thinking_message(thinking_id) # 清理思考消息 - # 继续循环, 视为非操作周期 + logger.error(f"{self._get_log_prefix()} 循环: 回复器工作失败: {e_replier}") + self._cleanup_thinking_message(thinking_id) # 清理思考消息 + # 继续循环, 视为非操作周期 if replier_result: # --- Sender --- @@ -272,13 +272,13 @@ class PFChatting: await self._sender(thinking_id, anchor_message, replier_result) logger.info(f"{self._get_log_prefix()} 循环: 发送器完成成功.") except Exception as e_sender: - logger.error(f"{self._get_log_prefix()} 循环: 发送器失败: {e_sender}") - self._cleanup_thinking_message(thinking_id) # 确保发送失败时清理 - # 继续循环, 视为非操作周期 + logger.error(f"{self._get_log_prefix()} 循环: 发送器失败: {e_sender}") + self._cleanup_thinking_message(thinking_id) # 确保发送失败时清理 + # 继续循环, 视为非操作周期 else: - # Replier failed to produce result - logger.warning(f"{self._get_log_prefix()} 循环: 回复器未产生结果. 跳过发送.") - self._cleanup_thinking_message(thinking_id) # 清理思考消息 + # Replier failed to produce result + logger.warning(f"{self._get_log_prefix()} 循环: 回复器未产生结果. 跳过发送.") + self._cleanup_thinking_message(thinking_id) # 清理思考消息 elif action == "emoji_reply": logger.info(f"{self._get_log_prefix()} PFChatting: 麦麦决定回复表情 ('{emoji_query}').") @@ -290,19 +290,19 @@ class PFChatting: except Exception as e_emoji: logger.error(f"{self._get_log_prefix()} 循环: 发送表情失败: {e_emoji}") else: - logger.warning(f"{self._get_log_prefix()} 循环: 无法发送表情, 无法获取锚点.") + logger.warning(f"{self._get_log_prefix()} 循环: 无法发送表情, 无法获取锚点.") elif action == "no_reply": logger.info(f"{self._get_log_prefix()} PFChatting: 麦麦决定不回复. 原因: {reasoning}") # Do nothing else, action_taken_this_cycle remains False elif action == "error": - logger.error(f"{self._get_log_prefix()} PFChatting: 麦麦回复出错. 原因: {reasoning}") - # 视为非操作周期 + logger.error(f"{self._get_log_prefix()} PFChatting: 麦麦回复出错. 原因: {reasoning}") + # 视为非操作周期 - else: # Unknown action - logger.warning(f"{self._get_log_prefix()} PFChatting: 麦麦做了奇怪的事情. 原因: {reasoning}") - # 视为非操作周期 + else: # Unknown action + logger.warning(f"{self._get_log_prefix()} PFChatting: 麦麦做了奇怪的事情. 原因: {reasoning}") + # 视为非操作周期 except Exception as e_cycle: # Catch errors occurring within the locked section (e.g., planner crash) @@ -310,9 +310,9 @@ class PFChatting: logger.error(traceback.format_exc()) # Ensure lock is released if an error occurs before the finally block if acquired_lock and self._processing_lock.locked(): - self._processing_lock.release() - acquired_lock = False # 防止在 finally 块中重复释放 - logger.warning(f"{self._get_log_prefix()} 由于循环周期中的错误释放了处理锁.") + self._processing_lock.release() + acquired_lock = False # 防止在 finally 块中重复释放 + logger.warning(f"{self._get_log_prefix()} 由于循环周期中的错误释放了处理锁.") finally: # Ensure the lock is always released after a cycle @@ -324,26 +324,28 @@ class PFChatting: cycle_duration = time.monotonic() - loop_cycle_start_time async with self._timer_lock: self._loop_timer -= cycle_duration - logger.debug(f"{self._get_log_prefix()} PFChatting: 麦麦聊了{cycle_duration:.2f}秒. 还能聊: {self._loop_timer:.1f}s.") + logger.debug( + f"{self._get_log_prefix()} PFChatting: 麦麦聊了{cycle_duration:.2f}秒. 还能聊: {self._loop_timer:.1f}s." + ) # --- Delay --- # Add a small delay, especially if no action was taken, to prevent busy-waiting try: if not action_taken_this_cycle and cycle_duration < 1.5: - # If nothing happened and cycle was fast, wait a bit longer - await asyncio.sleep(1.5 - cycle_duration) - elif cycle_duration < 0.2: # Minimum delay even if action was taken - await asyncio.sleep(0.2) + # If nothing happened and cycle was fast, wait a bit longer + await asyncio.sleep(1.5 - cycle_duration) + elif cycle_duration < 0.2: # Minimum delay even if action was taken + await asyncio.sleep(0.2) except asyncio.CancelledError: logger.info(f"{self._get_log_prefix()} Sleep interrupted, likely loop cancellation.") - break # Exit loop if cancelled during sleep + break # Exit loop if cancelled during sleep except asyncio.CancelledError: - logger.info(f"{self._get_log_prefix()} PFChatting: 麦麦的聊天被取消了") + logger.info(f"{self._get_log_prefix()} PFChatting: 麦麦的聊天被取消了") except Exception as e_loop_outer: - # Catch errors outside the main cycle lock (should be rare) - logger.error(f"{self._get_log_prefix()} PFChatting: 麦麦的聊天出错了: {e_loop_outer}") - logger.error(traceback.format_exc()) + # Catch errors outside the main cycle lock (should be rare) + logger.error(f"{self._get_log_prefix()} PFChatting: 麦麦的聊天出错了: {e_loop_outer}") + logger.error(traceback.format_exc()) finally: # Reset trigger count when loop finishes async with self._timer_lock: @@ -363,7 +365,7 @@ class PFChatting: observed_messages: List[dict] = [] tool_result_info = {} get_mid_memory_id = [] - send_emoji_from_tools = "" # Renamed for clarity + send_emoji_from_tools = "" # Renamed for clarity current_mind: Optional[str] = None # --- 获取最新的观察信息 --- @@ -371,8 +373,8 @@ class PFChatting: 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()...") - await observation.observe() # 主动观察以获取最新消息 - observed_messages = observation.talking_message # 获取更新后的消息列表 + await observation.observe() # 主动观察以获取最新消息 + observed_messages = observation.talking_message # 获取更新后的消息列表 logger.debug(f"{log_prefix}[Planner] 获取到 {len(observed_messages)} 条观察消息。") else: logger.warning(f"{log_prefix}[Planner] 无法获取 SubHeartflow 或 Observation 来获取消息。") @@ -385,26 +387,28 @@ class PFChatting: try: observation_context_text = "" if observed_messages: - context_texts = [msg.get('detailed_plain_text', '') for msg in observed_messages if msg.get('detailed_plain_text')] + 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 + # 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 + 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] + 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 + 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.") + logger.warning(f"{log_prefix}[Planner] Skipping tool use because SubHeartflow is not available.") except Exception as e_tool: logger.error(f"{log_prefix}[Planner] Tool use failed: {e_tool}") @@ -422,20 +426,19 @@ class PFChatting: ) 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 + 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 = "[心流思考出错]" - + logger.error(f"{log_prefix}[Planner] SubHeartflow thinking failed: {e_shf}") + logger.error(traceback.format_exc()) + current_mind = "[心流思考出错]" # --- 使用 LLM 进行决策 --- - action = "no_reply" # Default action + action = "no_reply" # Default action emoji_query = "" reasoning = "默认决策或获取决策失败" - llm_error = False # Flag for LLM failure + llm_error = False # Flag for LLM failure try: # 构建提示 (Now includes current_mind) @@ -447,7 +450,7 @@ class PFChatting: "model": self.planner_llm.model_name, "messages": [{"role": "user", "content": prompt}], "tools": PLANNER_TOOL_DEFINITION, - "tool_choice": {"type": "function", "function": {"name": "decide_reply_action"}}, # 强制调用此工具 + "tool_choice": {"type": "function", "function": {"name": "decide_reply_action"}}, # 强制调用此工具 } logger.debug(f"{log_prefix}[Planner] 发送 Planner LLM 请求...") @@ -457,32 +460,43 @@ class PFChatting: ) # 解析 LLM 响应 - if len(response) == 3: # 期望返回 content, reasoning_content, tool_calls + if len(response) == 3: # 期望返回 content, reasoning_content, tool_calls _, _, tool_calls = response if tool_calls and isinstance(tool_calls, list) and len(tool_calls) > 0: # 通常强制调用后只会有一个 tool_call tool_call = tool_calls[0] - if tool_call.get("type") == "function" and tool_call.get("function", {}).get("name") == "decide_reply_action": + if ( + tool_call.get("type") == "function" + and tool_call.get("function", {}).get("name") == "decide_reply_action" + ): try: arguments = json.loads(tool_call["function"]["arguments"]) action = arguments.get("action", "no_reply") reasoning = arguments.get("reasoning", "未提供理由") if action == "emoji_reply": # Planner's decision overrides tool's emoji if action is emoji_reply - emoji_query = arguments.get("emoji_query", send_emoji_from_tools) # Use tool emoji as default if planner asks for emoji - logger.info(f"{log_prefix}[Planner] LLM 决策: {action}, 理由: {reasoning}, EmojiQuery: '{emoji_query}'") + emoji_query = arguments.get( + "emoji_query", send_emoji_from_tools + ) # Use tool emoji as default if planner asks for emoji + logger.info( + f"{log_prefix}[Planner] LLM 决策: {action}, 理由: {reasoning}, EmojiQuery: '{emoji_query}'" + ) except json.JSONDecodeError as json_e: - logger.error(f"{log_prefix}[Planner] 解析工具参数失败: {json_e}. Arguments: {tool_call['function'].get('arguments')}") + logger.error( + f"{log_prefix}[Planner] 解析工具参数失败: {json_e}. Arguments: {tool_call['function'].get('arguments')}" + ) action = "error" reasoning = "工具参数解析失败" llm_error = True except Exception as parse_e: logger.error(f"{log_prefix}[Planner] 处理工具参数时出错: {parse_e}") action = "error" - reasoning = "处理工具参数时出错" + reasoning = "处理工具参数时出错" llm_error = True else: - logger.warning(f"{log_prefix}[Planner] LLM 未按预期调用 'decide_reply_action' 工具。Tool calls: {tool_calls}") + logger.warning( + f"{log_prefix}[Planner] LLM 未按预期调用 'decide_reply_action' 工具。Tool calls: {tool_calls}" + ) action = "error" reasoning = "LLM未调用预期工具" llm_error = True @@ -509,11 +523,11 @@ class PFChatting: return { "action": action, "reasoning": reasoning, - "emoji_query": emoji_query, # Specific query if action is emoji_reply + "emoji_query": emoji_query, # Specific query if action is emoji_reply "current_mind": current_mind, - "send_emoji_from_tools": send_emoji_from_tools, # Emoji suggested by pre-thinking tools + "send_emoji_from_tools": send_emoji_from_tools, # Emoji suggested by pre-thinking tools "observed_messages": observed_messages, - "llm_error": llm_error # Indicate if LLM decision process failed + "llm_error": llm_error, # Indicate if LLM decision process failed } async def _get_anchor_message(self, observed_messages: List[dict]) -> Optional[MessageRecv]: @@ -535,34 +549,47 @@ class PFChatting: # Attempt reconstruction from the last observed message dictionary anchor_message = MessageRecv(last_msg_dict, chat_stream=self.chat_stream) # Basic validation - if not (anchor_message and anchor_message.message_info and anchor_message.message_info.message_id and anchor_message.message_info.user_info): + if not ( + anchor_message + and anchor_message.message_info + and anchor_message.message_info.message_id + and anchor_message.message_info.user_info + ): raise ValueError("重构的 MessageRecv 缺少必要信息.") - logger.debug(f"{self._get_log_prefix()} 重构的锚点消息: ID={anchor_message.message_info.message_id}") + logger.debug( + f"{self._get_log_prefix()} 重构的锚点消息: ID={anchor_message.message_info.message_id}" + ) return anchor_message except Exception as e_reconstruct: - logger.warning(f"{self._get_log_prefix()} 从观察到的消息重构 MessageRecv 失败: {e_reconstruct}. 创建占位符.") + logger.warning( + f"{self._get_log_prefix()} 从观察到的消息重构 MessageRecv 失败: {e_reconstruct}. 创建占位符." + ) else: logger.warning(f"{self._get_log_prefix()} observed_messages 为空. 创建占位符锚点消息.") # --- Create Placeholder --- placeholder_id = f"mid_pf_{int(time.time() * 1000)}" - placeholder_user = UserInfo(user_id="system_trigger", user_nickname="System Trigger", platform=self.chat_stream.platform) + placeholder_user = UserInfo( + user_id="system_trigger", user_nickname="System Trigger", platform=self.chat_stream.platform + ) placeholder_msg_info = BaseMessageInfo( message_id=placeholder_id, platform=self.chat_stream.platform, group_info=self.chat_stream.group_info, user_info=placeholder_user, - time=time.time() + time=time.time(), ) placeholder_msg_dict = { "message_info": placeholder_msg_info.to_dict(), - "processed_plain_text": "[System Trigger Context]", # Placeholder text + "processed_plain_text": "[System Trigger Context]", # Placeholder text "raw_message": "", "time": placeholder_msg_info.time, } anchor_message = MessageRecv(placeholder_msg_dict) - anchor_message.update_chat_stream(self.chat_stream) # Associate with the stream - logger.info(f"{self._get_log_prefix()} Created placeholder anchor message: ID={anchor_message.message_info.message_id}") + anchor_message.update_chat_stream(self.chat_stream) # Associate with the stream + logger.info( + f"{self._get_log_prefix()} Created placeholder anchor message: ID={anchor_message.message_info.message_id}" + ) return anchor_message except Exception as e: @@ -579,7 +606,6 @@ class PFChatting: except Exception as e: logger.error(f"{self._get_log_prefix()} Error cleaning up thinking message {thinking_id}: {e}") - async def _sender(self, thinking_id: str, anchor_message: MessageRecv, replier_result: Dict[str, Any]): """ 发送器 (Sender): 使用HeartFC_Chat的方法发送生成的回复。 @@ -589,13 +615,13 @@ class PFChatting: """ # replier_result should contain 'response_set' and 'send_emoji' response_set = replier_result.get("response_set") - send_emoji = replier_result.get("send_emoji", "") # Emoji determined by tools, passed via replier + send_emoji = replier_result.get("send_emoji", "") # Emoji determined by tools, passed via replier if not response_set: - logger.error(f"{self._get_log_prefix()}[Sender-{thinking_id}] Called with empty response_set.") - # Clean up thinking message before raising error - self._cleanup_thinking_message(thinking_id) - raise ValueError("Sender called with no response_set") # Signal failure to loop + logger.error(f"{self._get_log_prefix()}[Sender-{thinking_id}] Called with empty response_set.") + # Clean up thinking message before raising error + self._cleanup_thinking_message(thinking_id) + raise ValueError("Sender called with no response_set") # Signal failure to loop first_bot_msg: Optional[MessageSending] = None send_success = False @@ -606,18 +632,22 @@ class PFChatting: first_bot_msg = await self.heartfc_chat._send_response_messages(anchor_message, response_set, thinking_id) if first_bot_msg: - send_success = True # Mark success + send_success = True # Mark success logger.info(f"{self._get_log_prefix()}[Sender-{thinking_id}] Successfully sent reply.") # --- Handle associated emoji (if determined by tools) --- if send_emoji: - logger.info(f"{self._get_log_prefix()}[Sender-{thinking_id}] Sending associated emoji: {send_emoji}") + logger.info( + f"{self._get_log_prefix()}[Sender-{thinking_id}] Sending associated emoji: {send_emoji}" + ) try: # Use first_bot_msg as anchor if available, otherwise fallback to original anchor emoji_anchor = first_bot_msg if first_bot_msg else anchor_message await self.heartfc_chat._handle_emoji(emoji_anchor, response_set, send_emoji) except Exception as e_emoji: - logger.error(f"{self._get_log_prefix()}[Sender-{thinking_id}] Failed to send associated emoji: {e_emoji}") + logger.error( + f"{self._get_log_prefix()}[Sender-{thinking_id}] Failed to send associated emoji: {e_emoji}" + ) # Log error but don't fail the whole send process for emoji failure # --- Update relationship --- @@ -625,16 +655,19 @@ class PFChatting: await self.heartfc_chat._update_relationship(anchor_message, response_set) logger.debug(f"{self._get_log_prefix()}[Sender-{thinking_id}] Updated relationship.") except Exception as e_rel: - logger.error(f"{self._get_log_prefix()}[Sender-{thinking_id}] Failed to update relationship: {e_rel}") + logger.error( + f"{self._get_log_prefix()}[Sender-{thinking_id}] Failed to update relationship: {e_rel}" + ) # Log error but don't fail the whole send process for relationship update failure else: - # Sending failed (e.g., _send_response_messages found thinking message already gone) - send_success = False - logger.warning(f"{self._get_log_prefix()}[Sender-{thinking_id}] Failed to send reply (maybe thinking message expired or was removed?).") - # No need to clean up thinking message here, _send_response_messages implies it's gone or handled - raise RuntimeError("Sending reply failed, _send_response_messages returned None.") # Signal failure - + # Sending failed (e.g., _send_response_messages found thinking message already gone) + send_success = False + logger.warning( + f"{self._get_log_prefix()}[Sender-{thinking_id}] Failed to send reply (maybe thinking message expired or was removed?)." + ) + # No need to clean up thinking message here, _send_response_messages implies it's gone or handled + raise RuntimeError("Sending reply failed, _send_response_messages returned None.") # Signal failure except Exception as e: # Catch potential errors during sending or post-send actions @@ -643,11 +676,10 @@ class PFChatting: # Ensure thinking message is cleaned up if send failed mid-way and wasn't handled if not send_success: self._cleanup_thinking_message(thinking_id) - raise # Re-raise the exception to signal failure to the loop + raise # Re-raise the exception to signal failure to the loop # No finally block needed for lock management - async def shutdown(self): """ Gracefully shuts down the PFChatting instance by cancelling the active loop task. @@ -660,13 +692,13 @@ class PFChatting: # Wait briefly for the task to acknowledge cancellation await asyncio.wait_for(self._loop_task, timeout=5.0) except asyncio.CancelledError: - logger.info(f"{self._get_log_prefix()} PF loop task cancelled successfully.") + logger.info(f"{self._get_log_prefix()} PF loop task cancelled successfully.") except asyncio.TimeoutError: - logger.warning(f"{self._get_log_prefix()} Timeout waiting for PF loop task cancellation.") + logger.warning(f"{self._get_log_prefix()} Timeout waiting for PF loop task cancellation.") except Exception as e: - logger.error(f"{self._get_log_prefix()} Error during loop task cancellation: {e}") + logger.error(f"{self._get_log_prefix()} Error during loop task cancellation: {e}") else: - logger.info(f"{self._get_log_prefix()} No active PF loop task found to cancel.") + logger.info(f"{self._get_log_prefix()} No active PF loop task found to cancel.") # Ensure loop state is reset even if task wasn't running or cancellation failed self._loop_active = False @@ -685,30 +717,43 @@ class PFChatting: # Add current mind state if available if current_mind: - prompt += f"\n你当前的内部想法是：\n---\n{current_mind}\n---\n\n" + prompt += f"\n你当前的内部想法是：\n---\n{current_mind}\n---\n\n" else: - prompt += "\n你当前没有特别的内部想法。\n" + prompt += "\n你当前没有特别的内部想法。\n" if observed_messages: - context_text = "\n".join([msg.get('detailed_plain_text', '') for msg in observed_messages if msg.get('detailed_plain_text')]) + context_text = "\n".join( + [msg.get("detailed_plain_text", "") for msg in observed_messages if msg.get("detailed_plain_text")] + ) prompt += "观察到的最新聊天内容如下：\n---\n" - prompt += context_text[:1500] # Limit context length + prompt += context_text[:1500] # Limit context length prompt += "\n---\n" else: prompt += "当前没有观察到新的聊天内容。\n" - prompt += "\n请结合你的内部想法和观察到的聊天内容，分析情况并使用 'decide_reply_action' 工具来决定你的最终行动。\n" + prompt += ( + "\n请结合你的内部想法和观察到的聊天内容，分析情况并使用 'decide_reply_action' 工具来决定你的最终行动。\n" + ) prompt += "决策依据：\n" prompt += "1. 如果聊天内容无聊、与你无关、或者你的内部想法认为不适合回复，选择 'no_reply'。\n" prompt += "2. 如果聊天内容值得回应，且适合用文字表达（参考你的内部想法），选择 'text_reply'。\n" - prompt += "3. 如果聊天内容或你的内部想法适合用一个表情来回应，选择 'emoji_reply' 并提供表情主题 'emoji_query'。\n" + prompt += ( + "3. 如果聊天内容或你的内部想法适合用一个表情来回应，选择 'emoji_reply' 并提供表情主题 'emoji_query'。\n" + ) prompt += "4. 如果你已经回复过消息，也没有人又回复你，选择'no_reply'。" prompt += "必须调用 'decide_reply_action' 工具并提供 'action' 和 'reasoning'。" - return prompt + return prompt # --- 回复器 (Replier) 的定义 --- # - async def _replier_work(self, observed_messages: List[dict], anchor_message: MessageRecv, thinking_id: str, current_mind: Optional[str], send_emoji: str) -> Optional[Dict[str, Any]]: + async def _replier_work( + self, + observed_messages: List[dict], + anchor_message: MessageRecv, + thinking_id: str, + current_mind: Optional[str], + send_emoji: str, + ) -> Optional[Dict[str, Any]]: """ 回复器 (Replier): 核心逻辑用于生成回复。 被 _run_pf_loop 直接调用和 await。 @@ -724,23 +769,23 @@ class PFChatting: # 注意：实际的生成调用是在 self.heartfc_chat.gpt.generate_response 中 response_set = await self.heartfc_chat.gpt.generate_response( anchor_message, - thinking_id + thinking_id, # current_mind 不再直接传递给 gpt.generate_response， # 因为 generate_response 内部会通过 thinking_id 或其他方式获取所需上下文 ) if not response_set: logger.warning(f"{log_prefix}[Replier-{thinking_id}] LLM生成了一个空回复集。") - return None # Indicate failure + return None # Indicate failure # --- 准备并返回结果 --- logger.info(f"{log_prefix}[Replier-{thinking_id}] 成功生成了回复集: {' '.join(response_set)[:50]}...") return { "response_set": response_set, - "send_emoji": send_emoji, # Pass through the emoji determined earlier (usually by tools) + "send_emoji": send_emoji, # Pass through the emoji determined earlier (usually by tools) } except Exception as e: logger.error(f"{log_prefix}[Replier-{thinking_id}] Unexpected error in replier_work: {e}") logger.error(traceback.format_exc()) - return None # Indicate failure \ No newline at end of file + return None # Indicate failure