Knowledge Graph 메모리 시스템 완벽 가이드

AI 에이전트에 장기 기억 능력을 부여하는 Knowledge Graph 메모리 시스템을 구축합니다. Working, Short-term, Long-term 메모리 아키텍처부터 Neo4j를 활용한 Temporal Knowledge Graph 구현까지 실습합니다.

1. 메모리 아키텍처 설계

김개발 씨는 요즘 AI 챗봇 프로젝트를 진행하고 있습니다. 그런데 이상한 점을 발견했습니다.

분명 5분 전에 "저는 김철수입니다"라고 말했는데, 챗봇이 "이름이 뭐예요?"라고 다시 묻는 것입니다. 마치 금붕어처럼 기억력이 없는 AI를 보며 김개발 씨는 고민에 빠졌습니다.

메모리 아키텍처는 AI 에이전트가 정보를 기억하고 활용하는 방식을 체계화한 설계입니다. 마치 우리 뇌가 작업 기억, 단기 기억, 장기 기억으로 나뉘듯이 AI도 비슷한 구조가 필요합니다.

이를 통해 AI는 맥락을 이해하고, 과거 대화를 기억하며, 사용자에 대한 지식을 축적할 수 있습니다.

다음 코드를 살펴봅시다.

from dataclasses import dataclass, field
from typing import Dict, List, Optional
from datetime import datetime, timedelta

@dataclass
class MemoryArchitecture:
    # Working Memory: 현재 대화의 즉각적인 컨텍스트
    working_memory: Dict[str, any] = field(default_factory=dict)
    working_memory_ttl: int = 300  # 5분

    # Short-term Memory: 세션 동안 유지되는 정보
    short_term_memory: List[Dict] = field(default_factory=list)
    short_term_capacity: int = 100  # 최대 100개 항목

    # Long-term Memory: 영구적으로 저장되는 지식
    long_term_memory_enabled: bool = True

    def process_input(self, user_input: str, context: dict):
        # 1단계: Working Memory에 즉시 저장
        self.working_memory["current_input"] = user_input
        self.working_memory["timestamp"] = datetime.now()

        # 2단계: 중요도 평가 후 Short-term으로 승격
        importance = self._evaluate_importance(user_input)
        if importance > 0.5:
            self._promote_to_short_term(user_input, context)

        # 3단계: 패턴 발견 시 Long-term으로 통합
        if self._detect_pattern():
            self._consolidate_to_long_term()

김개발 씨는 입사 6개월 차 AI 엔지니어입니다. 회사에서 고객 상담 챗봇을 개발하고 있는데, 사용자들의 불만이 끊이질 않습니다.

"방금 말했잖아요!", "왜 자꾸 똑같은 걸 물어봐요?" 이런 피드백이 매일 쏟아집니다. 선배 개발자 박시니어 씨가 김개발 씨의 코드를 살펴보더니 고개를 저었습니다.

"이건 기억 시스템이 없어서 그래요. 매번 대화가 리셋되니까 금붕어처럼 보이는 거죠." 그렇다면 메모리 아키텍처란 정확히 무엇일까요?

쉽게 비유하자면, 메모리 아키텍처는 마치 우리가 공부할 때 사용하는 책상, 서랍, 책장과 같습니다. 책상 위에는 지금 당장 필요한 자료만 올려놓습니다.

서랍에는 오늘 수업에서 쓸 노트와 필기구를 넣어둡니다. 책장에는 언제든 참고할 수 있는 교과서와 참고서를 보관합니다.

AI 메모리도 이와 똑같이 동작합니다. Working Memory는 책상 위의 자료입니다.

지금 이 순간 처리해야 할 정보만 담고 있습니다. 사용자가 "오늘 날씨 어때?"라고 물으면, 이 질문 자체가 Working Memory에 올라갑니다.

처리가 끝나면 곧바로 사라지거나 다음 정보로 교체됩니다. 보통 몇 초에서 몇 분 정도만 유지됩니다.

Short-term Memory는 서랍 속 노트입니다. 현재 대화 세션 동안 기억해야 할 정보들이 여기에 저장됩니다.

"저는 김철수입니다"라고 사용자가 말했다면, 이 정보는 Short-term Memory로 이동합니다. 세션이 끝나기 전까지는 "김철수 님, 무엇을 도와드릴까요?"라고 대답할 수 있게 됩니다.

Long-term Memory는 책장의 참고서입니다. 사용자가 매번 "저는 채식주의자입니다"라고 말한다면, 이 정보는 영구적으로 저장할 가치가 있습니다.

Long-term Memory에 저장된 정보는 세션이 바뀌어도, 심지어 몇 달이 지나도 기억됩니다. 위의 코드를 자세히 살펴보겠습니다.

MemoryArchitecture 클래스는 세 가지 메모리 계층을 모두 담고 있습니다. working_memory는 딕셔너리로 구현되어 빠른 접근이 가능합니다.

short_term_memory는 리스트로 구현되어 순서가 보존됩니다. long_term_memory_enabled 플래그는 영구 저장소 연결 여부를 나타냅니다.

process_input 메서드가 핵심입니다. 사용자 입력이 들어오면 먼저 Working Memory에 저장합니다.

그다음 중요도를 평가하여 0.5 이상이면 Short-term으로 승격시킵니다. 마지막으로 패턴이 발견되면 Long-term으로 통합합니다.

실제 현업에서는 이 아키텍처가 어떻게 활용될까요? 예를 들어 금융 상담 챗봇을 생각해봅시다.

고객이 "제 계좌 잔액 알려줘"라고 말하면 Working Memory가 즉시 처리합니다. "지난달에 해외여행 갔다 왔어요"라는 정보는 Short-term에 저장되어 추후 여행자 보험 상품을 추천할 때 활용됩니다.

"저는 위험을 싫어해요"라는 투자 성향은 Long-term에 저장되어 언제든 적합한 상품을 추천하는 데 사용됩니다. 주의할 점도 있습니다.

모든 정보를 Long-term에 저장하려는 유혹을 조심해야 합니다. 저장 공간은 유한하고, 검색 속도도 느려집니다.

정말 중요한 정보만 선별하여 저장하는 것이 핵심입니다. 다시 김개발 씨의 이야기로 돌아가 봅시다.

박시니어 씨의 설명을 들은 김개발 씨는 눈이 반짝였습니다. "그래서 사람처럼 기억하는 AI를 만들려면 이 세 가지 계층이 필요한 거군요!"

실전 팁

💡 - Working Memory는 가볍게, Short-term은 적당히, Long-term은 신중하게 설계하세요

각 계층 간 승격 기준(중요도 임계값)을 비즈니스 요구사항에 맞게 조정하세요

2. Neo4j 설치 및 기본 설정

메모리 아키텍처를 설계한 김개발 씨는 이제 Long-term Memory를 어디에 저장할지 고민에 빠졌습니다. 관계형 데이터베이스?

NoSQL? 그때 박시니어 씨가 말했습니다.

"지식 간의 관계가 중요하다면 그래프 데이터베이스가 답이야."

Neo4j는 노드와 관계로 데이터를 저장하는 그래프 데이터베이스입니다. 마치 마인드맵처럼 개념들이 서로 연결되어 있어서, "김철수의 친구의 취미"처럼 복잡한 관계도 한 번의 쿼리로 찾을 수 있습니다.

Knowledge Graph를 구현하기에 최적의 선택입니다.

다음 코드를 살펴봅시다.

# Docker로 Neo4j 설치 및 실행
# docker run -d --name neo4j \
#   -p 7474:7474 -p 7687:7687 \
#   -e NEO4J_AUTH=neo4j/password123 \
#   neo4j:latest

from neo4j import GraphDatabase

class Neo4jConnection:
    def __init__(self, uri: str, user: str, password: str):
        self._driver = GraphDatabase.driver(uri, auth=(user, password))
        self._verify_connection()

    def _verify_connection(self):
        # 연결 상태 확인
        with self._driver.session() as session:
            result = session.run("RETURN 1 AS test")
            print(f"Neo4j 연결 성공: {result.single()['test']}")

    def close(self):
        self._driver.close()

    def execute_query(self, query: str, parameters: dict = None):
        with self._driver.session() as session:
            result = session.run(query, parameters or {})
            return [record.data() for record in result]

# 사용 예시
neo4j_conn = Neo4jConnection(
    uri="bolt://localhost:7687",
    user="neo4j",
    password="password123"
)

김개발 씨는 Long-term Memory 저장소를 고민하다가 MySQL을 먼저 떠올렸습니다. 익숙하니까요.

하지만 곧 문제가 생겼습니다. "김철수가 좋아하는 음식을 추천한 식당의 주인이 아는 요리사"를 찾으려면 JOIN을 몇 번이나 해야 할까요?

박시니어 씨가 화이트보드에 그림을 그리기 시작했습니다. "관계형 DB는 표 형태야.

하지만 지식은 그물망처럼 연결되어 있지. 그래서 그래프 데이터베이스가 필요해." 그래프 데이터베이스를 이해하려면 먼저 노드와 관계를 알아야 합니다.

노드는 마치 포스트잇과 같습니다. "김철수", "피자", "서울"처럼 하나의 개체를 나타냅니다.

각 포스트잇에는 이름, 나이, 가격 같은 속성을 적을 수 있습니다. 관계는 포스트잇을 연결하는 실과 같습니다.

"김철수 --좋아한다--> 피자", "김철수 --거주한다--> 서울"처럼 노드 사이의 연결을 나타냅니다. 관계에도 "2023년부터"처럼 속성을 붙일 수 있습니다.

Neo4j를 설치하는 가장 간단한 방법은 Docker를 사용하는 것입니다. 위 코드의 주석 부분을 터미널에 입력하면 Neo4j가 실행됩니다.

7474 포트는 웹 브라우저 인터페이스용이고, 7687 포트는 애플리케이션 연결용입니다. Neo4jConnection 클래스는 Python에서 Neo4j에 접속하기 위한 래퍼입니다.

__init__ 메서드에서 드라이버를 초기화하고 연결을 검증합니다. _verify_connection 메서드는 간단한 쿼리를 실행하여 연결 상태를 확인합니다.

execute_query 메서드가 실제로 쿼리를 실행하는 부분입니다. Neo4j는 Cypher라는 쿼리 언어를 사용합니다.

