메모리와 상태 관리 — 에이전트가 기억하는 방법

에이전트가 기억을 잃는 문제

에이전트에게 "내 이름은 철수야"라고 말하고 다음 날 다시 접속하면 기억하지 못합니다. 컨텍스트 윈도우가 초기화되기 때문입니다.

실용적인 에이전트는 두 종류의 메모리가 필요합니다.

flowchart TB
    subgraph MEMORY["에이전트 메모리 체계"]
        direction LR
        
        subgraph SHORT["단기 메모리\n(현재 세션)"]
            S1["현재 대화 내용"]
            S2["작업 중인 데이터"]
            S3["도구 실행 결과"]
        end

        subgraph LONG["장기 메모리\n(세션 간 유지)"]
            L1["사용자 프로필\n이름, 선호도"]
            L2["과거 대화 요약"]
            L3["도메인 지식\n문서, 사실"]
        end
    end

    SHORT -->|"중요 정보 저장"| LONG
    LONG -->|"관련 정보 검색"| SHORT

---

단기 메모리: 컨텍스트 윈도우 관리

대화가 길어지면 컨텍스트 윈도우가 넘칩니다. 요약으로 해결합니다.

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

llm = ChatOpenAI(model="gpt-4o-mini")

def summarize_conversation(messages: list) -> str:
    """대화 기록을 요약해 컨텍스트를 줄입니다."""
    conversation_text = "\n".join([
        f"{m.type}: {m.content}" for m in messages
    ])
    
    summary = llm.invoke([
        SystemMessage("다음 대화를 핵심만 3문장으로 요약해줘."),
        HumanMessage(conversation_text)
    ])
    return summary.content

def trim_messages(messages: list, max_messages: int = 20) -> list:
    """메시지가 너무 많으면 오래된 것을 요약으로 대체합니다."""
    if len(messages) <= max_messages:
        return messages
    
    # 오래된 메시지 요약
    old_messages = messages[:-max_messages]
    recent_messages = messages[-max_messages:]
    
    summary = summarize_conversation(old_messages)
    
    return [
        SystemMessage(f"이전 대화 요약: {summary}"),
        *recent_messages
    ]

---

LangGraph 상태에 메모리 추가

from typing import Annotated, Optional
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    user_name: Optional[str]          # 사용자 이름
    task_context: Optional[str]       # 현재 작업 컨텍스트
    completed_tasks: list             # 완료된 작업 목록

from langchain_core.tools import tool

@tool
def remember_user_info(key: str, value: str) -> str:
    """사용자의 중요 정보를 기억합니다.
    
    Args:
        key: 정보 종류 (예: 'name', 'preference', 'goal')
        value: 저장할 값
    """
    # 실제 구현에서는 DB 저장
    return f"✅ 기억 완료: {key} = {value}"

@tool
def recall_user_info(key: str) -> str:
    """이전에 기억한 사용자 정보를 불러옵니다."""
    # 실제 구현에서는 DB 조회
    return f"{key}에 대한 정보: [저장된 값]"

---

장기 메모리: 벡터 DB 연동

import chromadb
from chromadb.utils import embedding_functions
from datetime import datetime

chroma_client = chromadb.PersistentClient(path="./agent_memory")
openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key=os.getenv("OPENAI_API_KEY"),
    model_name="text-embedding-3-small"
)

memory_collection = chroma_client.get_or_create_collection(
    name="agent_long_term_memory",
    embedding_function=openai_ef
)

def store_memory(content: str, memory_type: str, user_id: str):
    """중요한 정보를 장기 메모리에 저장합니다."""
    memory_id = f"{user_id}_{datetime.now().timestamp()}"
    
    memory_collection.add(
        ids=[memory_id],
        documents=[content],
        metadatas=[{
            "type": memory_type,   # "fact", "preference", "event"
            "user_id": user_id,
            "timestamp": datetime.now().isoformat()
        }]
    )

def retrieve_memory(query: str, user_id: str, top_k: int = 5) -> list[str]:
    """관련된 장기 기억을 검색합니다."""
    results = memory_collection.query(
        query_texts=[query],
        n_results=top_k,
        where={"user_id": user_id}
    )
    return results["documents"][0] if results["documents"] else []

---

메모리를 활용하는 에이전트 노드

flowchart TB
    INPUT["사용자 입력"] --> RECALL["장기 메모리 검색\n관련 과거 기억 로드"]
    RECALL --> AGENT["에이전트 노드\nLLM + 메모리 컨텍스트"]
    AGENT -->|"도구 호출"| TOOLS["도구 실행"]
    TOOLS --> AGENT
    AGENT -->|"중요 정보 발견"| STORE["장기 메모리 저장"]
    AGENT --> RESPONSE["응답 생성"]

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode, tools_condition

def agent_with_memory(state: AgentState):
    messages = state["messages"]
    user_id = state.get("user_id", "default")
    
    # 장기 메모리에서 관련 정보 검색
    last_user_message = messages[-1].content if messages else ""
    memories = retrieve_memory(last_user_message, user_id)
    
    # 메모리를 시스템 메시지에 주입
    memory_context = ""
    if memories:
        memory_context = f"\n\n[기억하고 있는 정보]\n" + "\n".join(memories)
    
    system_message = SystemMessage(
        f"당신은 사용자를 도와주는 AI 어시스턴트입니다.{memory_context}"
    )
    
    full_messages = [system_message] + messages
    response = llm_with_tools.invoke(full_messages)
    
    return {"messages": [response]}

---

메모리 유형별 설계

유형	저장 시점	검색 방법	예시
사실 메모리	사용자가 알려줄 때	키워드/의미 검색	"이름은 철수, 직업은 개발자"
선호도 메모리	사용자 피드백 시	카테고리 필터	"짧은 답변 선호"
에피소드 메모리	작업 완료 후	날짜 + 의미 검색	"2024-03 프로젝트 분석 완료"
지식 메모리	문서 인덱싱 시	의미 검색 (RAG)	회사 내부 문서

---

망각: 오래된 기억 정리

모든 것을 기억하면 노이즈가 됩니다.

from datetime import datetime, timedelta

def forget_old_memories(user_id: str, days_threshold: int = 90):
    """오래되고 중요하지 않은 메모리를 삭제합니다."""
    cutoff = (datetime.now() - timedelta(days=days_threshold)).isoformat()
    
    # 오래된 메모리 조회
    old_memories = memory_collection.get(
        where={
            "user_id": user_id,
            "type": "temporary",  # 임시 메모리만
            "timestamp": {"$lt": cutoff}
        }
    )
    
    if old_memories["ids"]:
        memory_collection.delete(ids=old_memories["ids"])
        return f"{len(old_memories['ids'])}개 오래된 메모리 삭제"
    
    return "삭제할 메모리 없음"

---

대화 요약 자동화

매 N턴마다 자동으로 오래된 대화를 요약합니다.

def should_summarize(messages: list, threshold: int = 15) -> bool:
    return len(messages) > threshold

def agent_node_with_summary(state: AgentState):
    messages = state["messages"]
    
    # 메시지가 너무 많으면 요약
    if should_summarize(messages):
        messages = trim_messages(messages)
    
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

---

정리

개념	내용
단기 메모리	현재 세션의 컨텍스트 윈도우
장기 메모리	벡터 DB에 영속 저장된 정보
대화 요약	컨텍스트 초과 방지 자동 압축
메모리 검색	현재 질문과 관련된 과거 기억 로드
망각	오래된/불필요한 메모리 정기 삭제

다음 편에서는 멀티 에이전트 패턴 — 여러 에이전트가 협업해 복잡한 작업을 처리하는 구조를 설계합니다.