Losita/mai-bot
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf：优化麦麦观察体验，优化推理检索体验

Browse Source
This commit is contained in:
SengokuCola
2026-05-07 20:15:14 +08:00
parent 2a7722f84e
commit 827cdbd441
23 changed files with 1206 additions and 376 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/config/config.py
View File

@@ -57,7 +57,7 @@ MODEL_CONFIG_PATH: Path = (CONFIG_DIR / "model_config.toml").resolve().absolute(
LEGACY_ENV_PATH: Path = (PROJECT_ROOT / ".env").resolve().absolute()
A_MEMORIX_LEGACY_CONFIG_PATH: Path = (CONFIG_DIR / "a_memorix.toml").resolve().absolute()
MMC_VERSION: str = "1.0.0-pre.14"
CONFIG_VERSION: str = "8.10.13"
CONFIG_VERSION: str = "8.10.15"
MODEL_CONFIG_VERSION: str = "1.16.0"
logger = get_logger("config")

9
src/config/official_configs.py
View File

@@ -2767,15 +2767,6 @@ class DebugConfig(ConfigBase):
__ui_label__ = "其他"
__ui_icon__ = "more-horizontal"
enable_maisaka_stage_board: bool = Field(
default=False,
json_schema_extra={
"x-widget": "switch",
"x-icon": "layout-dashboard",
},
)
"""是否启用 Maisaka 阶段看板"""
show_maisaka_thinking: bool = Field(
default=True,
json_schema_extra={

4
src/main.py
View File

@@ -16,7 +16,6 @@ from src.config.config import config_manager, global_config
from src.emoji_system.emoji_manager import emoji_manager
from src.learners.expression_auto_check_task import ExpressionAutoCheckTask
from src.manager.async_task_manager import async_task_manager
from src.maisaka.display.stage_status_board import disable_stage_status_board, enable_stage_status_board
from src.plugin_runtime.integration import get_plugin_runtime_manager
from src.prompt.prompt_manager import prompt_manager
from src.services.memory_flow_service import memory_automation_service
@@ -66,8 +65,6 @@ class MainSystem:
async def initialize(self) -> None:
"""初始化系统组件"""
if global_config.debug.enable_maisaka_stage_board:
enable_stage_status_board()
logger.info(t("startup.waking_up", nickname=global_config.bot.nickname))
self.webui_task = asyncio.create_task(self._run_webui_startup_sequence(), name="webui_startup")
@@ -191,7 +188,6 @@ async def main() -> None:
await system.initialize()
await system.schedule_tasks()
finally:
disable_stage_status_board()
emoji_manager.shutdown()
await memory_automation_service.shutdown()
await a_memorix_host_service.stop()

1
src/maisaka/builtin_tool/continue_tool.py
View File

@@ -31,7 +31,6 @@ async def handle_tool(
success=True,
content="当前对话继续进入下一轮思考和工具执行。",
metadata={
"pause_execution": True,
"timing_action": "continue",
},
)

4
src/maisaka/display/__init__.py
View File

@@ -11,8 +11,6 @@ from .display_utils import (
from .prompt_cli_renderer import PromptCLIVisualizer
from .prompt_preview_logger import PromptPreviewLogger
from .stage_status_board import (
disable_stage_status_board,
enable_stage_status_board,
remove_stage_status,
update_stage_status,
)
@@ -21,8 +19,6 @@ __all__ = [
"PromptCLIVisualizer",
"PromptPreviewLogger",
"build_tool_call_summary_lines",
"disable_stage_status_board",
"enable_stage_status_board",
"format_token_count",
"format_tool_call_for_display",
"get_request_panel_style",

132
src/maisaka/display/stage_status_board.py
View File

@@ -1,54 +1,20 @@
"""Maisaka 阶段状态看板。"""
"""Maisaka 阶段状态广播。"""
from __future__ import annotations
from pathlib import Path
from typing import Any, Optional
from typing import Any
import json
import os
import subprocess
import sys
import asyncio
import threading
import time
class MaisakaStageStatusBoard:
"""维护 Maisaka 阶段状态，并在独立终端中展示。"""
"""维护 Maisaka 阶段状态，并推送给 WebUI 麦麦观察。"""
def __init__(self) -> None:
self._lock = threading.Lock()
self._enabled = False
self._entries: dict[str, dict[str, Any]] = {}
self._viewer_process: Optional[subprocess.Popen[Any]] = None
self._state_file = Path("temp") / "maisaka_stage_status.json"
self._state_file.parent.mkdir(parents=True, exist_ok=True)
def enable(self) -> None:
"""启用阶段状态看板。"""
with self._lock:
if self._enabled:
return
self._enabled = True
self._write_state_locked()
self._ensure_viewer_process_locked()
def disable(self) -> None:
"""禁用阶段状态看板。"""
with self._lock:
self._enabled = False
self._entries.clear()
self._write_state_locked()
process = self._viewer_process
self._viewer_process = None
if process is not None and process.poll() is None:
try:
process.terminate()
except Exception:
pass
def update(
self,
@@ -62,16 +28,15 @@ class MaisakaStageStatusBoard:
) -> None:
"""更新一个会话的阶段状态。"""
now = time.time()
with self._lock:
if not self._enabled:
return
now = time.time()
current = self._entries.get(session_id, {})
previous_stage = str(current.get("stage") or "").strip()
stage_started_at = float(current.get("stage_started_at") or now)
if previous_stage != stage:
stage_started_at = now
self._entries[session_id] = {
payload = {
"session_id": session_id,
"session_name": session_name,
"stage": stage,
@@ -80,62 +45,53 @@ class MaisakaStageStatusBoard:
"agent_state": agent_state,
"stage_started_at": stage_started_at,
"updated_at": now,
"timestamp": now,
}
self._write_state_locked()
self._entries[session_id] = payload
self._schedule_stage_status_event(payload)
def remove(self, session_id: str) -> None:
"""移除一个会话的阶段状态。"""
with self._lock:
if not self._enabled:
return
self._entries.pop(session_id, None)
self._write_state_locked()
removed = self._entries.pop(session_id, None)
def _write_state_locked(self) -> None:
payload = {
"enabled": self._enabled,
"host_pid": os.getpid(),
"updated_at": time.time(),
"entries": list(self._entries.values()),
}
tmp_file = self._state_file.with_suffix(".tmp")
tmp_file.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
tmp_file.replace(self._state_file)
self._schedule_stage_removed_event(session_id, removed)
def _ensure_viewer_process_locked(self) -> None:
if not sys.platform.startswith("win"):
def snapshot(self) -> list[dict[str, Any]]:
"""返回当前所有聊天流的阶段状态快照。"""
with self._lock:
return [dict(entry) for entry in self._entries.values()]
@staticmethod
def _schedule_stage_status_event(payload: dict[str, Any]) -> None:
try:
from src.maisaka.monitor_events import emit_stage_status
asyncio.get_running_loop().create_task(emit_stage_status(**payload))
except RuntimeError:
return
if self._viewer_process is not None and self._viewer_process.poll() is None:
@staticmethod
def _schedule_stage_removed_event(session_id: str, removed: dict[str, Any] | None) -> None:
try:
from src.maisaka.monitor_events import emit_stage_removed
asyncio.get_running_loop().create_task(
emit_stage_removed(
session_id=session_id,
session_name=str((removed or {}).get("session_name") or ""),
)
)
except RuntimeError:
return
creationflags = getattr(subprocess, "CREATE_NEW_CONSOLE", 0)
viewer_script = Path(__file__).resolve().with_name("stage_status_viewer.py")
self._viewer_process = subprocess.Popen(
[
sys.executable,
str(viewer_script),
str(self._state_file.resolve()),
],
creationflags=creationflags,
cwd=str(Path.cwd()),
)
_stage_board = MaisakaStageStatusBoard()
def enable_stage_status_board() -> None:
"""启用控制台阶段状态看板。"""
_stage_board.enable()
def disable_stage_status_board() -> None:
"""禁用控制台阶段状态看板。"""
_stage_board.disable()
def update_stage_status(
*,
session_id: str,
@@ -145,7 +101,7 @@ def update_stage_status(
round_text: str = "",
agent_state: str = "",
) -> None:
"""更新控制台阶段状态。"""
"""更新 WebUI 麦麦观察中的阶段状态。"""
_stage_board.update(
session_id=session_id,
@@ -158,6 +114,12 @@ def update_stage_status(
def remove_stage_status(session_id: str) -> None:
"""移除控制台阶段状态。"""
"""移除 WebUI 麦麦观察中的阶段状态。"""
_stage_board.remove(session_id)
def get_stage_status_snapshot() -> list[dict[str, Any]]:
"""获取当前阶段状态快照。"""
return _stage_board.snapshot()

93
src/maisaka/display/stage_status_viewer.py
View File

@@ -1,93 +0,0 @@
"""Maisaka 阶段状态看板查看器。"""
from __future__ import annotations
from pathlib import Path
from typing import Any
import json
import os
import sys
import time
import traceback
def _clear_screen() -> None:
os.system("cls" if sys.platform.startswith("win") else "clear")
def _load_state(state_file: Path) -> dict[str, Any]:
if not state_file.exists():
return {}
try:
return json.loads(state_file.read_text(encoding="utf-8"))
except Exception:
return {}
def _render(state: dict[str, Any]) -> str:
entries = state.get("entries")
if not isinstance(entries, list):
entries = []
lines = ["Maisaka 阶段看板", "=" * 72, ""]
if not entries:
lines.append("当前没有活跃会话。")
return "\n".join(lines)
entries = sorted(
[entry for entry in entries if isinstance(entry, dict)],
key=lambda item: str(item.get("session_name") or item.get("session_id") or ""),
)
now = time.time()
for entry in entries:
session_name = str(entry.get("session_name") or entry.get("session_id") or "").strip() or "unknown"
session_id = str(entry.get("session_id") or "").strip()
stage = str(entry.get("stage") or "").strip() or "未知"
detail = str(entry.get("detail") or "").strip() or "-"
round_text = str(entry.get("round_text") or "").strip()
agent_state = str(entry.get("agent_state") or "").strip() or "-"
stage_started_at = float(entry.get("stage_started_at") or now)
elapsed = max(0.0, now - stage_started_at)
lines.append(f"Chat: {session_name}")
if session_id and session_id != session_name:
lines.append(f"ID: {session_id}")
lines.append(f"阶段: {stage}")
if round_text:
lines.append(f"轮次: {round_text}")
lines.append(f"详情: {detail}")
lines.append(f"状态: {agent_state}")
lines.append(f"阶段耗时: {elapsed:.1f}s")
lines.append("-" * 72)
return "\n".join(lines)
def main() -> int:
if len(sys.argv) < 2:
return 1
state_file = Path(sys.argv[1]).resolve()
log_file = state_file.with_name("maisaka_stage_status_viewer.log")
last_render = ""
while True:
try:
state = _load_state(state_file)
if not state.get("enabled", False):
return 0
rendered = _render(state)
if rendered != last_render:
_clear_screen()
print(rendered, flush=True)
last_render = rendered
time.sleep(0.5)
except Exception:
log_file.write_text(traceback.format_exc(), encoding="utf-8")
time.sleep(3)
return 1
if __name__ == "__main__":
raise SystemExit(main())

78
src/maisaka/monitor_events.py
View File

@@ -367,6 +367,47 @@ async def emit_session_start(
})
async def emit_stage_status(
*,
session_id: str,
session_name: str,
stage: str,
detail: str = "",
round_text: str = "",
agent_state: str = "",
stage_started_at: float,
updated_at: float,
timestamp: float,
) -> None:
"""广播单个聊天流的当前阶段状态。"""
await _broadcast("stage.status", {
"session_id": session_id,
"session_name": session_name,
"stage": stage,
"detail": detail,
"round_text": round_text,
"agent_state": agent_state,
"stage_started_at": stage_started_at,
"updated_at": updated_at,
"timestamp": timestamp,
})
async def emit_stage_removed(
*,
session_id: str,
session_name: str = "",
) -> None:
"""广播聊天流阶段状态移除事件。"""
await _broadcast("stage.removed", {
"session_id": session_id,
"session_name": session_name,
"timestamp": time.time(),
})
async def emit_message_ingested(
session_id: str,
speaker_name: str,
@@ -385,6 +426,26 @@ async def emit_message_ingested(
})
async def emit_message_sent(
session_id: str,
speaker_name: str,
content: str,
message_id: str,
timestamp: float,
source_kind: str = "",
) -> None:
"""广播 MaiSaka 自己发送的消息事件。"""
await _broadcast("message.sent", {
"session_id": session_id,
"speaker_name": speaker_name,
"content": content,
"message_id": message_id,
"source_kind": source_kind,
"timestamp": timestamp,
})
async def emit_cycle_start(
session_id: str,
cycle_id: int,
@@ -404,6 +465,23 @@ async def emit_cycle_start(
})
async def emit_cycle_end(
session_id: str,
cycle_id: int,
time_records: Dict[str, float],
agent_state: str,
) -> None:
"""广播单个推理循环结束事件。"""
await _broadcast("cycle.end", {
"session_id": session_id,
"cycle_id": cycle_id,
"time_records": _normalize_payload_value(time_records),
"agent_state": agent_state,
"timestamp": time.time(),
})
async def emit_timing_gate_result(
session_id: str,
cycle_id: int,

77
src/maisaka/reasoning_engine.py
View File

@@ -38,6 +38,7 @@ from .context_messages import (
from .history_post_processor import process_chat_history_after_cycle
from .history_utils import build_prefixed_message_sequence, build_session_message_visible_text
from .monitor_events import (
emit_cycle_end,
emit_cycle_start,
emit_message_ingested,
emit_planner_finalized,
@@ -418,14 +419,7 @@ class MaisakaReasoningEngine:
try:
while self._runtime._running:
queued_trigger = await self._runtime._internal_turn_queue.get()
message_triggered, timeout_triggered = self._drain_ready_turn_triggers(queued_trigger)
if self._runtime._agent_state == self._runtime._STATE_WAIT and not timeout_triggered:
self._runtime._message_turn_scheduled = False
logger.debug(
f"{self._runtime.log_prefix} 当前仍处于 wait 状态，忽略消息触发并继续等待超时"
)
continue
message_triggered = self._drain_ready_turn_triggers(queued_trigger)
if message_triggered:
await self._runtime._wait_for_message_quiet_period()
@@ -436,34 +430,17 @@ class MaisakaReasoningEngine:
if self._runtime._has_pending_messages()
else []
)
if not timeout_triggered and not cached_messages:
if not cached_messages:
continue
self._runtime._agent_state = self._runtime._STATE_RUNNING
self._runtime._update_stage_status(
"消息整理",
f"待处理消息 {len(cached_messages)} 条" if cached_messages else "准备复用超时锚点",
f"待处理消息 {len(cached_messages)} 条",
)
if cached_messages:
asyncio.create_task(self._runtime._trigger_batch_learning(cached_messages))
if timeout_triggered:
self._runtime._chat_history.append(
self._build_wait_completed_message(has_new_messages=True)
)
await self._ingest_messages(cached_messages)
anchor_message = cached_messages[-1]
else:
anchor_message = self._get_timeout_anchor_message()
if anchor_message is None:
logger.warning(
f"{self._runtime.log_prefix} 等待超时后缺少可复用的锚点消息，跳过本轮继续思考"
)
continue
logger.info(f"{self._runtime.log_prefix} 等待超时后开始新一轮思考")
if self._runtime._pending_wait_tool_call_id:
self._runtime._chat_history.append(
self._build_wait_completed_message(has_new_messages=False)
)
asyncio.create_task(self._runtime._trigger_batch_learning(cached_messages))
await self._ingest_messages(cached_messages)
anchor_message = cached_messages[-1]
try:
timing_gate_required = True
for round_index in range(self._runtime._max_internal_rounds):
@@ -716,6 +693,12 @@ class MaisakaReasoningEngine:
time_records=dict(completed_cycle.time_records),
agent_state=self._runtime._agent_state,
)
await emit_cycle_end(
session_id=self._runtime.session_id,
cycle_id=cycle_detail.cycle_id,
time_records=dict(completed_cycle.time_records),
agent_state=self._runtime._agent_state,
)
finally:
if self._runtime._agent_state == self._runtime._STATE_RUNNING:
self._runtime._agent_state = self._runtime._STATE_STOP
@@ -731,12 +714,11 @@ class MaisakaReasoningEngine:
def _drain_ready_turn_triggers(
self,
queued_trigger: Literal["message", "timeout"],
) -> tuple[bool, bool]:
"""合并当前已就绪的 turn 触发信号。"""
queued_trigger: Literal["message"],
) -> bool:
"""合并当前已就绪的消息触发信号。"""
message_triggered = queued_trigger == "message"
timeout_triggered = queued_trigger == "timeout"
while True:
try:
@@ -747,33 +729,8 @@ class MaisakaReasoningEngine:
if next_trigger == "message":
message_triggered = True
continue
if next_trigger == "timeout":
timeout_triggered = True
continue
return message_triggered, timeout_triggered
def _get_timeout_anchor_message(self) -> Optional[SessionMessage]:
"""在 wait 超时后复用最近一条真实用户消息作为锚点。"""
if self._runtime.message_cache:
return self._runtime.message_cache[-1]
return None
def _build_wait_completed_message(self, *, has_new_messages: bool) -> ToolResultMessage:
"""构造 wait 完成后的工具结果消息。"""
tool_call_id = self._runtime._pending_wait_tool_call_id or "wait_timeout"
self._runtime._pending_wait_tool_call_id = None
content = (
"等待已结束，期间收到了新的用户输入。请结合这些新消息继续下一轮思考。"
if has_new_messages
else "等待已超时，期间没有收到新的用户输入。请基于现有上下文继续下一轮思考。"
)
return ToolResultMessage(
content=content,
timestamp=datetime.now(),
tool_call_id=tool_call_id,
tool_name="wait",
)
return message_triggered
async def _ingest_messages(self, messages: list[SessionMessage]) -> None:
"""处理传入消息列表，将其转换为历史消息并加入聊天历史缓存。"""

83
src/maisaka/runtime.py
View File

@@ -46,7 +46,7 @@ from .display.display_utils import build_tool_call_summary_lines, format_token_c
from .display.prompt_cli_renderer import PromptCLIVisualizer
from .display.stage_status_board import remove_stage_status, update_stage_status
from .history_utils import drop_leading_orphan_tool_results
from .monitor_events import emit_session_start
from .monitor_events import emit_message_sent, emit_session_start
from .reasoning_engine import MaisakaReasoningEngine
from .reply_effect import ReplyEffectTracker
from .reply_effect.image_utils import extract_visual_attachments_from_sequence
@@ -62,7 +62,6 @@ class MaisakaHeartFlowChatting:
"""会话级别的 Maisaka 运行时。"""
_STATE_RUNNING: Literal["running"] = "running"
_STATE_WAIT: Literal["wait"] = "wait"
_STATE_STOP: Literal["stop"] = "stop"
def __init__(self, session_id: str):
@@ -85,7 +84,7 @@ class MaisakaHeartFlowChatting:
# Keep all original messages for batching and later learning.
self.message_cache: list[SessionMessage] = []
self._last_processed_index = 0
self._internal_turn_queue: asyncio.Queue[Literal["message", "timeout"]] = asyncio.Queue()
self._internal_turn_queue: asyncio.Queue[Literal["message"]] = asyncio.Queue()
self._mcp_manager: Optional[MCPManager] = None
self._mcp_host_bridge: Optional[MCPHostLLMBridge] = None
@@ -103,7 +102,6 @@ class MaisakaHeartFlowChatting:
self._talk_frequency_adjust = 1.0
self._reply_latency_measurement_started_at: Optional[float] = None
self._recent_reply_latencies: deque[tuple[float, float]] = deque()
self._wait_timeout_task: Optional[asyncio.Task[None]] = None
self._max_internal_rounds = MAX_INTERNAL_ROUNDS
configured_context_size = (
global_config.chat.max_context_size
@@ -111,8 +109,7 @@ class MaisakaHeartFlowChatting:
else global_config.chat.max_private_context_size
)
self._max_context_size = max(1, int(configured_context_size))
self._agent_state: Literal["running", "wait", "stop"] = self._STATE_STOP
self._pending_wait_tool_call_id: Optional[str] = None
self._agent_state: Literal["running", "stop"] = self._STATE_STOP
self._force_next_timing_continue = False
self._force_next_timing_message_id = ""
self._force_next_timing_reason = ""
@@ -211,7 +208,6 @@ class MaisakaHeartFlowChatting:
self._message_turn_scheduled = False
self._message_debounce_required = False
self._cancel_deferred_message_turn_task()
self._cancel_wait_timeout_task()
while not self._internal_turn_queue.empty():
_ = self._internal_turn_queue.get_nowait()
@@ -270,6 +266,11 @@ class MaisakaHeartFlowChatting:
source_kind=source_kind,
)
self._chat_history.append(history_message)
self._emit_monitor_message_sent(
message=message,
speaker_name=speaker_name,
source_kind=source_kind,
)
return True
except Exception as exc:
logger.warning(
@@ -278,6 +279,29 @@ class MaisakaHeartFlowChatting:
)
return False
def _emit_monitor_message_sent(
self,
*,
message: SessionMessage,
speaker_name: str,
source_kind: str,
) -> None:
"""异步广播 MaiSaka 自己发出的消息，供 WebUI 实时展示。"""
try:
asyncio.create_task(
emit_message_sent(
session_id=self.session_id,
speaker_name=speaker_name,
content=(message.processed_plain_text or "").strip(),
message_id=message.message_id,
timestamp=message.timestamp.timestamp(),
source_kind=source_kind,
)
)
except RuntimeError as exc:
logger.debug(f"{self.log_prefix} 广播已发送消息到监控面板失败: {exc}")
async def register_message(self, message: SessionMessage) -> None:
"""缓存一条新消息并唤醒主循环。"""
if self._running:
@@ -914,9 +938,6 @@ class MaisakaHeartFlowChatting:
def _schedule_message_turn(self) -> None:
"""为当前待处理消息安排一次内部 turn。"""
if self._agent_state == self._STATE_WAIT:
return
if not self._has_pending_messages() or self._message_turn_scheduled:
return
@@ -999,51 +1020,9 @@ class MaisakaHeartFlowChatting:
self._message_debounce_required = False
def _enter_wait_state(self, seconds: Optional[float] = None, tool_call_id: Optional[str] = None) -> None:
"""切换到等待状态。"""
self._agent_state = self._STATE_WAIT
self._pending_wait_tool_call_id = tool_call_id
self._message_turn_scheduled = False
self._cancel_deferred_message_turn_task()
self._cancel_wait_timeout_task()
if seconds is not None:
self._wait_timeout_task = asyncio.create_task(
self._schedule_wait_timeout(seconds=seconds, tool_call_id=tool_call_id)
)
def _enter_stop_state(self) -> None:
"""切换到停止状态。"""
self._agent_state = self._STATE_STOP
self._pending_wait_tool_call_id = None
self._cancel_wait_timeout_task()
def _cancel_wait_timeout_task(self) -> None:
"""取消当前 wait 对应的超时任务。"""
if self._wait_timeout_task is None:
return
self._wait_timeout_task.cancel()
self._wait_timeout_task = None
async def _schedule_wait_timeout(self, seconds: float, tool_call_id: Optional[str]) -> None:
"""在 wait 到期后向内部循环投递 timeout 触发。"""
try:
if seconds > 0:
await asyncio.sleep(seconds)
if not self._running:
return
if self._agent_state != self._STATE_WAIT:
return
if self._pending_wait_tool_call_id != tool_call_id:
return
logger.debug(f"{self.log_prefix} Maisaka 等待已超时")
self._agent_state = self._STATE_RUNNING
await self._internal_turn_queue.put("timeout")
except asyncio.CancelledError:
return
finally:
if self._wait_timeout_task is not None and self._pending_wait_tool_call_id == tool_call_id:
self._wait_timeout_task = None
async def _trigger_batch_learning(self, messages: list[SessionMessage]) -> None:
"""按同一批消息触发表达方式和黑话学习。"""

75
src/webui/routers/reasoning_process.py
View File

@@ -37,6 +37,7 @@ class ReasoningPromptListResponse(BaseModel):
page_size: int
stages: list[str] = Field(default_factory=list)
sessions: list[str] = Field(default_factory=list)
selected_session: str = ""
class ReasoningPromptContentResponse(BaseModel):
@@ -76,15 +77,54 @@ def _relative_posix_path(path: Path) -> str:
return path.relative_to(PROMPT_LOG_ROOT).as_posix()
def _collect_prompt_files() -> tuple[list[ReasoningPromptFile], list[str], list[str]]:
def _is_safe_name(name: str) -> bool:
path = Path(name)
return bool(name) and not path.is_absolute() and ".." not in path.parts and len(path.parts) == 1
def _list_stage_names() -> list[str]:
if not PROMPT_LOG_ROOT.is_dir():
return [], [], []
return []
return sorted(path.name for path in PROMPT_LOG_ROOT.iterdir() if path.is_dir() and _is_safe_name(path.name))
def _resolve_stage_name(stage: str) -> str:
normalized_stage = str(stage or "").strip()
if not normalized_stage or normalized_stage == "all":
return "planner"
if not _is_safe_name(normalized_stage):
raise HTTPException(status_code=400, detail="阶段名称不合法")
return normalized_stage
def _list_session_names(stage: str) -> list[str]:
stage_dir = PROMPT_LOG_ROOT / stage
if not stage_dir.is_dir():
return []
session_dirs = [path for path in stage_dir.iterdir() if path.is_dir() and _is_safe_name(path.name)]
session_dirs.sort(key=lambda path: path.stat().st_mtime, reverse=True)
return [path.name for path in session_dirs]
def _resolve_session_name(session: str, sessions: list[str]) -> str:
normalized_session = str(session or "").strip()
if not normalized_session or normalized_session in {"all", "auto"}:
return sessions[0] if sessions else ""
if not _is_safe_name(normalized_session):
raise HTTPException(status_code=400, detail="会话名称不合法")
return normalized_session if normalized_session in sessions else ""
def _collect_prompt_files(stage: str, session: str) -> list[ReasoningPromptFile]:
session_dir = PROMPT_LOG_ROOT / stage / session
if not session or not session_dir.is_dir():
return []
records: dict[tuple[str, str, str], dict[str, object]] = {}
stages: set[str] = set()
sessions: set[str] = set()
for file_path in PROMPT_LOG_ROOT.rglob("*"):
for file_path in session_dir.iterdir():
if not file_path.is_file() or file_path.suffix.lower() not in ALLOWED_SUFFIXES:
continue
@@ -97,17 +137,15 @@ def _collect_prompt_files() -> tuple[list[ReasoningPromptFile], list[str], list[
if len(parts) < 3:
continue
stage, session_id = parts[0], parts[1]
stage_name, session_id = parts[0], parts[1]
stem = file_path.stem
key = (stage, session_id, stem)
key = (stage_name, session_id, stem)
stat = file_path.stat()
stages.add(stage)
sessions.add(session_id)
record = records.setdefault(
key,
{
"stage": stage,
"stage": stage_name,
"session_id": session_id,
"stem": stem,
"timestamp": int(stem) if stem.isdigit() else None,
@@ -127,26 +165,26 @@ def _collect_prompt_files() -> tuple[list[ReasoningPromptFile], list[str], list[
items = [ReasoningPromptFile(**record) for record in records.values()]
items.sort(key=lambda item: (item.modified_at, item.timestamp or 0), reverse=True)
return items, sorted(stages), sorted(sessions)
return items
@router.get("/files", response_model=ReasoningPromptListResponse)
async def list_reasoning_prompt_files(
stage: str = Query("all"),
session: str = Query("all"),
stage: str = Query("planner"),
session: str = Query("auto"),
search: str = Query(""),
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=10, le=200),
):
"""列出 logs/maisaka_prompt 下的推理过程日志。"""
items, stages, sessions = _collect_prompt_files()
stages = _list_stage_names()
selected_stage = _resolve_stage_name(stage)
sessions = _list_session_names(selected_stage)
selected_session = _resolve_session_name(session, sessions)
items = _collect_prompt_files(selected_stage, selected_session)
normalized_search = search.strip().lower()
if stage != "all":
items = [item for item in items if item.stage == stage]
if session != "all":
items = [item for item in items if item.session_id == session]
if normalized_search:
items = [
item
@@ -167,6 +205,7 @@ async def list_reasoning_prompt_files(
page_size=page_size,
stages=stages,
sessions=sessions,
selected_session=selected_session,
)

286
src/webui/routers/system.py
View File

@@ -5,14 +5,19 @@
"""
from datetime import datetime
from typing import Optional
from pathlib import Path
from typing import Literal, Optional
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from pydantic import BaseModel, Field
from sqlalchemy import func, inspect, text
from sqlmodel import col, select
import os
import time
from src.common.database.database import engine, get_db_session
from src.common.database.database_model import Images, ImageType
from src.common.logger import get_logger
from src.config.config import MMC_VERSION
from src.webui.dashboard_update import (
@@ -27,6 +32,14 @@ router = APIRouter(prefix="/system", tags=["system"], dependencies=[Depends(requ
logger = get_logger("webui_system")
_start_time = time.time()
_PROJECT_ROOT = Path(__file__).resolve().parents[3]
_DATA_DIR = _PROJECT_ROOT / "data"
_IMAGE_DIR = _DATA_DIR / "images"
_EMOJI_DIR = _DATA_DIR / "emoji"
_EMOJI_THUMBNAIL_DIR = _DATA_DIR / "emoji_thumbnails"
_LOG_DIR = _PROJECT_ROOT / "logs"
_DATABASE_FILE = _DATA_DIR / "MaiBot.db"
_DATABASE_AUXILIARY_SUFFIXES = ("-wal", "-shm")
class RestartResponse(BaseModel):
@@ -56,6 +69,211 @@ class DashboardVersionResponse(BaseModel):
pypi_url: str = PYPI_PROJECT_URL
class CacheDirectoryStats(BaseModel):
"""本地缓存目录统计。"""
key: str
label: str
path: str
exists: bool
file_count: int
total_size: int
db_records: int = 0
class DatabaseFileStats(BaseModel):
"""数据库文件统计。"""
path: str
exists: bool
size: int
class DatabaseTableStats(BaseModel):
"""数据库表统计。"""
name: str
rows: int
class DatabaseStorageStats(BaseModel):
"""数据库存储统计。"""
files: list[DatabaseFileStats]
tables: list[DatabaseTableStats]
total_size: int
class LocalCacheStatsResponse(BaseModel):
"""本地缓存统计响应。"""
directories: list[CacheDirectoryStats]
database: DatabaseStorageStats
class LocalCacheCleanupRequest(BaseModel):
"""本地缓存清理请求。"""
target: Literal["images", "emoji", "logs"]
tables: list[Literal["llm_usage", "tool_records", "mai_messages"]] = Field(default_factory=list)
class LocalCacheCleanupResponse(BaseModel):
"""本地缓存清理响应。"""
success: bool
message: str
target: str
removed_files: int = 0
removed_bytes: int = 0
removed_records: int = 0
def _parse_version_parts(version: str | None) -> Optional[list[int]]:
"""将版本号转换为可比较的整数列表。"""
if not version:
return None
parts: list[int] = []
for raw_part in version.split("."):
if not raw_part.isdigit():
return None
parts.append(int(raw_part))
return parts
def _is_newer_version(latest: str | None, current: str | None) -> bool:
"""判断 latest 是否新于 current。"""
latest_parts = _parse_version_parts(latest)
current_parts = _parse_version_parts(current)
if latest_parts is None or current_parts is None:
return False
max_len = max(len(latest_parts), len(current_parts))
latest_parts.extend([0] * (max_len - len(latest_parts)))
current_parts.extend([0] * (max_len - len(current_parts)))
return latest_parts > current_parts
def _iter_files(directory: Path) -> list[Path]:
if not directory.exists() or not directory.is_dir():
return []
return [path for path in directory.rglob("*") if path.is_file()]
def _get_directory_size(directory: Path) -> tuple[int, int]:
files = _iter_files(directory)
total_size = 0
for file_path in files:
try:
total_size += file_path.stat().st_size
except OSError:
logger.warning(f"读取缓存文件大小失败: {file_path}")
return len(files), total_size
def _get_image_record_count(image_type: ImageType) -> int:
with get_db_session() as session:
statement = select(func.count()).select_from(Images).where(col(Images.image_type) == image_type)
return int(session.exec(statement).one())
def _build_directory_stats(key: str, label: str, path: Path, image_type: ImageType | None = None) -> CacheDirectoryStats:
file_count, total_size = _get_directory_size(path)
return CacheDirectoryStats(
key=key,
label=label,
path=str(path),
exists=path.exists(),
file_count=file_count,
total_size=total_size,
db_records=_get_image_record_count(image_type) if image_type is not None else 0,
)
def _get_database_files() -> list[DatabaseFileStats]:
db_paths = [_DATABASE_FILE, *[Path(f"{_DATABASE_FILE}{suffix}") for suffix in _DATABASE_AUXILIARY_SUFFIXES]]
result: list[DatabaseFileStats] = []
for db_path in db_paths:
exists = db_path.exists()
size = 0
if exists:
try:
size = db_path.stat().st_size
except OSError:
logger.warning(f"读取数据库文件大小失败: {db_path}")
result.append(DatabaseFileStats(path=str(db_path), exists=exists, size=size))
return result
def _get_database_table_stats() -> list[DatabaseTableStats]:
inspector = inspect(engine)
table_stats: list[DatabaseTableStats] = []
with engine.connect() as connection:
for table_name in inspector.get_table_names():
quoted_table_name = table_name.replace('"', '""')
rows = connection.execute(text(f'SELECT COUNT(*) FROM "{quoted_table_name}"')).scalar_one()
table_stats.append(DatabaseTableStats(name=table_name, rows=int(rows)))
return sorted(table_stats, key=lambda item: item.name)
def _build_database_stats() -> DatabaseStorageStats:
files = _get_database_files()
return DatabaseStorageStats(
files=files,
tables=_get_database_table_stats(),
total_size=sum(file.size for file in files),
)
def _remove_directory_contents(directory: Path) -> tuple[int, int]:
if not directory.exists() or not directory.is_dir():
return 0, 0
removed_files = 0
removed_bytes = 0
for file_path in _iter_files(directory):
try:
file_size = file_path.stat().st_size
file_path.unlink()
removed_files += 1
removed_bytes += file_size
except OSError as exc:
logger.warning(f"删除缓存文件失败: {file_path}, error={exc}")
for child in sorted(directory.rglob("*"), key=lambda item: len(item.parts), reverse=True):
if child.is_dir():
try:
child.rmdir()
except OSError:
pass
return removed_files, removed_bytes
def _delete_image_records(image_type: ImageType) -> int:
removed_records = 0
with get_db_session() as session:
statement = select(Images).where(col(Images.image_type) == image_type)
for record in session.exec(statement).all():
session.delete(record)
removed_records += 1
return removed_records
def _delete_log_records(table_names: list[str]) -> int:
allowed_tables = {"llm_usage", "tool_records", "mai_messages"}
invalid_tables = set(table_names) - allowed_tables
if invalid_tables:
raise ValueError(f"不支持清理这些表: {', '.join(sorted(invalid_tables))}")
removed_records = 0
with engine.begin() as connection:
for table_name in table_names:
quoted_table_name = table_name.replace('"', '""')
result = connection.execute(text(f'DELETE FROM "{quoted_table_name}"'))
removed_records += int(result.rowcount or 0)
return removed_records
@router.post("/restart", response_model=RestartResponse)
async def restart_maibot():
"""
@@ -120,6 +338,70 @@ async def get_dashboard_version(current_version: Optional[str] = None):
)
@router.get("/local-cache", response_model=LocalCacheStatsResponse)
async def get_local_cache_stats():
"""获取 data 目录下图片、表情包和数据库的本地存储情况。"""
try:
return LocalCacheStatsResponse(
directories=[
_build_directory_stats("images", "图片缓存", _IMAGE_DIR, ImageType.IMAGE),
_build_directory_stats("emoji", "表情包缓存", _EMOJI_DIR, ImageType.EMOJI),
_build_directory_stats("emoji_thumbnails", "表情包缩略图缓存", _EMOJI_THUMBNAIL_DIR),
_build_directory_stats("logs", "日志文件", _LOG_DIR),
],
database=_build_database_stats(),
)
except Exception as e:
logger.exception(f"获取本地缓存统计失败: {e}")
raise HTTPException(status_code=500, detail=f"获取本地缓存统计失败: {str(e)}") from e
@router.post("/local-cache/cleanup", response_model=LocalCacheCleanupResponse)
async def cleanup_local_cache(request: LocalCacheCleanupRequest):
"""清理指定的本地缓存区域。"""
try:
if request.target == "images":
removed_files, removed_bytes = _remove_directory_contents(_IMAGE_DIR)
removed_records = _delete_image_records(ImageType.IMAGE)
return LocalCacheCleanupResponse(
success=True,
message="图片缓存已清理",
target=request.target,
removed_files=removed_files,
removed_bytes=removed_bytes,
removed_records=removed_records,
)
if request.target == "emoji":
emoji_files, emoji_bytes = _remove_directory_contents(_EMOJI_DIR)
thumbnail_files, thumbnail_bytes = _remove_directory_contents(_EMOJI_THUMBNAIL_DIR)
removed_records = _delete_image_records(ImageType.EMOJI)
return LocalCacheCleanupResponse(
success=True,
message="表情包缓存已清理",
target=request.target,
removed_files=emoji_files + thumbnail_files,
removed_bytes=emoji_bytes + thumbnail_bytes,
removed_records=removed_records,
)
if not request.tables:
raise HTTPException(status_code=400, detail="请至少选择一个要清理的日志表")
removed_records = _delete_log_records(list(request.tables))
return LocalCacheCleanupResponse(
success=True,
message="日志记录已清理",
target=request.target,
removed_records=removed_records,
)
except HTTPException:
raise
except Exception as e:
logger.exception(f"清理本地缓存失败: {e}")
raise HTTPException(status_code=500, detail=f"清理本地缓存失败: {str(e)}") from e
# 可选：添加更多系统控制功能

9
src/webui/routers/websocket/unified.py
View File

@@ -159,6 +159,15 @@ async def _handle_maisaka_monitor_subscribe(connection_id: str, request_id: Opti
ok=True,
data={"domain": "maisaka_monitor", "topic": "main"},
)
from src.maisaka.display.stage_status_board import get_stage_status_snapshot
await websocket_manager.send_event(
connection_id,
domain="maisaka_monitor",
event="stage.snapshot",
topic="main",
data={"entries": get_stage_status_snapshot(), "timestamp": time.time()},
)
async def _handle_subscribe(connection_id: str, message: Dict[str, Any]) -> None: