feat:可选择是否压缩人设

This commit is contained in:
SengokuCola
2025-06-25 21:31:19 +08:00
parent cdb0912b5a
commit 81356e3813
6 changed files with 231 additions and 68 deletions

View File

@@ -240,7 +240,7 @@ class ExpressionSelector:
valid_expressions.append(expression)
# 对选中的表达方式count数+0.1
self.update_expression_count(chat_id, expression, 0.1)
self.update_expression_count(chat_id, expression, 0.0001)
# logger.info(f"LLM从{len(all_expressions)}个情境中选择了{len(valid_expressions)}个")
return valid_expressions

View File

@@ -19,6 +19,8 @@ from src.chat.utils.chat_message_builder import build_readable_messages, get_raw
from src.chat.express.exprssion_learner import get_expression_learner
import time
import random
import ast
from src.person_info.person_info import get_person_info_manager
from datetime import datetime
import re
@@ -277,6 +279,8 @@ class DefaultReplyer:
reply_data=None,
) -> str:
chat_stream = self.chat_stream
person_info_manager = get_person_info_manager()
bot_person_id = person_info_manager.get_person_id("system", "bot_id")
is_group_chat = bool(chat_stream.group_info)
@@ -389,8 +393,26 @@ class DefaultReplyer:
bot_nickname = f",也有人叫你{','.join(global_config.bot.alias_names)}"
else:
bot_nickname = ""
bot_core_personality = global_config.personality.personality_core
indentify_block = f"你的名字是{bot_name}{bot_nickname},你{bot_core_personality}"
short_impression = await person_info_manager.get_value(bot_person_id, "short_impression")
# 解析字符串形式的Python列表
try:
if isinstance(short_impression, str) and short_impression.strip():
short_impression = ast.literal_eval(short_impression)
elif not short_impression:
logger.warning("short_impression为空使用默认值")
short_impression = ["友好活泼", "人类"]
except (ValueError, SyntaxError) as e:
logger.error(f"解析short_impression失败: {e}, 原始值: {short_impression}")
short_impression = ["友好活泼", "人类"]
# 确保short_impression是列表格式且有足够的元素
if not isinstance(short_impression, list) or len(short_impression) < 2:
logger.warning(f"short_impression格式不正确: {short_impression}, 使用默认值")
short_impression = ["友好活泼", "人类"]
personality = short_impression[0]
identity = short_impression[1]
prompt_personality = personality + "" + identity
indentify_block = f"你的名字是{bot_name}{bot_nickname},你{prompt_personality}"
if sender:
reply_target_block = f"现在{sender}说的:{target}。引起了你的注意,你想要在群里发言或者回复这条消息。"

View File

@@ -11,6 +11,8 @@ import random
from src.person_info.person_info import get_person_info_manager
from src.chat.express.expression_selector import expression_selector
import re
import json
import ast
from src.person_info.relationship_manager import get_relationship_manager
@@ -91,13 +93,30 @@ class PromptBuilder:
enable_planner: bool = False,
available_actions=None,
) -> str:
core_personality = global_config.personality.personality_core
person_info_manager = get_person_info_manager()
bot_person_id = person_info_manager.get_person_id("system", "bot_id")
short_impression = await person_info_manager.get_value(bot_person_id, "short_impression")
prompt_personality = core_personality
if short_impression:
prompt_personality += short_impression
# 解析字符串形式的Python列表
try:
if isinstance(short_impression, str) and short_impression.strip():
short_impression = ast.literal_eval(short_impression)
elif not short_impression:
logger.warning("short_impression为空使用默认值")
short_impression = ["友好活泼", "人类"]
except (ValueError, SyntaxError) as e:
logger.error(f"解析short_impression失败: {e}, 原始值: {short_impression}")
short_impression = ["友好活泼", "人类"]
# 确保short_impression是列表格式且有足够的元素
if not isinstance(short_impression, list) or len(short_impression) < 2:
logger.warning(f"short_impression格式不正确: {short_impression}, 使用默认值")
short_impression = ["友好活泼", "人类"]
personality = short_impression[0]
identity = short_impression[1]
prompt_personality = personality + "" + identity
is_group_chat = bool(chat_stream.group_info)

