diff --git a/backend/api/route.py b/backend/api/route.py index 8d8a633..d458a1c 100644 --- a/backend/api/route.py +++ b/backend/api/route.py @@ -1,68 +1,16 @@ from fastapi import APIRouter -from ..core.items import ChatRequest -from ..tools.get_all_role_and_chat import get_all_role_and_chat -from ..core.models.chat_history import ChatHistory # 修改导入语句 +from .routes import presetsRoute, chatsRoute, worldbooksRoute router = APIRouter() -# 从本地读取所有的data内容 +# 注册子路由 +router.include_router(presetsRoute.router) +router.include_router(chatsRoute.router) +router.include_router(worldbooksRoute.router) + + +# 保留原有的其他路由 @router.get("/tool_bar/get_all_role_and_chat") def get_all_role_and_chat_endpoint(): - # 正确调用函数并返回结果 - return get_all_role_and_chat() - -# 根据rolename和chatname读取特定聊天记录 -@router.get("/chat_box/get_chat_history") -async def get_chat_history_endpoint(role_name: str, chat_name: str): - # 实例化工具类 - reader = ChatHistory.load_from_file(role_name, chat_name) - - return reader.to_chatbox_format() - - -# 从本地读取所有的预设列表内容 -@router.get("/presets/list") -async def get_presets_list(): - """ - 获取所有可用的预设列表 - 返回预设名称列表 - """ - import os - from pathlib import Path - - # 预设文件存储目录 - preset_dir = Path("data/preset") - - # 确保目录存在 - if not preset_dir.exists(): - return {"presets": []} - - # 获取所有.json文件 - preset_files = list(preset_dir.glob("*.json")) - - # 提取文件名(不带扩展名)作为预设名称 - presets = [file.stem for file in preset_files] - - return {"presets": presets} - - -# 从本地读取特定预设 -@router.get("/presets/{preset_name}") -async def get_preset_content(preset_name: str): - """ - 获取特定预设的完整内容 - """ - import os - from pathlib import Path - import json - - preset_path = Path("data/preset") / f"{preset_name}.json" - - if not preset_path.exists(): - raise HTTPException(status_code=404, detail="Preset not found") - - with open(preset_path, 'r', encoding='utf-8') as f: - preset_data = json.load(f) - - return preset_data - + from ..tools.get_all_role_and_chat import get_all_role_and_chat + return get_all_role_and_chat() diff --git a/backend/api/routes/chatsRoute.py b/backend/api/routes/chatsRoute.py new file mode 100644 index 0000000..4638f1b --- /dev/null +++ b/backend/api/routes/chatsRoute.py @@ -0,0 +1,193 @@ +from fastapi import APIRouter, HTTPException, status +from pathlib import Path +import json +from typing import List, Dict +from backend.core.models.chat_history import ChatHistory, Message + +router = APIRouter(prefix="/chats", tags=["chats"]) + + +# ========== 聊天历史基础路由 ========== + +@router.get("", response_model=Dict[str, List[Dict]]) +async def list_all_chats(): + """获取所有角色的所有聊天列表""" + data_dir = Path("data") + if not data_dir.exists(): + return {"chats": []} + + chats = [] + for role_dir in data_dir.iterdir(): + if role_dir.is_dir(): + for chat_file in role_dir.glob("*.jsonl"): + try: + with open(chat_file, 'r', encoding='utf-8') as f: + # 读取第一行获取元数据 + first_line = f.readline() + metadata = json.loads(first_line) + chats.append({ + "role_name": role_dir.name, + "chat_name": chat_file.stem, + "user_name": metadata.get("user_name", "User"), + "character_name": metadata.get("character_name", "Assistant"), + "last_modified": metadata.get("last_modified", ""), + "message_count": sum(1 for _ in f) # 统计剩余行数(消息数) + }) + except Exception as e: + continue # 跳过损坏的聊天文件 + return {"chats": chats} + + +@router.get("/{role_name}/{chat_name}") +async def get_chat(role_name: str, chat_name: str): + """获取指定聊天的完整内容""" + try: + chat_history = ChatHistory.load_from_file(role_name, chat_name) + return { + "metadata": chat_history.chat_metadata.dict(), + "messages": chat_history.to_chatbox_format() + } + except FileNotFoundError: + raise HTTPException(status_code=404, detail="Chat not found") + + +@router.post("/{role_name}", status_code=status.HTTP_201_CREATED) +async def create_chat(role_name: str, chat_name: str, metadata: Dict = None): + """创建新聊天""" + role_dir = Path("data") / role_name + role_dir.mkdir(parents=True, exist_ok=True) + chat_path = role_dir / f"{chat_name}.jsonl" + + if chat_path.exists(): + raise HTTPException(status_code=400, detail="Chat already exists") + + # 创建聊天历史对象 + chat_history = ChatHistory( + chat_metadata=metadata or {}, + messages=[] + ) + + # 保存到文件 + chat_history.save_to_file(role_name, chat_name) + return {"message": "Chat created successfully"} + + +@router.put("/{role_name}/{chat_name}") +async def update_chat(role_name: str, chat_name: str, update_data: Dict): + """更新聊天元数据""" + try: + chat_history = ChatHistory.load_from_file(role_name, chat_name) + + # 更新元数据 + if "metadata" in update_data: + for key, value in update_data["metadata"].items(): + if hasattr(chat_history.chat_metadata, key): + setattr(chat_history.chat_metadata, key, value) + + # 保存更改 + chat_history.save_to_file(role_name, chat_name) + return {"message": "Chat metadata updated successfully"} + except FileNotFoundError: + raise HTTPException(status_code=404, detail="Chat not found") + + +@router.delete("/{role_name}/{chat_name}") +async def delete_chat(role_name: str, chat_name: str): + """删除指定聊天""" + chat_path = Path("data") / role_name / f"{chat_name}.jsonl" + if not chat_path.exists(): + raise HTTPException(status_code=404, detail="Chat not found") + chat_path.unlink() + return {"message": "Chat deleted successfully"} + + +# ========== 聊天消息路由 ========== + +@router.get("/{role_name}/{chat_name}/messages") +async def list_messages(role_name: str, chat_name: str): + """获取聊天的所有消息""" + try: + chat_history = ChatHistory.load_from_file(role_name, chat_name) + return {"messages": chat_history.to_chatbox_format()} + except FileNotFoundError: + raise HTTPException(status_code=404, detail="Chat not found") + + +@router.get("/{role_name}/{chat_name}/messages/{floor}") +async def get_message(role_name: str, chat_name: str, floor: int): + """获取指定楼层的消息""" + try: + chat_history = ChatHistory.load_from_file(role_name, chat_name) + message = next((msg for msg in chat_history.messages if msg.floor == floor), None) + + if not message: + raise HTTPException(status_code=404, detail="Message not found") + + return message.dict() + except FileNotFoundError: + raise HTTPException(status_code=404, detail="Chat not found") + + +@router.post("/{role_name}/{chat_name}/messages", status_code=status.HTTP_201_CREATED) +async def add_message(role_name: str, chat_name: str, message_data: Dict): + """向聊天添加新消息""" + try: + chat_history = ChatHistory.load_from_file(role_name, chat_name) + + # 创建消息对象 + message = Message(**message_data) + + # 检查楼层是否已存在 + if any(msg.floor == message.floor for msg in chat_history.messages): + raise HTTPException(status_code=400, detail="Message floor already exists") + + # 添加消息 + chat_history.messages.append(message) + + # 保存更改 + chat_history.save_to_file(role_name, chat_name) + return {"message": "Message added successfully", "floor": message.floor} + except FileNotFoundError: + raise HTTPException(status_code=404, detail="Chat not found") + + +@router.put("/{role_name}/{chat_name}/messages/{floor}") +async def update_message(role_name: str, chat_name: str, floor: int, update_data: Dict): + """更新指定楼层的消息""" + try: + chat_history = ChatHistory.load_from_file(role_name, chat_name) + message = next((msg for msg in chat_history.messages if msg.floor == floor), None) + + if not message: + raise HTTPException(status_code=404, detail="Message not found") + + # 更新消息字段 + for key, value in update_data.items(): + if hasattr(message, key): + setattr(message, key, value) + + # 保存更改 + chat_history.save_to_file(role_name, chat_name) + return {"message": "Message updated successfully"} + except FileNotFoundError: + raise HTTPException(status_code=404, detail="Chat not found") + + +@router.delete("/{role_name}/{chat_name}/messages/{floor}") +async def delete_message(role_name: str, chat_name: str, floor: int): + """删除指定楼层的消息""" + try: + chat_history = ChatHistory.load_from_file(role_name, chat_name) + + # 查找并删除消息 + original_length = len(chat_history.messages) + chat_history.messages = [msg for msg in chat_history.messages if msg.floor != floor] + + if len(chat_history.messages) == original_length: + raise HTTPException(status_code=404, detail="Message not found") + + # 保存更改 + chat_history.save_to_file(role_name, chat_name) + return {"message": "Message deleted successfully"} + except FileNotFoundError: + raise HTTPException(status_code=404, detail="Chat not found") diff --git a/backend/api/routes/presetsRoute.py b/backend/api/routes/presetsRoute.py new file mode 100644 index 0000000..8200464 --- /dev/null +++ b/backend/api/routes/presetsRoute.py @@ -0,0 +1,264 @@ +from fastapi import APIRouter, HTTPException, status +from pathlib import Path +import json +from typing import List, Dict +from backend.core.models.PromptList import AIDesignSpec +from backend.core.models.PromptComponent import PromptComponent + +router = APIRouter(prefix="/presets", tags=["presets"]) + + +# ========== 预设基础路由 ========== + +@router.get("", response_model=Dict[str, List[Dict]]) +async def list_presets(): + """获取所有预设列表及其基本信息""" + preset_dir = Path("data/preset") + if not preset_dir.exists(): + return {"presets": []} + + presets = [] + for preset_file in preset_dir.glob("*.json"): + try: + with open(preset_file, 'r', encoding='utf-8') as f: + preset_data = json.load(f) + presets.append({ + "name": preset_file.stem, + "description": preset_data.get("description", ""), + "component_count": len(preset_data.get("prompts", [])), + "temperature": preset_data.get("temperature", 1.0) + }) + except Exception as e: + continue # 跳过损坏的预设文件 + return {"presets": presets} + + +@router.get("/{preset_name}") +async def get_preset(preset_name: str): + """获取指定预设的完整内容""" + preset_path = Path("data/preset") / f"{preset_name}.json" + if not preset_path.exists(): + raise HTTPException(status_code=404, detail="Preset not found") + + try: + with open(preset_path, 'r', encoding='utf-8') as f: + preset_data = json.load(f) + + # 转换为AIDesignSpec对象进行验证 + ai_design_spec = AIDesignSpec(**preset_data) + return ai_design_spec.dict() + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to load preset: {str(e)}") + + +@router.post("", status_code=status.HTTP_201_CREATED) +async def create_preset(preset_name: str, preset_data: Dict): + """创建新预设""" + preset_dir = Path("data/preset") + preset_dir.mkdir(parents=True, exist_ok=True) + preset_path = preset_dir / f"{preset_name}.json" + + if preset_path.exists(): + raise HTTPException(status_code=400, detail="Preset already exists") + + try: + # 验证并转换为AIDesignSpec对象 + ai_design_spec = AIDesignSpec(**preset_data) + + # 保存到文件 + with open(preset_path, 'w', encoding='utf-8') as f: + json.dump(ai_design_spec.dict(), f, ensure_ascii=False, indent=2) + + return {"message": "Preset created successfully", "name": preset_name} + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to create preset: {str(e)}") + + +@router.put("/{preset_name}") +async def update_preset(preset_name: str, update_data: Dict): + """更新预设配置""" + preset_path = Path("data/preset") / f"{preset_name}.json" + if not preset_path.exists(): + raise HTTPException(status_code=404, detail="Preset not found") + + try: + # 加载现有预设 + with open(preset_path, 'r', encoding='utf-8') as f: + preset_data = json.load(f) + + # 更新字段 + for key, value in update_data.items(): + preset_data[key] = value + + # 验证并转换为AIDesignSpec对象 + ai_design_spec = AIDesignSpec(**preset_data) + + # 保存更新 + with open(preset_path, 'w', encoding='utf-8') as f: + json.dump(ai_design_spec.dict(), f, ensure_ascii=False, indent=2) + + return {"message": "Preset updated successfully"} + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to update preset: {str(e)}") + + +@router.delete("/{preset_name}") +async def delete_preset(preset_name: str): + """删除指定预设""" + preset_path = Path("data/preset") / f"{preset_name}.json" + if not preset_path.exists(): + raise HTTPException(status_code=404, detail="Preset not found") + + try: + preset_path.unlink() + return {"message": "Preset deleted successfully"} + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to delete preset: {str(e)}") + + +# ========== 预设组件路由 ========== + +@router.get("/{preset_name}/components") +async def list_preset_components(preset_name: str): + """获取预设中的所有组件""" + preset_path = Path("data/preset") / f"{preset_name}.json" + if not preset_path.exists(): + raise HTTPException(status_code=404, detail="Preset not found") + + try: + with open(preset_path, 'r', encoding='utf-8') as f: + preset_data = json.load(f) + + # 获取组件列表 + components = preset_data.get("prompts", []) + return {"components": components} + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to load components: {str(e)}") + + +@router.get("/{preset_name}/components/{component_id}") +async def get_preset_component(preset_name: str, component_id: str): + """获取指定组件的详情""" + preset_path = Path("data/preset") / f"{preset_name}.json" + if not preset_path.exists(): + raise HTTPException(status_code=404, detail="Preset not found") + + try: + with open(preset_path, 'r', encoding='utf-8') as f: + preset_data = json.load(f) + + # 查找组件 + components = preset_data.get("prompts", []) + component = next((c for c in components if c.get("identifier") == component_id), None) + + if not component: + raise HTTPException(status_code=404, detail="Component not found") + + return component + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to load component: {str(e)}") + + +@router.post("/{preset_name}/components", status_code=status.HTTP_201_CREATED) +async def add_preset_component(preset_name: str, component_data: Dict): + """向预设添加新组件""" + preset_path = Path("data/preset") / f"{preset_name}.json" + if not preset_path.exists(): + raise HTTPException(status_code=404, detail="Preset not found") + + try: + # 加载预设数据 + with open(preset_path, 'r', encoding='utf-8') as f: + preset_data = json.load(f) + + # 验证组件数据 + component = PromptComponent(**component_data) + + # 检查组件ID是否已存在 + components = preset_data.get("prompts", []) + if any(c.get("identifier") == component.identifier for c in components): + raise HTTPException(status_code=400, detail="Component identifier already exists") + + # 添加组件 + components.append(component.dict()) + preset_data["prompts"] = components + + # 保存更新 + with open(preset_path, 'w', encoding='utf-8') as f: + json.dump(preset_data, f, ensure_ascii=False, indent=2) + + return {"message": "Component added successfully", "identifier": component.identifier} + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to add component: {str(e)}") + + +@router.put("/{preset_name}/components/{component_id}") +async def update_preset_component(preset_name: str, component_id: str, update_data: Dict): + """更新指定组件""" + preset_path = Path("data/preset") / f"{preset_name}.json" + if not preset_path.exists(): + raise HTTPException(status_code=404, detail="Preset not found") + + try: + # 加载预设数据 + with open(preset_path, 'r', encoding='utf-8') as f: + preset_data = json.load(f) + + # 查找并更新组件 + components = preset_data.get("prompts", []) + component_index = next((i for i, c in enumerate(components) if c.get("identifier") == component_id), None) + + if component_index is None: + raise HTTPException(status_code=404, detail="Component not found") + + # 更新组件字段 + for key, value in update_data.items(): + components[component_index][key] = value + + # 保存更新 + with open(preset_path, 'w', encoding='utf-8') as f: + json.dump(preset_data, f, ensure_ascii=False, indent=2) + + return {"message": "Component updated successfully"} + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to update component: {str(e)}") + + +@router.delete("/{preset_name}/components/{component_id}") +async def delete_preset_component(preset_name: str, component_id: str): + """从预设中删除指定组件""" + preset_path = Path("data/preset") / f"{preset_name}.json" + if not preset_path.exists(): + raise HTTPException(status_code=404, detail="Preset not found") + + try: + # 加载预设数据 + with open(preset_path, 'r', encoding='utf-8') as f: + preset_data = json.load(f) + + # 查找并删除组件 + components = preset_data.get("prompts", []) + original_length = len(components) + components = [c for c in components if c.get("identifier") != component_id] + + if len(components) == original_length: + raise HTTPException(status_code=404, detail="Component not found") + + # 更新预设数据 + preset_data["prompts"] = components + + # 保存更新 + with open(preset_path, 'w', encoding='utf-8') as f: + json.dump(preset_data, f, ensure_ascii=False, indent=2) + + return {"message": "Component deleted successfully"} + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to delete component: {str(e)}") diff --git a/backend/api/routes/worldbooksRoute.py b/backend/api/routes/worldbooksRoute.py new file mode 100644 index 0000000..a380e68 --- /dev/null +++ b/backend/api/routes/worldbooksRoute.py @@ -0,0 +1,170 @@ +from fastapi import APIRouter, HTTPException, status +from pathlib import Path +from typing import Dict, List +from backend.core.models.WorldBook import WorldBook +from backend.core.models.WorldItem import WorldInfoEntry + +router = APIRouter(prefix="/worldbooks", tags=["worldbooks"]) + + +# ========== 世界书基础路由 ========== + +@router.get("", response_model=Dict[str, List[Dict]]) +async def list_worldbooks(): + """获取所有世界书列表""" + worldbook_dir = Path("data/worldbooks") + if not worldbook_dir.exists(): + return {"worldbooks": []} + + worldbooks = [] + for wb_file in worldbook_dir.glob("*.json"): + try: + worldbook = WorldBook.from_sillytavern_json(str(wb_file)) + worldbooks.append(worldbook.to_summary_dict()) + except Exception as e: + print(f"警告: 跳过文件 {wb_file.name}，加载失败: {e}") + continue + + return {"worldbooks": worldbooks} + + +@router.get("/{worldbook_uid}") +async def get_worldbook(worldbook_uid: str): + """获取指定世界书完整内容""" + worldbook_path = Path("data/worldbooks") / f"{worldbook_uid}.json" + if not worldbook_path.exists(): + raise HTTPException(status_code=404, detail="WorldBook not found") + + try: + worldbook = WorldBook.from_sillytavern_json(str(worldbook_path)) + return worldbook.to_dict() + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to load worldbook: {str(e)}") + + +@router.post("", status_code=status.HTTP_201_CREATED) +async def create_worldbook(worldbook_data: Dict): + """创建新世界书""" + worldbook_dir = Path("data/worldbooks") + worldbook_dir.mkdir(parents=True, exist_ok=True) + + worldbook = WorldBook.from_dict(worldbook_data) + worldbook_path = worldbook_dir / f"{worldbook.uid}.json" + + if worldbook_path.exists(): + raise HTTPException(status_code=400, detail="WorldBook already exists") + + worldbook.to_sillytavern_json(str(worldbook_path)) + return {"message": "WorldBook created successfully", "uid": worldbook.uid} + + +@router.put("/{worldbook_uid}") +async def update_worldbook(worldbook_uid: str, update_data: Dict): + """更新世界书基本信息""" + worldbook_path = Path("data/worldbooks") / f"{worldbook_uid}.json" + if not worldbook_path.exists(): + raise HTTPException(status_code=404, detail="WorldBook not found") + + try: + worldbook = WorldBook.from_sillytavern_json(str(worldbook_path)) + for key, value in update_data.items(): + if hasattr(worldbook, key): + setattr(worldbook, key, value) + worldbook.to_sillytavern_json(str(worldbook_path)) + return {"message": "WorldBook updated successfully"} + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to update worldbook: {str(e)}") + + +@router.delete("/{worldbook_uid}") +async def delete_worldbook(worldbook_uid: str): + """删除世界书""" + worldbook_path = Path("data/worldbooks") / f"{worldbook_uid}.json" + if not worldbook_path.exists(): + raise HTTPException(status_code=404, detail="WorldBook not found") + worldbook_path.unlink() + return {"message": "WorldBook deleted successfully"} + + +# ========== 世界书条目路由 ========== + +@router.get("/{worldbook_uid}/entries") +async def list_worldbook_entries(worldbook_uid: str): + """获取世界书所有条目列表""" + worldbook_path = Path("data/worldbooks") / f"{worldbook_uid}.json" + if not worldbook_path.exists(): + raise HTTPException(status_code=404, detail="WorldBook not found") + + try: + worldbook = WorldBook.from_sillytavern_json(str(worldbook_path)) + return {"entries": [entry.dict() for entry in worldbook.entries]} + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to load entries: {str(e)}") + + +@router.get("/{worldbook_uid}/entries/{entry_uid}") +async def get_worldbook_entry(worldbook_uid: str, entry_uid: str): + """获取指定条目详情""" + worldbook_path = Path("data/worldbooks") / f"{worldbook_uid}.json" + if not worldbook_path.exists(): + raise HTTPException(status_code=404, detail="WorldBook not found") + + try: + worldbook = WorldBook.from_sillytavern_json(str(worldbook_path)) + entry = worldbook.get_entry(entry_uid) + if not entry: + raise HTTPException(status_code=404, detail="Entry not found") + return entry.dict() + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to load entry: {str(e)}") + + +@router.post("/{worldbook_uid}/entries", status_code=status.HTTP_201_CREATED) +async def add_worldbook_entry(worldbook_uid: str, entry_data: Dict): + """向世界书添加新条目""" + worldbook_path = Path("data/worldbooks") / f"{worldbook_uid}.json" + if not worldbook_path.exists(): + raise HTTPException(status_code=404, detail="WorldBook not found") + + try: + worldbook = WorldBook.from_sillytavern_json(str(worldbook_path)) + entry = WorldInfoEntry(**entry_data) + worldbook.add_entry(entry) + worldbook.to_sillytavern_json(str(worldbook_path)) + return {"message": "Entry added successfully", "entry_uid": entry.uid} + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to add entry: {str(e)}") + + +@router.put("/{worldbook_uid}/entries/{entry_uid}") +async def update_worldbook_entry(worldbook_uid: str, entry_uid: str, update_data: Dict): + """更新指定条目""" + worldbook_path = Path("data/worldbooks") / f"{worldbook_uid}.json" + if not worldbook_path.exists(): + raise HTTPException(status_code=404, detail="WorldBook not found") + + try: + worldbook = WorldBook.from_sillytavern_json(str(worldbook_path)) + if not worldbook.update_entry(entry_uid, **update_data): + raise HTTPException(status_code=404, detail="Entry not found") + worldbook.to_sillytavern_json(str(worldbook_path)) + return {"message": "Entry updated successfully"} + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to update entry: {str(e)}") + + +@router.delete("/{worldbook_uid}/entries/{entry_uid}") +async def delete_worldbook_entry(worldbook_uid: str, entry_uid: str): + """从世界书删除指定条目""" + worldbook_path = Path("data/worldbooks") / f"{worldbook_uid}.json" + if not worldbook_path.exists(): + raise HTTPException(status_code=404, detail="WorldBook not found") + + try: + worldbook = WorldBook.from_sillytavern_json(str(worldbook_path)) + if not worldbook.remove_entry(entry_uid): + raise HTTPException(status_code=404, detail="Entry not found") + worldbook.to_sillytavern_json(str(worldbook_path)) + return {"message": "Entry deleted successfully"} + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to delete entry: {str(e)}") diff --git a/backend/core/models/PromptComponent.py b/backend/core/models/PromptComponent.py index 42f8d15..1ca3b04 100644 --- a/backend/core/models/PromptComponent.py +++ b/backend/core/models/PromptComponent.py @@ -1,4 +1,5 @@ -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, validator +from typing import Dict, Any class PromptComponent(BaseModel): @@ -12,4 +13,48 @@ class PromptComponent(BaseModel): system_prompt: bool = Field(False, description="是否强制作为系统提示词处理") marker: bool = Field(False, description="是否为动态插入点占位符") + @validator('role') + def validate_role(cls, v): + """验证角色值是否在有效范围内""" + if v not in [0, 1, 2]: + raise ValueError("角色值必须是0(System)、1(User)或2(Assistant)") + return v + def update(self, **kwargs) -> None: + """ + 更新组件属性 + + 参数: + **kwargs: 要更新的字段和值 + + 异常: + ValueError: 当尝试更新identifier时抛出 + """ + if 'identifier' in kwargs: + raise ValueError("组件标识符不可修改") + + for key, value in kwargs.items(): + if hasattr(self, key): + setattr(self, key, value) + + def to_dict(self) -> Dict[str, Any]: + """ + 将组件转换为字典 + + 返回: + Dict[str, Any]: 组件的字典表示 + """ + return self.dict() + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> 'PromptComponent': + """ + 从字典创建组件实例 + + 参数: + data: 包含组件数据的字典 + + 返回: + PromptComponent: 组件实例 + """ + return cls(**data) diff --git a/backend/core/models/PromptList.py b/backend/core/models/PromptList.py index 4d4bd8f..d005928 100644 --- a/backend/core/models/PromptList.py +++ b/backend/core/models/PromptList.py @@ -1,6 +1,6 @@ -from pydantic import BaseModel, Field -import PromptComponent -from typing import List, Dict, Any +from pydantic import BaseModel, Field, validator +from typing import List, Dict, Any, Optional +from .PromptComponent import PromptComponent class AIDesignSpec(BaseModel): @@ -50,3 +50,167 @@ class AIDesignSpec(BaseModel): default_factory=list, description="组装说明书，定义构建最终提示词的顺序" ) + + @validator('prompts') + def validate_prompts_unique_identifier(cls, v): + """验证组件标识符唯一性""" + identifiers = [comp.identifier for comp in v] + if len(identifiers) != len(set(identifiers)): + raise ValueError("组件标识符必须唯一") + return v + + @validator('prompt_order') + def validate_prompt_order_exists(cls, v, values): + """验证prompt_order中的组件ID是否存在于prompts中""" + if 'prompts' in values: + prompt_ids = {comp.identifier for comp in values['prompts']} + invalid_ids = set(v) - prompt_ids + if invalid_ids: + raise ValueError(f"prompt_order中包含不存在的组件ID: {invalid_ids}") + return v + + # ========== 组件管理方法 ========== + + def add_component(self, component: PromptComponent) -> None: + """ + 添加新组件 + + 参数: + component: 要添加的组件 + + 异常: + ValueError: 当组件标识符已存在时抛出 + """ + if any(c.identifier == component.identifier for c in self.prompts): + raise ValueError(f"组件标识符 {component.identifier} 已存在") + self.prompts.append(component) + + def remove_component(self, identifier: str) -> bool: + """ + 移除指定组件 + + 参数: + identifier: 组件标识符 + + 返回: + bool: 是否成功移除 + """ + original_length = len(self.prompts) + self.prompts = [c for c in self.prompts if c.identifier != identifier] + + # 同时从prompt_order中移除 + self.prompt_order = [id for id in self.prompt_order if id != identifier] + + return len(self.prompts) < original_length + + def get_component(self, identifier: str) -> Optional[PromptComponent]: + """ + 获取指定组件 + + 参数: + identifier: 组件标识符 + + 返回: + Optional[PromptComponent]: 找到的组件，未找到返回None + """ + for component in self.prompts: + if component.identifier == identifier: + return component + return None + + def update_component(self, identifier: str, **kwargs) -> bool: + """ + 更新指定组件 + + 参数: + identifier: 组件标识符 + **kwargs: 要更新的字段 + + 返回: + bool: 是否成功更新 + """ + component = self.get_component(identifier) + if component is None: + return False + + component.update(**kwargs) + return True + + def list_components(self) -> List[Dict[str, Any]]: + """ + 列出所有组件 + + 返回: + List[Dict[str, Any]]: 组件字典列表 + """ + return [component.to_dict() for component in self.prompts] + + def reorder_components(self, new_order: List[str]) -> None: + """ + 重新排序组件 + + 参数: + new_order: 新的组件标识符顺序 + + 异常: + ValueError: 当包含不存在的组件ID时抛出 + """ + # 验证所有ID都存在 + existing_ids = {c.identifier for c in self.prompts} + invalid_ids = set(new_order) - existing_ids + + if invalid_ids: + raise ValueError(f"包含不存在的组件ID: {invalid_ids}") + + self.prompt_order = new_order + + def get_ordered_components(self) -> List[PromptComponent]: + """ + 获取按prompt_order排序的组件列表 + + 返回: + List[PromptComponent]: 排序后的组件列表 + """ + component_map = {c.identifier: c for c in self.prompts} + ordered_components = [] + + for identifier in self.prompt_order: + if identifier in component_map: + ordered_components.append(component_map[identifier]) + + # 添加未在prompt_order中的组件 + ordered_components.extend([ + c for c in self.prompts + if c.identifier not in self.prompt_order + ]) + + return ordered_components + + def to_dict(self) -> Dict[str, Any]: + """ + 将设计规范转换为字典 + + 返回: + Dict[str, Any]: 设计规范的字典表示 + """ + return self.dict() + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> 'AIDesignSpec': + """ + 从字典创建设计规范实例 + + 参数: + data: 包含设计规范数据的字典 + + 返回: + AIDesignSpec: 设计规范实例 + """ + # 处理prompts字段 + if 'prompts' in data: + data['prompts'] = [ + PromptComponent.from_dict(comp) if isinstance(comp, dict) else comp + for comp in data['prompts'] + ] + + return cls(**data) diff --git a/backend/core/models/WorldBook.py b/backend/core/models/WorldBook.py new file mode 100644 index 0000000..0808880 --- /dev/null +++ b/backend/core/models/WorldBook.py @@ -0,0 +1,391 @@ +import json +import os +from typing import Dict, List, Optional, Any +from pydantic import BaseModel, Field, validator +from .WorldItem import WorldInfoEntry, TriggerStrategy, PositionMode + + +class WorldBook(BaseModel): + """ + 世界书集合模型 + 管理多个世界书条目，支持导入导出 SillyTavern 格式 + """ + # 世界书基本信息 + uid: str = Field(..., description="世界书唯一标识符") + name: str = Field(..., description="世界书名称") + description: str = Field("", description="世界书描述") + + # 条目集合 + entries: List[WorldInfoEntry] = Field( + default_factory=list, + description="世界书条目列表" + ) + + # 全局配置 + enabled: bool = Field(True, description="是否启用") + scan_depth: int = Field(4, description="默认扫描深度") + rag_threshold: float = Field(0.75, description="默认RAG相似度阈值") + + @validator('entries') + def validate_entries_unique_uid(cls, v): + uids = [entry.uid for entry in v] + if len(uids) != len(set(uids)): + raise ValueError("条目 UID 必须唯一") + return v + + def add_entry(self, entry: WorldInfoEntry) -> None: + """ + 添加世界书条目 + + Args: + entry: 世界书条目对象 + """ + if any(e.uid == entry.uid for e in self.entries): + raise ValueError(f"条目 UID {entry.uid} 已存在") + self.entries.append(entry) + + def remove_entry(self, uid: str) -> bool: + """ + 移除世界书条目 + + Args: + uid: 条目 UID + + Returns: + bool: 是否成功移除 + """ + original_length = len(self.entries) + self.entries = [e for e in self.entries if e.uid != uid] + return len(self.entries) < original_length + + def get_entry(self, uid: str) -> Optional[WorldInfoEntry]: + """ + 获取指定 UID 的世界书条目 + + Args: + uid: 条目 UID + + Returns: + Optional[WorldInfoEntry]: 找到的条目，未找到返回 None + """ + for entry in self.entries: + if entry.uid == uid: + return entry + return None + + def update_entry(self, uid: str, **kwargs) -> bool: + """ + 更新世界书条目 + + Args: + uid: 条目 UID + **kwargs: 要更新的字段 + + Returns: + bool: 是否成功更新 + """ + entry = self.get_entry(uid) + if entry is None: + return False + + for key, value in kwargs.items(): + if hasattr(entry, key): + setattr(entry, key, value) + return True + + def filter_by_position(self, position_mode: PositionMode, position_value: int) -> List[WorldInfoEntry]: + """ + 根据位置筛选条目 + + Args: + position_mode: 位置模式 + position_value: 位置值 + + Returns: + List[WorldInfoEntry]: 筛选后的条目列表 + """ + return [ + entry for entry in self.entries + if entry.position_mode == position_mode and + (entry.position_anchor == position_value or + (position_mode == PositionMode.ABSOLUTE_DEPTH and entry.position_dx == position_value)) + ] + + def get_summary(self) -> Dict[str, Any]: + """ + 获取世界书概要信息 + + Returns: + Dict[str, Any]: 概要信息字典 + """ + return { + "uid": self.uid, + "name": self.name, + "description": self.description, + "entry_count": len(self.entries), + "enabled": self.enabled, + "scan_depth": self.scan_depth, + "rag_threshold": self.rag_threshold, + "trigger_strategies": { + strategy.value: sum(1 for e in self.entries if e.trigger_strategy == strategy) + for strategy in TriggerStrategy + } + } + + def to_summary_dict(self) -> Dict[str, Any]: + """ + 生成世界书摘要信息，用于列表显示 + + Returns: + Dict[str, Any]: 包含基本信息的字典 + """ + return { + "uid": self.uid, + "name": self.name, + "description": self.description, + "entry_count": len(self.entries), + "enabled": self.enabled + } + + @classmethod + def from_dict(cls, data: Dict) -> 'WorldBook': + """ + 从字典创建 WorldBook 对象，支持多种数据格式 + + Args: + data: 包含世界书数据的字典 + + Returns: + WorldBook: 世界书对象 + + Raises: + ValueError: 数据格式无效 + """ + # 处理包含 originalData 的格式 + if 'originalData' in data: + original_data = data['originalData'] + return cls( + uid=data.get('uid') or original_data.get('name', 'unknown'), + name=original_data.get('name', 'Unknown'), + description=original_data.get('description', ''), + enabled=data.get('enabled', True), + entries=[WorldInfoEntry(**entry) for entry in data.get('entries', {}).values()] + ) + + # 处理标准格式 + return cls(**data) + + def to_dict(self) -> Dict: + """ + 将 WorldBook 转换为字典，兼容多种格式 + + Returns: + Dict: 世界书数据字典 + """ + return { + 'uid': self.uid, + 'name': self.name, + 'description': self.description, + 'enabled': self.enabled, + 'entries': {entry.uid: entry.dict() for entry in self.entries} + } + + @classmethod + def from_sillytavern_json(cls, file_path: str) -> 'WorldBook': + """ + 从 SillyTavern 格式的 JSON 文件加载世界书 + + Args: + file_path: JSON 文件路径 + + Returns: + WorldBook: 世界书对象 + + Raises: + FileNotFoundError: 文件不存在 + ValueError: 格式不符合 SillyTavern 标准 + json.JSONDecodeError: JSON 解析错误 + """ + if not os.path.exists(file_path): + raise FileNotFoundError(f"世界书文件未找到: {file_path}") + + with open(file_path, 'r', encoding='utf-8') as f: + raw_data = json.load(f) + + # 创建世界书对象 + world_name = raw_data.get("name", "") + if not world_name: + # 如果 name 字段不存在或为空，从文件名中提取 + file_name = os.path.splitext(os.path.basename(file_path))[0] + world_name = file_name + + # 检查是否有 originalData 字段 + if "originalData" in raw_data: + # 使用 originalData 格式 + original_data = raw_data["originalData"] + world_book = cls( + uid=f"st_{os.path.basename(file_path)}_{int(os.path.getmtime(file_path))}", + name=world_name, + description=f"从 SillyTavern 导入: {file_path}" + ) + + # 转换 originalData 中的条目 + for entry_data in original_data.get("entries", []): + try: + world_entry = WorldInfoEntry( + uid=str(entry_data.get("id", "")), + key=entry_data.get("keys", []), + keysecondary=entry_data.get("secondary_keys", []), + comment=entry_data.get("comment", ""), + content=entry_data.get("content", ""), + trigger_strategy=TriggerStrategy.KEYWORD, + position_mode=PositionMode.ANCHOR if entry_data.get( + "position") == "after_char" else PositionMode.ABSOLUTE_DEPTH, + position_anchor=0, + position_dx=None, + constant=entry_data.get("constant", False), + selective=entry_data.get("selective", True), + order=entry_data.get("insertion_order", 100), + probability=entry_data.get("extensions", {}).get("probability", 100), + useProbability=entry_data.get("extensions", {}).get("useProbability", False), + group=entry_data.get("extensions", {}).get("group", None), + match_whole_words=entry_data.get("use_regex", False), + use_regex=entry_data.get("use_regex", False), + vectorized=False + ) + world_book.add_entry(world_entry) + except Exception as e: + print(f"警告: 跳过条目，解析失败: {e}") + else: + # 使用标准 SillyTavern 格式 + entries_dict = raw_data.get("entries", {}) + if not isinstance(entries_dict, dict): + raise ValueError("无效的世界书格式：'entries' 字段必须是一个字典。") + + if not any(key.isdigit() for key in entries_dict.keys()): + raise ValueError("无效的世界书格式：'entries' 中未找到条目。请确认是 SillyTavern 导出的格式。") + + world_book = cls( + uid=f"st_{os.path.basename(file_path)}_{int(os.path.getmtime(file_path))}", + name=world_name, + description=f"从 SillyTavern 导入: {file_path}" + ) + + # 转换标准格式的条目 + for uid, entry_data in entries_dict.items(): + try: + position = entry_data.get("position", 0) + position_mode = PositionMode.ANCHOR if position < 6 else PositionMode.ABSOLUTE_DEPTH + + world_entry = WorldInfoEntry( + uid=uid, + key=entry_data.get("key", []), + keysecondary=entry_data.get("keysecondary", []), + comment=entry_data.get("comment", ""), + content=entry_data.get("content", ""), + trigger_strategy=TriggerStrategy.KEYWORD, + position_mode=position_mode, + position_anchor=position if position < 6 else 0, + position_dx=position if position >= 6 else None, + constant=entry_data.get("constant", False), + selective=entry_data.get("selective", True), + order=entry_data.get("order", 100), + probability=entry_data.get("probability", 100), + useProbability=entry_data.get("useProbability", False), + group=entry_data.get("group", None), + match_whole_words=entry_data.get("matchWholeWords", False), + use_regex=entry_data.get("useRegex", False), + vectorized=False + ) + world_book.add_entry(world_entry) + except Exception as e: + print(f"警告: 跳过条目 {uid}，解析失败: {e}") + + return world_book + + def to_sillytavern_json(self, file_path: str) -> None: + """ + 导出为 SillyTavern 格式的 JSON 文件 + + Args: + file_path: 要保存的文件路径 + """ + entries_dict = {} + + for entry in self.entries: + # 映射 WorldInfoEntry 到 SillyTavern 格式 + position = entry.position_anchor if entry.position_mode == PositionMode.ANCHOR else entry.position_dx + + entries_dict[entry.uid] = { + "uid": entry.uid, + "key": entry.key, + "keysecondary": entry.keysecondary, + "comment": entry.comment, + "content": entry.content, + "constant": entry.constant, + "selective": entry.selective, + "position": position, + "order": entry.order, + "probability": entry.probability, + "useProbability": entry.useProbability, + "group": entry.group, + "matchWholeWords": entry.match_whole_words, + "useRegex": entry.use_regex, + "preventRecursion": True, + "excludeRecursion": True + } + + output_data = { + "entries": entries_dict, + "name": self.name + } + + with open(file_path, 'w', encoding='utf-8') as f: + json.dump(output_data, f, ensure_ascii=False, indent=2) + + def list_triggers_and_content(self) -> List[Dict[str, Any]]: + """ + 提取所有条目的触发关键词和内容，用于快速构建向量数据库或索引 + + Returns: + List[Dict[str, Any]]: 包含 trigger (key) 和 content 的列表 + """ + result = [] + for entry in self.entries: + result.append({ + "uid": entry.uid, + "comment": entry.comment, + "triggers": entry.key, + "content": entry.content, + "position": entry.position_anchor if entry.position_mode == PositionMode.ANCHOR else entry.position_dx, + "constant": entry.constant, + "trigger_strategy": entry.trigger_strategy.value + }) + return result + + +# --- 使用示例 --- +if __name__ == "__main__": + try: + # 从 SillyTavern 格式导入 + world_book = WorldBook.from_sillytavern_json('entries.json') + + # 打印概要 + summary = world_book.get_summary() + print(f"世界书名称: {summary['name']}") + print(f"条目数量: {summary['entry_count']}") + print(f"触发策略分布: {summary['trigger_strategies']}") + + # 列出所有条目的触发词和内容预览 + print("\n--- 条目预览 ---") + for item in world_book.list_triggers_and_content(): + triggers = item['triggers'] if item['triggers'] else ['(无关键词 - 常驻)'] + content_preview = item['content'][:50].replace('\n', ' ') + "..." + print(f"[{item['position']}] TRIGGERS: {triggers} -> CONTENT: {content_preview}") + + # 导出为 SillyTavern 格式 + world_book.to_sillytavern_json('exported_world.json') + print("\n✅ 世界书已导出为 exported_world.json") + + except Exception as e: + print(f"❌ 错误: {e}") diff --git a/backend/core/models/WorldItem.py b/backend/core/models/WorldItem.py new file mode 100644 index 0000000..94d7775 --- /dev/null +++ b/backend/core/models/WorldItem.py @@ -0,0 +1,70 @@ +from enum import Enum +from typing import List, Optional +from pydantic import BaseModel, Field, validator + + +class TriggerStrategy(str, Enum): + """ + 触发策略枚举 + """ + ALL = "all" # 全部触发 + KEYWORD = "keyword" # 传统关键词匹配 + RAG = "rag" # 向量检索触发 + CONDITION = "condition" # 逻辑条件触发 + + +class PositionMode(str, Enum): + """ + 位置模式枚举 + """ + ANCHOR = "anchor" # 锚点模式 (0-5) + ABSOLUTE_DEPTH = "dx" # 绝对深度模式 (Dx) + + +class WorldInfoEntry(BaseModel): + """ + 世界书条目模型 v2.0 + 支持 RAG、条件触发及 Dx 绝对位置 + 兼容 SillyTavern 格式 + """ + # --- 核心身份 --- + uid: str + key: List[str] = [] + keysecondary: List[str] = [] + comment: str = "" + content: str + + # --- 策略配置 --- + trigger_strategy: TriggerStrategy = TriggerStrategy.KEYWORD + condition_expr: Optional[str] = None # 条件表达式 + rag_threshold: float = 0.75 # RAG 相似度阈值 + + # --- 位置与扫描 --- + position_mode: PositionMode = PositionMode.ANCHOR + position_anchor: int = 0 # 锚点值 (0-5) + position_dx: Optional[int] = None # 绝对深度值 + scan_depth: int = 4 # 向前扫描 N 条消息 + + # --- 其他配置 --- + constant: bool = False + selective: bool = True + order: int = 100 + probability: int = 100 + useProbability: bool = False + + # 高级过滤 + group: Optional[str] = None + match_whole_words: bool = False + use_regex: bool = False + vectorized: bool = False # 是否已生成向量 embedding + + # SillyTavern 特定字段 + enabled: bool = True + position: str = "after_char" # SillyTavern 位置字符串 + insertion_order: int = 100 # SillyTavern 插入顺序 + + @validator('position_dx') + def validate_dx(cls, v, values): + if values.get('position_mode') == PositionMode.ABSOLUTE_DEPTH and v is None: + raise ValueError("当 position_mode 为 ABSOLUTE_DEPTH 时，必须指定 position_dx") + return v diff --git a/backend/core/models/chat_history.py b/backend/core/models/chat_history.py index 58a2599..ba5b48d 100644 --- a/backend/core/models/chat_history.py +++ b/backend/core/models/chat_history.py @@ -6,211 +6,211 @@ import json class Message(BaseModel): - """消息类，代表JSONL文件中的一行消息内容""" - name: str = Field(..., description="发送者名称") - is_user: bool = Field(..., description="是否为用户消息") - is_system: bool = Field(False, description="是否为系统消息") - send_date: str = Field( - default_factory=lambda: str(int(datetime.now().timestamp() * 1000)), - description="消息发送时间戳" - ) - floor: int = Field(0, description="对话楼层数") - swipes: List[str] = Field( - default_factory=list, - description="历史版本列表。用户消息：存编辑过的不同版本。AI消息：存重roll生成的不同版本" - ) - swipe_id: int = Field( - 0, - description="当前指针。指示当前显示的是 swipes 数组中的第几个（从 0 开始）" - ) - mes: str = Field(..., description="消息内容文本") - extra: Dict[str, Any] = Field( - default_factory=dict, - description="额外信息，包含推理内容、API、模型等" - ) - force_avatar: Optional[str] = Field(None, description="强制头像URL") - variables: List[Any] = Field(default_factory=list, description="消息变量列表") - variables_initialized: List[bool] = Field(default_factory=list, description="变量初始化状态数组") - is_ejs_processed: List[bool] = Field(default_factory=list, description="EJS处理状态数组") + """消息类，代表JSONL文件中的一行消息内容""" + name: str = Field(..., description="发送者名称") + is_user: bool = Field(..., description="是否为用户消息") + is_system: bool = Field(False, description="是否为系统消息") + send_date: str = Field( + default_factory=lambda: str(int(datetime.now().timestamp() * 1000)), + description="消息发送时间戳" + ) + floor: int = Field(0, description="对话楼层数") + swipes: List[str] = Field( + default_factory=list, + description="历史版本列表。用户消息：存编辑过的不同版本。AI消息：存重roll生成的不同版本" + ) + swipe_id: int = Field( + 0, + description="当前指针。指示当前显示的是 swipes 数组中的第几个（从 0 开始）" + ) + mes: str = Field(..., description="消息内容文本") + extra: Dict[str, Any] = Field( + default_factory=dict, + description="额外信息，包含推理内容、API、模型等" + ) + force_avatar: Optional[str] = Field(None, description="强制头像URL") + variables: List[Any] = Field(default_factory=list, description="消息变量列表") + variables_initialized: List[bool] = Field(default_factory=list, description="变量初始化状态数组") + is_ejs_processed: List[bool] = Field(default_factory=list, description="EJS处理状态数组") - # 以下属性仅在is_user为False时有值 - api: Optional[str] = Field(None, description="使用的API提供商") - model: Optional[str] = Field(None, description="使用的AI模型") - reasoning: Optional[str] = Field(None, description="推理内容") - reasoning_duration: Optional[float] = Field(None, description="推理耗时") - reasoning_signature: Optional[str] = Field(None, description="推理签名") - time_to_first_token: Optional[float] = Field(None, description="首Token响应时间") - bias: Optional[float] = Field(None, description="偏差值") + # 以下属性仅在is_user为False时有值 + api: Optional[str] = Field(None, description="使用的API提供商") + model: Optional[str] = Field(None, description="使用的AI模型") + reasoning: Optional[str] = Field(None, description="推理内容") + reasoning_duration: Optional[float] = Field(None, description="推理耗时") + reasoning_signature: Optional[str] = Field(None, description="推理签名") + time_to_first_token: Optional[float] = Field(None, description="首Token响应时间") + bias: Optional[float] = Field(None, description="偏差值") class ChatMetadata(BaseModel): - """聊天元数据类，包含整个聊天的共享属性""" - user_name: str = Field("User", description="用户名称") - character_name: str = Field("Assistant", description="角色名称") + """聊天元数据类，包含整个聊天的共享属性""" + user_name: str = Field("User", description="用户名称") + character_name: str = Field("Assistant", description="角色名称") - # 完整性校验相关 - integrity: str = Field("", description="完整性校验值") - chat_id_hash: str = Field("", description="聊天ID哈希值") + # 完整性校验相关 + integrity: str = Field("", description="完整性校验值") + chat_id_hash: str = Field("", description="聊天ID哈希值") - # 笔记相关 - note_prompt: str = Field("", description="作者笔记提示词") - note_interval: int = Field(0, description="笔记插入间隔数") - note_position: int = Field(0, description="笔记插入位置") - note_depth: int = Field(0, description="笔记插入深度") - # 0：System，1：User，2：Assistant - note_role: int = Field("", description="笔记使用角色类型") + # 笔记相关 + note_prompt: str = Field("", description="作者笔记提示词") + note_interval: int = Field(0, description="笔记插入间隔数") + note_position: int = Field(0, description="笔记插入位置") + note_depth: int = Field(0, description="笔记插入深度") + # 0：System，1：User，2：Assistant + note_role: int = Field("", description="笔记使用角色类型") - # 扩展信息 - extensions: Dict[str, Any] = Field( - default_factory=dict, - description="扩展信息，如LittleWhiteBox等" - ) - # 世界信息 - timedWorldInfo: Dict[str, Any] = Field( - default_factory=dict, - description="定时世界信息" - ) - # 变量 - variables: Dict[str, Any] = Field( - default_factory=dict, - description="变量字典" - ) - # 状态标记 - tainted: bool = Field(False, description="是否被修改标记") - lastInContextMessageId: int = Field(-1, description="最后上下文消息ID") + # 扩展信息 + extensions: Dict[str, Any] = Field( + default_factory=dict, + description="扩展信息，如LittleWhiteBox等" + ) + # 世界信息 + timedWorldInfo: Dict[str, Any] = Field( + default_factory=dict, + description="定时世界信息" + ) + # 变量 + variables: Dict[str, Any] = Field( + default_factory=dict, + description="变量字典" + ) + # 状态标记 + tainted: bool = Field(False, description="是否被修改标记") + lastInContextMessageId: int = Field(-1, description="最后上下文消息ID") class ChatHistory(BaseModel): - """聊天文件类，包含完整的聊天记录""" - chat_metadata: ChatMetadata = Field(..., description="聊天元数据，包含基本信息和配置") - messages: List[Message] = Field(default_factory=list, description="消息列表") + """聊天文件类，包含完整的聊天记录""" + chat_metadata: ChatMetadata = Field(..., description="聊天元数据，包含基本信息和配置") + messages: List[Message] = Field(default_factory=list, description="消息列表") - class Config: - arbitrary_types_allowed = True + class Config: + arbitrary_types_allowed = True - @classmethod # 类方法装饰器，表示这是一个类方法，可以通过类名直接调用 - def load_from_file(cls, role_name: str, chat_name: str, base_path: Path = None) -> 'ChatHistory': - """ - 从JSONL文件加载聊天历史 + @classmethod # 类方法装饰器，表示这是一个类方法，可以通过类名直接调用 + def load_from_file(cls, role_name: str, chat_name: str, base_path: Path = None) -> 'ChatHistory': + """ + 从JSONL文件加载聊天历史 - 参数: - role_name: 角色名称（文件夹名） - chat_name: 聊天名称（文件名，不含扩展名） - base_path: 基础路径，默认为配置中的DATA_PATH/chat + 参数: + role_name: 角色名称（文件夹名） + chat_name: 聊天名称（文件名，不含扩展名） + base_path: 基础路径，默认为配置中的DATA_PATH/chat - 返回: - ChatHistory: 加载的聊天历史对象 + 返回: + ChatHistory: 加载的聊天历史对象 - 异常: - FileNotFoundError: 当文件不存在时抛出 - json.JSONDecodeError: 当JSON解析失败时抛出 - """ - # 设置默认基础路径 - 如果未提供base_path，则从配置中获取默认路径 - if base_path is None: - from backend.core.config import settings # 延迟导入配置模块 - base_path = settings.DATA_PATH / "chat" # 构建默认路径 + 异常: + FileNotFoundError: 当文件不存在时抛出 + json.JSONDecodeError: 当JSON解析失败时抛出 + """ + # 设置默认基础路径 - 如果未提供base_path，则从配置中获取默认路径 + if base_path is None: + from backend.core.config import settings # 延迟导入配置模块 + base_path = settings.DATA_PATH / "chat" # 构建默认路径 - # 构建文件路径 - file_path = base_path / role_name / f"{chat_name}.jsonl" + # 构建文件路径 + file_path = base_path / role_name / f"{chat_name}.jsonl" - # 检查文件是否存在 - if not file_path.exists(): - raise FileNotFoundError(f"聊天文件不存在: {file_path}") + # 检查文件是否存在 + if not file_path.exists(): + raise FileNotFoundError(f"聊天文件不存在: {file_path}") - # 初始化结果数据 - messages = [] - metadata = None + # 初始化结果数据 + messages = [] + metadata = None - # 读取文件内容 - with open(file_path, 'r', encoding='utf-8') as f: - for line_num, line in enumerate(f): - try: - line_data = json.loads(line.strip()) + # 读取文件内容 + with open(file_path, 'r', encoding='utf-8') as f: + for line_num, line in enumerate(f): + try: + line_data = json.loads(line.strip()) - # 第一行是元数据 - if line_num == 0: - metadata = ChatMetadata(**line_data) - else: - # 后续行是消息 - messages.append(Message(**line_data)) - except json.JSONDecodeError: - continue + # 第一行是元数据 + if line_num == 0: + metadata = ChatMetadata(**line_data) + else: + # 后续行是消息 + messages.append(Message(**line_data)) + except json.JSONDecodeError: + continue - # 创建并返回ChatHistory对象 - return cls( - chat_metadata=metadata or ChatMetadata(), - messages=messages - ) + # 创建并返回ChatHistory对象 + return cls( + chat_metadata=metadata or ChatMetadata(), + messages=messages + ) - @classmethod - def load_from_jsonl(cls, file_path: Path) -> 'ChatHistory': - """ - 从JSONL文件加载聊天历史 + @classmethod + def load_from_jsonl(cls, file_path: Path) -> 'ChatHistory': + """ + 从JSONL文件加载聊天历史 - 参数: - file_path: JSONL文件路径 + 参数: + file_path: JSONL文件路径 - 返回: - ChatHistory: 加载的聊天历史对象 + 返回: + ChatHistory: 加载的聊天历史对象 - 异常: - FileNotFoundError: 当文件不存在时抛出 - json.JSONDecodeError: 当JSON解析失败时抛出 - """ - # 检查文件是否存在 - if not file_path.exists(): - raise FileNotFoundError(f"聊天文件不存在: {file_path}") + 异常: + FileNotFoundError: 当文件不存在时抛出 + json.JSONDecodeError: 当JSON解析失败时抛出 + """ + # 检查文件是否存在 + if not file_path.exists(): + raise FileNotFoundError(f"聊天文件不存在: {file_path}") - # 初始化结果数据 - messages = [] - metadata = None + # 初始化结果数据 + messages = [] + metadata = None - # 读取文件内容 - with open(file_path, 'r', encoding='utf-8') as f: - for line_num, line in enumerate(f): - try: - line_data = json.loads(line.strip()) + # 读取文件内容 + with open(file_path, 'r', encoding='utf-8') as f: + for line_num, line in enumerate(f): + try: + line_data = json.loads(line.strip()) - # 第一行是元数据 - if line_num == 0: - # 处理元数据中的嵌套结构 - if 'chat_metadata' in line_data: - metadata_dict = line_data['chat_metadata'] - # 合并顶层字段和chat_metadata中的字段 - metadata_dict.update(line_data) - metadata = ChatMetadata(**metadata_dict) - else: - metadata = ChatMetadata(**line_data) - else: - # 后续行是消息 - # 处理extra字段中的内容 - extra_data = line_data.get('extra', {}) + # 第一行是元数据 + if line_num == 0: + # 处理元数据中的嵌套结构 + if 'chat_metadata' in line_data: + metadata_dict = line_data['chat_metadata'] + # 合并顶层字段和chat_metadata中的字段 + metadata_dict.update(line_data) + metadata = ChatMetadata(**metadata_dict) + else: + metadata = ChatMetadata(**line_data) + else: + # 后续行是消息 + # 处理extra字段中的内容 + extra_data = line_data.get('extra', {}) - # 如果是AI消息(is_user=False)，将extra中的某些字段提升到顶层 - if not line_data.get('is_user', True): - ai_fields = ['api', 'model', 'reasoning', 'reasoning_duration', - 'reasoning_signature', 'time_to_first_token', 'bias'] - for field in ai_fields: - if field in extra_data: - line_data[field] = extra_data.pop(field) + # 如果是AI消息(is_user=False)，将extra中的某些字段提升到顶层 + if not line_data.get('is_user', True): + ai_fields = ['api', 'model', 'reasoning', 'reasoning_duration', + 'reasoning_signature', 'time_to_first_token', 'bias'] + for field in ai_fields: + if field in extra_data: + line_data[field] = extra_data.pop(field) - # 创建Message实例 - message = Message(**line_data) - # 将剩余的extra数据保存回extra字段 - message.extra = extra_data - messages.append(message) + # 创建Message实例 + message = Message(**line_data) + # 将剩余的extra数据保存回extra字段 + message.extra = extra_data + messages.append(message) - except json.JSONDecodeError: - continue + except json.JSONDecodeError: + continue - # 创建并返回ChatHistory对象 - return cls( - chat_metadata=metadata or ChatMetadata(), - messages=messages - ) + # 创建并返回ChatHistory对象 + return cls( + chat_metadata=metadata or ChatMetadata(), + messages=messages + ) - def to_chatbox_format(self) -> List[Dict[str, Any]]: - """ + def to_chatbox_format(self) -> List[Dict[str, Any]]: + """ 将聊天历史转换为适合前端chatbox显示的格式 返回: @@ -224,26 +224,53 @@ class ChatHistory(BaseModel): "swipe_id": int } """ - # 创建消息字典列表 - messages_list = [] - for msg in self.messages: - # 获取当前消息内容：优先从swipes数组中获取，如果不存在则使用mes - current_mes = msg.mes - if msg.swipes and 0 <= msg.swipe_id < len(msg.swipes): - current_mes = msg.swipes[msg.swipe_id] + # 创建消息字典列表 + messages_list = [] + for msg in self.messages: + # 获取当前消息内容：优先从swipes数组中获取，如果不存在则使用mes + current_mes = msg.mes + if msg.swipes and 0 <= msg.swipe_id < len(msg.swipes): + current_mes = msg.swipes[msg.swipe_id] - msg_dict = { - "name": msg.name, - "is_user": msg.is_user, - "floor": msg.floor, - "mes": current_mes, - "swipes": msg.swipes, - "swipe_id": msg.swipe_id - } - messages_list.append(msg_dict) + msg_dict = { + "name": msg.name, + "is_user": msg.is_user, + "floor": msg.floor, + "mes": current_mes, + "swipes": msg.swipes, + "swipe_id": msg.swipe_id + } + messages_list.append(msg_dict) - # 按floor排序 - messages_list.sort(key=lambda x: x["floor"]) + # 按floor排序 + messages_list.sort(key=lambda x: x["floor"]) - return messages_list + return messages_list + def save_to_file(self, role_name: str, chat_name: str, base_path: Path = None) -> None: + """ + 将聊天历史保存到JSONL文件 + + 参数: + role_name: 角色名称(文件夹名) + chat_name: 聊天名称(文件名，不含扩展名) + base_path: 基础路径，默认为data/chat + """ + # 设置默认基础路径 + if base_path is None: + base_path = Path("data") + + # 构建文件路径 + file_path = base_path / role_name / f"{chat_name}.jsonl" + + # 确保目录存在 + file_path.parent.mkdir(parents=True, exist_ok=True) + + # 写入文件 + with open(file_path, 'w', encoding='utf-8') as f: + # 写入元数据 + f.write(json.dumps(self.chat_metadata.dict(), ensure_ascii=False) + '\n') + + # 写入消息 + for message in self.messages: + f.write(json.dumps(message.dict(), ensure_ascii=False) + '\n') diff --git a/frontend-react/src/Store/Slices/LeftTabsSlices/WorldBookSlice.jsx b/frontend-react/src/Store/Slices/LeftTabsSlices/WorldBookSlice.jsx new file mode 100644 index 0000000..836409b --- /dev/null +++ b/frontend-react/src/Store/Slices/LeftTabsSlices/WorldBookSlice.jsx @@ -0,0 +1,268 @@ +import { create } from 'zustand'; + +const useWorldBookStore = create((set, get) => ({ + // 世界书列表 + worldBooks: [], + // 当前选中的世界书（用于编辑） + selectedWorldBook: null, + // 全局激活的世界书列表（在槽位中显示） + globalWorldBooks: [], + // 是否显示编辑面板 + showEditPanel: false, + // 当前编辑的条目 + editingEntry: null, + // 加载状态 + isLoading: false, + // 选中世界书的加载状态 + isSelecting: false, + // 错误信息 + error: null, + // 是否显示世界书下拉框 + showWorldBookDropdown: false, + // 是否显示添加世界书下拉框 + showAddWorldBookDropdown: false, + + // 从后端获取世界书列表 + fetchWorldBooks: async () => { + set({ isLoading: true, error: null }); + try { + const response = await fetch('/api/worldbooks'); + if (!response.ok) { + throw new Error('获取世界书列表失败'); + } + const data = await response.json(); + set({ + worldBooks: data.worldbooks, + globalWorldBooks: data.worldbooks.filter(book => book.enabled), + isLoading: false + }); + } catch (error) { + set({ error: error.message, isLoading: false }); + console.error('获取世界书列表失败:', error); + } + }, + + // 切换世界书选择下拉框显示 + toggleWorldBookDropdown: () => set((state) => ({ + showWorldBookDropdown: !state.showWorldBookDropdown + })), + + // 切换添加世界书下拉框显示 + toggleAddWorldBookDropdown: () => set((state) => ({ + showAddWorldBookDropdown: !state.showAddWorldBookDropdown + })), + + // 添加世界书到全局槽位 + addGlobalWorldBook: async (uid) => { + try { + const response = await fetch(`/api/worldbooks/${uid}`, { + method: 'PUT', + headers: { + 'Content-Type': 'application/json', + }, + body: JSON.stringify({ + enabled: true + }), + }); + + if (!response.ok) { + throw new Error('添加全局世界书失败'); + } + + // 更新本地状态 + set((state) => ({ + worldBooks: state.worldBooks.map(b => + b.uid === uid ? { ...b, enabled: true } : b + ), + globalWorldBooks: [...state.globalWorldBooks, state.worldBooks.find(b => b.uid === uid)] + })); + } catch (error) { + console.error('添加全局世界书失败:', error); + throw error; + } + }, + + // 从全局槽位移除世界书 + removeGlobalWorldBook: async (uid) => { + try { + const response = await fetch(`/api/worldbooks/${uid}`, { + method: 'PUT', + headers: { + 'Content-Type': 'application/json', + }, + body: JSON.stringify({ + enabled: false + }), + }); + + if (!response.ok) { + throw new Error('移除全局世界书失败'); + } + + // 更新本地状态 + set((state) => ({ + worldBooks: state.worldBooks.map(b => + b.uid === uid ? { ...b, enabled: false } : b + ), + globalWorldBooks: state.globalWorldBooks.filter(b => b.uid !== uid) + })); + } catch (error) { + console.error('移除全局世界书失败:', error); + throw error; + } + }, + + // 创建新世界书 + createWorldBook: async (name, description) => { + try { + const response = await fetch('/api/worldbooks', { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + }, + body: JSON.stringify({ + name, + description: description || '', + enabled: false // 新建的世界书默认不全局激活 + }), + }); + + if (!response.ok) { + throw new Error('创建世界书失败'); + } + + // 重新获取世界书列表 + await get().fetchWorldBooks(); + + return await response.json(); + } catch (error) { + console.error('创建世界书失败:', error); + throw error; + } + }, + + // 删除世界书 + deleteWorldBook: async (id) => { + try { + const response = await fetch(`/api/worldbooks/${id}`, { + method: 'DELETE', + }); + + if (!response.ok) { + throw new Error('删除世界书失败'); + } + + // 更新状态 + set((state) => ({ + worldBooks: state.worldBooks.filter(book => book.uid !== id), + selectedWorldBook: state.selectedWorldBook?.uid === id ? null : state.selectedWorldBook, + globalWorldBooks: state.globalWorldBooks.filter(book => book.uid !== id) + })); + } catch (error) { + console.error('删除世界书失败:', error); + throw error; + } + }, + + // 选择世界书（用于编辑） + selectWorldBook: async (id) => { + set({ isSelecting: true, error: null }); + try { + const response = await fetch(`/api/worldbooks/${id}`); + if (!response.ok) { + throw new Error('获取世界书详情失败'); + } + const data = await response.json(); + set({ + selectedWorldBook: data, + isSelecting: false + }); + } catch (error) { + set({ error: error.message, isSelecting: false }); + console.error('获取世界书详情失败:', error); + throw error; + } + }, + + // 添加条目 + addEntry: async (entry) => { + const state = get(); + if (!state.selectedWorldBook) return; + + try { + const response = await fetch(`/api/worldbooks/${state.selectedWorldBook.uid}/entries`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + }, + body: JSON.stringify(entry), + }); + + if (!response.ok) { + throw new Error('添加条目失败'); + } + + // 重新获取选中的世界书 + await get().selectWorldBook(state.selectedWorldBook.uid); + } catch (error) { + console.error('添加条目失败:', error); + throw error; + } + }, + + // 删除条目 + deleteEntry: async (entryId) => { + const state = get(); + if (!state.selectedWorldBook) return; + + try { + const response = await fetch(`/api/worldbooks/${state.selectedWorldBook.uid}/entries/${entryId}`, { + method: 'DELETE', + }); + + if (!response.ok) { + throw new Error('删除条目失败'); + } + + // 重新获取选中的世界书 + await get().selectWorldBook(state.selectedWorldBook.uid); + } catch (error) { + console.error('删除条目失败:', error); + throw error; + } + }, + + // 更新条目 + updateEntry: async (entryId, updatedEntry) => { + const state = get(); + if (!state.selectedWorldBook) return; + + try { + const response = await fetch(`/api/worldbooks/${state.selectedWorldBook.uid}/entries/${entryId}`, { + method: 'PUT', + headers: { + 'Content-Type': 'application/json', + }, + body: JSON.stringify(updatedEntry), + }); + + if (!response.ok) { + throw new Error('更新条目失败'); + } + + // 重新获取选中的世界书 + await get().selectWorldBook(state.selectedWorldBook.uid); + } catch (error) { + console.error('更新条目失败:', error); + throw error; + } + }, + + // 切换编辑面板显示 + toggleEditPanel: (show, entry = null) => set({ + showEditPanel: show !== undefined ? show : !get().showEditPanel, + editingEntry: entry + }) +})); + +export default useWorldBookStore; diff --git a/frontend-react/src/components/SideBarLeft/tab/WorldBook.jsx b/frontend-react/src/components/SideBarLeft/tab/WorldBook.jsx index 709f2e7..ecdd4ab 100644 --- a/frontend-react/src/components/SideBarLeft/tab/WorldBook.jsx +++ b/frontend-react/src/components/SideBarLeft/tab/WorldBook.jsx @@ -1,11 +1,291 @@ -import React from 'react'; +import React, { useEffect, useState } from 'react'; import '../tabcss/WorldBook.css'; +import useWorldBookStore from '../../../Store/Slices/LeftTabsSlices/WorldBookSlice'; const WorldBook = () => { + const { + worldBooks, + selectedWorldBook, + showEditPanel, + editingEntry, + isLoading, + isSelecting, + error, + showWorldBookDropdown, + showAddWorldBookDropdown, + globalWorldBooks, + fetchWorldBooks, + toggleWorldBookDropdown, + toggleAddWorldBookDropdown, + addGlobalWorldBook, + removeGlobalWorldBook, + createWorldBook, + deleteWorldBook, + selectWorldBook, + addEntry, + deleteEntry, + updateEntry, + toggleEditPanel, + } = useWorldBookStore(); + + const [newEntry, setNewEntry] = useState({ + name: '', + content: '', + enabled: true, + triggerStrategy: 'keyword', + insertPosition: 'after', + order: 0, + }); + + useEffect(() => { + fetchWorldBooks(); + }, [fetchWorldBooks]); + + const handleCreateWorldBook = async () => { + const name = prompt('请输入世界书名称:'); + if (name) { + await createWorldBook(name); + } + }; + + const handleAddEntry = async () => { + if (!selectedWorldBook) return; + await addEntry(newEntry); + setNewEntry({ + name: '', + content: '', + enabled: true, + triggerStrategy: 'keyword', + insertPosition: 'after', + order: 0, + }); + }; + + const handleEntryClick = (entry) => { + toggleEditPanel(true, entry); + }; + + const handleEntryUpdate = async (field, value) => { + if (!editingEntry) return; + const updatedEntry = { ...editingEntry, [field]: value }; + await updateEntry(editingEntry.uid, updatedEntry); + }; + return (
-

