A-MEM: Agentic Memory for LLM Agents 코드 분석

repo & 환경

논문: https://arxiv.org/abs/2502.12110
repo: https://github.com/WujiangXu/A-mem/tree/main

논문 구현 레포를 중점적으로 분석하였음. ChromaDB를 적용한 버전은 https://github.com/WujiangXu/A-mem-sys/, GitHub - agiresearch/A-mem: A-MEM: Agentic Memory for LLM Agents 참고

1. 논문의 핵심 아이디어

일반적인 RAG 메모리는 문서를 넣고 임베딩한 뒤 검색하는 정적인 구조라면, A-MEM은 새 메모리가 들어올 때 LLM이 노트를 구조화하고, 기존 메모리와 link를 연결하며, 관련된 과거 메모리의 설명까지 수정하게 한다.

새 상호작용
  → Note Construction
  → Link Generation
  → Memory Evolution
  → 임베딩 저장

사용자 질문
  → 유사도 top-k 검색
  → 링크된 이웃 메모리로 컨텍스트 확장
  → LLM 답변

1.1 Note Construction

하나의 메모리는 원문뿐만 아니라 다음의 요소들을 함께 갖는다.

content + timestamp + keywords + context + tags + embedding + links

keywords, context, tags는 LLM이 생성한다.

1.2 Link Generation

새 노트와 가장 가까운 기존 노트 top-k를 임베딩으로 찾고, LLM이 실제로 연결할 이웃을 결정한다.

1.3 Memory Evolution

새 메모리 때문에 기존 메모리의 의미가 달라질 수 있다. A-MEM은 필요할 경우 관련 이웃 메모리의 context와 속성을 갱신해 과거 메모리를 계속 재작성한다.

1.4 Retrieve Relative Memory

질문과 가까운 노트를 검색한 뒤, 그 노트에 연결된 이웃도 답변 컨텍스트로 가져온다.

2. 핵심 아이디어 구현

2.1 전체 실행 경로

A-mem/test_advanced.py의 흐름은 다음과 같다.

evaluate_dataset()
  └─ advancedMemAgent
       ├─ add_memory()
       │    └─ AgenticMemorySystem.add_note()
       │         ├─ MemoryNote.analyze_content()
       │         ├─ AgenticMemorySystem.process_memory()
       │         └─ SimpleEmbeddingRetriever.add_documents()
       └─ answer_question()
            ├─ generate_query_llm()
            ├─ find_related_memories_raw()
            └─ LLM 답변 생성

LoCoMo의 대화 turn을 하나씩 메모리로 추가하고, 질문마다 관련 메모리를 검색해 답을 만든 뒤 F1, BLEU, ROUGE 등을 계산한다.

2.2 Note Construction

위치: A-mem/memory_layer.py

• MemoryNote.analyze_content(): LLM으로 keywords, context, tags를 생성한다.
• AgenticMemorySystem.add_note(): 노트 생성부터 진화, 저장, 임베딩 추가까지의 쓰기 경로 전체를 오케스트레이션한다.

실제 검색 문서는 다음처럼 만들어진다.

"content:" + note.content
+ " context:" + note.context
+ " keywords: " + ", ".join(note.keywords)
+ " tags: " + ", ".join(note.tags)

2.3 Link Generation과 Memory Evolution

위치: AgenticMemorySystem.process_memory()

1. find_related_memories(note.content, k=5)로 기존 이웃을 찾는다.
2. 새 노트와 이웃 메모리들을 evolution_system_prompt에 넣는다.
3. LLM이 should_evolve, action( strengthen, update_neighbor)를 결정한다.
4. strengthen이면 새 노트의 links와 tags를 바꾼다.
5. update_neighbor이면 기존 이웃 노트의 context와 tags를 덮어쓴다.

기존 이웃 노트의 context와 tags는 즉시 변경되지만, 검색기가 보관하는 문서 문자열과 임베딩은 즉시 갱신되지 않는다. 따라서 다음 전체 재임베딩 전까지는 진화 이전의 representation을 기준으로 검색 후보가 선택될 수 있다. 기본 설정에서는 실제 진화가 100회 누적될 때 consolidate_memories()가 현재 메모리 상태로 검색기를 다시 만든다.

2.4 검색과 연결 이웃 확장

위치: SimpleEmbeddingRetriever, find_related_memories_raw()

• SimpleEmbeddingRetriever: SentenceTransformer 임베딩과 cosine similarity만 사용하는 in-memory 검색기.
• find_related_memories_raw(): top-k 노트를 찾고 각 노트의 links에 들어있는 이웃들도 답변 컨텍스트에 추가한다.

HybridRetriever도 정의돼 있지만 AgenticMemorySystem이 실제 생성하는 것은 SimpleEmbeddingRetriever다.

2.5 Robust 구현

위치: A-mem/memory_layer_robust.py, A-mem/llm_text_parsers.py

원 구현은 JSON Schema 응답 한 번으로 진화 결정을 받는다. Robust 구현은 LLM 호출을 아래처럼 나눈다.

1. evolution decision
2. strengthen details (조건부)
3. neighbor updates (조건부)

응답은 KEYWORDS:, CONTEXT:, TAGS: 같은 텍스트 프로토콜로 파싱한다.

2.6 실제 코드 읽기

2.6.1 Note Construction: MemoryNote

위치: A-mem/memory_layer.py의 MemoryNote.__init__,
MemoryNote.analyze_content

def __init__(
    self,
    content: str,
    id: Optional[str] = None,
    keywords: Optional[List[str]] = None,
    links: Optional[Dict] = None,
    importance_score: Optional[float] = None,
    retrieval_count: Optional[int] = None,
    timestamp: Optional[str] = None,
    last_accessed: Optional[str] = None,
    context: Optional[str] = None,
    evolution_history: Optional[List] = None,
    category: Optional[str] = None,
    tags: Optional[List[str]] = None,
    llm_controller: Optional[LLMController] = None
):
    # 입력: 원문 content와 선택적인 메타데이터, 메타데이터 생성에 쓸 LLM controller
    # 출력: 명시적 return은 없고, 초기화된 MemoryNote 객체의 필드를 self에 저장한다.

    self.content = content

    # keywords/context/category/tags 중 하나라도 없으면 LLM 분석을 실행한다.
    # !category 유무를 검사하지만 analyze_content()에서는 category를 만들지 않고 있음.!
    if llm_controller and any(param is None for param in [keywords, context, category, tags]):
        analysis = self.analyze_content(content, llm_controller)
        print("analysis", analysis)

        # 호출자가 값을 직접 주었다면 그 값을 우선하고, 없을 때만 LLM 결과를 쓴다.
        keywords = keywords or analysis["keywords"]
        context = context or analysis["context"]
        tags = tags or analysis["tags"]

    # 노트 저장용 식별자는 UUID다.
    self.id = id or str(uuid.uuid4())
    self.keywords = keywords or []

    # links는 뒤에서 LLM이 고른 이웃의 정수 인덱스를 담는다.
    self.links = links or []
    self.importance_score = importance_score or 1.0 # 안 쓰는 값인 듯
    self.retrieval_count = retrieval_count or 0

    # 입력 시각이 없으면 현재 시각을 사용한다.
    current_time = datetime.now().strftime("%Y%m%d%H%M")
    self.timestamp = timestamp or current_time
    self.last_accessed = last_accessed or current_time

    # 분석 실패 시 context는 General로 fallback한다.
    self.context = context or "General"

    # 모델이 context를 list로 반환해도 검색 문자열을 만들 수 있게 합친다.
    if isinstance(self.context, list):
        self.context = " ".join(self.context)

    self.evolution_history = evolution_history or []
    self.category = category or "Uncategorized"
    self.tags = tags or []

실제 LLM 응답을 파싱하는 핵심은 다음과 같다.
(프롬프트만 한글로 번역하였음.)

