14. 장기기억(10) - PostgreSQL의 memories 테이블과 ChromaDB연결 단계(4.1)

현재 우리가 해결하려는 문제는 딱 하나입니다.

새로운 Memory가 PostgreSQL에 저장될 때 생성된 memory_id를 받아서, 그 Memory를 ChromaDB에도 정확히 저장한다.

 

[요약]

# PostgreSQL의 memories 테이블과 ChromaDB연결 단계(4.1)
=> database.py -> PostgreSQL save_memory()가 생성된 memory_id를 반환하도록 수정
# step1. database.py에서 save_memory()를 찾아, 코드 교체하기(수정 코드 없음)
# step2. test_save_memory.py 만들어서, 테스트

---

오늘 진행할 범위

전체 Dual Write는 다음 4단계입니다.

STEP 7-1  database.py
          ↓
          PostgreSQL save_memory()가
          생성된 memory_id를 반환하도록 수정

STEP 7-2  memory.py
          ↓
          save_memories()가
          저장된 Memory 정보를 반환하도록 수정

STEP 7-3  agent_with_memory.py
          ↓
          PostgreSQL 저장 후
          ChromaDB에도 저장하도록 연결

STEP 7-4  통합 테스트
          ↓
          "새 기억 생성 → 두 DB 저장 → 검색" 확인

 

오늘은 우선 STEP 7-1만 확실하게 끝내겠습니다.

---

STEP 7-1. database.py 수정

1. 왜 save_memory()를 수정하는가?

현재 Agent에서:

Memory 추출
   ↓
save_memories()
   ↓
PostgreSQL

 

까지는 잘 됩니다.

실제로 방금 테스트에서도:

memory_id=35
memory_id=36

 

이 생성됐습니다.

그런데 Python 코드 입장에서는 이 35, 36이라는 번호를 다음 단계에서 사용해야 합니다.

왜냐하면 ChromaDB에서는:

memory_id = 35
       ↓
ChromaDB ID
memory_35

 

로 저장할 것이기 때문입니다.

따라서:

PostgreSQL INSERT
       ↓
"방금 저장한 Memory의 ID가 몇 번이지?"
       ↓
35
       ↓
ChromaDB에 memory_35로 저장

 

이라는 흐름이 필요합니다.

---

2. PostgreSQL의 RETURNING을 사용합니다

PostgreSQL에는 아주 좋은 기능이 있습니다.

 

RETURNING memory_id

 

입니다.

예를 들어:

 

INSERT INTO memories (...)
VALUES (...)
RETURNING memory_id;

 

라고 하면 PostgreSQL이 데이터를 저장한 후:

35

 

를 바로 돌려줍니다.

즉,

INSERT
  ↓
저장
  ↓
생성된 PK 반환

 

이 한 번에 가능합니다.

---

3. 현재 database.py에서 save_memory()를 찾으세요

현재 함수는 대략 이런 구조일 것입니다.

 

def save_memory(conn, user_id, memory_type, content, importance=5):
    ...

 

이 함수만 교체하겠습니다.

나머지 database.py는 건드리지 않습니다.

다음 코드로 바꿔주세요.

 

def save_memory(
    conn,
    user_id,
    memory_type,
    content,
    importance=5,
):
    """
    하나의 Memory를 PostgreSQL에 저장하고
    새로 생성된 memory_id를 반환한다.
    """

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

    with conn.cursor() as cur:
        cur.execute(
            sql,
            (
                user_id,
                memory_type,
                content,
                importance,
            ),
        )

        memory_id = cur.fetchone()[0]

    conn.commit()

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

    return memory_id

 

---

4. 여기서 한 줄씩 이해해 봅시다

① 함수

 

def save_memory(
    conn,
    user_id,
    memory_type,
    content,
    importance=5,
):

 

기존과 거의 같습니다.

우리가 바꾸는 핵심은 저장 후 memory_id를 반환한다는 것입니다.

---

② INSERT

 

INSERT INTO memories (
    user_id,
    memory_type,
    content,
    importance
)

 

PostgreSQL의 memories 테이블에 Memory를 저장합니다.

예를 들어 이번에 추출된 Memory가:

user_id     = 8
memory_type = goal
content     = 사용자는 저작권 미구매 고객에게...
importance  = 9

 

이라면 이것을 DB에 넣습니다.

---

5. 가장 중요한 부분

 

RETURNING memory_id;

 

이것이 이번 수정의 핵심입니다.

예를 들어 DB가:

INSERT 성공

 

하면서 자동으로:

memory_id = 37

 

을 만들었다고 합시다.

그러면:

 

RETURNING memory_id

 

덕분에 Python이 그 37을 받을 수 있습니다.

---

6. Python에서는 이것을 받습니다

 

memory_id = cur.fetchone()[0]

 

여기서 조금 천천히 보겠습니다.

PostgreSQL이 반환한 결과는 대략:

 

(37,)

 

같은 tuple입니다.