SQL과 비슷하지만 그래프에 특화되어 있습니다. 결과는 리스트 형태로 반환됩니다.

실무에서는 연결 풀링을 고려해야 합니다. 매번 새 연결을 만들면 성능이 떨어지기 때문입니다.

Neo4j 드라이버는 기본적으로 연결 풀을 관리하므로, 하나의 드라이버 인스턴스를 재사용하는 것이 좋습니다. 웹 브라우저에서 http://localhost:7474에 접속하면 Neo4j Browser를 사용할 수 있습니다.

여기서 시각적으로 그래프를 탐색하고 Cypher 쿼리를 테스트할 수 있습니다. 개발 초기에는 이 도구가 매우 유용합니다.

주의할 점이 있습니다. 프로덕션 환경에서는 반드시 비밀번호를 환경 변수로 관리해야 합니다.

코드에 비밀번호를 하드코딩하면 보안 사고로 이어질 수 있습니다. 김개발 씨는 Neo4j Browser에서 그래프가 시각화되는 것을 보고 감탄했습니다.

"와, 정말 마인드맵처럼 생겼네요! 이제 복잡한 관계도 쉽게 찾을 수 있겠어요."

실전 팁

💡 - 개발 환경에서는 Docker로 Neo4j를 실행하고, 프로덕션에서는 Neo4j Aura 클라우드 서비스를 고려하세요

Neo4j Browser의 :help 명령어로 Cypher 기본 문법을 익히세요

3. 엔터티 및 관계 스키마 설계

Neo4j 설치를 마친 김개발 씨는 바로 데이터를 넣으려고 했습니다. 하지만 박시니어 씨가 말렸습니다.

"잠깐, 스키마부터 설계해야지. 나중에 구조 바꾸려면 정말 힘들어."

스키마 설계는 어떤 종류의 노드와 관계를 사용할지 미리 정의하는 과정입니다. 마치 건물을 짓기 전에 설계도를 그리는 것과 같습니다.

잘 설계된 스키마는 효율적인 쿼리와 확장성의 기반이 됩니다.

다음 코드를 살펴봅시다.

from enum import Enum
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime

# 엔터티 타입 정의
class EntityType(Enum):
    USER = "User"           # 사용자
    CONCEPT = "Concept"     # 개념/지식
    EVENT = "Event"         # 이벤트/사건
    PREFERENCE = "Preference"  # 선호도
    CONTEXT = "Context"     # 대화 맥락

# 관계 타입 정의
class RelationType(Enum):
    KNOWS = "KNOWS"         # 알고 있다
    LIKES = "LIKES"         # 좋아한다
    DISLIKES = "DISLIKES"   # 싫어한다
    RELATED_TO = "RELATED_TO"  # 관련 있다
    MENTIONED_IN = "MENTIONED_IN"  # 언급되었다
    DERIVED_FROM = "DERIVED_FROM"  # 파생되었다

@dataclass
class Entity:
    id: str
    type: EntityType
    name: str
    properties: dict
    created_at: datetime = None
    updated_at: datetime = None
    confidence: float = 1.0  # 정보의 신뢰도

@dataclass
class Relationship:
    source_id: str
    target_id: str
    type: RelationType
    properties: dict
    strength: float = 1.0   # 관계의 강도
    valid_from: datetime = None
    valid_until: datetime = None  # Temporal 속성

김개발 씨는 처음에 대충 "사람", "물건", "장소" 정도로 분류하면 되겠다고 생각했습니다. 하지만 막상 구현하려니 애매한 경우가 너무 많았습니다.

"피자를 좋아한다"의 "피자"는 물건일까요, 개념일까요? 박시니어 씨가 설명했습니다.

"스키마 설계는 도메인을 깊이 이해하는 과정이야. 우리 서비스에서 어떤 정보가 중요한지 먼저 정리해봐." 엔터티 타입부터 살펴보겠습니다.

User는 시스템과 상호작용하는 사용자입니다. 각 사용자마다 별도의 Knowledge Graph가 구축됩니다.

이름, 나이, 직업 같은 기본 정보가 속성으로 저장됩니다. Concept은 추상적인 개념이나 지식을 나타냅니다.

"머신러닝", "이탈리안 요리", "건강한 생활" 같은 것들입니다. 가장 많이 사용되는 타입이며, 서로 계층 구조를 가질 수 있습니다.

Event는 특정 시점에 일어난 사건입니다. "2024년 1월 15일 회의", "지난주 여행" 같은 것들입니다.

시간 정보가 핵심 속성입니다. Preference는 사용자의 선호도를 명시적으로 나타냅니다.

"매운 음식 싫어함", "아침형 인간" 같은 정보입니다. AI가 개인화된 응답을 생성하는 데 핵심적인 역할을 합니다.

Context는 대화의 맥락 정보입니다. 특정 대화에서 논의된 주제, 분위기 등을 저장합니다.

이제 관계 타입을 살펴보겠습니다. KNOWS는 사용자가 어떤 개념을 알고 있음을 나타냅니다.

학습 시스템에서 유용합니다. LIKES와 DISLIKES는 선호도를 나타내며, 추천 시스템의 기반이 됩니다.

RELATED_TO는 범용적인 연관 관계입니다. 두 개념이 어떤 식으로든 관련이 있을 때 사용합니다.

MENTIONED_IN은 특정 대화에서 언급되었음을 추적합니다. DERIVED_FROM은 어떤 지식이 다른 지식에서 파생되었음을 나타냅니다.

코드에서 주목할 부분은 confidence와 strength 속성입니다. 모든 정보가 똑같이 확실한 것은 아닙니다.

"저는 김철수입니다"는 confidence가 1.0이지만, "아마 내일 비가 올 것 같아요"에서 추출한 정보는 0.3 정도일 수 있습니다. valid_from과 valid_until은 Temporal 속성입니다.

"작년까지는 서울에 살았어요"라는 정보는 유효 기간이 있습니다. 이를 통해 시간에 따른 변화를 추적할 수 있습니다.

실무에서 스키마를 설계할 때는 확장성을 고려해야 합니다. 처음부터 모든 경우를 예측할 수는 없으므로, properties 딕셔너리를 통해 유연성을 확보합니다.

김개발 씨는 스키마를 문서화하면서 뿌듯함을 느꼈습니다. "이제 어떤 데이터든 체계적으로 저장할 수 있겠어요!"

실전 팁

💡 - 엔터티 타입은 5-10개 이내로 유지하세요. 너무 세분화하면 관리가 어려워집니다

관계의 방향성을 명확히 정의하세요. "A가 B를 좋아한다"와 "B가 A를 좋아한다"는 다릅니다

4. Temporal Knowledge Graph 구현

스키마 설계를 마친 김개발 씨가 테스트 데이터를 넣다가 문제를 발견했습니다. "사용자가 예전에는 서울에 살았는데 지금은 부산에 산다면, 둘 다 저장해야 하나요?" 박시니어 씨가 웃으며 대답했습니다.

"좋은 질문이야. 그래서 Temporal Knowledge Graph가 필요해."

Temporal Knowledge Graph는 지식의 시간적 변화를 추적하는 그래프입니다. 마치 역사책처럼 "언제부터 언제까지" 정보가 유효했는지를 기록합니다.

이를 통해 과거의 맥락을 이해하고, 현재 유효한 정보만 조회할 수 있습니다.

다음 코드를 살펴봅시다.

from datetime import datetime, timezone
from typing import Optional, List

class TemporalKnowledgeGraph:
    def __init__(self, neo4j_connection):
        self.conn = neo4j_connection
        self._setup_constraints()

    def _setup_constraints(self):
        # 유니크 제약조건 및 인덱스 설정
        queries = [
            "CREATE CONSTRAINT IF NOT EXISTS FOR (e:Entity) REQUIRE e.id IS UNIQUE",
            "CREATE INDEX IF NOT EXISTS FOR (e:Entity) ON (e.valid_until)",
        ]
        for query in queries:
            self.conn.execute_query(query)

    def add_knowledge(self, entity: dict, valid_from: datetime = None):
        valid_from = valid_from or datetime.now(timezone.utc)
        query = """
        MERGE (e:Entity {id: $id})
        ON CREATE SET
            e.type = $type,
            e.name = $name,
            e.properties = $properties,
            e.valid_from = $valid_from,
            e.valid_until = null,
            e.created_at = datetime()
        ON MATCH SET
            e.updated_at = datetime()
        RETURN e
        """
        return self.conn.execute_query(query, {**entity, "valid_from": valid_from.isoformat()})

    def update_knowledge(self, entity_id: str, new_properties: dict):
        # 기존 버전을 종료하고 새 버전 생성 (버전 관리)
        now = datetime.now(timezone.utc)

        # 1. 기존 버전의 valid_until 설정
        close_query = """
        MATCH (e:Entity {id: $id})
        WHERE e.valid_until IS NULL
        SET e.valid_until = $now
        RETURN e
        """
        old_version = self.conn.execute_query(close_query, {"id": entity_id, "now": now.isoformat()})

        # 2. 새 버전 생성
        if old_version:
            new_id = f"{entity_id}_v{int(now.timestamp())}"
            return self.add_knowledge({**new_properties, "id": new_id}, now)

시간 여행 영화를 본 적 있으신가요? 과거로 돌아가면 그때의 세상이 그대로 있습니다.

Temporal Knowledge Graph도 비슷합니다. "2023년 1월의 김철수"와 "2024년 1월의 김철수"가 다를 수 있고, 둘 다 기록됩니다.

왜 이런 기능이 필요할까요? 김개발 씨의 챗봇을 예로 들어봅시다.

사용자가 "작년에 추천해줬던 식당 이름이 뭐였지?"라고 물었습니다. 만약 과거 정보를 덮어썼다면 대답할 수 없습니다.

하지만 Temporal Graph라면 "2023년 3월에 추천드린 '맛있는 집'을 말씀하시나요?"라고 답할 수 있습니다. _setup_constraints 메서드는 데이터 무결성을 보장합니다.

엔터티 ID의 유니크 제약조건과 valid_until 인덱스를 설정합니다. 인덱스는 "현재 유효한 정보"를 빠르게 찾기 위해 필수입니다.

add_knowledge 메서드는 새로운 지식을 추가합니다. MERGE 구문을 사용하여 이미 존재하면 업데이트하고, 없으면 생성합니다.