@staticmethod
def analyze_content(content: str, llm_controller: LLMController) -> Dict:
    # 입력: 분석할 원문 content, LLM 호출에 사용할 llm_controller
    # 출력: keywords, context, tags를 담은 Dict

    prompt = """다음 내용을 아래 기준에 따라 구조적으로 분석하세요.
        1. 가장 핵심적인 키워드를 식별하세요. 명사, 동사, 핵심 개념에 집중하세요.
        2. 핵심 주제와 맥락적 요소를 추출하세요.
        3. 관련 있는 분류 태그를 만드세요.

        응답은 다음 JSON 객체 형식으로 작성하세요.
        {
            "keywords": [
                // 핵심 개념과 용어를 포착하는 구체적이고 서로 구별되는 키워드
                // 중요도가 높은 순서로 정렬
                // 화자의 이름이나 시간을 나타내는 단어는 제외
                // 최소 세 개를 생성하되 불필요하게 중복하지 말 것
            ],
            "context":
                // 다음 내용을 요약한 한 문장:
                // - 주요 주제 또는 영역
                // - 핵심 주장 또는 요점
                // - 대상 독자 또는 목적
            ,
            "tags": [
                // 분류에 사용할 포괄적인 범주 또는 주제
                // 영역, 형식, 유형 태그를 포함
                // 최소 세 개를 생성하되 불필요하게 중복하지 말 것
            ]
        }

        분석할 내용:
        """ + content

    # 실제 코드는 세 필드를 required로 둔 JSON Schema와 함께 LLM을 호출한다.
    response = llm_controller.llm.get_completion(
        prompt,
        response_format={
            "type": "json_schema",
            "json_schema": {
                "name": "response",
                "schema": {
                    "type": "object",
                    "properties": {
                        "keywords": {"type": "array", "items": {"type": "string"}},
                        "context": {"type": "string"},
                        "tags": {"type": "array", "items": {"type": "string"}},
                    },
                    "required": ["keywords", "context", "tags"],
                    "additionalProperties": False,
                },
                "strict": True,
            },
        },
    )

... (json 파싱) ...

    return analysis

여기서 만든 keywords, context, tags는 이후 임베딩을 위한 문자열과 evolution prompt에 다시 들어간다.

2.6.2 임베딩 저장과 top-k 검색: SimpleEmbeddingRetriever

위치: A-mem/memory_layer.py의 SimpleEmbeddingRetriever

def __init__(self, model_name: str = 'all-MiniLM-L6-v2'):
    # 입력: SentenceTransformer에서 불러올 model_name
    # 출력: 명시적 return은 없고, 검색 모델과 빈 저장 공간을 self에 초기화한다.

    self.model = SentenceTransformer(model_name)

    # 별도 vector DB 없이 문서와 벡터를 프로세스 메모리에 둔다.
    self.corpus = []
    self.embeddings = None
    self.document_ids = {}

def add_documents(self, documents: List[str]):
    # 입력: 임베딩하고 저장할 문서 문자열의 List
    # 출력: 명시적 return은 없고, corpus와 embeddings를 내부 상태로 갱신한다.

    if not self.corpus:
        # 첫 입력은 corpus 전체를 설정하고 한 번에 임베딩한다.
        self.corpus = documents
        self.embeddings = self.model.encode(documents)
        self.document_ids = {doc: idx for idx, doc in enumerate(documents)}
    else:
        # 이후 입력은 새 문서만 임베딩해 기존 행렬 아래에 붙인다.
        start_idx = len(self.corpus)
        self.corpus.extend(documents)
        new_embeddings = self.model.encode(documents)
        if self.embeddings is None:
            self.embeddings = new_embeddings
        else:
            self.embeddings = np.vstack([self.embeddings, new_embeddings])

        # 문서의 배열 위치를 기록한다.
        for idx, doc in enumerate(documents):
            self.document_ids[doc] = start_idx + idx

def search(self, query: str, k: int = 5) -> List[Dict[str, float]]:
    # 입력: 검색 질의 query와 반환할 최대 결과 수 k
    # 출력: 타입 힌트와 달리 실제로는 유사도 상위 문서의 NumPy 인덱스 배열

    if not self.corpus:
        return []

    # 질문을 메모리와 같은 임베딩 공간으로 변환한다.
    query_embedding = self.model.encode([query])[0]

    # 모든 메모리와 cosine similarity를 계산한다.
    similarities = cosine_similarity([query_embedding], self.embeddings)[0]

    # 높은 점수부터 최대 k개의 배열 인덱스를 얻는다.
    top_k_indices = np.argsort(similarities)[-k:][::-1]

    # docstring/type hint와 달리 실제 반환값은 dict 목록이 아니라 NumPy 인덱스 배열이다.
    return top_k_indices

이 반환값이 UUID가 아니라 삽입 순서 인덱스라는 사실이 뒤의 링크 생성과 이웃 evolution에서 uuid를 사용하지 않는 배경이 된다.

2.6.3 노트 한 건을 저장하는 전체 쓰기 경로

위치: A-mem/memory_layer.py의 AgenticMemorySystem.add_note

def add_note(self, content: str, time: str = None, **kwargs) -> str:
    # 입력: 저장할 원문 content, 선택적인 시각 time과 MemoryNote 추가 인자
    # 출력: 저장을 마친 새 MemoryNote의 UUID 문자열

    # 생성자 안에서 먼저 keywords/context/tags를 생성한다.
    note = MemoryNote(
        content=content,
        llm_controller=self.llm_controller,
        timestamp=time,
        **kwargs
    )

    # 아직 self.memories에 넣기 전이므로 새 노트 자신은 이웃 후보가 되지 않는다.
    evo_label, note = self.process_memory(note)

    # evolution 처리가 끝난 새 노트를 UUID key로 저장한다.
    self.memories[note.id] = note

    # 원문과 구조화 메타데이터를 한 문자열로 만들어 검색기에 추가한다.
    self.retriever.add_documents([
        "content:" + note.content
        + " context:" + note.context
        + " keywords: " + ", ".join(note.keywords)
        + " tags: " + ", ".join(note.tags)
    ])

    # 실제 evolution이 발생했을 경우에만 재임베딩 주기를 카운트한다.
  	# 100이 되면 전체를 재임베딩한다.
    if evo_label == True:
        self.evo_cnt += 1
        if self.evo_cnt % self.evo_threshold == 0:
            self.consolidate_memories()

    return note.id

저장은 UUID key를 쓰지만 검색 결과와 링크는 정수 배열 위치를 쓴다.

2.6.4 Link Generation과 Memory Evolution

위치: A-mem/memory_layer.py의 AgenticMemorySystem.process_memory

process_memory()가 사용하는 evolution 프롬프트는
AgenticMemorySystem.__init__()에서 다음과 같이 정의된다.

def __init__(
    self,
    model_name: str = 'all-MiniLM-L6-v2',
    llm_backend: str = "sglang",
    llm_model: str = "gpt-4o-mini",
    evo_threshold: int = 100,
    api_key: Optional[str] = None,
    api_base: Optional[str] = None,
    sglang_host: str = "http://localhost",
    sglang_port: int = 30000
):
    # 입력: 임베딩 모델, LLM backend와 모델, 진화 임계값, backend 연결 설정
    # 출력: 명시적 return은 없고 메모리 저장소, 검색기, LLM, 진화 prompt를 초기화한다.

    self.memories = {}
    self.retriever = SimpleEmbeddingRetriever(model_name)
    self.llm_controller = LLMController(
        llm_backend,
        llm_model,
        api_key,
        api_base,
        sglang_host,
        sglang_port
    )

    self.evolution_system_prompt = '''
        당신은 지식 베이스를 관리하고 진화시키는 AI 메모리 진화 에이전트입니다.
        새 메모리 노트의 keywords와 context, 그리고 가장 가까운 여러 이웃 메모리를
        함께 분석하여 메모리를 어떻게 진화시킬지 결정하세요.

        새 메모리의 context:
        {context}
        content: {content}
        keywords: {keywords}

        가장 가까운 이웃 메모리:
        {nearest_neighbors_memories}

        다음 사항을 결정하세요.
        1. 다른 메모리와의 관계를 고려할 때 이 메모리를 진화시켜야 합니까?
        2. 어떤 action을 수행해야 합니까? (strengthen, update_neighbor)
           2.1 strengthen을 선택한다면 어느 메모리에 연결해야 하며,
               이 메모리의 갱신된 tags는 무엇입니까?
           2.2 update_neighbor를 선택한다면 메모리들을 종합적으로 이해한 결과를
               바탕으로 이웃의 context와 tags를 갱신하세요. 갱신할 필요가 없다면
               기존 값을 그대로 반환하세요. 입력된 이웃 순서대로 생성하세요.
        tags는 나중의 검색과 분류에 사용할 수 있도록 메모리 내용의 특징을
        반영해야 합니다.
        new_tags_neighborhood와 new_context_neighborhood의 길이는 각각 입력된
        이웃 수와 같아야 합니다.
        이웃 수는 {neighbor_number}개입니다.

        다음 구조의 JSON으로 결정 결과를 반환하세요.
        {{
            "should_evolve": True or False,
            "actions": ["strengthen", "update_neighbor"],
            "suggested_connections": ["neighbor_memory_ids"],
            "tags_to_update": ["tag_1", ..., "tag_n"],
            "new_context_neighborhood": ["new context", ..., "new context"],
            "new_tags_neighborhood": [
                ["tag_1", ..., "tag_n"],
                ...,
                ["tag_1", ..., "tag_n"]
            ]
        }}
    '''

    self.evo_cnt = 0
    self.evo_threshold = evo_threshold

