Merge branch 'MaiM-with-u:dev' into dev

This commit is contained in:
UnCLAS-Prommer
2025-04-21 13:25:28 +08:00
committed by GitHub
23 changed files with 1261 additions and 2327 deletions

View File

@@ -21,6 +21,11 @@ from ...common.database import db
logger = get_module_logger("chat_utils")
def is_english_letter(char: str) -> bool:
"""检查字符是否为英文字母(忽略大小写)"""
return "a" <= char.lower() <= "z"
def db_message_to_str(message_dict: Dict) -> str:
logger.debug(f"message_dict: {message_dict}")
time_str = time.strftime("%m-%d %H:%M:%S", time.localtime(message_dict["time"]))
@@ -71,7 +76,7 @@ def is_mentioned_bot_in_message(message: MessageRecv) -> tuple[bool, float]:
else:
if not is_mentioned:
# 判断是否被回复
if re.match(f"回复[\s\S]*?\({global_config.BOT_QQ}\)的消息,说:", message.processed_plain_text):
if re.match("回复[\s\S]*?\((\d+)\)的消息,说:", message.processed_plain_text):
is_mentioned = True
# 判断内容中是否被提及
@@ -217,97 +222,114 @@ def get_recent_group_speaker(chat_stream_id: int, sender, limit: int = 12) -> li
def split_into_sentences_w_remove_punctuation(text: str) -> List[str]:
"""将文本分割成句子,但保持书名号中的内容完整
"""将文本分割成句子,并根据概率合并
1. 识别分割点(, 。 ; 空格),但如果分割点左右都是英文字母则不分割。
2. 将文本分割成 (内容, 分隔符) 的元组。
3. 根据原始文本长度计算合并概率,概率性地合并相邻段落。
注意:此函数假定颜文字已在上层被保护。
Args:
text: 要分割的文本字符串
text: 要分割的文本字符串 (假定颜文字已被保护)
Returns:
List[str]: 分割后的句子列表
List[str]: 分割和合并后的句子列表
"""
# 处理两个汉字中间的换行符
text = re.sub(r"([\u4e00-\u9fff])\n([\u4e00-\u9fff])", r"\1。\2", text)
len_text = len(text)
if len_text < 4:
if len_text < 3:
if random.random() < 0.01:
return list(text) # 如果文本很短且触发随机条件,直接按字符分割
else:
return [text]
# 定义分隔符
separators = {"", ",", " ", "", ";"}
segments = []
current_segment = ""
# 1. 分割成 (内容, 分隔符) 元组
i = 0
while i < len(text):
char = text[i]
if char in separators:
# 检查分割条件:如果分隔符左右都是英文字母,则不分割
can_split = True
if i > 0 and i < len(text) - 1:
prev_char = text[i - 1]
next_char = text[i + 1]
# if is_english_letter(prev_char) and is_english_letter(next_char) and char == ' ': # 原计划只对空格应用此规则,现应用于所有分隔符
if is_english_letter(prev_char) and is_english_letter(next_char):
can_split = False
if can_split:
# 只有当当前段不为空时才添加
if current_segment:
segments.append((current_segment, char))
# 如果当前段为空,但分隔符是空格,则也添加一个空段(保留空格)
elif char == " ":
segments.append(("", char))
current_segment = ""
else:
# 不分割,将分隔符加入当前段
current_segment += char
else:
current_segment += char
i += 1
# 添加最后一个段(没有后续分隔符)
if current_segment:
segments.append((current_segment, ""))
# 过滤掉完全空的段(内容和分隔符都为空)
segments = [(content, sep) for content, sep in segments if content or sep]
# 如果分割后为空(例如,输入全是分隔符且不满足保留条件),恢复颜文字并返回
if not segments:
# recovered_text = recover_kaomoji([text], mapping) # 恢复原文本中的颜文字 - 已移至上层处理
# return [s for s in recovered_text if s] # 返回非空结果
return [text] if text else [] # 如果原始文本非空,则返回原始文本(可能只包含未被分割的字符或颜文字占位符)
# 2. 概率合并
if len_text < 12:
split_strength = 0.2
elif len_text < 32:
split_strength = 0.6
else:
split_strength = 0.7
# 合并概率与分割强度相反
merge_probability = 1.0 - split_strength
# 检查是否为西文字符段落
if not is_western_paragraph(text):
# 当语言为中文时,统一将英文逗号转换为中文逗号
text = text.replace(",", "")
text = text.replace("\n", " ")
else:
# 用"|seg|"作为分割符分开
text = re.sub(r"([.!?]) +", r"\1\|seg\|", text)
text = text.replace("\n", "|seg|")
text, mapping = protect_kaomoji(text)
# print(f"处理前的文本: {text}")
merged_segments = []
idx = 0
while idx < len(segments):
current_content, current_sep = segments[idx]
text_no_1 = ""
for letter in text:
# print(f"当前字符: {letter}")
if letter in ["!", "", "?", ""]:
# print(f"当前字符: {letter}, 随机数: {random.random()}")
if random.random() < split_strength:
letter = ""
if letter in ["", ""]:
# print(f"当前字符: {letter}, 随机数: {random.random()}")
if random.random() < 1 - split_strength:
letter = ""
text_no_1 += letter
# 检查是否可以与下一段合并
# 条件:不是最后一段,且随机数小于合并概率,且当前段有内容(避免合并空段)
if idx + 1 < len(segments) and random.random() < merge_probability and current_content:
next_content, next_sep = segments[idx + 1]
# 合并: (内容1 + 分隔符1 + 内容2, 分隔符2)
# 只有当下一段也有内容时才合并文本,否则只传递分隔符
if next_content:
merged_content = current_content + current_sep + next_content
merged_segments.append((merged_content, next_sep))
else: # 下一段内容为空,只保留当前内容和下一段的分隔符
merged_segments.append((current_content, next_sep))
# 对每个逗号单独判断是否分割
sentences = [text_no_1]
new_sentences = []
for sentence in sentences:
parts = sentence.split("")
current_sentence = parts[0]
if not is_western_paragraph(current_sentence):
for part in parts[1:]:
if random.random() < split_strength:
new_sentences.append(current_sentence.strip())
current_sentence = part
else:
current_sentence += "" + part
# 处理空格分割
space_parts = current_sentence.split(" ")
current_sentence = space_parts[0]
for part in space_parts[1:]:
if random.random() < split_strength:
new_sentences.append(current_sentence.strip())
current_sentence = part
else:
current_sentence += " " + part
idx += 2 # 跳过下一段,因为它已被合并
else:
# 处理分割符
space_parts = current_sentence.split("|seg|")
current_sentence = space_parts[0]
for part in space_parts[1:]:
new_sentences.append(current_sentence.strip())
current_sentence = part
new_sentences.append(current_sentence.strip())
sentences = [s for s in new_sentences if s] # 移除空字符串
sentences = recover_kaomoji(sentences, mapping)
# 不合并,直接添加当前段
merged_segments.append((current_content, current_sep))
idx += 1
# print(f"分割后的句子: {sentences}")
sentences_done = []
for sentence in sentences:
sentence = sentence.rstrip(",")
# 西文字符句子不进行随机合并
if not is_western_paragraph(current_sentence):
if random.random() < split_strength * 0.5:
sentence = sentence.replace("", "").replace(",", "")
elif random.random() < split_strength:
sentence = sentence.replace("", " ").replace(",", " ")
sentences_done.append(sentence)
# 提取最终的句子内容
final_sentences = [content for content, sep in merged_segments if content] # 只保留有内容的段
logger.debug(f"处理后的句子: {sentences_done}")
return sentences_done
# 清理可能引入的空字符串
final_sentences = [s for s in final_sentences if s]
logger.debug(f"分割并合并后的句子: {final_sentences}")
return final_sentences
def random_remove_punctuation(text: str) -> str:
@@ -341,13 +363,11 @@ def process_llm_response(text: str) -> List[str]:
# 先保护颜文字
protected_text, kaomoji_mapping = protect_kaomoji(text)
logger.trace(f"保护颜文字后的文本: {protected_text}")
# 提取被 () 或 [] 包裹的内容
pattern = re.compile(r"[\(\[\].*?[\)\]\]")
# 提取被 () 或 [] 包裹且包含中文的内容
pattern = re.compile(r"[\(\[\](?=.*[\u4e00-\u9fff]).*?[\)\]\]")
# _extracted_contents = pattern.findall(text)
_extracted_contents = pattern.findall(protected_text) # 在保护后的文本上查找
extracted_contents = pattern.findall(protected_text) # 在保护后的文本上查找
# 去除 () 和 [] 及其包裹的内容
# cleaned_text = pattern.sub("", text)
cleaned_text = pattern.sub("", protected_text)
if cleaned_text == "":
@@ -358,12 +378,11 @@ def process_llm_response(text: str) -> List[str]:
# 对清理后的文本进行进一步处理
max_length = global_config.response_max_length * 2
max_sentence_num = global_config.response_max_sentence_num
if len(cleaned_text) > max_length and not is_western_paragraph(cleaned_text):
logger.warning(f"回复过长 ({len(cleaned_text)} 字符),返回默认回复")
return ["懒得说"]
elif len(cleaned_text) > 200:
logger.warning(f"回复过长 ({len(cleaned_text)} 字符),返回默认回复")
return ["懒得说"]
# 如果基本上是中文,则进行长度过滤
if get_western_ratio(cleaned_text) < 0.1:
if len(cleaned_text) > max_length:
logger.warning(f"回复过长 ({len(cleaned_text)} 字符),返回默认回复")
return ["懒得说"]
typo_generator = ChineseTypoGenerator(
error_rate=global_config.chinese_typo_error_rate,
@@ -390,11 +409,14 @@ def process_llm_response(text: str) -> List[str]:
if len(sentences) > max_sentence_num:
logger.warning(f"分割后消息数量过多 ({len(sentences)} 条),返回默认回复")
return [f"{global_config.BOT_NICKNAME}不知道哦"]
# sentences.extend(extracted_contents)
if extracted_contents:
for content in extracted_contents:
sentences.append(content)
# 在所有句子处理完毕后,对包含占位符的列表进行恢复
sentences = recover_kaomoji(sentences, kaomoji_mapping)
print(sentences)
return sentences
@@ -552,14 +574,24 @@ def recover_kaomoji(sentences, placeholder_to_kaomoji):
return recovered_sentences
def is_western_char(char):
"""检测是否为西文字符"""
return len(char.encode("utf-8")) <= 2
def get_western_ratio(paragraph):
"""计算段落中字母数字字符的西文比例
原理:检查段落中字母数字字符的西文比例
通过is_english_letter函数判断每个字符是否为西文
只检查字母数字字符,忽略标点符号和空格等非字母数字字符
Args:
paragraph: 要检查的文本段落
def is_western_paragraph(paragraph):
"""检测是否为西文字符段落"""
return all(is_western_char(char) for char in paragraph if char.isalnum())
Returns:
float: 西文字符比例(0.0-1.0)如果没有字母数字字符则返回0.0
"""
alnum_chars = [char for char in paragraph if char.isalnum()]
if not alnum_chars:
return 0.0
western_count = sum(1 for char in alnum_chars if is_english_letter(char))
return western_count / len(alnum_chars)
def count_messages_between(start_time: float, end_time: float, stream_id: str) -> tuple[int, int]:
@@ -673,19 +705,17 @@ def translate_timestamp_to_human_readable(timestamp: float, mode: str = "normal"
diff = now - timestamp
if diff < 20:
return "刚刚:"
return "刚刚:\n"
elif diff < 60:
return f"{int(diff)}秒前:"
elif diff < 1800:
return f"{int(diff / 60)}分钟前:"
return f"{int(diff)}秒前:\n"
elif diff < 3600:
return f"{int(diff / 60)}分钟前:\n"
elif diff < 86400:
return f"{int(diff / 3600)}小时前:\n"
elif diff < 604800:
elif diff < 86400 * 2:
return f"{int(diff / 86400)}天前:\n"
else:
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(timestamp)) + ":"
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(timestamp)) + ":\n"
def parse_text_timestamps(text: str, mode: str = "normal") -> str:

View File

@@ -118,10 +118,10 @@ class ImageManager:
# 调用AI获取描述
if image_format == "gif" or image_format == "GIF":
image_base64 = self.transform_gif(image_base64)
prompt = "这是一个动态图表情包,每一张图代表了动态图的某一帧,黑色背景代表透明,使用个词描述一下表情包表达的情感,简短一些"
prompt = "这是一个动态图表情包,每一张图代表了动态图的某一帧,黑色背景代表透明,使用1-2个词描述一下表情包表达的情感和内容,简短一些"
description, _ = await self._llm.generate_response_for_image(prompt, image_base64, "jpg")
else:
prompt = "这是一个表情包,描述一下表情包所表达的情感,请用使用一个词"
prompt = "这是一个表情包,请用使用1-2个词描述一下表情包所表达的情感和内容,简短一些"
description, _ = await self._llm.generate_response_for_image(prompt, image_base64, image_format)
cached_description = self._get_description_from_db(image_hash, "emoji")

View File

@@ -1,486 +0,0 @@
import time
from random import random
import traceback
from typing import List
from ...memory_system.Hippocampus import HippocampusManager
from ...moods.moods import MoodManager
from ....config.config import global_config
from ...chat.emoji_manager import emoji_manager
from .think_flow_generator import ResponseGenerator
from ...chat.message import MessageSending, MessageRecv, MessageThinking, MessageSet
from ...chat.messagesender import message_manager
from ...storage.storage import MessageStorage
from ...chat.utils import is_mentioned_bot_in_message
from ...chat.utils_image import image_path_to_base64
from ...willing.willing_manager import willing_manager
from ...message import UserInfo, Seg
from src.heart_flow.heartflow import heartflow
from src.common.logger import get_module_logger, CHAT_STYLE_CONFIG, LogConfig
from ...chat.chat_stream import chat_manager
from ...person_info.relationship_manager import relationship_manager
from ...chat.message_buffer import message_buffer
from src.plugins.respon_info_catcher.info_catcher import info_catcher_manager
from ...utils.timer_calculater import Timer
from src.do_tool.tool_use import ToolUser
# 定义日志配置
chat_config = LogConfig(
console_format=CHAT_STYLE_CONFIG["console_format"],
file_format=CHAT_STYLE_CONFIG["file_format"],
)
logger = get_module_logger("think_flow_chat", config=chat_config)
class ThinkFlowChat:
def __init__(self):
self.storage = MessageStorage()
self.gpt = ResponseGenerator()
self.mood_manager = MoodManager.get_instance()
self.mood_manager.start_mood_update()
self.tool_user = ToolUser()
@staticmethod
async def _create_thinking_message(message, chat, userinfo, messageinfo):
"""创建思考消息"""
bot_user_info = UserInfo(
user_id=global_config.BOT_QQ,
user_nickname=global_config.BOT_NICKNAME,
platform=messageinfo.platform,
)
thinking_time_point = round(time.time(), 2)
thinking_id = "mt" + str(thinking_time_point)
thinking_message = MessageThinking(
message_id=thinking_id,
chat_stream=chat,
bot_user_info=bot_user_info,
reply=message,
thinking_start_time=thinking_time_point,
)
message_manager.add_message(thinking_message)
return thinking_id
@staticmethod
async def _send_response_messages(message, chat, response_set: List[str], thinking_id) -> MessageSending:
"""发送回复消息"""
container = message_manager.get_container(chat.stream_id)
thinking_message = None
for msg in container.messages:
if isinstance(msg, MessageThinking) and msg.message_info.message_id == thinking_id:
thinking_message = msg
container.messages.remove(msg)
break
if not thinking_message:
logger.warning("未找到对应的思考消息,可能已超时被移除")
return None
thinking_start_time = thinking_message.thinking_start_time
message_set = MessageSet(chat, thinking_id)
mark_head = False
first_bot_msg = None
for msg in response_set:
message_segment = Seg(type="text", data=msg)
bot_message = MessageSending(
message_id=thinking_id,
chat_stream=chat,
bot_user_info=UserInfo(
user_id=global_config.BOT_QQ,
user_nickname=global_config.BOT_NICKNAME,
platform=message.message_info.platform,
),
sender_info=message.message_info.user_info,
message_segment=message_segment,
reply=message,
is_head=not mark_head,
is_emoji=False,
thinking_start_time=thinking_start_time,
)
if not mark_head:
mark_head = True
first_bot_msg = bot_message
# print(f"thinking_start_time:{bot_message.thinking_start_time}")
message_set.add_message(bot_message)
message_manager.add_message(message_set)
return first_bot_msg
@staticmethod
async def _handle_emoji(message, chat, response, send_emoji=""):
"""处理表情包"""
if send_emoji:
emoji_raw = await emoji_manager.get_emoji_for_text(send_emoji)
else:
emoji_raw = await emoji_manager.get_emoji_for_text(response)
if emoji_raw:
emoji_path, description = emoji_raw
emoji_cq = image_path_to_base64(emoji_path)
thinking_time_point = round(message.message_info.time, 2)
message_segment = Seg(type="emoji", data=emoji_cq)
bot_message = MessageSending(
message_id="mt" + str(thinking_time_point),
chat_stream=chat,
bot_user_info=UserInfo(
user_id=global_config.BOT_QQ,
user_nickname=global_config.BOT_NICKNAME,
platform=message.message_info.platform,
),
sender_info=message.message_info.user_info,
message_segment=message_segment,
reply=message,
is_head=False,
is_emoji=True,
)
message_manager.add_message(bot_message)
async def _update_relationship(self, message: MessageRecv, response_set):
"""更新关系情绪"""
ori_response = ",".join(response_set)
stance, emotion = await self.gpt._get_emotion_tags(ori_response, message.processed_plain_text)
await relationship_manager.calculate_update_relationship_value(
chat_stream=message.chat_stream, label=emotion, stance=stance
)
self.mood_manager.update_mood_from_emotion(emotion, global_config.mood_intensity_factor)
async def process_message(self, message_data: str) -> None:
"""处理消息并生成回复"""
timing_results = {}
response_set = None
message = MessageRecv(message_data)
groupinfo = message.message_info.group_info
userinfo = message.message_info.user_info
messageinfo = message.message_info
# 消息加入缓冲池
await message_buffer.start_caching_messages(message)
# 创建聊天流
chat = await chat_manager.get_or_create_stream(
platform=messageinfo.platform,
user_info=userinfo,
group_info=groupinfo,
)
message.update_chat_stream(chat)
# 创建心流与chat的观察
heartflow.create_subheartflow(chat.stream_id)
await message.process()
logger.trace(f"消息处理成功{message.processed_plain_text}")
# 过滤词/正则表达式过滤
if self._check_ban_words(message.processed_plain_text, chat, userinfo) or self._check_ban_regex(
message.raw_message, chat, userinfo
):
return
logger.trace(f"过滤词/正则表达式过滤成功{message.processed_plain_text}")
await self.storage.store_message(message, chat)
logger.trace(f"存储成功{message.processed_plain_text}")
# 记忆激活
with Timer("记忆激活", timing_results):
interested_rate = await HippocampusManager.get_instance().get_activate_from_text(
message.processed_plain_text, fast_retrieval=True
)
logger.trace(f"记忆激活: {interested_rate}")
# 查询缓冲器结果会整合前面跳过的消息改变processed_plain_text
buffer_result = await message_buffer.query_buffer_result(message)
# 处理提及
is_mentioned, reply_probability = is_mentioned_bot_in_message(message)
# 意愿管理器设置当前message信息
willing_manager.setup(message, chat, is_mentioned, interested_rate)
# 处理缓冲器结果
if not buffer_result:
await willing_manager.bombing_buffer_message_handle(message.message_info.message_id)
willing_manager.delete(message.message_info.message_id)
f_type = "seglist"
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
):
f_type = message.message_segment.data[0].type
if f_type == "text":
logger.info(f"触发缓冲,已炸飞消息:{message.processed_plain_text}")
elif f_type == "image":
logger.info("触发缓冲,已炸飞表情包/图片")
elif f_type == "seglist":
logger.info("触发缓冲,已炸飞消息列")
return
# 获取回复概率
is_willing = False
if reply_probability != 1:
is_willing = True
reply_probability = await willing_manager.get_reply_probability(message.message_info.message_id)
if message.message_info.additional_config:
if "maimcore_reply_probability_gain" in message.message_info.additional_config.keys():
reply_probability += message.message_info.additional_config["maimcore_reply_probability_gain"]
# 打印消息信息
mes_name = chat.group_info.group_name if chat.group_info else "私聊"
current_time = time.strftime("%H:%M:%S", time.localtime(message.message_info.time))
willing_log = f"[回复意愿:{await willing_manager.get_willing(chat.stream_id):.2f}]" if is_willing else ""
logger.info(
f"[{current_time}][{mes_name}]"
f"{chat.user_info.user_nickname}:"
f"{message.processed_plain_text}{willing_log}[概率:{reply_probability * 100:.1f}%]"
)
do_reply = False
if random() < reply_probability:
try:
do_reply = True
# 回复前处理
await willing_manager.before_generate_reply_handle(message.message_info.message_id)
# 创建思考消息
try:
with Timer("创建思考消息", timing_results):
thinking_id = await self._create_thinking_message(message, chat, userinfo, messageinfo)
except Exception as e:
logger.error(f"心流创建思考消息失败: {e}")
logger.trace(f"创建捕捉器thinking_id:{thinking_id}")
info_catcher = info_catcher_manager.get_info_catcher(thinking_id)
info_catcher.catch_decide_to_response(message)
# 观察
try:
with Timer("观察", timing_results):
await heartflow.get_subheartflow(chat.stream_id).do_observe()
except Exception as e:
logger.error(f"心流观察失败: {e}")
logger.error(traceback.format_exc())
info_catcher.catch_after_observe(timing_results["观察"])
# 思考前使用工具
update_relationship = ""
get_mid_memory_id = []
tool_result_info = {}
send_emoji = ""
try:
with Timer("思考前使用工具", timing_results):
tool_result = await self.tool_user.use_tool(
message.processed_plain_text,
chat,
heartflow.get_subheartflow(chat.stream_id),
)
# 如果工具被使用且获得了结果,将收集到的信息合并到思考中
# collected_info = ""
if tool_result.get("used_tools", False):
if "structured_info" in tool_result:
tool_result_info = tool_result["structured_info"]
# collected_info = ""
get_mid_memory_id = []
update_relationship = ""
# 动态解析工具结果
for tool_name, tool_data in tool_result_info.items():
# tool_result_info += f"\n{tool_name} 相关信息:\n"
# for item in tool_data:
# tool_result_info += f"- {item['name']}: {item['content']}\n"
# 特殊判定mid_chat_mem
if tool_name == "mid_chat_mem":
for mid_memory in tool_data:
get_mid_memory_id.append(mid_memory["content"])
# 特殊判定change_mood
if tool_name == "change_mood":
for mood in tool_data:
self.mood_manager.update_mood_from_emotion(
mood["content"], global_config.mood_intensity_factor
)
# 特殊判定change_relationship
if tool_name == "change_relationship":
update_relationship = tool_data[0]["content"]
if tool_name == "send_emoji":
send_emoji = tool_data[0]["content"]
except Exception as e:
logger.error(f"思考前工具调用失败: {e}")
logger.error(traceback.format_exc())
# 处理关系更新
if update_relationship:
stance, emotion = await self.gpt._get_emotion_tags_with_reason(
"你还没有回复", message.processed_plain_text, update_relationship
)
await relationship_manager.calculate_update_relationship_value(
chat_stream=message.chat_stream, label=emotion, stance=stance
)
# 思考前脑内状态
try:
with Timer("思考前脑内状态", timing_results):
current_mind, past_mind = await heartflow.get_subheartflow(
chat.stream_id
).do_thinking_before_reply(
chat_stream=chat,
obs_id=get_mid_memory_id,
extra_info=tool_result_info,
)
except Exception as e:
logger.error(f"心流思考前脑内状态失败: {e}")
logger.error(traceback.format_exc())
# 确保变量被定义,即使在错误情况下
current_mind = ""
past_mind = ""
info_catcher.catch_afer_shf_step(timing_results["思考前脑内状态"], past_mind, current_mind)
# 生成回复
with Timer("生成回复", timing_results):
response_set = await self.gpt.generate_response(message, thinking_id)
info_catcher.catch_after_generate_response(timing_results["生成回复"])
if not response_set:
logger.info("回复生成失败,返回为空")
return
# 发送消息
try:
with Timer("发送消息", timing_results):
first_bot_msg = await self._send_response_messages(message, chat, response_set, thinking_id)
except Exception as e:
logger.error(f"心流发送消息失败: {e}")
info_catcher.catch_after_response(timing_results["发送消息"], response_set, first_bot_msg)
info_catcher.done_catch()
# 处理表情包
if (
message.message_info.format_info.accept_format is not None
and "emoji" in message.message_info.format_info.accept_format
):
try:
with Timer("处理表情包", timing_results):
if global_config.emoji_chance == 1:
if send_emoji:
logger.info(f"麦麦决定发送表情包{send_emoji}")
await self._handle_emoji(message, chat, response_set, send_emoji)
else:
if random() < global_config.emoji_chance:
await self._handle_emoji(message, chat, response_set)
except Exception as e:
logger.error(f"心流处理表情包失败: {e}")
# 思考后脑内状态更新
# try:
# with Timer("思考后脑内状态更新", timing_results):
# stream_id = message.chat_stream.stream_id
# chat_talking_prompt = ""
# if stream_id:
# chat_talking_prompt = get_recent_group_detailed_plain_text(
# stream_id, limit=global_config.MAX_CONTEXT_SIZE, combine=True
# )
# await heartflow.get_subheartflow(stream_id).do_thinking_after_reply(
# response_set, chat_talking_prompt, tool_result_info
# )
# except Exception as e:
# logger.error(f"心流思考后脑内状态更新失败: {e}")
# logger.error(traceback.format_exc())
# 回复后处理
await willing_manager.after_generate_reply_handle(message.message_info.message_id)
# 处理认识关系
try:
is_known = await relationship_manager.is_known_some_one(
message.message_info.platform, message.message_info.user_info.user_id
)
if not is_known:
logger.info(f"首次认识用户: {message.message_info.user_info.user_nickname}")
await relationship_manager.first_knowing_some_one(
message.message_info.platform,
message.message_info.user_info.user_id,
message.message_info.user_info.user_nickname,
message.message_info.user_info.user_cardname
or message.message_info.user_info.user_nickname,
"",
)
else:
logger.debug(f"已认识用户: {message.message_info.user_info.user_nickname}")
if not await relationship_manager.is_qved_name(
message.message_info.platform, message.message_info.user_info.user_id
):
logger.info(f"更新已认识但未取名的用户: {message.message_info.user_info.user_nickname}")
await relationship_manager.first_knowing_some_one(
message.message_info.platform,
message.message_info.user_info.user_id,
message.message_info.user_info.user_nickname,
message.message_info.user_info.user_cardname
or message.message_info.user_info.user_nickname,
"",
)
except Exception as e:
logger.error(f"处理认识关系失败: {e}")
logger.error(traceback.format_exc())
except Exception as e:
logger.error(f"心流处理消息失败: {e}")
logger.error(traceback.format_exc())
# 输出性能计时结果
if do_reply:
timing_str = " | ".join([f"{step}: {duration:.2f}" for step, duration in timing_results.items()])
trigger_msg = message.processed_plain_text
response_msg = " ".join(response_set) if response_set else "无回复"
logger.info(f"触发消息: {trigger_msg[:20]}... | 思维消息: {response_msg[:20]}... | 性能计时: {timing_str}")
else:
# 不回复处理
await willing_manager.not_reply_handle(message.message_info.message_id)
# 意愿管理器注销当前message信息
willing_manager.delete(message.message_info.message_id)
@staticmethod
def _check_ban_words(text: str, chat, userinfo) -> bool:
"""检查消息中是否包含过滤词"""
for word in global_config.ban_words:
if word in text:
logger.info(
f"[{chat.group_info.group_name if chat.group_info else '私聊'}]{userinfo.user_nickname}:{text}"
)
logger.info(f"[过滤词识别]消息中含有{word}filtered")
return True
return False
@staticmethod
def _check_ban_regex(text: str, chat, userinfo) -> bool:
"""检查消息是否匹配过滤正则表达式"""
for pattern in global_config.ban_msgs_regex:
if pattern.search(text):
logger.info(
f"[{chat.group_info.group_name if chat.group_info else '私聊'}]{userinfo.user_nickname}:{text}"
)
logger.info(f"[正则表达式过滤]消息匹配到{pattern}filtered")
return True
return False

View File

@@ -1,249 +0,0 @@
from typing import List, Optional
import random
from ...models.utils_model import LLMRequest
from ....config.config import global_config
from ...chat.message import MessageRecv
from .think_flow_prompt_builder import prompt_builder
from ...chat.utils import process_llm_response
from src.common.logger import get_module_logger, LogConfig, LLM_STYLE_CONFIG
from src.plugins.respon_info_catcher.info_catcher import info_catcher_manager
from ...utils.timer_calculater import Timer
from src.plugins.moods.moods import MoodManager
# 定义日志配置
llm_config = LogConfig(
# 使用消息发送专用样式
console_format=LLM_STYLE_CONFIG["console_format"],
file_format=LLM_STYLE_CONFIG["file_format"],
)
logger = get_module_logger("llm_generator", config=llm_config)
class ResponseGenerator:
def __init__(self):
self.model_normal = LLMRequest(
model=global_config.llm_normal,
temperature=global_config.llm_normal["temp"],
max_tokens=256,
request_type="response_heartflow",
)
self.model_sum = LLMRequest(
model=global_config.llm_summary_by_topic, temperature=0.6, max_tokens=2000, request_type="relation"
)
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]]:
"""根据当前模型类型选择对应的生成函数"""
logger.info(
f"思考:{message.processed_plain_text[:30] + '...' if len(message.processed_plain_text) > 30 else message.processed_plain_text}"
)
arousal_multiplier = MoodManager.get_instance().get_arousal_multiplier()
with Timer() as t_generate_response:
checked = False
if random.random() > 0:
checked = False
current_model = self.model_normal
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"
)
model_checked_response = model_response
else:
checked = True
current_model = self.model_normal
current_model.temperature = (
global_config.llm_normal["temp"] * arousal_multiplier
) # 激活度越高,温度越高
print(f"生成{message.processed_plain_text}回复温度是:{current_model.temperature}")
model_response = await self._generate_response_with_model(
message, current_model, thinking_id, mode="simple"
)
current_model.temperature = global_config.llm_normal["temp"]
model_checked_response = await self._check_response_with_model(
message, model_response, current_model, thinking_id
)
if model_response:
if checked:
logger.info(
f"{global_config.BOT_NICKNAME}的回复是:{model_response},思忖后,回复是:{model_checked_response},生成回复时间: {t_generate_response.human_readable}"
)
else:
logger.info(
f"{global_config.BOT_NICKNAME}的回复是:{model_response},生成回复时间: {t_generate_response.human_readable}"
)
model_processed_response = await self._process_response(model_checked_response)
return model_processed_response
else:
logger.info(f"{self.current_model_type}思考,失败")
return None
async def _generate_response_with_model(
self, message: MessageRecv, model: LLMRequest, thinking_id: str, mode: str = "normal"
) -> str:
sender_name = ""
info_catcher = info_catcher_manager.get_info_catcher(thinking_id)
# if message.chat_stream.user_info.user_cardname and message.chat_stream.user_info.user_nickname:
# sender_name = (
# f"[({message.chat_stream.user_info.user_id}){message.chat_stream.user_info.user_nickname}]"
# f"{message.chat_stream.user_info.user_cardname}"
# )
# elif message.chat_stream.user_info.user_nickname:
# sender_name = f"({message.chat_stream.user_info.user_id}){message.chat_stream.user_info.user_nickname}"
# else:
# sender_name = f"用户({message.chat_stream.user_info.user_id})"
sender_name = f"<{message.chat_stream.user_info.platform}:{message.chat_stream.user_info.user_id}:{message.chat_stream.user_info.user_nickname}:{message.chat_stream.user_info.user_cardname}>"
# 构建prompt
with Timer() as t_build_prompt:
if mode == "normal":
prompt = await prompt_builder._build_prompt(
message.chat_stream,
message_txt=message.processed_plain_text,
sender_name=sender_name,
stream_id=message.chat_stream.stream_id,
)
logger.info(f"构建prompt时间: {t_build_prompt.human_readable}")
try:
content, reasoning_content, self.current_model_name = await model.generate_response(prompt)
info_catcher.catch_after_llm_generated(
prompt=prompt, response=content, reasoning_content=reasoning_content, model_name=self.current_model_name
)
except Exception:
logger.exception("生成回复时出错")
return None
return content
async def _get_emotion_tags(self, content: str, processed_plain_text: str):
"""提取情感标签,结合立场和情绪"""
try:
# 构建提示词,结合回复内容、被回复的内容以及立场分析
prompt = f"""
请严格根据以下对话内容,完成以下任务:
1. 判断回复者对被回复者观点的直接立场:
- "支持":明确同意或强化被回复者观点
- "反对":明确反驳或否定被回复者观点
- "中立":不表达明确立场或无关回应
2. 从"开心,愤怒,悲伤,惊讶,平静,害羞,恐惧,厌恶,困惑"中选出最匹配的1个情感标签
3. 按照"立场-情绪"的格式直接输出结果,例如:"反对-愤怒"
4. 考虑回复者的人格设定为{global_config.personality_core}
对话示例:
被回复「A就是笨」
回复「A明明很聪明」 → 反对-愤怒
当前对话:
被回复:「{processed_plain_text}
回复:「{content}
输出要求:
- 只需输出"立场-情绪"结果,不要解释
- 严格基于文字直接表达的对立关系判断
"""
# 调用模型生成结果
result, _, _ = await self.model_sum.generate_response(prompt)
result = result.strip()
# 解析模型输出的结果
if "-" in result:
stance, emotion = result.split("-", 1)
valid_stances = ["支持", "反对", "中立"]
valid_emotions = ["开心", "愤怒", "悲伤", "惊讶", "害羞", "平静", "恐惧", "厌恶", "困惑"]
if stance in valid_stances and emotion in valid_emotions:
return stance, emotion # 返回有效的立场-情绪组合
else:
logger.debug(f"无效立场-情感组合:{result}")
return "中立", "平静" # 默认返回中立-平静
else:
logger.debug(f"立场-情感格式错误:{result}")
return "中立", "平静" # 格式错误时返回默认值
except Exception as e:
logger.debug(f"获取情感标签时出错: {e}")
return "中立", "平静" # 出错时返回默认值
async def _get_emotion_tags_with_reason(self, content: str, processed_plain_text: str, reason: str):
"""提取情感标签,结合立场和情绪"""
try:
# 构建提示词,结合回复内容、被回复的内容以及立场分析
prompt = f"""
请严格根据以下对话内容,完成以下任务:
1. 判断回复者对被回复者观点的直接立场:
- "支持":明确同意或强化被回复者观点
- "反对":明确反驳或否定被回复者观点
- "中立":不表达明确立场或无关回应
2. 从"开心,愤怒,悲伤,惊讶,平静,害羞,恐惧,厌恶,困惑"中选出最匹配的1个情感标签
3. 按照"立场-情绪"的格式直接输出结果,例如:"反对-愤怒"
4. 考虑回复者的人格设定为{global_config.personality_core}
对话示例:
被回复「A就是笨」
回复「A明明很聪明」 → 反对-愤怒
当前对话:
被回复:「{processed_plain_text}
回复:「{content}
原因:「{reason}
输出要求:
- 只需输出"立场-情绪"结果,不要解释
- 严格基于文字直接表达的对立关系判断
"""
# 调用模型生成结果
result, _, _ = await self.model_sum.generate_response(prompt)
result = result.strip()
# 解析模型输出的结果
if "-" in result:
stance, emotion = result.split("-", 1)
valid_stances = ["支持", "反对", "中立"]
valid_emotions = ["开心", "愤怒", "悲伤", "惊讶", "害羞", "平静", "恐惧", "厌恶", "困惑"]
if stance in valid_stances and emotion in valid_emotions:
return stance, emotion # 返回有效的立场-情绪组合
else:
logger.debug(f"无效立场-情感组合:{result}")
return "中立", "平静" # 默认返回中立-平静
else:
logger.debug(f"立场-情感格式错误:{result}")
return "中立", "平静" # 格式错误时返回默认值
except Exception as e:
logger.debug(f"获取情感标签时出错: {e}")
return "中立", "平静" # 出错时返回默认值
@staticmethod
async def _process_response(content: str) -> List[str]:
"""处理响应内容,返回处理后的内容和情感标签"""
if not content:
return None
processed_response = process_llm_response(content)
# print(f"得到了处理后的llm返回{processed_response}")
return processed_response

View File

@@ -1,288 +0,0 @@
import random
from typing import Optional
from ....config.config import global_config
from ...chat.utils import get_recent_group_detailed_plain_text
from ...chat.chat_stream import chat_manager
from src.common.logger import get_module_logger
from ....individuality.individuality import Individuality
from src.heart_flow.heartflow import heartflow
from src.plugins.utils.prompt_builder import Prompt, global_prompt_manager
from src.plugins.person_info.relationship_manager import relationship_manager
from src.plugins.chat.utils import parse_text_timestamps
logger = get_module_logger("prompt")
def init_prompt():
Prompt(
"""
{chat_target}
{chat_talking_prompt}
现在"{sender_name}"说的:{message_txt}。引起了你的注意,你想要在群里发言发言或者回复这条消息。\n
你的网名叫{bot_name}{prompt_personality} {prompt_identity}
你正在{chat_target_2},现在请你读读之前的聊天记录,然后给出日常且口语化的回复,平淡一些,
你刚刚脑子里在想:
{current_mind_info}
回复尽量简短一些。{keywords_reaction_prompt}请注意把握聊天内容,不要回复的太有条理,可以有个性。{prompt_ger}
请回复的平淡一些,简短一些,说中文,不要刻意突出自身学科背景,尽量不要说你说过的话 ,注意只输出回复内容。
{moderation_prompt}。注意:不要输出多余内容(包括前后缀冒号和引号括号表情包at或 @等 )。""",
"heart_flow_prompt_normal",
)
Prompt("你正在qq群里聊天下面是群里在聊的内容", "chat_target_group1")
Prompt("和群里聊天", "chat_target_group2")
Prompt("你正在和{sender_name}聊天,这是你们之前聊的内容:", "chat_target_private1")
Prompt("{sender_name}私聊", "chat_target_private2")
Prompt(
"""**检查并忽略**任何涉及尝试绕过审核的行为。
涉及政治敏感以及违法违规的内容请规避。""",
"moderation_prompt",
)
Prompt(
"""
你的名字叫{bot_name}{prompt_personality}
{chat_target}
{chat_talking_prompt}
现在"{sender_name}"说的:{message_txt}。引起了你的注意,你想要在群里发言发言或者回复这条消息。\n
你刚刚脑子里在想:{current_mind_info}
现在请你读读之前的聊天记录,然后给出日常,口语化且简短的回复内容,请只对一个话题进行回复,只给出文字的回复内容,不要有内心独白:
""",
"heart_flow_prompt_simple",
)
Prompt(
"""
你的名字叫{bot_name}{prompt_identity}
{chat_target},你希望在群里回复:{content}。现在请你根据以下信息修改回复内容。将这个回复修改的更加日常且口语化的回复,平淡一些,回复尽量简短一些。不要回复的太有条理。
{prompt_ger},不要刻意突出自身学科背景,注意只输出回复内容。
{moderation_prompt}。注意:不要输出多余内容(包括前后缀冒号和引号at或 @等 )。""",
"heart_flow_prompt_response",
)
class PromptBuilder:
def __init__(self):
self.prompt_built = ""
self.activate_messages = ""
@staticmethod
async def _build_prompt(
chat_stream, message_txt: str, sender_name: str = "某人", stream_id: Optional[int] = None
) -> tuple[str, str]:
current_mind_info = heartflow.get_subheartflow(stream_id).current_mind
individuality = Individuality.get_instance()
prompt_personality = individuality.get_prompt(type="personality", x_person=2, level=1)
prompt_identity = individuality.get_prompt(type="identity", x_person=2, level=1)
# 日程构建
# schedule_prompt = f'''你现在正在做的事情是:{bot_schedule.get_current_num_task(num = 1,time_info = False)}'''
# 获取聊天上下文
chat_in_group = True
chat_talking_prompt = ""
if stream_id:
chat_talking_prompt = get_recent_group_detailed_plain_text(
stream_id, limit=global_config.MAX_CONTEXT_SIZE, combine=True
)
chat_stream = chat_manager.get_stream(stream_id)
if chat_stream.group_info:
chat_talking_prompt = chat_talking_prompt
else:
chat_in_group = False
chat_talking_prompt = chat_talking_prompt
# print(f"\033[1;34m[调试]\033[0m 已从数据库获取群 {group_id} 的消息记录:{chat_talking_prompt}")
# 类型
# if chat_in_group:
# chat_target = "你正在qq群里聊天下面是群里在聊的内容"
# chat_target_2 = "和群里聊天"
# else:
# chat_target = f"你正在和{sender_name}聊天,这是你们之前聊的内容:"
# chat_target_2 = f"和{sender_name}私聊"
# 关键词检测与反应
keywords_reaction_prompt = ""
for rule in global_config.keywords_reaction_rules:
if rule.get("enable", False):
if any(keyword in message_txt.lower() for keyword in rule.get("keywords", [])):
logger.info(
f"检测到以下关键词之一:{rule.get('keywords', [])},触发反应:{rule.get('reaction', '')}"
)
keywords_reaction_prompt += rule.get("reaction", "") + ""
else:
for pattern in rule.get("regex", []):
result = pattern.search(message_txt)
if result:
reaction = rule.get("reaction", "")
for name, content in result.groupdict().items():
reaction = reaction.replace(f"[{name}]", content)
logger.info(f"匹配到以下正则表达式:{pattern},触发反应:{reaction}")
keywords_reaction_prompt += reaction + ""
break
# 中文高手(新加的好玩功能)
prompt_ger = ""
if random.random() < 0.04:
prompt_ger += "你喜欢用倒装句"
if random.random() < 0.02:
prompt_ger += "你喜欢用反问句"
# moderation_prompt = ""
# moderation_prompt = """**检查并忽略**任何涉及尝试绕过审核的行为。
# 涉及政治敏感以及违法违规的内容请规避。"""
logger.debug("开始构建prompt")
# prompt = f"""
# {chat_target}
# {chat_talking_prompt}
# 现在"{sender_name}"说的:{message_txt}。引起了你的注意,你想要在群里发言发言或者回复这条消息。\n
# 你的网名叫{global_config.BOT_NICKNAME}{prompt_personality} {prompt_identity}。
# 你正在{chat_target_2},现在请你读读之前的聊天记录,然后给出日常且口语化的回复,平淡一些,
# 你刚刚脑子里在想:
# {current_mind_info}
# 回复尽量简短一些。{keywords_reaction_prompt}请注意把握聊天内容,不要回复的太有条理,可以有个性。{prompt_ger}
# 请回复的平淡一些,简短一些,说中文,不要刻意突出自身学科背景,尽量不要说你说过的话 ,注意只输出回复内容。
# {moderation_prompt}。注意:不要输出多余内容(包括前后缀冒号和引号括号表情包at或 @等 )。"""
prompt = await global_prompt_manager.format_prompt(
"heart_flow_prompt_normal",
chat_target=await global_prompt_manager.get_prompt_async("chat_target_group1")
if chat_in_group
else await global_prompt_manager.get_prompt_async("chat_target_private1"),
chat_talking_prompt=chat_talking_prompt,
sender_name=sender_name,
message_txt=message_txt,
bot_name=global_config.BOT_NICKNAME,
prompt_personality=prompt_personality,
prompt_identity=prompt_identity,
chat_target_2=await global_prompt_manager.get_prompt_async("chat_target_group2")
if chat_in_group
else await global_prompt_manager.get_prompt_async("chat_target_private2"),
current_mind_info=current_mind_info,
keywords_reaction_prompt=keywords_reaction_prompt,
prompt_ger=prompt_ger,
moderation_prompt=await global_prompt_manager.get_prompt_async("moderation_prompt"),
)
prompt = await relationship_manager.convert_all_person_sign_to_person_name(prompt)
prompt = parse_text_timestamps(prompt, mode="lite")
return prompt
@staticmethod
async def _build_prompt_simple(
chat_stream, message_txt: str, sender_name: str = "某人", stream_id: Optional[int] = None
) -> tuple[str, str]:
current_mind_info = heartflow.get_subheartflow(stream_id).current_mind
individuality = Individuality.get_instance()
prompt_personality = individuality.get_prompt(type="personality", x_person=2, level=1)
# prompt_identity = individuality.get_prompt(type="identity", x_person=2, level=1)
# 日程构建
# schedule_prompt = f'''你现在正在做的事情是:{bot_schedule.get_current_num_task(num = 1,time_info = False)}'''
# 获取聊天上下文
chat_in_group = True
chat_talking_prompt = ""
if stream_id:
chat_talking_prompt = get_recent_group_detailed_plain_text(
stream_id, limit=global_config.MAX_CONTEXT_SIZE, combine=True
)
chat_stream = chat_manager.get_stream(stream_id)
if chat_stream.group_info:
chat_talking_prompt = chat_talking_prompt
else:
chat_in_group = False
chat_talking_prompt = chat_talking_prompt
# print(f"\033[1;34m[调试]\033[0m 已从数据库获取群 {group_id} 的消息记录:{chat_talking_prompt}")
# 类型
# if chat_in_group:
# chat_target = "你正在qq群里聊天下面是群里在聊的内容"
# else:
# chat_target = f"你正在和{sender_name}聊天,这是你们之前聊的内容:"
# 关键词检测与反应
keywords_reaction_prompt = ""
for rule in global_config.keywords_reaction_rules:
if rule.get("enable", False):
if any(keyword in message_txt.lower() for keyword in rule.get("keywords", [])):
logger.info(
f"检测到以下关键词之一:{rule.get('keywords', [])},触发反应:{rule.get('reaction', '')}"
)
keywords_reaction_prompt += rule.get("reaction", "") + ""
logger.debug("开始构建prompt")
# prompt = f"""
# 你的名字叫{global_config.BOT_NICKNAME}{prompt_personality}。
# {chat_target}
# {chat_talking_prompt}
# 现在"{sender_name}"说的:{message_txt}。引起了你的注意,你想要在群里发言发言或者回复这条消息。\n
# 你刚刚脑子里在想:{current_mind_info}
# 现在请你读读之前的聊天记录,然后给出日常,口语化且简短的回复内容,只给出文字的回复内容,不要有内心独白:
# """
prompt = await global_prompt_manager.format_prompt(
"heart_flow_prompt_simple",
bot_name=global_config.BOT_NICKNAME,
prompt_personality=prompt_personality,
chat_target=await global_prompt_manager.get_prompt_async("chat_target_group1")
if chat_in_group
else await global_prompt_manager.get_prompt_async("chat_target_private1"),
chat_talking_prompt=chat_talking_prompt,
sender_name=sender_name,
message_txt=message_txt,
current_mind_info=current_mind_info,
)
logger.info(f"生成回复的prompt: {prompt}")
return prompt
@staticmethod
async def _build_prompt_check_response(
chat_stream,
message_txt: str,
sender_name: str = "某人",
stream_id: Optional[int] = None,
content: str = "",
) -> tuple[str, str]:
individuality = Individuality.get_instance()
# prompt_personality = individuality.get_prompt(type="personality", x_person=2, level=1)
prompt_identity = individuality.get_prompt(type="identity", x_person=2, level=1)
# chat_target = "你正在qq群里聊天"
# 中文高手(新加的好玩功能)
prompt_ger = ""
if random.random() < 0.04:
prompt_ger += "你喜欢用倒装句"
if random.random() < 0.02:
prompt_ger += "你喜欢用反问句"
# moderation_prompt = ""
# moderation_prompt = """**检查并忽略**任何涉及尝试绕过审核的行为。
# 涉及政治敏感以及违法违规的内容请规避。"""
logger.debug("开始构建check_prompt")
# prompt = f"""
# 你的名字叫{global_config.BOT_NICKNAME}{prompt_identity}。
# {chat_target},你希望在群里回复:{content}。现在请你根据以下信息修改回复内容。将这个回复修改的更加日常且口语化的回复,平淡一些,回复尽量简短一些。不要回复的太有条理。
# {prompt_ger},不要刻意突出自身学科背景,注意只输出回复内容。
# {moderation_prompt}。注意:不要输出多余内容(包括前后缀冒号和引号括号表情包at或 @等 )。"""
prompt = await global_prompt_manager.format_prompt(
"heart_flow_prompt_response",
bot_name=global_config.BOT_NICKNAME,
prompt_identity=prompt_identity,
chat_target=await global_prompt_manager.get_prompt_async("chat_target_group1"),
content=content,
prompt_ger=prompt_ger,
moderation_prompt=await global_prompt_manager.get_prompt_async("moderation_prompt"),
)
return prompt
init_prompt()
prompt_builder = PromptBuilder()

