8장: 메모리 관리와 상태 유지

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
 
# 세션별 메모리 저장소
store = {}
 
def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]
 
# 메모리가 적용된 체인
chain_with_memory = RunnableWithMessageHistory(
    runnable=prompt | model,
    get_session_history=get_session_history,
    input_messages_key="question",
    history_messages_key="history",
)
 
# 대화
config = {"configurable": {"session_id": "user-abc"}}
 
response1 = chain_with_memory.invoke(
    {"question": "Python의 GIL에 대해 설명해줘"},
    config=config,
)
 
response2 = chain_with_memory.invoke(
    {"question": "그럼 멀티스레딩은 어떻게 해야 하나?"},
    config=config,
)
# 이전 대화 맥락이 자동으로 포함됨

from langchain_core.messages import trim_messages
 
# 최근 10개 메시지만 유지
trimmer = trim_messages(
    max_tokens=4000,
    strategy="last",
    token_counter=model,
    include_system=True,  # 시스템 메시지는 항상 유지
    allow_partial=False,
)
 
chain_with_window = (
    RunnablePassthrough.assign(
        history=lambda x: trimmer.invoke(x["history"])
    )
    | prompt
    | model
)

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
 
class SummaryMemory:
    def __init__(self, model, max_messages: int = 10):
        self.model = model
        self.max_messages = max_messages
        self.summary = ""
        self.messages: list = []
 
    async def add_message(self, message):
        self.messages.append(message)
 
        if len(self.messages) > self.max_messages:
            # 오래된 메시지를 요약
            old_messages = self.messages[:self.max_messages // 2]
            summary_prompt = f"""
이전 대화 요약: {self.summary}
 
추가 대화:
{self._format_messages(old_messages)}
 
위 내용을 간결하게 요약해주세요.
"""
            self.summary = str(
                await self.model.ainvoke(summary_prompt)
            )
            self.messages = self.messages[self.max_messages // 2:]
 
    def get_context(self) -> list:
        context = []
        if self.summary:
            context.append(
                SystemMessage(content=f"이전 대화 요약: {self.summary}")
            )
        context.extend(self.messages)
        return context

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
 
class LongTermMemory:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings(
            model="text-embedding-3-small"
        )
        self.vectorstore = Chroma(
            collection_name="user_memories",
            embedding_function=self.embeddings,
        )
 
    def store_memory(self, user_id: str, content: str, metadata: dict = None):
        """대화에서 중요한 정보를 장기 메모리에 저장"""
        meta = {"user_id": user_id, **(metadata or {})}
        self.vectorstore.add_texts(
            texts=[content],
            metadatas=[meta],
        )
 
    def recall(self, user_id: str, query: str, k: int = 5) -> list[str]:
        """관련 장기 메모리 검색"""
        results = self.vectorstore.similarity_search(
            query=query,
            k=k,
            filter={"user_id": user_id},
        )
        return [doc.page_content for doc in results]
 
# 사용 예시
memory = LongTermMemory()
 
# 대화 중 중요 정보 저장
memory.store_memory(
    user_id="user-123",
    content="사용자는 Python과 TypeScript를 주로 사용하며, AI 엔지니어링에 관심이 있다.",
    metadata={"type": "preference"},
)
 
# 새 세션에서 관련 기억 검색
relevant_memories = memory.recall(
    user_id="user-123",
    query="추천 프로그래밍 언어",
)

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.memory import InMemoryStore
 
# 크로스 스레드 메모리 저장소
memory_store = InMemoryStore()
 
app = graph.compile(
    checkpointer=PostgresSaver(conn_string),
    store=memory_store,
)
 
# 노드에서 메모리 스토어 접근
def agent_node(state, config, store):
    """에이전트 노드에서 장기 메모리 활용"""
    user_id = config["configurable"]["user_id"]
 
    # 기존 메모리 검색
    memories = store.search(
        namespace=("memories", user_id),
        query=state["messages"][-1].content,
    )
 
    # 메모리를 컨텍스트에 포함
    context = "\n".join(m.value["content"] for m in memories)
 
    response = model.invoke(
        f"사용자 컨텍스트: {context}\n\n{state['messages']}"
    )
 
    # 중요 정보 메모리에 저장
    store.put(
        namespace=("memories", user_id),
        key=f"memory-{len(memories)}",
        value={"content": "새로 학습한 정보..."},
    )
 
    return {"messages": [response]}

from llama_index.core.memory import ChatMemoryBuffer
 
# 토큰 기반 윈도우 메모리
memory = ChatMemoryBuffer.from_defaults(
    token_limit=3000,
)
 
# 대화 메시지 추가
memory.put(ChatMessage(role="user", content="안녕하세요"))
memory.put(ChatMessage(role="assistant", content="안녕하세요! 무엇을 도와드릴까요?"))
 
# 현재 메모리의 메시지 조회 (토큰 한도 내)
messages = memory.get()

from semantic_kernel.memory import SemanticTextMemory
from semantic_kernel.connectors.memory.azure_cognitive_search import (
    AzureCognitiveSearchMemoryStore,
)
 
# Azure Cognitive Search 기반 장기 메모리
memory_store = AzureCognitiveSearchMemoryStore(
    search_endpoint="https://myorg-search.search.windows.net",
    admin_key="your-admin-key",
)
 
memory = SemanticTextMemory(
    storage=memory_store,
    embeddings_generator=embedding_service,
)
 
# 메모리 저장
await memory.save_information(
    collection="user_preferences",
    id="pref-001",
    text="사용자는 Python과 AI 엔지니어링을 선호합니다",
)
 
# 메모리 검색
results = await memory.search(
    collection="user_preferences",
    query="추천 기술 스택",
    limit=5,
)

from haystack.components.memory import ChatMessageMemory
from haystack import Pipeline
 
# 대화 메모리 컴포넌트
memory = ChatMessageMemory()
 
pipeline = Pipeline()
pipeline.add_component("memory", memory)
pipeline.add_component("prompt_builder", prompt_builder)
pipeline.add_component("generator", generator)
 
pipeline.connect("memory.messages", "prompt_builder.history")
pipeline.connect("prompt_builder", "generator")

from pydantic import BaseModel
from typing import Optional
 
class UserContext(BaseModel):
    """사용자 컨텍스트 상태"""
    user_id: str
    name: str
    language: str = "ko"
    preferences: dict = {}
    current_task: Optional[str] = None
    task_progress: dict = {}
 
class ConversationState(BaseModel):
    """대화 상태"""
    user_context: UserContext
    messages: list = []
    summary: str = ""
    turn_count: int = 0
    total_tokens_used: int = 0
 
# LangGraph에서 구조화된 상태 활용
from langgraph.graph import StateGraph
 
class AppState(TypedDict):
    user_context: UserContext
    messages: Annotated[list, add_messages]
    current_tool_results: dict
    error_count: int
 
graph = StateGraph(AppState)

class LayeredMemory:
    """계층화된 메모리 시스템"""
 
    def __init__(self, model, vectorstore, db):
        self.short_term = []  # 현재 대화 버퍼
        self.model = model
        self.vectorstore = vectorstore  # 장기 메모리
        self.db = db  # 영구 저장소
 
    async def get_context(self, user_id: str, query: str) -> dict:
        """모든 메모리 계층에서 컨텍스트 수집"""
        # 1. 단기: 최근 대화
        recent = self.short_term[-10:]
 
        # 2. 장기: 관련 기억 검색
        memories = self.vectorstore.similarity_search(
            query=query,
            k=3,
            filter={"user_id": user_id},
        )
 
        # 3. 영구: 사용자 프로필
        profile = await self.db.get_user_profile(user_id)
 
        return {
            "recent_messages": recent,
            "relevant_memories": [m.page_content for m in memories],
            "user_profile": profile,
        }
 
    async def process_and_store(self, user_id: str, message: str, response: str):
        """대화 후 메모리 업데이트"""
        # 단기 메모리에 추가
        self.short_term.append({"role": "user", "content": message})
        self.short_term.append({"role": "assistant", "content": response})
 
        # 중요 정보 추출 및 장기 메모리 저장
        importance = await self._assess_importance(message, response)
        if importance > 0.7:
            self.vectorstore.add_texts(
                texts=[f"User: {message}\nAssistant: {response}"],
                metadatas=[{"user_id": user_id, "importance": importance}],
            )
 
        # 단기 메모리 한도 관리
        if len(self.short_term) > 20:
            await self._summarize_old_messages()