def process_memory(self, note: MemoryNote) -> bool:
    # 입력: 아직 self.memories에 저장되지 않은 새 MemoryNote
    # 출력: 타입 힌트와 달리 실제로는 (should_evolve: bool, note: MemoryNote) tuple

    # 새 노트의 content만 검색 질의로 쓴다.
    # 방금 생성한 context/keywords/tags는 이 후보 검색에 포함되지 않는다. (왜 이렇게 했을까 의문. 논문의 설명과 다름.)
    neighbor_memory, indices = self.find_related_memories(note.content, k=5)

    # 새 노트의 구조화 정보와 검색된 이웃을 진화 prompt에 합친다.
    prompt_memory = self.evolution_system_prompt.format(
        context=note.context,
        content=note.content,
        keywords=note.keywords,
        nearest_neighbors_memories=neighbor_memory,
        neighbor_number=len(indices)
    )

    # 실제 코드는 아래 여섯 필드를 required로 둔 JSON Schema를 전달한다.
    response = self.llm_controller.llm.get_completion(
        prompt_memory,
        response_format={
            "type": "json_schema",
            "json_schema": {
                "name": "response",
                "schema": {
                    "type": "object",
                    "properties": {
                        "should_evolve": {"type": "boolean"},
                        "actions": {"type": "array", "items": {"type": "string"}},
                        "suggested_connections": {
                            "type": "array",
                            "items": {"type": "integer"}
                        },
                        "new_context_neighborhood": {
                            "type": "array",
                            "items": {"type": "string"}
                        },
                        "tags_to_update": {
                            "type": "array",
                            "items": {"type": "string"}
                        },
                        "new_tags_neighborhood": {
                            "type": "array",
                            "items": {
                                "type": "array",
                                "items": {"type": "string"}
                            }
                        }
                    },
                    "required": [
                        "should_evolve",
                        "actions",
                        "suggested_connections",
                        "tags_to_update",
                        "new_context_neighborhood",
                        "new_tags_neighborhood"
                    ],
                    "additionalProperties": False
                },
                "strict": True
            }
        }
    )

    try:
        # 모델이 JSON 앞뒤에 설명을 붙였으면 첫 {와 마지막 } 사이만 남긴다.
        response_cleaned = response.strip()
        if not response_cleaned.startswith('{'):
            start_idx = response_cleaned.find('{')
            if start_idx != -1:
                response_cleaned = response_cleaned[start_idx:]
        if not response_cleaned.endswith('}'):
            end_idx = response_cleaned.rfind('}')
            if end_idx != -1:
                response_cleaned = response_cleaned[:end_idx+1]

        response_json = json.loads(response_cleaned)
    except json.JSONDecodeError as e:
        # 파싱에 실패해도 새 노트는 버리지 않고 "진화 없음"으로 저장 경로에 돌려준다.
        print(f"JSON parsing error: {e}")
        print(f"Raw response: {response}")
        return False, note

    # 이 boolean이 아래 상태 변경 전체의 gate다.
    should_evolve = response_json["should_evolve"]
    if should_evolve:
        actions = response_json["actions"]
        for action in actions:
            if action == "strengthen":
                # LLM이 선택한 이웃은 UUID가 아니라 정수 인덱스다.
                suggest_connections = response_json["suggested_connections"]
                new_tags = response_json["tags_to_update"]

                # 링크를 추가하고 새 노트의 tag 전체를 교체한다.
                note.links.extend(suggest_connections)
                note.tags = new_tags

            elif action == "update_neighbor":
                # 기존 이웃에 적용할 새 context와 tags를 받는다.
                new_context_neighborhood = response_json["new_context_neighborhood"]
                new_tags_neighborhood = response_json["new_tags_neighborhood"]

                # 검색 인덱스를 MemoryNote 객체와 UUID key에 다시 대응시키는 평행 배열이다.
                noteslist = list(self.memories.values())
                notes_id = list(self.memories.keys())

                # 모델이 이웃 수보다 짧은 결과를 주면 가능한 범위까지만 갱신한다.
                for i in range(min(len(indices), len(new_tags_neighborhood))):
                    tag = new_tags_neighborhood[i]

                    # 현재 이웃에 대응하는 새 context 원소가 없으면 기존 context를 유지한다.
                    if i < len(new_context_neighborhood):
                        context = new_context_neighborhood[i]
                    else:
                        context = noteslist[indices[i]].context

                    # 임베딩 검색 결과의 정수 위치로 기존 메모리를 찾는다.
                    memorytmp_idx = indices[i]
                    notetmp = noteslist[memorytmp_idx]

                    # 논문 설명과 달리 keywords는 건드리지 않고 tags/context만 교체한다.
                    notetmp.tags = tag
                    notetmp.context = context
                    self.memories[notes_id[memorytmp_idx]] = notetmp

    return should_evolve, note

진화된 이웃의 임베딩은 즉시 바뀌지 않는다. 다음 코드가 기본 100회의 진화마다
검색기를 새로 만들고 현재 메타데이터를 다시 임베딩한다.

def consolidate_memories(self):
    # 입력: 별도 인자 없음. self.memories의 현재 상태를 사용한다.
    # 출력: 명시적 return은 없고, self.retriever를 새 검색기로 교체한다.

    try:
        model_name = self.retriever.model.get_config_dict()['model_name']
    except (AttributeError, KeyError):
        model_name = 'all-MiniLM-L6-v2'

    # 기존 벡터 행렬을 버리고 같은 모델로 빈 검색기를 다시 만든다.
    self.retriever = SimpleEmbeddingRetriever(model_name)

    # 현재 시점의 content/context/keywords/tags를 전부 다시 encode한다.
    for memory in self.memories.values():
        metadata_text = (
            f"{memory.context} "
            f"{' '.join(memory.keywords)} "
            f"{' '.join(memory.tags)}"
        )
        self.retriever.add_documents([memory.content + " , " + metadata_text])

2.6.5 Relative Retrieval: 링크된 이웃까지 가져오기

위치: A-mem/memory_layer.py의 find_related_memories_raw

