이 글은 누구를 위한 것인가
- AI 어시스턴트가 이전 대화를 기억하게 하고 싶은 팀
- 사용자별로 개인화된 기억을 관리해야 하는 개발자
- MemGPT 같은 장기 메모리 시스템을 직접 구현하려는 엔지니어
들어가며
"저번에 말한 프로젝트 기억해요?"라고 AI에게 물었을 때 기억하지 못하면 사용자 경험이 깨진다. LLM의 컨텍스트 윈도우는 세션이 끝나면 사라진다. 외부 메모리 시스템이 필요하다.
이 글은 bluefoxdev.kr의 AI 메모리 시스템 을 참고하여 작성했습니다.
1. 메모리 유형 설계
[LLM 메모리 유형]
에피소딕 메모리 (Episodic):
특정 사건/대화 기억
"지난주에 Python 에러 질문했다"
저장: 날짜, 내용, 감정 상태
시맨틱 메모리 (Semantic):
사용자 프로파일, 사실
"사용자는 백엔드 개발자, Python 선호"
저장: 키-값 구조화 정보
절차적 메모리 (Procedural):
사용자 선호 방식
"짧게 답하는 것을 선호, 코드 예시 원함"
저장: 행동 패턴, 선호도
[메모리 저장 전략]
Working Memory: 현재 컨텍스트 (대화 기록)
→ LLM 컨텍스트 윈도우
Short-term Memory: 최근 N세션
→ Redis (빠른 접근, TTL 설정)
Long-term Memory: 영구 저장
→ 벡터 DB (pgvector, Pinecone)
→ RDB (사용자 프로파일)
[메모리 검색 트리거]
사용자 이름 언급 → 사용자 프로파일 로드
이전 프로젝트 언급 → 관련 에피소딕 로드
선호도 관련 질문 → 절차적 메모리 로드
2. 장기 메모리 시스템 구현
import anthropic
import json
import uuid
from dataclasses import dataclass, field
from datetime import datetime
import numpy as np
client = anthropic.Anthropic()
@dataclass
class Memory:
id: str = field(default_factory=lambda: str(uuid.uuid4()))
user_id: str = ""
type: str = "episodic" # episodic, semantic, procedural
content: str = ""
embedding: list[float] = field(default_factory=list)
importance: float = 0.5
created_at: datetime = field(default_factory=datetime.utcnow)
last_accessed: datetime = field(default_factory=datetime.utcnow)
access_count: int = 0
tags: list[str] = field(default_factory=list)
class MemoryStore:
"""벡터 DB 기반 장기 메모리"""
def __init__(self):
self.memories: list[Memory] = [] # 실제: pgvector/Pinecone
def _embed(self, text: str) -> list[float]:
import openai
oai = openai.OpenAI()
resp = oai.embeddings.create(input=text, model="text-embedding-3-small")
return resp.data[0].embedding
async def store(self, memory: Memory):
memory.embedding = self._embed(memory.content)
self.memories.append(memory)
async def retrieve(
self,
query: str,
user_id: str,
top_k: int = 5,
memory_type: str | None = None,
) -> list[Memory]:
query_emb = np.array(self._embed(query))
relevant = [
m for m in self.memories
if m.user_id == user_id and
(memory_type is None or m.type == memory_type)
]
if not relevant:
return []
scored = []
for memory in relevant:
emb = np.array(memory.embedding)
sim = np.dot(query_emb, emb) / (
np.linalg.norm(query_emb) * np.linalg.norm(emb) + 1e-8
)
# 시간 감쇠: 오래된 메모리는 점수 낮춤
days_old = (datetime.utcnow() - memory.last_accessed).days
decay = np.exp(-0.01 * days_old)
final_score = sim * decay * memory.importance
scored.append((final_score, memory))
scored.sort(key=lambda x: x[0], reverse=True)
# 접근 카운트 업데이트
result = [m for _, m in scored[:top_k]]
for m in result:
m.access_count += 1
m.last_accessed = datetime.utcnow()
return result
async def extract_and_store_memories(
conversation: list[dict],
user_id: str,
store: MemoryStore,
):
"""대화에서 기억할 정보 추출 및 저장"""
conversation_text = "\n".join(
f"{m['role']}: {m['content']}" for m in conversation[-10:]
)
response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=500,
messages=[{
"role": "user",
"content": f"""다음 대화에서 기억해야 할 정보를 추출하세요.
대화:
{conversation_text}
JSON으로 반환 (없으면 빈 배열):
[
{{
"type": "episodic/semantic/procedural",
"content": "기억할 내용",
"importance": 0.0-1.0,
"tags": ["태그1", "태그2"]
}}
]"""
}]
)
memories_data = json.loads(response.content[0].text)
for m_data in memories_data:
memory = Memory(
user_id=user_id,
type=m_data["type"],
content=m_data["content"],
importance=m_data["importance"],
tags=m_data.get("tags", []),
)
await store.store(memory)
async def chat_with_memory(
user_message: str,
user_id: str,
store: MemoryStore,
conversation_history: list[dict],
) -> str:
"""장기 메모리를 활용한 AI 응답"""
# 관련 메모리 검색
relevant_memories = await store.retrieve(
query=user_message,
user_id=user_id,
top_k=5,
)
memory_context = ""
if relevant_memories:
memory_context = "관련 기억:\n" + "\n".join(
f"- [{m.type}] {m.content}" for m in relevant_memories
)
system_prompt = f"""당신은 사용자의 개인 AI 어시스턴트입니다.
{memory_context}
위 기억을 참고하여 자연스럽게 대화하세요.
기억이 있으면 명시적으로 언급하지 말고 자연스럽게 활용하세요."""
messages = conversation_history + [{"role": "user", "content": user_message}]
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1000,
system=system_prompt,
messages=messages,
)
ai_response = response.content[0].text
# 대화에서 새 메모리 추출
updated_history = messages + [{"role": "assistant", "content": ai_response}]
await extract_and_store_memories(updated_history, user_id, store)
return ai_response
마무리
장기 메모리의 핵심은 선택적 기억이다. 모든 것을 기억하면 검색 품질이 떨어지고 비용이 증가한다. 중요도 스코어링과 시간 감쇠로 "잊을 것은 잊고 중요한 것은 오래 기억"하는 자연스러운 망각 메커니즘을 구현하면 인간의 기억에 가까워진다.