chenmoranzhi/SillyTavern_replica
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Add agent workflow engine foundation and theme-style page switcher.

Browse Source 
Introduce WorkflowEngine with state machine, tool registry, and builtin chat template; migrate stream chat to engine callbacks. Move page mode switching to TopBar actions cluster as a ThemeToggle-style dropdown (聊天/工作室/爽文/房间).

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
moranzhi
2026-05-31 02:30:43 +08:00
parent d6745b45a5
commit bc130d98f4
24 changed files with 1164 additions and 614 deletions
Download Patch File Download Diff File Expand all files Collapse all files

504
backend/services/chat_workflow_service.py
View File

@@ -17,6 +17,7 @@ try:
from backend.utils.llm_client import LLMClient
from backend.services.task_queue_manager import task_queue_manager, TaskType
from backend.core.config import settings
from backend.services.workflow_engine import workflow_engine
except ImportError:
from services.character_service import CharacterService
from services.worldbook_service import WorldBookService
@@ -24,6 +25,7 @@ except ImportError:
from utils.llm_client import LLMClient
from services.task_queue_manager import task_queue_manager, TaskType
from core.config import settings
from services.workflow_engine import workflow_engine
class ChatWorkflowService:
@@ -128,201 +130,17 @@ class ChatWorkflowService:
self,
request_data: Dict[str, Any]
) -> Dict[str, Any]:
"""
处理聊天请求的核心工作流
Args:
request_data: 前端发送的完整数据
Returns:
{
"success": bool,
"content": str, # 生成的回复内容
"error": str | None
}
"""
try:
# === 第1步：解析请求数据 ===
current_role = request_data.get("currentRole")
current_chat = request_data.get("currentChat")
user_message = request_data.get("mes", "")
if not current_role or not user_message:
return {
"success": False,
"content": "",
"error": "缺少必要的参数：currentRole 或 mes"
}
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:
# 如果没有提供角色卡数据，从后端加载
character = self.character_service.get_character_by_name(current_role)
if not character:
return {
"success": False,
"content": "",
"error": f"角色 '{current_role}' 不存在"
}
else:
# 使用前端提供的角色卡数据（可能包含用户修改）
try:
from backend.models.internal import CharacterCard
except ImportError:
from models.internal import CharacterCard
character = CharacterCard(**character_data)
print(f"[ChatWorkflow] 已加载角色卡: {character.name}")
# === 第3步：收集并激活世界书条目 ===
active_entries = await self._collect_and_activate_worldbooks(
request_data,
character
)
print(f"[ChatWorkflow] 激活了 {len(active_entries)} 个世界书条目")
# === 第4步：加载聊天历史 ===
chat_history = await self._load_chat_history(current_role, current_chat)
print(f"[ChatWorkflow] 加载了 {len(chat_history)} 条历史消息")
# === 第5步：组装提示词 ===
prompt_messages = self._assemble_prompt(
character,
chat_history,
user_message,
active_entries,
request_data
)
print(f"[ChatWorkflow] 组装了 {len(prompt_messages)} 条提示消息")
# === 第6步：调用LLM生成回复 ===
api_config = request_data.get("apiConfig", {})
preset_config = request_data.get("presetConfig", {})
stream_output = request_data.get("stream", False)
result = await self._generate_response(
prompt_messages,
api_config,
preset_config,
stream_output
)
generated_content = result["content"]
token_usage = result.get("usage", {})
duration = result.get("duration")
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)
try:
try:
from backend.services.token_usage_service import token_usage_service
from backend.models.internal import TokenUsageStatus
except ImportError:
from services.token_usage_service import token_usage_service
from models.internal import TokenUsageStatus
await token_usage_service.record_usage(
chat_id=chat_id,
role_name=current_role,
chat_name=request_data.get('currentChat', ''),
prompt_tokens=token_usage.get("prompt_tokens", 0),
completion_tokens=token_usage.get("completion_tokens", 0),
total_tokens=token_usage.get("total_tokens", 0),
status=TokenUsageStatus.COMPLETED,
floor=floor + 1, # AI 回复的楼层
duration=duration,
model=api_config.get("model"),
api_provider="openai", # TODO: 从 API URL 检测提供商
api_url=api_config.get("api_url") # ✅ 记录 API URL
)
except Exception as e:
print(f"[ChatWorkflow] 记录 Token 使用失败: {e}")
# === 第8步：启动异步并行任务 ===
# 创建任务ID
image_task_id = None
table_task_id = None
options = request_data.get("options", {})
if options.get("imageWorkflow", False):
import uuid
image_task_id = f"img_{uuid.uuid4().hex[:8]}"
await task_queue_manager.add_task(image_task_id, TaskType.IMAGE_WORKFLOW, chat_id)
if options.get("dynamicTable", False):
import uuid
table_task_id = f"tbl_{uuid.uuid4().hex[:8]}"
await task_queue_manager.add_task(table_task_id, TaskType.DYNAMIC_TABLE, chat_id)
await self._start_parallel_tasks(request_data, generated_content, image_task_id, table_task_id)
return {
"success": True,
"content": generated_content,
"error": None,
"activeEntries": active_entries,
"taskIds": {
"imageWorkflow": image_task_id,
"dynamicTable": table_task_id
}
}
except Exception as e:
print(f"[ChatWorkflow] 错误: {str(e)}")
import traceback
traceback.print_exc()
return {
"success": False,
"content": "",
"error": f"工作流执行失败: {str(e)}"
}
"""处理聊天请求 – delegates to WorkflowEngine.run_turn."""
result = await workflow_engine.run_turn(request_data, stream=False)
return {
"success": result.success,
"content": result.content,
"error": result.error,
"activeEntries": result.active_entries,
"taskIds": result.task_ids,
"engineRunId": result.run_id,
"workflowTemplateId": result.workflow_template_id,
}
def _normalize_worldbook_entry(self, entry_data: Dict[str, Any]) -> Dict[str, Any]:
"""
@@ -1211,280 +1029,24 @@ class ChatWorkflowService:
async def process_chat_request_stream(
self,
request_data: Dict[str, Any],
on_chunk
on_chunk,
on_worldbook_active=None,
on_tasks_created=None,
) -> Dict[str, Any]:
"""
处理流式聊天请求
Args:
request_data: 前端发送的完整数据
on_chunk: 回调函数，每次收到 chunk 时调用
Returns:
{
"success": bool,
"content": str, # 完整的生成内容
"error": str | None,
"activeEntries": List,
"taskIds": Dict
}
"""
try:
# === 第1步：解析请求数据 ===
current_role = request_data.get("currentRole")
current_chat = request_data.get("currentChat")
user_message = request_data.get("mes", "")
if not current_role or not user_message:
return {
"success": False,
"content": "",
"error": "缺少必要的参数：currentRole 或 mes"
}
print(f"\n{'#'*80}")
print(f"[ChatWorkflow-Stream] 🚀 开始处理请求")
print(f" - Role: {current_role}")
print(f" - Chat: {current_chat}")
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:
character = self.character_service.get_character_by_name(current_role)
if not character:
return {
"success": False,
"content": "",
"error": f"角色 '{current_role}' 不存在"
}
else:
try:
from backend.models.internal import CharacterCard
except ImportError:
from models.internal import CharacterCard
character = CharacterCard(**character_data)
print(f"[ChatWorkflow-Stream] ✅ 已加载角色卡: {character.name}")
# === 第3步：收集并激活世界书条目 ===
active_entries = await self._collect_and_activate_worldbooks(
request_data,
character
)
print(f"\n[ChatWorkflow-Stream] 📚 世界书激活结果: {len(active_entries)} 个条目")
for i, entry in enumerate(active_entries, 1):
print(f" {i}. {getattr(entry, 'name', 'N/A')} (UID: {getattr(entry, 'uid', 'N/A')})")
# === 第4步：加载聊天历史 ===
chat_history = await self._load_chat_history(current_role, current_chat)
print(f"[ChatWorkflow-Stream] 💬 聊天历史加载结果: {len(chat_history)} 条消息")
# === 第5步：组装提示词 ===
prompt_messages = self._assemble_prompt(
character,
chat_history,
user_message,
active_entries,
request_data
)
print(f"[ChatWorkflow-Stream] 📝 提示词组装结果: {len(prompt_messages)} 条消息")
for i, msg in enumerate(prompt_messages, 1):
role = getattr(msg, 'role', 'unknown')
content_preview = str(getattr(msg, 'content', ''))[:50]
print(f" {i}. [{role}] {content_preview}...")
# === 第6步：流式调用LLM生成回复 ===
api_config = request_data.get("apiConfig", {})
preset_config = request_data.get("presetConfig", {})
print(f"\n[ChatWorkflow-Stream] 🤖 开始流式调用 LLM")
print(f" - Model: {api_config.get('model', 'N/A')}")
print(f" - API URL: {api_config.get('api_url', 'N/A')[:50]}...")
print(f" - API Key: {'已设置' if api_config.get('api_key') else '⚠️ 未设置'}")
print(f" - Temperature: {preset_config.get('parameters', {}).get('temperature', 1.0)}")
print(f" - Max Tokens: {preset_config.get('parameters', {}).get('max_tokens', 30000)}")
print(f" - Request Timeout: {preset_config.get('parameters', {}).get('request_timeout', 60)}s")
print(f"{'~'*80}")
# ✅ 验证 API Key
if not api_config.get('api_key'):
print(f"[ChatWorkflow-Stream] ❌ 错误: API Key 为空！")
print(f" - 请检查配置文件中是否保存了 API Key")
print(f" - Profile ID: {request_data.get('currentProfile', {}).get('id', 'N/A')}")
raise Exception("API Key 未配置，请先在 API 配置页面保存密钥")
generated_content = ""
token_usage = {}
duration = 0
chunk_count = 0
start_time = time.time()
print(f"[ChatWorkflow-Stream] 📡 正在调用 llm_client.stream_chat()...")
try:
async for chunk_dict in self.llm_client.stream_chat(
messages=prompt_messages,
api_url=api_config.get("api_url", ""),
api_key=api_config.get("api_key", ""),
model=api_config.get("model", ""),
temperature=preset_config.get("parameters", {}).get("temperature", 1.0),
max_tokens=preset_config.get("parameters", {}).get("max_tokens", 30000),
request_timeout=preset_config.get("parameters", {}).get("request_timeout", 60) # ✅ 传递超时时间
):
# ✅ 处理 LLM 返回的字典格式数据
if isinstance(chunk_dict, dict):
if chunk_dict.get("type") == "chunk":
chunk_content = chunk_dict.get("content", "")
elif chunk_dict.get("type") == "usage":
# 跳过 usage 信息，不拼接到内容中
continue
else:
# 兼容旧格式或未知类型，尝试直接获取 content
chunk_content = chunk_dict.get("content", str(chunk_dict))
else:
# 如果直接返回字符串（兼容情况）
chunk_content = str(chunk_dict)
# ✅ 拼接纯文本内容
generated_content += chunk_content
chunk_count += 1
# 第一个 chunk 到达时记录
if chunk_count == 1:
first_chunk_time = time.time()
print(f"[ChatWorkflow-Stream] ✨ 收到第一个 chunk (耗时: {first_chunk_time - start_time:.2f}s)")
# 每20个chunk记录一次进度
if chunk_count % 20 == 0:
elapsed = time.time() - start_time
print(f"[ChatWorkflow-Stream] 📊 已接收 {chunk_count} 个 chunks, 当前长度: {len(generated_content)}, 耗时: {elapsed:.2f}s")
# ✅ 调用回调函数发送纯文本 chunk
await on_chunk(chunk_content)
except Exception as stream_error:
elapsed = time.time() - start_time
print(f"[ChatWorkflow-Stream] ❌ 流式调用失败 (耗时: {elapsed:.2f}s)")
print(f" - 错误类型: {type(stream_error).__name__}")
print(f" - 错误信息: {str(stream_error)}")
raise
elapsed = time.time() - start_time
print(f"{'~'*80}")
print(f"[ChatWorkflow-Stream] ✅ LLM 流式调用完成")
print(f" - 总 Chunks: {chunk_count}")
print(f" - 内容长度: {len(generated_content)}")
print(f" - 总耗时: {elapsed:.2f}s")
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)
try:
try:
from backend.services.token_usage_service import token_usage_service
from backend.models.internal import TokenUsageStatus
except ImportError:
from services.token_usage_service import token_usage_service
from models.internal import TokenUsageStatus
# 估算 token 数量
prompt_tokens = len(str(prompt_messages)) // 4
completion_tokens = len(generated_content) // 4
await token_usage_service.record_usage(
chat_id=chat_id,
role_name=current_role,
chat_name=request_data.get('currentChat', ''),
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
status=TokenUsageStatus.COMPLETED,
floor=floor + 1,
duration=duration,
model=api_config.get("model"),
api_provider="openai",
api_url=api_config.get("api_url") # ✅ 记录 API URL
)
except Exception as e:
print(f"[ChatWorkflow-Stream] 记录 Token 使用失败: {e}")
# === 第8步：启动异步并行任务 ===
image_task_id = None
table_task_id = None
options = request_data.get("options", {})
if options.get("imageWorkflow", False):
import uuid
image_task_id = f"img_{uuid.uuid4().hex[:8]}"
await task_queue_manager.add_task(image_task_id, TaskType.IMAGE_WORKFLOW, chat_id)
if options.get("dynamicTable", False):
import uuid
table_task_id = f"tbl_{uuid.uuid4().hex[:8]}"
await task_queue_manager.add_task(table_task_id, TaskType.DYNAMIC_TABLE, chat_id)
await self._start_parallel_tasks(request_data, generated_content, image_task_id, table_task_id)
return {
"success": True,
"content": generated_content,
"error": None,
"activeEntries": active_entries,
"taskIds": {
"imageWorkflow": image_task_id,
"dynamicTable": table_task_id
}
}
except Exception as e:
print(f"[ChatWorkflow-Stream] 错误: {str(e)}")
import traceback
traceback.print_exc()
return {
"success": False,
"content": "",
"error": f"工作流执行失败: {str(e)}"
}
"""处理流式聊天请求 – delegates to WorkflowEngine.run_turn with callbacks."""
result = await workflow_engine.run_turn(
request_data,
stream=True,
on_chunk=on_chunk,
on_worldbook_active=on_worldbook_active,
on_tasks_created=on_tasks_created,
)
return {
"success": result.success,
"content": result.content,
"error": result.error,
"activeEntries": result.active_entries,
"taskIds": result.task_ids,
"engineRunId": result.run_id,
"workflowTemplateId": result.workflow_template_id,
}

105
backend/services/state_machine_runner.py Normal file
View File

@@ -0,0 +1,105 @@
"""
JSON state machine runner for workflow templates.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional
try:
from backend.models.agent import RunEvent, RunEventType, TurnContext, WorkflowRun, RunStatus
from backend.services.tool_registry import ToolRegistry
except ImportError:
from models.agent import RunEvent, RunEventType, TurnContext, WorkflowRun, RunStatus
from services.tool_registry import ToolRegistry
class StateMachineRunner:
def __init__(
self,
definition: Dict[str, Any],
registry: ToolRegistry,
*,
on_event: Optional[Callable[[RunEvent], None]] = None,
) -> None:
self.definition = definition
self.registry = registry
self.on_event = on_event
self.states: Dict[str, Dict[str, Any]] = definition.get("states", {})
@classmethod
def from_file(cls, path: Path, registry: ToolRegistry, **kwargs) -> "StateMachineRunner":
with open(path, "r", encoding="utf-8") as f:
definition = json.load(f)
return cls(definition, registry, **kwargs)
def _emit(self, run: WorkflowRun, event_type: RunEventType, **payload: Any) -> RunEvent:
event = RunEvent(
run_id=run.id,
type=event_type,
state=run.current_state,
tool=payload.pop("tool", None),
payload=payload,
)
if self.on_event:
self.on_event(event)
return event
async def run(self, run: WorkflowRun, ctx: TurnContext) -> List[RunEvent]:
events: List[RunEvent] = []
original_on_event = self.on_event
def collect(event: RunEvent) -> None:
events.append(event)
if original_on_event:
original_on_event(event)
self.on_event = collect
initial = self.definition.get("initial")
if not initial:
raise ValueError("State machine missing 'initial' state")
current = initial
run.status = RunStatus.RUNNING
try:
while current:
state_def = self.states.get(current)
if not state_def:
raise ValueError(f"Unknown state: {current}")
run.current_state = current
events.append(self._emit(run, RunEventType.STATE_ENTER, state=current))
tool_name = state_def.get("tool")
if tool_name:
events.append(self._emit(run, RunEventType.TOOL_START, tool=tool_name))
await self.registry.execute(tool_name, ctx)
events.append(
self._emit(
run,
RunEventType.TOOL_END,
tool=tool_name,
success=True,
)
)
current = state_def.get("next")
if current == "end" or current is None:
break
run.status = RunStatus.COMPLETED
run.result_content = ctx.generated_content
events.append(self._emit(run, RunEventType.COMPLETE))
except Exception as exc:
run.status = RunStatus.FAILED
run.error = str(exc)
ctx.error = str(exc)
events.append(self._emit(run, RunEventType.ERROR, message=str(exc)))
raise
finally:
self.on_event = original_on_event
return events