def find_related_memories_raw(self, query: str, k: int = 5) -> List[MemoryNote]:
    # 입력: 검색 질의 query와 직접 검색할 최대 메모리 수 k
    # 출력: 타입 힌트와 달리 직접 검색된 메모리와 링크 이웃을 이어 붙인 문자열

    if not self.memories:
        return []

    # 질문과 직접 가까운 top-k 메모리의 배열 인덱스를 얻는다.
    indices = self.retriever.search(query, k)

    # 검색 인덱스를 삽입 순서의 MemoryNote 객체로 해석한다.
    all_memories = list(self.memories.values())
    memory_str = ""

    for i in indices:
        j = 0

        # 먼저 직접 검색된 메모리를 답변 컨텍스트에 붙인다.
        memory_str += (
            "talk start time:" + all_memories[i].timestamp
            + "memory content: " + all_memories[i].content
            + "memory context: " + all_memories[i].context
            + "memory keywords: " + str(all_memories[i].keywords)
            + "memory tags: " + str(all_memories[i].tags) + "\n"
        )

        # Link Generation이 저장한 정수 인덱스 목록을 읽는다.
        neighborhood = all_memories[i].links
        for neighbor in neighborhood:
            # 연결된 이웃 메모리도 같은 방식으로 컨텍스트에 추가한다.
            memory_str += (
                "talk start time:" + all_memories[neighbor].timestamp
                + "memory content: " + all_memories[neighbor].content
                + "memory context: " + all_memories[neighbor].context
                + "memory keywords: " + str(all_memories[neighbor].keywords)
                + "memory tags: " + str(all_memories[neighbor].tags) + "\n"
            )

            # append 후 제한을 검사하므로 링크가 충분하면 k+1개까지 들어간다.
            if j >= k:
                break
            j += 1

    return memory_str

만약 링크된 이웃들 중에 서로 중복되는 이웃이 존재한다면?

2.6.6 Robust 구현의 evolution

위치: A-mem/memory_layer_robust.py의 RobustMemoryNote.analyze_content,
RobustAgenticMemorySystem.process_memory

Robust Note Construction은 JSON Schema 대신 평문을 파싱 (KEYWORDS 등 키값 기반 파싱)

ANALYZE_CONTENT_PROMPT = """다음 내용을 분석하여 아래 항목을 제공하세요.
1. KEYWORDS: 가장 중요한 키워드입니다. 명사, 동사, 핵심 개념을 중심으로
   중요도가 높은 순서로 최소 세 개를 작성하세요. 화자 이름과 시간 표현은
   포함하지 마세요.
2. CONTEXT: 주요 주제, 핵심 요점, 목적을 요약한 한 문장입니다.
3. TAGS: 분류에 사용할 포괄적인 범주 또는 주제입니다. 영역, 형식, 유형을
   나타내는 태그를 최소 세 개 작성하세요.

각 header마다 하나의 section을 두고 반드시 다음 형식으로 응답하세요.

KEYWORDS: keyword1, keyword2, keyword3, ...
CONTEXT: 내용을 요약하는 한 문장
TAGS: tag1, tag2, tag3, ...

분석할 내용:
{content}"""

FOCUSED_KEYWORDS_PROMPT = """다음 텍스트의 주요 개념을 포착하는 키워드를
정확히 5개 나열하세요. 다른 설명 없이 쉼표로 구분한 키워드만 출력하세요.

텍스트: {content}"""

EVOLUTION_DECISION_PROMPT = """당신은 AI 메모리 진화 에이전트입니다.
새 메모리 노트와 가장 가까운 이웃들을 분석하여 진화가 필요한지 결정하세요.

새 메모리:
- Context: {context}
- Content: {content}
- Keywords: {keywords}

가장 가까운 이웃 메모리:
{nearest_neighbors_memories}

새 메모리와 이웃의 관계를 바탕으로 다음 중 하나를 결정하세요.
- NO_EVOLUTION: 독립적인 메모리이므로 변경할 필요가 없습니다.
- STRENGTHEN: 새 메모리를 일부 이웃과 연결하고 tags를 갱신해야 합니다.
- UPDATE_NEIGHBOR: 새롭게 이해한 내용을 바탕으로 이웃의 context와 tags를
  갱신해야 합니다.
- STRENGTHEN_AND_UPDATE: 연결 강화와 이웃 갱신을 모두 수행해야 합니다.

반드시 다음 형식으로 응답하세요.
DECISION: <NO_EVOLUTION, STRENGTHEN, UPDATE_NEIGHBOR,
           STRENGTHEN_AND_UPDATE 중 하나>
REASON: <간단한 이유>"""

STRENGTHEN_DETAILS_PROMPT = """새 메모리와 이웃을 바탕으로 갱신된 연결과
tags를 제시하세요.

새 메모리:
- Content: {content}
- Keywords: {keywords}

이웃 메모리:
{nearest_neighbors_memories}

새 메모리가 연결되어야 할 이웃의 index와 이 메모리를 가장 잘 설명하는 tags를
결정하세요.

반드시 다음 형식으로 응답하세요.
CONNECTIONS: 0, 2, 3
TAGS: tag1, tag2, tag3, ..."""

UPDATE_NEIGHBORS_PROMPT = """새 메모리와 이웃 메모리 전체를 종합적으로
이해한 결과를 바탕으로 각 이웃의 context와 tags를 갱신하세요.

새 메모리:
- Content: {content}
- Context: {context}

이웃 메모리:
{nearest_neighbors_memories}

0부터 {max_neighbor_idx}까지 각 이웃에 대해 갱신된 context와 tags를
제공하세요. 변경할 필요가 없다면 기존 값을 그대로 반복하세요.

이웃마다 하나의 block을 두고 반드시 다음 형식으로 응답하세요.

NEIGHBOR 0:
CONTEXT: 갱신된 context 문장
TAGS: tag1, tag2, tag3

NEIGHBOR 1:
CONTEXT: 갱신된 context 문장
TAGS: tag1, tag2, tag3

({neighbor_count}개 이웃 모두에 대해 계속 작성)"""

여기서 FOCUSED_KEYWORDS_PROMPT는 첫 분석 결과의 keywords가 비었을 때만
호출되고, 나머지 세 프롬프트는 decision 결과에 따라 조건부로 호출된다.

@staticmethod
def analyze_content(
    content: str,
    llm_controller: RobustLLMController
) -> Dict:
    # 입력: 분석할 원문 content, 평문 응답을 받을 RobustLLMController
    # 출력: keywords, context, tags를 담은 Dict

    prompt = ANALYZE_CONTENT_PROMPT.format(content=content)
    try:
        # 첫 번째 LLM 호출의 section marker를 파싱한다.
        response = llm_controller.llm.get_completion(prompt)
        analysis = parse_analyze_content(response, content)

        # keywords만 비었을 때는 keywords 전용 prompt로 한 번 더 요청한다.
        if not analysis["keywords"]:
            logger.info("Keywords empty after initial parse — retrying with focused prompt")
            retry_prompt = FOCUSED_KEYWORDS_PROMPT.format(content=content)
            retry_response = llm_controller.llm.get_completion(
                retry_prompt,
                temperature=0.3
            )
            from llm_text_parsers import _parse_list_items
            analysis["keywords"] = _parse_list_items(retry_response)

        # 빠진 필드를 보완하고 구조화 결과를 확정한다.
        analysis = validate_analysis_result(analysis, content)
        return analysis

    except Exception as e:
        # LLM이나 parser가 실패해도 원문 기반 휴리스틱으로 노트 저장을 계속한다.
        logger.error("Error analyzing content: %s", e)
        from llm_text_parsers import _heuristic_keywords, _heuristic_context
        return {
            "keywords": _heuristic_keywords(content),
            "context": _heuristic_context(content),
            "tags": _heuristic_keywords(content, 3),
        }

evolution은 최대 세 번의 작은 호출로 나뉜다.

