Memory-Augmented LLM 아키텍처¶
장기 대화와 컨텍스트 유지를 위한 메모리 기반 LLM 아키텍처 가이드.
개요¶
Memory-Augmented LLM은 외부 메모리 시스템을 통해 컨텍스트 윈도우의 한계를 극복하고, 장기 대화에서 일관성 있는 응답을 생성하는 아키텍처다.
| 구분 | 설명 |
|---|---|
| 핵심 문제 | Context window 한계 (128K-1M tokens) |
| 해결 방식 | 외부 메모리 + 동적 검색 |
| 적용 사례 | 장기 대화, 개인화 에이전트, 지식 관리 |
| 대표 시스템 | Mem0, MemGPT, Memorizing Transformers |
아키텍처 패턴¶
1. 기본 구조¶
┌─────────────────────────────────────────────────────────┐
│ Memory-Augmented LLM │
├─────────────────────────────────────────────────────────┤
│ │
│ User Query │
│ │ │
│ ▼ │
│ ┌─────────────┐ Query ┌─────────────────┐ │
│ │ Memory │◄────────────►│ Memory Store │ │
│ │ Retriever │ Retrieve │ │ │
│ └─────────────┘ │ - Short-term │ │
│ │ │ - Long-term │ │
│ │ Context │ - Episodic │ │
│ ▼ │ - Semantic │ │
│ ┌─────────────┐ └─────────────────┘ │
│ │ LLM │ ▲ │
│ │ (Frozen) │ │ │
│ └─────────────┘ │ │
│ │ │ │
│ │ Response │ Write │
│ ▼ │ │
│ ┌─────────────┐ │ │
│ │ Memory │─────────────────────┘ │
│ │ Writer │ │
│ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────┘
2. 메모리 유형¶
| 유형 | 설명 | 유지 기간 | 예시 |
|---|---|---|---|
| Short-term | 현재 대화 컨텍스트 | 세션 내 | 최근 10개 대화 |
| Long-term | 영구 저장 정보 | 영구 | 사용자 선호도 |
| Episodic | 특정 이벤트/경험 | 중장기 | "지난주 회의 내용" |
| Semantic | 일반화된 지식 | 영구 | "사용자는 Python을 선호" |
| Procedural | 작업 수행 방법 | 영구 | "보고서 작성 순서" |
주요 시스템¶
Mem0¶
Mem0는 그래프 기반 메모리로 LLM의 장기 대화 일관성을 향상시킨다.
아키텍처:
┌─────────────────────────────────────────┐
│ Mem0 │
├─────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────────┐ │
│ │ Memory │ │ Graph Memory │ │
│ │ Extractor │──►│ │ │
│ │ (LLM) │ │ Entities ──────│──┼─► Neo4j
│ └─────────────┘ │ Relations ─────│ │
│ │ Properties ────│ │
│ └─────────────────┘ │
│ │
│ ┌─────────────┐ ┌─────────────────┐ │
│ │ Vector │ │ Embedding │ │
│ │ Store │◄──│ Model │ │
│ │ (Qdrant) │ │ │ │
│ └─────────────┘ └─────────────────┘ │
│ │
└─────────────────────────────────────────┘
사용법:
from mem0 import Memory
# 초기화
m = Memory()
# 메모리 추가
m.add("사용자는 커피를 좋아하고 아침에 마신다", user_id="user123")
m.add("사용자는 Python 백엔드 개발자이다", user_id="user123")
# 검색
results = m.search("사용자의 음료 선호도는?", user_id="user123")
# → "사용자는 커피를 좋아하고 아침에 마신다"
# 그래프 쿼리 (관계 기반)
related = m.get_relations(entity="user123", relation_type="PREFERS")
MemGPT¶
OS의 가상 메모리처럼 계층적 메모리 관리:
┌─────────────────────────────────────────┐
│ MemGPT │
├─────────────────────────────────────────┤
│ │
│ Main Context (In-context memory) │
│ ┌─────────────────────────────────────┐│
│ │ System Prompt │ Working Memory ││
│ │ │ (recent msgs) ││
│ └─────────────────────────────────────┘│
│ ▲ │ │
│ │ Page In │ Page Out │
│ │ ▼ │
│ ┌─────────────────────────────────────┐│
│ │ Archival Memory ││
│ │ (Vector DB - unlimited) ││
│ └─────────────────────────────────────┘│
│ │
│ Functions: │
│ - core_memory_append() │
│ - core_memory_replace() │
│ - archival_memory_insert() │
│ - archival_memory_search() │
│ │
└─────────────────────────────────────────┘
Memorizing Transformers¶
KV Cache를 외부 메모리로 확장:
# 개념적 구조
class MemorizingTransformer:
def __init__(self, base_model, memory_size=65536):
self.model = base_model
self.knn_memory = KNNMemory(size=memory_size)
def forward(self, x):
# Local attention
local_out = self.model.attention(x)
# KNN retrieval from external memory
retrieved = self.knn_memory.search(x, k=32)
# Gated fusion
gate = self.gate_network(local_out, retrieved)
output = gate * local_out + (1 - gate) * retrieved
return output
메모리 연산¶
1. Memory Extraction¶
대화에서 저장할 정보 추출:
EXTRACTION_PROMPT = """
다음 대화에서 장기적으로 기억할 가치가 있는 정보를 추출하세요.
대화:
{conversation}
추출할 정보 유형:
1. 사용자 선호도 (좋아하는 것, 싫어하는 것)
2. 사실 정보 (이름, 직업, 위치 등)
3. 관계 정보 (가족, 친구, 동료)
4. 이벤트 (계획, 약속, 과거 경험)
JSON 형식으로 출력:
"""
2. Memory Consolidation¶
중복 제거 및 정보 통합:
def consolidate_memories(memories: List[Memory]) -> List[Memory]:
# 유사한 메모리 그룹화
clusters = cluster_by_similarity(memories, threshold=0.85)
consolidated = []
for cluster in clusters:
if len(cluster) == 1:
consolidated.append(cluster[0])
else:
# LLM으로 여러 메모리 통합
merged = llm.merge_memories(cluster)
consolidated.append(merged)
return consolidated
3. Memory Retrieval¶
쿼리에 맞는 메모리 검색:
def retrieve_memories(query: str, user_id: str, k: int = 5):
# 1. Vector similarity search
vector_results = vector_store.search(
query_embedding=embed(query),
filter={"user_id": user_id},
k=k * 2
)
# 2. Recency weighting
scored = []
for mem in vector_results:
recency_score = decay_function(mem.timestamp)
final_score = mem.similarity * 0.7 + recency_score * 0.3
scored.append((mem, final_score))
# 3. Diversity sampling (MMR)
diverse = maximal_marginal_relevance(scored, k=k)
return diverse
그래프 기반 메모리¶
Entity-Relation 모델링¶
┌──────────────────────────────────────────────────────┐
│ Graph Memory │
├──────────────────────────────────────────────────────┤
│ │
│ [User: John] │
│ │ │
│ WORKS_AT │
│ │ │
│ ▼ │
│ [Company: TechCorp] ──LOCATED_IN──► [City: SF] │
│ │ │
│ HAS_PROJECT │
│ │ │
│ ▼ │
│ [Project: ML Pipeline] │
│ │ │
│ USES_TECH │
│ │ │
│ ▼ │
│ [Tech: Python] ◄──PREFERS── [User: John] │
│ │
└──────────────────────────────────────────────────────┘
쿼리 예시¶
// 사용자의 모든 관련 컨텍스트 조회
MATCH (u:User {id: "john"})-[r]->(n)
RETURN u, type(r), n
ORDER BY r.timestamp DESC
LIMIT 20
// 특정 프로젝트 관련 기억
MATCH path = (u:User)-[:WORKS_ON]->(p:Project)-[*1..2]->(related)
WHERE p.name = "ML Pipeline"
RETURN path
구현 패턴¶
1. Conversation Memory Manager¶
class ConversationMemoryManager:
def __init__(self, llm, vector_store, graph_store):
self.llm = llm
self.vector_store = vector_store
self.graph_store = graph_store
self.short_term = [] # 최근 N개 메시지
def add_message(self, role: str, content: str, user_id: str):
# Short-term에 추가
self.short_term.append({"role": role, "content": content})
if len(self.short_term) > 10:
self._flush_to_long_term(user_id)
def _flush_to_long_term(self, user_id: str):
# 오래된 메시지에서 메모리 추출
old_messages = self.short_term[:-5]
self.short_term = self.short_term[-5:]
# LLM으로 중요 정보 추출
memories = self.llm.extract_memories(old_messages)
# 저장
for mem in memories:
self.vector_store.add(mem.text, mem.embedding, user_id)
if mem.entities:
self.graph_store.add_relations(mem.entities, user_id)
def get_context(self, query: str, user_id: str) -> str:
# Long-term memory 검색
relevant = self.vector_store.search(query, user_id, k=5)
# Short-term + Long-term 결합
context = self._format_context(self.short_term, relevant)
return context
2. Hierarchical Memory¶
class HierarchicalMemory:
def __init__(self):
self.levels = {
"immediate": deque(maxlen=5), # 현재 대화
"session": deque(maxlen=50), # 세션 내
"persistent": VectorStore(), # 영구 저장
}
def add(self, memory: Memory):
self.levels["immediate"].append(memory)
# Summarization cascade
if len(self.levels["immediate"]) >= 5:
summary = self._summarize(self.levels["immediate"])
self.levels["session"].append(summary)
if len(self.levels["session"]) >= 50:
consolidated = self._consolidate(self.levels["session"])
self.levels["persistent"].add(consolidated)
def retrieve(self, query: str, k: int = 10):
results = []
# 모든 레벨에서 검색
results.extend(list(self.levels["immediate"]))
results.extend(self._search_recent(self.levels["session"], k//2))
results.extend(self.levels["persistent"].search(query, k//2))
return self._rank_and_dedupe(results, k)
성능 최적화¶
검색 지연 시간¶
| 방법 | 지연 시간 | 설명 |
|---|---|---|
| In-memory vector | 1-5ms | 소규모 (< 100K) |
| Qdrant/Milvus | 5-20ms | 대규모 |
| Graph query | 10-50ms | 복잡한 관계 |
| LLM extraction | 200-500ms | 실시간 추출 |
최적화 전략¶
# 1. Async prefetch
async def prefetch_context(user_id: str, likely_topics: List[str]):
tasks = [
memory_store.search(topic, user_id)
for topic in likely_topics
]
results = await asyncio.gather(*tasks)
return flatten(results)
# 2. Cache layer
@lru_cache(maxsize=1000)
def get_user_context(user_id: str, query_hash: str):
return memory_store.search(query_hash, user_id)
# 3. Batch extraction
def batch_extract_memories(conversations: List[Conversation]):
# 여러 대화를 배치로 처리
return llm.batch_extract(conversations)
평가 지표¶
| 지표 | 설명 | 측정 방법 |
|---|---|---|
| Recall@k | 관련 메모리 검색률 | 정답 메모리 중 검색된 비율 |
| Coherence | 대화 일관성 | 이전 대화 참조 정확도 |
| Latency | 응답 지연 | 메모리 검색 + 생성 시간 |
| Memory Efficiency | 저장 효율 | 중복/노이즈 비율 |
벤치마크 결과 (Mem0, 2026)¶
| 모델 | LoCoMo (F1) | LongMemEval (Acc) |
|---|---|---|
| GPT-4 (vanilla) | 0.42 | 58% |
| GPT-4 + RAG | 0.51 | 67% |
| GPT-4 + Mem0 | 0.67 | 78% |
| Claude + Mem0 | 0.72 | 81% |
적용 사례¶
1. 개인화 어시스턴트¶
- 사용자 선호도 학습
- 이전 대화 참조
- 맥락 있는 추천
2. 고객 서비스 봇¶
- 고객 이력 파악
- 이슈 추적
- 해결 패턴 학습
3. 협업 에이전트¶
- 프로젝트 컨텍스트 유지
- 팀 지식 공유
- 의사결정 이력 관리
한계 및 고려사항¶
| 한계 | 대응 방안 |
|---|---|
| 프라이버시 | 사용자별 격리, 암호화 |
| 환각된 메모리 | 신뢰도 점수, 검증 파이프라인 |
| 메모리 충돌 | 타임스탬프 기반 우선순위 |
| 저장 비용 | 압축, 요약, 만료 정책 |
요약¶
Memory-Augmented LLM은 외부 메모리 시스템을 통해 LLM의 컨텍스트 한계를 극복한다. Mem0 같은 그래프 기반 접근법은 엔티티 관계를 모델링하여 복잡한 장기 대화에서 일관성을 유지한다. 효과적인 메모리 추출, 통합, 검색 파이프라인 설계가 핵심이다.