반응형
DB를 다루는 Memory Layer 자체를 먼저 완성하고 테스트한 다음 Agent에 붙이는 것이 가장 좋습니다.
1. 두 단계를 나누는 이유
지금 제시하신 두 단계는 사실 서로 다른 문제입니다.
1단계: Memory Layer 만들기
사용자 대화
↓
extract_memory()
↓
"이 내용 중 장기기억할 것이 무엇인가?"
↓
save_memory()
↓
PostgreSQL
↓
search_memory()
↓
관련 기억 검색
↓
build_memory_context()
↓
Agent가 사용할 문자열여기서는 "기억 시스템 자체가 제대로 작동하는가?"를 배우는 것입니다.
2단계: Agent에 Memory Layer 연결
사용자
↓
Agent
↓
Memory 검색
↓
OpenAI
↓
답변
↓
중요한 정보 발견
↓
Memory 저장여기서는 "Agent가 기억을 자동으로 사용하는가?"를 배우는 것입니다.
따라서 저는 다음 순서를 추천합니다.
현재 완료
│
▼
PostgreSQL 기본 CRUD
│
▼
★ 지금 단계
Memory Layer
│
├── extract_memory()
├── save_memory()
├── search_memory()
└── build_memory_context()
│
▼
Memory Layer 단독 테스트
│
▼
★ 다음 단계
Agent + Long-term Memory 통합
│
▼
실제 장기기억 Agent2. 이번 단계에서 중요한 설계 원칙
이번에는 database.py를 없애지 않습니다.
오히려 역할을 분리합니다.
database.py
↓
"DB와 이야기하는 사람"
memory.py
↓
"기억을 관리하는 사람"
test_memory.py
↓
"기억 시스템을 시험하는 사람"이렇게 만드는 것이 좋습니다.
최종 구조는:
agent_project/
│
├── .env
│
├── database.py
│
├── memory.py
│
└── test_memory.py입니다.
3. 각 파일의 역할
database.py
이미 만든 파일입니다.
PostgreSQL
▲
│
database.pyDB의 CRUD만 담당합니다.
memory.py
이번에 만들 파일입니다.
LLM
│
▼
extract_memory()
│
▼
save_memory()
│
▼
PostgreSQL
│
▼
search_memory()
│
▼
build_memory_context()즉 "무엇을 기억할 것인가?"와 "어떻게 기억을 사용할 것인가?"를 담당합니다.
4. 먼저 memory.py부터 만들겠습니다
여기서 중요한 부분이 하나 있습니다.
우리가 단순히 LLM에게:
"이 내용을 요약해줘."라고 하는 것이 아닙니다.
이번에는:
"이 대화에서 장기기억으로 저장할 가치가 있는 정보가 있는가?"
를 판단하게 합니다.
예를 들어:
사용자:
오늘 점심은 김치찌개를 먹었어.→ 장기기억할 필요 없음.
반면:
사용자:
나는 Python으로 AI Agent를 개발하고 있어.→ 장기기억할 가치가 있음.
또:
사용자:
코드를 설명할 때 전체 흐름을 먼저 설명해주는 방식을 좋아해.→ 장기기억할 가치가 있음.
이 차이를 LLM에게 판단시키는 것입니다.
5. memory.py
다음 코드를 그대로 만들어보세요.
# ============================================================
# memory.py : Long-term Memory 관리 모듈
# 기능
# 1. extract_memory() - 대화에서 장기기억할 정보를 LLM에게 추출시킨다.
# 2. save_memory() - 추출된 기억을 PostgreSQL에 저장한다.
# 3. search_memory() - 특정 사용자의 기억을 검색한다.
# 4. 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에 저장한다."""
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에서 중요도가 높은 기억부터 가져온다.
"""
memories = get_memories(
conn,
user_id=user_id,
limit=limit,
)
return memories
# ============================================================
# 7. 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)
6. 여기서 아주 중요한 구조
이제 memory.py 안에서 데이터가 이렇게 흐릅니다.
사용자 대화
│
▼
extract_memory()
│
▼
┌────────────────────┐
│ 중요한 기억인가? │
└─────────┬──────────┘
│
기억할 가치 있음
│
▼
save_memory()
│
▼
database.py
│
▼
PostgreSQL반대로 기억을 사용할 때는:
PostgreSQL
│
▼
search_memory()
│
▼
build_memory_context()
│
▼
"사용자 장기기억:"
│
▼
OpenAI Agent입니다.
7. 그런데 지금 바로 Agent에 붙이면 안 됩니다
여기가 중요합니다.
지금은 test_memory.py를 먼저 만들어서 Memory Layer만 테스트하겠습니다.
왜냐하면 문제가 생겼을 때:
LLM 문제인가?
DB 문제인가?
Memory 문제인가?
Agent 문제인가?를 구분할 수 있어야 하기 때문입니다.
8. test_memory.py 만들기
# ============================================================
# test_memory.py : Long-term Memory 기능 테스트
# 테스트 순서
# 1. PostgreSQL 연결
# 2. 테스트 사용자 생성
# 3. 대화 입력
# 4. LLM Memory Extraction
# 5. Memory 저장
# 6. PostgreSQL Memory 조회
# 7. Memory Context 생성
# 8. 결과 출력
# ============================================================
from database import (connect_db, create_user, get_user, close_connection)
from memory import (extract_memory, save_memories, search_memory, build_memory_context)
# ============================================================
# 1. 시작
# ============================================================
print("\n")
print("="*70)
print("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="Memory Test User",
location="Seoul",
preferences=("AI Agent와 Python에 관심이 있음")
)
print(f"테스트 user_id = {user_id}")
# ========================================================
# 4. 사용자 확인
# ========================================================
user = get_user(conn, user_id)
print("\n사용자 정보: ")
print(user)
# ========================================================
# 5. 테스트 대화
# ========================================================
print("\n")
print("="*70)
print("2. 테스트 대화")
print("="*70)
conversation = """
사용자: 나는 요즘 Python으로 AI Agent를 만들고 있어.
특히 Agent의 Memory 기능을 공부하고 있어.
사용자: 앞으로 실제 업무에서 사용할 수 있는 AI Agent를 만드는 것이 목표야.
사용자: 그리고 코드를 설명받을 때는 전체적인 흐름을 먼저 설명해주고,
그 다음에 세부 코드를 자세하게 설명해주는 방식을 좋아해.
사용자: 오늘 점심에는 김치찌개를 먹었어.
"""
print(conversation)
# ========================================================
# 6. Memory Extraction
# ========================================================
print("\n")
print("="*70)
print("3. Memory Extraction")
print("="*70)
memories = extract_memory(conversation)
print("\n추출된 Memory")
for memory in memories:
print(f"\nType: {memory.get('memory_type')}")
print(f"Content: {memory.get('content')}")
print(f"Importance: {memory.get('importance')}")
# ========================================================
# 7. Memory 저장
# ========================================================
print("\n")
print("="*70)
print("4. PostgreSQL에 Memory 저장")
print("="*70)
saved_ids = save_memories(conn, user_id, memories)
print(f"\n저장된 Memory ID: {saved_ids}")
# ========================================================
# 8. Memory 검색
# ========================================================
print("\n")
print("="*70)
print("5. PostgreSQL에서 Memory 검색")
print("="*70)
stored_memories = search_memory(conn, user_id, limit=10)
for memory in stored_memories:
print("\n")
print(f"memory_id: {memory[0]}")
print(f"type: {memory[1]}")
print(f"content: {memory[2]}")
print(f"importance: {memory[3]}")
print(f"created_at: {memory[4]}")
print("-"*70)
# ========================================================
# 9. Memory Context 생성
# ========================================================
print("\n")
print("="*70)
print("6. Agent용 Memory Context")
print("="*70)
memory_context = build_memory_context(stored_memories)
print("\n")
print(memory_context)
finally:
# ========================================================
# 10. DB 종료
# ========================================================
close_connection(conn)
print("\n")
print("="*70)
print("Memory Test 종료")
print("="*70)
9. 실행
폴더 구조가 다음과 같은지 확인하세요.
agent_project/
│
├── .env
│
├── database.py
│
├── memory.py
│
└── test_memory.py그리고:
python test_memory.py를 실행합니다.
10. 가장 중요한 테스트 결과
LLM이 정상적으로 판단한다면 대략 이런 기억들이 나올 것입니다.
[interest]
사용자는 Python과 AI Agent 개발에 관심이 있다.
[project]
사용자는 실제 업무에서 사용할 수 있는 AI Agent를 만들고 싶어 한다.
[preference]
사용자는 코드 설명을 받을 때 전체 흐름을 먼저 설명하고
그 다음 세부 내용을 자세하게 설명받는 것을 선호한다.반면:
오늘 점심에는 김치찌개를 먹었어.는 기억에서 제외될 가능성이 높습니다.
바로 이것이 이번 단계의 핵심 테스트입니다.
11. build_memory_context()는 무엇인가?
예를 들어 PostgreSQL에서:
memory 1
Python과 AI Agent에 관심
memory 2
실제 업무용 Agent 개발 목표
memory 3
전체 흐름을 먼저 설명받는 것을 선호를 가져왔다고 합시다.
build_memory_context()가 이것을:
다음은 사용자의 장기기억입니다.
답변에 도움이 될 경우 자연스럽게 활용하세요.
- [interest] 사용자는 Python과 AI Agent 개발에 관심이 있다. (중요도: 8)
- [project] 사용자는 실제 업무에서 사용할 수 있는 AI Agent를 만들고 싶어 한다. (중요도: 10)
- [preference] 사용자는 코드 설명을 받을 때 전체 흐름을 먼저 설명하고 그 다음 세부 내용을 자세하게 설명받는 것을 선호한다. (중요도: 9)라는 하나의 문자열로 만들어 줍니다.
이 문자열을 다음 단계에서 Agent에게 전달하는 것입니다.
12. 그런데 search_memory()가 아직 진짜 "검색"은 아닙니다
여기서 하나 짚고 넘어가겠습니다.
현재:
search_memory()는 실제로는:
SELECT ...
FROM memories
WHERE user_id = ?
ORDER BY importance DESC
LIMIT ?입니다.
즉:
"이 사용자의 기억을 중요도 순으로 가져와라."
입니다.
아직:
"사용자의 질문과 관련된 기억만 찾아라."
는 아닙니다.
이 차이가 매우 중요합니다.
13. 진짜 Memory Search는 다음 단계에서 발전시킵니다
예를 들어 사용자가:
"내가 Python을 왜 공부하고 있었지?"라고 하면 모든 기억을 가져오는 것이 아니라:
질문
↓
"Python"
↓
관련 Memory 검색
↓
Python / AI Agent 관련 기억
↓
OpenAI가 되어야 합니다.
이것이 나중에 Embedding + Vector Search로 발전합니다.
구조가:
현재
PostgreSQL
↓
중요도 기반 검색
다음
PostgreSQL + Vector
↓
의미 기반 검색
최종
PostgreSQL + pgvector
↓
Embedding
↓
Semantic Search로 발전합니다.
14. 지금 단계에서는 이것만 확실히 이해하면 됩니다
현재 우리가 만든 것은:
MEMORY SYSTEM
│
┌─────────────┼─────────────┐
│ │ │
▼ ▼ ▼
Extract Save Search
│ │ │
▼ ▼ ▼
LLM PostgreSQL PostgreSQL
│
▼
Memory Context입니다.
이것을 먼저 독립적으로 성공시키는 것이 중요합니다.
15. 그리고 다음 단계에서 Agent와 연결
이번 단계가 성공하면 그때 비로소:
👤 사용자
"내가 어떤 공부를 하고 있었지?"
│
▼
Agent
│
▼
search_memory()
│
▼
PostgreSQL
│
▼
"Python / AI Agent 공부"
│
▼
build_memory_context()
│
▼
OpenAI
│
▼
"Python으로 AI Agent를
공부하고 계셨습니다."그리고 새로운 대화에서:
👤
"앞으로 업무용 Agent를 만들고 싶어."라고 하면:
↓
extract_memory()
↓
"업무용 Agent 개발 목표"
↓
save_memory()
↓
PostgreSQL이 자동으로 일어나도록 만들 것입니다.
지금 단계의 학습 목표
지금은 Agent를 만들었다고 생각하지 마시고, 다음 4개를 완전히 이해하는 데 집중하세요.
① extract_memory()
→ 무엇을 기억할지 판단
② save_memory()
→ 기억을 DB에 저장
③ search_memory()
→ 저장된 기억 가져오기
④ build_memory_context()
→ AI가 사용할 수 있는 형태로 변환이 4개가 제대로 돌아가면 다음 단계에서 Agent Loop에 이 Memory Layer를 끼워 넣는 작업은 생각보다 간단합니다.
그리고 한 가지는 기억해 두시면 좋습니다. 지금 search_memory()는 "관련성 검색"이 아니라 "사용자의 기억 목록 조회" 단계입니다. 이 차이를 일부러 남겨둔 것입니다. 다음 단계에서 Agent를 연결한 뒤, 그 다음 발전 단계로 pgvector + Embedding을 붙이면 "Python에 관한 기억", "프로젝트에 관한 기억"처럼 의미를 이해해서 관련 기억을 찾는 장기 메모리로 발전시킬 수 있습니다.
반응형
'Agentic AI > Memory' 카테고리의 다른 글
| 5. 장기 메모리(2)(Long-term Memory) - Python으로 PostgreSQL을 장기 기억 저장소처럼 사용하기 (0) | 2026.09.03 |
|---|---|
| 4. 장기 메모리(1)(Long-term Memory) - Docker & PostgreSQL 실행하기 (0) | 2026.09.01 |
| 2. [Summary Memory] 대화 이력 슬라이딩 윈도우 및 자동 요약 압축 시스템 (0) | 2026.08.31 |
| 1. [Short-term Memory] 대화 이력 자동 압축 및 요약 시스템 (0) | 2026.08.31 |