따라서:

 

cur.fetchone()

 

결과가:

 

(37,)

 

이고,

 

cur.fetchone()[0]

 

은:

 

37

 

이 됩니다.

그래서:

 

memory_id = 37

 

이 되는 것입니다.

---

7. 왜 commit()은 그 뒤에 있나요?

 

conn.commit()

 

으로 실제 트랜잭션을 확정합니다.

전체 흐름은:

INSERT 실행
   ↓
memory_id 반환
   ↓
commit
   ↓
저장 완료

 

입니다.

---

8. 그리고 마지막에

 

return memory_id

 

를 합니다.

이것이 아주 중요합니다.

예를 들어:

 

memory_id = save_memory(
    conn,
    user_id=8,
    memory_type="goal",
    content="AI Agent를 만들고 싶다.",
    importance=9,
)

 

라고 하면:

 

memory_id

 

에:

37

 

이 들어옵니다.

앞으로 이것을 ChromaDB에 넘길 수 있습니다.

---

9. 이제 우리가 만들려는 최종 연결이 보입니다

현재는:

save_memory()
      ↓
PostgreSQL 저장
      ↓
끝

 

이지만 수정 후에는:

save_memory()
      ↓
PostgreSQL INSERT
      ↓
memory_id 반환
      ↓
Python이 memory_id를 받음
      ↓
ChromaDB에 전달

 

이 됩니다.

---

STEP 7-1 테스트

여기서 바로 Agent를 실행하지 않는 것이 좋습니다.

먼저 save_memory() 자체가 정확히 memory_id를 반환하는지 테스트하겠습니다.

프로젝트 폴더:

 

cd C:\Dev\book_agentic_ai\agent_project

 

그리고 새로운 테스트 파일:

test_save_memory.py

 

를 만드세요.

전체 코드는 다음과 같습니다.

 

from database import (
    connect_db,
    get_user_by_name,
    save_memory,
)


TEST_USER_NAME = "스티븐김"


def main():

    print("=" * 70)
    print("PostgreSQL save_memory() 반환값 테스트")
    print("=" * 70)

    # --------------------------------------------------
    # STEP 1. PostgreSQL 연결
    # --------------------------------------------------

    print("\n[STEP 1] PostgreSQL 연결")

    conn = connect_db()

    try:

        # --------------------------------------------------
        # STEP 2. 사용자 조회
        # --------------------------------------------------

        print("\n[STEP 2] 테스트 사용자 조회")

        user = get_user_by_name(
            conn,
            TEST_USER_NAME,
        )

        if user is None:
            print(
                f"❎ 사용자를 찾을 수 없습니다: "
                f"{TEST_USER_NAME}"
            )
            return

        user_id = user[0]

        print("☑️ 사용자 발견")
        print(f"   user_id   : {user_id}")
        print(f"   user_name : {user[1]}")

        # --------------------------------------------------
        # STEP 3. 테스트 Memory 저장
        # --------------------------------------------------

        print("\n[STEP 3] 테스트 Memory 저장")

        memory_id = save_memory(
            conn=conn,
            user_id=user_id,
            memory_type="test",
            content="Dual Write 테스트를 위한 임시 Memory입니다.",
            importance=1,
        )

        # --------------------------------------------------
        # STEP 4. 반환값 확인
        # --------------------------------------------------

        print("\n[STEP 4] 반환된 memory_id 확인")

        print(f"   반환된 memory_id : {memory_id}")
        print(f"   자료형           : {type(memory_id)}")

        # --------------------------------------------------
        # STEP 5. 최종 확인
        # --------------------------------------------------

        if memory_id is not None:
            print("\n🎉 테스트 성공!")
            print(
                "save_memory()가 "
                "PostgreSQL에 저장된 Memory의 "
                "memory_id를 정상적으로 반환했습니다."
            )
        else:
            print("\n❎ 테스트 실패")
            print("memory_id가 반환되지 않았습니다.")

    finally:

        conn.close()

        print("\n📴 PostgreSQL 연결 종료")


if __name__ == "__main__":
    main()

 

---

10. 테스트할 때 왜 "test" Memory를 하나 넣나요?

아주 중요한 질문입니다.

이번 테스트의 목적은:

save_memory()가 ID를 반환하는가?

를 확인하는 것입니다.

그래서 실제 사용자의 중요한 기억을 넣을 필요가 없습니다.

우리는 일부러:

memory_type = test
importance = 1
content = Dual Write 테스트를 위한 임시 Memory입니다.

 

라는 테스트 데이터를 하나 넣습니다.

---

11. 실행

PowerShell에서:

 

python test_save_memory.py

 

정상이라면 대략 다음처럼 나옵니다.

======================================================================
PostgreSQL save_memory() 반환값 테스트
======================================================================

[STEP 1] PostgreSQL 연결
☑️ PostgreSQL 연결 성공

[STEP 2] 테스트 사용자 조회
☑️ 사용자 발견
   user_id   : 8
   user_name : 스티븐김

