重构路由架构

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)

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)

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}")

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

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')