def process_memory(self, note: RobustMemoryNote) -> tuple:
    # 입력: 아직 저장되지 않은 새 RobustMemoryNote
    # 출력: (진화 여부 bool, 처리된 RobustMemoryNote) tuple

    # 원 구현처럼 새 노트의 content만으로 기존 이웃을 찾는다.
    neighbor_memory, indices = self.find_related_memories(note.content, k=5)

    # 첫 노트처럼 이웃이 없으면 진화 LLM을 호출하지 않는다.
    if len(indices) == 0:
        return False, note

    try:
        # 1차 호출: 어떤 진화가 필요한지만 결정한다.
        decision_prompt = EVOLUTION_DECISION_PROMPT.format(
            context=note.context,
            content=note.content,
            keywords=note.keywords,
            nearest_neighbors_memories=neighbor_memory,
        )
        decision_response = self.llm_controller.llm.get_completion(decision_prompt)
        decision = parse_evolution_decision(decision_response)

        # 진화가 없으면 뒤의 두 LLM 호출을 모두 생략한다.
        if decision["decision"] == "NO_EVOLUTION":
            return False, note

        should_strengthen = decision["decision"] in (
            "STRENGTHEN",
            "STRENGTHEN_AND_UPDATE"
        )
        should_update = decision["decision"] in (
            "UPDATE_NEIGHBOR",
            "STRENGTHEN_AND_UPDATE"
        )

        # 2차 호출: strengthen이 필요할 때만 링크와 새 tags를 받는다.
        if should_strengthen:
            strengthen_prompt = STRENGTHEN_DETAILS_PROMPT.format(
                content=note.content,
                keywords=note.keywords,
                nearest_neighbors_memories=neighbor_memory,
            )
            strengthen_response = self.llm_controller.llm.get_completion(
                strengthen_prompt
            )
            strengthen = parse_strengthen_details(strengthen_response)

            note.links.extend(strengthen["connections"])
            if strengthen["tags"]:
                note.tags = strengthen["tags"]

        # 3차 호출: 기존 이웃 갱신이 필요할 때만 실행한다.
        if should_update:
            update_prompt = UPDATE_NEIGHBORS_PROMPT.format(
                content=note.content,
                context=note.context,
                nearest_neighbors_memories=neighbor_memory,
                max_neighbor_idx=len(indices) - 1,
                neighbor_count=len(indices),
            )
            update_response = self.llm_controller.llm.get_completion(update_prompt)
            neighbor_updates = parse_update_neighbors(
                update_response,
                len(indices)
            )

            noteslist = list(self.memories.values())
            notes_id = list(self.memories.keys())
            for i in range(min(len(indices), len(neighbor_updates))):
                upd = neighbor_updates[i]
                memorytmp_idx = indices[i]

                # 잘못된 인덱스를 받아도 전체 저장을 중단하지 않는다.
                if memorytmp_idx >= len(noteslist):
                    continue

                notetmp = noteslist[memorytmp_idx]
                if upd["tags"]:
                    notetmp.tags = upd["tags"]
                if upd["context"]:
                    notetmp.context = upd["context"]
                self.memories[notes_id[memorytmp_idx]] = notetmp

        return True, note

    except Exception as e:
        # 진화만 포기하고 새 메모리 자체는 저장할 수 있게 돌려준다.
        logger.error(
            "Evolution failed for note %s: %s — storing without evolution",
            note.id,
            e
        )
        return False, note

2.6.7 LoCoMo QA에서 검색된 메모리를 답변에 넣는 부분

위치: A-mem/test_advanced.py의 advancedMemAgent.generate_query_llm,
advancedMemAgent.answer_question

LoCoMo QA는 사용자 질문을 곧바로 검색하지 않고 먼저 검색용 keywords로
바꾼다.

def generate_query_llm(self, question):
    # 입력: LoCoMo의 자연어 질문 question
    # 출력: JSON 응답에서 꺼낸 검색 keywords 문자열

    prompt = f"""다음 질문을 바탕으로 여러 키워드를 생성하되,
            'cosmos'를 구분자로 사용하세요.

            질문: {question}

            선택한 텍스트를 "keywords" field에 넣은 JSON 객체로 응답하세요.

            응답 형식 예:
            {{"keywords": "keyword1, keyword2, keyword3"}}"""

    response = self.retriever_llm.llm.get_completion(
        prompt,
        response_format={
            "type": "json_schema",
            "json_schema": {
                "name": "response",
                "schema": {
                    "type": "object",
                    "properties": {
                        "keywords": {"type": "string"}
                    },
                    "required": ["keywords"],
                    "additionalProperties": False
                },
                "strict": True
            }
        }
    )

    try:
        response = json.loads(response)["keywords"]
    except:
        response = response.strip()
    return response

프롬프트 문장은 'cosmos'를 구분자로 요구하지만 예시는 쉼표를 사용한다. 코드도 별도의 cosmos 분리를 하지 않고 반환 문자열 전체를 검색 질의로 넘긴다.

Robust 평가 경로의 generate_query_llm()은 이 모순을 없애고 쉼표로 구분된 평문을 직접 요청한다.

def generate_query_llm(self, question):
    # 입력: LoCoMo의 자연어 질문 question
    # 출력: parser가 정리한 검색 keyword 문자열

    prompt = f"""다음 질문을 바탕으로 여러 키워드를 생성하고 쉼표로 구분하세요.

질문: {question}

키워드:"""

    response = self.retriever_llm.llm.get_completion(prompt)
    result = parse_keywords_response(response)
    return result

def answer_question(self, question: str, category: int, answer: str) -> str:
    # 입력: 평가 질문 question, 질문 유형 category, 정답 후보로 쓰일 answer
    # 출력: 타입 힌트와 달리 (LLM 응답, 실제 prompt, 검색 원문 context) tuple

    # 질문을 그대로 임베딩하지 않고 LLM으로 검색 keywords를 먼저 만든다.
    keywords = self.generate_query_llm(question)

    # find_related_memories_raw를 호출하므로 top-k에 링크 이웃까지 합쳐진다.
    raw_context = self.retrieve_memory(keywords, k=self.retrieve_k)
    context = raw_context

    assert category in [1, 2, 3, 4, 5]

    # 기본 temperature는 0.7이고 category 5 (Adversarial)만 별도 설정값을 쓴다.
    # category 5: 대화 기록에 실제 정답이 없는 질문
    # category 5에 대해서만 선택형 질문임.
    temperature = 0.7

    if category == 5:
        # adversarial 질문은 정답과 "언급되지 않음"의 순서를 무작위로 섞는다.
        # 답변할 수 없는 질문이기 때문에 '대화에서 언급되지 않음'이 정답임.
        answer_tmp = []
        if random.random() < 0.5:
            answer_tmp.append("대화에서 언급되지 않음")
            answer_tmp.append(answer)
        else:
            answer_tmp.append(answer)
            answer_tmp.append("대화에서 언급되지 않음")

        user_prompt = f"""다음 컨텍스트를 바탕으로 질문에 답하세요.
                        컨텍스트: {context}
                        질문: {question}

                        올바른 답을 선택하세요:
                        {answer_tmp[0]} 또는 {answer_tmp[1]}
                        짧은 답변:"""
        temperature = self.temperature_c5

    elif category == 2: # Temporal: 사건의 날짜·시점·기간·순서를 추론하는 질문
        # 시간 추론 질문은 대화 날짜를 이용해 근사 날짜를 답하도록 요구한다.
        user_prompt = f"""다음 컨텍스트를 바탕으로 질문에 답하세요.
                        근사 날짜를 답할 때 대화의 날짜를 사용하세요.
                        가능한 경우 대화에 나온 단어를 사용하여 최대한 짧게 답하고,
                        주어는 사용하지 마세요.

                        컨텍스트: {context}
                        질문: {question}
                        짧은 답변:"""

    elif category == 3: # Open-domain: 대화 내용에 상식이나 외부 지식을 결합해 추론하는 질문
        # open-domain 질문도 검색된 대화 context 안의 표현을 우선한다.
        user_prompt = f"""다음 컨텍스트를 바탕으로 아래 질문에 짧은 구 형태로
                        답하세요. 가능하면 컨텍스트에 있는 단어를 그대로 사용하세요.

                        컨텍스트: {context}
                        질문: {question}
                        짧은 답변:"""

    else: 
        # category 1과 4가 사용하는 기본 짧은 답변 프롬프트다.
        # category 1, 4: Multi-hop, Single-hop
        user_prompt = f"""다음 컨텍스트를 바탕으로 아래 질문에 짧은 구 형태로
                        답하세요. 가능하면 컨텍스트에 있는 단어를 그대로 사용하세요.

                        컨텍스트: {context}
                        질문: {question}
                        짧은 답변:"""

    # 최종 답변은 같은 memory system의 LLM이 생성한다.
    response = self.memory_system.llm_controller.llm.get_completion(
        user_prompt,
        response_format={
            "type": "json_schema",
            "json_schema": {
                "name": "response",
                "schema": {
                    "type": "object",
                    "properties": {"answer": {"type": "string"}},
                    "required": ["answer"],
                    "additionalProperties": False
                },
                "strict": True
            }
        },
        temperature=temperature
    )

    # 평가 결과에는 답뿐 아니라 실제 prompt와 검색 context도 남긴다.
    return response, user_prompt, raw_context

