优化prompt格式&代码结构

提示分三类:人设、背景、要求
pull/260/head
Hosigus 2025-03-12 14:47:34 +08:00
parent f7e8050fd7
commit 712505f6ca
1 changed files with 171 additions and 122 deletions

View File

@ -1,6 +1,6 @@
import random import random
import time import time
from typing import Optional from typing import Optional, List
from loguru import logger from loguru import logger
from ...common.database import Database from ...common.database import Database
@ -18,25 +18,140 @@ class PromptBuilder:
self.activate_messages = '' self.activate_messages = ''
self.db = Database.get_instance() self.db = Database.get_instance()
async def _build_prompt(self, async def _build_prompt(self,
message_txt: str, message_txt: str,
sender_name: str = "某人", sender_name: str = "某人",
relationship_value: float = 0.0, relationship_value: float = 0.0,
stream_id: Optional[int] = None) -> tuple[str, str]: stream_id: Optional[int] = None) -> tuple[str, str]:
"""构建prompt """构建prompt
Args: Args:
message_txt: 消息文本 message_txt: 消息文本
sender_name: 发送者昵称 sender_name: 发送者昵称
relationship_value: 关系值 relationship_value: 关系值
group_id: 群组ID stream_id: 消息上下文ID
Returns: Returns:
str: 构建好的prompt str: 构建好的prompt
""" """
# 先禁用关系
# 开始构建prompt
role_core = await self._build_role_core_prompt(message_txt, sender_name, relationship_value, stream_id)
chat_context = await self._build_chat_context_prompt(message_txt, sender_name, relationship_value, stream_id)
constraints = await self._build_constraints_prompt(message_txt, sender_name, relationship_value, stream_id)
# 合并prompt
prompt = '\n\n'.join((
'# 角色核心设定\n* ' + '\n* '.join(role_core),
'# 对话背景\n* ' + '\n* '.join(chat_context),
'# 要求\n* ' + '\n* '.join(constraints),
'# 目标: 发言或回复用户'
))
'''读空气prompt处理 [未启用]'''
# activate_prompt_chec
# extra_check_info = f"请注意把握k = f"以上是群里正在进行的聊天,昵称为 '{sender_name}' 的用户说的:{message_txt}。引起了你的注意,你和他{relation_prompt},你想要{relation_prompt_2},但是这不一定是合适的时机,请你决定是否要回应这条消息。"
# #
# # prompt_personality_check = ''群里的聊天内容的基础上,综合群内的氛围,例如,和{global_config.BOT_NICKNAME}相关的话题要积极回复,如果是at自己的消息一定要回复如果自己正在和别人聊天一定要回复其他话题如果合适搭话也可以回复如果认为应该回复请输出yes否则输出no请注意是决定是否需要回复而不是编写回复内容除了yes和no不要输出任何回复内容。"
# if personality_choice < probability_1: # 第一种人格
# prompt_personality_check = f'''你的网名叫{global_config.BOT_NICKNAME}{personality[0]}, 你正在浏览qq群{promt_info_prompt} {activate_prompt_check} {extra_check_info}'''
# elif personality_choice < probability_1 + probability_2: # 第二种人格
# prompt_personality_check = f'''你的网名叫{global_config.BOT_NICKNAME}{personality[1]}, 你正在浏览qq群{promt_info_prompt} {activate_prompt_check} {extra_check_info}'''
# else: # 第三种人格
# prompt_personality_check = f'''你的网名叫{global_config.BOT_NICKNAME}{personality[2]}, 你正在浏览qq群{promt_info_prompt} {activate_prompt_check} {extra_check_info}'''
# prompt_check_if_response = f"{prompt_info}\n{prompt_date}\n{chat_talking_prompt}\n{prompt_personality_check}"
prompt_check_if_response = ''
return prompt, prompt_check_if_response
async def _build_role_core_prompt(self,
message_txt: str,
sender_name: str = "某人",
relationship_value: float = 0.0,
stream_id: Optional[int] = None) -> List[str]:
role_core_prompt = []
# 人格选择
personality = random.choices(global_config.PROMPT_PERSONALITY, global_config.PERSONALITY_PROBS)[0]
role_core_prompt.append(personality)
role_core_prompt.append(f'你的网名叫{global_config.BOT_NICKNAME}')
if len(global_config.BOT_ALIAS_NAMES) > 0:
role_core_prompt.append(f'别人也叫你:{"/".join(global_config.BOT_ALIAS_NAMES)}')
# 心情
role_core_prompt.append(MoodManager.get_instance().get_prompt())
# 日程构建
current_date = time.strftime("%Y-%m-%d", time.localtime())
current_time = time.strftime("%H:%M:%S", time.localtime())
bot_schedule_now_time, bot_schedule_now_activity = bot_schedule.get_current_task()
schedule_str = '\n'.join([f"\t- {time} {activity}"
for time, activity in bot_schedule.today_schedule.items()])
role_core_prompt.append(f'今天是{current_date},现在是{current_time},你今天的日程是:\n{schedule_str}')
role_core_prompt.append(f'你现在正在{bot_schedule_now_activity}')
# 知识构建 [未启用]
start_time = time.time()
knowledge_prompt = ''
promt_info_prompt = ''
knowledge_prompt = await self.get_prompt_info(message_txt, threshold=0.5)
if knowledge_prompt:
knowledge_prompt = f'''你有以下这些[知识]{knowledge_prompt}请你记住上面的[
知识]之后可能会用到-'''
end_time = time.time()
logger.debug(f"知识检索耗时: {(end_time - start_time):.3f}")
# 中文高手(新加的好玩功能)
prompt_ger = ''
if random.random() < 0.04:
prompt_ger += '你喜欢用倒装句'
if random.random() < 0.02:
prompt_ger += '你喜欢用反问句'
if random.random() < 0.01:
prompt_ger += '你喜欢用文言文'
if len(prompt_ger) > 0:
role_core_prompt.append(prompt_ger)
return role_core_prompt
async def _build_chat_context_prompt(self,
message_txt: str,
sender_name: str = "某人",
relationship_value: float = 0.0,
stream_id: Optional[int] = None) -> List[str]:
chat_context_prompt = []
# 获取聊天上下文
chat_in_group = True
chat_talking_prompt = ''
if stream_id:
chat_context: List[str] = get_recent_group_detailed_plain_text(self.db, stream_id,
limit=global_config.MAX_CONTEXT_SIZE,
combine=False)
chat_stream = chat_manager.get_stream(stream_id)
if chat_stream.group_info:
chat_talking_prompt = "以下是群里正在聊的内容:\n\t- "
else:
chat_in_group = False
chat_talking_prompt = f"以下是你正在和{sender_name}私聊的内容:\n"
# print(f"\033[1;34m[调试]\033[0m 已从数据库获取群 {stream_id} 的消息记录:{chat_talking_prompt}")
chat_talking_prompt += "\n\t- ".join(chat_context)
chat_context_prompt.append(chat_talking_prompt)
# [未启用]
if chat_in_group:
prompt_in_group = f"你正在浏览{chat_stream.platform}"
else:
prompt_in_group = f"你正在{chat_stream.platform}上和{sender_name}私聊"
# 使用新的记忆获取方法
memory_prompt = await self._build_memory_prompt(message_txt)
if len(memory_prompt) > 0:
chat_context_prompt.append(memory_prompt)
# 关系 [未启用]
if 0 > 30: if 0 > 30:
relation_prompt = "关系特别特别好,你很喜欢喜欢他" relation_prompt = "关系特别特别好,你很喜欢喜欢他"
relation_prompt_2 = "热情发言或者回复" relation_prompt_2 = "热情发言或者回复"
@ -47,47 +162,17 @@ class PromptBuilder:
relation_prompt = "关系一般" relation_prompt = "关系一般"
relation_prompt_2 = "发言或者回复" relation_prompt_2 = "发言或者回复"
# 开始构建prompt # 激活prompt构建
activate_prompt = ''
if chat_in_group:
activate_prompt = f"现在昵称为 '{sender_name}' 的用户说的:{message_txt}。引起了你的注意,你和ta{relation_prompt},你想要{relation_prompt_2}"
else:
activate_prompt = f"现在昵称为 '{sender_name}' 的用户说的:{message_txt}。引起了你的注意,你和ta{relation_prompt},你想要{relation_prompt_2}"
chat_context_prompt.append(activate_prompt)
# 心情 return chat_context_prompt
mood_manager = MoodManager.get_instance()
mood_prompt = mood_manager.get_prompt()
# 日程构建 async def _build_memory_prompt(self, message_txt: str) -> str:
current_date = time.strftime("%Y-%m-%d", time.localtime())
current_time = time.strftime("%H:%M:%S", time.localtime())
bot_schedule_now_time, bot_schedule_now_activity = bot_schedule.get_current_task()
prompt_date = f'''今天是{current_date},现在是{current_time},你今天的日程是:\n{bot_schedule.today_schedule}\n你现在正在{bot_schedule_now_activity}\n'''
# 知识构建
start_time = time.time()
prompt_info = ''
promt_info_prompt = ''
prompt_info = await self.get_prompt_info(message_txt, threshold=0.5)
if prompt_info:
prompt_info = f'''你有以下这些[知识]{prompt_info}请你记住上面的[
知识]之后可能会用到-'''
end_time = time.time()
logger.debug(f"知识检索耗时: {(end_time - start_time):.3f}")
# 获取聊天上下文
chat_in_group=True
chat_talking_prompt = ''
if stream_id:
chat_talking_prompt = get_recent_group_detailed_plain_text(self.db, 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 = f"以下是群里正在聊天的内容:\n{chat_talking_prompt}"
else:
chat_in_group=False
chat_talking_prompt = f"以下是你正在和{sender_name}私聊的内容:\n{chat_talking_prompt}"
# print(f"\033[1;34m[调试]\033[0m 已从数据库获取群 {group_id} 的消息记录:{chat_talking_prompt}")
# 使用新的记忆获取方法
memory_prompt = '' memory_prompt = ''
start_time = time.time() start_time = time.time()
@ -103,9 +188,9 @@ class PromptBuilder:
# 格式化记忆内容 # 格式化记忆内容
memory_items = [] memory_items = []
for memory in relevant_memories: for memory in relevant_memories:
memory_items.append(f"关于「{memory['topic']}」的记忆:{memory['content']}") memory_items.append(f"\t- 关于「{memory['topic']}」的记忆:{memory['content']}")
memory_prompt = "看到这些聊天,你想起来:\n" + "\n".join(memory_items) + "\n" memory_prompt = "看到这些聊天,你想起来:\n" + "\n".join(memory_items)
# 打印调试信息 # 打印调试信息
logger.debug("[记忆检索]找到以下相关记忆:") logger.debug("[记忆检索]找到以下相关记忆:")
@ -115,15 +200,38 @@ class PromptBuilder:
end_time = time.time() end_time = time.time()
logger.info(f"回忆耗时: {(end_time - start_time):.3f}") logger.info(f"回忆耗时: {(end_time - start_time):.3f}")
# 激活prompt构建 return memory_prompt
activate_prompt = ''
if chat_in_group:
activate_prompt = f"以上是群里正在进行的聊天,{memory_prompt} 现在昵称为 '{sender_name}' 的用户说的:{message_txt}。引起了你的注意,你和ta{relation_prompt},{mood_prompt},你想要{relation_prompt_2}"
else:
activate_prompt = f"以上是你正在和{sender_name}私聊的内容,{memory_prompt} 现在昵称为 '{sender_name}' 的用户说的:{message_txt}。引起了你的注意,你和ta{relation_prompt},{mood_prompt},你想要{relation_prompt_2}"
# 关键词检测与反应 async def _build_constraints_prompt(self,
# 获取当前规则中(实际命中的关键词-反应) message_txt: str,
sender_name: str = "某人",
relationship_value: float = 0.0,
stream_id: Optional[int] = None) -> List[str]:
constraints_prompt = []
keyword_prompt = self._build_keywords_reaction_prompt(message_txt)
if len(keyword_prompt) > 0:
constraints_prompt.append(keyword_prompt)
# 额外信息要求
constraints_prompt.append('请给出日常且口语化的回复,平淡一些,尽量简短一些,尤其注意在没明确提到时不要过多提及自身的背景.')
constraints_prompt.append('不要直接回复别人发的表情包.')
constraints_prompt.append('请注意把握群里的聊天内容,不要刻意突出自身学科背景,不要回复的太有条理,可以有个性.')
constraints_prompt.append('只输出回复内容,不要输出其他任何内容,包括前后缀、冒号、引号、括号、表情等.')
constraints_prompt.append('请表现你自己的见解和观点,不要一昧迎合,可以表现个性,尽量简短一些.')
return constraints_prompt
def _build_keywords_reaction_prompt(self, message_txt: str) -> str:
"""关键词检测与反应
Returns: 没命中返回空字符串命中时返回prompt格式如下
群聊的发言命中了预设规则你需要按照要求的反应回复:
- 命中关键字: 'test', '测试'; 回复要求: '回答测试成功'
- 命中关键字: '人机'; 回复要求: '否定这一事实,攻击或反问他。注意,只有认为你是机器人才回复,否则不要否认'
"""
# 获取当前规则 (实际命中的关键词-反应)
triggered_keyword_prompt_arr = [] triggered_keyword_prompt_arr = []
lower_msg_text = message_txt.lower() lower_msg_text = message_txt.lower()
for rule in global_config.keywords_reaction_rules: for rule in global_config.keywords_reaction_rules:
@ -135,73 +243,14 @@ class PromptBuilder:
logger.info(f"检测到关键词:{matched_kws},触发反应:{reaction}") logger.info(f"检测到关键词:{matched_kws},触发反应:{reaction}")
triggered_keyword_prompt_arr.append((matched_kws, reaction)) triggered_keyword_prompt_arr.append((matched_kws, reaction))
"""
群聊的发言有命中预设规则你需要按照要求的反应回复:
- 命中关键字: 'test', 'hello'; 回复要求: '回答测试成功'
- 命中关键字: '人机'; 回复要求: '否定这一事实,攻击或反问他。注意,只有认为你是机器人才回复,否则不要否认'
"""
keywords_reaction_prompt = '' keywords_reaction_prompt = ''
if len(triggered_keyword_prompt_arr) > 0: if len(triggered_keyword_prompt_arr) > 0:
keywords_reaction_prompt = '群聊的发言有命中,你需要按照预设的反应回复:\n' \ keywords_reaction_prompt = '群聊的发言命中了预设规则,你需要按照要求的反应回复:\n' \
+ "\n".join(f'- 命中关键字: {", ".join(map(repr, matched_kws))}; 回复要求: {reaction!r}' for matched_kws, reaction in triggered_keyword_prompt_arr) + "\n".join(
f'\t - 命中关键字: {", ".join(map(repr, matched_kws))}; 回复要求: {reaction!r}' for
matched_kws, reaction in triggered_keyword_prompt_arr)
# 人格选择 return keywords_reaction_prompt
personality = random.choices(global_config.PROMPT_PERSONALITY, global_config.PERSONALITY_PROBS)[0]
prompt_personality = f'{activate_prompt}你的网名叫{global_config.BOT_NICKNAME},你还有很多别名:{"/".join(global_config.BOT_ALIAS_NAMES)}'
personality_choice = random.random()
if chat_in_group:
prompt_in_group=f"你正在浏览{chat_stream.platform}"
else:
prompt_in_group=f"你正在{chat_stream.platform}上和{sender_name}私聊"
if personality_choice < probability_1: # 第一种人格
prompt_personality += f'''{personality}, 你正在浏览qq群,{promt_info_prompt},
现在请你给出日常且口语化的回复平淡一些尽量简短一些{keywords_reaction_prompt}
请注意把握群里的聊天内容不要刻意突出自身学科背景不要回复的太有条理可以有个性'''
elif personality_choice < probability_1 + probability_2: # 第二种人格
prompt_personality += f'''{personality}, 你正在浏览qq群{promt_info_prompt},
现在请你给出日常且口语化的回复请表现你自己的见解不要一昧迎合尽量简短一些{keywords_reaction_prompt}
请你表达自己的见解和观点可以有个性'''
else: # 第三种人格
prompt_personality += f'''{personality}, 你正在浏览qq群{promt_info_prompt},
现在请你给出日常且口语化的回复请表现你自己的见解不要一昧迎合尽量简短一些{keywords_reaction_prompt}
请你表达自己的见解和观点可以有个性'''
# 中文高手(新加的好玩功能)
prompt_ger = ''
if random.random() < 0.04:
prompt_ger += '你喜欢用倒装句'
if random.random() < 0.02:
prompt_ger += '你喜欢用反问句'
if random.random() < 0.01:
prompt_ger += '你喜欢用文言文'
# 额外信息要求
extra_info = '''但是记得回复平淡一些,简短一些,尤其注意在没明确提到时不要过多提及自身的背景, 不要直接回复别人发的表情包,记住不要输出多余内容(包括前后缀,冒号和引号,括号,表情等),只需要输出回复内容就好,不要输出其他任何内容'''
# 合并prompt
prompt = ""
prompt += f"{prompt_info}\n"
prompt += f"{prompt_date}\n"
prompt += f"{chat_talking_prompt}\n"
prompt += f"{prompt_personality}\n"
prompt += f"{prompt_ger}\n"
prompt += f"{extra_info}\n"
'''读空气prompt处理'''
activate_prompt_check = f"以上是群里正在进行的聊天,昵称为 '{sender_name}' 的用户说的:{message_txt}。引起了你的注意,你和他{relation_prompt},你想要{relation_prompt_2},但是这不一定是合适的时机,请你决定是否要回应这条消息。"
prompt_personality_check = ''
extra_check_info = f"请注意把握群里的聊天内容的基础上,综合群内的氛围,例如,和{global_config.BOT_NICKNAME}相关的话题要积极回复,如果是at自己的消息一定要回复如果自己正在和别人聊天一定要回复其他话题如果合适搭话也可以回复如果认为应该回复请输出yes否则输出no请注意是决定是否需要回复而不是编写回复内容除了yes和no不要输出任何回复内容。"
if personality_choice < probability_1: # 第一种人格
prompt_personality_check = f'''你的网名叫{global_config.BOT_NICKNAME}{personality[0]}, 你正在浏览qq群{promt_info_prompt} {activate_prompt_check} {extra_check_info}'''
elif personality_choice < probability_1 + probability_2: # 第二种人格
prompt_personality_check = f'''你的网名叫{global_config.BOT_NICKNAME}{personality[1]}, 你正在浏览qq群{promt_info_prompt} {activate_prompt_check} {extra_check_info}'''
else: # 第三种人格
prompt_personality_check = f'''你的网名叫{global_config.BOT_NICKNAME}{personality[2]}, 你正在浏览qq群{promt_info_prompt} {activate_prompt_check} {extra_check_info}'''
prompt_check_if_response = f"{prompt_info}\n{prompt_date}\n{chat_talking_prompt}\n{prompt_personality_check}"
return prompt, prompt_check_if_response
def _build_initiative_prompt_select(self, group_id, probability_1=0.8, probability_2=0.1): def _build_initiative_prompt_select(self, group_id, probability_1=0.8, probability_2=0.1):
current_date = time.strftime("%Y-%m-%d", time.localtime()) current_date = time.strftime("%Y-%m-%d", time.localtime())