View File

@@ -37,6 +37,9 @@ class PersonalityConfig(ConfigBase):
personality_sides: list[str] = field(default_factory=lambda: [])
"""人格侧写"""
compress_personality: bool = True
"""是否压缩人格压缩后会精简人格信息节省token消耗并提高回复性能但是会丢失一些信息如果人设不长可以关闭"""
@dataclass
class IdentityConfig(ConfigBase):
@@ -45,6 +48,9 @@ class IdentityConfig(ConfigBase):
identity_detail: list[str] = field(default_factory=lambda: [])
"""身份特征"""
compress_indentity: bool = True
"""是否压缩身份压缩后会精简身份信息节省token消耗并提高回复性能但是会丢失一些信息如果不长可以关闭"""
@dataclass
class RelationshipConfig(ConfigBase):

View File

@@ -58,7 +58,7 @@ class Individuality:
self.name = bot_nickname
# 检查配置变化,如果变化则清空
await self._check_config_and_clear_if_changed(
personality_changed, identity_changed = await self._check_config_and_clear_if_changed(
bot_nickname, personality_core, personality_sides, identity_detail
)
@@ -96,11 +96,60 @@ class Individuality:
await person_info_manager.update_one_field(
self.bot_person_id, "impression", impression_text, data=update_data
)
logger.info("已将完整人设更新到bot的impression中")
# 创建压缩版本的short_impression
asyncio.create_task(self._create_compressed_impression(personality_core, personality_sides, identity_detail))
logger.debug("已将完整人设更新到bot的impression中")
# 根据变化情况决定是否重新创建
personality_result = None
identity_result = None
if personality_changed:
logger.info("检测到人格配置变化,重新生成压缩版本")
personality_result = await self._create_personality(personality_core, personality_sides)
else:
logger.info("人格配置未变化,使用缓存版本")
# 从缓存中获取已有的personality结果
existing_short_impression = await person_info_manager.get_value(self.bot_person_id, "short_impression")
if existing_short_impression:
try:
existing_data = json.loads(existing_short_impression) if isinstance(existing_short_impression, str) else existing_short_impression
if isinstance(existing_data, list) and len(existing_data) >= 1:
personality_result = existing_data[0]
except (json.JSONDecodeError, TypeError, IndexError):
logger.warning("无法解析现有的short_impression将重新生成人格部分")
personality_result = await self._create_personality(personality_core, personality_sides)
else:
logger.info("未找到现有的人格缓存,重新生成")
personality_result = await self._create_personality(personality_core, personality_sides)
if identity_changed:
logger.info("检测到身份配置变化,重新生成压缩版本")
identity_result = await self._create_identity(identity_detail)
else:
logger.info("身份配置未变化,使用缓存版本")
# 从缓存中获取已有的identity结果
existing_short_impression = await person_info_manager.get_value(self.bot_person_id, "short_impression")
if existing_short_impression:
try:
existing_data = json.loads(existing_short_impression) if isinstance(existing_short_impression, str) else existing_short_impression
if isinstance(existing_data, list) and len(existing_data) >= 2:
identity_result = existing_data[1]
except (json.JSONDecodeError, TypeError, IndexError):
logger.warning("无法解析现有的short_impression将重新生成身份部分")
identity_result = await self._create_identity(identity_detail)
else:
logger.info("未找到现有的身份缓存,重新生成")
identity_result = await self._create_identity(identity_detail)
result = [personality_result, identity_result]
# 更新short_impression字段
if personality_result and identity_result:
person_info_manager = get_person_info_manager()
await person_info_manager.update_one_field(self.bot_person_id, "short_impression", result)
logger.info("已将人设构建")
else:
logger.error("人设构建失败")
asyncio.create_task(self.express_style.extract_and_store_personality_expressions())
def to_dict(self) -> dict:
@@ -271,31 +320,61 @@ class Individuality:
def _get_config_hash(
self, bot_nickname: str, personality_core: str, personality_sides: list, identity_detail: list
) -> str:
"""获取当前personality和identity配置的哈希值"""
config_data = {
) -> tuple[str, str]:
"""获取personality和identity配置的哈希值
Returns:
tuple: (personality_hash, identity_hash)
"""
# 人格配置哈希
personality_config = {
"nickname": bot_nickname,
"personality_core": personality_core,
"personality_sides": sorted(personality_sides),
"identity_detail": sorted(identity_detail),
"compress_personality": global_config.personality.compress_personality,
}
config_str = json.dumps(config_data, sort_keys=True)
return hashlib.md5(config_str.encode("utf-8")).hexdigest()
personality_str = json.dumps(personality_config, sort_keys=True)
personality_hash = hashlib.md5(personality_str.encode("utf-8")).hexdigest()
# 身份配置哈希
identity_config = {
"identity_detail": sorted(identity_detail),
"compress_identity": global_config.identity.compress_indentity,
}
identity_str = json.dumps(identity_config, sort_keys=True)
identity_hash = hashlib.md5(identity_str.encode("utf-8")).hexdigest()
return personality_hash, identity_hash
async def _check_config_and_clear_if_changed(
self, bot_nickname: str, personality_core: str, personality_sides: list, identity_detail: list
):
"""检查配置是否发生变化,如果变化则清空info_list"""
) -> tuple[bool, bool]:
"""检查配置是否发生变化,如果变化则清空相应缓存
Returns:
tuple: (personality_changed, identity_changed)
"""
person_info_manager = get_person_info_manager()
current_hash = self._get_config_hash(bot_nickname, personality_core, personality_sides, identity_detail)
current_personality_hash, current_identity_hash = self._get_config_hash(
bot_nickname, personality_core, personality_sides, identity_detail
)
meta_info = self._load_meta_info()
stored_hash = meta_info.get("config_hash")
stored_personality_hash = meta_info.get("personality_hash")
stored_identity_hash = meta_info.get("identity_hash")
if current_hash != stored_hash:
logger.info("检测到人格配置发生变化,将清空原有的关键词缓存。")
personality_changed = current_personality_hash != stored_personality_hash
identity_changed = current_identity_hash != stored_identity_hash
# 清空数据库中的info_list
if personality_changed:
logger.info("检测到人格配置发生变化")
if identity_changed:
logger.info("检测到身份配置发生变化")
# 如果任何一个发生变化都需要清空info_list因为这影响整体人设
if personality_changed or identity_changed:
logger.info("将清空原有的关键词缓存")
update_data = {
"platform": "system",
"user_id": "bot_id",
@@ -304,9 +383,14 @@ class Individuality:
}
await person_info_manager.update_one_field(self.bot_person_id, "info_list", [], data=update_data)
# 更新元信息文件,重置计数器
new_meta_info = {"config_hash": current_hash}
self._save_meta_info(new_meta_info)
# 更新元信息文件
new_meta_info = {
"personality_hash": current_personality_hash,
"identity_hash": current_identity_hash,
}
self._save_meta_info(new_meta_info)
return personality_changed, identity_changed
def _load_meta_info(self) -> dict:
"""从JSON文件中加载元信息"""
@@ -368,8 +452,8 @@ class Individuality:
logger.error(f"解析info_list失败: {info_list_json}")
return keywords
async def _create_compressed_impression(
self, personality_core: str, personality_sides: list, identity_detail: list
async def _create_personality(
self, personality_core: str, personality_sides: list
) -> str:
"""使用LLM创建压缩版本的impression
@@ -381,30 +465,62 @@ class Individuality:
Returns:
str: 压缩后的impression文本
"""
logger.info("正在构建人格.........")
# 核心人格保持不变
compressed_parts = []
personality_parts = []
if personality_core:
compressed_parts.append(f"{personality_core}")
personality_parts.append(f"{personality_core}")
# 准备需要压缩的内容
content_to_compress = []
if personality_sides:
content_to_compress.append(f"人格特质: {''.join(personality_sides)}")
if identity_detail:
content_to_compress.append(f"身份背景: {''.join(identity_detail)}")
if global_config.personality.compress_personality:
personality_to_compress = []
if personality_sides:
personality_to_compress.append(f"人格特质: {''.join(personality_sides)}")
prompt = f"""请将以下人格信息进行简洁压缩,保留主要内容,用简练的中文表达:
{personality_to_compress}
if not content_to_compress:
# 如果没有需要压缩的内容,直接返回核心人格
result = "".join(compressed_parts)
return result + "" if result else ""
要求:
1. 保持原意不变,尽量使用原文
2. 尽量简洁不超过30字
3. 直接输出压缩后的内容,不要解释"""
# 使用LLM压缩其他内容
try:
compress_content = "".join(content_to_compress)
response, (_, _) = await self.model.generate_response_async(
prompt=prompt,
)
if response.strip():
personality_parts.append(response.strip())
logger.info(f"精简人格侧面: {response.strip()}")
else:
logger.error(f"使用LLM压缩人设时出错: {response}")
if personality_parts:
personality_result = "".join(personality_parts)
else:
personality_result = personality_core
else:
personality_result = personality_core
if personality_sides:
personality_result += "".join(personality_sides)
prompt = f"""请将以下人设信息进行简洁压缩,保留主要内容,用简练的中文表达:
return personality_result
{compress_content}
async def _create_identity(self, identity_detail: list) -> str:
"""使用LLM创建压缩版本的impression
"""
logger.info("正在构建身份.........")
if global_config.identity.compress_indentity:
identity_to_compress = []
if identity_detail:
identity_to_compress.append(f"身份背景: {''.join(identity_detail)}")
prompt = f"""请将以下身份信息进行简洁压缩,保留主要内容,用简练的中文表达:
{identity_to_compress}
要求:
1. 保持原意不变,尽量使用原文
@@ -416,20 +532,16 @@ class Individuality:
)
if response.strip():
compressed_parts.append(response.strip())
logger.info(f"精简人格侧面: {response.strip()}")
identity_result = response.strip()
logger.info(f"精简身份: {identity_result}")
else:
logger.error(f"使用LLM压缩人设时出错: {response}")
except Exception as e:
logger.error(f"使用LLM压缩人设时出错: {e}")
logger.error(f"使用LLM压缩身份时出错: {response}")
else:
identity_result = "".join(identity_detail)
return identity_result
result = "".join(compressed_parts)
# 更新short_impression字段
if result:
person_info_manager = get_person_info_manager()
await person_info_manager.update_one_field(self.bot_person_id, "short_impression", result)
logger.info("已将压缩人设更新到bot的short_impression中")
individuality = None

