世界书部分基本完成

This commit is contained in:
2026-04-07 01:56:13 +08:00
parent e8dedb5ec4
commit 4f9cf4b725
10 changed files with 2505 additions and 862 deletions

View File

@@ -1,8 +1,14 @@
import json
import os
import logging
from typing import Dict, List, Optional, Any
from pathlib import Path
from pydantic import BaseModel, Field, validator
from .WorldItem import WorldInfoEntry, TriggerStrategy, PositionMode
from .WorldItem import WorldInfoEntry, TriggerStrategy
from backend.core.config import settings
# 配置日志
logger = logging.getLogger(__name__)
class WorldBook(BaseModel):
@@ -11,40 +17,99 @@ class WorldBook(BaseModel):
管理多个世界书条目,支持导入导出 SillyTavern 格式
"""
# 世界书基本信息
uid: str = Field(..., description="世界书唯一标识符")
name: str = Field(..., description="世界书名称")
description: str = Field("", description="世界书描述")
# 条目集合
entries: List[WorldInfoEntry] = Field(
default_factory=list,
description="世界书条目列表"
entries: Dict[str, WorldInfoEntry] = Field(
default_factory=dict,
description="世界书条目字典对象 (Key-Value Map)"
)
# 全局配置
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]
"""验证条目 UID 的唯一性"""
uids = [entry.uid for entry in v.values()]
if len(uids) != len(set(uids)):
logger.error("验证失败: 条目 UID 必须唯一")
raise ValueError("条目 UID 必须唯一")
return v
@classmethod
def get_file_path(cls, name: str) -> str:
"""
根据世界书名称获取文件路径
Args:
name: 世界书名称
Returns:
str: 完整的文件路径
"""
# 使用配置中的 WORLDBOOKS_PATH
return str(settings.WORLDBOOKS_PATH / f"{name}.json")
@classmethod
def exists(cls, name: str) -> bool:
"""
检查指定名称的世界书文件是否存在
Args:
name: 世界书名称
Returns:
bool: 文件是否存在
"""
file_path = cls.get_file_path(name)
return os.path.exists(file_path)
@classmethod
def create_empty(cls, name: str) -> 'WorldBook':
"""
创建并保存一个空白的世界书
Args:
name: 世界书名称
Returns:
WorldBook: 创建的世界书对象
Raises:
ValueError: 世界书已存在
IOError: 文件写入失败
"""
# 检查世界书是否已存在
if cls.exists(name):
raise ValueError(f"世界书 '{name}' 已存在")
# 创建空白世界书对象
world_book = cls(
name=name,
)
# 保存世界书
world_book.save()
logger.info(f"创建空白世界书: {name}")
return world_book
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:
Raises:
ValueError: 条目 UID 已存在
"""
entry_key = str(entry.uid)
if entry_key in self.entries:
error_msg = f"添加条目失败: 条目 UID {entry.uid} 已存在于世界书 {self.name}"
logger.error(error_msg)
raise ValueError(error_msg)
self.entries[entry_key] = entry
logger.debug(f"已添加条目: UID={entry.uid}, 世界书={self.name}")
def remove_entry(self, uid: int) -> bool:
"""
移除世界书条目
@@ -54,11 +119,15 @@ class WorldBook(BaseModel):
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
entry_key = str(uid)
if entry_key in self.entries:
del self.entries[entry_key]
logger.info(f"已从世界书 {self.name} 移除条目: UID={uid}")
return True
logger.warning(f"尝试移除不存在的条目: 世界书 {self.name} 中未找到 UID={uid}")
return False
def get_entry(self, uid: str) -> Optional[WorldInfoEntry]:
def get_entry(self, uid: int) -> Optional[WorldInfoEntry]:
"""
获取指定 UID 的世界书条目
@@ -68,12 +137,15 @@ class WorldBook(BaseModel):
Returns:
Optional[WorldInfoEntry]: 找到的条目,未找到返回 None
"""
for entry in self.entries:
if entry.uid == uid:
return entry
return None
entry_key = str(uid)
entry = self.entries.get(entry_key)
if entry:
logger.debug(f"从世界书 {self.name} 获取条目: UID={uid}")
else:
logger.debug(f"在世界书 {self.name} 中未找到条目: UID={uid}")
return entry
def update_entry(self, uid: str, **kwargs) -> bool:
def update_entry(self, uid: int, **kwargs) -> bool:
"""
更新世界书条目
@@ -86,30 +158,32 @@ class WorldBook(BaseModel):
"""
entry = self.get_entry(uid)
if entry is None:
logger.warning(f"更新条目失败: 在世界书 {self.name} 中未找到 UID={uid}")
return False
for key, value in kwargs.items():
if hasattr(entry, key):
setattr(entry, key, value)
logger.info(f"已更新世界书 {self.name} 中的条目: UID={uid}, 更新字段={list(kwargs.keys())}")
return True
def filter_by_position(self, position_mode: PositionMode, position_value: int) -> List[WorldInfoEntry]:
def filter_by_position(self, position: int) -> List[WorldInfoEntry]:
"""
根据位置筛选条目
Args:
position_mode: 位置模式
position_value: 位置值
position: 位置
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))
filtered_entries = [
entry for entry in self.entries.values()
if entry.position == position
]
logger.debug(
f"在世界书 {self.name} 中按位置筛选: 值={position}, 结果数量={len(filtered_entries)}")
return filtered_entries
def get_summary(self) -> Dict[str, Any]:
"""
@@ -118,19 +192,17 @@ class WorldBook(BaseModel):
Returns:
Dict[str, Any]: 概要信息字典
"""
return {
"uid": self.uid,
summary = {
"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)
strategy.value: sum(1 for e in self.entries.values()
if strategy in e.trigger_config.get_enabled_triggers())
for strategy in TriggerStrategy
}
}
logger.debug(f"获取世界书 {self.name} 的概要信息")
return summary
def to_summary_dict(self) -> Dict[str, Any]:
"""
@@ -139,209 +211,122 @@ class WorldBook(BaseModel):
Returns:
Dict[str, Any]: 包含基本信息的字典
"""
return {
"uid": self.uid,
summary = {
"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)
logger.debug(f"生成世界书 {self.name} 的摘要信息")
return summary
def to_dict(self) -> Dict:
"""
将 WorldBook 转换为字典,兼容多种格式
将 WorldBook 转换为字典
Returns:
Dict: 世界书数据字典
"""
return {
'uid': self.uid,
result = {
'name': self.name,
'description': self.description,
'enabled': self.enabled,
'entries': {entry.uid: entry.dict() for entry in self.entries}
'entries': {uid: entry.dict() for uid, entry in self.entries.items()}
}
logger.debug(f"将世界书 {self.name} 转换为字典")
return result
@classmethod
def from_sillytavern_json(cls, file_path: str) -> 'WorldBook':
@classmethod
def load(cls, name: str) -> 'WorldBook':
"""
SillyTavern 格式的 JSON 文件加载世界书
从文件加载世界书(只有 entries 字段的格式)
Args:
file_path: JSON 文件路径
name: 世界书名称
Returns:
WorldBook: 世界书对象
Raises:
FileNotFoundError: 文件不存在
ValueError: 格式不符合 SillyTavern 标准
ValueError: 格式不符合标准
json.JSONDecodeError: JSON 解析错误
"""
file_path = cls.get_file_path(name)
if not os.path.exists(file_path):
raise FileNotFoundError(f"世界书文件未找到: {file_path}")
error_msg = f"世界书文件未找到: {file_path}"
logger.error(error_msg)
raise FileNotFoundError(error_msg)
with open(file_path, 'r', encoding='utf-8') as f:
raw_data = json.load(f)
try:
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
# 世界书名称始终使用文件名(不包括后缀名)
world_name = 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 字段
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 导出的格式。")
error_msg = "无效的世界书格式:'entries' 字段必须是一个字典。"
logger.error(error_msg)
raise ValueError(error_msg)
# 创建世界书对象
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_entry = WorldInfoEntry.from_sillytavern_data(entry_data)
world_book.add_entry(world_entry)
except Exception as e:
print(f"警告: 跳过条目 {uid},解析失败: {e}")
logger.warning(f"跳过条目 {uid},解析失败: {e}")
return world_book
logger.info(
f"从文件加载世界书: 文件={file_path}, 名称={world_name}, 条目数={len(world_book.entries)}")
def to_sillytavern_json(self, file_path: str) -> None:
return world_book
except json.JSONDecodeError as e:
error_msg = f"JSON 解析错误: {str(e)}"
logger.error(error_msg)
raise ValueError(error_msg)
except Exception as e:
error_msg = f"从文件加载世界书失败: {str(e)}"
logger.error(error_msg)
raise ValueError(error_msg)
def save(self) -> None:
"""
导出为 SillyTavern 格式的 JSON 文件
保存世界书到文件(只有 entries 字段的格式)
如果文件不存在,会创建新文件;如果文件存在,会更新现有文件
Args:
file_path: 要保存的文件路径
Raises:
IOError: 文件写入失败
"""
entries_dict = {}
file_path = self.get_file_path(self.name)
for entry in self.entries:
# 映射 WorldInfoEntry 到 SillyTavern 格式
position = entry.position_anchor if entry.position_mode == PositionMode.ANCHOR else entry.position_dx
# 确保目录存在
os.makedirs(Path(file_path).parent, exist_ok=True)
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
try:
# 转换为标准格式
entries_dict = {}
for uid, entry in self.entries.items():
entries_dict[uid] = entry.to_sillytavern_dict()
output_data = {
"entries": entries_dict
}
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)
with open(file_path, 'w', encoding='utf-8') as f:
json.dump(output_data, f, ensure_ascii=False, indent=2)
logger.info(
f"世界书已保存: 文件={file_path}, 名称={self.name}, 条目数={len(self.entries)}")
except Exception as e:
error_msg = f"保存世界书失败: {str(e)}"
logger.error(error_msg)
raise IOError(error_msg)
def list_triggers_and_content(self) -> List[Dict[str, Any]]:
"""
@@ -351,24 +336,108 @@ class WorldBook(BaseModel):
List[Dict[str, Any]]: 包含 trigger (key) 和 content 的列表
"""
result = []
for entry in self.entries:
for uid, entry in self.entries.items():
# 获取启用的触发策略
enabled_triggers = entry.trigger_config.get_enabled_triggers()
# 检查是否启用了关键词触发
keyword_enabled, keyword_config = entry.trigger_config.get_trigger(TriggerStrategy.KEYWORD)
triggers = keyword_config.key if keyword_enabled and keyword_config else []
# 检查是否启用了永久触发
constant_enabled, _ = entry.trigger_config.get_trigger(TriggerStrategy.CONSTANT)
result.append({
"uid": entry.uid,
"comment": entry.comment,
"triggers": entry.key,
"triggers": triggers,
"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
"position": entry.position,
"constant": constant_enabled,
"trigger_strategies": [strategy.value for strategy in enabled_triggers]
})
logger.debug(f"列出世界书 {self.name} 的触发词和内容: 条目数={len(result)}")
return result
def get_all_entries(self) -> List[Dict[str, Any]]:
"""
获取所有条目的核心信息(包括已禁用的条目)
Returns:
List[Dict[str, Any]]: 包含核心信息的条目列表
"""
result = []
for uid, entry in self.entries.items():
# 获取启用的触发策略
enabled_triggers = entry.trigger_config.get_enabled_triggers()
# 检查是否启用了永久触发
constant_enabled, _ = entry.trigger_config.get_trigger(TriggerStrategy.CONSTANT)
result.append({
"uid": entry.uid,
"comment": entry.comment,
"content": entry.content,
"constant": constant_enabled,
"position": entry.position,
"order": entry.order,
"role": entry.role,
"disable": entry.disable,
"trigger_strategies": [strategy.value for strategy in enabled_triggers],
"trigger_params": entry.get_trigger_params()
})
logger.debug(f"获取世界书 {self.name} 的所有条目: 条目数={len(result)}")
return result
def merge_from_book(self, other_book: 'WorldBook') -> None:
"""
从另一个世界书合并条目
Args:
other_book: 要合并的世界书对象
"""
for uid, entry in other_book.entries.items():
if uid in self.entries:
# 更新现有条目
for key, value in entry.dict().items():
if key != 'uid': # 不更新 UID
setattr(self.entries[uid], key, value)
else:
# 添加新条目
self.add_entry(entry)
logger.info(f"合并世界书: 从 {other_book.name} 合并到 {self.name}")
def to_sillytavern_json(self, file_path: str) -> None:
"""
导出为 SillyTavern 格式的 JSON 文件
Args:
file_path: 导出文件路径
"""
# 转换为 SillyTavern 格式
entries_dict = {}
for uid, entry in self.entries.items():
entries_dict[uid] = entry.to_sillytavern_dict()
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)
logger.info(f"导出世界书为 SillyTavern 格式: 文件={file_path}")
# --- 使用示例 ---
if __name__ == "__main__":
try:
# 从 SillyTavern 格式导入
world_book = WorldBook.from_sillytavern_json('entries.json')
# 创建空白世界书
world_book = WorldBook.create_empty("test_worldbook", "测试世界书")
# 加载世界书
world_book = WorldBook.load("test_worldbook")
# 打印概要
summary = world_book.get_summary()
@@ -383,9 +452,9 @@ if __name__ == "__main__":
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")
# 保存世界书
world_book.save()
print(f"\n✅ 世界书已保存")
except Exception as e:
print(f"❌ 错误: {e}")

View File

@@ -1,70 +1,389 @@
import logging
from enum import Enum
from typing import List, Optional
from typing import List, Optional, Dict, Any, Union
from pydantic import BaseModel, Field, validator
from pydantic import Extra
# 配置日志
logger = logging.getLogger(__name__)
class TriggerStrategy(str, Enum):
"""
触发策略枚举
"""
ALL = "all" # 全部触发
KEYWORD = "keyword" # 传统关键词匹配
CONSTANT = "constant" # 永久触发
KEYWORD = "keyword" # 关键词匹配触发
RAG = "rag" # 向量检索触发
CONDITION = "condition" # 逻辑条件触发
class PositionMode(str, Enum):
class KeywordTriggerConfig(BaseModel):
"""
位置模式枚举
关键词触发配置
"""
ANCHOR = "anchor" # 锚点模式 (0-5)
ABSOLUTE_DEPTH = "dx" # 绝对深度模式 (Dx)
key: List[str] = Field(default_factory=list, description="主关键词数组")
keysecondary: List[str] = Field(default_factory=list, description="次要关键词数组")
selective: bool = Field(True, description="是否开启选择性匹配")
selectiveLogic: int = Field(0, description="逻辑模式 (0=OR, 1=AND)")
matchWholeWords: bool = Field(False, description="是否全词匹配")
caseSensitive: bool = Field(False, description="是否区分大小写")
class RAGTriggerConfig(BaseModel):
"""
RAG触发配置
"""
threshold: float = Field(0.75, description="RAG 相似度阈值")
top_k: int = Field(5, description="返回的匹配条目数")
query_template: Optional[str] = Field(None, description="检索用的查询模板")
class ConditionTriggerConfig(BaseModel):
"""
条件触发配置
"""
variable_a: str = Field(..., description="变量a")
operator: str = Field(..., description="运算符 (>, <, =, >=, <=, !=)")
variable_b: str = Field(..., description="变量b")
class TriggerConfig(BaseModel):
"""
触发配置
使用字典结构,键为触发策略,值为[是否启用, 对应配置]的列表
"""
triggers: Dict[TriggerStrategy, List[
Union[bool, Optional[Union[KeywordTriggerConfig, RAGTriggerConfig, ConditionTriggerConfig]]]]] = Field(
default_factory=lambda: {
TriggerStrategy.CONSTANT: [True, None],
TriggerStrategy.KEYWORD: [False, None],
TriggerStrategy.RAG: [False, None],
TriggerStrategy.CONDITION: [False, None]
},
description="触发配置字典,键为触发策略,值为[是否启用, 对应配置]"
)
class Config:
extra = Extra.forbid # 禁止额外字段
def set_trigger(self, strategy: TriggerStrategy, enabled: bool,
config: Optional[Union[KeywordTriggerConfig, RAGTriggerConfig, ConditionTriggerConfig]] = None
):
"""
设置触发策略
Args:
strategy: 触发策略
enabled: 是否启用
config: 对应的配置对象
"""
self.triggers[strategy] = [enabled, config]
def get_trigger(self, strategy: TriggerStrategy) -> List[
Union[bool, Optional[Union[KeywordTriggerConfig, RAGTriggerConfig, ConditionTriggerConfig]]]]:
"""
获取触发策略
Args:
strategy: 触发策略
Returns:
List: [是否启用, 对应配置]
"""
return self.triggers.get(strategy, [False, None])
def get_enabled_triggers(self) -> List[TriggerStrategy]:
"""
获取所有启用的触发策略
Returns:
List[TriggerStrategy]: 启用的触发策略列表
"""
return [strategy for strategy, (enabled, _) in self.triggers.items() if enabled]
class WorldInfoEntry(BaseModel):
"""
世界书条目模型 v2.0
支持 RAG、条件触发及 Dx 绝对位置
兼容 SillyTavern 格式
世界书条目模型
完整兼容所有可能的属性字段
"""
# --- 核心身份 ---
uid: str
key: List[str] = []
keysecondary: List[str] = []
comment: str = ""
content: str
# A. 基础定义 (Base Fields)
uid: int = Field(..., description="唯一标识符")
key: List[str] = Field(default_factory=list, description="条目名")
content: str = Field(..., description="注入到 Prompt 的实际文本内容")
comment: str = Field("", description="备注")
# --- 策略配置 ---
trigger_strategy: TriggerStrategy = TriggerStrategy.KEYWORD
condition_expr: Optional[str] = None # 条件表达式
rag_threshold: float = 0.75 # RAG 相似度阈值
# B. 注入与排序 (Injection & Order)
position: int = Field(0, description="插入位置")
order: int = Field(100, description="注入顺序权重")
depth: int = Field(4, description="扫描深度")
scanDepth: Optional[int] = Field(None, description="显式扫描深度")
addMemo: bool = Field(False, description="是否添加备忘录标记")
# --- 位置与扫描 ---
position_mode: PositionMode = PositionMode.ANCHOR
position_anchor: int = 0 # 锚点值 (0-5)
position_dx: Optional[int] = None # 绝对深度值
scan_depth: int = 4 # 向前扫描 N 条消息
# C. 触发配置 (Trigger Configuration)
trigger_config: TriggerConfig = Field(
default_factory=TriggerConfig,
description="触发配置"
)
# --- 其他配置 ---
constant: bool = False
selective: bool = True
order: int = 100
probability: int = 100
useProbability: bool = False
# D. 高级匹配 (Advanced Matching)
role: int = Field(0, description="角色匹配 (0=Both, 1=User, 2=Assistant)")
useGroupScoring: bool = Field(False, description="是否使用分组评分")
# 高级过滤
group: Optional[str] = None
match_whole_words: bool = False
use_regex: bool = False
vectorized: bool = False # 是否已生成向量 embedding
# E. 分组设置 (Grouping)
group: Optional[str] = Field(None, description="分组名称")
groupOverride: bool = Field(False, description="是否覆盖分组限制")
groupWeight: int = Field(100, description="分组权重")
# SillyTavern 特定字段
enabled: bool = True
position: str = "after_char" # SillyTavern 位置字符串
insertion_order: int = 100 # SillyTavern 插入顺序
# F. 递归与防抖 (Recursion)
excludeRecursion: bool = Field(True, description="排除递归")
preventRecursion: bool = Field(True, description="防止递归")
delayUntilRecursion: bool = Field(False, description="延迟直到递归发生")
@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
# G. 状态与元数据 (State & Metadata)
disable: bool = Field(False, description="是否禁用该条目")
ignoreBudget: bool = Field(False, description="忽略 Token 预算")
outletName: Optional[str] = Field(None, description="出口名称")
automationId: Optional[str] = Field(None, description="自动化 ID")
sticky: int = Field(0, description="粘性值")
cooldown: int = Field(0, description="冷却时间")
delay: int = Field(0, description="延迟触发")
triggers: List[str] = Field(default_factory=list, description="触发器数组")
displayIndex: int = Field(0, description="UI 显示索引")
vectorized: bool = Field(False, description="是否已向量化")
# H. 兼容性字段 (Compatibility)
matchPersonaDescription: bool = Field(False, description="匹配人设描述")
matchCharacterDescription: bool = Field(False, description="匹配角色描述")
matchCharacterPersonality: bool = Field(False, description="匹配角色性格")
matchCharacterDepthPrompt: bool = Field(False, description="匹配深度提示词")
matchScenario: bool = Field(False, description="匹配场景")
matchCreatorNotes: bool = Field(False, description="匹配作者注释")
# I. 嵌套对象 (Nested Objects)
characterFilter: Optional[Dict[str, Any]] = Field(None, description="角色过滤器配置")
extensions: Optional[Dict[str, Any]] = Field(None, description="扩展属性")
@classmethod
def from_sillytavern_data(cls, data: Dict[str, Any]) -> 'WorldInfoEntry':
"""
从 SillyTavern 格式的数据创建条目
Args:
data: SillyTavern 格式的条目数据
Returns:
WorldInfoEntry: 世界书条目对象
Raises:
ValueError: 数据格式无效
"""
try:
# 提取基本字段
uid = int(data.get("uid", data.get("id", 0))) # 条目唯一标识符
key = data.get("key", data.get("keys", [])) # 条目触发关键词数组,用于匹配和触发该条目
keysecondary = data.get("keysecondary", data.get("secondary_keys", [])) # 次要关键词数组,用于扩展触发条件
comment = data.get("comment", "") # 条目备注说明,用于描述条目的用途和特点
content = data.get("content", "") # 条目的实际内容,将被注入到提示词中
constant = data.get("constant", False) # 是否常驻注入true表示始终注入
selective = data.get("selective", True) # 是否启用选择性匹配true表示仅关键词匹配时注入
order = data.get("order", data.get("insertion_order", 100)) # 条目注入顺序权重,数字越小越优先
# 处理 position 字段,确保为整数类型
position = data.get("position", 0)
if isinstance(position, str):
# 如果是字符串,尝试转换为整数
try:
position = int(position)
except ValueError:
# 如果转换失败,使用默认值
logger.warning(f"条目 {uid} 的 position 字段值 '{position}' 无法转换为整数,使用默认值 0")
position = 0
# 处理触发配置,默认使用永久触发
try:
trigger_config = TriggerConfig()
# 设置永久触发
trigger_config.set_trigger(TriggerStrategy.CONSTANT, constant)
# 设置关键词触发
if not constant and key:
keyword_config = KeywordTriggerConfig(
key=key,
keysecondary=keysecondary,
selective=selective,
selectiveLogic=data.get("selectiveLogic", 0),
matchWholeWords=data.get("matchWholeWords", False),
caseSensitive=data.get("caseSensitive", False)
)
trigger_config.set_trigger(TriggerStrategy.KEYWORD, True, keyword_config)
# 设置RAG触发
if "rag_threshold" in data or "top_k" in data or "query_template" in data:
rag_config = RAGTriggerConfig(
threshold=data.get("rag_threshold", 0.75),
top_k=data.get("top_k", 5),
query_template=data.get("query_template", None)
)
trigger_config.set_trigger(TriggerStrategy.RAG, True, rag_config)
# 设置条件触发
if "variable_a" in data and "operator" in data and "variable_b" in data:
condition_config = ConditionTriggerConfig(
variable_a=data.get("variable_a", ""),
operator=data.get("operator", "="),
variable_b=data.get("variable_b", "")
)
trigger_config.set_trigger(TriggerStrategy.CONDITION, True, condition_config)
except Exception as e:
# 如果触发配置解析失败,使用默认的永久触发配置
logger.warning(f"条目 {uid} 的触发配置解析失败: {str(e)},使用默认的永久触发配置")
trigger_config = TriggerConfig()
trigger_config.set_trigger(TriggerStrategy.CONSTANT, True)
# 处理其他字段
probability = data.get("probability", 100)
useProbability = data.get("useProbability", False)
group = data.get("group", None)
use_regex = data.get("useRegex", False)
# 处理 extensions 字段(如果存在)
extensions = data.get("extensions", {})
if extensions:
probability = extensions.get("probability", probability)
useProbability = extensions.get("useProbability", useProbability)
group = extensions.get("group", group)
# 创建条目对象
entry = cls(
uid=uid,
key=key, # 保留key字段用于兼容
comment=comment,
content=content,
position=position,
order=order,
trigger_config=trigger_config,
group=group,
probability=probability,
useProbability=useProbability,
use_regex=use_regex
)
# 复制其他字段(如果存在)
for key, value in data.items():
if hasattr(entry, key) and key not in ["uid", "key", "comment", "content",
"constant", "position", "order", "probability",
"useProbability", "group", "use_regex",
"keysecondary", "selective", "selectiveLogic",
"matchWholeWords", "caseSensitive"]:
setattr(entry, key, value)
return entry
except Exception as e:
logger.error(f"从 SillyTavern 数据创建条目失败: {str(e)}")
raise ValueError(f"从 SillyTavern 数据创建条目失败: {str(e)}")
def to_sillytavern_dict(self) -> Dict[str, Any]:
"""
转换为 SillyTavern 格式的字典
Returns:
Dict[str, Any]: SillyTavern 格式的条目数据
"""
result = {
"uid": self.uid,
"key": self.key,
"comment": self.comment,
"content": self.content,
"position": self.position,
"order": self.order,
"probability": self.probability,
"useProbability": self.useProbability,
"group": self.group,
"useRegex": getattr(self, 'use_regex', False),
"preventRecursion": self.preventRecursion,
"excludeRecursion": self.excludeRecursion
}
# 根据触发策略设置对应的字段
try:
# 检查永久触发是否启用
constant_enabled, _ = self.trigger_config.get_trigger(TriggerStrategy.CONSTANT)
result["constant"] = constant_enabled
# 检查关键词触发是否启用
keyword_enabled, keyword_config = self.trigger_config.get_trigger(TriggerStrategy.KEYWORD)
if keyword_enabled and keyword_config:
result["selective"] = keyword_config.selective
result["keysecondary"] = keyword_config.keysecondary
result["selectiveLogic"] = keyword_config.selectiveLogic
result["matchWholeWords"] = keyword_config.matchWholeWords
result["caseSensitive"] = keyword_config.caseSensitive
# 检查RAG触发是否启用
rag_enabled, rag_config = self.trigger_config.get_trigger(TriggerStrategy.RAG)
if rag_enabled and rag_config:
result["rag_threshold"] = rag_config.threshold
result["top_k"] = rag_config.top_k
result["query_template"] = rag_config.query_template
# 检查条件触发是否启用
condition_enabled, condition_config = self.trigger_config.get_trigger(TriggerStrategy.CONDITION)
if condition_enabled and condition_config:
result["variable_a"] = condition_config.variable_a
result["operator"] = condition_config.operator
result["variable_b"] = condition_config.variable_b
except Exception as e:
# 如果触发配置导出失败,使用默认的永久触发配置
logger.warning(f"条目 {self.uid} 的触发配置导出失败: {str(e)},使用默认的永久触发配置")
result["constant"] = True
return result
def get_trigger_params(self) -> Dict[str, Any]:
"""
获取触发策略所需的参数
Returns:
Dict[str, Any]: 触发参数字典
"""
params = {}
try:
# 获取所有启用的触发策略
enabled_triggers = self.trigger_config.get_enabled_triggers()
# 处理关键词触发
if TriggerStrategy.KEYWORD in enabled_triggers:
_, keyword_config = self.trigger_config.get_trigger(TriggerStrategy.KEYWORD)
if keyword_config:
params["key"] = keyword_config.key
params["keysecondary"] = keyword_config.keysecondary
params["selective"] = keyword_config.selective
params["selectiveLogic"] = keyword_config.selectiveLogic
params["matchWholeWords"] = keyword_config.matchWholeWords
params["caseSensitive"] = keyword_config.caseSensitive
# 处理RAG触发
if TriggerStrategy.RAG in enabled_triggers:
_, rag_config = self.trigger_config.get_trigger(TriggerStrategy.RAG)
if rag_config:
params["threshold"] = rag_config.threshold
params["top_k"] = rag_config.top_k
params["query_template"] = rag_config.query_template
params["vectorized"] = self.vectorized
# 处理条件触发
if TriggerStrategy.CONDITION in enabled_triggers:
_, condition_config = self.trigger_config.get_trigger(TriggerStrategy.CONDITION)
if condition_config:
params["variable_a"] = condition_config.variable_a
params["operator"] = condition_config.operator
params["variable_b"] = condition_config.variable_b
except Exception as e:
# 如果获取触发参数失败,返回空字典,表示使用默认的永久触发
logger.warning(f"条目 {self.uid} 的触发参数获取失败: {str(e)},使用默认的永久触发")
return params

View File

@@ -102,7 +102,7 @@ class ChatHistory(BaseModel):
"""获取所有角色的所有聊天列表"""
data_dir = cls.get_data_path()
if not data_dir.exists():
return {"chats": []}
return {"chat": []}
chats = []
for role_dir in data_dir.iterdir():
@@ -123,7 +123,7 @@ class ChatHistory(BaseModel):
})
except Exception:
continue # 跳过损坏的聊天文件
return {"chats": chats}
return {"chat": chats}
@classmethod
async def get_chat(cls, role_name: str, chat_name: str) -> Dict[str, Any]: