File size: 3,835 Bytes
c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 c75184e 8a1fc38 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | """
Chat Memory Service — Firestore-backed persistent chat history.
Provides per-session chat history with Firestore persistence and
in-memory fallback when Firestore is unavailable.
Replaces the MongoDB implementation from Phase 1.
"""
import logging
from datetime import datetime, timezone
from typing import List, Dict
from backend.src.config import settings
from backend.src.services.firebase_service import get_firestore_client
try:
from langchain_core.messages import HumanMessage, AIMessage
except ImportError:
from langchain.schema import HumanMessage, AIMessage
logger = logging.getLogger("math_assistant.memory")
class FirestoreChatMemory:
"""Per-session chat memory with Firestore persistence and in-memory fallback."""
def __init__(self, uid: str = "anonymous", session_id: str = "default"):
self.uid = uid
self.session_id = session_id
self._memory: List[Dict] = []
self.db = get_firestore_client()
def _get_collection_ref(self):
"""Get a reference to the session's messages subcollection."""
if not self.db:
return None
return self.db.collection("users").document(self.uid).collection("sessions").document(self.session_id).collection("messages")
def add_message(self, role: str, content: str):
"""Add a message to chat history."""
msg = {
"role": role,
"content": content,
"timestamp": datetime.now(timezone.utc),
}
col_ref = self._get_collection_ref()
if col_ref is not None:
try:
col_ref.add(msg)
return
except Exception as e:
logger.error(f"Failed to add message to Firestore: {e}")
# Fallback to in-memory
msg["session_id"] = self.session_id
self._memory.append(msg)
def get_history(self, limit: int = 20) -> List[Dict]:
"""Retrieve recent chat history for this session."""
col_ref = self._get_collection_ref()
if col_ref is not None:
try:
from firebase_admin import firestore
docs = (
col_ref.order_by("timestamp", direction=firestore.Query.DESCENDING)
.limit(limit)
.stream()
)
msgs = [doc.to_dict() for doc in docs]
msgs.reverse()
return msgs
except Exception as e:
logger.error(f"Failed to get history from Firestore: {e}")
# Fallback to in-memory
return self._memory[-limit:]
def get_langchain_messages(self, limit: int = 10):
"""Convert chat history to LangChain message objects."""
history = self.get_history(limit)
result = []
for msg in history:
if msg.get("role") == "human":
result.append(HumanMessage(content=msg.get("content", "")))
elif msg.get("role") == "assistant":
result.append(AIMessage(content=msg.get("content", "")))
return result
def clear_history(self):
"""Clear all messages for this session."""
col_ref = self._get_collection_ref()
if col_ref is not None:
try:
docs = col_ref.limit(500).stream()
batch = self.db.batch()
count = 0
for doc in docs:
batch.delete(doc.reference)
count += 1
if count > 0:
batch.commit()
except Exception as e:
logger.error(f"Failed to clear history in Firestore: {e}")
self._memory.clear()
# Maintain backward compatibility name for the MathAIEngine import during transition
MongoDBChatMemory = FirestoreChatMemory
|