433 lines
16 KiB
Python
433 lines
16 KiB
Python
# file_manager.py
|
||
|
||
import json
|
||
import os
|
||
import uuid
|
||
import shutil # 用于删除目录及其内容
|
||
import logging
|
||
from config import get_config # 导入config模块的获取配置函数
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# --- 辅助函数：JSON 和 JSONL 文件操作 ---
|
||
|
||
|
||
def _load_json_file(file_path, default_content=None):
|
||
"""
|
||
加载JSON文件。如果文件不存在，则创建并写入默认内容（如果提供）。
|
||
:param file_path: JSON文件路径。
|
||
:param default_content: 文件不存在时要写入的默认Python对象。
|
||
:return: 文件内容（Python对象）或None（如果加载失败）。
|
||
"""
|
||
if not os.path.exists(file_path):
|
||
os.makedirs(os.path.dirname(file_path), exist_ok=True) # 确保目录存在
|
||
if default_content is not None:
|
||
logger.info(f"文件不存在，创建并写入默认内容到：{file_path}")
|
||
_save_json_file(file_path, default_content)
|
||
else:
|
||
logger.warning(f"尝试加载的JSON文件不存在且未提供默认内容：{file_path}")
|
||
return {} # 返回空字典，以提高健壮性
|
||
|
||
try:
|
||
with open(file_path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except json.JSONDecodeError as e:
|
||
logger.error(f"JSON文件解码失败：{file_path} - {e}")
|
||
return None
|
||
except IOError as e:
|
||
logger.error(f"读取JSON文件失败：{file_path} - {e}")
|
||
return None
|
||
|
||
|
||
def _save_json_file(file_path, data):
|
||
"""
|
||
保存Python对象到JSON文件。
|
||
:param file_path: JSON文件路径。
|
||
:param data: 要保存的Python对象。
|
||
"""
|
||
os.makedirs(os.path.dirname(file_path), exist_ok=True) # 确保目录存在
|
||
try:
|
||
with open(file_path, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
logger.debug(f"JSON文件已保存：{file_path}")
|
||
return True
|
||
except IOError as e:
|
||
logger.error(f"保存JSON文件失败：{file_path} - {e}")
|
||
return False
|
||
|
||
|
||
def _append_jsonl_file(file_path, data):
|
||
"""
|
||
追加一条JSON记录到JSONL文件。
|
||
:param file_path: JSONL文件路径。
|
||
:param data: 要追加的Python对象。
|
||
"""
|
||
os.makedirs(os.path.dirname(file_path), exist_ok=True) # 确保目录存在
|
||
try:
|
||
with open(file_path, "a", encoding="utf-8") as f:
|
||
f.write(json.dumps(data, ensure_ascii=False) + "\n")
|
||
logger.debug(f"JSONL记录已追加到：{file_path}")
|
||
return True
|
||
except IOError as e:
|
||
logger.error(f"追加JSONL文件失败：{file_path} - {e}")
|
||
return False
|
||
|
||
|
||
def _read_jsonl_file(file_path, limit=None):
|
||
"""
|
||
读取JSONL文件的所有记录。
|
||
:param file_path: JSONL文件路径。
|
||
:param limit: 可选，限制读取的最后N条记录。
|
||
:return: 包含所有记录的列表。
|
||
"""
|
||
if not os.path.exists(file_path):
|
||
return []
|
||
|
||
records = []
|
||
try:
|
||
with open(file_path, "r", encoding="utf-8") as f:
|
||
lines = f.readlines()
|
||
if limit:
|
||
lines = lines[-limit:] # 只读取最后N行
|
||
for line in lines:
|
||
try:
|
||
records.append(json.loads(line.strip()))
|
||
except json.JSONDecodeError as e:
|
||
logger.warning(
|
||
f"跳过JSONL文件中的损坏行：{file_path} - {line.strip()} - {e}"
|
||
)
|
||
return records
|
||
except IOError as e:
|
||
logger.error(f"读取JSONL文件失败：{file_path} - {e}")
|
||
return []
|
||
|
||
|
||
# --- 默认文件内容 ---
|
||
|
||
# 默认特征文件内容
|
||
DEFAULT_FEATURE_CONTENT = {
|
||
"角色名称": "通用AI助手",
|
||
"角色描述": "你是一个有帮助的AI助手，旨在提供清晰、准确、友好的回答。你没有预设的特定角色或个性，能够灵活适应各种对话主题。",
|
||
"心理特征": {
|
||
"核心哲学与世界观": {"描述": "以提供信息、解决问题为核心，保持中立客观。"},
|
||
"价值观与道德观": {"描述": "遵循伦理原则，避免偏见，确保信息准确。"},
|
||
"决策风格": {"描述": "基于逻辑和可用信息进行决策。"},
|
||
"情绪反应与表达": {"描述": "保持平静、专业、不带个人情感。"},
|
||
"人际互动与关系处理": {"描述": "以帮助用户为目标，保持礼貌和尊重。"},
|
||
"动机与目标": {"描述": "提供有价值的信息和解决方案。"},
|
||
"自我认知": {"描述": "认知自己是AI，不具备人类情感和意识。"},
|
||
"应对机制": {"描述": "遇到未知问题时，会请求更多信息或承认能力限制。"},
|
||
},
|
||
"语言特征": {
|
||
"词汇与措辞": {"描述": "清晰、简洁、准确，避免模糊不清的表达。"},
|
||
"句式结构与复杂度": {"描述": "常用清晰直接的句式，逻辑性强。"},
|
||
"语气与风格": {"描述": "专业、友善、乐于助人。"},
|
||
"修辞手法与模式": {"描述": "多使用直接陈述、解释说明。"},
|
||
"互动模式": {"描述": "响应式，根据用户输入提供信息或引导对话。"},
|
||
},
|
||
"重要人际关系": [],
|
||
"角色弧线总结": "作为一个持续学习和改进的AI，不断提升服务质量。",
|
||
}
|
||
|
||
# 默认记忆文件内容
|
||
DEFAULT_MEMORY_CONTENT = {
|
||
"long_term_facts": ["此记忆集为默认记忆，未包含特定用户或AI的长期事实。"],
|
||
"short_term_events": [],
|
||
}
|
||
|
||
# --- 路径管理函数 ---
|
||
|
||
|
||
def _get_dir_path(base_dir_config_key):
|
||
"""根据配置获取目录路径"""
|
||
config = get_config()
|
||
return config["Application"][base_dir_config_key]
|
||
|
||
|
||
def get_features_dir():
|
||
"""获取特征文件根目录"""
|
||
return _get_dir_path("FEATURES_DIR")
|
||
|
||
|
||
def get_memories_dir():
|
||
"""获取记忆文件根目录"""
|
||
return _get_dir_path("MEMORIES_DIR")
|
||
|
||
|
||
def get_chat_logs_dir():
|
||
"""获取聊天记录文件根目录"""
|
||
return _get_dir_path("CHAT_LOGS_DIR")
|
||
|
||
|
||
def get_role_features_file_path(role_id):
|
||
"""获取指定角色ID的特征文件路径"""
|
||
return os.path.join(get_features_dir(), role_id, "features.json")
|
||
|
||
|
||
def get_memory_data_file_path(memory_id):
|
||
"""获取指定记忆ID的记忆数据文件路径"""
|
||
return os.path.join(get_memories_dir(), memory_id, "memory.json")
|
||
|
||
|
||
def get_chat_log_file_path(chat_log_id):
|
||
"""获取指定聊天记录ID的聊天记录文件路径"""
|
||
# 聊天记录ID通常与角色ID和记忆ID关联，或者直接就是session ID
|
||
# 这里简化为直接以chat_log_id为文件名
|
||
return os.path.join(get_chat_logs_dir(), f"{chat_log_id}.jsonl")
|
||
|
||
|
||
# --- 角色和记忆管理 ---
|
||
|
||
|
||
def list_roles():
|
||
"""
|
||
列出所有可用的角色ID和名称。
|
||
角色名称从其features.json中获取。
|
||
:return: 列表，每个元素为 {"id": "role_id", "name": "角色名称"}
|
||
"""
|
||
roles = []
|
||
features_dir = get_features_dir()
|
||
if not os.path.exists(features_dir):
|
||
os.makedirs(features_dir) # 确保目录存在
|
||
return []
|
||
|
||
for role_id in os.listdir(features_dir):
|
||
role_path = os.path.join(features_dir, role_id)
|
||
features_file = get_role_features_file_path(role_id)
|
||
if os.path.isdir(role_path) and os.path.exists(features_file):
|
||
features = _load_json_file(features_file)
|
||
if features:
|
||
roles.append({"id": role_id, "name": features.get("角色名称", role_id)})
|
||
else:
|
||
logger.warning(
|
||
f"无法加载角色 {role_id} 的 features.json，可能文件损坏。"
|
||
)
|
||
else:
|
||
logger.warning(f"目录 {role_path} 或文件 {features_file} 不完整，跳过。")
|
||
return roles
|
||
|
||
|
||
def create_role(role_id, role_name=None):
|
||
"""
|
||
创建一个新的角色。
|
||
:param role_id: 新角色的唯一ID。
|
||
:param role_name: 新角色的名称，如果未提供，则默认为role_id。
|
||
:return: True如果成功，False如果失败或ID已存在。
|
||
"""
|
||
role_path = os.path.join(get_features_dir(), role_id)
|
||
if os.path.exists(role_path):
|
||
logger.warning(f"角色ID '{role_id}' 已存在，无法创建。")
|
||
return False
|
||
|
||
os.makedirs(role_path)
|
||
features_file = get_role_features_file_path(role_id)
|
||
|
||
default_content = DEFAULT_FEATURE_CONTENT.copy()
|
||
default_content["角色名称"] = role_name if role_name else role_id
|
||
|
||
success = _save_json_file(features_file, default_content)
|
||
if success:
|
||
logger.info(f"角色 '{role_id}' 创建成功。")
|
||
return success
|
||
|
||
|
||
def delete_role(role_id):
|
||
"""
|
||
删除一个角色及其所有相关文件。
|
||
:param role_id: 要删除的角色ID。
|
||
:return: True如果成功，False如果失败。
|
||
"""
|
||
role_path = os.path.join(get_features_dir(), role_id)
|
||
if not os.path.exists(role_path):
|
||
logger.warning(f"角色ID '{role_id}' 不存在，无法删除。")
|
||
return False
|
||
|
||
try:
|
||
shutil.rmtree(role_path)
|
||
logger.info(f"角色 '{role_id}' 已成功删除。")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"删除角色 '{role_id}' 失败: {e}")
|
||
return False
|
||
|
||
|
||
def role_exists(role_id):
|
||
"""
|
||
检查指定角色ID的角色是否存在。
|
||
:param role_id: 角色ID。
|
||
:return: True如果存在，False如果不存在。
|
||
"""
|
||
return os.path.exists(get_role_features_file_path(role_id))
|
||
|
||
|
||
def list_memories():
|
||
"""
|
||
列出所有可用的记忆ID和名称。
|
||
记忆名称从其memory.json中获取（如果模型能生成name字段，否则使用ID）。
|
||
:return: 列表，每个元素为 {"id": "memory_id", "name": "记忆名称"}
|
||
"""
|
||
memories = []
|
||
memories_dir = get_memories_dir()
|
||
if not os.path.exists(memories_dir):
|
||
os.makedirs(memories_dir) # 确保目录存在
|
||
return []
|
||
|
||
for memory_id in os.listdir(memories_dir):
|
||
memory_path = os.path.join(memories_dir, memory_id)
|
||
memory_file = get_memory_data_file_path(memory_id)
|
||
if os.path.isdir(memory_path) and os.path.exists(memory_file):
|
||
memory_data = _load_json_file(memory_file)
|
||
# 确保 memory_data 不是 None 且不是空字典
|
||
if memory_data is not None and memory_data != {}:
|
||
# 假设 memory.json 中可能有一个 'name' 字段来标识记忆集
|
||
memories.append(
|
||
{"id": memory_id, "name": memory_data.get("name", memory_id)}
|
||
)
|
||
else:
|
||
logger.warning(
|
||
f"无法加载记忆 {memory_id} 的 memory.json，可能文件损或为空，或内容为空。"
|
||
)
|
||
else:
|
||
logger.warning(f"目录 {memory_path} 或文件 {memory_file} 不完整，跳过。")
|
||
return memories
|
||
|
||
|
||
def create_memory(memory_id, memory_name=None):
|
||
"""
|
||
创建一个新的记忆集。
|
||
:param memory_id: 新记忆集的唯一ID。
|
||
:param memory_name: 新记忆集的名称，如果未提供，则默认为memory_id。
|
||
:return: True如果成功，False如果失败或ID已存在。
|
||
"""
|
||
memory_path = os.path.join(get_memories_dir(), memory_id)
|
||
if os.path.exists(memory_path):
|
||
logger.warning(f"记忆ID '{memory_id}' 已存在，无法创建。")
|
||
return False
|
||
|
||
os.makedirs(memory_path)
|
||
memory_file = get_memory_data_file_path(memory_id)
|
||
|
||
default_content = DEFAULT_MEMORY_CONTENT.copy()
|
||
if memory_name:
|
||
default_content["name"] = memory_name # 记忆集本身可以有一个名称
|
||
# 更新 long_term_facts 中的默认描述，包含记忆名称
|
||
if default_content["long_term_facts"]:
|
||
default_content["long_term_facts"][
|
||
0
|
||
] = f"此记忆集名为 '{memory_name}'，目前未包含特定用户或AI的长期事实。"
|
||
else:
|
||
default_content["long_term_facts"].append(
|
||
f"此记忆集名为 '{memory_name}'，目前未包含特定用户或AI的长期事实。"
|
||
)
|
||
|
||
success = _save_json_file(memory_file, default_content)
|
||
if success:
|
||
logger.info(f"记忆集 '{memory_id}' 创建成功。")
|
||
return success
|
||
|
||
|
||
def delete_memory(memory_id):
|
||
"""
|
||
删除一个记忆集及其所有相关文件。
|
||
:param memory_id: 要删除的记忆ID。
|
||
:return: True如果成功，False如果失败。
|
||
"""
|
||
memory_path = os.path.join(get_memories_dir(), memory_id)
|
||
if not os.path.exists(memory_path):
|
||
logger.warning(f"记忆ID '{memory_id}' 不存在，无法删除。")
|
||
return False
|
||
|
||
try:
|
||
shutil.rmtree(memory_path)
|
||
logger.info(f"记忆集 '{memory_id}' 已成功删除。")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"删除记忆集 '{memory_id}' 失败: {e}")
|
||
return False
|
||
|
||
|
||
def memory_exists(memory_id):
|
||
"""
|
||
检查指定记忆ID的记忆集是否存在。
|
||
:param memory_id: 记忆ID。
|
||
:return: True如果存在，False如果不存在。
|
||
"""
|
||
return os.path.exists(get_memory_data_file_path(memory_id))
|
||
|
||
|
||
# --- 加载/保存当前活动会话的文件 ---
|
||
|
||
|
||
def load_active_features(role_id):
|
||
"""加载当前激活角色的特征数据"""
|
||
file_path = get_role_features_file_path(role_id)
|
||
return _load_json_file(file_path, DEFAULT_FEATURE_CONTENT.copy())
|
||
|
||
|
||
def save_active_features(role_id, features_data):
|
||
"""保存当前激活角色的特征数据"""
|
||
file_path = get_role_features_file_path(role_id)
|
||
return _save_json_file(file_path, features_data)
|
||
|
||
|
||
def load_active_memory(memory_id):
|
||
"""加载当前激活记忆集的记忆数据"""
|
||
file_path = get_memory_data_file_path(memory_id)
|
||
return _load_json_file(file_path, DEFAULT_MEMORY_CONTENT.copy())
|
||
|
||
|
||
def save_active_memory(memory_id, memory_data):
|
||
"""保存当前激活记忆集的记忆数据"""
|
||
file_path = get_memory_data_file_path(memory_id)
|
||
return _save_json_file(file_path, memory_data)
|
||
|
||
|
||
def append_chat_log(chat_log_id, entry):
|
||
"""向当前激活聊天记录文件追加条目"""
|
||
file_path = get_chat_log_file_path(chat_log_id)
|
||
return _append_jsonl_file(file_path, entry)
|
||
|
||
|
||
def read_chat_log(chat_log_id, limit=None):
|
||
"""读取当前激活聊天记录的所有条目"""
|
||
file_path = get_chat_log_file_path(chat_log_id)
|
||
return _read_jsonl_file(file_path, limit)
|
||
|
||
|
||
def delete_chat_log(chat_log_id):
|
||
"""删除指定聊天记录文件"""
|
||
file_path = get_chat_log_file_path(chat_log_id)
|
||
if os.path.exists(file_path):
|
||
try:
|
||
os.remove(file_path)
|
||
logger.info(f"聊天记录文件 '{file_path}' 已删除。")
|
||
return True
|
||
except OSError as e:
|
||
logger.error(f"删除聊天记录文件 '{file_path}' 失败: {e}")
|
||
return False
|
||
logger.warning(f"聊天记录文件 '{file_path}' 不存在，无需删除。")
|
||
return False
|
||
|
||
|
||
# --- 确保默认角色和记忆存在 ---
|
||
|
||
|
||
def ensure_initial_data():
|
||
"""
|
||
确保 'default_role' 和 'default_memory' 存在，如果它们不存在。
|
||
这个函数会在应用启动时调用。
|
||
"""
|
||
config_data = get_config()["Session"]
|
||
default_role_id = config_data["CURRENT_ROLE_ID"]
|
||
default_memory_id = config_data["CURRENT_MEMORY_ID"]
|
||
|
||
if not os.path.exists(get_role_features_file_path(default_role_id)):
|
||
logger.info(f"默认角色 '{default_role_id}' 不存在，正在创建。")
|
||
create_role(default_role_id, config_data.get("DEFAULT_ROLE_NAME", "通用AI助手"))
|
||
|
||
if not os.path.exists(get_memory_data_file_path(default_memory_id)):
|
||
logger.info(f"默认记忆集 '{default_memory_id}' 不存在，正在创建。")
|
||
create_memory(
|
||
default_memory_id, config_data.get("DEFAULT_MEMORY_NAME", "通用记忆集")
|
||
)
|