49
backend/services/tool_registry.py Normal file
View File

@@ -0,0 +1,49 @@
"""
Tool registry for workflow engine steps.
"""
from __future__ import annotations
from typing import Any, Awaitable, Callable, Dict, Optional
try:
from backend.models.agent import TurnContext, ToolSpec
except ImportError:
from models.agent import TurnContext, ToolSpec
ToolHandler = Callable[[TurnContext], Awaitable[None]]
class ToolRegistry:
def __init__(self) -> None:
self._tools: Dict[str, ToolHandler] = {}
self._specs: Dict[str, ToolSpec] = {}
def register(
self,
name: str,
handler: ToolHandler,
*,
description: str = "",
parameters: Optional[Dict[str, Any]] = None,
) -> None:
self._tools[name] = handler
self._specs[name] = ToolSpec(
name=name,
description=description,
parameters=parameters or {},
)
def get(self, name: str) -> ToolHandler:
if name not in self._tools:
raise KeyError(f"Unknown tool: {name}")
return self._tools[name]
def list_specs(self) -> list[ToolSpec]:
return list(self._specs.values())
async def execute(self, name: str, ctx: TurnContext) -> None:
handler = self.get(name)
await handler(ctx)
default_tool_registry = ToolRegistry()

1
backend/services/tools/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Workflow chat tools package."""

261
backend/services/tools/chat_tools.py Normal file
View File

@@ -0,0 +1,261 @@
"""
Chat workflow tools extracted from ChatWorkflowService.
"""
from __future__ import annotations
import asyncio
import time
import uuid
from typing import Any, Dict, List
try:
from backend.models.agent import TurnContext
from backend.models.internal import CharacterCard, TokenUsageStatus
from backend.models.regex_rules import RegexPlacement
from backend.services.character_service import CharacterService
from backend.services.regex_service import regex_service
from backend.services.task_queue_manager import TaskType, task_queue_manager
from backend.services.token_usage_service import token_usage_service
from backend.core.config import settings
except ImportError:
from models.agent import TurnContext
from models.internal import CharacterCard, TokenUsageStatus
from models.regex_rules import RegexPlacement
from services.character_service import CharacterService
from services.regex_service import regex_service
from services.task_queue_manager import TaskType, task_queue_manager
from services.token_usage_service import token_usage_service
from core.config import settings
_character_service = CharacterService()
_workflow_service = None
def _get_workflow_service():
"""Lazy init to avoid circular import with chat_workflow_service."""
global _workflow_service
if _workflow_service is None:
try:
from backend.services.chat_workflow_service import ChatWorkflowService
except ImportError:
from services.chat_workflow_service import ChatWorkflowService
_workflow_service = ChatWorkflowService()
return _workflow_service
async def regex_apply_user_input(ctx: TurnContext) -> None:
processed = regex_service.apply_rules_by_placement(
text=ctx.user_message,
placement=RegexPlacement.USER_INPUT.value,
character_name=ctx.current_role,
preset_name=ctx.preset_name,
message_depth=0,
is_for_llm=True,
is_markdown_rendered=False,
)
if processed != ctx.user_message:
print("[WorkflowTool] Applied user-input regex rules")
ctx.user_message = processed
async def load_character(ctx: TurnContext) -> None:
character_data = ctx.request_data.get("characterData")
if not character_data:
character = _character_service.get_character_by_name(ctx.current_role)
if not character:
raise ValueError(f"角色 '{ctx.current_role}' 不存在")
else:
character = CharacterCard(**character_data)
ctx.character = character
print(f"[WorkflowTool] Loaded character: {character.name}")
async def activate_worldbook(ctx: TurnContext) -> None:
svc = _get_workflow_service()
active_entries = await svc._collect_and_activate_worldbooks(
ctx.request_data,
ctx.character,
)
ctx.active_entries = active_entries
print(f"[WorkflowTool] Activated {len(active_entries)} worldbook entries")
if ctx.callbacks and ctx.callbacks.on_worldbook_active:
entries_payload = [
entry.model_dump() if hasattr(entry, "model_dump") else entry
for entry in active_entries
]
await ctx.callbacks.on_worldbook_active(entries_payload)
async def load_chat_history(ctx: TurnContext) -> None:
svc = _get_workflow_service()
chat_history = await svc._load_chat_history(
ctx.current_role,
ctx.current_chat,
)
ctx.chat_history = chat_history
print(f"[WorkflowTool] Loaded {len(chat_history)} history messages")
async def build_prompt_messages(ctx: TurnContext) -> None:
svc = _get_workflow_service()
prompt_messages = svc._assemble_prompt(
ctx.character,
ctx.chat_history,
ctx.user_message,
ctx.active_entries,
ctx.request_data,
)
ctx.prompt_messages = prompt_messages
print(f"[WorkflowTool] Built {len(prompt_messages)} prompt messages")
async def llm_main_reply(ctx: TurnContext) -> None:
svc = _get_workflow_service()
api_config = ctx.request_data.get("apiConfig", {})
preset_config = ctx.request_data.get("presetConfig", {})
if ctx.stream:
if not api_config.get("api_key"):
raise ValueError("API Key 未配置，请先在 API 配置页面保存密钥")
generated_content = ""
chunk_count = 0
start_time = time.time()
async for chunk_dict in svc.llm_client.stream_chat(
messages=ctx.prompt_messages,
api_url=api_config.get("api_url", ""),
api_key=api_config.get("api_key", ""),
model=api_config.get("model", ""),
temperature=preset_config.get("parameters", {}).get("temperature", 1.0),
max_tokens=preset_config.get("parameters", {}).get("max_tokens", 30000),
request_timeout=preset_config.get("parameters", {}).get("request_timeout", 60),
):
if isinstance(chunk_dict, dict):
if chunk_dict.get("type") == "chunk":
chunk_content = chunk_dict.get("content", "")
elif chunk_dict.get("type") == "usage":
continue
else:
chunk_content = chunk_dict.get("content", str(chunk_dict))
else:
chunk_content = str(chunk_dict)
generated_content += chunk_content
chunk_count += 1
if ctx.callbacks and ctx.callbacks.on_chunk:
await ctx.callbacks.on_chunk(chunk_content)
ctx.duration = time.time() - start_time
ctx.generated_content = generated_content
ctx.token_usage = {
"prompt_tokens": len(str(ctx.prompt_messages)) // 4,
"completion_tokens": len(generated_content) // 4,
"total_tokens": (len(str(ctx.prompt_messages)) // 4)
+ (len(generated_content) // 4),
}
print(
f"[WorkflowTool] Stream LLM complete: {chunk_count} chunks, "
f"{len(generated_content)} chars"
)
else:
result = await svc._generate_response(
ctx.prompt_messages,
api_config,
preset_config,
stream=False,
)
ctx.generated_content = result["content"]
ctx.token_usage = result.get("usage", {})
ctx.duration = result.get("duration", 0.0)
print(f"[WorkflowTool] LLM complete: {len(ctx.generated_content)} chars")
async def regex_apply_ai_output(ctx: TurnContext) -> None:
processed = regex_service.apply_rules_by_placement(
text=ctx.generated_content,
placement=RegexPlacement.AI_OUTPUT.value,
character_name=ctx.current_role,
preset_name=ctx.preset_name,
message_depth=0,
is_for_llm=False,
is_markdown_rendered=False,
)
if processed != ctx.generated_content:
print("[WorkflowTool] Applied AI-output regex rules")
ctx.generated_content = processed
async def record_token_usage(ctx: TurnContext) -> None:
chat_id = f"{ctx.current_role}/{ctx.current_chat}"
floor = ctx.request_data.get("floor", 0)
api_config = ctx.request_data.get("apiConfig", {})
try:
await token_usage_service.record_usage(
chat_id=chat_id,
role_name=ctx.current_role,
chat_name=ctx.current_chat,
prompt_tokens=ctx.token_usage.get("prompt_tokens", 0),
completion_tokens=ctx.token_usage.get("completion_tokens", 0),
total_tokens=ctx.token_usage.get("total_tokens", 0),
status=TokenUsageStatus.COMPLETED,
floor=floor + 1,
duration=ctx.duration,
model=api_config.get("model"),
api_provider="openai",
api_url=api_config.get("api_url"),
)
except Exception as exc:
print(f"[WorkflowTool] Token usage recording failed: {exc}")
async def enqueue_parallel_tasks(ctx: TurnContext) -> None:
chat_id = f"{ctx.current_role}/{ctx.current_chat}"
options = ctx.request_data.get("options", {})
image_task_id = None
table_task_id = None
if options.get("imageWorkflow", False):
image_task_id = f"img_{uuid.uuid4().hex[:8]}"
await task_queue_manager.add_task(image_task_id, TaskType.IMAGE_WORKFLOW, chat_id)
if options.get("dynamicTable", False):
table_task_id = f"tbl_{uuid.uuid4().hex[:8]}"
await task_queue_manager.add_task(table_task_id, TaskType.DYNAMIC_TABLE, chat_id)
ctx.task_ids = {
"imageWorkflow": image_task_id,
"dynamicTable": table_task_id,
}
if ctx.callbacks and ctx.callbacks.on_tasks_created:
if image_task_id or table_task_id:
await ctx.callbacks.on_tasks_created(ctx.task_ids)
# Fire-and-forget parallel workers (same as legacy service)
svc = _get_workflow_service()
asyncio.create_task(
svc._start_parallel_tasks(
ctx.request_data,
ctx.generated_content,
image_task_id,
table_task_id,
)
)
def register_chat_tools(registry) -> None:
"""Register all chat workflow tools on the given registry."""
registry.register("regex_apply_user_input", regex_apply_user_input, description="Apply user-input regex")
registry.register("load_character", load_character, description="Load character card")
registry.register("activate_worldbook", activate_worldbook, description="Activate worldbook entries")
registry.register("load_chat_history", load_chat_history, description="Load chat history")
registry.register("build_prompt_messages", build_prompt_messages, description="Assemble LLM prompt")
registry.register("llm_main_reply", llm_main_reply, description="Call main LLM (supports stream)")
registry.register("regex_apply_ai_output", regex_apply_ai_output, description="Apply AI-output regex")
registry.register("record_token_usage", record_token_usage, description="Persist token usage")
registry.register("enqueue_parallel_tasks", enqueue_parallel_tasks, description="Enqueue parallel tasks")

170
backend/services/workflow_engine.py Normal file
View File

@@ -0,0 +1,170 @@
"""
Workflow engine – orchestrates template loading, state machine execution, and run persistence.
"""
from __future__ import annotations
import json
import uuid
from datetime import datetime
from pathlib import Path
from typing import Any, Awaitable, Callable, Dict, List, Optional
try:
from backend.core.config import settings
from backend.models.agent import (
ChatRunBinding,
ChatTurnResult,
RunEvent,
RunStatus,
TurnCallbacks,
TurnContext,
WorkflowRun,
WorkflowTemplate,
WorkflowTemplateKind,
)
from backend.services.state_machine_runner import StateMachineRunner
from backend.services.tool_registry import ToolRegistry, default_tool_registry
from backend.services.tools.chat_tools import register_chat_tools
except ImportError:
from core.config import settings
from models.agent import (
ChatRunBinding,
ChatTurnResult,
RunEvent,
RunStatus,
TurnCallbacks,
TurnContext,
WorkflowRun,
WorkflowTemplate,
WorkflowTemplateKind,
)
from services.state_machine_runner import StateMachineRunner
from services.tool_registry import ToolRegistry, default_tool_registry
from services.tools.chat_tools import register_chat_tools
class WorkflowEngine:
def __init__(self, registry: Optional[ToolRegistry] = None) -> None:
self.registry = registry or default_tool_registry
if not self.registry.list_specs():
register_chat_tools(self.registry)
def _template_dir(self, template_id: str) -> Path:
return settings.AGENT_TEMPLATES_PATH / template_id
def load_template(self, template_id: str = WorkflowTemplateKind.BUILTIN_CHAT.value) -> WorkflowTemplate:
template_path = self._template_dir(template_id) / "template.json"
with open(template_path, "r", encoding="utf-8") as f:
data = json.load(f)
return WorkflowTemplate(**data)
def _run_dir(self, role_name: str, chat_name: str) -> Path:
return settings.AGENT_RUNS_PATH / "chat" / role_name / chat_name
def _persist_run(self, run: WorkflowRun, events: List[RunEvent]) -> None:
run_dir = self._run_dir(run.binding.role_name, run.binding.chat_name)
run_dir.mkdir(parents=True, exist_ok=True)
run_file = run_dir / "run.json"
run.finished_at = datetime.now().isoformat()
with open(run_file, "w", encoding="utf-8") as f:
json.dump(run.model_dump(), f, ensure_ascii=False, indent=2)
events_file = run_dir / "events.jsonl"
with open(events_file, "a", encoding="utf-8") as f:
for event in events:
f.write(json.dumps(event.model_dump(), ensure_ascii=False) + "\n")
async def run_turn(
self,
request_data: Dict[str, Any],
*,
stream: bool = False,
on_chunk: Optional[Callable[[str], Awaitable[None]]] = None,
on_worldbook_active: Optional[Callable[[List[Any]], Awaitable[None]]] = None,
on_tasks_created: Optional[Callable[[Dict[str, Any]], Awaitable[None]]] = None,
template_id: str = WorkflowTemplateKind.BUILTIN_CHAT.value,
) -> ChatTurnResult:
current_role = request_data.get("currentRole", "")
current_chat = request_data.get("currentChat", "")
user_message = request_data.get("mes", "")
if not current_role or not user_message:
return ChatTurnResult(
success=False,
error="缺少必要的参数：currentRole 或 mes",
workflow_template_id=template_id,
)
preset_config = request_data.get("presetConfig", {})
preset_name = preset_config.get("selectedPreset")
run_id = uuid.uuid4().hex
binding = ChatRunBinding(
role_name=current_role,
chat_name=current_chat or "",
template_id=template_id,
)
run = WorkflowRun(
id=run_id,
template_id=template_id,
binding=binding,
status=RunStatus.PENDING,
)
callbacks = TurnCallbacks(
on_chunk=on_chunk,
on_worldbook_active=on_worldbook_active,
on_tasks_created=on_tasks_created,
)
ctx = TurnContext(
request_data=request_data,
template_id=template_id,
run_id=run_id,
stream=stream,
callbacks=callbacks,
current_role=current_role,
current_chat=current_chat or "",
user_message=user_message,
preset_name=preset_name,
)
template = self.load_template(template_id)
sm_path = self._template_dir(template_id) / template.state_machine_path
runner = StateMachineRunner.from_file(sm_path, self.registry)
try:
events = await runner.run(run, ctx)
self._persist_run(run, events)
active_entries = [
entry.model_dump() if hasattr(entry, "model_dump") else entry
for entry in ctx.active_entries
]
return ChatTurnResult(
success=True,
content=ctx.generated_content,
active_entries=active_entries,
task_ids=ctx.task_ids,
run_id=run_id,
workflow_template_id=template_id,
)
except Exception as exc:
run.status = RunStatus.FAILED
run.error = str(exc)
try:
self._persist_run(run, [])
except Exception:
pass
return ChatTurnResult(
success=False,
error=f"工作流执行失败: {exc}",
run_id=run_id,
workflow_template_id=template_id,
)
# Module-level singleton
workflow_engine = WorkflowEngine()