remove:移除了冗余代码

This commit is contained in:
SengokuCola
2025-07-24 03:33:36 +08:00
parent b839f8ba6c
commit 677b17754a
5 changed files with 177 additions and 253 deletions

View File

@@ -2,6 +2,7 @@ import ast
import json
import os
import hashlib
import time
from src.common.logger import get_logger
from src.config.config import global_config
@@ -9,8 +10,6 @@ from src.llm_models.utils_model import LLMRequest
from src.person_info.person_info import get_person_info_manager
from rich.traceback import install
from .personality import Personality
install(extra_lines=3)
logger = get_logger("individuality")
@@ -20,12 +19,10 @@ class Individuality:
"""个体特征管理类"""
def __init__(self):
# 正常初始化实例属性
self.personality: Personality = None # type: ignore
self.name = ""
self.bot_person_id = ""
self.meta_info_file_path = "data/personality/meta.json"
self.personality_data_file_path = "data/personality/personality_data.json"
self.model = LLMRequest(
model=global_config.model.utils,
@@ -33,14 +30,7 @@ class Individuality:
)
async def initialize(self) -> None:
"""初始化个体特征
Args:
bot_nickname: 机器人昵称
personality_core: 人格核心特点
personality_side: 人格侧面描述
identity: 身份细节描述
"""
"""初始化个体特征"""
bot_nickname = global_config.bot.nickname
personality_core = global_config.personality.personality_core
personality_side = global_config.personality.personality_side
@@ -56,124 +46,58 @@ class Individuality:
bot_nickname, personality_core, personality_side, identity
)
# 初始化人格(现在包含身份)
self.personality = Personality.initialize(
bot_nickname=bot_nickname,
personality_core=personality_core,
personality_side=personality_side,
identity=identity,
compress_personality=global_config.personality.compress_personality,
compress_identity=global_config.personality.compress_identity,
)
logger.info("正在构建人设信息")
logger.info("正在将所有人设写入impression")
# 将所有人设写入impression
impression_parts = []
if personality_core:
impression_parts.append(f"核心人格: {personality_core}")
if personality_side:
impression_parts.append(f"人格侧面: {personality_side}")
if identity:
impression_parts.append(f"身份: {identity}")
logger.info(f"impression_parts: {impression_parts}")
# 如果配置有变化,重新生成压缩版本
if personality_changed or identity_changed:
logger.info("检测到配置变化,重新生成压缩版本")
personality_result = await self._create_personality(personality_core, personality_side)
identity_result = await self._create_identity(identity)
else:
logger.info("配置未变化,使用缓存版本")
# 从文件中获取已有的结果
personality_result, identity_result = self._get_personality_from_file()
if not personality_result or not identity_result:
logger.info("未找到有效缓存,重新生成")
personality_result = await self._create_personality(personality_core, personality_side)
identity_result = await self._create_identity(identity)
impression_text = "".join(impression_parts)
if impression_text:
impression_text += ""
# 保存到文件
if personality_result and identity_result:
self._save_personality_to_file(personality_result, identity_result)
logger.info("已将人设构建并保存到文件")
else:
logger.error("人设构建失败")
if impression_text:
# 如果任何一个发生变化都需要清空数据库中的info_list因为这影响整体人设
if personality_changed or identity_changed:
logger.info("将清空数据库中原有的关键词缓存")
update_data = {
"platform": "system",
"user_id": "bot_id",
"person_name": self.name,
"nickname": self.name,
}
await person_info_manager.update_one_field(
self.bot_person_id, "impression", impression_text, data=update_data
)
logger.debug("已将完整人设更新到bot的impression中")
# 根据变化情况决定是否重新创建
personality_result = None
identity_result = None
if personality_changed:
logger.info("检测到人格配置变化,重新生成压缩版本")
personality_result = await self._create_personality(personality_core, personality_side)
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 = ast.literal_eval(existing_short_impression) # type: ignore
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_side)
else:
logger.info("未找到现有的人格缓存,重新生成")
personality_result = await self._create_personality(personality_core, personality_side)
if identity_changed:
logger.info("检测到身份配置变化,重新生成压缩版本")
identity_result = await self._create_identity(identity)
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 = ast.literal_eval(existing_short_impression) # type: ignore
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)
else:
logger.info("未找到现有的身份缓存,重新生成")
identity_result = await self._create_identity(identity)
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("人设构建失败")
await person_info_manager.update_one_field(self.bot_person_id, "info_list", [], data=update_data)
async def get_personality_block(self) -> str:
person_info_manager = get_person_info_manager()
bot_person_id = person_info_manager.get_person_id("system", "bot_id")
bot_name = global_config.bot.nickname
if global_config.bot.alias_names:
bot_nickname = f",也有人叫你{','.join(global_config.bot.alias_names)}"
else:
bot_nickname = ""
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
personality, identity = self._get_personality_from_file()
# 确保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 = f"{personality}{identity}"
return f"你的名字是{bot_name}{bot_nickname},你{prompt_personality}"
if not personality or not identity:
logger.warning(f"personality或identity为空: {personality}, {identity}, 使用默认值")
personality = "友好活泼"
identity = "人类"
prompt_personality = f"{personality}\n{identity}"
return f"你的名字是{bot_name}{bot_nickname},你{prompt_personality}"
def _get_config_hash(
self, bot_nickname: str, personality_core: str, personality_side: str, identity: str
@@ -188,7 +112,7 @@ class Individuality:
"nickname": bot_nickname,
"personality_core": personality_core,
"personality_side": personality_side,
"compress_personality": self.personality.compress_personality if self.personality else True,
"compress_personality": global_config.personality.compress_personality,
}
personality_str = json.dumps(personality_config, sort_keys=True)
personality_hash = hashlib.md5(personality_str.encode("utf-8")).hexdigest()
@@ -196,7 +120,7 @@ class Individuality:
# 身份配置哈希
identity_config = {
"identity": identity,
"compress_identity": self.personality.compress_identity if self.personality else True,
"compress_identity": global_config.personality.compress_identity,
}
identity_str = json.dumps(identity_config, sort_keys=True)
identity_hash = hashlib.md5(identity_str.encode("utf-8")).hexdigest()
@@ -269,6 +193,53 @@ class Individuality:
except IOError as e:
logger.error(f"保存meta_info文件失败: {e}")
def _load_personality_data(self) -> dict:
"""从JSON文件中加载personality数据"""
if os.path.exists(self.personality_data_file_path):
try:
with open(self.personality_data_file_path, "r", encoding="utf-8") as f:
return json.load(f)
except (json.JSONDecodeError, IOError) as e:
logger.error(f"读取personality_data文件失败: {e}, 将创建新文件。")
return {}
return {}
def _save_personality_data(self, personality_data: dict):
"""将personality数据保存到JSON文件"""
try:
os.makedirs(os.path.dirname(self.personality_data_file_path), exist_ok=True)
with open(self.personality_data_file_path, "w", encoding="utf-8") as f:
json.dump(personality_data, f, ensure_ascii=False, indent=2)
logger.debug(f"已保存personality数据到文件: {self.personality_data_file_path}")
except IOError as e:
logger.error(f"保存personality_data文件失败: {e}")
def _get_personality_from_file(self) -> tuple[str, str]:
"""从文件获取personality数据
Returns:
tuple: (personality, identity)
"""
personality_data = self._load_personality_data()
personality = personality_data.get("personality", "友好活泼")
identity = personality_data.get("identity", "人类")
return personality, identity
def _save_personality_to_file(self, personality: str, identity: str):
"""保存personality数据到文件
Args:
personality: 压缩后的人格描述
identity: 压缩后的身份描述
"""
personality_data = {
"personality": personality,
"identity": identity,
"bot_nickname": self.name,
"last_updated": int(time.time())
}
self._save_personality_data(personality_data)
async def _create_personality(self, personality_core: str, personality_side: str) -> str:
# sourcery skip: merge-list-append, move-assign
"""使用LLM创建压缩版本的impression
@@ -288,7 +259,7 @@ class Individuality:
personality_parts.append(f"{personality_core}")
# 准备需要压缩的内容
if self.personality.compress_personality:
if global_config.personality.compress_personality:
personality_to_compress = f"人格特质: {personality_side}"
prompt = f"""请将以下人格信息进行简洁压缩,保留主要内容,用简练的中文表达:
@@ -323,7 +294,7 @@ class Individuality:
"""使用LLM创建压缩版本的impression"""
logger.info("正在构建身份.........")
if self.personality.compress_identity:
if global_config.personality.compress_identity:
identity_to_compress = f"身份背景: {identity}"
prompt = f"""请将以下身份信息进行简洁压缩,保留主要内容,用简练的中文表达:

View File

@@ -1,90 +0,0 @@
from dataclasses import dataclass
from typing import Dict, Optional
@dataclass
class Personality:
"""人格特质类"""
bot_nickname: str # 机器人昵称
personality_core: str # 人格核心特点
personality_side: str # 人格侧面描述
identity: Optional[str] # 身份细节描述
compress_personality: bool # 是否压缩人格
compress_identity: bool # 是否压缩身份
_instance = None
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, personality_core: str = "", personality_side: str = "", identity: Optional[str] = None):
self.personality_core = personality_core
self.personality_side = personality_side
self.identity = identity
self.compress_personality = True
self.compress_identity = True
@classmethod
def get_instance(cls) -> "Personality":
"""获取Personality单例实例
Returns:
Personality: 单例实例
"""
if cls._instance is None:
cls._instance = cls()
return cls._instance
@classmethod
def initialize(
cls,
bot_nickname: str,
personality_core: str,
personality_side: str,
identity: Optional[str] = None,
compress_personality: bool = True,
compress_identity: bool = True,
) -> "Personality":
"""初始化人格特质
Args:
bot_nickname: 机器人昵称
personality_core: 人格核心特点
personality_side: 人格侧面描述
identity: 身份细节描述
compress_personality: 是否压缩人格
compress_identity: 是否压缩身份
Returns:
Personality: 初始化后的人格特质实例
"""
instance = cls.get_instance()
instance.bot_nickname = bot_nickname
instance.personality_core = personality_core
instance.personality_side = personality_side
instance.identity = identity
instance.compress_personality = compress_personality
instance.compress_identity = compress_identity
return instance
def to_dict(self) -> Dict:
"""将人格特质转换为字典格式"""
return {
"bot_nickname": self.bot_nickname,
"personality_core": self.personality_core,
"personality_side": self.personality_side,
"identity": self.identity,
"compress_personality": self.compress_personality,
"compress_identity": self.compress_identity,
}
@classmethod
def from_dict(cls, data: Dict) -> "Personality":
"""从字典创建人格特质实例"""
instance = cls.get_instance()
for key, value in data.items():
setattr(instance, key, value)
return instance