Agentic AI/Memory
10. 장기기억(7)- PostgreSQL의 memories 테이블과 ChromaDB연결 단계(1)
아톨
2026. 9. 12. 16:03
반응형
지금은 단순히 코드를 조금 고치는 단계가 아니라, PostgreSQL과 ChromaDB를 처음으로 실제 연결하는 단계이므로 한 번에 Agent까지 바꾸지 않는 것이 좋습니다.
[요약]
# PostgreSQL의 memories 테이블과 ChromaDB연결 단계(1)
===> memory.py, vector_store.py 수정/보완하고 테스트하기
# step1. memory.py 수정
# step1.1. 7. Memory tuple → dictionary 과 8. Memory tuple 목록 → dictionary 목록 추가.
# step1.2. 기존의 7. Memory Context 생성 => 9.Memory Context 생성로 변경하고 코드 수정
# step2. vector_store.py를 보완(1)(2)(3)(4)
# step3. test_postgres_to_chroma.py를 만들어서 테스트
===> cd C:\Dev\book_agentic_ai\agent_projec폴더에서
===> python test_postgres_to_chroma.py
이번 단계의 목표를 먼저 명확하게 잡겠습니다.
이번 단계의 목표:
PostgreSQL에 이미 저장되어 있는 Memory를 가져와서 → ChromaDB에 정상적으로 저장하고 → 의미 검색까지 해본다.
Agent는 이번 단계에서는 아직 건드리지 않습니다.
전체 진행 그림
우리가 지금부터 만드는 것은 다음입니다.
PostgreSQL
장기기억 원본
│
│ get_memories()
↓
Memory tuple
│
│ 변환
↓
Memory dictionary
│
│ add_memories()
↓
ChromaDB
의미 기반 검색
│
│ query
↓
관련 Memory 반환그리고 이번 단계를 끝내면 다음 단계에서:
사용자 질문
↓
ChromaDB 의미 검색
↓
관련 Memory
↓
LLM
↓
답변으로 연결할 수 있습니다.
먼저 반드시 고칠 것: memory.py
현재 가장 중요한 오류는 제가 앞서 말씀드린 tuple 인덱스 문제입니다.
database.py의 get_memories()는 현재:
SELECT
memory_id,
user_id,
memory_type,
content,
importance,
created_at순서입니다.
따라서 하나의 memory는:
[0] memory_id
[1] user_id
[2] memory_type
[3] content
[4] importance
[5] created_at입니다.
그런데 현재 build_memory_context()는 예전 구조를 기준으로 작성되어 있습니다.
이것부터 수정하겠습니다.
① memory.py 수정
이번에는 memory.py 전체를 아래 코드로 교체하는 것을 권합니다.
# ============================================================
# memory.py : Long-term Memory 관리 모듈
# 기능
# 1. extract_memory() - 대화에서 장기기억할 정보를 LLM에게 추출시킨다.
# 2. save_memory() - 추출된 기억을 PostgreSQL에 저장한다.
# 3. save_memories() - 여러 개의 기억을 PostgreSQL에 저장한다.
# 4. search_memory() - 특정 사용자의 기억을 검색한다.
# 5. build_memory_context() - 검색된 기억을 Agent가 사용할 문자열로 만든다.
# ============================================================
import os
import json
from dotenv import load_dotenv
from openai import OpenAI
from database import (
save_memory as db_save_memory,
get_memories
)
# ============================================================
# 1. 환경 설정
# ============================================================
load_dotenv()
MODEL_NAME = os.getenv(
"MODEL_NAME",
"gpt-5.4-mini"
)
# ============================================================
# 2. OpenAI Client
# ============================================================
client = OpenAI()
# ============================================================
# 3. Memory 추출
# ============================================================
def extract_memory(conversation):
"""
대화 내용에서 장기적으로 기억할 가치가 있는 정보를 LLM을 이용하여 추출한다.
반환값:
[
{
"memory_type":"interest",
"content":"사용자는 Python과 AI Agent에 관심이 있다.",
"importance":8
}
]
기억할 내용이 없으면 []을 반환한다.
"""
# --------------------------------------------------------
# LLM에게 전달할 Prompt
# --------------------------------------------------------
System_prompt = """
당신은 AI Agent의 장기기억을 관리하는 Memory Manager입니다.
사용자의 대화에서 앞으로의 대화에도 도움이 될 만한 중요한 정보를 찾아내세요.
특히, 다음 정보를 장기기억으로 저장할 가치가 있습니다.
1. 사용자의 지속적인 관심사
2. 사용자의 목표
3. 사용자의 장기적인 계획
4. 사용자의 선호
5. 사용자의 작업 방식
6. 반복적으로 도움이 될 개인 설정
7. 현재 진행중인 중요한 프로젝트
반대로, 다음 정보는 저장하지 마세요.
1. 단순한 인사
2. 일회성 질문
3. 중요하지 않은 잡담
4. 단순한 현재 상황
5. 미래 대화에 도움이 되지 않는 정보
각 기억에는 다음 정보를 포함하세요.
memory_type:
- interest
- goal
- preference
- fact
- other
importance: 1~10 사이의 숫자
반드시 JSON 배열만 반환하세요.
예:
[
{
"memory_type":"interest",
"content":"사용자는 Python과 AI Agent 개발에 관심이 있다.",
"importance":8
}
]
기억할 내용이 없다면: []
"""
# --------------------------------------------------------
# LLM 호출
# --------------------------------------------------------
response = client.responses.create(
model=MODEL_NAME,
input=[
{
"role":"system",
"content":System_prompt,
},
{
"role":"user",
"content":conversation,
}
]
)
# --------------------------------------------------------
# LLM 결과
# --------------------------------------------------------
result = response.output_text.strip()
print("\n")
print("="*70)
print("🧠 LLM Memory Extraction 결과")
print("="*70)
print(result)
# --------------------------------------------------------
# JSON 반환
# --------------------------------------------------------
try:
memories = json.loads(result)
except json.JSONDecodeError:
print("❎ LLM 결과를 JSON으로 변환하지 못했습니다.")
return []
# --------------------------------------------------------
# 결과 검증
# --------------------------------------------------------
if not isinstance(memories, list):
print("❎ Memory 결과가 리스트가 아닙니다.")
return []
return memories
# ============================================================
# 4. Memory 하나 저장
# ============================================================
def save_memory(conn, user_id, memory):
"""extract_memory()가 생성한 하나의 기억을 PostgreSQL에 저장한다."""
memory_type = memory.get("memory_type","other")
content = memory.get("content")
importance = memory.get("importance", 5)
# --------------------------------------------------------
# 기본 검증
# --------------------------------------------------------
if not content:
print("⚠️ content가 없는 Memory는 저장하지 않습니다.")
return None
# --------------------------------------------------------
# importance 범위 보정
# --------------------------------------------------------
try:
importance = int(importance)
except (TypeError, ValueError):
importance = 5
importance = max(1, min(10, importance))
# --------------------------------------------------------
# database.py의 save_memory() 호출
# --------------------------------------------------------
memory_id = db_save_memory(
conn,
user_id=user_id,
memory_type=memory_type,
content=content,
importance=importance,
)
return memory_id
# ============================================================
# 5. Memory 여러 개 저장
# ============================================================
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
# ============================================================
# 6. Memory 검색
# ============================================================
def search_memory(conn, user_id, limit=10):
"""
특정 사용자의 장기기억을 검색한다.
현재 단계에서는 PostgreSQL에서 중요도가 높은 기억부터 가져온다.
※ 다음 단계에서 이 부분을 ChromaDB 의미 검색 방식으로 교체할 예정이다.
"""
memories = get_memories(
conn,
user_id=user_id,
limit=limit,
)
return memories
# PostgreSQL의 memories 테이블과 ChromaDB연결 단계(1)
# step1. memory.py 수정
# step1.1. 7. Memory tuple → dictionary 과 8. Memory tuple 목록 → dictionary 목록 추가.
# ============================================================
# 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
)
반환:
{
"memory_id": ...,
"user_id": ...,
"memory_type": ...,
"content": ...,
"importance": ...,
"created_at": ...
}
"""
return {
"meomory_id":row[0],
"user_id":row[1],
"memory_type": row[2],
"content": row[3],
"importance": row[4],
"created_at": row[5],
}
# ============================================================
# 8. Memory tuple 목록 → dictionary 목록
# ============================================================
def memory_tuples_to_dicts(rows):
"""
PostgreSQL에서 가져온 여러 Memory tuple을
dictionary 목록으로 변환한다.
"""
return [
memory_tuple_to_dict(row)
for row in rows
]
# PostgreSQL의 memories 테이블과 ChromaDB연결 단계(1)
# step1. memory.py 수정
# step1.2. 기존의 7. Memory Context 생성 => 9.Memory Context 생성로 변경하고 코드 수정
# ============================================================
# 9. Memory Context 생성
# ============================================================
# def build_memory_context(memories):
# """PostgreSQL에서 가져온 Memory를 LLM에게 전달하기 좋은 문자열로 변환한다."""
# if not memories:
# return ("현재 저장되어 있는 장기기억이 없습니다.")
# lines = []
# lines.append("다음은 사용자의 장기기억입니다.")
# lines.append("답변에 도움이 될 경우 자연스럽게 활용하세요.")
# lines.append("")
# for memory in memories:
# memory_id = memory[0]
# memory_type = memory[1]
# content = memory[2]
# importance = memory[3]
# created_at = memory[4]
# lines.append(
# f"- [{memory_type}]"
# f"{content}"
# f"(중요도: {importance})"
# )
# return "\n".join(lines)
def build_memory_context(memories):
"""
PostgreSQL에서 가져온 Memory를
LLM에게 전달하기 좋은 문자열로 변환한다.
현재 Memory tuple 구조:
[0] memory_id
[1] user_id
[2] memory_type
[3] content
[4] importance
[5] created_at
"""
# --------------------------------------------------------
# Memory가 없는 경우
# --------------------------------------------------------
if not memories:
return (
"현재 저장되어 있는 장기기억이 없습니다."
)
# --------------------------------------------------------
# Context 생성
# --------------------------------------------------------
lines = []
lines.append("다음은 사용자의 장기기억입니다.")
lines.append("답변에 도움이 될 경우 자연스럽게 활용하세요.")
lines.append("")
# --------------------------------------------------------
# 각각의 Memory 처리
# --------------------------------------------------------
for memory in memories:
memory_id = memory[0]
user_id = memory[1]
memory_type = memory[2]
content = memory[3]
importance = memory[4]
created_at = memory[5]
lines.append(
f"- [{memory_type}] "
f"{content} "
f"(중요도: {importance})"
)
return "\n".join(lines)
①-1. 이번 수정에서 무엇이 달라졌는가?
중요한 것만 짚어보겠습니다.
첫 번째
기존:
System_prompt = """수정:
system_prompt = """파이썬에서는 변수명에 대문자/소문자를 자유롭게 사용할 수 있지만, 일반적으로 지역 변수는 snake_case로 쓰는 것이 좋습니다.
두 번째 — 가장 중요
기존에는:
memory_id = memory[0]
memory_type = memory[1]
content = memory[2]
importance = memory[3]
created_at = memory[4]였습니다.
이것을:
memory_id = memory[0]
user_id = memory[1]
memory_type = memory[2]
content = memory[3]
importance = memory[4]
created_at = memory[5]로 수정했습니다.
이것이 필수 수정사항입니다.
①-2. 그리고 memory.py에 변환 함수를 넣은 이유
이번에 추가한:
def memory_tuple_to_dict(row):와
def memory_tuples_to_dicts(rows):는 앞으로 상당히 중요하게 사용합니다.
PostgreSQL에서:
tuple형태로 나오는 것을
dictionary로 바꾸는 겁니다.
예를 들어 DB에서:
(
7,
1,
"interest",
"사용자는 Python과 AI Agent를 학습하고 있다.",
8,
datetime(...)
)가 나오면:
{
"memory_id": 7,
"user_id": 1,
"memory_type": "interest",
"content": "사용자는 Python과 AI Agent를 학습하고 있다.",
"importance": 8,
"created_at": datetime(...)
}가 됩니다.
이렇게 하면 이후 ChromaDB에 넣기가 훨씬 쉬워집니다.
② 이제 vector_store.py를 보완합니다
현재 vector_store.py는 기본적인 기능이 잘 되어 있습니다.
다만 이번에는 PostgreSQL에서 가져온 Memory를 ChromaDB에 넣는 역할을 명확하게 추가하겠습니다.
그리고 한 가지 원칙을 지키겠습니다.
PostgreSQL의 memory_id를 ChromaDB의 ID로 그대로 연결한다.
예를 들어:
PostgreSQL
memory_id = 17이면 ChromaDB:
memory_17이 됩니다.
그러면 나중에:
ChromaDB 검색 결과
↓
memory_17
↓
PostgreSQL memory_id = 17로 원본을 찾을 수 있습니다.
vector_store.py 전체 교체
# ============================================================
# vector_store.py
# PostgreSQL의 기억을 ChromaDB에 저장하고 검색하는 모듈
# 역할: PostgreSQL -> Memory 원본 -> ChromaDB -> 의미 기반 검색
# ============================================================
from pathlib import Path
import chromadb
# ============================================================
# 1. ChromaDB 설정
# ============================================================
CHROMA_DIR = Path("./chroma_db")
COLLECTION_NAME = "memories"
# ============================================================
# 2. ChromaDB 컬렉션 가져오기
# ============================================================
def get_collection():
"""ChromaDB의 memories 컬렉션을 가져온다. 컬렉션이 없으면 새로 생성한다."""
client = chromadb.PersistentClient(path=str(CHROMA_DIR))
collection = client.get_or_create_collection(name=COLLECTION_NAME)
return collection
# ============================================================
# test_vector_store.py 만들고 수정(3): PostgreSQL Memory tuple → dictionary
# ============================================================
def memory_tuple_to_dict(row):
"""
PostgreSQL에서 조회한 기억 tuple을 ChromaDB 저장용 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],
}
# ============================================================
# 3. 기억 하나를 ChromaDB에 저장
# ============================================================
def add_memory(
memory_id,
user_id,
content,
memory_type=None,
importance=5
):
"""
기억 하나를 ChromaDB에 저장한다.
Parameters
----------
memory_id: int
PostgreSQL memories 테이블의 memory_id
user_id: int
PostgreSQL user_profiles 테이블의 user_id
content: str
실제 기억 내용
memory_type: str
기억 유형
importance: int
기억 중요도
"""
collection = get_collection()
chroma_id = f"memory_{memory_id}"
collection.upsert(
ids=[chroma_id],
documents=[content],
metadatas=[
{
"memory_id":memory_id,
"user_id":user_id,
"memory_type":memory_type or "general",
"importance":importance
}
]
)
print(f"☑️ChromaDB 저장 완료: {chroma_id}")
# ============================================================
# 4. 여러 기억을 ChromaDB에 저장
# ============================================================
def add_memories(memories):
"""
여러 기억을 한번에 ChromaDB에 저장한다.
memories는 다음과 같은 dictionary 목록을 기대한다.
[
{
"memory_id": 1,
"user_id": 1,
"content": "...",
"memory_type": "goal",
"importance": 8
}
]
"""
if not memories:
print(
"⚠️ ChromaDB에 저장할 Memory가 없습니다."
)
return
collection = get_collection()
ids=[]
documents=[]
metadatas=[]
# --------------------------------------------------------
# Memory 하나씩 변환
# --------------------------------------------------------
for memory in memories:
memory_id = memory["memory_id"]
##### # step2. vector_store.py를 보완(1)
user_id = memory["user_id"]
content = memory["content"]
memory_type = memory.get("memory_type", "general")
importance = memory.get("importance", 5)
########################### 여기까지
# ----------------------------------------------------
# ChromaDB ID
# ----------------------------------------------------
ids.append(f"memory_{memory_id}")
# ----------------------------------------------------
# 실제 검색 대상 문서
# ----------------------------------------------------
# documents.append(memory["content"])
##### # step2. vector_store.py를 보완(2) - 위 내용을 아래로 수정
documents.append(content)
########################### 여기까지
# ----------------------------------------------------
# 검색 필터 등에 사용할 Metadata
# ----------------------------------------------------
# metadatas.append(
# {
# "memory_id":memory_id,
# "user_id":memory["user_id"],
# "memory_type":memory.get("memory_type", "general"),
# "importance":memory.get("importance", 5)
# }
# )
##### # step2. vector_store.py를 보완(3) - 위 내용을 아래로 수정
metadatas.append(
{
"memory_id": memory_id,
"user_id": user_id,
"memory_type": memory_type,
"importance": importance,
}
)
########################### 여기까지
# --------------------------------------------------------
# ChromaDB에 한꺼번에 저장
# --------------------------------------------------------
collection.upsert(
ids=ids,
documents=documents,
metadatas=metadatas
)
print(f"☑️ChromaDB에 {len(memories)}개 기억 저장 완료")
##### # step2. vector_store.py를 보완(4) - PostgreSQL Memory를 ChromaDB로 동기화
# ============================================================
# PostgreSQL Memory를 ChromaDB로 동기화
# ============================================================
def sync_memories_to_chroma(memories):
"""
PostgreSQL에서 가져온 Memory 목록을 ChromaDB로 동기화한다.
memories는 PostgreSQL의 get_memories() 반환값인 tuple 목록을 받는다.
처리 과정:
PostgreSQL tuple
↓
dictionary 변환
↓
ChromaDB 저장
"""
# --------------------------------------------------------
# Memory가 없는 경우
# --------------------------------------------------------
if not memories:
print("⚠️ 동기화할 Memory가 없습니다.")
return
# --------------------------------------------------------
# tuple → dictionary
# --------------------------------------------------------
memory_dicts = [
memory_tuple_to_dict(row)
for row in memories
]
# --------------------------------------------------------
# ChromaDB 저장
# --------------------------------------------------------
add_memories(memory_dicts)
#################여기까지
# ============================================================
# 5. 의미 기반 기억 검색
# ============================================================
def search_similar_memories(
query,
user_id=None,
limit=5
):
"""
query와 의미가 비슷한 기억을 검색한다.
user_id가 지정되면 해당 사용자의 기억만 검색한다.
Parameters
----------
query : str
검색할 문장
user_id : int, optional
사용자 ID
limit : int
검색 결과 개수
"""
collection = get_collection()
# --------------------------------------------------------
# 기본 검색 조건
# --------------------------------------------------------
query_kwargs = {
"query_texts":[query],
"n_results":limit
}
# --------------------------------------------------------
# user_id 필터
# --------------------------------------------------------
if user_id is not None:
query_kwargs["where"] = {"user_id":user_id}
# --------------------------------------------------------
# ChromaDB 검색
# --------------------------------------------------------
results = collection.query(**query_kwargs)
return results
②-1. 이번에 sync_memories_to_chroma()를 만든 이유
이 함수가 이번 단계의 핵심입니다.
def sync_memories_to_chroma(memories):사용하는 방법은:
memories = get_memories(...)
sync_memories_to_chroma(memories)입니다.
그러면 내부적으로:
PostgreSQL
↓
get_memories()
↓
[
tuple,
tuple,
tuple
]
↓
memory_tuple_to_dict()
↓
[
dict,
dict,
dict
]
↓
add_memories()
↓
ChromaDB가 됩니다.
②-2. created_at을 ChromaDB metadata에 넣지 않은 이유
현재 PostgreSQL에는:
created_at이 있습니다.
그런데 이번 단계에서는 ChromaDB metadata에 넣지 않았습니다.
현재:
metadatas.append(
{
"memory_id": memory_id,
"user_id": user_id,
"memory_type": memory_type,
"importance": importance,
}
)까지만 합니다.
이유는 간단합니다.
created_at은 PostgreSQL의 원본 데이터에서 관리하는 것이 더 좋기 때문입니다.
그리고 ChromaDB에서는 일단:
memory_id
user_id
memory_type
importance정도만 있으면 검색과 필터링 학습에 충분합니다.
③ 이제 테스트 코드를 만듭니다
여기가 매우 중요합니다.
Agent를 실행하지 않습니다.
먼저 독립적인 테스트를 만들어서:
PostgreSQL → ChromaDB
가 정말 되는지를 확인합니다.
파일 이름은:
test_postgres_to_chroma.py로 하겠습니다.
현재:
agent_project/밑에 새로 만드세요.
agent_project/
│
├─ database.py
├─ memory.py
├─ vector_store.py
├─ agent_with_memory.py
├─ test_chroma.py
├─ test_postgres_to_chroma.py ← 새 파일
│
├─ chroma_db/
│
└─ .envtest_postgres_to_chroma.py
전체 코드는 다음과 같습니다.
# ============================================================
# test_postgres_to_chroma.py: PostgreSQL → ChromaDB 연동 테스트
#
# 테스트 흐름:#
# 1. PostgreSQL 연결
# 2. 테스트 사용자 찾기
# 3. PostgreSQL Memory 조회
# 4. Memory 출력
# 5. PostgreSQL Memory → ChromaDB 동기화
# 6. ChromaDB에 저장된 Memory 개수 확인
# 7. 의미 검색 테스트
# ============================================================
from database import (
connect_db,
get_user_by_name,
get_memories,
)
from vector_store import (
get_collection,
sync_memories_to_chroma,
search_similar_memories,
)
# ============================================================
# 1. 설정
# ============================================================
TEST_USER_NAME = "스티븐김"
# ============================================================
# 2. 메인 함수
# ============================================================
def main():
print("=" * 70)
print("PostgreSQL → ChromaDB 연동 테스트")
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]
user_name = user[1]
location = user[2]
preferences = user[3]
print(f"☑️ 사용자 발견")
print(f" user_id : {user_id}")
print(f" user_name : {user_name}")
print(f" location : {location}")
print(f" preferences : {preferences}")
# ====================================================
# STEP 3. PostgreSQL Memory 조회
# ====================================================
print("\n[STEP 3] PostgreSQL Memory 조회")
memories = get_memories(
conn,
user_id=user_id,
limit=20,
)
# ----------------------------------------------------
# Memory가 없는 경우
# ----------------------------------------------------
if not memories:
print("⚠️ PostgreSQL에 저장된 Memory가 없습니다.")
return
# ----------------------------------------------------
# Memory 출력
# ----------------------------------------------------
print(
f"☑️ PostgreSQL에서 "
f"{len(memories)}개 Memory 발견"
)
print()
for memory in memories:
print(f"memory_id : {memory[0]}")
print(f"user_id : {memory[1]}")
print(f"memory_type : {memory[2]}")
print(f"content : {memory[3]}")
print(f"importance : {memory[4]}")
print(f"created_at : {memory[5]}")
print("-" * 50)
# ====================================================
# STEP 4. PostgreSQL → ChromaDB 동기화
# ====================================================
print("\n[STEP 4] PostgreSQL → ChromaDB 동기화")
sync_memories_to_chroma(memories)
# ====================================================
# STEP 5. ChromaDB 상태 확인
# ====================================================
print("\n[STEP 5] ChromaDB 상태 확인")
collection = get_collection()
count = collection.count()
print(
f"☑️ ChromaDB 저장된 전체 Memory 개수: "
f"{count}"
)
# ====================================================
# STEP 6. ChromaDB 의미 검색
# ====================================================
print("\n[STEP 6] ChromaDB 의미 검색")
query = (
"나는 파이썬과 AI 에이전트 개발을 "
"공부하고 있다."
)
print(f"검색 질문: {query}")
results = search_similar_memories(
query=query,
user_id=user_id,
limit=5,
)
# ====================================================
# STEP 7. 검색 결과 출력
# ====================================================
print("\n[STEP 7] 검색 결과")
print("-" * 70)
print("IDs: ")
print(results["ids"])
print()
print("Documents:")
print(results["documents"])
print()
print("Metadatas:")
print(results["metadatas"])
print()
print("Distances:")
print(results["distances"])
print("-" * 70)
# ====================================================
# 완료
# ====================================================
print("\n🎉 PostgreSQL → ChromaDB 테스트 완료")
finally:
# ====================================================
# PostgreSQL 연결 종료
# ====================================================
conn.close()
print("\n📴 PostgreSQL 연결 종료")
# ============================================================
# 프로그램 시작점
# ============================================================
if __name__ == "__main__":
main()
③-1. 이 테스트에서 중요한 부분
이 테스트는 Agent가 아닙니다.
이 점을 꼭 이해하시면 좋겠습니다.
현재는:
Agent를 테스트하는 것이 아니라
데이터 파이프라인을 테스트하는 것입니다.
즉:
PostgreSQL
↓
Memory 가져오기
↓
Dictionary 변환
↓
ChromaDB 저장
↓
Vector Search만 검증합니다.
④ 이제 실행합니다
프로젝트 폴더에서:
cd C:\Dev\book_agentic_ai\agent_project그리고:
python test_postgres_to_chroma.py를 실행하세요.
④-1. 정상적으로 나온다면
예를 들어 PostgreSQL에 다음 Memory가 있다면:
memory_id : 1
user_id : 1
memory_type : interest
content : 사용자는 Python과 AI Agent를 학습하고 있다.
importance : 8다음과 비슷한 결과가 나와야 합니다.
======================================================================
PostgreSQL → ChromaDB 연동 테스트
======================================================================
[STEP 1] PostgreSQL 연결
☑️ PostgreSQL 연결 성공
[STEP 2] 테스트 사용자 조회
☑️ 사용자 발견
user_id : 1
user_name : 스티븐김
location : 대한민국
preferences : 저작권 안내 자동전화AI
[STEP 3] PostgreSQL Memory 조회
☑️ PostgreSQL에서 3개 Memory 발견
memory_id : 3
user_id : 1
memory_type : interest
content : 사용자는 Python과 AI Agent를 학습하고 있다.
importance : 8
created_at : ...
--------------------------------------------------
[STEP 4] PostgreSQL → ChromaDB 동기화
☑️ ChromaDB에 3개 기억 저장 완료
[STEP 5] ChromaDB 상태 확인
☑️ ChromaDB 저장된 전체 Memory 개수: 6
[STEP 6] ChromaDB 의미 검색
검색 질문:
나는 파이썬과 AI 에이전트 개발을 공부하고 있다.
[STEP 7] 검색 결과
IDs:
[['memory_3', 'memory_1', ...]]
Documents:
[['사용자는 Python과 AI Agent를 학습하고 있다.',
...]]
Metadatas:
[[{'memory_id': 3,
'user_id': 1,
'memory_type': 'interest',
'importance': 8},
...]]
Distances:
[[0.2..., 0.7..., ...]]
----------------------------------------------------------------------
🎉 PostgreSQL → ChromaDB 테스트 완료이런 식으로 나오면 핵심 연결이 성공한 것입니다.
그런데 한 가지 주의할 점
현재 test_chroma.py에서 이미:
collection.upsert(
ids=["memory_1", "memory_2", "memory_3"],를 실행했습니다.
따라서 현재 chroma_db에는 테스트용 Memory가 이미 있을 가능성이 높습니다.
즉:
ChromaDB count = PostgreSQL Memory 개수가 반드시 같지는 않습니다.
예를 들어:
기존 테스트 데이터 3개
+
PostgreSQL Memory 5개
=
8개가 될 수 있습니다.
이건 오류가 아닙니다.
오히려 여기서 아주 중요한 개념이 하나 있습니다
우리가 upsert()를 사용하는 이유가 바로 이것입니다.
예를 들어 PostgreSQL에:
memory_id = 5가 있다면 ChromaDB ID는:
memory_5입니다.
첫 번째 실행:
collection.upsert(
ids=["memory_5"],
...
)→ 새로 생성
두 번째 실행:
collection.upsert(
ids=["memory_5"],
...
)→ 같은 ID이므로 새로운 중복 Memory가 생기는 것이 아니라 기존 값을 업데이트
합니다.
따라서 테스트를 여러 번 실행해도:
memory_5
memory_5
memory_5
memory_5처럼 계속 쌓이지 않습니다.
이것은 앞으로 Memory 시스템을 만들 때 굉장히 중요한 특징입니다.
지금 단계에서 test_chroma.py는 어떻게 할까?
현재 test_chroma.py는 버리지 마세요.
그 파일은 ChromaDB 자체가 제대로 설치되고 동작하는지 확인하는 기초 테스트입니다.
현재:
test_chroma.py는:
"ChromaDB 자체 테스트"
이고,
이번에 만든:
test_postgres_to_chroma.py는:
"PostgreSQL과 ChromaDB의 연결 테스트"
입니다.
둘은 역할이 다릅니다.
지금까지의 구조를 다시 정리하면
database.py
PostgreSQL 담당사용자
Memory
저장
조회memory.py
Memory Logic 담당LLM
↓
Memory 추출
↓
Memory 저장
↓
Memory Context그리고 이번에:
tuple → dictionary변환 기능을 추가했습니다.
vector_store.py
ChromaDB 담당Memory 저장
Memory 검색
user_id 필터그리고 이번에:
PostgreSQL Memory
↓
sync_memories_to_chroma()
↓
ChromaDB를 추가했습니다.
test_chroma.py
ChromaDB 자체 테스트test_postgres_to_chroma.py
PostgreSQL
↓
ChromaDB통합 테스트입니다.
그런데 아직 Agent는 그대로입니다
현재 agent_with_memory.py는 여전히:
memories = search_memory(
conn,
user_id=user_id,
limit=10,
)을 사용합니다.
즉 아직은:
사용자 질문
↓
PostgreSQL에서 중요도 높은 Memory 10개
↓
LLM입니다.
이번 단계에서는 이것을 아직 바꾸지 않습니다.
왜냐하면 먼저:
PostgreSQL
↓
ChromaDB
↓
의미 검색이 확실히 동작하는 것을 확인해야 하기 때문입니다.
그리고 다음 단계가 진짜 재미있는 부분입니다
이번 테스트가 성공하면 다음 단계에서는 memory.py의:
def search_memory(conn, user_id, limit=10):을 바꿉니다.
현재:
PostgreSQL
↓
중요도 순서
↓
10개에서
사용자 질문
↓
ChromaDB
↓
의미적으로 가까운 Memory
↓
5개로 바꾸게 됩니다.
그러면 Agent가 예를 들어:
"내가 요즘 어떤 공부를 하고 있지?"
라고 물었을 때,
단순히 최근 Memory 10개를 가져오는 것이 아니라,
질문
"내가 요즘 어떤 공부를 하고 있지?"
↓
Embedding
↓
ChromaDB
↓
┌──────────────────────────────┐
│ Python과 AI Agent 학습 │ ← 높은 관련성
│ PostgreSQL 학습 │ ← 관련성
│ HWP 자동문서 생성 │
│ 여행 │
└──────────────────────────────┘처럼 질문과 의미가 가까운 기억을 찾아서 LLM에게 제공하게 됩니다.
이것이 이번 학습에서 중요한 분기점입니다.
따라서 지금은 여기까지 하고 실행 결과를 확인하면 됩니다
지금 해야 할 일은 딱 4가지입니다.
1. memory.py를 위 코드로 교체
2. vector_store.py를 위 코드로 교체
3. test_postgres_to_chroma.py를 새로 생성
4. 실행
[테스트 결과 예시] - 다음장에서 해설.
PostgreSQL → ChromaDB 연동 테스트
[STEP 1] PostgreSQL 연결
☑️ PostgreSQL 연결 성공
[STEP 2] 테스트 사용자 조회
☑️ 사용자 발견
user_id : 8
user_name : 스티븐김
location : 대한민국
preferences : Python과 AI Agent 학습
[STEP 3] PostgreSQL Memory 조회
☑️ PostgreSQL에서 8개 Memory 발견
memory_id : 30
user_id : 8
memory_type : goal
content : 사용자는 저작권 미구매 고객에게 자동으로 저작권 안내 전화를 걸고 구매 요청을 하는 AI를 만드는 것을 목표로 하고 있다.
importance : 9
created_at : 2026-09-05 13:21:36.822445
memory_id : 28
user_id : 8
memory_type : goal
content : 사용자의 목표는 저작권 미구매 고객에게 자동으로 저작권 안내 전화를 걸고 구매 요청을 하는 AI를 만드는 것이다.
importance : 9
created_at : 2026-09-05 13:21:16.167284
memory_id : 27
user_id : 8
memory_type : project
content : 사용자는 Python으로 AI Agent를 만들고 있으며, 저작권 미구매 고객에게 자동으로 저작권 안내 전화를 걸고 구매 요청을 하는 AI를 만들고 싶어 한다.
importance : 9
created_at : 2026-09-05 13:10:38.705845
memory_id : 23
user_id : 8
memory_type : goal
content : 사용자는 저작권 미구매 고객에게 자동으로 저작권 안내 전화를 걸고 구매 요청을 하는 AI를 만들고 싶어 한다.
importance : 9
created_at : 2026-09-05 12:54:34.591357
memory_id : 31
user_id : 8
memory_type : project
content : 사용자는 Python으로 AI Agent를 개발하여 저작권 미구매 고객에게 자동으로 전화를 걸도록 설계하고 있다.
importance : 8
created_at : 2026-09-05 13:22:11.034977
memory_id : 29
user_id : 8
memory_type : interest
content : 사용자는 Python으로 AI Agent를 만들고 있다.
importance : 8
created_at : 2026-09-05 13:21:25.354024
memory_id : 19
user_id : 8
memory_type : interest
content : 사용자는 Python으로 AI Agent를 만들고 있다.
importance : 8
created_at : 2026-09-05 12:04:07.015184
memory_id : 32
user_id : 8
memory_type : interest
content : 사용자는 Python을 활용한 AI Agent 개발에 관심이 있다.
importance : 7
created_at : 2026-09-05 13:22:11.038135
[STEP 4] PostgreSQL → ChromaDB 동기화
☑️ChromaDB에 8개 기억 저장 완료
[STEP 5] ChromaDB 상태 확인
☑️ ChromaDB 저장된 전체 Memory 개수: 15
[STEP 6] ChromaDB 의미 검색
검색 질문: 나는 파이썬과 AI 에이전트 개발을 공부하고 있다.
[STEP 7] 검색 결과
IDs:
[['memory_27', 'memory_29', 'memory_19', 'memory_32', 'memory_31']]
Documents:
[['사용자는 Python으로 AI Agent를 만들고 있으며, 저작권 미구매 고객에게 자동으로 저작권 안내 전화를 걸고 구매 요청을 하는 AI를 만들고 싶어 한다.', '사용자는 Python으로 AI Agent를 만들고 있다.', '사용자는 Python으로 AI Agent를 만들고 있다.', '사용자는 Python을 활용한 AI Agent 개발에 관심이 있다.', '사용자는 Python으로 AI Agent를 개발하여 저작권 미구매 고객에게 자동으로 전화를 걸도록 설계하고 있다.']]
Metadatas:
[[{'memory_type': 'project', 'user_id': 8, 'memory_id': 27, 'importance': 9}, {'memory_id': 29, 'importance': 8, 'memory_type': 'interest', 'user_id': 8}, {'importance': 8, 'memory_id': 19, 'memory_type': 'interest', 'user_id': 8}, {'memory_id': 32, 'user_id': 8, 'importance': 7, 'memory_type': 'interest'}, {'memory_type': 'project', 'user_id': 8, 'importance': 8, 'memory_id': 31}]]
Distances:
[[0.7492944598197937, 0.7528510689735413, 0.7528510689735413, 0.7566316723823547, 0.7820292711257935]]
🎉 PostgreSQL → ChromaDB 테스트 완료
📴 PostgreSQL 연결 종료
반응형