test_advanced_robust.py의 category별 답변 프롬프트도 위의 네 분기와 같은 지시를
사용한다. 마찬가지로 JSON Schema를 요구하지 않고 평문 응답을 바로 받는다.

• category==5에 대해서는 adversarial_answer를 answer 값으로 사용. (LLM이 잘못된 선택을 하도록 유도하는 오답)

!image.png

!image.png

다른 구현에서도 category 5에 대해서 비슷하게 평가.(https://github.com/aiming-lab/SimpleMem/blob/main/test_locomo10.py)

locomo 공식 구현에서는 다음과 같이 평가 (https://github.com/snap-research/locomo/blob/main/task_eval/evaluation.py)

!image.png

그런데 프롬프트에는 해당 지시가 주석 처리되어 있음.

!image.png

3. 논문 설명과 실제 구현이 어긋나는 부분

Note Construction의 timestamp 입력

논문 수식 (2)는 원문 $c_i$, timestamp $t_i$, prompt $P_{s1}$을 LLM에 함께
넣어 keywords $K_i$, tags $G_i$, context $X_i$를 생성한다고 설명한다.

그러나 MemoryNote.__init__()은 timestamp를 받더라도
analyze_content(content, llm_controller)만 호출한다. analyze_content()의
prompt에도 content만 들어가므로 timestamp는 메타데이터 생성에 사용되지 않는다.

Link Generation 후보 검색에서 새 노트의 표현

논문 수식 (3)은 새 노트의 content, keywords, tags, context를 합쳐 임베딩 $e_n$을 만들고, 3.2절의 수식 (4)~(5)는 이 $e_n$으로 기존 노트와 cosine similarity를 계산한다고 설명한다.

코드에서는 MemoryNote 생성이 끝나 note.context, note.keywords,
note.tags가 이미 존재하는데도 다음처럼 원문만 검색 질의로 전달한다.

neighbor_memory, indices = self.find_related_memories(
    note.content,
    k=5
)

따라서 기존 노트의 검색 벡터에는 content와 메타데이터가 들어가지만, 새 노트의 후보 검색 쿼리 벡터에는 content만 들어간다는 차이점이 있다.

Memory Evolution의 갱신 대상 필드

논문 3.3절은 각 이웃 노트에 대해 context, keywords, tags를 갱신할지 결정한다고
명시한다. 반면 원 구현과 Robust 구현은 기존 이웃의 context와 tags만
교체하고 keywords는 갱신하지 않는다.

Link Generation과 Memory Evolution의 호출 구조

논문 본문은 Link Generation을 수식 (6)과 $P_{s2}$, Memory Evolution을 수식
(7)과 $P_{s3}$로 분리하고, 부록 B.2와 B.3에도 별도 prompt를 제시한다.

원 구현의 process_memory()는 하나의 evolution_system_prompt를 한 번
호출하여 strengthen과 update_neighbor를 같은 JSON 응답에서 결정한다.
따라서 논문의 단계 구분이 실제 호출 구조에서는 합쳐져 있다. Robust 구현은
decision, strengthen, neighbor update를 조건부 호출로 다시 나누지만, 이것도
논문의 $P_{s2}$, $P_{s3}$를 그대로 구현한 구조는 아니다.

4. 무엇을 어떻게 돌려봤고, 무엇을 확인했나?

• 코드

  # multi_turn_memory_demo.py
  # 실행: python multi_turn_memory_demo.py
  """짧은 멀티턴 대화로 A-MEM의 메모리 진화를 보여준다."""
  
  from __future__ import annotations
  
  import argparse
  
  from memory_layer_robust import RobustAgenticMemorySystem
  
  DIALOGUE = [
      (
          "202607201000",
          "사용자: 나는 주말마다 여의도 한강공원에서 10km 달리기를 좋아해.\n"
          "어시스턴트: 꾸준히 달리네. 한강 보면서 뛰면 기분도 좋겠다.",
      ),
      (
          "202607211000",
          "사용자: 가을에 열리는 10km 대회에 참가하려고 요즘 기록을 줄이는 중이야.\n"
          "어시스턴트: 목표가 생겼구나. 요즘은 어느 정도 기록이 나와?",
      ),
      (
          "202607221000",
          "사용자: 그런데 무릎이 아파서 요즘은 달리기를 쉬고 잠실 한강공원에서 천천히 걸어.\n"
          "어시스턴트: 아이고, 기록보다 회복이 먼저겠다. 병원에는 가 봤어?",
      ),
      (
          "202607231000",
          "사용자: 병원에서 2주 동안 뛰지 말고 스트레칭만 하라고 했어. 가을 대회 참가는 아직 모르겠어.\n"
          "어시스턴트: 잘 다녀왔네. 대회는 무릎 상태를 보고 천천히 결정해도 되겠다.",
      ),
      (
          "202607241000",
          "사용자: 무릎이 조금 좋아져서 오늘 잠실에서 3km만 천천히 뛰었어. 당분간 10km는 무리하지 않을 거야.\n"
          "어시스턴트: 좋아지고 있다니 다행이다. 당분간은 지금처럼 짧게 뛰는 게 좋겠네.",
      ),
  ]
  
  def snapshot(memory_system) -> dict:
      return {
          memory_id: {
              "content": note.content,
              "context": note.context,
              "tags": list(note.tags),
              "links": list(note.links),
          }
          for memory_id, note in memory_system.memories.items()
      }
  
  def print_evolution(turn: int, before: dict, memory_system, new_id: str) -> None:
      memory_ids = list(memory_system.memories)
      note = memory_system.memories[new_id]
      new_number = memory_ids.index(new_id) + 1
      linked_numbers = [f"#{index + 1}" for index in note.links]
  
      print(f"\n----- TURN {turn} 메모리 반영 결과 -----")
      print(f"[새 메모리 #{new_number}]")
      print(f"새 메모리 keywords: {note.keywords}")
      print(f"새 메모리 context: {note.context}")
      print(f"새 메모리 tags: {note.tags}")
      print(f"새 메모리 links: {', '.join(linked_numbers) if linked_numbers else '없음'}")
  
      changed_numbers = []
      for number, (memory_id, old) in enumerate(before.items(), 1):
          current = memory_system.memories[memory_id]
          changes = []
          for field in ("context", "tags", "links"):
              new_value = getattr(current, field)
              if old[field] != new_value:
                  changes.append((field, old[field], new_value))
  
          if changes:
              changed_numbers.append(f"#{number}")
              print(f"\n[진화한 기존 메모리 #{number}]")
              print(f"원문: {old['content']}")
              for field, old_value, new_value in changes:
                  print(f"{field} 변경 전: {old_value}")
                  print(f"{field} 변경 후: {new_value}")
  
      print(f"\nTURN {turn} 요약")
      print(f"- 생성: 메모리 #{new_number}")
      print(f"- 연결: {', '.join(linked_numbers) if linked_numbers else '없음'}")
      print(f"- 진화: {', '.join(changed_numbers) if changed_numbers else '없음'}")
  
  def run_dialogue(dialogue=DIALOGUE, memory_system=None):
      memory_system = memory_system or RobustAgenticMemorySystem(
          model_name="nlpai-lab/KURE-v1",
          llm_backend="ollama",
          llm_model="gemma4:e2b",
      )
      for turn, (timestamp, content) in enumerate(dialogue, 1):
          print(f"\n\n===== TURN {turn} 입력 =====")
          print(content)
          before = snapshot(memory_system)
          new_id = memory_system.add_note(content, time=timestamp)
          print_evolution(turn, before, memory_system, new_id)
      return memory_system
  
  def main() -> None:
      parser = argparse.ArgumentParser(description=__doc__)
      parser.add_argument("--llm-model", default="gemma4:e2b")
      parser.add_argument("--embedding-model", default="nlpai-lab/KURE-v1")
      args = parser.parse_args()
      memory_system = RobustAgenticMemorySystem(
          model_name=args.embedding_model,
          llm_backend="ollama",
          llm_model=args.llm_model,
      )
      run_dialogue(memory_system=memory_system)
  
  if __name__ == "__main__":
      main()
  

• 간단한 멀티턴 예제(robust 버전) (gemma4:e2b)

  ===== TURN 1 입력 =====
  사용자: 나는 주말마다 여의도 한강공원에서 10km 달리기를 좋아해.
  어시스턴트: 꾸준히 달리네. 한강 보면서 뛰면 기분도 좋겠다.
  
  [LLM 중간 응답 #1 | gemma4:e2b]
  KEYWORDS: 달리기, 한강공원, 주말, 운동
  CONTEXT: A user enjoys running 10km every weekend at Yeouido Hangang Park.
  TAGS: 운동, 야외활동, 건강
  
  ----- TURN 1 메모리 반영 결과 -----
  [새 메모리 #1]
  새 메모리 keywords: ['달리기', '한강공원', '주말', '운동']
  새 메모리 context: A user enjoys running 10km every weekend at Yeouido Hangang Park.
  새 메모리 tags: ['운동', '야외활동', '건강']
  새 메모리 links: 없음
  
  TURN 1 요약
  - 생성: 메모리 #1
  - 연결: 없음
  - 진화: 없음
  
  ===== TURN 2 입력 =====
  사용자: 가을에 열리는 10km 대회에 참가하려고 요즘 기록을 줄이는 중이야.
  어시스턴트: 목표가 생겼구나. 요즘은 어느 정도 기록이 나와?
  
  [LLM 중간 응답 #2 | gemma4:e2b]
  KEYWORDS: 대회, 기록, 목표
  CONTEXT: A user is trying to improve their performance for an upcoming 10km race by currently working on reducing their time.
  TAGS: 운동, 목표 설정, 대화
  
  [LLM 중간 응답 #3 | gemma4:e2b]
  DECISION: STRENGTHEN
  REASON: The new memory about race performance and time reduction is related to the neighbor's memory about running 10km, suggesting a connection to fitness goals.
  
  [LLM 중간 응답 #4 | gemma4:e2b]
  CONNECTIONS: 0
  TAGS: ['운동', '건강']
  
  ----- TURN 2 메모리 반영 결과 -----
  [새 메모리 #2]
  새 메모리 keywords: ['대회', '기록', '목표']
  새 메모리 context: A user is trying to improve their performance for an upcoming 10km race by currently working on reducing their time.
  새 메모리 tags: ['운동', '건강']
  새 메모리 links: #1
  
  TURN 2 요약
  - 생성: 메모리 #2
  - 연결: #1
  - 진화: 없음
  
  ===== TURN 3 입력 =====
  사용자: 그런데 무릎이 아파서 요즘은 달리기를 쉬고 잠실 한강공원에서 천천히 걸어.
  어시스턴트: 아이고, 기록보다 회복이 먼저겠다. 병원에는 가 봤어?
  
  [LLM 중간 응답 #5 | gemma4:e2b]
  KEYWORDS: 무릎 통증, 달리기 중단, 회복, 산책
  CONTEXT: A user is resting from running due to knee pain and is now walking slowly at Jamsil Hangang Park for recovery.
  TAGS: 건강, 운동, 신체 회복
  
  [LLM 중간 응답 #6 | gemma4:e2b]
  DECISION: STRENGTHEN
  REASON: The new memory discusses knee pain and walking for recovery, which relates to the general theme of '운동' and '건강' found in the nearest neighbors about running and race goals. It should be linked to these themes.
  
  [LLM 중간 응답 #7 | gemma4:e2b]
  CONNECTIONS: 0
  TAGS: 무릎 통증, 회복, 산책
  
  ----- TURN 3 메모리 반영 결과 -----
  [새 메모리 #3]
  새 메모리 keywords: ['무릎 통증', '달리기 중단', '회복', '산책']
  새 메모리 context: A user is resting from running due to knee pain and is now walking slowly at Jamsil Hangang Park for recovery.
  새 메모리 tags: ['무릎 통증', '회복', '산책']
  새 메모리 links: #1
  
  TURN 3 요약
  - 생성: 메모리 #3
  - 연결: #1
  - 진화: 없음
  
  ===== TURN 4 입력 =====
  사용자: 병원에서 2주 동안 뛰지 말고 스트레칭만 하라고 했어. 가을 대회 참가는 아직 모르겠어.
  어시스턴트: 잘 다녀왔네. 대회는 무릎 상태를 보고 천천히 결정해도 되겠다.
  
  [LLM 중간 응답 #8 | gemma4:e2b]
  KEYWORDS: 스트레칭, 대회 참가, 무릎 상태, 휴식
  CONTEXT: A user was advised by a hospital to only stretch for two weeks instead of running, and the assistant suggested deciding on the competition based on knee condition.
  TAGS: 건강 정보, 운동 조언, 의학 관련, 대화
  
  [LLM 중간 응답 #9 | gemma4:e2b]
  DECISION: STRENGTHEN_AND_UPDATE
  REASON: The new memory strongly relates to the existing memories about knee issues, rest, and running advice. It should be linked to Memory Index 2 (knee pain/rest) and Memory Index 0 (running/park activity). Tags should reflect the specific context of medical advice and recovery.
  
  [LLM 중간 응답 #10 | gemma4:e2b]
  CONNECTIONS: 1, 2, 0
  TAGS: 스트레칭, 대회 참가, 무릎 상태, 휴식, 운동, 건강
  
  [LLM 중간 응답 #11 | gemma4:e2b]
  NEIGHBOR 0:
  CONTEXT: A user enjoys running 10km every weekend at Yeouido Hangang Park, but is currently advised by a hospital to only stretch for two weeks instead of running.
  TAGS: 운동, 야외활동, 건강, 스트레칭
  
  NEIGHBOR 1:
  CONTEXT: A user is trying to improve their performance for an upcoming 10km race, but has been advised by a hospital to only stretch for two weeks instead of running due to knee concerns.
  TAGS: 대회, 기록, 목표, 운동, 건강, 무릎 통증
  
  NEIGHBOR 2:
  CONTEXT: A user is resting from running due to knee pain and is now walking slowly at Jamsil Hangang Park for recovery, following medical advice to limit activity.
  TAGS: 무릎 통증, 회복, 산책, 운동
  
  ----- TURN 4 메모리 반영 결과 -----
  [새 메모리 #4]
  새 메모리 keywords: ['스트레칭', '대회 참가', '무릎 상태', '휴식']
  새 메모리 context: A user was advised by a hospital to only stretch for two weeks instead of running, and the assistant suggested deciding on the competition based on knee condition.
  새 메모리 tags: ['스트레칭', '대회 참가', '무릎 상태', '휴식', '운동', '건강']
  새 메모리 links: #2, #3, #1
  
  [진화한 기존 메모리 #1]
  원문: 사용자: 나는 주말마다 여의도 한강공원에서 10km 달리기를 좋아해.
  어시스턴트: 꾸준히 달리네. 한강 보면서 뛰면 기분도 좋겠다.
  context 변경 전: A user enjoys running 10km every weekend at Yeouido Hangang Park.
  context 변경 후: A user is resting from running due to knee pain and is now walking slowly at Jamsil Hangang Park for recovery, following medical advice to limit activity.
  tags 변경 전: ['운동', '야외활동', '건강']
  tags 변경 후: ['무릎 통증', '회복', '산책', '운동']
  
  [진화한 기존 메모리 #2]
  원문: 사용자: 가을에 열리는 10km 대회에 참가하려고 요즘 기록을 줄이는 중이야.
  어시스턴트: 목표가 생겼구나. 요즘은 어느 정도 기록이 나와?
  context 변경 전: A user is trying to improve their performance for an upcoming 10km race by currently working on reducing their time.
  context 변경 후: A user enjoys running 10km every weekend at Yeouido Hangang Park, but is currently advised by a hospital to only stretch for two weeks instead of running.
  tags 변경 전: ['운동', '건강']
  tags 변경 후: ['운동', '야외활동', '건강', '스트레칭']
  
  [진화한 기존 메모리 #3]
  원문: 사용자: 그런데 무릎이 아파서 요즘은 달리기를 쉬고 잠실 한강공원에서 천천히 걸어.
  어시스턴트: 아이고, 기록보다 회복이 먼저겠다. 병원에는 가 봤어?
  context 변경 전: A user is resting from running due to knee pain and is now walking slowly at Jamsil Hangang Park for recovery.
  context 변경 후: A user is trying to improve their performance for an upcoming 10km race, but has been advised by a hospital to only stretch for two weeks instead of running due to knee concerns.
  tags 변경 전: ['무릎 통증', '회복', '산책']
  tags 변경 후: ['대회', '기록', '목표', '운동', '건강', '무릎 통증']
  
  TURN 4 요약
  - 생성: 메모리 #4
  - 연결: #2, #3, #1
  - 진화: #1, #2, #3
  
  ===== TURN 5 입력 =====
  사용자: 무릎이 조금 좋아져서 오늘 잠실에서 3km만 천천히 뛰었어. 당분간 10km는 무리하지 않을 거야.
  어시스턴트: 좋아지고 있다니 다행이다. 당분간은 지금처럼 짧게 뛰는 게 좋겠네.
  
  [LLM 중간 응답 #12 | gemma4:e2b]
  KEYWORDS: 무릎, 달리기, 운동, 조절
  CONTEXT: The user ran a short distance due to improved knee condition and plans to limit the distance for the time being.
  TAGS: 건강, 운동, 피드백
  
  [LLM 중간 응답 #13 | gemma4:e2b]
  DECISION: STRENGTHEN_AND_UPDATE
  REASON: The new memory directly relates to the context of the nearest neighbors regarding knee issues and running adjustments. It strengthens the theme of managing running activity based on physical condition.
  
  [LLM 중간 응답 #14 | gemma4:e2b]
  CONNECTIONS: 0, 2, 3
  TAGS: 무릎, 달리기, 운동, 회복
  
  [LLM 중간 응답 #15 | gemma4:e2b]
  NEIGHBOR 0:
  CONTEXT: A user is resting from running due to knee pain and is now walking slowly at Jamsil Hangang Park for recovery, following medical advice to limit activity, and is currently running a short distance due to improved knee condition.
  TAGS: '무릎 통증', '회복', '산책', '운동'
  
  NEIGHBOR 1:
  CONTEXT: A user enjoys running 10km every weekend at Yeouido Hangang Park, but has been advised by a hospital to only stretch for two weeks instead of running, and is currently running a short distance due to improved knee condition.
  TAGS: '운동', '야외활동', '건강', '스트레칭'
  
  NEIGHBOR 2:
  CONTEXT: A user is trying to improve their performance for an upcoming 10km race, has been advised by a hospital to only stretch for two weeks instead of running, and is currently running a short distance due to improved knee condition.
  TAGS: '대회', '기록', '목표', '운동', '건강', '무릎 통증'
  
  NEIGHBOR 3:
  CONTEXT: A user was advised by a hospital to only stretch for two weeks instead of running, and is currently running a short distance due to improved knee condition.
  TAGS: '스트레칭', '대회 참가', '무릎 상태', '휴식', '운동', '건강'
  
  ----- TURN 5 메모리 반영 결과 -----
  [새 메모리 #5]
  새 메모리 keywords: ['무릎', '달리기', '운동', '조절']
  새 메모리 context: The user ran a short distance due to improved knee condition and plans to limit the distance for the time being.
  새 메모리 tags: ['무릎', '달리기', '운동', '회복']
  새 메모리 links: #1, #3, #4
  
  [진화한 기존 메모리 #1]
  원문: 사용자: 나는 주말마다 여의도 한강공원에서 10km 달리기를 좋아해.
  어시스턴트: 꾸준히 달리네. 한강 보면서 뛰면 기분도 좋겠다.
  context 변경 전: A user is resting from running due to knee pain and is now walking slowly at Jamsil Hangang Park for recovery, following medical advice to limit activity.
  context 변경 후: A user was advised by a hospital to only stretch for two weeks instead of running, and is currently running a short distance due to improved knee condition.
  tags 변경 전: ['무릎 통증', '회복', '산책', '운동']
  tags 변경 후: ['스트레칭', '대회 참가', '무릎 상태', '휴식', '운동', '건강']
  
  [진화한 기존 메모리 #2]
  원문: 사용자: 가을에 열리는 10km 대회에 참가하려고 요즘 기록을 줄이는 중이야.
  어시스턴트: 목표가 생겼구나. 요즘은 어느 정도 기록이 나와?
  context 변경 전: A user enjoys running 10km every weekend at Yeouido Hangang Park, but is currently advised by a hospital to only stretch for two weeks instead of running.
  context 변경 후: A user is trying to improve their performance for an upcoming 10km race, has been advised by a hospital to only stretch for two weeks instead of running, and is currently running a short distance due to improved knee condition.
  tags 변경 전: ['운동', '야외활동', '건강', '스트레칭']
  tags 변경 후: ['대회', '기록', '목표', '운동', '건강', '무릎 통증']
  
  [진화한 기존 메모리 #3]
  원문: 사용자: 그런데 무릎이 아파서 요즘은 달리기를 쉬고 잠실 한강공원에서 천천히 걸어.
  어시스턴트: 아이고, 기록보다 회복이 먼저겠다. 병원에는 가 봤어?
  context 변경 전: A user is trying to improve their performance for an upcoming 10km race, but has been advised by a hospital to only stretch for two weeks instead of running due to knee concerns.
  context 변경 후: A user is resting from running due to knee pain and is now walking slowly at Jamsil Hangang Park for recovery, following medical advice to limit activity, and is currently running a short distance due to improved knee condition.
  tags 변경 전: ['대회', '기록', '목표', '운동', '건강', '무릎 통증']
  tags 변경 후: ['무릎 통증', '회복', '산책', '운동']
  
  [진화한 기존 메모리 #4]
  원문: 사용자: 병원에서 2주 동안 뛰지 말고 스트레칭만 하라고 했어. 가을 대회 참가는 아직 모르겠어.
  어시스턴트: 잘 다녀왔네. 대회는 무릎 상태를 보고 천천히 결정해도 되겠다.
  context 변경 전: A user was advised by a hospital to only stretch for two weeks instead of running, and the assistant suggested deciding on the competition based on knee condition.
  context 변경 후: A user enjoys running 10km every weekend at Yeouido Hangang Park, but has been advised by a hospital to only stretch for two weeks instead of running, and is currently running a short distance due to improved knee condition.
  tags 변경 전: ['스트레칭', '대회 참가', '무릎 상태', '휴식', '운동', '건강']
  tags 변경 후: ['운동', '야외활동', '건강', '스트레칭']
  
  TURN 5 요약
  - 생성: 메모리 #5
  - 연결: #1, #3, #4
  - 진화: #1, #2, #3, #4

  • 턴이 길어질수록 context가 풍부해짐(길어짐) → context 잘 요약해서 출력하는 것이 중요할 듯.
  • keywords와 tags의 차이점을 잘 모르겠다.
  • 이웃 연결의 최대 길이를 제한할 필요가 있을 것 같다.

5. 결론

A-MEM의 핵심은 메모리를 정적으로 쌓는 것이 아니라, 새 메모리가 들어올 때 관련 메모리를 연결하고 기존 메모리를 다시 해석하는 데 있다. 실제 코드는 구조화 노트, embedding 후보 검색, LLM 기반 메모리 연결, 진화, 연결 이웃 확장으로 이를 구현한다.

실제 구현 코드에서는 턴이 길어질수록 context가 길어진다는 문제, 메모리 간 중복 연결 문제 방지가 필요해 보인다.