正则相关

This commit is contained in:
2026-05-06 00:40:18 +08:00
parent f843a74715
commit 9faccc2c03
32 changed files with 4814 additions and 346 deletions

View File

@@ -389,36 +389,95 @@ async def _save_messages(
try:
from datetime import datetime
# 1. 保存用户消息
user_message = {
"id": f"msg_{datetime.now().timestamp()}_user",
"name": request_data.get("userName", "User"),
"is_user": True,
"is_system": False,
"sendDate": datetime.now().isoformat(),
"mes": request_data.get("mes", ""),
"chatId": f"{role_name}/{chat_name}",
"floor": request_data.get("floor", 0)
}
# ✅ 检查是否是重roll模式targetFloor 存在且不为 null
target_floor = request_data.get("floor")
is_reroll = target_floor is not None
chat_service.add_message(role_name, chat_name, user_message)
# 2. 保存AI回复
ai_message = {
"id": f"msg_{datetime.now().timestamp()}_ai",
"name": request_data.get("characterName", role_name),
"is_user": False,
"is_system": False,
"sendDate": datetime.now().isoformat(),
"mes": ai_response,
"chatId": f"{role_name}/{chat_name}",
"floor": request_data.get("floor", 0) + 1
}
chat_service.add_message(role_name, chat_name, ai_message)
print(f"[WebSocket] 消息已保存: {role_name}/{chat_name}")
if is_reroll:
# ✅ 重roll模式更新现有消息的 swipes 数组
print(f"[WebSocket] 🔄 重roll模式更新楼层 {target_floor} 的 swipes")
# 获取现有的消息
existing_message = chat_service.get_message(role_name, chat_name, target_floor)
if not existing_message:
print(f"[WebSocket] ⚠️ 找不到楼层 {target_floor} 的消息,创建新消息")
# 如果找不到,创建新消息(兼容处理)
ai_message = {
"id": f"msg_{datetime.now().timestamp()}_ai",
"name": request_data.get("characterName", role_name),
"is_user": False,
"is_system": False,
"sendDate": datetime.now().isoformat(),
"mes": ai_response,
"chatId": f"{role_name}/{chat_name}",
"floor": target_floor,
"swipes": [ai_response],
"swipe_id": 0
}
chat_service.add_message(role_name, chat_name, ai_message)
else:
# ✅ 更新 swipes 数组
existing_swipes = existing_message.get("swipes", [])
current_mes = existing_message.get("mes", "")
# 构建新的 swipes 数组
updated_swipes = list(existing_swipes) # 复制现有swipes
# 如果当前 mes 不在 swipes 中,先添加它
if current_mes and current_mes not in updated_swipes:
updated_swipes.append(current_mes)
print(f"[WebSocket] 📝 将当前内容添加到 swipes")
# 添加新生成的内容
updated_swipes.append(ai_response)
print(f"[WebSocket] 📊 Swipes 更新: {len(existing_swipes)} -> {len(updated_swipes)}")
# 更新消息
update_data = {
"mes": ai_response, # 显示最新内容
"swipes": updated_swipes, # 更新 swipes 数组
"swipe_id": len(updated_swipes) - 1 # 自动切换到新版本
}
chat_service.update_message(role_name, chat_name, target_floor, update_data)
print(f"[WebSocket] ✅ 楼层 {target_floor} 已更新swipes 数量: {len(updated_swipes)}")
else:
# ✅ 正常模式创建新的用户消息和AI消息
print(f"[WebSocket] 正常模式,创建新消息")
# 1. 保存用户消息
user_message = {
"id": f"msg_{datetime.now().timestamp()}_user",
"name": request_data.get("userName", "User"),
"is_user": True,
"is_system": False,
"sendDate": datetime.now().isoformat(),
"mes": request_data.get("mes", ""),
"chatId": f"{role_name}/{chat_name}",
"floor": request_data.get("floor", 0)
}
chat_service.add_message(role_name, chat_name, user_message)
# 2. 保存AI回复
ai_message = {
"id": f"msg_{datetime.now().timestamp()}_ai",
"name": request_data.get("characterName", role_name),
"is_user": False,
"is_system": False,
"sendDate": datetime.now().isoformat(),
"mes": ai_response,
"chatId": f"{role_name}/{chat_name}",
"floor": request_data.get("floor", 0) + 1
}
chat_service.add_message(role_name, chat_name, ai_message)
print(f"[WebSocket] ✅ 新消息已保存: {role_name}/{chat_name}")
except Exception as e:
print(f"[WebSocket] 保存消息失败: {e}")
import traceback
traceback.print_exc()
# 不抛出异常,避免影响主流程

View File

@@ -15,7 +15,7 @@ from services.system_settings_service import system_settings_service
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/regex", tags=["regex"])
router = APIRouter(prefix="/regex", tags=["regex"])
# ==================== 数据模型 ====================
@@ -146,13 +146,68 @@ async def add_rule(request: RuleUpdateRequest):
@router.delete("/rules/{rule_id}")
async def delete_rule(rule_id: str):
"""删除规则(需要指定作用域和名称)"""
# TODO: 实现具体的删除逻辑
return {
"success": False,
"message": "暂未实现,请使用规则集级别的删除"
}
async def delete_rule(rule_id: str, scope: str = "global", name: Optional[str] = None):
"""
删除规则
Args:
rule_id: 规则ID
scope: 作用域 (global/character/preset)
name: 角色名或预设名scope 为 character/preset 时需要)
"""
try:
from models.regex_rules import RegexScope
scope_map = {
"global": RegexScope.GLOBAL,
"character": RegexScope.CHARACTER,
"preset": RegexScope.PRESET
}
scope_enum = scope_map.get(scope, RegexScope.GLOBAL)
# 找到包含该规则的规则集
if scope_enum == RegexScope.GLOBAL:
rulesets = regex_service.global_rulesets
elif scope_enum == RegexScope.CHARACTER:
if not name:
raise ValueError("删除角色规则需要提供角色名称")
rulesets = {name: regex_service.character_rulesets.get(name)} if name in regex_service.character_rulesets else {}
elif scope_enum == RegexScope.PRESET:
if not name:
raise ValueError("删除预设规则需要提供预设名称")
rulesets = {name: regex_service.preset_rulesets.get(name)} if name in regex_service.preset_rulesets else {}
# 查找并删除规则
deleted = False
for ruleset_name, ruleset in rulesets.items():
if not ruleset:
continue
original_count = len(ruleset.rules)
ruleset.rules = [r for r in ruleset.rules if r.id != rule_id]
if len(ruleset.rules) < original_count:
# 保存更新后的规则集
regex_service.save_ruleset(ruleset, scope_enum, ruleset_name if scope_enum != RegexScope.GLOBAL else None)
deleted = True
break
if not deleted:
return {
"success": False,
"message": "未找到指定的规则"
}
# 重新加载规则
regex_service._load_all_rules()
return {
"success": True,
"message": "规则已删除"
}
except Exception as e:
logger.error(f"删除规则失败: {e}")
raise HTTPException(status_code=500, detail=str(e))
# ==================== 规则导入导出 ====================
@@ -160,7 +215,7 @@ async def delete_rule(rule_id: str):
@router.post("/import")
async def import_rules(file: UploadFile = File(...)):
"""
导入正则规则(支持 SillyTavern 格式)
导入正则规则(支持 SillyTavern 格式)- 文件上传方式
可以导入:
1. 单个规则文件JSON 数组)
@@ -197,6 +252,56 @@ async def import_rules(file: UploadFile = File(...)):
raise HTTPException(status_code=500, detail=str(e))
@router.post("/import-from-preset")
async def import_rules_from_preset(request: dict):
"""
从预设导入正则规则 - JSON 数据方式
Request Body:
{
"rules": [...], // SillyTavern 格式的 regex_scripts 数组
"scope": "preset", // 作用域global/character/preset
"presetName": "预设名称" // 当 scope 为 preset 时需要
}
"""
try:
rules_data = request.get("rules", [])
scope_str = request.get("scope", "global")
preset_name = request.get("presetName")
if not rules_data or not isinstance(rules_data, list):
raise ValueError("无效的规则数据")
# 转换作用域字符串为枚举
scope_map = {
"global": RegexScope.GLOBAL,
"character": RegexScope.CHARACTER,
"preset": RegexScope.PRESET
}
scope = scope_map.get(scope_str, RegexScope.GLOBAL)
# 转换 SillyTavern 格式
name = preset_name or "imported_rules"
ruleset = regex_service._convert_sillytavern_format(rules_data, name, scope)
# 保存规则集
regex_service.save_ruleset(ruleset, scope, name)
# 重新加载规则
regex_service._load_all_rules()
logger.info(f"✅ 从预设导入 {len(rules_data)} 条正则规则到 {scope.value}: {name}")
return {
"success": True,
"message": f"成功导入 {len(rules_data)} 条正则规则",
"rulesetId": ruleset.id
}
except Exception as e:
logger.error(f"从预设导入规则失败: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.get("/export/global")
async def export_global_rules():
"""导出所有全局规则"""

View File

@@ -84,7 +84,7 @@ class RegexRule(BaseModel):
runOnEdit: bool = Field(True, description="用户编辑消息时是否重新应用")
# 消息深度控制
minDepth: int = Field(0, ge=0, description="最小消息深度(从最新消息开始计数)")
minDepth: Optional[int] = Field(None, ge=0, description="最小消息深度(从最新消息开始计数None 表示无限制")
maxDepth: Optional[int] = Field(None, ge=0, description="最大消息深度None 表示无限制)")
# 作用域配置

View File

@@ -172,6 +172,41 @@ class ChatService:
logger.error(f"读取聊天失败 {role_name}/{chat_name}: {str(e)}")
raise
def get_message(self, role_name: str, chat_name: str, floor: int) -> Dict:
"""
获取指定楼层的消息
Args:
role_name: 角色名称
chat_name: 聊天名称
floor: 楼层号
Returns:
Dict: 消息数据,如果不存在则返回 None
"""
chat_file = self.chat_dir / role_name / f"{chat_name}.jsonl"
if not chat_file.exists():
return None
try:
with open(chat_file, 'r', encoding='utf-8') as f:
lines = f.readlines()
# 找到对应的消息行floor + 1因为第0行是header
message_line_index = floor + 1
if message_line_index >= len(lines):
return None
# 解析并返回消息
msg_data = json.loads(lines[message_line_index])
return msg_data
except Exception as e:
logger.error(f"获取消息失败 {role_name}/{chat_name}/{floor}: {str(e)}")
return None
def create_chat(self, role_name: str, chat_name: str, metadata: Dict = None) -> Dict:
"""
创建新聊天

View File

@@ -156,6 +156,29 @@ class ChatWorkflowService:
print(f"[ChatWorkflow] 开始处理请求: role={current_role}, chat={current_chat}")
# ✅ 获取预设名称(用于加载预设绑定的正则规则)
preset_config = request_data.get("presetConfig", {})
preset_name = preset_config.get("selectedPreset")
# ✅ 第1.5步:应用用户输入的正则规则
from services.regex_service import regex_service
from models.regex_rules import RegexPlacement
processed_user_message = regex_service.apply_rules_by_placement(
text=user_message,
placement=RegexPlacement.USER_INPUT.value,
character_name=current_role,
preset_name=preset_name,
message_depth=0,
is_for_llm=True, # ✅ 用户输入会发送给LLM
is_markdown_rendered=False
)
if processed_user_message != user_message:
print(f"[Regex] ✅ 已应用用户输入正则规则")
user_message = processed_user_message
# === 第2步加载角色卡 ===
character_data = request_data.get("characterData")
if not character_data:
@@ -217,6 +240,21 @@ class ChatWorkflowService:
print(f"[ChatWorkflow] 生成完成,内容长度: {len(generated_content)}")
print(f"[ChatWorkflow] Token 使用: {token_usage}")
# ✅ 第6.5步:应用 AI 输出的正则规则
processed_ai_output = regex_service.apply_rules_by_placement(
text=generated_content,
placement=RegexPlacement.AI_OUTPUT.value,
character_name=current_role,
preset_name=preset_name,
message_depth=0,
is_for_llm=False, # ✅ AI输出是显示给用户的
is_markdown_rendered=False
)
if processed_ai_output != generated_content:
print(f"[Regex] ✅ 已应用 AI 输出正则规则")
generated_content = processed_ai_output
# === 第7步记录 Token 使用 ===
chat_id = f"{current_role}/{request_data.get('currentChat', '')}"
floor = request_data.get("floor", 0)
@@ -710,30 +748,243 @@ class ChatWorkflowService:
"""
组装提示词
使用PromptAssembler将角色卡、世界书、聊天历史等组装LLM消息列表
根据预设组件promptComponents动态组装LLM消息列表
"""
# 从预设配置中获取prompt components
preset_config = request_data.get("presetConfig", {})
prompt_components = preset_config.get("promptComponents", [])
# 创建配置
config = PromptConfig(
an_position="after_history", # 默认在历史之后
an_depth=4,
post_history_instructions=None
)
# ✅ 检查是否启用调试模式
debug_prompt = request_data.get("debugPrompt", False)
# 组装提示词
messages = self.prompt_assembler.assemble(
character=character,
chat_history=chat_history,
user_input=user_message,
active_entries=active_entries,
config=config
)
if debug_prompt:
print(f"\n{'='*80}")
print(f"[Prompt Debug] 🧩 预设组件配置")
print(f"{'='*80}")
print(f"[Prompt Debug] 组件数量: {len(prompt_components)}")
for i, comp in enumerate(prompt_components, 1):
print(f" {i}. {comp.get('name', 'Unknown')} (enabled={comp.get('enabled', True)}, type={comp.get('type', 'N/A')})")
print(f"{'='*80}\n")
# ✅ 如果有预设组件,使用预设组件组装
if prompt_components and len(prompt_components) > 0:
return self._assemble_prompt_from_components(
character,
chat_history,
user_message,
active_entries,
prompt_components,
debug_prompt
)
else:
# ✅ 否则使用默认的 SillyTavern 规范组装
print(f"[PromptAssembler] ⚠️ 未检测到预设组件使用默认SillyTavern规范")
# 创建配置
config = PromptConfig(
an_position="after_history", # 默认在历史之后
an_depth=4,
post_history_instructions=None
)
# 组装提示词
messages = self.prompt_assembler.assemble(
character=character,
chat_history=chat_history,
user_input=user_message,
active_entries=active_entries,
config=config
)
return messages
def _assemble_prompt_from_components(
self,
character,
chat_history: List,
user_message: str,
active_entries: List,
prompt_components: List[Dict],
debug_prompt: bool = False
) -> List:
"""
根据预设组件组装提示词
Args:
character: 角色卡数据
chat_history: 聊天历史
user_message: 用户输入
active_entries: 激活的世界书条目
prompt_components: 预设组件列表
debug_prompt: 是否输出调试信息
Returns:
List[BaseMessage]: 组装好的消息列表
"""
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
messages = []
# ✅ 按顺序处理每个启用的组件
enabled_components = [comp for comp in prompt_components if comp.get('enabled', True)]
if debug_prompt:
print(f"\n[Prompt Debug] 🔄 开始组装 {len(enabled_components)} 个启用的组件")
for i, component in enumerate(enabled_components, 1):
comp_name = component.get('name', 'Unknown')
comp_type = component.get('type', 'text') # text, system, user, assistant
comp_content = component.get('content', '')
if debug_prompt:
print(f"\n[Prompt Debug] --- 组件 {i}: {comp_name} ---")
print(f" 类型: {comp_type}")
print(f" 内容长度: {len(comp_content)} 字符")
# ✅ 替换模板变量
processed_content = self._process_component_content(
comp_content,
character,
chat_history,
user_message,
active_entries
)
if debug_prompt:
print(f" 处理后长度: {len(processed_content)} 字符")
if len(processed_content) < 500:
print(f" 完整内容:\n{processed_content}")
else:
print(f" 预览:\n{processed_content[:200]}...")
# ✅ 跳过空内容
if not processed_content or processed_content.strip() == "":
if debug_prompt:
print(f" ⚠️ 内容为空,跳过")
continue
# ✅ 根据组件类型创建对应的消息
if comp_type == 'system':
messages.append(SystemMessage(content=processed_content))
elif comp_type == 'user':
messages.append(HumanMessage(content=processed_content))
elif comp_type == 'assistant':
messages.append(AIMessage(content=processed_content))
else:
# 默认为 system
messages.append(SystemMessage(content=processed_content))
# ✅ 最后添加用户输入(如果还没有添加)
# 检查最后一个消息是否是用户输入
if messages and isinstance(messages[-1], HumanMessage):
# 已经包含用户输入,不需要再添加
pass
else:
# 添加用户输入
messages.append(HumanMessage(content=user_message))
if debug_prompt:
print(f"\n[Prompt Debug] ✅ 组装完成,总消息数: {len(messages)}")
print(f"{'='*80}\n")
return messages
def _process_component_content(
self,
content: str,
character,
chat_history: List,
user_message: str,
active_entries: List
) -> str:
"""
处理组件内容,替换模板变量
支持的变量:
- {{char}}: 角色名称
- {{user}}: 用户名称(暂时用 User
- {{description}}: 角色描述
- {{personality}}: 角色性格
- {{scenario}}: 场景
- {{mes_example}}: 对话示例
- {{first_mes}}: 第一条消息
- {{history}}: 聊天历史
- {{world_info}}: 世界书信息
Args:
content: 原始内容
character: 角色卡数据
chat_history: 聊天历史
user_message: 用户输入
active_entries: 激活的世界书条目
Returns:
str: 处理后的内容
"""
if not content:
return ""
# 替换角色相关变量
content = content.replace('{{char}}', getattr(character, 'name', 'Character'))
content = content.replace('{{user}}', 'User') # TODO: 从配置获取用户名
content = content.replace('{{description}}', getattr(character, 'description', ''))
content = content.replace('{{personality}}', getattr(character, 'personality', ''))
content = content.replace('{{scenario}}', getattr(character, 'scenario', ''))
content = content.replace('{{mes_example}}', getattr(character, 'mes_example', ''))
content = content.replace('{{first_mes}}', getattr(character, 'first_mes', ''))
# 替换聊天历史
if '{{history}}' in content:
history_text = self._format_chat_history(chat_history)
content = content.replace('{{history}}', history_text)
# 替换世界书信息
if '{{world_info}}' in content and active_entries:
world_info_text = self._format_world_info(active_entries)
content = content.replace('{{world_info}}', world_info_text)
return content
def _format_chat_history(self, chat_history: List) -> str:
"""
格式化聊天历史为文本
Args:
chat_history: 聊天历史列表
Returns:
str: 格式化后的历史文本
"""
lines = []
for msg in chat_history:
# ✅ 过滤已被总结的空消息
if hasattr(msg, 'is_summarized') and msg.is_summarized and (msg.mes == "" or msg.mes.strip() == ""):
continue
name = getattr(msg, 'name', 'User' if msg.is_user else 'Assistant')
mes = getattr(msg, 'mes', '')
lines.append(f"{name}: {mes}")
return "\n".join(lines)
def _format_world_info(self, active_entries: List) -> str:
"""
格式化世界书信息为文本
Args:
active_entries: 激活的世界书条目列表
Returns:
str: 格式化后的世界书文本
"""
lines = []
for entry in active_entries:
name = getattr(entry, 'name', 'Unknown')
content = getattr(entry, 'content', '')
lines.append(f"[{name}]\n{content}")
return "\n\n".join(lines)
async def _generate_response(
self,
prompt_messages: List,
@@ -978,6 +1229,29 @@ class ChatWorkflowService:
print(f" - Message Length: {len(user_message)}")
print(f"{'#'*80}\n")
# ✅ 获取预设名称(用于加载预设绑定的正则规则)
preset_config = request_data.get("presetConfig", {})
preset_name = preset_config.get("selectedPreset")
# ✅ 第1.5步:应用用户输入的正则规则
from services.regex_service import regex_service
from models.regex_rules import RegexPlacement
processed_user_message = regex_service.apply_rules_by_placement(
text=user_message,
placement=RegexPlacement.USER_INPUT.value,
character_name=current_role,
preset_name=preset_name,
message_depth=0,
is_for_llm=True, # ✅ 用户输入会发送给LLM
is_markdown_rendered=False
)
if processed_user_message != user_message:
print(f"[Regex] ✅ 已应用用户输入正则规则")
user_message = processed_user_message
# === 第2步加载角色卡 ===
character_data = request_data.get("characterData")
if not character_data:
@@ -1109,6 +1383,21 @@ class ChatWorkflowService:
print(f" - 平均速度: {len(generated_content)/elapsed if elapsed > 0 else 0:.0f} chars/s")
print(f"{'#'*80}\n")
# ✅ 第6.5步:应用 AI 输出的正则规则
processed_ai_output = regex_service.apply_rules_by_placement(
text=generated_content,
placement=RegexPlacement.AI_OUTPUT.value,
character_name=current_role,
preset_name=preset_name,
message_depth=0,
is_for_llm=False, # ✅ AI输出是显示给用户的
is_markdown_rendered=False
)
if processed_ai_output != generated_content:
print(f"[Regex] ✅ 已应用 AI 输出正则规则")
generated_content = processed_ai_output
# === 第7步记录 Token 使用(估算) ===
chat_id = f"{current_role}/{request_data.get('currentChat', '')}"
floor = request_data.get("floor", 0)

View File

@@ -201,7 +201,9 @@ class RegexService:
placement: int,
character_name: Optional[str] = None,
preset_name: Optional[str] = None,
message_depth: int = 0
message_depth: int = 0,
is_for_llm: bool = False, # ✅ 新增是否发送给LLM
is_markdown_rendered: bool = False # ✅ 新增是否已Markdown渲染
) -> str:
"""
根据 placement 应用正则规则
@@ -212,6 +214,8 @@ class RegexService:
character_name: 当前角色卡名称
preset_name: 当前预设名称
message_depth: 消息深度
is_for_llm: 是否用于发送给 LLM影响 promptOnly 逻辑)
is_markdown_rendered: 是否是 Markdown 渲染后的内容(影响 markdownOnly 逻辑)
Returns:
处理后的文本
@@ -220,6 +224,14 @@ class RegexService:
result = text
for rule in rules:
# ✅ 检查 promptOnly如果规则只应用于LLM但当前不是LLM场景则跳过
if rule.promptOnly and not is_for_llm:
continue
# ✅ 检查 markdownOnly如果规则只应用于Markdown渲染但当前不是渲染后则跳过
if rule.markdownOnly and not is_markdown_rendered:
continue
# 检查此规则是否适用于当前 placement
if placement not in [p.value for p in rule.placement]:
continue