valid_from은 이 정보가 언제부터 유효한지, valid_until이 null이면 현재도 유효함을 의미합니다. update_knowledge 메서드가 핵심입니다.

정보가 변경되면 기존 버전을 삭제하지 않습니다. 대신 valid_until을 현재 시간으로 설정하여 "종료"시키고, 새로운 버전을 만듭니다.

이것이 버전 관리의 핵심입니다. 새 버전의 ID는 원본 ID에 타임스탬프를 붙여 생성합니다.

예를 들어 user_123의 새 버전은 user_123_v1704067200 같은 형태가 됩니다. 이렇게 하면 모든 버전을 고유하게 식별할 수 있습니다.

실무에서 Temporal Graph를 사용할 때 주의할 점이 있습니다. 버전이 무한정 쌓이면 저장 공간과 쿼리 성능에 문제가 생깁니다.

따라서 오래된 버전은 주기적으로 아카이빙하거나 요약하는 정책이 필요합니다. 또한 "현재 유효한 정보"를 조회할 때는 항상 valid_until IS NULL 조건을 포함해야 합니다.

이 조건을 빠뜨리면 과거 정보까지 모두 반환되어 혼란스러운 결과가 나옵니다. 김개발 씨는 감탄했습니다.

"이제 AI가 '예전에 말씀하셨던'이라는 표현을 쓸 수 있겠네요! 진짜 기억하는 것처럼요."

실전 팁

💡 - 현재 유효한 정보 조회 시 WHERE valid_until IS NULL 조건을 습관화하세요

버전 폭발을 방지하기 위해 사소한 변경은 무시하거나 통합하는 로직을 추가하세요

5. 메모리 조회 및 업데이트 API

Temporal Knowledge Graph를 구현한 김개발 씨는 이제 실제로 사용할 API가 필요했습니다. "CRUD 정도면 되겠지?"라고 생각했지만, 박시니어 씨가 고개를 저었습니다.

"Knowledge Graph는 단순 CRUD가 아니야. 관계 탐색과 추론이 핵심이지."

메모리 API는 Knowledge Graph에 지식을 저장하고 조회하는 인터페이스입니다. 단순한 키-값 저장이 아니라, 관계를 따라 탐색하고 연관 정보를 함께 가져오는 것이 특징입니다.

마치 도서관 사서가 관련 책들을 함께 추천해주는 것과 같습니다.

다음 코드를 살펴봅시다.

from typing import List, Dict, Optional
from datetime import datetime, timezone

class MemoryAPI:
    def __init__(self, temporal_graph):
        self.graph = temporal_graph

    # 지식 저장
    def remember(self, user_id: str, subject: str, predicate: str,
                 obj: str, confidence: float = 1.0) -> str:
        query = """
        MERGE (u:User {id: $user_id})
        MERGE (s:Entity {name: $subject})
        MERGE (o:Entity {name: $object})
        MERGE (s)-[r:%s {
            confidence: $confidence,
            created_at: datetime(),
            valid_from: datetime(),
            user_id: $user_id
        }]->(o)
        RETURN id(r) as relation_id
        """ % predicate.upper()

        result = self.graph.conn.execute_query(query, {
            "user_id": user_id, "subject": subject,
            "object": obj, "confidence": confidence
        })
        return result[0]["relation_id"] if result else None

    # 관련 지식 조회 (2단계 깊이까지)
    def recall(self, user_id: str, query_entity: str, depth: int = 2) -> List[Dict]:
        query = """
        MATCH (u:User {id: $user_id})
        MATCH path = (start:Entity {name: $query})-[*1..%d]-(connected)
        WHERE ALL(r IN relationships(path) WHERE r.valid_until IS NULL)
        RETURN DISTINCT connected.name as entity,
               [r IN relationships(path) | type(r)] as relations,
               length(path) as distance
        ORDER BY distance
        LIMIT 20
        """ % depth

        return self.graph.conn.execute_query(query, {
            "user_id": user_id, "query": query_entity
        })

    # 특정 관계 검색
    def find_relation(self, user_id: str, subject: str, predicate: str) -> List[Dict]:
        query = """
        MATCH (s:Entity {name: $subject})-[r:%s]->(o:Entity)
        WHERE r.user_id = $user_id AND r.valid_until IS NULL
        RETURN o.name as object, r.confidence as confidence
        """ % predicate.upper()

        return self.graph.conn.execute_query(query, {
            "user_id": user_id, "subject": subject
        })

API 설계는 사용자 경험을 결정합니다. 아무리 좋은 데이터베이스라도 API가 불편하면 개발자들이 사용하지 않습니다.

김개발 씨는 처음에 create_node, create_edge, get_node 같은 저수준 API를 만들었습니다. 하지만 실제로 사용하려니 너무 번거로웠습니다.

"피자를 좋아한다"를 저장하려면 노드 2개를 만들고, 엣지를 연결하고, 속성을 설정해야 했습니다. 박시니어 씨가 조언했습니다.

"도메인 언어로 생각해봐. '기억한다', '떠올린다'처럼 자연스러운 동사를 써봐." remember 메서드는 이름 그대로 "기억하기"입니다.

주어-서술어-목적어 형태로 지식을 저장합니다. "김철수-좋아한다-피자"처럼요.

내부적으로는 노드와 관계를 자동으로 생성하거나 연결합니다. MERGE 구문이 핵심입니다.

이미 존재하는 노드면 재사용하고, 없으면 새로 만듭니다. 따라서 "김철수"를 여러 번 저장해도 노드는 하나만 생성됩니다.

recall 메서드는 "떠올리기"입니다. 하나의 개념에서 시작해서 연결된 모든 지식을 탐색합니다.

depth 파라미터로 탐색 깊이를 조절할 수 있습니다. 예를 들어 "피자"를 조회하면 "피자-재료-치즈", "피자-원산지-이탈리아", "이탈리아-수도-로마"처럼 연결된 정보들이 줄줄이 나옵니다.

depth=2면 2단계까지, depth=3이면 3단계까지 탐색합니다. 쿼리의 [*1..%d] 부분은 가변 길이 경로 탐색입니다.

1부터 depth까지의 모든 경로를 찾습니다. WHERE ALL(r IN relationships(path) WHERE r.valid_until IS NULL) 조건은 현재 유효한 관계만 포함하도록 필터링합니다.

find_relation 메서드는 특정 관계를 직접 검색합니다. "김철수가 좋아하는 것"을 찾고 싶을 때 사용합니다.

recall보다 정확하고 빠르지만, 탐색 범위가 좁습니다. 실무에서는 이 세 가지 API를 조합해서 사용합니다.

먼저 find_relation으로 직접적인 관계를 찾고, 없으면 recall로 넓게 탐색하는 식입니다. 주의할 점이 있습니다.

depth를 너무 크게 설정하면 성능이 급격히 나빠집니다. 그래프 탐색은 지수적으로 복잡도가 증가하기 때문입니다.

대부분의 경우 depth=2 또는 3이면 충분합니다. 김개발 씨는 API를 테스트하며 뿌듯해했습니다.

"이제 정말 AI가 기억하는 것 같아요!"

실전 팁

💡 - recall의 depth는 2-3으로 제한하고, 필요시 점진적으로 늘리세요

자주 사용하는 쿼리 패턴은 별도 메서드로 추출하여 성능 최적화하세요

6. 메모리 통합 및 아카이빙 로직

메모리 시스템이 잘 동작하자 새로운 문제가 생겼습니다. 몇 주 만에 노드가 수만 개로 늘어난 것입니다.

"피자를 좋아해요", "피자 먹고 싶어요", "피자가 최고야"가 모두 별개로 저장되어 있었습니다. 김개발 씨는 한숨을 쉬었습니다.

"이걸 어떻게 정리하죠?"

메모리 통합은 비슷한 지식들을 하나로 합치는 과정입니다. 아카이빙은 오래되거나 덜 중요한 정보를 별도 저장소로 옮기는 것입니다.

마치 책장을 정리하듯이, 비슷한 책은 묶고 안 읽는 책은 창고로 보내는 것과 같습니다.

다음 코드를 살펴봅시다.

from datetime import datetime, timedelta
from collections import defaultdict

class MemoryConsolidator:
    def __init__(self, memory_api, similarity_threshold: float = 0.8):
        self.api = memory_api
        self.threshold = similarity_threshold

    def consolidate_similar_memories(self, user_id: str):
        # 유사한 엔터티들을 찾아서 통합
        query = """
        MATCH (e1:Entity)-[r1]->(target)
        MATCH (e2:Entity)-[r2]->(target)
        WHERE e1.name <> e2.name
          AND r1.user_id = $user_id
          AND r2.user_id = $user_id
          AND type(r1) = type(r2)
        WITH e1, e2, collect(target) as shared_targets
        WHERE size(shared_targets) >= 2
        RETURN e1.name as entity1, e2.name as entity2,
               size(shared_targets) as overlap
        ORDER BY overlap DESC
        """
        candidates = self.api.graph.conn.execute_query(query, {"user_id": user_id})

        for candidate in candidates:
            if self._calculate_similarity(candidate["entity1"], candidate["entity2"]) > self.threshold:
                self._merge_entities(user_id, candidate["entity1"], candidate["entity2"])

    def archive_old_memories(self, user_id: str, days_threshold: int = 90):
        # 오래된 메모리를 아카이브로 이동
        cutoff_date = datetime.now() - timedelta(days=days_threshold)

        query = """
        MATCH (e:Entity)-[r]-()
        WHERE r.user_id = $user_id
          AND r.created_at < $cutoff
          AND r.access_count < 3
        SET r.archived = true, r.archived_at = datetime()
        RETURN count(r) as archived_count
        """
        result = self.api.graph.conn.execute_query(query, {
            "user_id": user_id,
            "cutoff": cutoff_date.isoformat()
        })
        return result[0]["archived_count"] if result else 0

    def _merge_entities(self, user_id: str, entity1: str, entity2: str):
        # entity2의 관계를 entity1로 이전 후 entity2 삭제
        query = """
        MATCH (e1:Entity {name: $entity1})
        MATCH (e2:Entity {name: $entity2})-[r]->(target)
        WHERE r.user_id = $user_id
        MERGE (e1)-[new_r:MERGED_FROM]->(target)
        SET new_r = properties(r), new_r.merged_from = $entity2
        DELETE r
        """
        self.api.graph.conn.execute_query(query, {
            "user_id": user_id, "entity1": entity1, "entity2": entity2
        })

