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

feat：为心流增加知识和知识缓存

Browse Source
This commit is contained in:
SengokuCola
2025-04-05 21:00:54 +08:00
parent db14d9c39b
commit 8cc4e19947
3 changed files with 360 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

300
src/heart_flow/sub_heartflow.py
View File

@@ -8,6 +8,9 @@ import time
from src.plugins.schedule.schedule_generator import bot_schedule
from src.plugins.memory_system.Hippocampus import HippocampusManager
from src.common.logger import get_module_logger, LogConfig, SUB_HEARTFLOW_STYLE_CONFIG # noqa: E402
from src.plugins.chat.utils import get_embedding
from src.common.database import db
from typing import Union
subheartflow_config = LogConfig(
# 使用海马体专用样式
@@ -53,6 +56,8 @@ class SubHeartflow:
self.is_active = False
self.observations: list[Observation] = []
self.running_knowledges = []
def add_observation(self, observation: Observation):
"""添加一个新的observation对象到列表中，如果已存在相同id的observation则不添加"""
@@ -98,49 +103,49 @@ class SubHeartflow:
logger.info(f"子心流 {self.subheartflow_id} 已经5分钟没有激活，正在销毁...")
break # 退出循环，销毁自己
async def do_a_thinking(self):
current_thinking_info = self.current_mind
mood_info = self.current_state.mood
# async def do_a_thinking(self):
# current_thinking_info = self.current_mind
# mood_info = self.current_state.mood
observation = self.observations[0]
chat_observe_info = observation.observe_info
# print(f"chat_observe_info：{chat_observe_info}")
# observation = self.observations[0]
# chat_observe_info = observation.observe_info
# # print(f"chat_observe_info：{chat_observe_info}")
# 调取记忆
related_memory = await HippocampusManager.get_instance().get_memory_from_text(
text=chat_observe_info, max_memory_num=2, max_memory_length=2, max_depth=3, fast_retrieval=False
)
# # 调取记忆
# related_memory = await HippocampusManager.get_instance().get_memory_from_text(
# text=chat_observe_info, max_memory_num=2, max_memory_length=2, max_depth=3, fast_retrieval=False
# )
if related_memory:
related_memory_info = ""
for memory in related_memory:
related_memory_info += memory[1]
else:
related_memory_info = ""
# if related_memory:
# related_memory_info = ""
# for memory in related_memory:
# related_memory_info += memory[1]
# else:
# related_memory_info = ""
# print(f"相关记忆：{related_memory_info}")
# # print(f"相关记忆：{related_memory_info}")
schedule_info = bot_schedule.get_current_num_task(num=1, time_info=False)
# schedule_info = bot_schedule.get_current_num_task(num=1, time_info=False)
prompt = ""
prompt += f"你刚刚在做的事情是：{schedule_info}\n"
# prompt += f"麦麦的总体想法是：{self.main_heartflow_info}\n\n"
prompt += f"你{self.personality_info}\n"
if related_memory_info:
prompt += f"你想起来你之前见过的回忆：{related_memory_info}。\n以上是你的回忆，不一定是目前聊天里的人说的，也不一定是现在发生的事情，请记住。\n"
prompt += f"刚刚你的想法是{current_thinking_info}。\n"
prompt += "-----------------------------------\n"
prompt += f"现在你正在上网，和qq群里的网友们聊天，群里正在聊的话题是：{chat_observe_info}\n"
prompt += f"你现在{mood_info}\n"
prompt += "现在你接下去继续思考，产生新的想法，不要分点输出，输出连贯的内心独白，不要太长，"
prompt += "但是记得结合上述的消息，要记得维持住你的人设，关注聊天和新内容，不要思考太多:"
reponse, reasoning_content = await self.llm_model.generate_response_async(prompt)
# prompt = ""
# prompt += f"你刚刚在做的事情是：{schedule_info}\n"
# # prompt += f"麦麦的总体想法是：{self.main_heartflow_info}\n\n"
# prompt += f"你{self.personality_info}\n"
# if related_memory_info:
# prompt += f"你想起来你之前见过的回忆：{related_memory_info}。\n以上是你的回忆，不一定是目前聊天里的人说的，也不一定是现在发生的事情，请记住。\n"
# prompt += f"刚刚你的想法是{current_thinking_info}。\n"
# prompt += "-----------------------------------\n"
# prompt += f"现在你正在上网，和qq群里的网友们聊天，群里正在聊的话题是：{chat_observe_info}\n"
# prompt += f"你现在{mood_info}\n"
# prompt += "现在你接下去继续思考，产生新的想法，不要分点输出，输出连贯的内心独白，不要太长，"
# prompt += "但是记得结合上述的消息，要记得维持住你的人设，关注聊天和新内容，不要思考太多:"
# reponse, reasoning_content = await self.llm_model.generate_response_async(prompt)
self.update_current_mind(reponse)
# self.update_current_mind(reponse)
self.current_mind = reponse
logger.debug(f"prompt:\n{prompt}\n")
logger.info(f"麦麦的脑内状态：{self.current_mind}")
# self.current_mind = reponse
# logger.debug(f"prompt:\n{prompt}\n")
# logger.info(f"麦麦的脑内状态：{self.current_mind}")
async def do_observe(self):
observation = self.observations[0]
@@ -166,6 +171,13 @@ class SubHeartflow:
else:
related_memory_info = ""
related_info,grouped_results = await self.get_prompt_info(chat_observe_info + message_txt, 0.4)
print(related_info)
for topic, results in grouped_results.items():
for result in results:
print(result)
self.running_knowledges.append(result)
# print(f"相关记忆：{related_memory_info}")
schedule_info = bot_schedule.get_current_num_task(num=1, time_info=False)
@@ -176,6 +188,8 @@ class SubHeartflow:
prompt += f"你刚刚在做的事情是：{schedule_info}\n"
if related_memory_info:
prompt += f"你想起来你之前见过的回忆：{related_memory_info}。\n以上是你的回忆，不一定是目前聊天里的人说的，也不一定是现在发生的事情，请记住。\n"
if related_info:
prompt += f"你想起你知道：{related_info}\n"
prompt += f"刚刚你的想法是{current_thinking_info}。\n"
prompt += "-----------------------------------\n"
prompt += f"现在你正在上网，和qq群里的网友们聊天，群里正在聊的话题是：{chat_observe_info}\n"
@@ -249,6 +263,222 @@ class SubHeartflow:
def update_current_mind(self, reponse):
self.past_mind.append(self.current_mind)
self.current_mind = reponse
async def get_prompt_info(self, message: str, threshold: float):
start_time = time.time()
related_info = ""
logger.debug(f"获取知识库内容，元消息：{message[:30]}...，消息长度: {len(message)}")
# 1. 先从LLM获取主题，类似于记忆系统的做法
topics = []
# try:
# # 先尝试使用记忆系统的方法获取主题
# hippocampus = HippocampusManager.get_instance()._hippocampus
# topic_num = min(5, max(1, int(len(message) * 0.1)))
# topics_response = await hippocampus.llm_topic_judge.generate_response(hippocampus.find_topic_llm(message, topic_num))
# # 提取关键词
# topics = re.findall(r"<([^>]+)>", topics_response[0])
# if not topics:
# topics = []
# else:
# topics = [
# topic.strip()
# for topic in ",".join(topics).replace("，", ",").replace("、", ",").replace(" ", ",").split(",")
# if topic.strip()
# ]
# logger.info(f"从LLM提取的主题: {', '.join(topics)}")
# except Exception as e:
# logger.error(f"从LLM提取主题失败: {str(e)}")
# # 如果LLM提取失败，使用jieba分词提取关键词作为备选
# words = jieba.cut(message)
# topics = [word for word in words if len(word) > 1][:5]
# logger.info(f"使用jieba提取的主题: {', '.join(topics)}")
# 如果无法提取到主题，直接使用整个消息
if not topics:
logger.info("未能提取到任何主题，使用整个消息进行查询")
embedding = await get_embedding(message, request_type="info_retrieval")
if not embedding:
logger.error("获取消息嵌入向量失败")
return ""
related_info = self.get_info_from_db(embedding, limit=3, threshold=threshold)
logger.info(f"知识库检索完成，总耗时: {time.time() - start_time:.3f}秒")
return related_info, {}
# 2. 对每个主题进行知识库查询
logger.info(f"开始处理{len(topics)}个主题的知识库查询")
# 优化：批量获取嵌入向量，减少API调用
embeddings = {}
topics_batch = [topic for topic in topics if len(topic) > 0]
if message: # 确保消息非空
topics_batch.append(message)
# 批量获取嵌入向量
embed_start_time = time.time()
for text in topics_batch:
if not text or len(text.strip()) == 0:
continue
try:
embedding = await get_embedding(text, request_type="info_retrieval")
if embedding:
embeddings[text] = embedding
else:
logger.warning(f"获取'{text}'的嵌入向量失败")
except Exception as e:
logger.error(f"获取'{text}'的嵌入向量时发生错误: {str(e)}")
logger.info(f"批量获取嵌入向量完成，耗时: {time.time() - embed_start_time:.3f}秒")
if not embeddings:
logger.error("所有嵌入向量获取失败")
return ""
# 3. 对每个主题进行知识库查询
all_results = []
query_start_time = time.time()
# 首先添加原始消息的查询结果
if message in embeddings:
original_results = self.get_info_from_db(embeddings[message], limit=3, threshold=threshold, return_raw=True)
if original_results:
for result in original_results:
result["topic"] = "原始消息"
all_results.extend(original_results)
logger.info(f"原始消息查询到{len(original_results)}条结果")
# 然后添加每个主题的查询结果
for topic in topics:
if not topic or topic not in embeddings:
continue
try:
topic_results = self.get_info_from_db(embeddings[topic], limit=3, threshold=threshold, return_raw=True)
if topic_results:
# 添加主题标记
for result in topic_results:
result["topic"] = topic
all_results.extend(topic_results)
logger.info(f"主题'{topic}'查询到{len(topic_results)}条结果")
except Exception as e:
logger.error(f"查询主题'{topic}'时发生错误: {str(e)}")
logger.info(f"知识库查询完成，耗时: {time.time() - query_start_time:.3f}秒，共获取{len(all_results)}条结果")
# 4. 去重和过滤
process_start_time = time.time()
unique_contents = set()
filtered_results = []
for result in all_results:
content = result["content"]
if content not in unique_contents:
unique_contents.add(content)
filtered_results.append(result)
# 5. 按相似度排序
filtered_results.sort(key=lambda x: x["similarity"], reverse=True)
# 6. 限制总数量（最多10条）
filtered_results = filtered_results[:10]
logger.info(f"结果处理完成，耗时: {time.time() - process_start_time:.3f}秒，过滤后剩余{len(filtered_results)}条结果")
# 7. 格式化输出
if filtered_results:
format_start_time = time.time()
grouped_results = {}
for result in filtered_results:
topic = result["topic"]
if topic not in grouped_results:
grouped_results[topic] = []
grouped_results[topic].append(result)
# 按主题组织输出
for topic, results in grouped_results.items():
related_info += f"【主题: {topic}】\n"
for i, result in enumerate(results, 1):
similarity = result["similarity"]
content = result["content"].strip()
# 调试：为内容添加序号和相似度信息
# related_info += f"{i}. [{similarity:.2f}] {content}\n"
related_info += f"{content}\n"
related_info += "\n"
logger.info(f"格式化输出完成，耗时: {time.time() - format_start_time:.3f}秒")
logger.info(f"知识库检索总耗时: {time.time() - start_time:.3f}秒")
return related_info,grouped_results
def get_info_from_db(self, query_embedding: list, limit: int = 1, threshold: float = 0.5, return_raw: bool = False) -> Union[str, list]:
if not query_embedding:
return "" if not return_raw else []
# 使用余弦相似度计算
pipeline = [
{
"$addFields": {
"dotProduct": {
"$reduce": {
"input": {"$range": [0, {"$size": "$embedding"}]},
"initialValue": 0,
"in": {
"$add": [
"$$value",
{
"$multiply": [
{"$arrayElemAt": ["$embedding", "$$this"]},
{"$arrayElemAt": [query_embedding, "$$this"]},
]
},
]
},
}
},
"magnitude1": {
"$sqrt": {
"$reduce": {
"input": "$embedding",
"initialValue": 0,
"in": {"$add": ["$$value", {"$multiply": ["$$this", "$$this"]}]},
}
}
},
"magnitude2": {
"$sqrt": {
"$reduce": {
"input": query_embedding,
"initialValue": 0,
"in": {"$add": ["$$value", {"$multiply": ["$$this", "$$this"]}]},
}
}
},
}
},
{"$addFields": {"similarity": {"$divide": ["$dotProduct", {"$multiply": ["$magnitude1", "$magnitude2"]}]}}},
{
"$match": {
"similarity": {"$gte": threshold} # 只保留相似度大于等于阈值的结果
}
},
{"$sort": {"similarity": -1}},
{"$limit": limit},
{"$project": {"content": 1, "similarity": 1}},
]
results = list(db.knowledges.aggregate(pipeline))
logger.debug(f"知识库查询结果数量: {len(results)}")
if not results:
return "" if not return_raw else []
if return_raw:
return results
else:
# 返回所有找到的内容，用换行分隔
return "\n".join(str(result["content"]) for result in results)
# subheartflow = SubHeartflow()

44
src/plugins/chat_module/reasoning_chat/reasoning_prompt_builder.py
View File

@@ -167,30 +167,30 @@ class PromptBuilder:
# 1. 先从LLM获取主题，类似于记忆系统的做法
topics = []
try:
# 先尝试使用记忆系统的方法获取主题
hippocampus = HippocampusManager.get_instance()._hippocampus
topic_num = min(5, max(1, int(len(message) * 0.1)))
topics_response = await hippocampus.llm_topic_judge.generate_response(hippocampus.find_topic_llm(message, topic_num))
# try:
# # 先尝试使用记忆系统的方法获取主题
# hippocampus = HippocampusManager.get_instance()._hippocampus
# topic_num = min(5, max(1, int(len(message) * 0.1)))
# topics_response = await hippocampus.llm_topic_judge.generate_response(hippocampus.find_topic_llm(message, topic_num))
# 提取关键词
topics = re.findall(r"<([^>]+)>", topics_response[0])
if not topics:
topics = []
else:
topics = [
topic.strip()
for topic in ",".join(topics).replace("，", ",").replace("、", ",").replace(" ", ",").split(",")
if topic.strip()
]
# # 提取关键词
# topics = re.findall(r"<([^>]+)>", topics_response[0])
# if not topics:
# topics = []
# else:
# topics = [
# topic.strip()
# for topic in ",".join(topics).replace("，", ",").replace("、", ",").replace(" ", ",").split(",")
# if topic.strip()
# ]
logger.info(f"从LLM提取的主题: {', '.join(topics)}")
except Exception as e:
logger.error(f"从LLM提取主题失败: {str(e)}")
# 如果LLM提取失败，使用jieba分词提取关键词作为备选
words = jieba.cut(message)
topics = [word for word in words if len(word) > 1][:5]
logger.info(f"使用jieba提取的主题: {', '.join(topics)}")
# logger.info(f"从LLM提取的主题: {', '.join(topics)}")
# except Exception as e:
# logger.error(f"从LLM提取主题失败: {str(e)}")
# # 如果LLM提取失败，使用jieba分词提取关键词作为备选
# words = jieba.cut(message)
# topics = [word for word in words if len(word) > 1][:5]
# logger.info(f"使用jieba提取的主题: {', '.join(topics)}")
# 如果无法提取到主题，直接使用整个消息
if not topics:

121
src/plugins/chat_module/think_flow_chat/think_flow_chat.py
View File

@@ -236,59 +236,84 @@ class ThinkFlowChat:
do_reply = False
if random() < reply_probability:
do_reply = True
# 创建思考消息
timer1 = time.time()
thinking_id = await self._create_thinking_message(message, chat, userinfo, messageinfo)
timer2 = time.time()
timing_results["创建思考消息"] = timer2 - timer1
# 观察
timer1 = time.time()
await heartflow.get_subheartflow(chat.stream_id).do_observe()
timer2 = time.time()
timing_results["观察"] = timer2 - timer1
# 思考前脑内状态
timer1 = time.time()
await heartflow.get_subheartflow(chat.stream_id).do_thinking_before_reply(message.processed_plain_text)
timer2 = time.time()
timing_results["思考前脑内状态"] = timer2 - timer1
# 生成回复
timer1 = time.time()
response_set = await self.gpt.generate_response(message)
timer2 = time.time()
timing_results["生成回复"] = timer2 - timer1
try:
do_reply = True
# 创建思考消息
try:
timer1 = time.time()
thinking_id = await self._create_thinking_message(message, chat, userinfo, messageinfo)
timer2 = time.time()
timing_results["创建思考消息"] = timer2 - timer1
except Exception as e:
logger.error(f"心流创建思考消息失败: {e}")
try:
# 观察
timer1 = time.time()
await heartflow.get_subheartflow(chat.stream_id).do_observe()
timer2 = time.time()
timing_results["观察"] = timer2 - timer1
except Exception as e:
logger.error(f"心流观察失败: {e}")
if not response_set:
logger.info("为什么生成回复失败？")
return
# 思考前脑内状态
try:
timer1 = time.time()
await heartflow.get_subheartflow(chat.stream_id).do_thinking_before_reply(message.processed_plain_text)
timer2 = time.time()
timing_results["思考前脑内状态"] = timer2 - timer1
except Exception as e:
logger.error(f"心流思考前脑内状态失败: {e}")
# 生成回复
timer1 = time.time()
response_set = await self.gpt.generate_response(message)
timer2 = time.time()
timing_results["生成回复"] = timer2 - timer1
# 发送消息
timer1 = time.time()
await self._send_response_messages(message, chat, response_set, thinking_id)
timer2 = time.time()
timing_results["发送消息"] = timer2 - timer1
if not response_set:
logger.info("为什么生成回复失败？")
return
# 处理表情包
timer1 = time.time()
await self._handle_emoji(message, chat, response_set)
timer2 = time.time()
timing_results["处理表情包"] = timer2 - timer1
# 发送消息
try:
timer1 = time.time()
await self._send_response_messages(message, chat, response_set, thinking_id)
timer2 = time.time()
timing_results["发送消息"] = timer2 - timer1
except Exception as e:
logger.error(f"心流发送消息失败: {e}")
# 更新心流
timer1 = time.time()
await self._update_using_response(message, response_set)
timer2 = time.time()
timing_results["更新心流"] = timer2 - timer1
# 处理表情包
try:
timer1 = time.time()
await self._handle_emoji(message, chat, response_set)
timer2 = time.time()
timing_results["处理表情包"] = timer2 - timer1
except Exception as e:
logger.error(f"心流处理表情包失败: {e}")
# 更新关系情绪
timer1 = time.time()
await self._update_relationship(message, response_set)
timer2 = time.time()
timing_results["更新关系情绪"] = timer2 - timer1
# 更新心流
try:
timer1 = time.time()
await self._update_using_response(message, response_set)
timer2 = time.time()
timing_results["更新心流"] = timer2 - timer1
except Exception as e:
logger.error(f"心流更新失败: {e}")
# 更新关系情绪
try:
timer1 = time.time()
await self._update_relationship(message, response_set)
timer2 = time.time()
timing_results["更新关系情绪"] = timer2 - timer1
except Exception as e:
logger.error(f"心流更新关系情绪失败: {e}")
except Exception as e:
logger.error(f"心流处理消息失败: {e}")
# 输出性能计时结果
if do_reply: