Files
kefu/wechat_rpa/conversation_store.py
2026-07-16 12:11:05 +08:00

154 lines
5.8 KiB
Python

"""
会话档案存储
============
按「会话指纹」持久化每个客户会话的对话档案(客户消息 + 我方回复 + 上次画面快照),
写入 conversations.json,机器人重启后上下文不丢失。
有了档案之后,机器人的提取策略从「每次翻屏复制大段历史」变为:
- 首次遇到某会话:翻屏提取一次可见历史,建档;
- 之后每次:只提取最新一屏,与档案中的上次画面快照做增量比对,
只把【新增的消息】发给 AI,完整上下文由档案历史直接提供。
"""
import json
import os
import time
import threading
# 每个会话最多留存的消息条数(防止档案无限膨胀;AI 实际使用条数由 AI_CONTEXT_MAX_ROUNDS 决定)
MAX_MESSAGES_PER_SESSION = 200
# 画面快照最多留存的行数(用于增量比对)
MAX_SNAPSHOT_LINES = 80
class ConversationStore:
def __init__(self, path: str):
self.path = path
self._lock = threading.Lock()
self._data = {}
self._mtime = 0
self._load()
# ── 持久化 ────────────────────────────────────────────────────────────────
def _load(self):
try:
if os.path.exists(self.path):
with open(self.path, encoding="utf-8") as f:
self._data = json.load(f)
try:
self._mtime = os.path.getmtime(self.path)
except Exception:
self._mtime = 0
else:
self._data = {}
self._mtime = 0
except Exception:
self._data = {}
self._mtime = 0
def _maybe_reload(self):
"""GUI 删除档案后,监听进程下次读写前自动同步磁盘。"""
try:
mt = os.path.getmtime(self.path) if os.path.exists(self.path) else 0
except Exception:
return
if mt != self._mtime:
self._load()
def save(self):
"""原子写入磁盘(先写临时文件再替换,避免中途崩溃损坏档案)。"""
with self._lock:
tmp = self.path + ".tmp"
try:
with open(tmp, "w", encoding="utf-8") as f:
json.dump(self._data, f, ensure_ascii=False, indent=1)
os.replace(tmp, self.path)
try:
self._mtime = os.path.getmtime(self.path)
except Exception:
pass
except Exception:
pass
# ── 档案访问 ──────────────────────────────────────────────────────────────
def _entry(self, fp_hex: str) -> dict:
self._maybe_reload()
return self._data.setdefault(fp_hex, {
"history": [], # [{"role": "user"/"assistant", "content": str, "ts": float}]
"last_lines": [], # 上次提取时聊天框的画面快照(按行),用于增量比对
"updated": 0,
})
def has_record(self, fp_hex: str) -> bool:
self._maybe_reload()
e = self._data.get(fp_hex)
return bool(e and (e.get("history") or e.get("last_lines")))
def history(self, fp_hex: str) -> list:
"""返回该会话的消息历史(列表为活引用,请勿直接修改,用 append)。"""
return self._entry(fp_hex)["history"]
def append(self, fp_hex: str, role: str, content: str):
e = self._entry(fp_hex)
e["history"].append({"role": role, "content": content, "ts": time.time()})
if len(e["history"]) > MAX_MESSAGES_PER_SESSION:
e["history"] = e["history"][-MAX_MESSAGES_PER_SESSION:]
e["updated"] = time.time()
def last_lines(self, fp_hex: str) -> list:
return self._entry(fp_hex)["last_lines"]
def set_last_lines(self, fp_hex: str, lines: list):
e = self._entry(fp_hex)
e["last_lines"] = list(lines)[-MAX_SNAPSHOT_LINES:]
e["updated"] = time.time()
def list_sessions(self, limit: int = 200) -> list:
"""按最近更新排序,返回会话摘要列表。"""
self._maybe_reload()
items = []
for sid, entry in (self._data or {}).items():
if not isinstance(entry, dict):
continue
hist = entry.get("history") or []
preview = ""
for m in reversed(hist):
c = (m.get("content") or "").strip()
if c:
preview = c.replace("\n", " ")[:80]
break
if not preview:
lines = entry.get("last_lines") or []
preview = " ".join(lines[-3:])[:80] if lines else ""
items.append({
"session_id": sid,
"updated": entry.get("updated") or 0,
"message_count": len(hist),
"preview": preview,
})
items.sort(key=lambda x: x.get("updated") or 0, reverse=True)
return items[: max(1, min(limit, 500))]
def delete(self, fp_hex: str) -> bool:
"""删除指定会话档案。"""
self._maybe_reload()
with self._lock:
if fp_hex not in self._data:
return False
del self._data[fp_hex]
self.save()
return True
def clear_all(self) -> int:
"""清空全部会话档案,返回删除条数。"""
self._maybe_reload()
with self._lock:
n = len(self._data)
self._data = {}
self.save()
return n
def count(self) -> int:
self._maybe_reload()
return len(self._data or {})