인간의 뇌도 비슷한 과정을 거칩니다. 수면 중에 뇌는 하루 동안 쌓인 기억을 정리합니다.

비슷한 기억은 통합하고, 중요하지 않은 기억은 잊어버립니다. 이것을 기억 통합이라고 합니다.

AI 메모리 시스템도 마찬가지입니다. 정리하지 않으면 쓸모없는 정보가 쌓여서 검색 성능이 떨어지고, 정확한 정보를 찾기 어려워집니다.

consolidate_similar_memories 메서드는 유사한 엔터티를 찾습니다. "피자"와 "Pizza", "피자요리"처럼 비슷하지만 다르게 저장된 것들입니다.

유사도를 판단하는 기준은 공유 타겟입니다. 두 엔터티가 같은 대상과 같은 종류의 관계를 맺고 있다면 유사할 가능성이 높습니다.

예를 들어 "피자"와 "Pizza"가 모두 "치즈", "토마토", "이탈리아"와 연결되어 있다면 같은 개념일 확률이 높습니다. WHERE size(shared_targets) >= 2 조건은 최소 2개 이상의 공유 타겟이 있어야 후보로 고려한다는 의미입니다.

너무 느슨하면 관련 없는 엔터티가 합쳐질 수 있습니다. archive_old_memories 메서드는 오래된 메모리를 정리합니다.

두 가지 기준을 사용합니다. 첫째, 생성된 지 90일이 지났는지.

둘째, 접근 횟수가 3회 미만인지. 둘 다 해당하면 "잘 안 쓰는 오래된 정보"로 판단합니다.

아카이브된 메모리는 삭제되지 않습니다. archived = true 플래그만 설정됩니다.

나중에 필요하면 복원할 수 있습니다. 완전 삭제는 별도의 정책으로 훨씬 나중에 진행합니다.

_merge_entities 메서드는 실제 통합을 수행합니다. entity2의 모든 관계를 entity1으로 이전합니다.

MERGED_FROM 관계 타입과 merged_from 속성으로 통합 이력을 추적할 수 있습니다. 실무에서는 이 작업을 배치 작업으로 스케줄링합니다.

매일 새벽 3시처럼 사용자가 적은 시간에 실행합니다. 실시간으로 하면 서비스 성능에 영향을 줄 수 있기 때문입니다.

주의할 점이 있습니다. 자동 통합은 실수할 수 있습니다.

"Apple(회사)"와 "apple(과일)"을 합쳐버릴 수도 있습니다. 따라서 통합 전에 로그를 남기고, 문제가 발견되면 롤백할 수 있어야 합니다.

김개발 씨는 통합 로직을 실행하고 결과를 확인했습니다. 수만 개였던 노드가 수천 개로 줄었습니다.

"와, 훨씬 깔끔해졌어요!"

실전 팁

💡 - 통합 작업은 반드시 로그를 남기고 롤백 가능하게 설계하세요

similarity_threshold를 너무 낮게 설정하면 다른 개념이 합쳐질 위험이 있습니다

7. 대화형 에이전트에 메모리 시스템 통합

모든 구성 요소가 준비되었습니다. 이제 마지막 단계입니다.

김개발 씨는 설렘 반 걱정 반으로 물었습니다. "이 모든 걸 어떻게 챗봇에 연결하죠?" 박시니어 씨가 미소를 지었습니다.

"자, 이제 진짜 마법이 시작되는 거야."

메모리 통합 에이전트는 대화 중에 자동으로 정보를 추출하고 저장하며, 응답 생성 시 관련 기억을 활용합니다. 마치 오래된 친구처럼 과거 대화를 기억하고, 사용자의 취향을 파악하여 개인화된 응답을 제공합니다.

다음 코드를 살펴봅시다.

import re
from typing import List, Dict, Optional
from dataclasses import dataclass

@dataclass
class ConversationTurn:
    user_message: str
    assistant_response: str
    extracted_facts: List[Dict]
    retrieved_memories: List[Dict]

class MemoryEnabledAgent:
    def __init__(self, memory_api, llm_client):
        self.memory = memory_api
        self.llm = llm_client
        self.conversation_history: List[ConversationTurn] = []

    async def chat(self, user_id: str, message: str) -> str:
        # 1단계: 관련 기억 검색
        relevant_memories = self._retrieve_relevant_memories(user_id, message)

        # 2단계: 메모리를 포함한 프롬프트 구성
        prompt = self._build_prompt_with_memory(message, relevant_memories)

        # 3단계: LLM 응답 생성
        response = await self.llm.generate(prompt)

        # 4단계: 대화에서 새로운 사실 추출 및 저장
        extracted_facts = self._extract_and_store_facts(user_id, message, response)

        # 5단계: 대화 기록 저장
        self.conversation_history.append(ConversationTurn(
            user_message=message,
            assistant_response=response,
            extracted_facts=extracted_facts,
            retrieved_memories=relevant_memories
        ))

        return response

    def _retrieve_relevant_memories(self, user_id: str, message: str) -> List[Dict]:
        # 메시지에서 키워드 추출 후 관련 메모리 검색
        keywords = self._extract_keywords(message)
        memories = []
        for keyword in keywords[:3]:  # 상위 3개 키워드만
            memories.extend(self.memory.recall(user_id, keyword, depth=2))
        return memories[:10]  # 최대 10개 반환

    def _build_prompt_with_memory(self, message: str, memories: List[Dict]) -> str:
        memory_context = "\n".join([
            f"- {m['entity']}: {m['relations']}" for m in memories
        ]) if memories else "관련 기억 없음"

        return f"""당신은 사용자와의 과거 대화를 기억하는 AI 어시스턴트입니다.

관련 기억:
{memory_context}

사용자 메시지: {message}

위 기억을 참고하여 개인화된 응답을 생성하세요."""

    def _extract_and_store_facts(self, user_id: str, message: str, response: str) -> List[Dict]:
        # 간단한 패턴 매칭으로 사실 추출 (실제로는 NLP/LLM 사용)
        facts = []
        patterns = [
            (r"저는 (.+)입니다", "USER", "IS_A"),
            (r"(.+)을 좋아해요", "USER", "LIKES"),
            (r"(.+)에 살아요", "USER", "LIVES_IN"),
        ]

        for pattern, subject, predicate in patterns:
            match = re.search(pattern, message)
            if match:
                obj = match.group(1)
                self.memory.remember(user_id, subject, predicate, obj, confidence=0.9)
                facts.append({"subject": subject, "predicate": predicate, "object": obj})

        return facts

드디어 모든 퍼즐 조각이 맞춰지는 순간입니다. MemoryEnabledAgent 클래스는 메모리 시스템과 LLM을 연결하는 다리 역할을 합니다.

memory_api는 앞서 만든 Knowledge Graph API이고, llm_client는 GPT나 Claude 같은 언어 모델 클라이언트입니다. chat 메서드가 핵심입니다.

사용자 메시지가 들어오면 5단계 파이프라인을 거칩니다. 첫 번째 단계는 관련 기억 검색입니다.

메시지에서 키워드를 추출하고, 각 키워드와 연관된 메모리를 Knowledge Graph에서 가져옵니다. "오늘 점심 뭐 먹을까?"라는 메시지에서 "점심", "먹다" 같은 키워드를 추출하고, 사용자의 음식 취향을 검색합니다.

두 번째 단계는 프롬프트 구성입니다. 검색된 메모리를 LLM 프롬프트에 포함시킵니다.

"이 사용자는 피자를 좋아하고, 매운 음식을 싫어합니다"처럼요. LLM은 이 정보를 참고하여 응답을 생성합니다.

세 번째 단계는 응답 생성입니다. LLM이 메모리를 참고하여 개인화된 답변을 만듭니다.

"지난번에 좋아하신다고 하셨던 피자 어떠세요?"처럼 과거 대화를 반영할 수 있습니다. 네 번째 단계는 사실 추출입니다.

사용자 메시지에서 저장할 만한 정보를 찾아냅니다. "저는 개발자입니다"라고 말하면 (USER, IS_A, 개발자) 트리플이 추출되어 저장됩니다.

다섯 번째 단계는 대화 기록 저장입니다. 이번 대화에서 무슨 일이 있었는지 기록합니다.

나중에 분석하거나 디버깅할 때 유용합니다. _extract_and_store_facts 메서드는 간단한 정규표현식을 사용합니다.

실제 프로덕션에서는 더 정교한 NLP 기법이나 LLM 기반 추출을 사용합니다. 하지만 원리는 동일합니다.

confidence=0.9로 설정한 것에 주목하세요. 자동 추출이므로 100% 확신할 수 없습니다.

나중에 사용자가 "아니, 그건 틀렸어"라고 말하면 confidence를 낮추거나 삭제할 수 있습니다. 실무에서는 몇 가지 추가 고려사항이 있습니다.

메모리 검색 시간이 길어지면 응답 지연이 발생합니다. 따라서 검색을 비동기로 처리하거나, 캐싱을 적용해야 합니다.

또한 너무 많은 메모리를 프롬프트에 넣으면 LLM의 컨텍스트 제한을 초과할 수 있습니다. 김개발 씨는 완성된 에이전트를 테스트했습니다.

"안녕, 나 김철수야"라고 말한 후, 한참 뒤에 "내 이름 기억해?"라고 물었습니다. 에이전트가 대답했습니다.

"네, 김철수 님이시죠!" 김개발 씨의 눈이 반짝였습니다. 박시니어 씨가 어깨를 두드렸습니다.

"축하해. 이제 네 AI는 진짜 기억할 수 있어."

실전 팁

💡 - 메모리 검색과 LLM 호출을 비동기로 병렬 처리하면 응답 속도가 빨라집니다

프롬프트에 넣는 메모리 개수를 제한하여 컨텍스트 오버플로우를 방지하세요

이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!

#Python#KnowledgeGraph#Neo4j#MemorySystem#AIAgent#AI Engineering

Knowledge Graph 메모리 시스템 완벽 가이드

