60 lines
1.5 KiB
Python
60 lines
1.5 KiB
Python
# ============================================================
|
||
# 长任务进度 / Thinking 日志（moduleId: core-progress, 可重生 ✅）
|
||
# 同步代码 emit → 网关 SSE 轮询 drain → 前端 Thinking 面板
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import threading
|
||
import time
|
||
from collections import deque
|
||
from typing import Any
|
||
|
||
_lock = threading.Lock()
|
||
_buf: deque[dict[str, Any]] = deque(maxlen=200)
|
||
_active = False
|
||
|
||
|
||
def begin_progress() -> None:
|
||
"""一轮对话开始：清空缓冲并启用（跨线程可见）。"""
|
||
global _active
|
||
with _lock:
|
||
_buf.clear()
|
||
_active = True
|
||
|
||
|
||
def end_progress() -> None:
|
||
global _active
|
||
with _lock:
|
||
_active = False
|
||
|
||
|
||
def emit_thinking(
|
||
step: str,
|
||
detail: str = "",
|
||
*,
|
||
status: str = "running",
|
||
pct: float | None = None,
|
||
) -> None:
|
||
"""推送一条 Thinking 步骤（线程安全，供 SQL 扫描等后台线程调用）。"""
|
||
with _lock:
|
||
if not _active:
|
||
return
|
||
ev: dict[str, Any] = {
|
||
"type": "thinking",
|
||
"step": step,
|
||
"detail": detail or "",
|
||
"status": status, # running | done | warn
|
||
"ts": time.time(),
|
||
}
|
||
if pct is not None:
|
||
ev["pct"] = max(0.0, min(100.0, float(pct)))
|
||
_buf.append(ev)
|
||
|
||
|
||
def drain_thinking() -> list[dict[str, Any]]:
|
||
"""取出并清空当前缓冲。"""
|
||
with _lock:
|
||
items = list(_buf)
|
||
_buf.clear()
|
||
return items
|