본문 바로가기
Agentic AI/Memory

5. 장기 메모리(2)(Long-term Memory) - Python으로 PostgreSQL을 장기 기억 저장소처럼 사용하기

by 아톨 2026. 9. 3.
반응형

이제부터는 "DB를 사용할 수 있는 Agent의 기본기"를 제대로 만드는 단계입니다.

이번 단계의 목표는 OpenAI나 Agent를 아직 붙이지 않는 것입니다. 먼저 Python이 PostgreSQL을 장기 기억 저장소처럼 사용할 수 있는지를 확실하게 테스트합니다.

 

[전체 요약]

# 1. Docker PostgreSQL에 접속
docker exec -it agent-postgres psql -U agent_user -d agent_db
===> agent_db=#

# 2. 사용자 테이블
CREATE TABLE IF NOT EXISTS user_profiles (
    user_id SERIAL PRIMARY KEY,
    user_name VARCHAR(50) NOT NULL,
    location VARCHAR(100),
    preferences TEXT
);

# 3. 기억 테이블
CREATE TABLE IF NOT EXISTS memories (
    memory_id SERIAL PRIMARY KEY,
    user_id INTEGER NOT NULL REFERENCES user_profiles(user_id),
    memory_type VARCHAR(50),
    content TEXT NOT NULL,
    importance INTEGER DEFAULT 5,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

# 4. 테이블이 제대로 만들어졌는지 확인(PostgreSQL에서)
1) \dt
===> List of relations 확인
2)\d user_profiles
===> Table "public.user_profiles" 확인
3)\d memories
===> Table "public.memories" 확인
4)\q
===> PostgreSQL을 빠져나오기

# 5. database.py 만들기

# 6. .env 만들기

# 7. 테스트 프로그램 만들기(test_memory_db.py)
===> 실제 Agent가 아니라 DB 기능을 시험하는 실험 프로그램

# 8. 실행: python test_memory_db.py

# 9. 한번 더실행

# 10. Docker를 꺼도 데이터가 남아 있는지 테스트
1) 현재 Docker Container 확인: docker ps
2) PostgreSQL 중지: docker stop agent-postgres
3) 다시 확인(PostgreSQL 없어졌을것임): docker ps
4) 다시 PostgreSQL 실행: docker start agent-postgres
5) 확인: docker ps
6) 실행: python test_memory_db.py
===> DB 데이터가 계속 남아 있음. Docker Container와 DB 데이터 저장소인 Volume이 별개라는 개념 이해

 

전체 흐름은:

Docker
  ↓
PostgreSQL
  ↓
database.py
  ↓
Python에서 DB 조작
  ↓
test_memory_db.py
  ↓
사용자 생성
  ↓
사용자 조회
  ↓
기억 저장
  ↓
기억 조회
 

입니다.


1. 먼저 전체 구조를 이해합시다

이번에 만들 폴더를 이렇게 생각하면 됩니다.

agent_project/
│
├── .env
│
├── database.py
│
└── test_memory_db.py
 

역할은 명확하게 나눕니다.

database.py

DB 담당자입니다.

database.py

connect_db()
    ↓
PostgreSQL 연결

create_user()
    ↓
사용자 저장

get_user()
    ↓
사용자 조회

save_memory()
    ↓
장기 기억 저장

get_memories()
    ↓
장기 기억 조회

close_connection()
    ↓
DB 연결 종료
 

test_memory_db.py

실험 담당자입니다.

test_memory_db.py

database.py를 가져옴
        ↓
함수 호출
        ↓
결과 확인
 

이렇게 분리하는 습관이 중요합니다.

나중에는:

Agent
  ↓
database.py
  ↓
PostgreSQL
 

가 되기 때문입니다.


2. 먼저 PostgreSQL 테이블을 준비합니다

앞에서 만들었던 user_profiles에 더해서 memories 테이블을 만들겠습니다.

Docker PostgreSQL에 접속합니다.

 
docker exec -it agent-postgres psql -U agent_user -d agent_db
 

그러면:

agent_db=#
 

가 나옵니다.


3. 사용자 테이블

다음 SQL을 실행하세요.

 
CREATE TABLE IF NOT EXISTS user_profiles (
    user_id SERIAL PRIMARY KEY,
    user_name VARCHAR(50) NOT NULL,
    location VARCHAR(100),
    preferences TEXT
);
 