# Knowledge Graph 메모리 시스템 완벽 가이드 AI 에이전트에 장기 기억 능력을 부여하는 Knowledge Graph 메모리 시스템을 구축합니다. Working, Short-term, Long-term 메모리 아키텍처부터 Neo4j를 활용한 Temporal Knowledge Graph 구현까지 실습합니다. --- ## 목차 1. [메모리_아키텍처_설계](#메모리-아키텍처-설계) 2. [Neo4j_설치_및_기본_설정](#neo4j-설치-및-기본-설정) 3. [엔터티_및_관계_스키마_설계](#엔터티-및-관계-스키마-설계) 4. [Temporal_Knowledge_Graph_구현](#temporal-knowledge-graph-구현) 5. [메모리_조회_및_업데이트_API](#메모리-조회-및-업데이트-api) 6. [메모리_통합_및_아카이빙_로직](#메모리-통합-및-아카이빙-로직) 7. [대화형_에이전트에_메모리_시스템_통합](#대화형-에이전트에-메모리-시스템-통합) --- ## 1. 메모리_아키텍처_설계 김개발 씨는 요즘 AI 챗봇 프로젝트를 진행하고 있습니다. 그런데 이상한 점을 발견했습니다. 분명 5분 전에 "저는 김철수입니다"라고 말했는데, 챗봇이 "이름이 뭐예요?"라고 다시 묻는 것입니다. 마치 금붕어처럼 기억력이 없는 AI를 보며 김개발 씨는 고민에 빠졌습니다. **메모리 아키텍처**는 AI 에이전트가 정보를 기억하고 활용하는 방식을 체계화한 설계입니다. 마치 우리 뇌가 작업 기억, 단기 기억, 장기 기억으로 나뉘듯이 AI도 비슷한 구조가 필요합니다. 이를 통해 AI는 맥락을 이해하고, 과거 대화를 기억하며, 사용자에 대한 지식을 축적할 수 있습니다. 다음 코드를 살펴봅시다. ```python from dataclasses import dataclass, field from typing import Dict, List, Optional from datetime import datetime, timedelta @dataclass class MemoryArchitecture: # Working Memory: 현재 대화의 즉각적인 컨텍스트 working_memory: Dict[str, any] = field(default_factory=dict) working_memory_ttl: int = 300 # 5분 # Short-term Memory: 세션 동안 유지되는 정보 short_term_memory: List[Dict] = field(default_factory=list) short_term_capacity: int = 100 # 최대 100개 항목 # Long-term Memory: 영구적으로 저장되는 지식 long_term_memory_enabled: bool = True def process_input(self, user_input: str, context: dict): # 1단계: Working Memory에 즉시 저장 self.working_memory["current_input"] = user_input self.working_memory["timestamp"] = datetime.now() # 2단계: 중요도 평가 후 Short-term으로 승격 importance = self._evaluate_importance(user_input) if importance > 0.5: self._promote_to_short_term(user_input, context) # 3단계: 패턴 발견 시 Long-term으로 통합 if self._detect_pattern(): self._consolidate_to_long_term() ``` 김개발 씨는 입사 6개월 차 AI 엔지니어입니다. 회사에서 고객 상담 챗봇을 개발하고 있는데, 사용자들의 불만이 끊이질 않습니다. "방금 말했잖아요!", "왜 자꾸 똑같은 걸 물어봐요?" 이런 피드백이 매일 쏟아집니다. 선배 개발자 박시니어 씨가 김개발 씨의 코드를 살펴보더니 고개를 저었습니다. "이건 기억 시스템이 없어서 그래요. 매번 대화가 리셋되니까 금붕어처럼 보이는 거죠." 그렇다면 **메모리 아키텍처**란 정확히 무엇일까요? 쉽게 비유하자면, 메모리 아키텍처는 마치 우리가 공부할 때 사용하는 **책상, 서랍, 책장**과 같습니다. 책상 위에는 지금 당장 필요한 자료만 올려놓습니다. 서랍에는 오늘 수업에서 쓸 노트와 필기구를 넣어둡니다. 책장에는 언제든 참고할 수 있는 교과서와 참고서를 보관합니다. AI 메모리도 이와 똑같이 동작합니다. **Working Memory**는 책상 위의 자료입니다. 지금 이 순간 처리해야 할 정보만 담고 있습니다. 사용자가 "오늘 날씨 어때?"라고 물으면, 이 질문 자체가 Working Memory에 올라갑니다. 처리가 끝나면 곧바로 사라지거나 다음 정보로 교체됩니다. 보통 몇 초에서 몇 분 정도만 유지됩니다. **Short-term Memory**는 서랍 속 노트입니다. 현재 대화 세션 동안 기억해야 할 정보들이 여기에 저장됩니다. "저는 김철수입니다"라고 사용자가 말했다면, 이 정보는 Short-term Memory로 이동합니다. 세션이 끝나기 전까지는 "김철수 님, 무엇을 도와드릴까요?"라고 대답할 수 있게 됩니다. **Long-term Memory**는 책장의 참고서입니다. 사용자가 매번 "저는 채식주의자입니다"라고 말한다면, 이 정보는 영구적으로 저장할 가치가 있습니다. Long-term Memory에 저장된 정보는 세션이 바뀌어도, 심지어 몇 달이 지나도 기억됩니다. 위의 코드를 자세히 살펴보겠습니다. `MemoryArchitecture` 클래스는 세 가지 메모리 계층을 모두 담고 있습니다. `working_memory`는 딕셔너리로 구현되어 빠른 접근이 가능합니다. `short_term_memory`는 리스트로 구현되어 순서가 보존됩니다. `long_term_memory_enabled` 플래그는 영구 저장소 연결 여부를 나타냅니다. `process_input` 메서드가 핵심입니다. 사용자 입력이 들어오면 먼저 Working Memory에 저장합니다. 그다음 중요도를 평가하여 0.5 이상이면 Short-term으로 승격시킵니다. 마지막으로 패턴이 발견되면 Long-term으로 통합합니다. 실제 현업에서는 이 아키텍처가 어떻게 활용될까요? 예를 들어 금융 상담 챗봇을 생각해봅시다. 고객이 "제 계좌 잔액 알려줘"라고 말하면 Working Memory가 즉시 처리합니다. "지난달에 해외여행 갔다 왔어요"라는 정보는 Short-term에 저장되어 추후 여행자 보험 상품을 추천할 때 활용됩니다. "저는 위험을 싫어해요"라는 투자 성향은 Long-term에 저장되어 언제든 적합한 상품을 추천하는 데 사용됩니다. 주의할 점도 있습니다. 모든 정보를 Long-term에 저장하려는 유혹을 조심해야 합니다. 저장 공간은 유한하고, 검색 속도도 느려집니다. 정말 중요한 정보만 선별하여 저장하는 것이 핵심입니다. 다시 김개발 씨의 이야기로 돌아가 봅시다. 박시니어 씨의 설명을 들은 김개발 씨는 눈이 반짝였습니다. "그래서 사람처럼 기억하는 AI를 만들려면 이 세 가지 계층이 필요한 거군요!" **실전 팁** 💡 - Working Memory는 가볍게, Short-term은 적당히, Long-term은 신중하게 설계하세요 - 각 계층 간 승격 기준(중요도 임계값)을 비즈니스 요구사항에 맞게 조정하세요 --- ## 2. Neo4j_설치_및_기본_설정 메모리 아키텍처를 설계한 김개발 씨는 이제 Long-term Memory를 어디에 저장할지 고민에 빠졌습니다. 관계형 데이터베이스? NoSQL? 그때 박시니어 씨가 말했습니다. "지식 간의 관계가 중요하다면 그래프 데이터베이스가 답이야." **Neo4j**는 노드와 관계로 데이터를 저장하는 그래프 데이터베이스입니다. 마치 마인드맵처럼 개념들이 서로 연결되어 있어서, "김철수의 친구의 취미"처럼 복잡한 관계도 한 번의 쿼리로 찾을 수 있습니다. Knowledge Graph를 구현하기에 최적의 선택입니다. 다음 코드를 살펴봅시다. ```python # Docker로 Neo4j 설치 및 실행 # docker run -d --name neo4j \ # -p 7474:7474 -p 7687:7687 \ # -e NEO4J_AUTH=neo4j/password123 \ # neo4j:latest from neo4j import GraphDatabase class Neo4jConnection: def __init__(self, uri: str, user: str, password: str): self._driver = GraphDatabase.driver(uri, auth=(user, password)) self._verify_connection() def _verify_connection(self): # 연결 상태 확인 with self._driver.session() as session: result = session.run("RETURN 1 AS test") print(f"Neo4j 연결 성공: {result.single()['test']}") def close(self): self._driver.close() def execute_query(self, query: str, parameters: dict = None): with self._driver.session() as session: result = session.run(query, parameters or {}) return [record.data() for record in result] # 사용 예시 neo4j_conn = Neo4jConnection( uri="bolt://localhost:7687", user="neo4j", password="password123" ) ``` 김개발 씨는 Long-term Memory 저장소를 고민하다가 MySQL을 먼저 떠올렸습니다. 익숙하니까요. 하지만 곧 문제가 생겼습니다. "김철수가 좋아하는 음식을 추천한 식당의 주인이 아는 요리사"를 찾으려면 JOIN을 몇 번이나 해야 할까요? 박시니어 씨가 화이트보드에 그림을 그리기 시작했습니다. "관계형 DB는 표 형태야. 하지만 지식은 그물망처럼 연결되어 있지. 그래서 **그래프 데이터베이스**가 필요해." 그래프 데이터베이스를 이해하려면 먼저 **노드**와 **관계**를 알아야 합니다. **노드**는 마치 포스트잇과 같습니다. "김철수", "피자", "서울"처럼 하나의 개체를 나타냅니다. 각 포스트잇에는 이름, 나이, 가격 같은 속성을 적을 수 있습니다. **관계**는 포스트잇을 연결하는 실과 같습니다. "김철수 --좋아한다--> 피자", "김철수 --거주한다--> 서울"처럼 노드 사이의 연결을 나타냅니다. 관계에도 "2023년부터"처럼 속성을 붙일 수 있습니다. Neo4j를 설치하는 가장 간단한 방법은 **Docker**를 사용하는 것입니다. 위 코드의 주석 부분을 터미널에 입력하면 Neo4j가 실행됩니다. 7474 포트는 웹 브라우저 인터페이스용이고, 7687 포트는 애플리케이션 연결용입니다. `Neo4jConnection` 클래스는 Python에서 Neo4j에 접속하기 위한 래퍼입니다. `__init__` 메서드에서 드라이버를 초기화하고 연결을 검증합니다. `_verify_connection` 메서드는 간단한 쿼리를 실행하여 연결 상태를 확인합니다. `execute_query` 메서드가 실제로 쿼리를 실행하는 부분입니다. Neo4j는 **Cypher**라는 쿼리 언어를 사용합니다. SQL과 비슷하지만 그래프에 특화되어 있습니다. 결과는 리스트 형태로 반환됩니다. 실무에서는 연결 풀링을 고려해야 합니다. 매번 새 연결을 만들면 성능이 떨어지기 때문입니다. Neo4j 드라이버는 기본적으로 연결 풀을 관리하므로, 하나의 드라이버 인스턴스를 재사용하는 것이 좋습니다. 웹 브라우저에서 `http://localhost:7474`에 접속하면 Neo4j Browser를 사용할 수 있습니다. 여기서 시각적으로 그래프를 탐색하고 Cypher 쿼리를 테스트할 수 있습니다. 개발 초기에는 이 도구가 매우 유용합니다. 주의할 점이 있습니다. 프로덕션 환경에서는 반드시 비밀번호를 환경 변수로 관리해야 합니다. 코드에 비밀번호를 하드코딩하면 보안 사고로 이어질 수 있습니다. 김개발 씨는 Neo4j Browser에서 그래프가 시각화되는 것을 보고 감탄했습니다. "와, 정말 마인드맵처럼 생겼네요! 이제 복잡한 관계도 쉽게 찾을 수 있겠어요." **실전 팁** 💡 - 개발 환경에서는 Docker로 Neo4j를 실행하고, 프로덕션에서는 Neo4j Aura 클라우드 서비스를 고려하세요 - Neo4j Browser의 `:help` 명령어로 Cypher 기본 문법을 익히세요 --- ## 3. 엔터티_및_관계_스키마_설계 Neo4j 설치를 마친 김개발 씨는 바로 데이터를 넣으려고 했습니다. 하지만 박시니어 씨가 말렸습니다. "잠깐, 스키마부터 설계해야지. 나중에 구조 바꾸려면 정말 힘들어." **스키마 설계**는 어떤 종류의 노드와 관계를 사용할지 미리 정의하는 과정입니다. 마치 건물을 짓기 전에 설계도를 그리는 것과 같습니다. 잘 설계된 스키마는 효율적인 쿼리와 확장성의 기반이 됩니다. 다음 코드를 살펴봅시다. ```python from enum import Enum from dataclasses import dataclass from typing import List, Optional from datetime import datetime # 엔터티 타입 정의 class EntityType(Enum): USER = "User" # 사용자 CONCEPT = "Concept" # 개념/지식 EVENT = "Event" # 이벤트/사건 PREFERENCE = "Preference" # 선호도 CONTEXT = "Context" # 대화 맥락 # 관계 타입 정의 class RelationType(Enum): KNOWS = "KNOWS" # 알고 있다 LIKES = "LIKES" # 좋아한다 DISLIKES = "DISLIKES" # 싫어한다 RELATED_TO = "RELATED_TO" # 관련 있다 MENTIONED_IN = "MENTIONED_IN" # 언급되었다 DERIVED_FROM = "DERIVED_FROM" # 파생되었다 @dataclass class Entity: id: str type: EntityType name: str properties: dict created_at: datetime = None updated_at: datetime = None confidence: float = 1.0 # 정보의 신뢰도 @dataclass class Relationship: source_id: str target_id: str type: RelationType properties: dict strength: float = 1.0 # 관계의 강도 valid_from: datetime = None valid_until: datetime = None # Temporal 속성 ``` 김개발 씨는 처음에 대충 "사람", "물건", "장소" 정도로 분류하면 되겠다고 생각했습니다. 하지만 막상 구현하려니 애매한 경우가 너무 많았습니다. "피자를 좋아한다"의 "피자"는 물건일까요, 개념일까요? 박시니어 씨가 설명했습니다. "스키마 설계는 도메인을 깊이 이해하는 과정이야. 우리 서비스에서 어떤 정보가 중요한지 먼저 정리해봐." **엔터티 타입**부터 살펴보겠습니다. **User**는 시스템과 상호작용하는 사용자입니다. 각 사용자마다 별도의 Knowledge Graph가 구축됩니다. 이름, 나이, 직업 같은 기본 정보가 속성으로 저장됩니다. **Concept**은 추상적인 개념이나 지식을 나타냅니다. "머신러닝", "이탈리안 요리", "건강한 생활" 같은 것들입니다. 가장 많이 사용되는 타입이며, 서로 계층 구조를 가질 수 있습니다. **Event**는 특정 시점에 일어난 사건입니다. "2024년 1월 15일 회의", "지난주 여행" 같은 것들입니다. 시간 정보가 핵심 속성입니다. **Preference**는 사용자의 선호도를 명시적으로 나타냅니다. "매운 음식 싫어함", "아침형 인간" 같은 정보입니다. AI가 개인화된 응답을 생성하는 데 핵심적인 역할을 합니다. **Context**는 대화의 맥락 정보입니다. 특정 대화에서 논의된 주제, 분위기 등을 저장합니다. 이제 **관계 타입**을 살펴보겠습니다. **KNOWS**는 사용자가 어떤 개념을 알고 있음을 나타냅니다. 학습 시스템에서 유용합니다. **LIKES**와 **DISLIKES**는 선호도를 나타내며, 추천 시스템의 기반이 됩니다. **RELATED_TO**는 범용적인 연관 관계입니다. 두 개념이 어떤 식으로든 관련이 있을 때 사용합니다. **MENTIONED_IN**은 특정 대화에서 언급되었음을 추적합니다. **DERIVED_FROM**은 어떤 지식이 다른 지식에서 파생되었음을 나타냅니다. 코드에서 주목할 부분은 **confidence**와 **strength** 속성입니다. 모든 정보가 똑같이 확실한 것은 아닙니다. "저는 김철수입니다"는 confidence가 1.0이지만, "아마 내일 비가 올 것 같아요"에서 추출한 정보는 0.3 정도일 수 있습니다. **valid_from**과 **valid_until**은 Temporal 속성입니다. "작년까지는 서울에 살았어요"라는 정보는 유효 기간이 있습니다. 이를 통해 시간에 따른 변화를 추적할 수 있습니다. 실무에서 스키마를 설계할 때는 확장성을 고려해야 합니다. 처음부터 모든 경우를 예측할 수는 없으므로, `properties` 딕셔너리를 통해 유연성을 확보합니다. 김개발 씨는 스키마를 문서화하면서 뿌듯함을 느꼈습니다. "이제 어떤 데이터든 체계적으로 저장할 수 있겠어요!" **실전 팁** 💡 - 엔터티 타입은 5-10개 이내로 유지하세요. 너무 세분화하면 관리가 어려워집니다 - 관계의 방향성을 명확히 정의하세요. "A가 B를 좋아한다"와 "B가 A를 좋아한다"는 다릅니다 --- ## 4. Temporal_Knowledge_Graph_구현 스키마 설계를 마친 김개발 씨가 테스트 데이터를 넣다가 문제를 발견했습니다. "사용자가 예전에는 서울에 살았는데 지금은 부산에 산다면, 둘 다 저장해야 하나요?" 박시니어 씨가 웃으며 대답했습니다. "좋은 질문이야. 그래서 Temporal Knowledge Graph가 필요해." **Temporal Knowledge Graph**는 지식의 시간적 변화를 추적하는 그래프입니다. 마치 역사책처럼 "언제부터 언제까지" 정보가 유효했는지를 기록합니다. 이를 통해 과거의 맥락을 이해하고, 현재 유효한 정보만 조회할 수 있습니다. 다음 코드를 살펴봅시다. ```python from datetime import datetime, timezone from typing import Optional, List class TemporalKnowledgeGraph: def __init__(self, neo4j_connection): self.conn = neo4j_connection self._setup_constraints() def _setup_constraints(self): # 유니크 제약조건 및 인덱스 설정 queries = [ "CREATE CONSTRAINT IF NOT EXISTS FOR (e:Entity) REQUIRE e.id IS UNIQUE", "CREATE INDEX IF NOT EXISTS FOR (e:Entity) ON (e.valid_until)", ] for query in queries: self.conn.execute_query(query) def add_knowledge(self, entity: dict, valid_from: datetime = None): valid_from = valid_from or datetime.now(timezone.utc) query = """ MERGE (e:Entity {id: $id}) ON CREATE SET e.type = $type, e.name = $name, e.properties = $properties, e.valid_from = $valid_from, e.valid_until = null, e.created_at = datetime() ON MATCH SET e.updated_at = datetime() RETURN e """ return self.conn.execute_query(query, {**entity, "valid_from": valid_from.isoformat()}) def update_knowledge(self, entity_id: str, new_properties: dict): # 기존 버전을 종료하고 새 버전 생성 (버전 관리) now = datetime.now(timezone.utc) # 1. 기존 버전의 valid_until 설정 close_query = """ MATCH (e:Entity {id: $id}) WHERE e.valid_until IS NULL SET e.valid_until = $now RETURN e """ old_version = self.conn.execute_query(close_query, {"id": entity_id, "now": now.isoformat()}) # 2. 새 버전 생성 if old_version: new_id = f"{entity_id}_v{int(now.timestamp())}" return self.add_knowledge({**new_properties, "id": new_id}, now) ``` 시간 여행 영화를 본 적 있으신가요? 과거로 돌아가면 그때의 세상이 그대로 있습니다. **Temporal Knowledge Graph**도 비슷합니다. "2023년 1월의 김철수"와 "2024년 1월의 김철수"가 다를 수 있고, 둘 다 기록됩니다. 왜 이런 기능이 필요할까요? 김개발 씨의 챗봇을 예로 들어봅시다. 사용자가 "작년에 추천해줬던 식당 이름이 뭐였지?"라고 물었습니다. 만약 과거 정보를 덮어썼다면 대답할 수 없습니다. 하지만 Temporal Graph라면 "2023년 3월에 추천드린 '맛있는 집'을 말씀하시나요?"라고 답할 수 있습니다. `_setup_constraints` 메서드는 데이터 무결성을 보장합니다. 엔터티 ID의 유니크 제약조건과 `valid_until` 인덱스를 설정합니다. 인덱스는 "현재 유효한 정보"를 빠르게 찾기 위해 필수입니다. `add_knowledge` 메서드는 새로운 지식을 추가합니다. **MERGE** 구문을 사용하여 이미 존재하면 업데이트하고, 없으면 생성합니다. `valid_from`은 이 정보가 언제부터 유효한지, `valid_until`이 null이면 현재도 유효함을 의미합니다. `update_knowledge` 메서드가 핵심입니다. 정보가 변경되면 기존 버전을 삭제하지 않습니다. 대신 `valid_until`을 현재 시간으로 설정하여 "종료"시키고, 새로운 버전을 만듭니다. 이것이 **버전 관리**의 핵심입니다. 새 버전의 ID는 원본 ID에 타임스탬프를 붙여 생성합니다. 예를 들어 `user_123`의 새 버전은 `user_123_v1704067200` 같은 형태가 됩니다. 이렇게 하면 모든 버전을 고유하게 식별할 수 있습니다. 실무에서 Temporal Graph를 사용할 때 주의할 점이 있습니다. 버전이 무한정 쌓이면 저장 공간과 쿼리 성능에 문제가 생깁니다. 따라서 오래된 버전은 주기적으로 아카이빙하거나 요약하는 정책이 필요합니다. 또한 "현재 유효한 정보"를 조회할 때는 항상 `valid_until IS NULL` 조건을 포함해야 합니다. 이 조건을 빠뜨리면 과거 정보까지 모두 반환되어 혼란스러운 결과가 나옵니다. 김개발 씨는 감탄했습니다. "이제 AI가 '예전에 말씀하셨던'이라는 표현을 쓸 수 있겠네요! 진짜 기억하는 것처럼요." **실전 팁** 💡 - 현재 유효한 정보 조회 시 `WHERE valid_until IS NULL` 조건을 습관화하세요 - 버전 폭발을 방지하기 위해 사소한 변경은 무시하거나 통합하는 로직을 추가하세요 --- ## 5. 메모리_조회_및_업데이트_API Temporal Knowledge Graph를 구현한 김개발 씨는 이제 실제로 사용할 API가 필요했습니다. "CRUD 정도면 되겠지?"라고 생각했지만, 박시니어 씨가 고개를 저었습니다. "Knowledge Graph는 단순 CRUD가 아니야. 관계 탐색과 추론이 핵심이지." **메모리 API**는 Knowledge Graph에 지식을 저장하고 조회하는 인터페이스입니다. 단순한 키-값 저장이 아니라, 관계를 따라 탐색하고 연관 정보를 함께 가져오는 것이 특징입니다. 마치 도서관 사서가 관련 책들을 함께 추천해주는 것과 같습니다. 다음 코드를 살펴봅시다. ```python from typing import List, Dict, Optional from datetime import datetime, timezone class MemoryAPI: def __init__(self, temporal_graph): self.graph = temporal_graph # 지식 저장 def remember(self, user_id: str, subject: str, predicate: str, obj: str, confidence: float = 1.0) -> str: query = """ MERGE (u:User {id: $user_id}) MERGE (s:Entity {name: $subject}) MERGE (o:Entity {name: $object}) MERGE (s)-[r:%s { confidence: $confidence, created_at: datetime(), valid_from: datetime(), user_id: $user_id }]->(o) RETURN id(r) as relation_id """ % predicate.upper() result = self.graph.conn.execute_query(query, { "user_id": user_id, "subject": subject, "object": obj, "confidence": confidence }) return result[0]["relation_id"] if result else None # 관련 지식 조회 (2단계 깊이까지) def recall(self, user_id: str, query_entity: str, depth: int = 2) -> List[Dict]: query = """ MATCH (u:User {id: $user_id}) MATCH path = (start:Entity {name: $query})-[*1..%d]-(connected) WHERE ALL(r IN relationships(path) WHERE r.valid_until IS NULL) RETURN DISTINCT connected.name as entity, [r IN relationships(path) | type(r)] as relations, length(path) as distance ORDER BY distance LIMIT 20 """ % depth return self.graph.conn.execute_query(query, { "user_id": user_id, "query": query_entity }) # 특정 관계 검색 def find_relation(self, user_id: str, subject: str, predicate: str) -> List[Dict]: query = """ MATCH (s:Entity {name: $subject})-[r:%s]->(o:Entity) WHERE r.user_id = $user_id AND r.valid_until IS NULL RETURN o.name as object, r.confidence as confidence """ % predicate.upper() return self.graph.conn.execute_query(query, { "user_id": user_id, "subject": subject }) ``` API 설계는 사용자 경험을 결정합니다. 아무리 좋은 데이터베이스라도 API가 불편하면 개발자들이 사용하지 않습니다. 김개발 씨는 처음에 `create_node`, `create_edge`, `get_node` 같은 저수준 API를 만들었습니다. 하지만 실제로 사용하려니 너무 번거로웠습니다. "피자를 좋아한다"를 저장하려면 노드 2개를 만들고, 엣지를 연결하고, 속성을 설정해야 했습니다. 박시니어 씨가 조언했습니다. "도메인 언어로 생각해봐. '기억한다', '떠올린다'처럼 자연스러운 동사를 써봐." **remember** 메서드는 이름 그대로 "기억하기"입니다. 주어-서술어-목적어 형태로 지식을 저장합니다. "김철수-좋아한다-피자"처럼요. 내부적으로는 노드와 관계를 자동으로 생성하거나 연결합니다. **MERGE** 구문이 핵심입니다. 이미 존재하는 노드면 재사용하고, 없으면 새로 만듭니다. 따라서 "김철수"를 여러 번 저장해도 노드는 하나만 생성됩니다. **recall** 메서드는 "떠올리기"입니다. 하나의 개념에서 시작해서 연결된 모든 지식을 탐색합니다. `depth` 파라미터로 탐색 깊이를 조절할 수 있습니다. 예를 들어 "피자"를 조회하면 "피자-재료-치즈", "피자-원산지-이탈리아", "이탈리아-수도-로마"처럼 연결된 정보들이 줄줄이 나옵니다. depth=2면 2단계까지, depth=3이면 3단계까지 탐색합니다. 쿼리의 `[*1..%d]` 부분은 가변 길이 경로 탐색입니다. 1부터 depth까지의 모든 경로를 찾습니다. `WHERE ALL(r IN relationships(path) WHERE r.valid_until IS NULL)` 조건은 현재 유효한 관계만 포함하도록 필터링합니다. **find_relation** 메서드는 특정 관계를 직접 검색합니다. "김철수가 좋아하는 것"을 찾고 싶을 때 사용합니다. recall보다 정확하고 빠르지만, 탐색 범위가 좁습니다. 실무에서는 이 세 가지 API를 조합해서 사용합니다. 먼저 find_relation으로 직접적인 관계를 찾고, 없으면 recall로 넓게 탐색하는 식입니다. 주의할 점이 있습니다. depth를 너무 크게 설정하면 성능이 급격히 나빠집니다. 그래프 탐색은 지수적으로 복잡도가 증가하기 때문입니다. 대부분의 경우 depth=2 또는 3이면 충분합니다. 김개발 씨는 API를 테스트하며 뿌듯해했습니다. "이제 정말 AI가 기억하는 것 같아요!" **실전 팁** 💡 - recall의 depth는 2-3으로 제한하고, 필요시 점진적으로 늘리세요 - 자주 사용하는 쿼리 패턴은 별도 메서드로 추출하여 성능 최적화하세요 --- ## 6. 메모리_통합_및_아카이빙_로직 메모리 시스템이 잘 동작하자 새로운 문제가 생겼습니다. 몇 주 만에 노드가 수만 개로 늘어난 것입니다. "피자를 좋아해요", "피자 먹고 싶어요", "피자가 최고야"가 모두 별개로 저장되어 있었습니다. 김개발 씨는 한숨을 쉬었습니다. "이걸 어떻게 정리하죠?" **메모리 통합**은 비슷한 지식들을 하나로 합치는 과정입니다. **아카이빙**은 오래되거나 덜 중요한 정보를 별도 저장소로 옮기는 것입니다. 마치 책장을 정리하듯이, 비슷한 책은 묶고 안 읽는 책은 창고로 보내는 것과 같습니다. 다음 코드를 살펴봅시다. ```python from datetime import datetime, timedelta from collections import defaultdict class MemoryConsolidator: def __init__(self, memory_api, similarity_threshold: float = 0.8): self.api = memory_api self.threshold = similarity_threshold def consolidate_similar_memories(self, user_id: str): # 유사한 엔터티들을 찾아서 통합 query = """ MATCH (e1:Entity)-[r1]->(target) MATCH (e2:Entity)-[r2]->(target) WHERE e1.name e2.name AND r1.user_id = $user_id AND r2.user_id = $user_id AND type(r1) = type(r2) WITH e1, e2, collect(target) as shared_targets WHERE size(shared_targets) >= 2 RETURN e1.name as entity1, e2.name as entity2, size(shared_targets) as overlap ORDER BY overlap DESC """ candidates = self.api.graph.conn.execute_query(query, {"user_id": user_id}) for candidate in candidates: if self._calculate_similarity(candidate["entity1"], candidate["entity2"]) > self.threshold: self._merge_entities(user_id, candidate["entity1"], candidate["entity2"]) def archive_old_memories(self, user_id: str, days_threshold: int = 90): # 오래된 메모리를 아카이브로 이동 cutoff_date = datetime.now() - timedelta(days=days_threshold) query = """ MATCH (e:Entity)-[r]-() WHERE r.user_id = $user_id AND r.created_at (target) WHERE r.user_id = $user_id MERGE (e1)-[new_r:MERGED_FROM]->(target) SET new_r = properties(r), new_r.merged_from = $entity2 DELETE r """ self.api.graph.conn.execute_query(query, { "user_id": user_id, "entity1": entity1, "entity2": entity2 }) ``` 인간의 뇌도 비슷한 과정을 거칩니다. 수면 중에 뇌는 하루 동안 쌓인 기억을 정리합니다. 비슷한 기억은 통합하고, 중요하지 않은 기억은 잊어버립니다. 이것을 **기억 통합**이라고 합니다. AI 메모리 시스템도 마찬가지입니다. 정리하지 않으면 쓸모없는 정보가 쌓여서 검색 성능이 떨어지고, 정확한 정보를 찾기 어려워집니다. **consolidate_similar_memories** 메서드는 유사한 엔터티를 찾습니다. "피자"와 "Pizza", "피자요리"처럼 비슷하지만 다르게 저장된 것들입니다. 유사도를 판단하는 기준은 **공유 타겟**입니다. 두 엔터티가 같은 대상과 같은 종류의 관계를 맺고 있다면 유사할 가능성이 높습니다. 예를 들어 "피자"와 "Pizza"가 모두 "치즈", "토마토", "이탈리아"와 연결되어 있다면 같은 개념일 확률이 높습니다. `WHERE size(shared_targets) >= 2` 조건은 최소 2개 이상의 공유 타겟이 있어야 후보로 고려한다는 의미입니다. 너무 느슨하면 관련 없는 엔터티가 합쳐질 수 있습니다. **archive_old_memories** 메서드는 오래된 메모리를 정리합니다. 두 가지 기준을 사용합니다. 첫째, 생성된 지 90일이 지났는지. 둘째, 접근 횟수가 3회 미만인지. 둘 다 해당하면 "잘 안 쓰는 오래된 정보"로 판단합니다. 아카이브된 메모리는 삭제되지 않습니다. `archived = true` 플래그만 설정됩니다. 나중에 필요하면 복원할 수 있습니다. 완전 삭제는 별도의 정책으로 훨씬 나중에 진행합니다. **_merge_entities** 메서드는 실제 통합을 수행합니다. entity2의 모든 관계를 entity1으로 이전합니다. `MERGED_FROM` 관계 타입과 `merged_from` 속성으로 통합 이력을 추적할 수 있습니다. 실무에서는 이 작업을 **배치 작업**으로 스케줄링합니다. 매일 새벽 3시처럼 사용자가 적은 시간에 실행합니다. 실시간으로 하면 서비스 성능에 영향을 줄 수 있기 때문입니다. 주의할 점이 있습니다. 자동 통합은 실수할 수 있습니다. "Apple(회사)"와 "apple(과일)"을 합쳐버릴 수도 있습니다. 따라서 통합 전에 로그를 남기고, 문제가 발견되면 롤백할 수 있어야 합니다. 김개발 씨는 통합 로직을 실행하고 결과를 확인했습니다. 수만 개였던 노드가 수천 개로 줄었습니다. "와, 훨씬 깔끔해졌어요!" **실전 팁** 💡 - 통합 작업은 반드시 로그를 남기고 롤백 가능하게 설계하세요 - similarity_threshold를 너무 낮게 설정하면 다른 개념이 합쳐질 위험이 있습니다 --- ## 7. 대화형_에이전트에_메모리_시스템_통합 모든 구성 요소가 준비되었습니다. 이제 마지막 단계입니다. 김개발 씨는 설렘 반 걱정 반으로 물었습니다. "이 모든 걸 어떻게 챗봇에 연결하죠?" 박시니어 씨가 미소를 지었습니다. "자, 이제 진짜 마법이 시작되는 거야." **메모리 통합 에이전트**는 대화 중에 자동으로 정보를 추출하고 저장하며, 응답 생성 시 관련 기억을 활용합니다. 마치 오래된 친구처럼 과거 대화를 기억하고, 사용자의 취향을 파악하여 개인화된 응답을 제공합니다. 다음 코드를 살펴봅시다. ```python import re from typing import List, Dict, Optional from dataclasses import dataclass @dataclass class ConversationTurn: user_message: str assistant_response: str extracted_facts: List[Dict] retrieved_memories: List[Dict] class MemoryEnabledAgent: def __init__(self, memory_api, llm_client): self.memory = memory_api self.llm = llm_client self.conversation_history: List[ConversationTurn] = [] async def chat(self, user_id: str, message: str) -> str: # 1단계: 관련 기억 검색 relevant_memories = self._retrieve_relevant_memories(user_id, message) # 2단계: 메모리를 포함한 프롬프트 구성 prompt = self._build_prompt_with_memory(message, relevant_memories) # 3단계: LLM 응답 생성 response = await self.llm.generate(prompt) # 4단계: 대화에서 새로운 사실 추출 및 저장 extracted_facts = self._extract_and_store_facts(user_id, message, response) # 5단계: 대화 기록 저장 self.conversation_history.append(ConversationTurn( user_message=message, assistant_response=response, extracted_facts=extracted_facts, retrieved_memories=relevant_memories )) return response def _retrieve_relevant_memories(self, user_id: str, message: str) -> List[Dict]: # 메시지에서 키워드 추출 후 관련 메모리 검색 keywords = self._extract_keywords(message) memories = [] for keyword in keywords[:3]: # 상위 3개 키워드만 memories.extend(self.memory.recall(user_id, keyword, depth=2)) return memories[:10] # 최대 10개 반환 def _build_prompt_with_memory(self, message: str, memories: List[Dict]) -> str: memory_context = "\n".join([ f"- {m['entity']}: {m['relations']}" for m in memories ]) if memories else "관련 기억 없음" return f"""당신은 사용자와의 과거 대화를 기억하는 AI 어시스턴트입니다. 관련 기억: {memory_context} 사용자 메시지: {message} 위 기억을 참고하여 개인화된 응답을 생성하세요.""" def _extract_and_store_facts(self, user_id: str, message: str, response: str) -> List[Dict]: # 간단한 패턴 매칭으로 사실 추출 (실제로는 NLP/LLM 사용) facts = [] patterns = [ (r"저는 (.+)입니다", "USER", "IS_A"), (r"(.+)을 좋아해요", "USER", "LIKES"), (r"(.+)에 살아요", "USER", "LIVES_IN"), ] for pattern, subject, predicate in patterns: match = re.search(pattern, message) if match: obj = match.group(1) self.memory.remember(user_id, subject, predicate, obj, confidence=0.9) facts.append({"subject": subject, "predicate": predicate, "object": obj}) return facts ``` 드디어 모든 퍼즐 조각이 맞춰지는 순간입니다. **MemoryEnabledAgent** 클래스는 메모리 시스템과 LLM을 연결하는 다리 역할을 합니다. `memory_api`는 앞서 만든 Knowledge Graph API이고, `llm_client`는 GPT나 Claude 같은 언어 모델 클라이언트입니다. **chat** 메서드가 핵심입니다. 사용자 메시지가 들어오면 5단계 파이프라인을 거칩니다. 첫 번째 단계는 **관련 기억 검색**입니다. 메시지에서 키워드를 추출하고, 각 키워드와 연관된 메모리를 Knowledge Graph에서 가져옵니다. "오늘 점심 뭐 먹을까?"라는 메시지에서 "점심", "먹다" 같은 키워드를 추출하고, 사용자의 음식 취향을 검색합니다. 두 번째 단계는 **프롬프트 구성**입니다. 검색된 메모리를 LLM 프롬프트에 포함시킵니다. "이 사용자는 피자를 좋아하고, 매운 음식을 싫어합니다"처럼요. LLM은 이 정보를 참고하여 응답을 생성합니다. 세 번째 단계는 **응답 생성**입니다. LLM이 메모리를 참고하여 개인화된 답변을 만듭니다. "지난번에 좋아하신다고 하셨던 피자 어떠세요?"처럼 과거 대화를 반영할 수 있습니다. 네 번째 단계는 **사실 추출**입니다. 사용자 메시지에서 저장할 만한 정보를 찾아냅니다. "저는 개발자입니다"라고 말하면 `(USER, IS_A, 개발자)` 트리플이 추출되어 저장됩니다. 다섯 번째 단계는 **대화 기록 저장**입니다. 이번 대화에서 무슨 일이 있었는지 기록합니다. 나중에 분석하거나 디버깅할 때 유용합니다. `_extract_and_store_facts` 메서드는 간단한 정규표현식을 사용합니다. 실제 프로덕션에서는 더 정교한 NLP 기법이나 LLM 기반 추출을 사용합니다. 하지만 원리는 동일합니다. confidence=0.9로 설정한 것에 주목하세요. 자동 추출이므로 100% 확신할 수 없습니다. 나중에 사용자가 "아니, 그건 틀렸어"라고 말하면 confidence를 낮추거나 삭제할 수 있습니다. 실무에서는 몇 가지 추가 고려사항이 있습니다. 메모리 검색 시간이 길어지면 응답 지연이 발생합니다. 따라서 검색을 비동기로 처리하거나, 캐싱을 적용해야 합니다. 또한 너무 많은 메모리를 프롬프트에 넣으면 LLM의 컨텍스트 제한을 초과할 수 있습니다. 김개발 씨는 완성된 에이전트를 테스트했습니다. "안녕, 나 김철수야"라고 말한 후, 한참 뒤에 "내 이름 기억해?"라고 물었습니다. 에이전트가 대답했습니다. "네, 김철수 님이시죠!" 김개발 씨의 눈이 반짝였습니다. 박시니어 씨가 어깨를 두드렸습니다. "축하해. 이제 네 AI는 진짜 기억할 수 있어." **실전 팁** 💡 - 메모리 검색과 LLM 호출을 비동기로 병렬 처리하면 응답 속도가 빨라집니다 - 프롬프트에 넣는 메모리 개수를 제한하여 컨텍스트 오버플로우를 방지하세요 --- 이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!

카테고리: AI Engineering

언어: Python

태그: Python, KnowledgeGraph, Neo4j, MemorySystem, AIAgent, AI Engineering

작성자: AI Generated

CodeDeck

Knowledge Graph 메모리 시스템 완벽 가이드

목차

1. 메모리 아키텍처 설계

2. Neo4j 설치 및 기본 설정

3. 엔터티 및 관계 스키마 설계

4. Temporal Knowledge Graph 구현

5. 메모리 조회 및 업데이트 API

6. 메모리 통합 및 아카이빙 로직

7. 대화형 에이전트에 메모리 시스템 통합

댓글 (0)

함께 보면 좋은 카드 뉴스

Production-Grade AI Agent System 완벽 가이드

LLM-as-a-Judge 평가 시스템 완벽 가이드

Agent-Optimized Tool Set 설계 실습 가이드

Multi-Agent Research Assistant 실습 가이드

Context Manager 구현하기 실습 가이드