View File

@@ -1,5 +1,5 @@
[inner]
version = "2.27.0"
version = "2.28.0"
#----以下是给开发人员阅读的,如果你只是部署了麦麦,不需要阅读----
#如果你想要修改配置文件请在修改后将version的值进行变更
@@ -18,25 +18,29 @@ nickname = "麦麦" # 麦麦的昵称
alias_names = ["麦叠", "牢麦"] # 麦麦的别名
[personality]
personality_core = "是一个积极向上的女大学生" # 建议50字以内
# 建议50字以内,描述人格的核心特质
personality_core = "是一个积极向上的女大学生"
# 人格的细节可以描述人格的一些侧面条数任意不能为0不宜太多
personality_sides = [
"用一句话或几句话描述人格的一些细节",
"用一句话或几句话描述人格的一些细节",
"用一句话或几句话描述人格的一些细节",
"用一句话或几句话描述人格的一些侧面",
"用一句话或几句话描述人格的一些侧面",
"用一句话或几句话描述人格的一些侧面",
]
# 条数任意不能为0
# 身份特点
compress_personality = true # 是否压缩人格压缩后会精简人格信息节省token消耗并提高回复性能但是会丢失一些信息如果人设不长可以关闭
[identity]
#アイデンティティがない 生まれないらららら
[identity]
# 可以描述外貌,性别,身高,职业,属性等等描述,条数任意不能为0
identity_detail = [
"年龄为19岁",
"是女孩子",
"身高为160cm",
"有橙色的短发",
]
# 可以描述外貌,性别,身高,职业,属性等等描述
# 条数任意不能为0
compress_indentity = true # 是否压缩身份压缩后会精简身份信息节省token消耗并提高回复性能但是会丢失一些信息如果不长可以关闭
[expression]
# 表达方式