gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix：修复人格表达生成太固定的问题

Browse Source
This commit is contained in:
SengokuCola
2025-06-03 00:45:54 +08:00
parent 75cb982de8
commit 14893ede27
3 changed files with 344 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
src/individuality/expression_style.py
View File

@@ -6,6 +6,7 @@ from src.chat.utils.prompt_builder import Prompt, global_prompt_manager
from typing import List, Tuple
import os
import json
from datetime import datetime
logger = get_logger("expressor")
@@ -45,11 +46,30 @@ class PersonalityExpression:
if os.path.exists(self.meta_file_path):
try:
with open(self.meta_file_path, "r", encoding="utf-8") as f:
return json.load(f)
meta_data = json.load(f)
# 检查是否有last_update_time字段
if "last_update_time" not in meta_data:
logger.warning(f"{self.meta_file_path} 中缺少last_update_time字段，将重新开始。")
# 清空并重写元数据文件
self._write_meta_data({"last_style_text": None, "count": 0, "last_update_time": None})
# 清空并重写表达文件
if os.path.exists(self.expressions_file_path):
with open(self.expressions_file_path, "w", encoding="utf-8") as f:
json.dump([], f, ensure_ascii=False, indent=2)
logger.debug(f"已清空表达文件: {self.expressions_file_path}")
return {"last_style_text": None, "count": 0, "last_update_time": None}
return meta_data
except json.JSONDecodeError:
logger.warning(f"无法解析 {self.meta_file_path} 中的JSON数据，将重新开始。")
return {"last_style_text": None, "count": 0}
return {"last_style_text": None, "count": 0}
# 清空并重写元数据文件
self._write_meta_data({"last_style_text": None, "count": 0, "last_update_time": None})
# 清空并重写表达文件
if os.path.exists(self.expressions_file_path):
with open(self.expressions_file_path, "w", encoding="utf-8") as f:
json.dump([], f, ensure_ascii=False, indent=2)
logger.debug(f"已清空表达文件: {self.expressions_file_path}")
return {"last_style_text": None, "count": 0, "last_update_time": None}
return {"last_style_text": None, "count": 0, "last_update_time": None}
def _write_meta_data(self, data):
os.makedirs(os.path.dirname(self.meta_file_path), exist_ok=True)
@@ -84,7 +104,7 @@ class PersonalityExpression:
if count >= self.max_calculations:
logger.debug(f"对于风格 '{current_style_text}' 已达到最大计算次数 ({self.max_calculations})。跳过提取。")
# 即使跳过，也更新元数据以反映当前风格已被识别且计数已满
self._write_meta_data({"last_style_text": current_style_text, "count": count})
self._write_meta_data({"last_style_text": current_style_text, "count": count, "last_update_time": meta_data.get("last_update_time")})
return
# 构建prompt
@@ -99,30 +119,63 @@ class PersonalityExpression:
except Exception as e:
logger.error(f"个性表达方式提取失败: {e}")
# 如果提取失败，保存当前的风格和未增加的计数
self._write_meta_data({"last_style_text": current_style_text, "count": count})
self._write_meta_data({"last_style_text": current_style_text, "count": count, "last_update_time": meta_data.get("last_update_time")})
return
logger.info(f"个性表达方式提取response: {response}")
# chat_id用personality
expressions = self.parse_expression_response(response, "personality")
# 转为dict并count=100
result = []
for _, situation, style in expressions:
result.append({"situation": situation, "style": style, "count": 100})
# 超过50条时随机删除多余的，只保留50条
if len(result) > 50:
remove_count = len(result) - 50
remove_indices = set(random.sample(range(len(result)), remove_count))
result = [item for idx, item in enumerate(result) if idx not in remove_indices]
if response != "":
expressions = self.parse_expression_response(response, "personality")
# 读取已有的表达方式
existing_expressions = []
if os.path.exists(self.expressions_file_path):
try:
with open(self.expressions_file_path, "r", encoding="utf-8") as f:
existing_expressions = json.load(f)
except (json.JSONDecodeError, FileNotFoundError):
logger.warning(f"无法读取或解析 {self.expressions_file_path}，将创建新的表达文件。")
# 创建新的表达方式
new_expressions = []
for _, situation, style in expressions:
new_expressions.append({"situation": situation, "style": style, "count": 1})
# 合并表达方式，如果situation和style相同则累加count
merged_expressions = existing_expressions.copy()
for new_expr in new_expressions:
found = False
for existing_expr in merged_expressions:
if (existing_expr["situation"] == new_expr["situation"] and
existing_expr["style"] == new_expr["style"]):
existing_expr["count"] += new_expr["count"]
found = True
break
if not found:
merged_expressions.append(new_expr)
# 超过50条时随机删除多余的，只保留50条
if len(merged_expressions) > 50:
remove_count = len(merged_expressions) - 50
remove_indices = set(random.sample(range(len(merged_expressions)), remove_count))
merged_expressions = [item for idx, item in enumerate(merged_expressions) if idx not in remove_indices]
with open(self.expressions_file_path, "w", encoding="utf-8") as f:
json.dump(result, f, ensure_ascii=False, indent=2)
logger.info(f"已写入{len(result)}条表达到{self.expressions_file_path}")
with open(self.expressions_file_path, "w", encoding="utf-8") as f:
json.dump(merged_expressions, f, ensure_ascii=False, indent=2)
logger.info(f"已写入{len(merged_expressions)}条表达到{self.expressions_file_path}")
# 成功提取后更新元数据
count += 1
self._write_meta_data({"last_style_text": current_style_text, "count": count})
logger.info(f"成功处理。风格 '{current_style_text}' 的计数现在是 {count}。")
# 成功提取后更新元数据
count += 1
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self._write_meta_data({
"last_style_text": current_style_text,
"count": count,
"last_update_time": current_time
})
logger.info(f"成功处理。风格 '{current_style_text}' 的计数现在是 {count}，最后更新时间：{current_time}。")
else:
logger.warning(f"个性表达方式提取失败，模型返回空内容: {response}")
def parse_expression_response(self, response: str, chat_id: str) -> List[Tuple[str, str, str]]:
"""