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

🤖 自动格式化代码 [skip ci]

Browse Source
This commit is contained in:
github-actions[bot]
2025-04-18 03:37:20 +00:00
parent c0dcd578c9
commit dfe788c65c
12 changed files with 610 additions and 487 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
src/heart_flow/observation.py
View File

@@ -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):

52
src/heart_flow/sub_heartflow.py
View File

@@ -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()