358 lines
14 KiB
Python
358 lines
14 KiB
Python
from typing import Optional
|
||
from .personality import Personality
|
||
from .identity import Identity
|
||
from .expression_style import PersonalityExpression
|
||
import random
|
||
import json
|
||
import os
|
||
import hashlib
|
||
from rich.traceback import install
|
||
from src.common.logger import get_logger
|
||
from src.person_info.person_info import get_person_info_manager
|
||
|
||
install(extra_lines=3)
|
||
|
||
logger = get_logger("individuality")
|
||
|
||
|
||
class Individuality:
|
||
"""个体特征管理类"""
|
||
|
||
def __init__(self):
|
||
# 正常初始化实例属性
|
||
self.personality: Optional[Personality] = None
|
||
self.identity: Optional[Identity] = None
|
||
self.express_style: PersonalityExpression = PersonalityExpression()
|
||
|
||
self.name = ""
|
||
self.bot_person_id = ""
|
||
self.meta_info_file_path = "data/personality/meta.json"
|
||
|
||
async def initialize(
|
||
self,
|
||
bot_nickname: str,
|
||
personality_core: str,
|
||
personality_sides: list,
|
||
identity_detail: list,
|
||
) -> None:
|
||
"""初始化个体特征
|
||
|
||
Args:
|
||
bot_nickname: 机器人昵称
|
||
personality_core: 人格核心特点
|
||
personality_sides: 人格侧面描述
|
||
identity_detail: 身份细节描述
|
||
"""
|
||
person_info_manager = get_person_info_manager()
|
||
self.bot_person_id = person_info_manager.get_person_id("system", "bot_id")
|
||
self.name = bot_nickname
|
||
|
||
# 检查配置变化，如果变化则清空
|
||
await self._check_config_and_clear_if_changed(bot_nickname, personality_core, personality_sides, identity_detail)
|
||
|
||
# 初始化人格
|
||
self.personality = Personality.initialize(
|
||
bot_nickname=bot_nickname, personality_core=personality_core, personality_sides=personality_sides
|
||
)
|
||
|
||
# 初始化身份
|
||
self.identity = Identity(identity_detail=identity_detail)
|
||
|
||
# 将所有人设写入impression
|
||
impression_parts = []
|
||
if personality_core:
|
||
impression_parts.append(f"核心人格: {personality_core}")
|
||
if personality_sides:
|
||
impression_parts.append(f"人格侧面: {'、'.join(personality_sides)}")
|
||
if identity_detail:
|
||
impression_parts.append(f"身份: {'、'.join(identity_detail)}")
|
||
|
||
impression_text = "。".join(impression_parts)
|
||
if impression_text:
|
||
impression_text += "。"
|
||
|
||
if impression_text:
|
||
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.info("已将完整人设更新到bot的impression中")
|
||
|
||
await self.express_style.extract_and_store_personality_expressions()
|
||
|
||
def to_dict(self) -> dict:
|
||
"""将个体特征转换为字典格式"""
|
||
return {
|
||
"personality": self.personality.to_dict() if self.personality else None,
|
||
"identity": self.identity.to_dict() if self.identity else None,
|
||
}
|
||
|
||
@classmethod
|
||
def from_dict(cls, data: dict) -> "Individuality":
|
||
"""从字典创建个体特征实例"""
|
||
instance = cls()
|
||
if data.get("personality"):
|
||
instance.personality = Personality.from_dict(data["personality"])
|
||
if data.get("identity"):
|
||
instance.identity = Identity.from_dict(data["identity"])
|
||
return instance
|
||
|
||
def get_personality_prompt(self, level: int, x_person: int = 2) -> str:
|
||
"""
|
||
获取人格特征的prompt
|
||
|
||
Args:
|
||
level (int): 详细程度 (1: 核心, 2: 核心+随机侧面, 3: 核心+所有侧面)
|
||
x_person (int, optional): 人称代词 (0: 无人称, 1: 我, 2: 你). 默认为 2.
|
||
|
||
Returns:
|
||
str: 生成的人格prompt字符串
|
||
"""
|
||
if x_person not in [0, 1, 2]:
|
||
return "无效的人称代词，请使用 0 (无人称), 1 (我) 或 2 (你)。"
|
||
if not self.personality:
|
||
return "人格特征尚未初始化。"
|
||
|
||
if x_person == 2:
|
||
p_pronoun = "你"
|
||
prompt_personality = f"{p_pronoun}{self.personality.personality_core}"
|
||
elif x_person == 1:
|
||
p_pronoun = "我"
|
||
prompt_personality = f"{p_pronoun}{self.personality.personality_core}"
|
||
else: # x_person == 0
|
||
# 对于无人称，直接描述核心特征
|
||
prompt_personality = f"{self.personality.personality_core}"
|
||
|
||
# 根据level添加人格侧面
|
||
if level >= 2 and self.personality.personality_sides:
|
||
personality_sides = list(self.personality.personality_sides)
|
||
random.shuffle(personality_sides)
|
||
if level == 2:
|
||
prompt_personality += f"，有时也会{personality_sides[0]}"
|
||
elif level == 3:
|
||
sides_str = "、".join(personality_sides)
|
||
prompt_personality += f"，有时也会{sides_str}"
|
||
prompt_personality += "。"
|
||
return prompt_personality
|
||
|
||
def get_identity_prompt(self, level: int, x_person: int = 2) -> str:
|
||
"""
|
||
获取身份特征的prompt
|
||
|
||
Args:
|
||
level (int): 详细程度 (1: 随机细节, 2: 所有细节, 3: 同2)
|
||
x_person (int, optional): 人称代词 (0: 无人称, 1: 我, 2: 你). 默认为 2.
|
||
|
||
Returns:
|
||
str: 生成的身份prompt字符串
|
||
"""
|
||
if x_person not in [0, 1, 2]:
|
||
return "无效的人称代词，请使用 0 (无人称), 1 (我) 或 2 (你)。"
|
||
if not self.identity:
|
||
return "身份特征尚未初始化。"
|
||
|
||
if x_person == 2:
|
||
i_pronoun = "你"
|
||
elif x_person == 1:
|
||
i_pronoun = "我"
|
||
else: # x_person == 0
|
||
i_pronoun = "" # 无人称
|
||
|
||
identity_parts = []
|
||
|
||
# 根据level添加身份细节
|
||
if level >= 1 and self.identity.identity_detail:
|
||
identity_detail = list(self.identity.identity_detail)
|
||
random.shuffle(identity_detail)
|
||
if level == 1:
|
||
identity_parts.append(f"{identity_detail[0]}")
|
||
elif level >= 2:
|
||
details_str = "、".join(identity_detail)
|
||
identity_parts.append(f"{details_str}")
|
||
|
||
if identity_parts:
|
||
details_str = "，".join(identity_parts)
|
||
if x_person in [1, 2]:
|
||
return f"{i_pronoun}，{details_str}。"
|
||
else: # x_person == 0
|
||
# 无人称时，直接返回细节，不加代词和开头的逗号
|
||
return f"{details_str}。"
|
||
else:
|
||
if x_person in [1, 2]:
|
||
return f"{i_pronoun}的身份信息不完整。"
|
||
else: # x_person == 0
|
||
return "身份信息不完整。"
|
||
|
||
def get_prompt(self, level: int, x_person: int = 2) -> str:
|
||
"""
|
||
获取合并的个体特征prompt
|
||
|
||
Args:
|
||
level (int): 详细程度 (1: 核心/随机细节, 2: 核心+随机侧面/全部细节, 3: 全部)
|
||
x_person (int, optional): 人称代词 (0: 无人称, 1: 我, 2: 你). 默认为 2.
|
||
|
||
Returns:
|
||
str: 生成的合并prompt字符串
|
||
"""
|
||
if x_person not in [0, 1, 2]:
|
||
return "无效的人称代词，请使用 0 (无人称), 1 (我) 或 2 (你)。"
|
||
|
||
if not self.personality or not self.identity:
|
||
return "个体特征尚未完全初始化。"
|
||
|
||
# 调用新的独立方法
|
||
prompt_personality = self.get_personality_prompt(level, x_person)
|
||
prompt_identity = self.get_identity_prompt(level, x_person)
|
||
|
||
# 移除可能存在的错误信息，只合并有效的 prompt
|
||
valid_prompts = []
|
||
if "尚未初始化" not in prompt_personality and "无效的人称" not in prompt_personality:
|
||
valid_prompts.append(prompt_personality)
|
||
if (
|
||
"尚未初始化" not in prompt_identity
|
||
and "无效的人称" not in prompt_identity
|
||
and "信息不完整" not in prompt_identity
|
||
):
|
||
# 从身份 prompt 中移除代词和句号，以便更好地合并
|
||
identity_content = prompt_identity
|
||
if x_person == 2 and identity_content.startswith("你，"):
|
||
identity_content = identity_content[2:]
|
||
elif x_person == 1 and identity_content.startswith("我，"):
|
||
identity_content = identity_content[2:]
|
||
# 对于 x_person == 0，身份提示不带前缀，无需移除
|
||
|
||
if identity_content.endswith("。"):
|
||
identity_content = identity_content[:-1]
|
||
valid_prompts.append(identity_content)
|
||
|
||
# --- 合并 Prompt ---
|
||
final_prompt = " ".join(valid_prompts)
|
||
|
||
return final_prompt.strip()
|
||
|
||
def get_traits(self, factor):
|
||
"""
|
||
获取个体特征的特质
|
||
"""
|
||
if factor == "openness":
|
||
return self.personality.openness
|
||
elif factor == "conscientiousness":
|
||
return self.personality.conscientiousness
|
||
elif factor == "extraversion":
|
||
return self.personality.extraversion
|
||
elif factor == "agreeableness":
|
||
return self.personality.agreeableness
|
||
elif factor == "neuroticism":
|
||
return self.personality.neuroticism
|
||
return None
|
||
|
||
def _get_config_hash(self, bot_nickname: str, personality_core: str, personality_sides: list, identity_detail: list) -> str:
|
||
"""获取当前personality和identity配置的哈希值"""
|
||
config_data = {
|
||
"nickname": bot_nickname,
|
||
"personality_core": personality_core,
|
||
"personality_sides": sorted(personality_sides),
|
||
"identity_detail": sorted(identity_detail)
|
||
}
|
||
config_str = json.dumps(config_data, sort_keys=True)
|
||
return hashlib.md5(config_str.encode("utf-8")).hexdigest()
|
||
|
||
async def _check_config_and_clear_if_changed(
|
||
self, bot_nickname: str, personality_core: str, personality_sides: list, identity_detail: list
|
||
):
|
||
"""检查配置是否发生变化，如果变化则清空info_list"""
|
||
person_info_manager = get_person_info_manager()
|
||
current_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")
|
||
|
||
if current_hash != stored_hash:
|
||
logger.info("检测到人格配置发生变化，将清空原有的关键词缓存。")
|
||
|
||
# 清空数据库中的info_list
|
||
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, "info_list", [], data=update_data)
|
||
|
||
# 更新元信息文件，重置计数器
|
||
new_meta_info = {"config_hash": current_hash}
|
||
self._save_meta_info(new_meta_info)
|
||
|
||
def _load_meta_info(self) -> dict:
|
||
"""从JSON文件中加载元信息"""
|
||
if os.path.exists(self.meta_info_file_path):
|
||
try:
|
||
with open(self.meta_info_file_path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except (json.JSONDecodeError, IOError) as e:
|
||
logger.error(f"读取meta_info文件失败: {e}, 将创建新文件。")
|
||
return {}
|
||
return {}
|
||
|
||
def _save_meta_info(self, meta_info: dict):
|
||
"""将元信息保存到JSON文件"""
|
||
try:
|
||
os.makedirs(os.path.dirname(self.meta_info_file_path), exist_ok=True)
|
||
with open(self.meta_info_file_path, "w", encoding="utf-8") as f:
|
||
json.dump(meta_info, f, ensure_ascii=False, indent=2)
|
||
except IOError as e:
|
||
logger.error(f"保存meta_info文件失败: {e}")
|
||
|
||
async def get_keyword_info(self, keyword: str) -> str:
|
||
"""获取指定关键词的信息
|
||
|
||
Args:
|
||
keyword: 关键词
|
||
|
||
Returns:
|
||
str: 随机选择的一条信息，如果没有则返回空字符串
|
||
"""
|
||
person_info_manager = get_person_info_manager()
|
||
info_list_json = await person_info_manager.get_value(self.bot_person_id, "info_list")
|
||
if info_list_json:
|
||
try:
|
||
# get_value might return a pre-deserialized list if it comes from a cache,
|
||
# or a JSON string if it comes from DB.
|
||
info_list = json.loads(info_list_json) if isinstance(info_list_json, str) else info_list_json
|
||
|
||
for item in info_list:
|
||
if isinstance(item, dict) and item.get("info_type") == keyword:
|
||
return item.get("info_content", "")
|
||
except (json.JSONDecodeError, TypeError):
|
||
logger.error(f"解析info_list失败: {info_list_json}")
|
||
return ""
|
||
return ""
|
||
|
||
async def get_all_keywords(self) -> list:
|
||
"""获取所有已缓存的关键词列表"""
|
||
person_info_manager = get_person_info_manager()
|
||
info_list_json = await person_info_manager.get_value(self.bot_person_id, "info_list")
|
||
keywords = []
|
||
if info_list_json:
|
||
try:
|
||
info_list = json.loads(info_list_json) if isinstance(info_list_json, str) else info_list_json
|
||
for item in info_list:
|
||
if isinstance(item, dict) and "info_type" in item:
|
||
keywords.append(item["info_type"])
|
||
except (json.JSONDecodeError, TypeError):
|
||
logger.error(f"解析info_list失败: {info_list_json}")
|
||
return keywords
|
||
|
||
|
||
individuality = None
|
||
|
||
|
||
def get_individuality():
|
||
global individuality
|
||
if individuality is None:
|
||
individuality = Individuality()
|
||
return individuality
|