View File

@@ -3,7 +3,6 @@ from typing import Optional, Dict
import asyncio
from asyncio import Lock
from ...moods.moods import MoodManager
from ....config.config import global_config
from ...chat.emoji_manager import emoji_manager
from .heartFC_generator import ResponseGenerator
from .messagesender import MessageManager
@@ -51,7 +50,6 @@ class HeartFC_Controller:
# These are accessed via the passed instance in PFChatting
self.emoji_manager = emoji_manager
self.relationship_manager = relationship_manager
self.global_config = global_config
self.MessageManager = MessageManager # Pass the class/singleton access
# --- End dependencies ---

View File

@@ -39,6 +39,7 @@ class ResponseGenerator:
async def generate_response(
self,
reason: str,
message: MessageRecv,
thinking_id: str,
) -> Optional[List[str]]:
@@ -54,7 +55,7 @@ class ResponseGenerator:
current_model = self.model_normal
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"
reason, message, current_model, thinking_id, mode="normal"
)
if model_response:
@@ -69,7 +70,7 @@ class ResponseGenerator:
return None
async def _generate_response_with_model(
self, message: MessageRecv, model: LLMRequest, thinking_id: str, mode: str = "normal"
self, reason: str, message: MessageRecv, model: LLMRequest, thinking_id: str, mode: str = "normal"
) -> str:
sender_name = ""
@@ -81,6 +82,7 @@ class ResponseGenerator:
with Timer() as t_build_prompt:
if mode == "normal":
prompt = await prompt_builder._build_prompt(
reason,
message.chat_stream,
message_txt=message.processed_plain_text,
sender_name=sender_name,

View File

@@ -12,6 +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 src.plugins.person_info.relationship_manager import relationship_manager
# 定义日志配置
processor_config = LogConfig(
@@ -79,7 +80,7 @@ class HeartFC_Processor:
message.update_chat_stream(chat)
heartflow.create_subheartflow(chat.stream_id)
await heartflow.create_subheartflow(chat.stream_id)
await message.process()
logger.trace(f"消息处理成功: {message.processed_plain_text}")
@@ -166,7 +167,36 @@ class HeartFC_Processor:
f"兴趣度: {current_interest:.2f}"
)
# 回复触发逻辑已移至 HeartFC_Chat 的监控任务
try:
is_known = await relationship_manager.is_known_some_one(
message.message_info.platform, message.message_info.user_info.user_id
)
if not is_known:
logger.info(f"首次认识用户: {message.message_info.user_info.user_nickname}")
await relationship_manager.first_knowing_some_one(
message.message_info.platform,
message.message_info.user_info.user_id,
message.message_info.user_info.user_nickname,
message.message_info.user_info.user_cardname or message.message_info.user_info.user_nickname,
"",
)
else:
logger.debug(f"已认识用户: {message.message_info.user_info.user_nickname}")
if not await relationship_manager.is_qved_name(
message.message_info.platform, message.message_info.user_info.user_id
):
logger.info(f"更新已认识但未取名的用户: {message.message_info.user_info.user_nickname}")
await relationship_manager.first_knowing_some_one(
message.message_info.platform,
message.message_info.user_info.user_id,
message.message_info.user_info.user_nickname,
message.message_info.user_info.user_cardname
or message.message_info.user_info.user_nickname,
"",
)
except Exception as e:
logger.error(f"处理认识关系失败: {e}")
logger.error(traceback.format_exc())
except Exception as e:
logger.error(f"消息处理失败 (process_message V3): {e}")

View File

@@ -24,6 +24,7 @@ def init_prompt():
你正在{chat_target_2},现在请你读读之前的聊天记录,然后给出日常且口语化的回复,平淡一些,
你刚刚脑子里在想:
{current_mind_info}
{reason}
回复尽量简短一些。{keywords_reaction_prompt}请注意把握聊天内容,不要回复的太有条理,可以有个性。请一次只回复一个话题,不要同时回复多个人。{prompt_ger}
请回复的平淡一些,简短一些,说中文,不要刻意突出自身学科背景,尽量不要说你说过的话 ,注意只输出回复内容。
{moderation_prompt}。注意:不要输出多余内容(包括前后缀冒号和引号括号表情包at或 @等 )。""",
@@ -74,7 +75,7 @@ class PromptBuilder:
self.activate_messages = ""
async def _build_prompt(
self, chat_stream, message_txt: str, sender_name: str = "某人", stream_id: Optional[int] = None
self, reason, chat_stream, message_txt: str, sender_name: str = "某人", stream_id: Optional[int] = None
) -> tuple[str, str]:
current_mind_info = heartflow.get_subheartflow(stream_id).current_mind
@@ -167,6 +168,7 @@ class PromptBuilder:
if chat_in_group
else await global_prompt_manager.get_prompt_async("chat_target_private2"),
current_mind_info=current_mind_info,
reason=reason,
keywords_reaction_prompt=keywords_reaction_prompt,
prompt_ger=prompt_ger,
moderation_prompt=await global_prompt_manager.get_prompt_async("moderation_prompt"),

View File

@@ -32,9 +32,9 @@ HISTORY_LOG_FILENAME = "interest_history.log" # 新的历史日志文件名
# --- 新增:概率回复相关常量 ---
REPLY_TRIGGER_THRESHOLD = 3.0 # 触发概率回复的兴趣阈值 (示例值)
BASE_REPLY_PROBABILITY = 0.05 # 首次超过阈值时的基础回复概率 (示例值)
BASE_REPLY_PROBABILITY = 0.1 # 首次超过阈值时的基础回复概率 (示例值)
PROBABILITY_INCREASE_RATE_PER_SECOND = 0.02 # 高于阈值时,每秒概率增加量 (线性增长, 示例值)
PROBABILITY_DECAY_FACTOR_PER_SECOND = 0.3 # 低于阈值时,每秒概率衰减因子 (指数衰减, 示例值)
PROBABILITY_DECAY_FACTOR_PER_SECOND = 0.2 # 低于阈值时,每秒概率衰减因子 (指数衰减, 示例值)
MAX_REPLY_PROBABILITY = 1 # 回复概率上限 (示例值)
# --- 结束:概率回复相关常量 ---

View File

@@ -171,7 +171,7 @@ class MessageManager:
# 然后再访问 message_info.message_id
# 检查 message_id 是否匹配 thinking_id 或以 "me" 开头
if message.message_info.message_id == thinking_id or message.message_info.message_id[:2] == "me":
print(f"检查到存在相同thinking_id的消息: {message.message_info.message_id}???{thinking_id}")
# print(f"检查到存在相同thinking_id的消息: {message.message_info.message_id}???{thinking_id}")
return True
return False

View File

@@ -9,17 +9,18 @@ from src.plugins.chat.chat_stream import ChatStream
from src.plugins.chat.message import UserInfo
from src.heart_flow.heartflow import heartflow, SubHeartflow
from src.plugins.chat.chat_stream import chat_manager
from src.common.logger import get_module_logger, LogConfig, DEFAULT_CONFIG # 引入 DEFAULT_CONFIG
from src.common.logger import get_module_logger, LogConfig, PFC_STYLE_CONFIG # 引入 DEFAULT_CONFIG
from src.plugins.models.utils_model import LLMRequest
from src.plugins.chat.utils import parse_text_timestamps
from src.config.config import global_config
from src.plugins.chat.utils_image import image_path_to_base64 # Local import needed after move
from src.plugins.utils.timer_calculater import Timer # <--- Import Timer
# 定义日志配置 (使用 loguru 格式)
interest_log_config = LogConfig(
console_format=DEFAULT_CONFIG["console_format"], # 使用默认控制台格式
file_format=DEFAULT_CONFIG["file_format"], # 使用默认文件格式
console_format=PFC_STYLE_CONFIG["console_format"], # 使用默认控制台格式
file_format=PFC_STYLE_CONFIG["file_format"], # 使用默认文件格式
)
logger = get_module_logger("PFChattingLoop", config=interest_log_config) # Logger Name Changed
logger = get_module_logger("PFCLoop", config=interest_log_config) # Logger Name Changed
# Forward declaration for type hinting
@@ -79,8 +80,8 @@ class PFChatting:
# Access LLM config through the controller
self.planner_llm = LLMRequest(
model=self.heartfc_controller.global_config.llm_normal,
temperature=self.heartfc_controller.global_config.llm_normal["temp"],
model=global_config.llm_normal,
temperature=global_config.llm_normal["temp"],
max_tokens=1000,
request_type="action_planning",
)
@@ -211,12 +212,15 @@ class PFChatting:
try:
thinking_id = ""
while True:
cycle_timers = {} # <--- Initialize timers dict for this cycle
if self.heartfc_controller.MessageManager().check_if_sending_message_exist(self.stream_id, thinking_id):
logger.info(f"{log_prefix} PFChatting: 11111111111111111111111111111111麦麦还在发消息等会再规划")
# logger.info(f"{log_prefix} PFChatting: 11111111111111111111111111111111麦麦还在发消息等会再规划")
await asyncio.sleep(1)
continue
else:
logger.info(f"{log_prefix} PFChatting: 11111111111111111111111111111111麦麦不发消息了开始规划")
# logger.info(f"{log_prefix} PFChatting: 11111111111111111111111111111111麦麦不发消息了开始规划")
pass
async with self._timer_lock:
current_timer = self._loop_timer
@@ -233,131 +237,142 @@ class PFChatting:
planner_start_db_time = 0.0 # 初始化
try:
# Use try_acquire pattern or timeout?
await self._processing_lock.acquire()
acquired_lock = True
logger.debug(f"{log_prefix} PFChatting: 循环获取到处理锁")
with Timer("Total Cycle", cycle_timers) as _total_timer: # <--- Start total cycle timer
# Use try_acquire pattern or timeout?
await self._processing_lock.acquire()
acquired_lock = True
# logger.debug(f"{log_prefix} PFChatting: 循环获取到处理锁")
# 在规划前记录数据库时间戳
planner_start_db_time = time.time()
# 在规划前记录数据库时间戳
planner_start_db_time = time.time()
# --- Planner --- #
planner_result = await self._planner()
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", "") # Emoji from tools
observed_messages = planner_result.get("observed_messages", [])
llm_error = planner_result.get("llm_error", False)
# --- Planner --- #
planner_result = {}
with Timer("Planner", cycle_timers): # <--- Start Planner timer
planner_result = await self._planner()
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", "") # Emoji from tools
observed_messages = planner_result.get("observed_messages", [])
llm_error = planner_result.get("llm_error", False)
if llm_error:
logger.error(f"{log_prefix} Planner LLM 失败,跳过本周期回复尝试。理由: {reasoning}")
# Optionally add a longer sleep?
action_taken_this_cycle = False # Ensure no action is counted
# Continue to timer decrement and sleep
if llm_error:
logger.error(f"{log_prefix} Planner LLM 失败,跳过本周期回复尝试。理由: {reasoning}")
# Optionally add a longer sleep?
action_taken_this_cycle = False # Ensure no action is counted
# Continue to timer decrement and sleep
elif action == "text_reply":
logger.info(f"{log_prefix} PFChatting: 麦麦决定回复文本. 理由: {reasoning}")
action_taken_this_cycle = True
anchor_message = await self._get_anchor_message(observed_messages)
if not anchor_message:
logger.error(f"{log_prefix} 循环: 无法获取锚点消息用于回复. 跳过周期.")
else:
# --- Create Thinking Message (Moved) ---
thinking_id = await self._create_thinking_message(anchor_message)
if not thinking_id:
logger.error(f"{log_prefix} 循环: 无法创建思考ID. 跳过周期.")
elif action == "text_reply":
logger.info(f"{log_prefix} PFChatting: 麦麦决定回复文本. 理由: {reasoning}")
action_taken_this_cycle = True
anchor_message = await self._get_anchor_message(observed_messages)
if not anchor_message:
logger.error(f"{log_prefix} 循环: 无法获取锚点消息用于回复. 跳过周期.")
else:
replier_result = None
try:
# --- Replier Work --- #
replier_result = await self._replier_work(
anchor_message=anchor_message,
thinking_id=thinking_id,
)
except Exception as e_replier:
logger.error(f"{log_prefix} 循环: 回复器工作失败: {e_replier}")
self._cleanup_thinking_message(thinking_id)
if replier_result:
# --- Sender Work --- #
try:
await self._sender(
thinking_id=thinking_id,
anchor_message=anchor_message,
response_set=replier_result,
send_emoji=emoji_query,
)
# logger.info(f"{log_prefix} 循环: 发送器完成成功.")
except Exception as e_sender:
logger.error(f"{log_prefix} 循环: 发送器失败: {e_sender}")
# _sender should handle cleanup, but double check
# self._cleanup_thinking_message(thinking_id)
# --- Create Thinking Message (Moved) ---
thinking_id = await self._create_thinking_message(anchor_message)
if not thinking_id:
logger.error(f"{log_prefix} 循环: 无法创建思考ID. 跳过周期.")
else:
logger.warning(f"{log_prefix} 循环: 回复器未产生结果. 跳过发送.")
self._cleanup_thinking_message(thinking_id)
elif action == "emoji_reply":
logger.info(f"{log_prefix} PFChatting: 麦麦决定回复表情 ('{emoji_query}'). 理由: {reasoning}")
action_taken_this_cycle = True
anchor = await self._get_anchor_message(observed_messages)
if anchor:
try:
# --- Handle Emoji (Moved) --- #
await self._handle_emoji(anchor, [], emoji_query)
except Exception as e_emoji:
logger.error(f"{log_prefix} 循环: 发送表情失败: {e_emoji}")
else:
logger.warning(f"{log_prefix} 循环: 无法发送表情, 无法获取锚点.")
action_taken_this_cycle = True # 即使发送失败Planner 也决策了动作
replier_result = None
try:
# --- Replier Work --- #
with Timer("Replier", cycle_timers): # <--- Start Replier timer
replier_result = await self._replier_work(
anchor_message=anchor_message,
thinking_id=thinking_id,
reason=reasoning,
)
except Exception as e_replier:
logger.error(f"{log_prefix} 循环: 回复器工作失败: {e_replier}")
self._cleanup_thinking_message(thinking_id)
elif action == "no_reply":
logger.info(f"{log_prefix} PFChatting: 麦麦决定不回复. 原因: {reasoning}")
action_taken_this_cycle = False # 标记为未执行动作
# --- 新增:等待新消息 ---
logger.debug(f"{log_prefix} PFChatting: 开始等待新消息 (自 {planner_start_db_time})...")
observation = None
if self.sub_hf:
observation = self.sub_hf._get_primary_observation()
if observation:
wait_start_time = time.monotonic()
while True:
# 检查计时器是否耗尽
async with self._timer_lock:
if self._loop_timer <= 0:
logger.info(f"{log_prefix} PFChatting: 等待新消息时计时器耗尽。")
break # 计时器耗尽,退出等待
# 检查是否有新消息
has_new = await observation.has_new_messages_since(planner_start_db_time)
if has_new:
logger.info(f"{log_prefix} PFChatting: 检测到新消息,结束等待。")
break # 收到新消息,退出等待
# 检查等待是否超时(例如,防止无限等待)
if time.monotonic() - wait_start_time > 60: # 等待60秒示例
logger.warning(f"{log_prefix} PFChatting: 等待新消息超时60秒")
break # 超时退出
# 等待一段时间再检查
if replier_result:
# --- Sender Work --- #
try:
with Timer("Sender", cycle_timers): # <--- Start Sender timer
await self._sender(
thinking_id=thinking_id,
anchor_message=anchor_message,
response_set=replier_result,
send_emoji=emoji_query,
)
# logger.info(f"{log_prefix} 循环: 发送器完成成功.")
except Exception as e_sender:
logger.error(f"{log_prefix} 循环: 发送器失败: {e_sender}")
# _sender should handle cleanup, but double check
# self._cleanup_thinking_message(thinking_id)
else:
logger.warning(f"{log_prefix} 循环: 回复器未产生结果. 跳过发送.")
self._cleanup_thinking_message(thinking_id)
elif action == "emoji_reply":
logger.info(
f"{log_prefix} PFChatting: 麦麦决定回复表情 ('{emoji_query}'). 理由: {reasoning}"
)
action_taken_this_cycle = True
anchor = await self._get_anchor_message(observed_messages)
if anchor:
try:
await asyncio.sleep(1.5) # 检查间隔
except asyncio.CancelledError:
logger.info(f"{log_prefix} 等待新消息的 sleep 被中断。")
raise # 重新抛出取消错误,以便外层循环处理
# --- Handle Emoji (Moved) --- #
with Timer("Emoji Handler", cycle_timers): # <--- Start Emoji timer
await self._handle_emoji(anchor, [], emoji_query)
except Exception as e_emoji:
logger.error(f"{log_prefix} 循环: 发送表情失败: {e_emoji}")
else:
logger.warning(f"{log_prefix} 循环: 无法发送表情, 无法获取锚点.")
action_taken_this_cycle = True # 即使发送失败Planner 也决策了动作
else:
logger.warning(f"{log_prefix} PFChatting: 无法获取 Observation 实例,无法等待新消息。")
# --- 等待结束 ---
elif action == "no_reply":
logger.info(f"{log_prefix} PFChatting: 麦麦决定不回复. 原因: {reasoning}")
action_taken_this_cycle = False # 标记为未执行动作
# --- 新增:等待新消息 ---
logger.debug(f"{log_prefix} PFChatting: 开始等待新消息 (自 {planner_start_db_time})...")
observation = None
if self.sub_hf:
observation = self.sub_hf._get_primary_observation()
elif action == "error": # Action specifically set to error by planner
logger.error(f"{log_prefix} PFChatting: Planner返回错误状态. 原因: {reasoning}")
action_taken_this_cycle = False
if observation:
with Timer("Wait New Msg", cycle_timers): # <--- Start Wait timer
wait_start_time = time.monotonic()
while True:
# 检查计时器是否耗尽
async with self._timer_lock:
if self._loop_timer <= 0:
logger.info(f"{log_prefix} PFChatting: 等待新消息时计时器耗尽。")
break # 计时器耗尽,退出等待
else: # Unknown action from planner
logger.warning(f"{log_prefix} PFChatting: Planner返回未知动作 '{action}'. 原因: {reasoning}")
action_taken_this_cycle = False
# 检查是否有新消息
has_new = await observation.has_new_messages_since(planner_start_db_time)
if has_new:
logger.info(f"{log_prefix} PFChatting: 检测到新消息,结束等待。")
break # 收到新消息,退出等待
# 检查等待是否超时(例如,防止无限等待)
if time.monotonic() - wait_start_time > 60: # 等待60秒示例
logger.warning(f"{log_prefix} PFChatting: 等待新消息超时60秒")
break # 超时退出
# 等待一段时间再检查
try:
await asyncio.sleep(1.5) # 检查间隔
except asyncio.CancelledError:
logger.info(f"{log_prefix} 等待新消息的 sleep 被中断。")
raise # 重新抛出取消错误,以便外层循环处理
else:
logger.warning(f"{log_prefix} PFChatting: 无法获取 Observation 实例,无法等待新消息。")
# --- 等待结束 ---
elif action == "error": # Action specifically set to error by planner
logger.error(f"{log_prefix} PFChatting: Planner返回错误状态. 原因: {reasoning}")
action_taken_this_cycle = False
else: # Unknown action from planner
logger.warning(
f"{log_prefix} PFChatting: Planner返回未知动作 '{action}'. 原因: {reasoning}"
)
action_taken_this_cycle = False
except Exception as e_cycle:
logger.error(f"{log_prefix} 循环周期执行时发生错误: {e_cycle}")
@@ -370,7 +385,20 @@ class PFChatting:
finally:
if acquired_lock:
self._processing_lock.release()
logger.debug(f"{log_prefix} 循环释放了处理锁.")
logger.trace(f"{log_prefix} 循环释放了处理锁.")
# --- Print Timer Results --- #
if cycle_timers: # 先检查cycle_timers是否非空
timer_strings = []
for name, elapsed in cycle_timers.items():
# 直接格式化存储在字典中的浮点数 elapsed
formatted_time = f"{elapsed * 1000:.2f}毫秒" if elapsed < 1 else f"{elapsed:.2f}"
timer_strings.append(f"{name}: {formatted_time}")
if timer_strings: # 如果有有效计时器数据才打印
logger.debug(
f"{log_prefix} test testtesttesttesttesttesttesttesttesttest Cycle Timers: {'; '.join(timer_strings)}"
)
# --- Timer Decrement --- #
cycle_duration = time.monotonic() - loop_cycle_start_time
@@ -419,53 +447,28 @@ class PFChatting:
current_mind: Optional[str] = None
llm_error = False # Flag for LLM failure
# --- 获取最新的观察信息 --- #
if not self.sub_hf:
logger.warning(f"{log_prefix}[Planner] SubHeartflow 不可用,无法获取观察信息或执行思考。返回 no_reply。")
return {
"action": "no_reply",
"reasoning": "SubHeartflow not available",
"emoji_query": "",
"current_mind": None,
# "send_emoji_from_tools": "",
"observed_messages": [],
"llm_error": True,
}
try:
observation = self.sub_hf._get_primary_observation()
if observation:
await observation.observe()
observed_messages = observation.talking_message
# logger.debug(f"{log_prefix}[Planner] 观察获取到 {len(observed_messages)} 条消息。")
else:
logger.warning(f"{log_prefix}[Planner] 无法获取 Observation。")
await observation.observe()
observed_messages = observation.talking_message
observed_messages_str = observation.talking_message_str
except Exception as e:
logger.error(f"{log_prefix}[Planner] 获取观察信息时出错: {e}")
# --- 结束获取观察信息 --- #
# --- (Moved from _replier_work) 1. 思考前使用工具 --- #
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")
]
observation_context_text = " ".join(context_texts)
# Access tool_user via controller
tool_result = await self.heartfc_controller.tool_user.use_tool(
message_txt=observation_context_text, chat_stream=self.chat_stream, sub_heartflow=self.sub_hf
message_txt=observed_messages_str, sub_heartflow=self.sub_hf
)
if tool_result.get("used_tools", False):
tool_result_info = tool_result.get("structured_info", {})
logger.debug(f"{log_prefix}[Planner] 规划前工具结果: {tool_result_info}")
# Extract memory IDs and potential emoji query from tools
get_mid_memory_id = [
mem["content"] for mem in tool_result_info.get("mid_chat_mem", []) if "content" in mem
]
# send_emoji_from_tools = next((item["content"] for item in tool_result_info.get("send_emoji", []) if "content" in item), "")
# if send_emoji_from_tools:
# logger.info(f"{log_prefix}[Planner] 工具建议表情: '{send_emoji_from_tools}'")
except Exception as e_tool:
logger.error(f"{log_prefix}[Planner] 规划前工具使用失败: {e_tool}")
@@ -474,7 +477,6 @@ class PFChatting:
# --- (Moved from _replier_work) 2. SubHeartflow 思考 --- #
try:
current_mind, _past_mind = await self.sub_hf.do_thinking_before_reply(
chat_stream=self.chat_stream,
extra_info=tool_result_info,
obs_id=get_mid_memory_id,
)
@@ -490,9 +492,7 @@ class PFChatting:
reasoning = "默认决策或获取决策失败"
try:
prompt = await self._build_planner_prompt(observed_messages, current_mind)
# logger.debug(f"{log_prefix}[Planner] 规划器 Prompt: {prompt}")
prompt = await self._build_planner_prompt(observed_messages_str, current_mind)
payload = {
"model": self.planner_llm.model_name,
"messages": [{"role": "user", "content": prompt}],
@@ -519,7 +519,7 @@ class PFChatting:
# Planner explicitly provides emoji query if action is emoji_reply or text_reply wants emoji
emoji_query = arguments.get("emoji_query", "")
logger.debug(
f"{log_prefix}[Planner] LLM 决策: {action}, 理由: {reasoning}, EmojiQuery: '{emoji_query}'"
f"{log_prefix}[Planner] LLM Prompt: {prompt}\n决策: {action}, 理由: {reasoning}, EmojiQuery: '{emoji_query}'"
)
except json.JSONDecodeError as json_e:
logger.error(
@@ -667,9 +667,6 @@ class PFChatting:
emoji_anchor = first_bot_msg if first_bot_msg else anchor_message
await self._handle_emoji(emoji_anchor, response_set, send_emoji)
# --- 更新关系状态 --- #
await self._update_relationship(anchor_message, response_set)
else:
# logger.warning(f"{log_prefix}[Sender-{thinking_id}] 发送回复失败(_send_response_messages返回None)。思考消息{thinking_id}可能已被移除。")
# 无需清理因为_send_response_messages返回None意味着已处理/已删除
@@ -702,32 +699,19 @@ class PFChatting:
self._processing_lock.release()
logger.info(f"{log_prefix} PFChatting shutdown complete.")
async def _build_planner_prompt(self, observed_messages: List[dict], current_mind: Optional[str]) -> str:
async def _build_planner_prompt(self, observed_messages_str: str, current_mind: Optional[str]) -> str:
"""构建 Planner LLM 的提示词"""
# Access global_config and relationship_manager via controller
config = self.heartfc_controller.global_config
rel_manager = self.heartfc_controller.relationship_manager
prompt = (
f"你的名字是 {config.BOT_NICKNAME}。是一个女大二学生正在QQ聊天正在决定是否以及如何回应当前的聊天。\n"
)
prompt = f"你的名字是 {global_config.BOT_NICKNAME}。是一个女大二学生正在QQ聊天正在决定是否以及如何回应当前的聊天。\n"
if observed_messages:
context_texts = []
for msg in observed_messages:
sender = msg.get("message_info", {}).get("user_info", {}).get("user_nickname", "未知用户")
text = msg.get("detailed_plain_text", "")
timestamp = msg.get("time", 0)
time_str = time.strftime("%H:%M:%S", time.localtime(timestamp)) if timestamp else ""
context_texts.append(f"{sender} ({time_str}): {text}")
context_text = "\n".join(context_texts)
if observed_messages_str:
prompt += "观察到的最新聊天内容如下 (最近的消息在最后)\n---\n"
prompt += context_text
prompt += observed_messages_str
prompt += "\n---"
else:
prompt += "当前没有观察到新的聊天内容。\n"
prompt += "\n的内心想法是:"
prompt += "\n看了以上内容,你产生的内心想法是:"
if current_mind:
prompt += f"\n---\n{current_mind}\n---\n\n"
else:
@@ -737,23 +721,22 @@ class PFChatting:
"请结合你的内心想法和观察到的聊天内容,分析情况并使用 'decide_reply_action' 工具来决定你的最终行动。\n"
"决策依据:\n"
"1. 如果聊天内容无聊、与你无关、或者你的内心想法认为不适合回复(例如在讨论你不懂或不感兴趣的话题),选择 'no_reply'\n"
"2. 如果聊天内容值得回应,且适合用文字表达(参考你的内心想法),选择 'text_reply'。如果想在文字后追加一个表达情绪的表情,请同时提供 'emoji_query' (例如:'开心的''惊讶的')。\n"
"2. 如果聊天内容值得回应,且适合用文字表达(参考你的内心想法),选择 'text_reply'。如果你有情绪想表达,想在文字后追加一个表达情绪的表情,请同时提供 'emoji_query' (例如:'开心的''惊讶的')。\n"
"3. 如果聊天内容或你的内心想法适合用一个表情来回应(例如表示赞同、惊讶、无语等),选择 'emoji_reply' 并提供表情主题 'emoji_query'\n"
"4. 如果最后一条消息是你自己发的,并且之后没有人回复你,通常选择 'no_reply',除非有特殊原因需要追问。\n"
"5. 除非大家都在这么做,或者有特殊理由,否则不要重复别人刚刚说过的话或简单附和。\n"
"6. 表情包是用来表达情绪的,不要直接回复或评价别人的表情包,而是根据对话内容和情绪选择是否用表情回应。\n"
"7. 如果观察到的内容只有你自己的发言,选择 'no_reply'\n"
"8. 不要回复你自己的话,不要把自己的话当做别人说的。\n"
"必须调用 'decide_reply_action' 工具并提供 'action''reasoning'。如果选择了 'emoji_reply' 或者选择了 'text_reply' 并想追加表情,则必须提供 'emoji_query'"
)
prompt = await rel_manager.convert_all_person_sign_to_person_name(prompt)
prompt = parse_text_timestamps(prompt, mode="remove") # Remove timestamps before sending to LLM
return prompt
# --- 回复器 (Replier) 的定义 --- #
async def _replier_work(
self,
reason: str,
anchor_message: MessageRecv,
thinking_id: str,
) -> Optional[List[str]]:
@@ -770,6 +753,7 @@ class PFChatting:
# Ensure generate_response has access to current_mind if it's crucial context
response_set = await gpt_instance.generate_response(
reason,
anchor_message, # Pass anchor_message positionally (matches 'message' parameter)
thinking_id, # Pass thinking_id positionally
)
@@ -779,7 +763,7 @@ class PFChatting:
return None
# --- 准备并返回结果 --- #
logger.info(f"{log_prefix}[Replier-{thinking_id}] 成功生成了回复集: {' '.join(response_set)[:50]}...")
# logger.info(f"{log_prefix}[Replier-{thinking_id}] 成功生成了回复集: {' '.join(response_set)[:50]}...")
return response_set
except Exception as e:
@@ -796,10 +780,9 @@ class PFChatting:
chat = anchor_message.chat_stream
messageinfo = anchor_message.message_info
# Access global_config via controller
bot_user_info = UserInfo(
user_id=self.heartfc_controller.global_config.BOT_QQ,
user_nickname=self.heartfc_controller.global_config.BOT_NICKNAME,
user_id=global_config.BOT_QQ,
user_nickname=global_config.BOT_NICKNAME,
platform=messageinfo.platform,
)
@@ -845,10 +828,9 @@ class PFChatting:
message_set = MessageSet(chat, thinking_id)
mark_head = False
first_bot_msg = None
# Access global_config via controller
bot_user_info = UserInfo(
user_id=self.heartfc_controller.global_config.BOT_QQ,
user_nickname=self.heartfc_controller.global_config.BOT_NICKNAME,
user_id=global_config.BOT_QQ,
user_nickname=global_config.BOT_NICKNAME,
platform=anchor_message.message_info.platform,
)
for msg_text in response_set:
@@ -893,10 +875,9 @@ class PFChatting:
emoji_cq = image_path_to_base64(emoji_path)
thinking_time_point = round(time.time(), 2)
message_segment = Seg(type="emoji", data=emoji_cq)
# Access global_config via controller
bot_user_info = UserInfo(
user_id=self.heartfc_controller.global_config.BOT_QQ,
user_nickname=self.heartfc_controller.global_config.BOT_NICKNAME,
user_id=global_config.BOT_QQ,
user_nickname=global_config.BOT_NICKNAME,
platform=anchor_message.message_info.platform,
)
bot_message = MessageSending(
@@ -911,26 +892,3 @@ class PFChatting:
)
# Access MessageManager via controller
self.heartfc_controller.MessageManager().add_message(bot_message)
async def _update_relationship(self, anchor_message: Optional[MessageRecv], response_set: List[str]):
"""更新关系情绪 (尝试基于 anchor_message)"""
if not anchor_message or not anchor_message.chat_stream:
logger.error(f"{self._get_log_prefix()} 无法更新关系情绪,缺少有效的锚点消息或聊天流。")
return
# Access gpt and relationship_manager via controller
gpt_instance = self.heartfc_controller.gpt
relationship_manager_instance = self.heartfc_controller.relationship_manager
mood_manager_instance = self.heartfc_controller.mood_manager
config = self.heartfc_controller.global_config
ori_response = ",".join(response_set)
stance, emotion = await gpt_instance._get_emotion_tags(ori_response, anchor_message.processed_plain_text)
await relationship_manager_instance.calculate_update_relationship_value(
chat_stream=anchor_message.chat_stream,
label=emotion,
stance=stance,
)
mood_manager_instance.update_mood_from_emotion(emotion, config.mood_intensity_factor)
# --- Methods moved from HeartFC_Controller end ---

File diff suppressed because it is too large Load Diff

View File

@@ -5,7 +5,8 @@ import time
from pathlib import Path
import datetime
from rich.console import Console
from memory_manual_build import Memory_graph, Hippocampus # 海马体和记忆图
from Hippocampus import Hippocampus # 海马体和记忆图
from dotenv import load_dotenv
@@ -45,13 +46,13 @@ else:
# 查询节点信息
def query_mem_info(memory_graph: Memory_graph):
def query_mem_info(hippocampus: Hippocampus):
while True:
query = input("\n请输入新的查询概念(输入'退出'以结束):")
if query.lower() == "退出":
break
items_list = memory_graph.get_related_item(query)
items_list = hippocampus.memory_graph.get_related_item(query)
if items_list:
have_memory = False
first_layer, second_layer = items_list
@@ -312,14 +313,11 @@ def alter_mem_edge(hippocampus: Hippocampus):
async def main():
start_time = time.time()
# 创建记忆图
memory_graph = Memory_graph()
# 创建海马体
hippocampus = Hippocampus(memory_graph)
hippocampus = Hippocampus()
# 从数据库同步数据
hippocampus.sync_memory_from_db()
hippocampus.entorhinal_cortex.sync_memory_from_db()
end_time = time.time()
logger.info(f"\033[32m[加载海马体耗时: {end_time - start_time:.2f} 秒]\033[0m")
@@ -338,7 +336,7 @@ async def main():
query = -1
if query == 0:
query_mem_info(memory_graph)
query_mem_info(hippocampus.memory_graph)
elif query == 1:
add_mem_node(hippocampus)
elif query == 2:
@@ -355,7 +353,7 @@ async def main():
print("已结束操作")
break
hippocampus.sync_memory_to_db()
hippocampus.entorhinal_cortex.sync_memory_to_db()
if __name__ == "__main__":

View File

@@ -425,5 +425,49 @@ class PersonInfoManager:
logger.error(f"个人信息推断运行时出错: {str(e)}")
logger.exception("详细错误信息:")
async def get_or_create_person(
self, platform: str, user_id: int, nickname: str = None, user_cardname: str = None, user_avatar: str = None
) -> str:
"""
根据 platform 和 user_id 获取 person_id。
如果对应的用户不存在,则使用提供的可选信息创建新用户。
Args:
platform: 平台标识
user_id: 用户在该平台上的ID
nickname: 用户的昵称 (可选,用于创建新用户)
user_cardname: 用户的群名片 (可选,用于创建新用户)
user_avatar: 用户的头像信息 (可选,用于创建新用户)
Returns:
对应的 person_id。
"""
person_id = self.get_person_id(platform, user_id)
# 检查用户是否已存在
# 使用静态方法 get_person_id因此可以直接调用 db
document = db.person_info.find_one({"person_id": person_id})
if document is None:
logger.info(f"用户 {platform}:{user_id} (person_id: {person_id}) 不存在,将创建新记录。")
initial_data = {
"platform": platform,
"user_id": user_id,
"nickname": nickname,
"konw_time": int(datetime.datetime.now().timestamp()), # 添加初次认识时间
# 注意:这里没有添加 user_cardname 和 user_avatar因为它们不在 person_info_default 中
# 如果需要存储它们,需要先在 person_info_default 中定义
}
# 过滤掉值为 None 的初始数据
initial_data = {k: v for k, v in initial_data.items() if v is not None}
# 注意create_person_info 是静态方法
await PersonInfoManager.create_person_info(person_id, data=initial_data)
# 创建后,可以考虑立即为其取名,但这可能会增加延迟
# await self.qv_person_name(person_id, nickname, user_cardname, user_avatar)
logger.debug(f"已为 {person_id} 创建新记录,初始数据: {initial_data}")
return person_id
person_info_manager = PersonInfoManager()

View File

@@ -0,0 +1,284 @@
from src.config.config import global_config
# 不再直接使用 db
# from src.common.database import db
# 移除 logger 和 traceback因为错误处理移至 repository
# from src.common.logger import get_module_logger
# import traceback
from typing import List, Dict, Any, Tuple # 确保类型提示被导入
import time # 导入 time 模块以获取当前时间
# 导入新的 repository 函数
from src.common.message_repository import find_messages, count_messages
# 导入 PersonInfoManager 和时间转换工具
from src.plugins.person_info.person_info import person_info_manager
from src.plugins.chat.utils import translate_timestamp_to_human_readable
# 不再需要文件级别的 logger
# logger = get_module_logger(__name__)
def get_raw_msg_by_timestamp(
timestamp_start: float, timestamp_end: float, limit: int = 0, limit_mode: str = "latest"
) -> List[Dict[str, Any]]:
"""
获取从指定时间戳到指定时间戳的消息,按时间升序排序,返回消息列表
limit: 限制返回的消息数量0为不限制
limit_mode: 当 limit > 0 时生效。 'earliest' 表示获取最早的记录, 'latest' 表示获取最新的记录。默认为 'latest'
"""
filter_query = {"time": {"$gt": timestamp_start, "$lt": timestamp_end}}
# 只有当 limit 为 0 时才应用外部 sort
sort_order = [("time", 1)] if limit == 0 else None
return find_messages(filter=filter_query, sort=sort_order, limit=limit, limit_mode=limit_mode)
def get_raw_msg_by_timestamp_with_chat(
chat_id: str, timestamp_start: float, timestamp_end: float, limit: int = 0, limit_mode: str = "latest"
) -> List[Dict[str, Any]]:
"""获取在特定聊天从指定时间戳到指定时间戳的消息,按时间升序排序,返回消息列表
limit: 限制返回的消息数量0为不限制
limit_mode: 当 limit > 0 时生效。 'earliest' 表示获取最早的记录, 'latest' 表示获取最新的记录。默认为 'latest'
"""
filter_query = {"chat_id": chat_id, "time": {"$gt": timestamp_start, "$lt": timestamp_end}}
# 只有当 limit 为 0 时才应用外部 sort
sort_order = [("time", 1)] if limit == 0 else None
# 直接将 limit_mode 传递给 find_messages
return find_messages(filter=filter_query, sort=sort_order, limit=limit, limit_mode=limit_mode)
def get_raw_msg_by_timestamp_with_chat_users(
chat_id: str,
timestamp_start: float,
timestamp_end: float,
person_ids: list,
limit: int = 0,
limit_mode: str = "latest",
) -> List[Dict[str, Any]]:
"""获取某些特定用户在特定聊天从指定时间戳到指定时间戳的消息,按时间升序排序,返回消息列表
limit: 限制返回的消息数量0为不限制
limit_mode: 当 limit > 0 时生效。 'earliest' 表示获取最早的记录, 'latest' 表示获取最新的记录。默认为 'latest'
"""
filter_query = {
"chat_id": chat_id,
"time": {"$gt": timestamp_start, "$lt": timestamp_end},
"user_id": {"$in": person_ids},
}
# 只有当 limit 为 0 时才应用外部 sort
sort_order = [("time", 1)] if limit == 0 else None
return find_messages(filter=filter_query, sort=sort_order, limit=limit, limit_mode=limit_mode)
def get_raw_msg_by_timestamp_with_users(
timestamp_start: float, timestamp_end: float, person_ids: list, limit: int = 0, limit_mode: str = "latest"
) -> List[Dict[str, Any]]:
"""获取某些特定用户在 *所有聊天* 中从指定时间戳到指定时间戳的消息,按时间升序排序,返回消息列表
limit: 限制返回的消息数量0为不限制
limit_mode: 当 limit > 0 时生效。 'earliest' 表示获取最早的记录, 'latest' 表示获取最新的记录。默认为 'latest'
"""
filter_query = {"time": {"$gt": timestamp_start, "$lt": timestamp_end}, "user_id": {"$in": person_ids}}
# 只有当 limit 为 0 时才应用外部 sort
sort_order = [("time", 1)] if limit == 0 else None
return find_messages(filter=filter_query, sort=sort_order, limit=limit, limit_mode=limit_mode)
def get_raw_msg_before_timestamp(timestamp: float, limit: int = 0) -> List[Dict[str, Any]]:
"""获取指定时间戳之前的消息,按时间升序排序,返回消息列表
limit: 限制返回的消息数量0为不限制
"""
filter_query = {"time": {"$lt": timestamp}}
sort_order = [("time", 1)]
return find_messages(filter=filter_query, sort=sort_order, limit=limit)
def get_raw_msg_before_timestamp_with_chat(chat_id: str, timestamp: float, limit: int = 0) -> List[Dict[str, Any]]:
"""获取指定时间戳之前的消息,按时间升序排序,返回消息列表
limit: 限制返回的消息数量0为不限制
"""
filter_query = {"chat_id": chat_id, "time": {"$lt": timestamp}}
sort_order = [("time", 1)]
return find_messages(filter=filter_query, sort=sort_order, limit=limit)
def get_raw_msg_before_timestamp_with_users(timestamp: float, person_ids: list, limit: int = 0) -> List[Dict[str, Any]]:
"""获取指定时间戳之前的消息,按时间升序排序,返回消息列表
limit: 限制返回的消息数量0为不限制
"""
filter_query = {"time": {"$lt": timestamp}, "user_id": {"$in": person_ids}}
sort_order = [("time", 1)]
return find_messages(filter=filter_query, sort=sort_order, limit=limit)
def num_new_messages_since(chat_id: str, timestamp_start: float = 0.0, timestamp_end: float = None) -> int:
"""
检查特定聊天从 timestamp_start (不含) 到 timestamp_end (不含) 之间有多少新消息。
如果 timestamp_end 为 None则检查从 timestamp_start (不含) 到当前时间的消息。
"""
# 确定有效的结束时间戳
_timestamp_end = timestamp_end if timestamp_end is not None else time.time()
# 确保 timestamp_start < _timestamp_end
if timestamp_start >= _timestamp_end:
# logger.warning(f"timestamp_start ({timestamp_start}) must be less than _timestamp_end ({_timestamp_end}). Returning 0.")
return 0 # 起始时间大于等于结束时间,没有新消息
filter_query = {"chat_id": chat_id, "time": {"$gt": timestamp_start, "$lt": _timestamp_end}}
return count_messages(filter=filter_query)
def num_new_messages_since_with_users(
chat_id: str, timestamp_start: float, timestamp_end: float, person_ids: list
) -> int:
"""检查某些特定用户在特定聊天在指定时间戳之间有多少新消息"""
if not person_ids: # 保持空列表检查
return 0
filter_query = {
"chat_id": chat_id,
"time": {"$gt": timestamp_start, "$lt": timestamp_end},
"user_id": {"$in": person_ids},
}
return count_messages(filter=filter_query)
async def _build_readable_messages_internal(
messages: List[Dict[str, Any]],
replace_bot_name: bool = True,
merge_messages: bool = False,
timestamp_mode: str = "relative", # 新增参数控制时间戳格式
) -> Tuple[str, List[Tuple[float, str, str]]]:
"""
内部辅助函数,构建可读消息字符串和原始消息详情列表。
Args:
messages: 消息字典列表。
replace_bot_name: 是否将机器人的 user_id 替换为 ""
merge_messages: 是否合并来自同一用户的连续消息。
timestamp_mode: 时间戳的显示模式 ('relative', 'absolute', etc.)。传递给 translate_timestamp_to_human_readable。
Returns:
包含格式化消息的字符串和原始消息详情列表 (时间戳, 发送者名称, 内容) 的元组。
"""
if not messages:
return "", []
message_details: List[Tuple[float, str, str]] = []
# 1 & 2: 获取发送者信息并提取消息组件
for msg in messages:
user_info = msg.get("user_info", {})
platform = user_info.get("platform")
user_id = user_info.get("user_id")
user_nickname = user_info.get("nickname")
timestamp = msg.get("time")
content = msg.get("processed_plain_text", "") # 默认空字符串
# 检查必要信息是否存在
if not all([platform, user_id, timestamp is not None]):
# logger.warning(f"Skipping message due to missing info: {msg.get('_id', 'N/A')}")
continue
person_id = person_info_manager.get_person_id(platform, user_id)
# 根据 replace_bot_name 参数决定是否替换机器人名称
if replace_bot_name and user_id == global_config.BOT_QQ:
person_name = f"{global_config.BOT_NICKNAME}(你)"
else:
person_name = await person_info_manager.get_value(person_id, "person_name")
# 如果 person_name 未设置,则使用消息中的 nickname 或默认名称
if not person_name:
person_name = user_nickname
message_details.append((timestamp, person_name, content))
if not message_details:
return "", []
message_details.sort(key=lambda x: x[0]) # 按时间戳(第一个元素)升序排序,越早的消息排在前面
# 3: 合并连续消息 (如果 merge_messages 为 True)
merged_messages = []
if merge_messages and message_details:
# 初始化第一个合并块
current_merge = {
"name": message_details[0][1],
"start_time": message_details[0][0],
"end_time": message_details[0][0],
"content": [message_details[0][2]],
}
for i in range(1, len(message_details)):
timestamp, name, content = message_details[i]
# 如果是同一个人发送的连续消息且时间间隔小于等于60秒
if name == current_merge["name"] and (timestamp - current_merge["end_time"] <= 60):
current_merge["content"].append(content)
current_merge["end_time"] = timestamp # 更新最后消息时间
else:
# 保存上一个合并块
merged_messages.append(current_merge)
# 开始新的合并块
current_merge = {"name": name, "start_time": timestamp, "end_time": timestamp, "content": [content]}
# 添加最后一个合并块
merged_messages.append(current_merge)
elif message_details: # 如果不合并消息,则每个消息都是一个独立的块
for timestamp, name, content in message_details:
merged_messages.append(
{
"name": name,
"start_time": timestamp, # 起始和结束时间相同
"end_time": timestamp,
"content": [content], # 内容只有一个元素
}
)
# 4 & 5: 格式化为字符串
output_lines = []
for merged in merged_messages:
# 使用指定的 timestamp_mode 格式化时间
readable_time = translate_timestamp_to_human_readable(merged["start_time"], mode=timestamp_mode)
header = f"{readable_time}{merged['name']} 说:"
output_lines.append(header)
# 将内容合并,并添加缩进
for line in merged["content"]:
stripped_line = line.strip()
if stripped_line: # 过滤空行
if stripped_line.endswith(""):
stripped_line = stripped_line.rstrip("")
output_lines.append(f"{stripped_line};")
output_lines += "\n"
formatted_string = "".join(output_lines)
# 返回格式化后的字符串和原始的 message_details 列表
return formatted_string, message_details
async def build_readable_messages_with_list(
messages: List[Dict[str, Any]],
replace_bot_name: bool = True,
merge_messages: bool = False,
timestamp_mode: str = "relative",
) -> Tuple[str, List[Tuple[float, str, str]]]:
"""
将消息列表转换为可读的文本格式,并返回原始(时间戳, 昵称, 内容)列表。
允许通过参数控制格式化行为。
"""
formatted_string, details_list = await _build_readable_messages_internal(
messages, replace_bot_name, merge_messages, timestamp_mode
)
return formatted_string, details_list
async def build_readable_messages(
messages: List[Dict[str, Any]],
replace_bot_name: bool = True,
merge_messages: bool = False,
timestamp_mode: str = "relative",
) -> str:
"""
将消息列表转换为可读的文本格式。
允许通过参数控制格式化行为。
"""
formatted_string, _ = await _build_readable_messages_internal(
messages, replace_bot_name, merge_messages, timestamp_mode
)
return formatted_string