Chatbot Memory Architecture: 장기 기억과 사용자 프로필 기반 개인화 대화 시스템 구축

from langchain.memory import (
    ConversationBufferMemory,
    ConversationBufferWindowMemory,
    ConversationSummaryMemory,
    ConversationSummaryBufferMemory,
    VectorStoreRetrieverMemory,
)
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import ConversationChain

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 1. Buffer Memory - 전체 대화 이력 저장
buffer_memory = ConversationBufferMemory(
    return_messages=True,
    memory_key="history",
)

# 2. Window Buffer Memory - 최근 5턴만 유지
window_memory = ConversationBufferWindowMemory(
    k=5,
    return_messages=True,
    memory_key="history",
)

# 3. Summary Memory - 대화 요약 유지
summary_memory = ConversationSummaryMemory(
    llm=llm,
    return_messages=True,
    memory_key="history",
)

# 4. Summary Buffer Memory - 하이브리드 (최근 대화 원본 + 오래된 대화 요약)
summary_buffer_memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=1000,  # 이 토큰 수를 초과하면 요약 시작
    return_messages=True,
    memory_key="history",
)

# 5. Vector Store Memory - 의미 기반 검색
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(
    collection_name="conversation_memory",
    embedding_function=embeddings,
    persist_directory="./memory_db",
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

vector_memory = VectorStoreRetrieverMemory(
    retriever=retriever,
    memory_key="history",
    input_key="input",
)

# 실제 대화 체인 구성
conversation = ConversationChain(
    llm=llm,
    memory=summary_buffer_memory,  # 선택한 메모리 타입 적용
    verbose=True,
)

# 대화 실행
response1 = conversation.predict(input="안녕하세요, 저는 김영주입니다. Python 개발자예요.")
response2 = conversation.predict(input="최근에 LangChain으로 RAG 시스템을 만들고 있어요.")
response3 = conversation.predict(input="제 이름이 뭐라고 했죠?")
print(response3)  # "김영주"라고 올바르게 응답

import json
import hashlib
from datetime import datetime, timedelta
from typing import Any
from dataclasses import dataclass, field
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma


@dataclass
class MemoryEntry:
    """단일 메모리 항목을 나타내는 데이터 클래스"""
    content: str
    timestamp: datetime
    importance: float  # 0.0 ~ 1.0
    access_count: int = 0
    last_accessed: datetime = field(default_factory=datetime.now)
    memory_type: str = "episodic"  # episodic, semantic, procedural
    metadata: dict = field(default_factory=dict)

    @property
    def recency_score(self) -> float:
        """시간 경과에 따른 감쇠 점수 (망각 곡선 시뮬레이션)"""
        hours_elapsed = (datetime.now() - self.last_accessed).total_seconds() / 3600
        decay_rate = 0.1
        return max(0.0, 1.0 * (2.718 ** (-decay_rate * hours_elapsed)))

    @property
    def composite_score(self) -> float:
        """중요도, 최신성, 접근 빈도를 조합한 복합 점수"""
        frequency_score = min(1.0, self.access_count / 10)
        return (
            0.4 * self.importance +
            0.35 * self.recency_score +
            0.25 * frequency_score
        )


class HierarchicalMemoryManager:
    """계층적 메모리 관리자: 단기 / 작업 / 장기 메모리를 통합 관리"""

    def __init__(self, user_id: str):
        self.user_id = user_id
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

        # 단기 메모리: 현재 세션의 최근 대화
        self.short_term: list[BaseMessage] = []
        self.short_term_limit = 10

        # 작업 메모리: 현재 대화의 핵심 정보 (요약)
        self.working_memory: str = ""

        # 장기 메모리: 벡터 DB 기반
        self.long_term_store = Chroma(
            collection_name=f"long_term_{user_id}",
            embedding_function=self.embeddings,
            persist_directory=f"./memory/{user_id}",
        )

        # 메모리 인덱스 (메타데이터 관리)
        self.memory_index: dict[str, MemoryEntry] = {}

    def add_interaction(self, human_msg: str, ai_msg: str) -> None:
        """새로운 대화 상호작용을 메모리에 추가"""
        # 단기 메모리에 추가
        self.short_term.append(HumanMessage(content=human_msg))
        self.short_term.append(AIMessage(content=ai_msg))

        # 단기 메모리 한도 초과 시 오래된 대화를 장기 메모리로 이전
        if len(self.short_term) > self.short_term_limit * 2:
            self._consolidate_to_long_term()

        # 작업 메모리 업데이트
        self._update_working_memory(human_msg, ai_msg)

    def _consolidate_to_long_term(self) -> None:
        """단기 메모리의 오래된 대화를 요약하여 장기 메모리로 이전"""
        old_messages = self.short_term[:4]  # 가장 오래된 2턴
        self.short_term = self.short_term[4:]

        # 대화 내용 요약
        conversation_text = "\n".join(
            f"{'User' if isinstance(m, HumanMessage) else 'AI'}: {m.content}"
            for m in old_messages
        )
        summary_prompt = f"다음 대화에서 기억할 가치가 있는 핵심 정보를 추출하세요:\n{conversation_text}"
        summary = self.llm.invoke(summary_prompt).content

        # 중요도 평가
        importance = self._evaluate_importance(summary)

        # 장기 메모리에 저장
        memory_id = hashlib.md5(summary.encode()).hexdigest()
        self.long_term_store.add_texts(
            texts=[summary],
            metadatas=[{
                "memory_id": memory_id,
                "user_id": self.user_id,
                "timestamp": datetime.now().isoformat(),
                "importance": importance,
                "type": "conversation_summary",
            }],
            ids=[memory_id],
        )
        self.memory_index[memory_id] = MemoryEntry(
            content=summary,
            timestamp=datetime.now(),
            importance=importance,
        )

    def _evaluate_importance(self, content: str) -> float:
        """메모리 내용의 중요도를 LLM으로 평가 (0.0 ~ 1.0)"""
        prompt = (
            f"다음 정보의 중요도를 0.0에서 1.0 사이의 숫자로만 평가하세요. "
            f"사용자의 개인 정보, 선호도, 반복되는 패턴은 높은 점수를, "
            f"일반적인 인사말이나 사소한 대화는 낮은 점수를 주세요.\n"
            f"정보: {content}\n점수:"
        )
        response = self.llm.invoke(prompt).content.strip()
        try:
            return max(0.0, min(1.0, float(response)))
        except ValueError:
            return 0.5

    def _update_working_memory(self, human_msg: str, ai_msg: str) -> None:
        """작업 메모리(현재 대화 요약)를 업데이트"""
        prompt = (
            f"현재 대화 요약:\n{self.working_memory}\n\n"
            f"새로운 대화:\nUser: {human_msg}\nAI: {ai_msg}\n\n"
            f"위 내용을 반영하여 대화 요약을 업데이트하세요. 3-5문장으로 핵심만 포함하세요:"
        )
        self.working_memory = self.llm.invoke(prompt).content

    def retrieve_relevant_memories(self, query: str, k: int = 5) -> list[str]:
        """쿼리와 관련된 장기 메모리를 검색"""
        results = self.long_term_store.similarity_search_with_score(query, k=k)
        memories = []
        for doc, score in results:
            memory_id = doc.metadata.get("memory_id")
            if memory_id and memory_id in self.memory_index:
                self.memory_index[memory_id].access_count += 1
                self.memory_index[memory_id].last_accessed = datetime.now()
            memories.append(doc.page_content)
        return memories

    def build_context(self, current_query: str) -> str:
        """현재 쿼리에 대한 전체 컨텍스트를 구성"""
        relevant_memories = self.retrieve_relevant_memories(current_query)
        short_term_text = "\n".join(
            f"{'User' if isinstance(m, HumanMessage) else 'AI'}: {m.content}"
            for m in self.short_term[-6:]  # 최근 3턴
        )
        context = (
            f"## 사용자에 대한 장기 기억\n"
            + "\n".join(f"- {m}" for m in relevant_memories)
            + f"\n\n## 현재 대화 요약\n{self.working_memory}"
            + f"\n\n## 최근 대화\n{short_term_text}"
        )
        return context

from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser


class UserPreferences(BaseModel):
    """사용자 선호도 프로필"""
    response_style: Optional[str] = Field(
        None, description="선호하는 응답 스타일 (간결/상세/코드중심)"
    )
    language_level: Optional[str] = Field(
        None, description="기술 수준 (초급/중급/고급)"
    )
    interests: list[str] = Field(
        default_factory=list, description="관심 분야 목록"
    )
    preferred_language: Optional[str] = Field(
        None, description="선호하는 프로그래밍 언어"
    )
    communication_tone: Optional[str] = Field(
        None, description="선호하는 대화 톤 (격식/반말/친근)"
    )


class UserProfile(BaseModel):
    """통합 사용자 프로필"""
    user_id: str
    name: Optional[str] = None
    occupation: Optional[str] = None
    company: Optional[str] = None
    location: Optional[str] = None
    preferences: UserPreferences = Field(default_factory=UserPreferences)
    known_facts: list[str] = Field(
        default_factory=list, description="사용자에 대해 알려진 사실들"
    )
    interaction_count: int = 0
    first_interaction: Optional[datetime] = None
    last_interaction: Optional[datetime] = None
    topics_discussed: list[str] = Field(default_factory=list)
    updated_at: Optional[datetime] = None


class ProfileUpdater:
    """대화에서 자동으로 사용자 프로필을 업데이트하는 시스템"""

    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        self.parser = JsonOutputParser()

    def extract_profile_updates(
        self, conversation: str, current_profile: UserProfile
    ) -> dict:
        """대화 내용에서 프로필 업데이트 정보를 추출"""
        prompt = ChatPromptTemplate.from_template(
            "당신은 대화에서 사용자 정보를 추출하는 전문가입니다.\n\n"
            "현재 사용자 프로필:\n{current_profile}\n\n"
            "최근 대화:\n{conversation}\n\n"
            "대화에서 새로 발견된 사용자 정보를 JSON으로 추출하세요.\n"
            "변경 없는 필드는 포함하지 마세요.\n"
            "추출 가능한 필드: name, occupation, company, location, "
            "interests (리스트에 추가), known_facts (리스트에 추가), "
            "response_style, language_level, preferred_language\n\n"
            "JSON 형식으로만 응답하세요:"
        )
        chain = prompt | self.llm | self.parser
        updates = chain.invoke({
            "current_profile": current_profile.model_dump_json(indent=2),
            "conversation": conversation,
        })
        return updates

    def apply_updates(
        self, profile: UserProfile, updates: dict
    ) -> UserProfile:
        """추출된 업데이트를 프로필에 적용"""
        if "name" in updates:
            profile.name = updates["name"]
        if "occupation" in updates:
            profile.occupation = updates["occupation"]
        if "company" in updates:
            profile.company = updates["company"]
        if "location" in updates:
            profile.location = updates["location"]
        if "interests" in updates:
            for interest in updates["interests"]:
                if interest not in profile.preferences.interests:
                    profile.preferences.interests.append(interest)
        if "known_facts" in updates:
            for fact in updates["known_facts"]:
                if fact not in profile.known_facts:
                    profile.known_facts.append(fact)
        if "response_style" in updates:
            profile.preferences.response_style = updates["response_style"]
        if "language_level" in updates:
            profile.preferences.language_level = updates["language_level"]
        if "preferred_language" in updates:
            profile.preferences.preferred_language = updates["preferred_language"]

        profile.interaction_count += 1
        profile.last_interaction = datetime.now()
        profile.updated_at = datetime.now()
        if not profile.first_interaction:
            profile.first_interaction = datetime.now()

        return profile


# 사용 예시
updater = ProfileUpdater()
profile = UserProfile(user_id="user_001")

conversation = """
User: 안녕하세요, 김영주입니다. 네이버에서 백엔드 개발하고 있어요.
AI: 반갑습니다 김영주님! 백엔드 개발자시군요.
User: 네, Python과 Go를 주로 사용합니다. 요즘 LangChain에 관심이 많아요.
AI: LangChain은 LLM 애플리케이션 개발에 정말 유용한 프레임워크죠!
"""

updates = updater.extract_profile_updates(conversation, profile)
# 결과: {"name": "김영주", "occupation": "백엔드 개발자", "company": "네이버",
#        "interests": ["LangChain"], "preferred_language": "Python"}
profile = updater.apply_updates(profile, updates)

from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.tools import tool


@dataclass
class CoreMemory:
    """항상 컨텍스트에 포함되는 핵심 메모리"""
    persona: str = "당신은 친절한 AI 어시스턴트입니다."
    user_info: str = "아직 사용자에 대해 알려진 정보가 없습니다."
    max_chars: int = 2000

    def update_persona(self, new_content: str) -> str:
        if len(new_content) > self.max_chars:
            return "Error: 코어 메모리 용량 초과"
        self.persona = new_content
        return f"Persona 업데이트 완료: {new_content[:50]}..."

    def update_user_info(self, new_content: str) -> str:
        if len(new_content) > self.max_chars:
            return "Error: 코어 메모리 용량 초과"
        self.user_info = new_content
        return f"User info 업데이트 완료: {new_content[:50]}..."

    def append_user_info(self, additional_info: str) -> str:
        updated = f"{self.user_info}\n- {additional_info}"
        if len(updated) > self.max_chars:
            return "Error: 코어 메모리 용량 초과. 아카이브로 이전하세요."
        self.user_info = updated
        return f"User info에 추가 완료: {additional_info}"


class MemGPTStyleAgent:
    """MemGPT 아키텍처를 시뮬레이션하는 에이전트"""

    def __init__(self, user_id: str):
        self.user_id = user_id
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

        # 3계층 메모리 초기화
        self.core_memory = CoreMemory()
        self.recall_memory: list[dict] = []  # 전체 대화 이력
        self.archival_memory = Chroma(
            collection_name=f"archival_{user_id}",
            embedding_function=self.embeddings,
            persist_directory=f"./archival/{user_id}",
        )

        # 메모리 관리 도구 정의
        self.tools = self._define_tools()

    def _define_tools(self) -> list:
        core = self.core_memory
        archival = self.archival_memory
        recall = self.recall_memory

        @tool
        def core_memory_append(info: str) -> str:
            """코어 메모리의 사용자 정보에 새 항목을 추가합니다."""
            return core.append_user_info(info)

        @tool
        def core_memory_replace(old_text: str, new_text: str) -> str:
            """코어 메모리의 특정 텍스트를 새 텍스트로 교체합니다."""
            if old_text in core.user_info:
                core.user_info = core.user_info.replace(old_text, new_text)
                return f"교체 완료: '{old_text}' -> '{new_text}'"
            return f"Error: '{old_text}'를 코어 메모리에서 찾을 수 없습니다."

        @tool
        def archival_memory_insert(content: str) -> str:
            """장기 아카이브 메모리에 정보를 저장합니다."""
            archival.add_texts(
                texts=[content],
                metadatas=[{
                    "timestamp": datetime.now().isoformat(),
                    "user_id": self.user_id,
                }],
            )
            return f"아카이브 저장 완료: {content[:50]}..."

        @tool
        def archival_memory_search(query: str, k: int = 3) -> str:
            """아카이브 메모리에서 관련 정보를 검색합니다."""
            results = archival.similarity_search(query, k=k)
            if not results:
                return "아카이브에서 관련 정보를 찾을 수 없습니다."
            return "\n".join(
                f"[{i+1}] {doc.page_content}" for i, doc in enumerate(results)
            )

        @tool
        def conversation_search(query: str) -> str:
            """과거 대화 이력에서 키워드로 검색합니다."""
            matches = [
                entry for entry in recall
                if query.lower() in entry["content"].lower()
            ]
            if not matches:
                return "관련 대화를 찾을 수 없습니다."
            return "\n".join(
                f"[{entry['timestamp']}] {entry['role']}: {entry['content']}"
                for entry in matches[-5:]
            )

        return [
            core_memory_append,
            core_memory_replace,
            archival_memory_insert,
            archival_memory_search,
            conversation_search,
        ]

    def build_system_prompt(self) -> str:
        """시스템 프롬프트에 코어 메모리를 포함"""
        return (
            f"# 시스템 지침\n{self.core_memory.persona}\n\n"
            f"# 사용자 정보 (코어 메모리)\n{self.core_memory.user_info}\n\n"
            f"# 메모리 관리 지침\n"
            f"- 사용자에 대한 새로운 정보를 발견하면 core_memory_append를 사용하세요.\n"
            f"- 기존 정보가 변경되면 core_memory_replace를 사용하세요.\n"
            f"- 상세한 기술 정보나 긴 내용은 archival_memory_insert로 저장하세요.\n"
            f"- 과거 대화를 참조해야 하면 conversation_search를 사용하세요.\n"
        )

    def chat(self, user_message: str) -> str:
        """사용자 메시지를 처리하고 응답을 생성"""
        # 리콜 메모리에 사용자 메시지 기록
        self.recall_memory.append({
            "role": "user",
            "content": user_message,
            "timestamp": datetime.now().isoformat(),
        })

        # LLM에 코어 메모리 + 도구와 함께 요청
        response = self.llm.bind_tools(self.tools).invoke([
            {"role": "system", "content": self.build_system_prompt()},
            {"role": "user", "content": user_message},
        ])

        # 도구 호출이 있으면 실행 (실제 구현에서는 에이전트 루프)
        ai_response = response.content or "메모리를 업데이트했습니다."

        # 리콜 메모리에 AI 응답 기록
        self.recall_memory.append({
            "role": "assistant",
            "content": ai_response,
            "timestamp": datetime.now().isoformat(),
        })

        return ai_response

import numpy as np
from typing import Optional
from dataclasses import dataclass
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from rank_bm25 import BM25Okapi
from sentence_transformers import CrossEncoder


@dataclass
class SearchResult:
    content: str
    score: float
    source: str  # "dense", "sparse", "hybrid"
    metadata: dict


class HybridMemoryRetriever:
    """하이브리드 검색 + 리랭킹 기반 메모리 검색 파이프라인"""

    def __init__(self, collection_name: str):
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.vectorstore = Chroma(
            collection_name=collection_name,
            embedding_function=self.embeddings,
        )
        # BM25를 위한 문서 코퍼스
        self.documents: list[str] = []
        self.doc_metadata: list[dict] = []
        self.bm25: Optional[BM25Okapi] = None

        # Cross-encoder 리랭커
        self.reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

    def add_memory(self, content: str, metadata: dict) -> None:
        """메모리를 벡터 DB와 BM25 인덱스 모두에 추가"""
        # 벡터 DB에 추가
        self.vectorstore.add_texts(
            texts=[content], metadatas=[metadata]
        )
        # BM25 인덱스에 추가
        self.documents.append(content)
        self.doc_metadata.append(metadata)
        # BM25 인덱스 재구축
        tokenized_docs = [doc.split() for doc in self.documents]
        self.bm25 = BM25Okapi(tokenized_docs)

    def _dense_search(self, query: str, k: int = 20) -> list[SearchResult]:
        """밀집 벡터 검색 (의미적 유사도)"""
        results = self.vectorstore.similarity_search_with_score(query, k=k)
        return [
            SearchResult(
                content=doc.page_content,
                score=1.0 / (1.0 + score),  # 거리를 유사도로 변환
                source="dense",
                metadata=doc.metadata,
            )
            for doc, score in results
        ]

    def _sparse_search(self, query: str, k: int = 20) -> list[SearchResult]:
        """희소 벡터 검색 (BM25 키워드 매칭)"""
        if self.bm25 is None:
            return []
        tokenized_query = query.split()
        scores = self.bm25.get_scores(tokenized_query)
        top_indices = np.argsort(scores)[-k:][::-1]
        return [
            SearchResult(
                content=self.documents[i],
                score=float(scores[i]),
                source="sparse",
                metadata=self.doc_metadata[i],
            )
            for i in top_indices if scores[i] > 0
        ]

    def _reciprocal_rank_fusion(
        self,
        dense_results: list[SearchResult],
        sparse_results: list[SearchResult],
        k: int = 60,
        dense_weight: float = 0.6,
        sparse_weight: float = 0.4,
    ) -> list[SearchResult]:
        """RRF(Reciprocal Rank Fusion)로 두 검색 결과를 결합"""
        doc_scores: dict[str, float] = {}
        doc_map: dict[str, SearchResult] = {}

        for rank, result in enumerate(dense_results):
            rrf_score = dense_weight / (k + rank + 1)
            doc_scores[result.content] = doc_scores.get(result.content, 0) + rrf_score
            doc_map[result.content] = result

        for rank, result in enumerate(sparse_results):
            rrf_score = sparse_weight / (k + rank + 1)
            doc_scores[result.content] = doc_scores.get(result.content, 0) + rrf_score
            if result.content not in doc_map:
                doc_map[result.content] = result

        sorted_docs = sorted(doc_scores.items(), key=lambda x: x[1], reverse=True)
        return [
            SearchResult(
                content=content,
                score=score,
                source="hybrid",
                metadata=doc_map[content].metadata,
            )
            for content, score in sorted_docs
        ]

    def _rerank(
        self, query: str, candidates: list[SearchResult], top_k: int = 5
    ) -> list[SearchResult]:
        """Cross-encoder로 후보를 재정렬"""
        if not candidates:
            return []
        pairs = [(query, r.content) for r in candidates]
        scores = self.reranker.predict(pairs)
        for i, score in enumerate(scores):
            candidates[i].score = float(score)
        candidates.sort(key=lambda x: x.score, reverse=True)
        return candidates[:top_k]

    def search(
        self,
        query: str,
        top_k: int = 5,
        use_reranking: bool = True,
    ) -> list[SearchResult]:
        """전체 하이브리드 검색 + 리랭킹 파이프라인"""
        # 1단계: 밀집 + 희소 검색
        dense_results = self._dense_search(query, k=20)
        sparse_results = self._sparse_search(query, k=20)

        # 2단계: RRF 결합
        fused_results = self._reciprocal_rank_fusion(
            dense_results, sparse_results
        )

        # 3단계: 리랭킹 (선택적)
        if use_reranking and fused_results:
            return self._rerank(query, fused_results[:15], top_k=top_k)
        return fused_results[:top_k]


# 사용 예시
retriever = HybridMemoryRetriever("user_memories")
retriever.add_memory(
    "사용자는 Python 백엔드 개발자로 FastAPI를 주로 사용한다.",
    {"type": "profile", "importance": 0.9}
)
retriever.add_memory(
    "지난주 LangChain의 LCEL 파이프라인에 대해 질문했다.",
    {"type": "conversation", "importance": 0.7}
)
results = retriever.search("FastAPI 관련 이전 대화")