世界书

- {/* 在这里实现世界书的具体内容 */} + {/* 全局世界书槽位 */} +
+
+ 全局世界书 +
+ + {showAddWorldBookDropdown && ( +
+
+ 新建世界书 +
+ {worldBooks + .filter(book => !globalWorldBooks.find(gb => gb.uid === book.uid)) + .map(book => ( +
{ + addGlobalWorldBook(book.uid); + toggleAddWorldBookDropdown(false); + }} + > + {book.name} +
+ ))} +
+ )} +
+
+
+ {globalWorldBooks.map(book => ( +
selectWorldBook(book.uid)} + > + {book.name} + {selectedWorldBook?.uid === book.uid && ( + { + e.stopPropagation(); + removeGlobalWorldBook(book.uid); + }} + > + ✕ + + )} +
+ ))} +
+
+ + {/* 世界书选择区域 */} +
+
+ + {showWorldBookDropdown && ( +
+ {worldBooks.map(book => ( +
{ + selectWorldBook(book.uid); + toggleWorldBookDropdown(false); + }} + > + {book.name} +
+ ))} +
+ )} +
+ {selectedWorldBook && ( + + )} +
+ + {/* 条目列表区域 */} + {isLoading ? ( +
加载中...
+ ) : error ? ( +
{error}
+ ) : selectedWorldBook ? ( +
+ {isSelecting ? ( +
加载条目中...
+ ) : selectedWorldBook.entries && selectedWorldBook.entries.length > 0 ? ( + selectedWorldBook.entries.map(entry => ( +
handleEntryClick(entry)} + > +
+ {entry.name} + + {entry.enabled ? '启用' : '禁用'} + +
+
+ 策略: {entry.triggerStrategy} + 位置: {entry.insertPosition} + 顺序: {entry.order} +
+
+ )) + ) : ( +
暂无条目
+ )} + +
+ ) : ( +
请选择一个世界书
+ )} + + {/* 编辑面板 */} + {showEditPanel && editingEntry && ( +
+
+

编辑条目

+ +
+ +
+ + handleEntryUpdate('name', e.target.value)} + /> +
+ +
+ +