memoryweave 35G→185M VACUUM + bge 按需化 + CBM 周更 (2026-09-05)
- memoryweave LanceDB VACUUM (lancedb 0.38+pylance 11, t.optimize(cleanup_older_than=0)):
memories 16.46G→49M (5.9万版本残留), episodes 17.18G→83M (1.8万版本), 总 35G→185M
数据完整: memories=11246, memory_search/memory_write 全链路验证通过
快照: /home/muc/.memoryweave.snapshot-20260905 (cp -al 硬链接, 稳定后删)
- bge-embed 按需化: disable 常驻 + bge-proxy 拉起逻辑 (_ensure_local)
- bge-proxy.service 去 Wants bge-embed; zhiyid.service Wants 改 bge-proxy
- proxy 新增实发失败降级: health OK 但 embedding 500 → 标故障 + 300s 冷却 + 切 local
(实测 104 embedding 500 → 自动拉起 bge-embed → 5s 返回 200)
- CBM 索引 cron 94b65de5e843: 每日 3:40 → 每周日 3:00 (避免每日 5G 内存峰值)
This commit is contained in:
小唯 A06
parent
55a37e2097
commit
e58549425b
|
|
@ -16,9 +16,10 @@ import json
|
|||
import logging
|
||||
import os
|
||||
import socket
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
from http.server import HTTPServer, BaseHTTPRequestHandler
|
||||
from http.server import HTTPServer, BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from urllib import request as urlrequest
|
||||
from urllib.error import URLError
|
||||
|
||||
|
|
@ -38,7 +39,14 @@ logging.basicConfig(
|
|||
log = logging.getLogger("bge-proxy")
|
||||
|
||||
_state_lock = threading.Lock()
|
||||
_state = {"remote_ok": None, "last_check": 0.0, "using": None}
|
||||
_state = {"remote_ok": None, "last_check": 0.0, "using": None,
|
||||
"local_started_by_proxy": False, "remote_ok_since": None,
|
||||
"remote_retry_after": 0.0} # embedding 实发失败后的冷却截止（秒，300s）
|
||||
|
||||
# 按需拉起本地 fallback 的参数（2026-09-05: bge-embed 不再常驻，104 故障时才拉起）
|
||||
LOCAL_START_TIMEOUT = float(os.environ.get("LOCAL_START_TIMEOUT", "30")) # 拉起后等待就绪上限
|
||||
LOCAL_STOP_GRACE = float(os.environ.get("LOCAL_STOP_GRACE", "60")) # 远端稳定多久后停本地
|
||||
BGE_EMBED_SERVICE = os.environ.get("BGE_EMBED_SERVICE", "bge-embed.service")
|
||||
|
||||
|
||||
def _probe_remote() -> bool:
|
||||
|
|
@ -54,23 +62,97 @@ def _probe_remote() -> bool:
|
|||
return False
|
||||
|
||||
|
||||
def _local_healthy() -> bool:
|
||||
"""探测本地 fallback (8001) /health。"""
|
||||
try:
|
||||
req = urlrequest.Request(LOCAL_EMBED_URL + HEALTH_PATH)
|
||||
with urlrequest.urlopen(req, timeout=1.5) as r:
|
||||
if r.status != 200:
|
||||
return False
|
||||
return "ok" in r.read(200).decode("utf-8", "ignore")
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def _ensure_local() -> bool:
|
||||
"""远端不可用时确保本地可用：未启动则 systemctl --user 拉起 bge-embed。
|
||||
|
||||
返回 True = 本地可用；False = 本地拉起失败/超时（转发将 502）。
|
||||
"""
|
||||
if _local_healthy():
|
||||
return True
|
||||
with _state_lock:
|
||||
if not _state["local_started_by_proxy"]:
|
||||
log.info("远端不可用 → 拉起 %s (按需)", BGE_EMBED_SERVICE)
|
||||
try:
|
||||
subprocess.Popen(
|
||||
["systemctl", "--user", "start", BGE_EMBED_SERVICE],
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
||||
)
|
||||
_state["local_started_by_proxy"] = True
|
||||
except Exception as e:
|
||||
log.error("拉起 %s 失败: %s", BGE_EMBED_SERVICE, e)
|
||||
# 轮询等待模型加载（int8 onnx ~10-20s）
|
||||
deadline = time.time() + LOCAL_START_TIMEOUT
|
||||
while time.time() < deadline:
|
||||
if _local_healthy():
|
||||
return True
|
||||
time.sleep(1)
|
||||
log.error("本地 fallback %s 拉起超时 (%ss)", BGE_EMBED_SERVICE, LOCAL_START_TIMEOUT)
|
||||
return False
|
||||
|
||||
|
||||
def _maybe_stop_local(remote_ok: bool):
|
||||
"""远端已恢复且稳定 → 停掉 proxy 拉起的本地 fallback，回到省电模式。"""
|
||||
now = time.time()
|
||||
with _state_lock:
|
||||
if remote_ok and _state["local_started_by_proxy"]:
|
||||
if _state["remote_ok_since"] is None:
|
||||
_state["remote_ok_since"] = now
|
||||
elif now - _state["remote_ok_since"] >= LOCAL_STOP_GRACE:
|
||||
log.info("远端已稳定 %ss → 停 %s (释放内存)", LOCAL_STOP_GRACE, BGE_EMBED_SERVICE)
|
||||
try:
|
||||
subprocess.Popen(
|
||||
["systemctl", "--user", "stop", BGE_EMBED_SERVICE],
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
||||
)
|
||||
_state["local_started_by_proxy"] = False
|
||||
_state["remote_ok_since"] = None
|
||||
except Exception as e:
|
||||
log.error("停止 %s 失败: %s", BGE_EMBED_SERVICE, e)
|
||||
elif not remote_ok:
|
||||
_state["remote_ok_since"] = None
|
||||
|
||||
|
||||
def choose_target() -> tuple[str, bool]:
|
||||
"""返回 (目标URL, 是否远端)。带 3s 冷却 + 状态缓存。"""
|
||||
"""返回 (目标URL, 是否远端)。带 3s 冷却 + 状态缓存 + 按需拉起本地 fallback。"""
|
||||
global _state
|
||||
now = time.time()
|
||||
with _state_lock:
|
||||
if _state["remote_ok"] is not None and now - _state["last_check"] < 3.0:
|
||||
cache_fresh = _state["remote_ok"] is not None and now - _state["last_check"] < 3.0
|
||||
in_cooldown = _state["remote_ok"] is False and now < _state["remote_retry_after"]
|
||||
if cache_fresh or in_cooldown:
|
||||
remote_ok = _state["remote_ok"]
|
||||
else:
|
||||
remote_ok = _probe_remote()
|
||||
_state["remote_ok"] = remote_ok
|
||||
_state["last_check"] = now
|
||||
target = REMOTE_EMBED_URL if remote_ok else LOCAL_EMBED_URL
|
||||
using = "remote(104-GPU)" if remote_ok else "local(106-CPU)"
|
||||
_maybe_stop_local(remote_ok)
|
||||
if remote_ok:
|
||||
using = "remote(104-GPU)"
|
||||
if using != _state["using"]:
|
||||
log.info("切换 → %s", using)
|
||||
with _state_lock:
|
||||
_state["using"] = using
|
||||
return REMOTE_EMBED_URL, True
|
||||
# 远端不可用 → 确保本地可用
|
||||
local_ok = _ensure_local()
|
||||
using = "local(106-CPU)" if local_ok else "local(106-CPU, 未就绪)"
|
||||
if using != _state["using"]:
|
||||
log.info("切换 → %s", using)
|
||||
with _state_lock:
|
||||
_state["using"] = using
|
||||
return target, remote_ok
|
||||
return LOCAL_EMBED_URL, False
|
||||
|
||||
|
||||
def _forward(url: str, method: str, body: bytes, headers: dict) -> tuple[int, dict, bytes]:
|
||||
|
|
@ -103,6 +185,21 @@ class Handler(BaseHTTPRequestHandler):
|
|||
status, resp_headers, resp_body = _forward(
|
||||
target + self.path, self.command, body, dict(self.headers)
|
||||
)
|
||||
# 远端 embedding 实发失败（500/502/503/504，health 探测不出的故障）
|
||||
# → 标故障 + 30s 冷却 + 立即降级 local 重试一次
|
||||
if status >= 500 and target.startswith(REMOTE_EMBED_URL):
|
||||
with _state_lock:
|
||||
_state["remote_ok"] = False
|
||||
# 300s 冷却: 104 可能出现 health OK 但 embedding 500（实发故障），
|
||||
# 短冷却会导致反复 remote→500→local 抖动并可能反复启停 bge-embed
|
||||
_state["remote_retry_after"] = time.time() + 300
|
||||
_state["using"] = None # 强制下次 log 切换
|
||||
log.info("远端 embedding 实发失败(%s) → 降级 local 重试", status)
|
||||
if _ensure_local():
|
||||
target = LOCAL_EMBED_URL
|
||||
status, resp_headers, resp_body = _forward(
|
||||
target + self.path, self.command, body, dict(self.headers)
|
||||
)
|
||||
self.send_response(status)
|
||||
# 只转发安全响应头；Content-Length 必须用实际长度
|
||||
for k, v in resp_headers.items():
|
||||
|
|
@ -112,7 +209,11 @@ class Handler(BaseHTTPRequestHandler):
|
|||
self.send_header("X-BGE-Proxy", "remote-104" if target.startswith(REMOTE_EMBED_URL) else "local-106")
|
||||
self.end_headers()
|
||||
if self.command != "HEAD":
|
||||
self.wfile.write(resp_body)
|
||||
try:
|
||||
self.wfile.write(resp_body)
|
||||
except (BrokenPipeError, ConnectionResetError):
|
||||
# 客户端提前断开（探测超时/取消）——不影响其他请求（多线程）
|
||||
pass
|
||||
|
||||
do_GET = _handle
|
||||
do_POST = _handle
|
||||
|
|
@ -131,11 +232,13 @@ def _startup_probe():
|
|||
|
||||
|
||||
if __name__ == "__main__":
|
||||
log.info("bge 故障切换代理启动 0.0.0.0:%s", PROXY_PORT)
|
||||
log.info("bge 故障切换代理启动 0.0.0.0:%s (Threading)", PROXY_PORT)
|
||||
log.info(" 远端: %s", REMOTE_EMBED_URL)
|
||||
log.info(" 本地 fallback: %s", LOCAL_EMBED_URL)
|
||||
_startup_probe()
|
||||
server = HTTPServer(("0.0.0.0", PROXY_PORT), Handler)
|
||||
# ThreadingHTTPServer: 单请求卡住不阻塞 /health 探测和其他请求
|
||||
server = ThreadingHTTPServer(("0.0.0.0", PROXY_PORT), Handler)
|
||||
server.daemon_threads = True
|
||||
try:
|
||||
server.serve_forever()
|
||||
except KeyboardInterrupt:
|
||||
|
|
|
|||
Loading…
Reference in New Issue