에이전트 메모리: 단기, 장기 및 에피소드 메모리

메인 썸네일 이미지 프롬프트: 어두운 배경에 네온 톤으로 표현된 인간 뇌의 단면 일러스트레이션. 세 영역이 명확하게 구분되고 라벨이 붙어 있음. 해마(Hippocampus) 영역은 파란색으로 빛나며 "에피소드 메모리 (Episodic Memory): 무슨 일이 일어났는가"라고 라벨링됨. 전전두엽 피질(Prefrontal cortex)은 주황색으로 빛나며 "작업 메모리 (Working Memory): 내가 지금 무엇을 하고 있는가"라고 라벨링됨. 분산된 노드 네트워크는 초록색으로 빛나며 "의미론적 메모리 (Semantic Memory): 내가 무엇을 알고 있는가"라고 라벨링됨. 화살표는 영역 간의 정보 흐름을 보여줌. 과학적이면서도 접근하기 쉬우며, 메모리 아키텍처(Memory architecture)가 신경학적이고 시각적으로 표현됨.

메모리 아키텍처 다이어그램 이미지 프롬프트: 어두운 배경에 수직으로 배열된 네 개의 저장 상자. 상단: "인-컨텍스트 윈도우 (In-Context Window, 작업 메모리)" — 가장 빠르고 가장 작으며 일시적임, RAM 칩 아이콘으로 표시됨. 두 번째: "외부 벡터 저장소 (External Vector Store, 의미론적 메모리)" — 빠른 검색, 영구적임, 검색 아이콘이 있는 실린더로 표시됨. 세 번째: "키-값 저장소 (Key-Value Store, 에피소드 메모리)" — 구조화된 사실, 데이터베이스 아이콘으로 표시됨. 하단: "파인튜닝된 가중치 (Fine-Tuned Weights, 절차적 메모리)" — 업데이트가 가장 느리고 가장 영구적임, 잠금 장치가 있는 뇌로 표시됨. 상자 사이의 읽기/쓰기 속도를 보여주는 화살표. 깔끔하고 기술적이며, 계층 구조가 통찰력을 제공함.

메모리 검색 흐름 이미지 프롬프트: 왼쪽의 에이전트에게 쿼리(Query)가 도착함. 네 개의 평행한 화살표가 오른쪽의 네 가지 메모리 소스로 향함: 대화 기록 (짧은 채팅 말풍선), 벡터 데이터베이스 (의미론적 검색 시각화), 구조화된 데이터베이스 (테이블 아이콘), 모델 가중치 (뇌 아이콘). 각 소스는 관련 항목을 반환함. 오른쪽의 "메모리 퓨전 (Memory Fusion)" 박스가 결과들을 결합함. 에이전트는 풍부해진 컨텍스트(Context)를 확인함. 여러 저장소로부터의 검색이 곧 아키텍처임.

LLM(대규모 언어 모델)과의 모든 대화는 제로(Zero) 상태에서 시작됩니다.

당신은 프로젝트를 설명합니다. 당신의 선호도를 설명합니다. 당신의 제약 사항을 설명합니다. 컨텍스트를 제공하는 데 5분을 소비합니다. 그리고 내일 다시 돌아옵니다. 그 모든 과정을 다시 반복해야 합니다.

모델은 세션 사이의 그 어떤 것도 기억하지 못합니다. 컨텍스트 윈도우 (Context Window)는 닫힙니다. 상태 (State)는 사라집니다. 모든 상호작용은 에이전트의 업무 첫날과 같습니다.

인간의 생산성은 기억에 의존합니다. 우리는 지난번에 무엇이 효과적이었는지 기억합니다. 과거의 경험을 바탕으로 무언가를 구축합니다. 우리는 우리의 도구, 동료, 그리고 반복되는 문제들을 알고 있습니다. 우리는 매일 처음부터 다시 시작하지 않습니다.

메모리를 가진 에이전트도 이와 같이 동작합니다. 그들은 과거의 대화를 기억합니다. 관련 사실을 회상합니다. 성공적인 전략을 저장합니다. 시간이 지남에 따라 사용자의 선호도와 프로젝트 컨텍스트 (Project Context)에 대한 모델을 구축합니다.

이 포스트에서는 네 가지 유형의 에이전트 메모리 (Agent Memory)를 처음부터 모두 구축합니다.

네 가지 유형의 에이전트 메모리

import os
import json
import time
...

유형 1: 작업 메모리 (Working Memory, 대화 기록)

class WorkingMemory:
    """
    컨텍스트 윈도우 (Context Window) 내에 존재하는 단기 메모리 (Short-term memory).
...

유형 2: 의미론적 메모리 (Semantic Memory, 벡터 스토어)

from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity

...

유형 3: 에피소드 메모리 (Episodic Memory, 구조화된 저장소)

import sqlite3
from contextlib import contextmanager

...

모두 통합하기: 메모리 증강 에이전트 (Memory-Augmented Agent)

class MemoryAgent:
    """
    네 가지 메모리 유형이 모두 통합된 완전한 에이전트.
    축적된 메모리를 기반으로 응답을 개인화함.
    """

    def __init__(self, agent_id: str = "agent_default",
                 model: str = "claude-3-5-haiku-20241022"):
        self.agent_id = agent_id
        self.client   = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
        self.model    = model

        self.working_memory  = WorkingMemory(max_turns=15)
        self.semantic_memory = SemanticMemory(
            persist_path=f"./memory_{agent_id}/semantic")
        self.episodic_memory = EpisodicMemory(
            db_path=f"./memory_{agent_id}/episodic.db")

        self._session_id = hashlib.md5(
            str(time.time()).encode()).hexdigest()[:8]

def _build_memory_context(self, query: str) -> str:
"""관련 메모리들을 컨텍스트 블록으로 조립합니다."""
parts = []

    prefs = self.episodic_memory.get_all_preferences()
    if prefs:
        parts.append("User preferences: " + 
                     "; ".join(f"{k}={v}" for k, v in prefs.items()))

    key_facts = self.episodic_memory.get_facts_by_category("identity")
    key_facts += self.episodic_memory.get_facts_by_category("project")
    if key_facts:
        facts_str = "; ".join(f"{f['key']}={f['value']}" for f in key_facts[:5])
        parts.append(f"Known facts: {facts_str}")

    relevant_memories = self.semantic_memory.recall(query, top_k=3)
    if relevant_memories:
        mem_str = "\n".join(
            f"- [{m['type']}] {m['content']}" for m in relevant_memories)
        parts.append(f"Relevant past context:\n{mem_str}")

    recent_episodes = self.episodic_memory.get_recent_episodes(3, success_only=True)
    if recent_episodes:
        ep_str = "; ".join(ep["action"][:40] for ep in recent_episodes)
        parts.append(f"Recent successful actions: {ep_str}")

    return "\n\n".join(parts) if parts else ""

def chat(self, user_message: str, verbose: bool = False) -> str:
    self.working_memory.add("user", user_message)

    memory_context = self._build_memory_context(user_message)

    system = f"""You are a helpful AI assistant with memory of past interactions.

Use the provided context to personalize your responses.

{f'Memory context:{chr(10)}{memory_context}' if memory_context else ''}

Adapt your response to the user's known preferences and expertise level."""

response = self.client.messages.create(
model = self.model,
max_tokens = 800,
system = system,
messages = self.working_memory.get_messages()
)
answer = response.content[0].text
self.working_memory.add("assistant", answer)

    self.semantic_memory.remember(
        f"User asked: {user_message[:100]}",
        memory_type = "conversation",
        metadata    = {"session": self._session_id}
    )