cfc/cfc-langgraph/app/memory/store.py

import os
import logging
from datetime import datetime

logger = logging.getLogger(__name__)


class MemoryManager:
    """轻量级记忆管理器 - 仅保存对话历史到向量数据库
    
    不再使用 langchain.memory（已废弃），直接实现核心功能：
    - Layer 3 - 语义记忆：每次对话后向量化存储，供后续会话召回
    """

    def __init__(self, embeddings):
        """初始化记忆管理器
        
        Args:
            embeddings: 嵌入模型实例 (OpenAIEmbeddings 或其他兼容接口)
        """
        self.embeddings = embeddings
        # Layer 3 向量库：历史对话记忆
        from langchain_chroma import Chroma
        self.memory_vectorstore = Chroma(
            collection_name="user_memory",
            embedding_function=self.embeddings,
            persist_directory=os.path.expanduser("~/.cfc/langgraph/memory"),
        )

    async def save_conversation(self, user_id: int, conversation_id: str,
                                messages: list[dict]):
        """会话结束后保存到向量记忆库"""
        texts = []
        for msg in messages:
            role = msg.get("role", "unknown")
            content = msg.get("content", "")
            texts.append(f"[{role}] {content}")

        full_text = "\n".join(texts)
        metadata = {
            "user_id": user_id,
            "conversation_id": conversation_id,
            "timestamp": str(datetime.now()),
        }

        await self.memory_vectorstore.aadd_texts(
            texts=[full_text],
            metadatas=[metadata],
        )
        self.memory_vectorstore.persist()
        logger.info("已保存对话到向量记忆：conv=%s, user=%s", conversation_id, user_id)

    async def recall(self, user_id: int, query: str, k: int = 3) -> list[str]:
        """语义召回：查询与 query 最相似的历史对话片段"""
        results = self.memory_vectorstore.similarity_search(
            query,
            k=k,
            filter={"user_id": user_id},
        )
        return [doc.page_content for doc in results]