fix(memory): use JSON file for conversation memory (Zvec can't do scalar-only query)

- Replace Zvec conversation memory with JSON-backed dict
- Thread-safe with _threading.Lock
- Keyed by speaker_id, capped at 20 turns per speaker
- Load/save on each conversation turn
- Restore history on voiceprint re-identification

asr_agent/voiceprint.py

@@ -219,111 +219,74 @@ def get_store() -> VoiceprintStore:
 
 
 # ═══════════════════════════════════════════════════════════════
-#  Conversation Memory — speaker-aware chat history in Zvec
+#  Conversation Memory — speaker-aware chat history (JSON)
 # ═══════════════════════════════════════════════════════════════
 
-_MEM_COLLECTION = None
-_MEM_DIM = 64  # small projection for topic similarity
+import json as _json
+import threading as _threading
 
+_MEM_LOCK = _threading.Lock()
+_MEMORY: dict[str, list[dict]] = {}
+_MEMORY_LOADED = False
 
-def _get_memory_collection():
-    global _MEM_COLLECTION
-    if _MEM_COLLECTION is not None:
-        return _MEM_COLLECTION
-    try:
-        import zvec
-        mem_path = os.path.join(VP_DB_PATH, "conversations")
-        os.makedirs(mem_path, exist_ok=True)
 
-        schema = zvec.CollectionSchema(
-            name="conversations",
-            fields=[
-                zvec.FieldSchema(name="speaker_name", data_type=zvec.DataType.STRING),
-                zvec.FieldSchema(name="speaker_id", data_type=zvec.DataType.STRING),
-                zvec.FieldSchema(name="user_query", data_type=zvec.DataType.STRING),
-                zvec.FieldSchema(name="reply", data_type=zvec.DataType.STRING),
-                zvec.FieldSchema(name="timestamp", data_type=zvec.DataType.FLOAT64),
-            ],
-            vectors=[
-                zvec.VectorSchema(
-                    name="embedding",
-                    data_type=zvec.DataType.VECTOR_FP32,
-                    dimension=_MEM_DIM,
-                    index_param=zvec.HnswIndexParam(metric_type=zvec.MetricType.COSINE),
-                ),
-            ],
-        )
-        _MEM_COLLECTION = zvec.create_and_open(path=mem_path, schema=schema)
-        _MEM_COLLECTION.optimize()
-        logger.info("Conversation memory DB opened at %s (%d entries)",
-                    mem_path, _MEM_COLLECTION.stats.row_count)
+def _load_memories():
+    global _MEMORY, _MEMORY_LOADED
+    path = os.path.join(VP_DB_PATH, "conversations.json")
+    try:
+        if os.path.exists(path):
+            with open(path) as f:
+                _MEMORY = _json.load(f)
+        logger.info("Loaded conversation memory: %d speakers", len(_MEMORY))
     except Exception:
-        logger.exception("Failed to open conversation memory DB")
-        return None
-    return _MEM_COLLECTION
+        pass
+    _MEMORY_LOADED = True
 
 
-def _text_embedding(text: str) -> list[float]:
-    """Simple text embedding via character n-gram hash projection."""
-    vec = np.zeros(_MEM_DIM, dtype=np.float32)
-    for i in range(len(text) - 2):
-        h = hash(text[i:i + 3]) % _MEM_DIM
-        vec[h] += 1.0
-    norm = np.linalg.norm(vec) + 1e-8
-    return (vec / norm).tolist()
+def _save_memories():
+    path = os.path.join(VP_DB_PATH, "conversations.json")
+    try:
+        os.makedirs(VP_DB_PATH, exist_ok=True)
+        with open(path, "w") as f:
+            _json.dump(_MEMORY, f, ensure_ascii=False, indent=2)
+    except Exception:
+        logger.exception("Failed to save conversation memory")
 
 
 def save_conversation(speaker_id: str, speaker_name: str, user_query: str, reply: str):
     """Save a conversation turn to persistent memory."""
-    col = _get_memory_collection()
-    if col is None:
-        return
-    try:
-        import zvec
-        full_text = user_query + " " + reply
-        doc_id = f"{speaker_id}_{int(time.time() * 1000)}"
-        col.insert(zvec.Doc(
-            id=doc_id,
-            vectors={"embedding": _text_embedding(full_text)},
-            fields={
-                "speaker_name": speaker_name,
-                "speaker_id": speaker_id,
-                "user_query": user_query,
-                "reply": reply,
-                "timestamp": time.time(),
-            },
-        ))
-    except Exception:
-        logger.exception("Failed to save conversation")
+    with _MEM_LOCK:
+        if not _MEMORY_LOADED:
+            _load_memories()
+        if speaker_id not in _MEMORY:
+            _MEMORY[speaker_id] = []
+        entry = {
+            "name": speaker_name,
+            "query": user_query,
+            "reply": reply,
+            "time": time.time(),
+        }
+        _MEMORY[speaker_id].append(entry)
+        _MEMORY[speaker_id] = _MEMORY[speaker_id][-20:]  # keep last 20
+        _save_memories()
+    logger.info("Memory saved: %s (%d turns)", speaker_id, len(_MEMORY[speaker_id]))
 
 
 def load_conversation_context(speaker_id: str | None, speaker_name: str | None, limit: int = 5) -> str:
-    """Retrieve recent conversation history for a speaker. Returns formatted context string."""
+    """Retrieve recent conversation history for a speaker."""
     if not speaker_id:
         return ""
-    col = _get_memory_collection()
-    if col is None:
-        return ""
-    try:
-        import zvec
-        result = col.query(
-            queries=zvec.Query(
-                field_name="embedding",
-                vector=[0.0] * _MEM_DIM,  # dummy vector for scalar-only query
-            ),
-            topk=limit,
-            filter=f'speaker_id == "{speaker_id}"',
-        )
-        if not result:
+    with _MEM_LOCK:
+        if not _MEMORY_LOADED:
+            _load_memories()
+        entries = _MEMORY.get(speaker_id, [])
+        if not entries:
             return ""
         history = []
-        for doc in reversed(result):
-            q = doc.fields.get("user_query", "")
-            r = doc.fields.get("reply", "")
+        for e in entries[-limit:]:
+            q, r = e.get("query", ""), e.get("reply", "")
             if q and r:
                 history.append(f"用户: {q}\n助手: {r}")
         if history:
             return "以下是之前和该用户的对话记录，请参考上下文回应:\n" + "\n".join(history)
-    except Exception:
-        logger.exception("Failed to load conversation context")
     return ""

asr_agent/worker.py

@@ -648,6 +648,13 @@ class Worker:
                 if speaker_name:
                     logger.info("VP: identified %s (sim=%.3f)", speaker_name, _sim)
                     await self._send({"type": "speaker", "name": speaker_name, "confidence": round(_sim, 3)})
+                    # Restore conversation history from persistent memory
+                    try:
+                        hist_data = load_conversation_context(speaker_id, speaker_name, limit=10)
+                        if hist_data:
+                            logger.info("Loaded %d history turns for %s", hist_data.count("用户:"), speaker_name)
+                    except Exception:
+                        pass
                 else:
                     auto_id = f"user_{int(time.time() * 1000) % 1000000:06d}"
                     auto_name = f"用户{auto_id[-3:]}"