지금 단계는 단순히 "코드가 실행되느냐"가 아니라,
PostgreSQL에 저장된 Memory를 ChromaDB에도 정확히 전달할 수 있는 구조를 만드는 단계
라고 이해하면 좋습니다.
[요약]
# PostgreSQL의 memories 테이블과 ChromaDB연결 단계(4.2)
=> memory.py -> save_memories()가 저장된 Memory 정보를 반환하도록 수정
# step 1. memory.py 에서 save_memories()만 수정
# step 2. 기 만들어 놓은, test_save_memories.py 실행해서 테스트
1. 먼저 현재 memory.py의 전체 구조를 보겠습니다
현재 구조는 크게 이렇게 되어 있습니다.
memory.py
│
├─ 1. 환경 설정
│
├─ 2. OpenAI Client
│
├─ 3. extract_memory()
│ └─ 대화 → Memory 추출
│
├─ 4. save_memory()
│ └─ Memory 1개 → PostgreSQL 저장
│
├─ 5. save_memories()
│ └─ Memory 여러 개 → PostgreSQL 저장
│
├─ 6. search_memory()
│ └─ ChromaDB에서 의미 검색
│
├─ chroma_results_to_memories()
│ └─ Chroma 결과 → Python dictionary
│
├─ 7. memory_tuple_to_dict()
│ └─ PostgreSQL tuple → dictionary
│
├─ 8. memory_tuples_to_dicts()
│ └─ PostgreSQL tuple 목록 → dictionary 목록
│
└─ 9. build_memory_context()
└─ 검색 Memory → LLM Context이 구조 자체는 상당히 좋습니다.
특히 지금까지 진행하면서 데이터 형태가
PostgreSQL
↓
tuple
↓
dictionary
↓
ChromaDB
↓
dictionary
↓
LLM Context로 정리되고 있습니다.
2. 그런데 현재 memory.py에서 반드시 수정해야 할 것이 하나 있습니다
여기입니다.
현재 코드:
return {
"meomory_id":row[0],
"user_id":row[1],
"memory_type": row[2],
"content": row[3],
"importance": row[4],
"created_at": row[5],
}여기:
"meomory_id"가 오타입니다.
정확히는:
"memory_id"여야 합니다.
즉,
def memory_tuple_to_dict(row):
return {
"memory_id": row[0],
"user_id": row[1],
"memory_type": row[2],
"content": row[3],
"importance": row[4],
"created_at": row[5],
}로 수정해야 합니다.
왜 중요할까요?
ChromaDB에서 우리가 사용하는 핵심 ID가:
memory_id이기 때문입니다.
예를 들어 PostgreSQL에서:
memory_id = 37이면 ChromaDB에서는:
memory_37이라는 ID를 사용합니다.
따라서
"meomory_id"라는 잘못된 이름이 들어가면 나중에 이런 코드에서 문제가 생길 수 있습니다.
memory["memory_id"]즉, 지금 발견한 이 오타는 지금 바로 수정하는 것이 좋습니다.
3. 그리고 핵심 질문: save_memories()를 수정하는 것이 맞는가?
결론부터 말씀드리면:
네. 수정하는 것이 맞습니다.
현재 save_memories()는:
def save_memories(conn, user_id, memories):
saved_ids = []
for memory in memories:
memory_id = save_memory(
conn,
user_id,
memory
)
if memory_id is not None:
saved_ids.append(memory_id)
return saved_ids입니다.
그러면 결과가:
[38, 39]처럼 ID만 반환됩니다.
그런데 앞으로 ChromaDB에 저장하려면 이것만으로는 부족합니다.
ChromaDB의 add_memories()는 다음과 같은 정보가 필요합니다.
{
"memory_id": 38,
"user_id": 8,
"memory_type": "interest",
"content": "...",
"importance": 8
}즉,
ID만 있으면 부족합니다.
4. 그래서 우리가 방금 save_memories()를 바꾸려고 한 것입니다
우리가 원하는 결과는:
[
{
"memory_id": 38,
"user_id": 8,
"memory_type": "interest",
"content": "사용자는 AI Agent 개발에 관심이 있다.",
"importance": 8
},
{
"memory_id": 39,
"user_id": 8,
"memory_type": "goal",
"content": "사용자는 AI Agent 개발 역량을 높이려 한다.",
"importance": 9
}
]입니다.
이렇게 하면 다음 단계에서:
save_memories()
│
↓
PostgreSQL 저장
│
↓
saved_memories
│
↓
ChromaDB 저장이라는 흐름을 만들 수 있습니다.
5. 따라서 save_memory()는 현재 구조를 유지하는 것이 좋습니다
현재 save_memory()는 상당히 잘 되어 있습니다.
def save_memory(conn, user_id, memory):
memory_type = memory.get("memory_type","other")
content = memory.get("content")
importance = memory.get("importance", 5)그리고:
if not content:
print("⚠️ content가 없는 Memory는 저장하지 않습니다.")
return None좋습니다.
또:
try:
importance = int(importance)
except (TypeError, ValueError):
importance = 5
importance = max(1, min(10, importance))이것도 좋습니다.
LLM이 혹시:
"importance": 15라고 만들어도:
10으로 제한하고,
"importance": 0이면:
1로 보정합니다.
그리고 가장 중요한 부분:
memory_id = db_save_memory(
conn,
user_id=user_id,
memory_type=memory_type,
content=content,
importance=importance,
)여기서 database.py의 save_memory()가 실제 PostgreSQL의 memory_id를 반환하도록 우리가 앞에서 수정했습니다.
그래서:
return memory_id가 가능합니다.
이 부분은 그대로 유지하면 됩니다.
6. 이제 save_memories()만 수정합니다
현재:
def save_memories(conn, user_id, memories):
"""
여러 개의 Memory를 PostgreSQL에 저장한다.
반환값: 저장된 memory_id 목록
"""
saved_ids = []
for memory in memories:
memory_id = save_memory(conn, user_id, memory)
if memory_id is not None:
saved_ids.append(memory_id)
return saved_ids이 부분을 통째로 교체하는 것을 권합니다.
수정 코드
# ============================================================
# 5. Memory 여러 개 저장
# ============================================================
def save_memories(conn, user_id, memories):
"""
여러 개의 Memory를 PostgreSQL에 저장하고
실제로 저장된 Memory 정보를 반환한다.
반환값:
[
{
"memory_id": 35,
"user_id": 8,
"memory_type": "interest",
"content": "...",
"importance": 8,
},
...
]
"""
saved_memories = []
for memory in memories:
memory_type = memory.get(
"memory_type",
"other",
)
content = memory.get(
"content",
"",
)
importance = memory.get(
"importance",
5,
)
# ----------------------------------------------------
# PostgreSQL에 저장
# ----------------------------------------------------
memory_id = db_save_memory(
conn=conn,
user_id=user_id,
memory_type=memory_type,
content=content,
importance=importance,
)
# ----------------------------------------------------
# 실제 저장된 Memory 정보를 dictionary로 구성
# ----------------------------------------------------
saved_memories.append(
{
"memory_id": memory_id,
"user_id": user_id,
"memory_type": memory_type,
"content": content,
"importance": importance,
}
)
return saved_memories7. 그런데 여기서 한 가지 생각해볼 부분이 있습니다
혹시 이런 생각이 들 수 있습니다.
"이미 save_memory()가 있는데 왜 save_memories()에서 똑같이 memory_type, content, importance를 다시 꺼내지?"
맞습니다.
사실 현재 구조에서는 조금 중복됩니다.
우리가 이미:
save_memory()라는 함수를 가지고 있기 때문입니다.
그래서 저는 학습 단계에서는 조금 더 깔끔하게 다음처럼 만드는 것도 추천합니다.
def save_memories(conn, user_id, memories):
saved_memories = []
for memory in memories:
memory_id = save_memory(
conn,
user_id,
memory,
)
if memory_id is None:
continue
saved_memories.append(
{
"memory_id": memory_id,
"user_id": user_id,
"memory_type": memory.get(
"memory_type",
"other",
),
"content": memory.get(
"content",
"",
),
"importance": memory.get(
"importance",
5,
),
}
)
return saved_memories저라면 이쪽을 더 추천합니다.
왜냐하면 데이터 검증의 책임을 save_memory()에 계속 유지할 수 있기 때문입니다.
8. 그런데 왜 제가 앞서 제안한 코드에서는 db_save_memory()를 직접 호출했을까요?
여기에는 설계적인 이유가 있습니다.
현재 save_memory()는:
memory.py에 있고,
그 내부에서:
db_save_memory()를 호출합니다.
즉:
save_memories()
↓
save_memory()
↓
db_save_memory()
↓
PostgreSQL구조입니다.
이것이 더 자연스럽습니다.
따라서 지금 학습하고 있는 단계에서는 save_memories()가 save_memory()를 호출하는 구조를 유지하는 것이 더 좋습니다.
그러므로 최종적으로 저는 다음 버전을 추천합니다.
9. 제가 최종 추천하는 save_memories() 코드
# ============================================================
# 5. Memory 여러 개 저장
# ============================================================
def save_memories(conn, user_id, memories):
"""
여러 개의 Memory를 PostgreSQL에 저장하고
실제로 저장된 Memory 정보를 반환한다.
반환값:
[
{
"memory_id": 35,
"user_id": 8,
"memory_type": "interest",
"content": "...",
"importance": 8,
},
...
]
"""
saved_memories = []
for memory in memories:
# ----------------------------------------------------
# 1. 하나의 Memory를 PostgreSQL에 저장
# ----------------------------------------------------
memory_id = save_memory(
conn,
user_id,
memory,
)
# ----------------------------------------------------
# 2. 저장 실패한 Memory는 제외
# ----------------------------------------------------
if memory_id is None:
continue
# ----------------------------------------------------
# 3. 실제 저장된 Memory 정보를 dictionary로 구성
# ----------------------------------------------------
saved_memories.append(
{
"memory_id": memory_id,
"user_id": user_id,
"memory_type": memory.get(
"memory_type",
"other",
),
"content": memory.get(
"content",
"",
),
"importance": memory.get(
"importance",
5,
),
}
)
return saved_memories이렇게 하면 역할이 명확합니다.
save_memory()
Memory 1개를 검증하고 PostgreSQL에 저장
→ memory_id 반환save_memories()
Memory 여러 개를 반복해서 저장
→ 저장된 Memory 정보를 모아서 반환이렇게 되는 것입니다.
10. 그리고 memory_tuple_to_dict() 오타도 같이 수정합니다
현재:
def memory_tuple_to_dict(row):
return {
"meomory_id":row[0],
"user_id":row[1],
"memory_type": row[2],
"content": row[3],
"importance": row[4],
"created_at": row[5],
}다음으로 수정:
# ============================================================
# 7. Memory tuple → dictionary
# ============================================================
def memory_tuple_to_dict(row):
"""
PostgreSQL에서 가져온 Memory tuple을
dictionary로 변환한다.
PostgreSQL tuple 구조:
(
memory_id,
user_id,
memory_type,
content,
importance,
created_at
)
"""
return {
"memory_id": row[0],
"user_id": row[1],
"memory_type": row[2],
"content": row[3],
"importance": row[4],
"created_at": row[5],
}이것만 수정하면 됩니다.
11. 그러면 현재 memory.py의 중요한 데이터 흐름이 이렇게 됩니다
이 부분이 이번 단계에서 가장 중요합니다.
① LLM이 Memory를 추출
new_memories = extract_memory(conversation)결과:
[
{
"memory_type": "interest",
"content": "사용자는 AI Agent 개발에 관심이 있다.",
"importance": 8
}
]② save_memories() 호출
saved_memories = save_memories(
conn,
user_id,
new_memories,
)③ PostgreSQL에 저장
예를 들어 DB가:
memory_id = 40을 생성합니다.
④ save_memories()가 결과를 반환
[
{
"memory_id": 40,
"user_id": 8,
"memory_type": "interest",
"content": "사용자는 AI Agent 개발에 관심이 있다.",
"importance": 8
}
]12. 바로 여기까지가 이번 memory.py 수정의 목적입니다
여기서 굉장히 중요한 점이 있습니다.
이번 단계에서는 아직 ChromaDB에 저장하지 않습니다.
즉 memory.py의 책임은:
LLM Memory 추출
↓
PostgreSQL 저장
↓
저장된 Memory 정보를 반환까지만 합니다.
그 다음:
saved_memories
↓
ChromaDB는 agent_with_memory.py가 담당하게 합니다.
이것이 제가 지금 단계에서 추천하는 구조입니다.
13. 왜 memory.py에서 ChromaDB까지 저장하지 않을까요?
물론 기술적으로는 가능합니다.
예를 들어:
def save_memories(...):
PostgreSQL 저장
ChromaDB 저장으로 만들 수도 있습니다.
하지만 그러면 memory.py가 점점 많은 일을 담당하게 됩니다.
memory.py
├─ LLM
├─ PostgreSQL
├─ ChromaDB
├─ 검색
├─ Context
...나중에는 관리하기 어려워집니다.
지금 구조에서는:
database.py
↓
PostgreSQL 담당
vector_store.py
↓
ChromaDB 담당
memory.py
↓
Memory 의미/변환 담당
agent_with_memory.py
↓
전체 흐름을 연결로 나누는 것이 좋습니다.
이것을 관심사의 분리(Separation of Concerns)라고 생각하면 됩니다.
14. 현재 코드에서 get_memories는 어떻게 해야 할까요?
현재 상단에:
from database import (
save_memory as db_save_memory,
get_memories
)가 있습니다.
그런데 현재 search_memory()는 PostgreSQL의 get_memories()를 사용하지 않습니다.
현재:
def search_memory(query, user_id, limit=5):
results = search_similar_memories(
query=query,
user_id=user_id,
limit=limit,
)
memories = chroma_results_to_memories(results)
return memories이기 때문입니다.
따라서 지금은:
get_memories가 사용되지 않습니다.
하지만 당장 삭제하지 않는 것을 추천합니다.
왜냐하면 지금 우리가 PostgreSQL → ChromaDB 구조를 배우는 과정이고,
memory_tuples_to_dicts()도 PostgreSQL 데이터를 ChromaDB에 넣는 과정에서 사용할 수 있기 때문입니다.
조금 뒤 구조가 안정되면 정리해도 됩니다.
15. 또 하나: chroma_results_to_memories()는 현재 아주 잘 되어 있습니다
이 부분:
memories.append({
"memory_id": metadata["memory_id"],
"user_id": metadata["user_id"],
"memory_type": metadata.get("memory_type", "general"),
"content": documents[i],
"importance": metadata.get("importance", 5),
"distance": distances[i],
})좋습니다.
결과적으로 ChromaDB가 반환하는 복잡한 결과를 우리가 사용하기 좋은:
{
"memory_id": ...,
"user_id": ...,
"memory_type": ...,
"content": ...,
"importance": ...,
"distance": ...
}형태로 통일합니다.
그리고 build_memory_context()는:
memory["memory_id"]
memory["memory_type"]
memory["content"]
memory["importance"]를 사용합니다.
즉 현재:
ChromaDB
↓
chroma_results_to_memories()
↓
dictionary
↓
build_memory_context()가 정확히 연결되어 있습니다.
16. 따라서 지금 memory.py에서 실제로 수정할 것은 딱 2가지입니다
수정 ① save_memories()
기존:
saved_ids = []에서
saved_memories = []로 바꾸고, 실제 Memory dictionary를 반환합니다.
수정 ② 오타
"meomory_id"↓
"memory_id"17. 나머지는 지금 건드리지 않는 것을 추천합니다
특히 다음은 그대로 둡니다.
extract_memory()그대로.
save_memory()그대로.
chroma_results_to_memories()그대로.
search_memory()그대로.
build_memory_context()그대로.
즉 이번 단계에서는 두 군데만 수정합니다.
18. 수정 후 전체적인 구조
수정 후 memory.py의 핵심 흐름은:
┌─────────────────┐
│ extract_memory │
│ LLM Memory │
│ 추출 │
└────────┬────────┘
│
▼
┌─────────────────┐
│ save_memories │
└────────┬────────┘
│
┌───────────┴───────────┐
▼ ▼
┌────────────────┐ saved_memories
│ save_memory │ │
└───────┬────────┘ │
▼ │
┌────────────────┐ │
│ PostgreSQL │ │
│ Source of │ │
│ Truth │ │
└────────────────┘ │
│
▼
[다음 단계]
ChromaDB 저장그리고 검색은 반대 방향입니다.
사용자 질문
│
▼
search_memory()
│
▼
ChromaDB
│
▼
의미적으로 가까운 Memory
│
▼
chroma_results_to_memories()
│
▼
dictionary
│
▼
build_memory_context()
│
▼
LLM이제 구조가 상당히 명확해졌습니다.
19. 지금은 바로 테스트하는 것이 좋습니다
따라서 지금은 제가 앞서 드린 test_save_memories.py를 실행하는 순서가 맞습니다.
수정 후:
python test_save_memories.py를 실행하세요.
정상이라면 대략:
[STEP 4] save_memories() 실행
🧠기억 저장 완료 (memory_id=38)
🧠기억 저장 완료 (memory_id=39)
[STEP 5] 저장 결과 확인
[1]
memory_id : 38
user_id : 8
memory_type : test
content : save_memories() 첫 번째 테스트 Memory입니다.
importance : 1
[2]
memory_id : 39
user_id : 8
memory_type : test
content : save_memories() 두 번째 테스트 Memory입니다.
importance : 2
🎉 테스트 성공!처럼 나와야 합니다.
여기서 중요한 것은 단순히 "성공"이라는 문구가 나오는 것이 아니라, saved_memories가 실제 dictionary 목록으로 반환되는지를 확인하는 것입니다.
그리고 다음 단계의 의미
이 테스트가 성공하면 이제 상당히 중요한 지점에 도달합니다.
지금까지는:
PostgreSQL ← Memory 저장까지만 되어 있었습니다.
다음 단계에서는:
새 Memory
↓
PostgreSQL
↓
ChromaDB라는 Dual Write를 완성하게 됩니다.
즉, 우리가 처음 설계했던
PostgreSQL은 원본(Source of Truth), ChromaDB는 검색용 Index
라는 구조가 실제 Agent 안에서 완성되기 시작합니다.
저는 다음 단계에서 agent_with_memory.py까지 연결한 뒤, PostgreSQL과 ChromaDB의 memory_id가 실제로 동일하게 유지되는지 확인하는 통합 테스트까지 이어가고 싶습니다.
[결과 예시]
======================================================================
memory.py save_memories() 테스트
======================================================================
[STEP 1] PostgreSQL 연결
☑️ PostgreSQL 연결 성공
[STEP 2] 테스트 사용자 조회
☑️ 사용자 발견
user_id : 8
user_name : 스티븐김
[STEP 3] 테스트 Memory 준비
테스트 Memory 개수 : 2
[STEP 4] save_memories() 실행
🧠기억 저장 완료 (memory_id=39)
🧠기억 저장 완료 (memory_id=40)
[STEP 5] 저장 결과 확인
----------------------------------------------------------------------
[1]
memory_id : 39
user_id : 8
memory_type : test
content : save_memories() 첫 번째 테스트 Memory입니다.
importance : 1
----------------------------------------------------------------------
[2]
memory_id : 40
user_id : 8
memory_type : test
content : save_memories() 두 번째 테스트 Memory입니다.
importance : 2
----------------------------------------------------------------------
🎉 테스트 성공!
save_memories()가 모든 Memory를 PostgreSQL에 저장하고 memory_id를 포함한 결과를 정상적으로 반환했습니다.
📴 PostgreSQL 연결 종료
'Agentic AI > Memory' 카테고리의 다른 글
| 17. 중간 점검(2) (0) | 2026.09.12 |
|---|---|
| 16. 장기기억(12)- PostgreSQL의 memories 테이블과 ChromaDB연결 단계(4.3) (0) | 2026.09.12 |
| 14. 장기기억(10) - PostgreSQL의 memories 테이블과 ChromaDB연결 단계(4.1) (0) | 2026.09.12 |
| 13. 중간 점검(1) (0) | 2026.09.12 |
| 12. 장기기억(9) - PostgreSQL의 memories 테이블과 ChromaDB연결 단계(3) (0) | 2026.09.12 |