[STEP 3] 테스트 Memory 저장
🧠기억 저장 완료 (memory_id=37)

[STEP 4] 반환된 memory_id 확인
   반환된 memory_id : 37
   자료형           : <class 'int'>

🎉 테스트 성공!
save_memory()가 PostgreSQL에 저장된 Memory의 memory_id를 정상적으로 반환했습니다.

📴 PostgreSQL 연결 종료

 

여기서 37이라는 숫자는 당연히 사용자 환경에 따라 달라집니다.

현재 마지막 Memory가 36이었으므로 보통 37이 될 가능성이 높습니다.

---

12. 그런데 테스트 후 DB에 테스트 Memory가 하나 생깁니다

이번에는:

memory_id=37
memory_type=test

 

같은 데이터가 들어갑니다.

이것은 정상입니다.

나중에 PostgreSQL을 확인할 때:

test
Dual Write 테스트를 위한 임시 Memory입니다.

 

라는 데이터가 하나 보일 것입니다.

지금은 삭제하지 마세요.

왜냐하면 다음 단계에서 이 Memory를 가지고 PostgreSQL → ChromaDB 연결을 설명할 수도 있기 때문입니다.

---

13. 이번 단계에서 반드시 확인해야 하는 것

테스트 결과에서 다음 세 가지를 봐주세요.

① 저장 성공

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

 

② 반환값 존재

반환된 memory_id : 37

 

③ 자료형

<class 'int'>

 

입니다.

특히 세 번째도 중요합니다.

우리가 ChromaDB ID를:

 

chroma_id = f"memory_{memory_id}"

 

처럼 만들 것이기 때문에 memory_id가 정상적인 숫자로 반환되는 것이 좋습니다.

---

14. 지금 전체 구조에서 어디까지 왔는가?

이번 수정까지 성공하면:

┌─────────────────────────────┐
│        PostgreSQL           │
│                             │
│  Memory 저장                │
│       ↓                     │
│  memory_id 생성             │
│       ↓                     │
│  Python으로 반환             │
└─────────────┬───────────────┘
              │
              │ memory_id
              ↓
       ┌──────────────┐
       │  다음 단계    │
       │  ChromaDB    │
       └──────────────┘

 

가 됩니다.

아직 ChromaDB에 넣지는 않습니다.

이번 단계에서는 PostgreSQL이 "내가 방금 저장한 Memory는 37번이다"라고 알려주는 것까지만 만드는 것입니다.

---

15. 그리고 이것이 왜 좋은 개발 방법인가?

사용자님이 지금 공부하고 있는 Agent 개발에서는 이 방식이 상당히 중요합니다.

우리가 한 번에:

database.py
memory.py
vector_store.py
agent_with_memory.py

 

를 모두 수정해 버리면 문제가 생겼을 때:

어느 파일에서 문제가 발생했는지?

찾기가 어려워집니다.

그래서 지금처럼:

① database.py 수정
      ↓
② database.py 단독 테스트
      ↓
③ 성공
      ↓
④ memory.py 수정
      ↓
⑤ memory.py 테스트
      ↓
⑥ 성공
      ↓
⑦ Agent 연결

 

방식으로 가는 것입니다.

이것이 실제 소프트웨어 개발에서도 상당히 중요한 incremental integration 방식입니다.

---

다음 단계가 무엇인지 미리 보면

이번 테스트가 성공하면 다음은 STEP 7-2입니다.

현재:

extract_memory()
       ↓
save_memories()
       ↓
PostgreSQL

 

을 다음처럼 바꿉니다.

extract_memory()
       ↓
save_memories()
       ↓
┌──────────────────────┐
│ PostgreSQL 저장       │
│ memory_id 확보        │
└──────────┬───────────┘
           ↓
저장된 Memory dictionary
           ↓
┌──────────────────────┐
│ ChromaDB 저장         │
└──────────────────────┘

 

그리고 그때 비로소 우리가 이미 만들어 놓은:

 

add_memories()

 

를 연결하게 됩니다.

---

지금은 여기까지만 해주세요

먼저 database.py의 save_memory()를 위 코드로 교체하고,

 

python test_save_memory.py

 

를 실행해 보세요.

 

[결과 예시]

======================================================================
PostgreSQL save_memory() 반환값 테스트
======================================================================

[STEP 1] PostgreSQL 연결
☑️ PostgreSQL 연결 성공

[STEP 2] 테스트 사용자 조회
☑️ 사용자 발견
   user_id   : 8
   user_name : 스티븐김

[STEP 3] 테스트 Memory 저장
🧠기억 저장 완료 (memory_id=37)

[STEP 4] 반환된 memory_id 확인
   반환된 memory_id : 37
   자료형           : <class 'int'>

🎉 테스트 성공!
save_memory()가 PostgreSQL에 저장된 Memory의 memory_id를 정상적으로 반환했습니다.

📴 PostgreSQL 연결 종료