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_id4. 여기서 한 줄씩 이해해 봅시다
① 함수
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 연결 종료