여기서 중요한 것은:

 
user_id SERIAL PRIMARY KEY
 

입니다.

PostgreSQL이 자동으로:

1
2
3
4
...
 

사용자 ID를 만들어줍니다.


4. 기억 테이블

이제 memories 테이블을 만듭니다.

 
CREATE TABLE IF NOT EXISTS memories (
    memory_id SERIAL PRIMARY KEY,
    user_id INTEGER NOT NULL REFERENCES user_profiles(user_id),
    memory_type VARCHAR(50),
    content TEXT NOT NULL,
    importance INTEGER DEFAULT 5,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
 

각 컬럼을 이해하는 것이 중요합니다.

컬럼의미
memory_id 기억의 고유번호
user_id 이 기억이 누구의 것인지
memory_type preference, goal, fact 등
content 실제 기억 내용
importance 기억의 중요도
created_at 저장된 시간

예를 들어:

user_id = 1
memory_type = preference
content = "코드 설명을 자세하게 듣는 것을 선호한다."
importance = 8
 

이라면:

1번 사용자가 "코드 설명을 자세하게 듣는 것을 선호한다"는 사실을 중요도 8로 기억하고 있다.

는 의미입니다.


5. 두 테이블의 관계

이 부분은 반드시 이해하고 넘어가는 것이 좋습니다.

user_profiles
────────────────────
user_id = 1
user_name = Kim
location = Seoul
        │
        │
        │ 1:N 관계
        ▼
memories
────────────────────
memory_id = 1
user_id = 1
content = Python을 공부하고 있다

memory_id = 2
user_id = 1
content = AI Agent에 관심이 있다

memory_id = 3
user_id = 1
content = 코드 설명을 자세히 듣는 것을 선호한다
 

즉,

한 명의 사용자에게 여러 개의 기억이 연결됩니다.

이것이:

 
user_id INTEGER REFERENCES user_profiles(user_id)
 

의 의미입니다.


6. 테이블이 제대로 만들어졌는지 확인

PostgreSQL에서:

 
\dt
 

실행하세요.

다음과 같이 보여야 합니다.

public | memories      | table
public | user_profiles | table
 

그리고:

 
\d user_profiles
 
 
\d memories
 

로 구조를 확인할 수 있습니다.

다 확인했으면:

 
\q
 

로 PostgreSQL을 빠져나옵니다.


7. 이제 database.py를 만듭니다

이번 코드에서는 각 함수가 무엇을 하는지 명확하게 보이도록 작성하겠습니다.

 
    # ============================================================
    # database.py : PostgreSQL 장기 메모리 관리를 담당하는 모듈
    # 기능
    # 1. connect_db()
    # 2. create_user()
    # 3. get_user()
    # 4. save_memory()
    # 5. get_memories()
    # 6. close_connection()
    # ============================================================
    import os
    import psycopg
    from dotenv import load_dotenv

    # ============================================================
    # 1. 환경설정
    # ============================================================
    load_dotenv()

    # PostgreSQL 접속 정보
    DB_HOST = os.getenv("DB_HOST", "localhost")
    DB_PORT = os.getenv("DB_PORT", "5432")
    DB_NAME = os.getenv("DB_NAME", "agent_db")
    DB_USER = os.getenv("DB_USER", "agent_user")
    DB_PASSWORD = os.getenv("DB_PASSWORD", "agent_password")

    # ============================================================
    # 2. DB 연결
    # ============================================================
    def connect_db():
        """
        PostgreSQL 데이터베이스에 연결한다.
        반환값: psycopg Connection 객체
        """
        conn = psycopg.connect(
            host = DB_HOST,
            port = DB_PORT,
            dbname = DB_NAME,
            user = DB_USER,
            password = DB_PASSWORD
        )
        print("☑️ PostgreSQL 연결 성공")
        return conn

    # ============================================================
    # 3. 사용자 생성
    # ============================================================
    def create_user(conn, user_name, location=None, preferences=None):
        """
        새로운 사용자를 생성한다.
        매개변수:
            conn        :DB 연결 객체
            user_name   :사용자 이름
            location    :위치
            preferences :사용자 선호
        반환값: 생성된 user_id
        """

        sql = """
            INSERT INTO
                user_profiles (user_name, location, preferences)
            VALUES
                (%s, %s, %s)
            RETURNING
                user_id;
        """

        with conn.cursor() as cursor:
            cursor.execute(sql, (user_name, location, preferences))
            user_id = cursor.fetchone()[0]

        # DB에 실제 반영
        conn.commit()

        print(f"☑️ 사용자 생성 완료 (user_id = {user_id})")
        return user_id

    # ============================================================
    # 4. 사용자 조회
    # ============================================================
    def get_user(conn, user_id):
        """
        user_id를 이용해서 사용자 정보를 조회한다.
        반환값: 사용자 정보(tuple). 없으면 None
        """

        sql = """
            SELECT
                user_id, user_name, location, preferences
            FROM
                user_profiles
            WHERE
                user_id = %s;    
        """

        with conn.cursor() as cursor:
            cursor.execute(sql, (user_id,))
            user = cursor.fetchone()

        return user

    # ============================================================
    # 5. 기억 저장
    # ============================================================
    def save_memory(conn, user_id, memory_type, content, importance=5):
        """
        사용자의 장기 기억을 저장한다.
        매개변수:
            conn: DB 연결 개체
            user_id: 사용자 ID
            memory_type: 기억 종류
            content: 기억 내용
            importance: 중요도(기본값 5)
        반환값: 생성된 memory_id
        """

        sql = """
            INSERT INTO
                memories (user_id, memory_type, content, importance)
            VALUES
                (%s, %s, %s, %s)
            RETURNING
                memory_id;
        """

        with conn.cursor() as cursor:
            cursor.execute(sql, (user_id, memory_type, content, importance))
            memory_id = cursor.fetchone()[0]

        #DB에 실제 반영
        conn.commit()

        print(f"🧠기억 저장 완료 (memory_id={memory_id})")

        return memory_id

    # ============================================================
    # 6. 기억 조회
    # ============================================================
    def get_memories(conn, user_id, limit=10):
        """
        특정 사용자의 장기 기억을 조회한다.
        중요도가 높은 기억을 우선하고, 같은 중요도라면 최근 기억부터 가져온다.
        반환값: 기억 목록
        """

        sql = """
            SELECT
                memory_id, memory_type, content, importance, created_at
            FROM
                memories
            WHERE
                user_id = %s
            ORDER BY
                importance DESC, created_at DESC
            LIMIT %s;
        """

        with conn.cursor() as cursor:
            cursor.execute(sql, (user_id, limit))
            memories = cursor.fetchall()

        return memories

    # ============================================================
    # 7. DB 연결 종료
    # ============================================================
    def close_connection(conn):
        """ PostgreSQL 연결을 종료한다."""

        if conn is not None:
            conn.close()
            print("📴 PostgreSQL 연결 종료")


이제 이 파일 하나가 PostgreSQL을 다루는 전담 모듈이 됩니다.


8. .env도 만들어 봅시다

프로젝트 폴더에:

.env
 

파일을 만들고:

DB_HOST=localhost
DB_PORT=5432
DB_NAME=agent_db
DB_USER=agent_user
DB_PASSWORD=agent_password
 

를 넣습니다.

그러면 코드에:

 
password="agent_password"
 

처럼 DB 비밀번호를 직접 적지 않아도 됩니다.


9. 이제 테스트 프로그램을 만듭니다

이제 가장 중요한 test_memory_db.py입니다.

 
    # ============================================================
    # test_memory_db.py : PostgreSQL 장기 메모리 기능 테스트
    # 테스트 순서
    # 1. 사용자 생성
    # 2. 사용자 조회
    # 3. 기억 저장
    # 4. 기억 여러 개 저장
    # 5. 기억 조회
    # 6. 결과 출력
    # ============================================================

    from database import (
        connect_db,
        create_user,
        get_user,
        save_memory,
        get_memories,
        close_connection
    )

    # ============================================================
    # 1. 프로그램 시작
    # ============================================================
    print("\n")
    print("="*70)
    print("PostgreSQL Long-term Memory Test")
    print("="*70)

    # ============================================================
    # 2. DB 연결
    # ============================================================
    conn = connect_db()

    try:
        # ============================================================
        # 3. 사용자 생성
        # ============================================================
        print("\n")
        print("="*70)
        print("1. 사용자 생성")
        print("="*70)

        user_id = create_user(
            conn,
            user_name="Kim",
            location="Seoul",
            preferences=(
                "Python과 AI Agent에 관심이 있으며 "
                "코드 설명을 자세하게 듣는 것을 선호"
            )
        )

        # ============================================================
        # 4. 사용자 조회
        # ============================================================
        print("\n")
        print("="*70)
        print("2. 사용자 조회")
        print("="*70)

        user = get_user(conn, user_id)
        if user is not None:
            print(f"user_id: {user[0]}")
            print(f"user_name: {user[1]}")
            print(f"location: {user[2]}")
            print(f"preferences: {user[3]}")
        else:
            print("❎ 사용자를 찾을 수 없습니다.")

        # ============================================================
        # 5. 첫 번째 기억 저장
        # ============================================================
        print("\n")
        print("="*70)
        print("3. 첫 번째 기억 저장")
        print("="*70)

        memory_id = save_memory(
            conn,
            user_id=user_id,
            memory_type="interest",
            content=("사용자는 AI Agent 개발에 관심이 있다."),
            importance=8
        )
        print(f"저장된 memory_id: {memory_id}")

        # ============================================================
        # 6. 여러 개의 기억 저장
        # ============================================================
        print("\n")
        print("="*70)
        print("4. 여러 개의 기억 저장")
        print("="*70)

        save_memory(
            conn,
            user_id=user_id,
            memory_type="preference",
            content=("사용자는 코드 설명을 자세하게 듣는 것을 선호한다."),
            importance=9
        )

        save_memory(
            conn,
            user_id=user_id,
            memory_type="goal",
            content=("사용자는 실제 업무에 사용할 AI Agent를 만들고 싶어 한다."),
            importance=10
        )

        save_memory(
            conn,
            user_id=user_id,
            memory_type="study",
            content=(
                "사용자는 Python을 이용하여 "
                "AI Agent를 공부하고 있다."
            ),
            importance=8
        )

        save_memory(
            conn,
            user_id=user_id,
            memory_type="interest",
            content=(
                "사용자는 PostgreSQL을 이용한 "
                "장기 메모리 구현을 학습하고 있다."
            ),
            importance=9
        )

        # ============================================================
        # 7. 기억 조회
        # ============================================================
        print("\n")
        print("="*70)
        print("5. 기억 조회")
        print("="*70)

        memories = get_memories(
            conn,
            user_id=user_id,
            limit=10
        )

        # ============================================================
        # 8. 결과 출력
        # ============================================================
        print("\n")
        print("="*70)
        print("6. 저장된 장기 기억")
        print("="*70)

        for memory in memories:
            memory_id = memory[0]
            memory_type = memory[1]
            content = memory[2]
            importance = memory[3]
            created_at = memory[4]

            print("\n")
            print(f"memory_id : {memory_id}")
            print(f"type      : {memory_type}")
            print(f"content   : {content}")
            print(f"importance: {importance}")
            print(f"created_at: {created_at}")
            print("-"*70)

    finally:
        # ============================================================
        # 9. DB 연결 종료
        # ============================================================
        close_connection(conn)

    print("\n")
    print("="*70)
    print("테스트 종료")
    print("="*70)

 


10. 이제 실행합니다

폴더 구조가:

agent_project/
│
├── .env
├── database.py
└── test_memory_db.py
 

라면 터미널에서:

 
python test_memory_db.py
 

를 실행합니다.


11. 정상 실행되면 어떤 일이 일어날까?

첫 번째:

======================================================================
PostgreSQL Long-term Memory Test
======================================================================

✅ PostgreSQL 연결 성공
 

그리고:

1. 사용자 생성

✅ 사용자 생성 완료 (user_id=1)
 

그 다음:

2. 사용자 조회

user_id     : 1
user_name   : Kim
location    : Seoul
preferences : Python과 AI Agent에 관심이 있으며...
 

그리고 기억 저장:

3. 첫 번째 기억 저장

🧠 기억 저장 완료 (memory_id=1)
 

여러 개 저장:

4. 여러 개의 기억 저장

🧠 기억 저장 완료 (memory_id=2)
🧠 기억 저장 완료 (memory_id=3)
🧠 기억 저장 완료 (memory_id=4)
🧠 기억 저장 완료 (memory_id=5)
 

그리고:

6. 저장된 장기 기억
 

이런 식으로 출력될 것입니다.

memory_id : 3
type      : goal
content   : 사용자는 실제 업무에 사용할 AI Agent를 만들고 싶어 한다.
importance: 10
created_at: ...
 

중요도가 높은 것부터 나오게 했기 때문에 importance=10인 기억이 먼저 나옵니다.


12. 그런데 여기서 아주 중요한 것을 확인해야 합니다

테스트 프로그램을 종료하세요.

테스트 종료
 

그리고 다시:

 
python test_memory_db.py
 

실행해 보세요.

이번에는:

user_id=2
 

가 될 가능성이 높습니다.

왜냐하면 새로운 사용자를 또 생성했기 때문입니다.

그리고:

memory_id=6
7
8
...
 

처럼 계속 증가할 것입니다.

이것이 바로 DB의 특징입니다.

Python 프로그램이 종료되어도:

Python 종료
       ↓
messages 사라짐
 

하지만:

PostgreSQL
       ↓
데이터 그대로 유지
 

됩니다.


13. Docker를 꺼도 데이터가 남아 있는지 테스트

이것도 한번 해보면 좋습니다.

현재 Docker Container를 확인:

 
docker ps
 

그리고 PostgreSQL을 중지:

 
docker stop agent-postgres
 

다시 확인:

 
docker ps
 

PostgreSQL이 없어졌을 것입니다.

그리고 다시:

 
docker start agent-postgres
 

실행합니다.

확인:

 
docker ps
 

그리고:

 
python test_memory_db.py
 

를 실행해 보세요.

만약 앞에서 제가 추천한:

-v agent_postgres_data:/var/lib/postgresql/data
 

를 사용했다면 DB 데이터가 계속 남아 있습니다.

이 실험을 통해 Docker Container와 DB 데이터 저장소인 Volume이 별개라는 개념도 자연스럽게 이해할 수 있습니다.


14. 지금 코드에서 정말 중요한 Python 개념

이번 코드에서 특히 주목해야 할 부분이 있습니다.

conn

 
conn = connect_db()
 

이것은:

PostgreSQL과 연결된 통로

입니다.

Python
   │
   │ conn
   ▼
PostgreSQL
 

이라고 생각하세요.


cursor

예를 들어:

 
with conn.cursor() as cursor:
 

는 SQL을 실행하기 위한 도구입니다.

conn
 │
 │ 연결
 ▼
PostgreSQL

cursor
 │
 │ SQL 실행
 ▼
SELECT / INSERT / UPDATE / DELETE
 

입니다.


15. %s가 매우 중요합니다

예를 들어:

 
cursor.execute(
    sql,
    (
        user_name,
        location,
        preferences
    )
)
 

그리고 SQL은:

 
VALUES (%s, %s, %s)
 

입니다.

이것은 단순한 문자열 조합이 아닙니다.

예를 들어 이렇게 하면 안 됩니다.

 
sql = f"""
INSERT INTO user_profiles
VALUES ('{user_name}')
"""
 

왜냐하면 SQL Injection 같은 문제가 발생할 수 있기 때문입니다.

psycopg에서는:

 
cursor.execute(
    sql,
    (user_name,)
)
 

형태로 값을 전달하는 습관을 들이는 것이 좋습니다.


16. commit()도 중요합니다

우리가:

 
cursor.execute(...)
 

했다고 해서 반드시 DB에 최종 반영되는 것은 아닙니다.

그래서:

 
conn.commit()
 

을 실행합니다.

쉽게 말하면:

INSERT
   ↓
DB 작업 준비
   ↓
commit()
   ↓
실제 반영
 

이라고 이해하면 됩니다.


17. RETURNING user_id는 왜 사용하는가?

이 부분:

 
INSERT INTO user_profiles (...)
VALUES (...)
RETURNING user_id;
 

그리고:

 
user_id = cursor.fetchone()[0]
 

는 상당히 중요한 패턴입니다.

DB가 자동으로:

user_id = 17
 

을 만들어줬다면 Python이 그 번호를 알아야 합니다.

그래서:

INSERT
 ↓
PostgreSQL이 ID 생성
 ↓
RETURNING user_id
 ↓
Python이 17을 받음
 

이 되는 것입니다.

그래서 이후:

 
save_memory(
    conn,
    user_id=17,
    ...
)
 

처럼 연결할 수 있습니다.


18. 지금 우리가 만든 것은 사실상 "Memory API"입니다

아직 FastAPI는 아니지만 개념적으로는 이미 API처럼 되어 있습니다.

database.py

create_user()
       ↓
사용자 생성

get_user()
       ↓
사용자 검색

save_memory()
       ↓
기억 저장

get_memories()
       ↓
기억 검색
 

나중에는 Agent가:

 
save_memory(
    conn,
    user_id,
    "preference",
    "사용자는 코드 설명을 자세히 듣는 것을 선호한다.",
    9
)
 

처럼 호출하면 됩니다.


19. 그리고 이것이 진짜 Agent로 연결됩니다

현재까지:

사용자
 ↓
Chat Loop
 ↓
messages
 ↓
Summary Memory
 

까지 만들었습니다.

이번에는:

PostgreSQL
 ↓
Long-term Memory
 

를 만든 것입니다.

그러면 최종 구조는:

                         ┌───────────────┐
                         │    사용자     │
                         └───────┬───────┘
                                 │
                                 ▼
                         ┌───────────────┐
                         │     Agent     │
                         └───────┬───────┘
                                 │
              ┌──────────────────┼──────────────────┐
              │                  │                  │
              ▼                  ▼                  ▼
       Recent History       Summary Memory    Long-term Memory
       Python messages      요약된 기억       PostgreSQL
              │                  │                  │
              └──────────────────┼──────────────────┘
                                 │
                                 ▼
                           OpenAI Model
                                 │
                                 ▼
                              답변
 

이 구조가 우리가 지금까지 공부해 온 것들이 하나로 합쳐지는 지점입니다.


20. 다만 지금 단계에서 한 가지는 일부러 하지 않았습니다

아직:

사용자의 말을 보고
"이건 기억해야겠다"
 

라고 LLM이 판단하는 기능은 넣지 않았습니다.

현재는 사람이 직접:

 
save_memory(...)
 

를 호출하고 있습니다.

이게 오히려 좋습니다.

왜냐하면 지금은 먼저:

Python
  ↕
PostgreSQL
 

을 확실히 이해해야 하기 때문입니다.

그 다음 단계에서:

사용자:
"나는 Python을 공부하고 있어."

       ↓

OpenAI

       ↓

"이 정보는 장기 기억으로 저장할 가치가 있다."

       ↓

save_memory()

       ↓

PostgreSQL
 

이라는 Memory Extraction을 붙이면 됩니다.


다음 단계에서 만들 것이 핵심입니다

지금 단계가 정상적으로 작동하면 다음에는 database.py를 크게 바꾸지 않고 다음과 같은 구조를 만들면 됩니다.

memory.py

extract_memory()
       ↓
LLM에게 중요한 정보인지 판단

save_memory()
       ↓
PostgreSQL 저장

search_memory()
       ↓
관련 기억 검색

build_memory_context()
       ↓
검색된 기억을 Agent에게 전달
 

그리고 최종적으로:

👤 사용자
"나는 요즘 Python으로 AI Agent를 만들고 있어."

              ↓

        ┌───────────┐
        │   Agent   │
        └─────┬─────┘
              │
       "중요한 정보인가?"
              │
             YES
              │
              ▼
       PostgreSQL 저장
              │
              ▼
       다음날 Agent 실행
              │
              ▼
사용자:
"내가 뭘 공부하고 있었지?"
              │
              ▼
       PostgreSQL 검색
              │
              ▼
       관련 기억 발견
              │
              ▼
"Python으로 AI Agent를
 공부하고 계셨습니다."
 

가 됩니다.

지금은 여기까지 먼저 직접 실행해 보시는 것을 강하게 권합니다. 특히 docker run → PostgreSQL 접속 → 테이블 생성 → database.py→test_memory_db.py 실행의 흐름을 한 번 직접 경험하면, 이후에 LLM을 DB와 연결하는 부분이 훨씬 쉽게 이해됩니다.

반응형