에이전트 장기 메모리 설계: 에피소딕·시맨틱 분리와 망각 전략

메모리 계층 설계: 두 가지 저장소

장기 메모리를 단일 벡터 DB에 모두 저장하는 방식은 간단하지만 두 가지 문제를 일으킨다. 첫째, 검색 시 에피소딕("어제 사용자가 요청한 것")과 시맨틱("사용자의 선호도 일반화") 정보가 혼재되어 관련도 계산이 흐려진다. 둘째, 오래된 에피소드가 최신 컨텍스트를 오염시킨다.

권장 계층 구조:

• 에피소딕 메모리: 개별 대화 세션의 요약본. TTL 7~30일. PostgreSQL + pgvector 또는 Redis Vector로 관리.
• 시맨틱 메모리: 반복 패턴에서 추출한 사실·선호도. TTL 무제한 또는 버전 관리. user_id + fact_key 기준 upsert.
• 워킹 메모리: 현재 컨텍스트 윈도우. 최근 N턴 + 검색된 메모리 스니펫만 포함.

망각 전략: TTL과 중요도 기반 압축

모든 메모리를 영구 보존하면 검색 노이즈가 증가하고 개인정보 리스크도 커진다. 다음 세 가지 망각 기법을 조합한다.

1. 시간 기반 TTL: 에피소딕 메모리는 30일 후 자동 만료. 갱신 없이 재접근되지 않은 메모리 우선 삭제.
2. 중요도 스코어링: 세션 종료 시 Claude에게 해당 대화의 핵심 사실을 1~5점으로 평가 요청. 3점 미만은 에피소딕에만 저장, 4점 이상은 시맨틱으로 승격.
3. 압축 배치: 주 1회 오래된 에피소딕 메모리를 배치로 요약·통합 후 원본 삭제. Batch API와 결합하면 비용 최소화.

import anthropic
from datetime import datetime, timedelta

client = anthropic.Anthropic()

def extract_and_score_memory(conversation_turns: list[dict]) -> dict:
    """세션 종료 시 메모리 추출 및 중요도 평가"""
    history_text = "\n".join(
        f"{t['role']}: {t['content']}" for t in conversation_turns
    )
    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=512,
        system=(
            "당신은 메모리 관리자입니다. 대화에서 장기 보존 가치가 있는 "
            "사실·선호도·결정사항을 JSON으로 추출하세요.\n"
            "형식: {\"facts\": [{\"content\": str, \"score\": 1-5, \"type\": \"episodic|semantic\"}]}"
        ),
        messages=[{"role": "user", "content": f"대화:\n{history_text}"}],
    )
    raw = response.content[0].text
    import json, re
    match = re.search(r"\{.*\}", raw, re.DOTALL)
    return json.loads(match.group()) if match else {"facts": []}

def store_memories(user_id: str, facts: list[dict], db_client) -> None:
    now = datetime.utcnow()
    for fact in facts:
        ttl = None if fact["type"] == "semantic" else now + timedelta(days=30)
        db_client.upsert(
            collection="memories",
            data={
                "user_id": user_id,
                "content": fact["content"],
                "score": fact["score"],
                "type": fact["type"],
                "expires_at": ttl,
                "created_at": now,
            }
        )

# 실제 세션 종료 후 호출 예시
# turns = [{"role": "user", "content": "..."}, ...]
# result = extract_and_score_memory(turns)
# store_memories("user-123", result["facts"], db)

운영 체크리스트

• [ ] 검색 시 필터 우선: 벡터 유사도만으로 검색하지 말고 user_id + type + expires_at > now 메타 필터 선적용
• [ ] 토큰 예산 설정: 워킹 메모리에 삽입할 메모리 스니펫은 최대 500토큰으로 제한
• [ ] 점수 임계값 튜닝: score ≥ 4를 시맨틱 승격 기준으로 시작, A/B 평가 후 조정
• [ ] GDPR 삭제 API: user_id 기준 전체 메모리 즉시 삭제 엔드포인트 필수 구현
• [ ] 압축 배치 모니터링: 주간 압축 후 검색 정밀도(Precision@5) 비교로 품질 회귀 감지
• [ ] 실패 모드: 메모리 DB 장애 시 워킹 메모리만으로 graceful degradation 처리