Agentic AI/Memory
7. 장기기억(4)-Memory Layer를 Agent Loop에 연결하기
아톨
2026. 9. 6. 12:39
반응형
지금까지는 “장기기억을 저장하고 꺼내는 Memory Layer”를 독립적으로 만드는 단계였습니다.
이제 다음 단계는 자연스럽게 다음과 같습니다.
[요약]
# 1. agent_with_memory.py 생성, 그리고 수정(장기기억용(3))
# 2. database.py 수정(장기기억용(1)(2))
# 3. 테스트 실시
다음 단계: Memory Layer를 Agent Loop에 연결하기
지금까지의 구조는 다음과 같습니다.
사용자 대화
↓
extract_memory()
↓
중요한 기억 추출
↓
PostgreSQL 저장
↓
search_memory()
↓
기억 조회
↓
build_memory_context()이제 이것을 실제 Agent에 연결합니다.
사용자 질문
↓
Agent
↓
① 관련 장기기억 검색
↓
② 최근 대화 + 장기기억을 함께 모델에 전달
↓
③ 모델 답변
↓
④ 이번 대화에서 중요한 기억 추출
↓
⑤ PostgreSQL 저장
↓
답변 반환핵심은 Agent가 대답하기 전에 기억을 읽고, 대답한 후에는 기억을 저장한다는 것입니다.
1. 먼저 전체 구조를 이해하기
예를 들어 첫 번째 날에 사용자가 이렇게 말한다고 가정하겠습니다.
사용자:
나는 요즘 Python으로 AI Agent를 만들고 있어.
장기적으로는 회사 업무에 사용할 Agent를 만들고 싶어.
코드를 설명할 때는 전체 흐름을 먼저 설명해 주면 이해하기 쉬워.Agent는 이 대화를 처리한 뒤 다음과 같은 기억을 저장할 수 있습니다.
[interest]
사용자는 Python과 AI Agent 개발에 관심이 있다.
[goal]
사용자는 회사 업무에 사용할 AI Agent를 만들고 싶어 한다.
[preference]
사용자는 코드 설명을 들을 때 전체 흐름을 먼저 설명받는 것을 선호한다.그리고 다음 날 사용자가 이렇게 질문합니다.
사용자:
내가 요즘 무엇을 공부하고 있었지?Agent는 먼저 PostgreSQL에서 관련 기억을 찾습니다.
검색 결과:
사용자는 Python과 AI Agent 개발에 관심이 있다.
사용자는 회사 업무에 사용할 AI Agent를 만들고 싶어 한다.그다음 모델에 다음과 같이 전달합니다.
시스템 지침:
당신은 사용자의 장기기억을 참고하여 답변하는 AI Agent입니다.
사용자의 장기기억:
- 사용자는 Python과 AI Agent 개발에 관심이 있다.
- 사용자는 회사 업무에 사용할 AI Agent를 만들고 싶어 한다.
최근 대화:
사용자: 내가 요즘 무엇을 공부하고 있었지?그러면 모델은 다음처럼 답할 수 있습니다.
최근에는 Python으로 AI Agent를 만드는 공부를 하고 계셨습니다.
장기적으로는 회사 업무에 사용할 수 있는 Agent를 만들려는 목표가 있으셨고,
코드를 공부할 때는 전체 흐름을 먼저 이해하는 방식을 선호하셨습니다.이것이 바로 장기기억이 실제 Agent의 답변에 사용되는 과정입니다.
2. 현재 구조에서 새로 필요한 것
기존 파일은 다음과 같습니다.
agent_project/
├── .env
├── database.py
├── memory.py
└── test_memory.py여기에 Agent 연결용 파일을 추가합니다.
agent_project/
├── .env
├── database.py
├── memory.py
├── test_memory.py
└── agent_with_memory.py각 파일의 역할은 다음과 같습니다.
파일역할
| database.py | PostgreSQL 연결 및 SQL 실행 |
| memory.py | 기억 추출, 저장, 검색, 문맥 생성 |
| test_memory.py | Memory Layer 단독 테스트 |
| agent_with_memory.py | 실제 Agent와 Memory Layer 연결 |
구조적으로 보면 다음과 같습니다.
agent_with_memory.py
│
├── memory.py
│ │
│ └── database.py
│
└── OpenAI API즉, Agent는 직접 SQL을 작성하지 않습니다.
Agent
↓
memory.py
↓
database.py
↓
PostgreSQL이렇게 분리하는 것이 중요합니다.
3. Agent가 기억을 사용하는 시점
Agent Loop 안에서 기억은 크게 두 번 사용됩니다.
① 모델 호출 전: 기억 검색
memories = search_memory(conn, user_id)
memory_context = build_memory_context(memories)그리고 모델 입력에 포함합니다.
model_input = [
{
"role": "system",
"content": SYSTEM_PROMPT
},
{
"role": "system",
"content": memory_context
},
*history
]여기서 중요한 점은 장기기억을 user 메시지로 넣는 것이 아니라, 별도의 system 메시지로 전달하는 것입니다.
② 모델 답변 후: 기억 저장
모델의 답변이 끝난 뒤 이번 대화를 Memory Manager에 전달합니다.
conversation = f"""
사용자: {user_message}
AI: {assistant_answer}
"""그리고:
memories = extract_memory(conversation)
save_memories(conn, user_id, memories)이렇게 하면 Agent는 대화를 처리할 때마다 중요한 정보를 저장할 수 있습니다.
4. 우선 가장 단순한 연결 버전부터 만들기
처음부터 기존의 복잡한 Agent Loop 전체를 수정하면 문제가 생겼을 때 원인을 찾기 어렵습니다.
따라서 이번 단계에서는 다음과 같은 간단한 Memory Agent를 먼저 만들어보겠습니다.
1. 사용자 입력 받기
2. PostgreSQL에서 기존 기억 읽기
3. 기억을 모델에 전달
4. 모델 답변 받기
5. 이번 대화에서 기억 추출
6. PostgreSQL에 저장
7. 답변 출력아래 코드는 학습용으로 단순화한 버전입니다.
5. agent_with_memory.py
import os
from dotenv import load_dotenv
from openai import OpenAI
from database import connect_db, create_user
from memory import(extract_memory, save_memories, search_memory, build_memory_context)
# ============================================================
# 1. 환경 설정
# ============================================================
load_dotenv()
MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o-mini")
client = OpenAI()
# ============================================================
# 2. 시스템 프롬프트
# ============================================================
SYSTEM_PROMPT="""
당신은 사용자의 장기기억을 참고하여 답변하는 AI Agent입니다.
사용자의 장기기억이 제공되면 답변에 도움이 되는 경우 자연스럽게 활용하세요.
단, 장기기억에 없는 내용을 기억하고 있다고 주장하지 마세요.
기억이 불확실하면 솔직하게 모른다고 답하세요.
답변은 친절하고 이해하기 쉽게 작성하세요.
"""
# ============================================================
# 3. 모델 호출 함수
# ============================================================
def call_model(user_message, memory_context):
""" 사용자 질문과 장기기억을 모델에 전달하고 답변을 반환합니다."""
response = client.responses.create(
model=MODEL_NAME,
input=[
{
"role":"system",
"content":SYSTEM_PROMPT,
},
{
"role":"system",
"content":memory_context,
},
{
"role":"user",
"content":user_message,
},
],
)
return response.output_text.strip()
# ============================================================
# 4. 한 번의 대화 처리
# ============================================================
def run_once(conn, user_id, user_message):
"""
사용자 질문 1회를 처리합니다.
처리 순서:
1. 기존 기억 검색
2. 기억 문맥 생성
3. 모델 호출
4. 이번 대화에서 기억 추출
5. 기억 저장
6. 답변 반환
"""
# --------------------------------------------------------
# 4.1. 1단계: 기존 장기기억 검색
# --------------------------------------------------------
memories = search_memory(
conn,
user_id=user_id,
limit=10,
)
# --------------------------------------------------------
# 4.2. 2단계: 모델에게 전달할 기억 문맥 생성
# --------------------------------------------------------
memory_context = build_memory_context(memories)
print("\n")
print("="*70)
print("📜현재 Agent가 참고하는 장기기억")
print("="*70)
print(memory_context)
# --------------------------------------------------------
# 4.3. 3단계: 모델 호출
# --------------------------------------------------------
answer = call_model(
user_message=user_message,
memory_context=memory_context,
)
# --------------------------------------------------------
# 4.4. 4단계: 이번 대화에서 중요한 기억 추출
# --------------------------------------------------------
conversation = f"""
사용자: {user_message}
AI: {answer}
"""
new_memories = extract_memory(conversation)
# --------------------------------------------------------
# 4.5. 5단계: 추출한 기억을 PostgreSQL에 저장
# --------------------------------------------------------
saved_ids = save_memories(
conn,
user_id=user_id,
memories=new_memories,
)
print("="*70)
print("🧠새롭게 저장된 기억")
print("="*70)
print(memory_context)
if saved_ids:
print(f"저장된 Memory ID: {saved_ids}")
else:
print("새롭게 저장된 기억이 없습니다.")
# --------------------------------------------------------
# 4.6. 6단계: 답변 반환
# --------------------------------------------------------
return answer
# ============================================================
# 5. 메인 실행
# ============================================================
def main():
print("="*70)
print("🤖Memory Agent 시작")
print("="*70)
# --------------------------------------------------------
# PostgreSQL 연결
# --------------------------------------------------------
conn = connect_db()
# --------------------------------------------------------
# 테스트 사용자 생성
# --------------------------------------------------------
# user_id = create_user(
# conn,
# user_name="스티븐김",
# location="대한민국",
# preferences="Python과 AI Agent 학습",
# )
# print(f"\n현재 사용자 ID: {user_id}")
# ### 장기기억 테스트용(추가(3)!!!)
from database import (get_or_create_user)
user_id = get_or_create_user(
conn,
user_name="스티븐김",
location="대한민국",
preferences="저작권 안내 자동전화AI",
)
# --------------------------------------------------------
# 첫 번째 질문
# --------------------------------------------------------
user_message = input("\n사용자 질문: ")
answer = run_once(
conn=conn,
user_id=user_id,
user_message=user_message,
)
print("\n")
print("="*70)
print("🤖Agent 답변")
print("="*70)
print(answer)
conn.close()
if __name__=="__main__":
main()
6. 코드의 핵심을 단계별로 해설
6-1. connect_db()
conn = connect_db()이 코드는 PostgreSQL 데이터베이스에 연결합니다.
Python 프로그램
↓
localhost:5432
↓
PostgreSQL
↓
agent_dbconn은 데이터베이스 연결 객체입니다.
앞으로 다음 작업에 사용됩니다.
search_memory(conn, ...)
save_memories(conn, ...)6-2. 사용자 ID 생성
user_id = create_user(
conn,
user_name="김현정",
location="대한민국",
preferences="Python과 AI Agent 학습"
)이 부분은 테스트를 위해 사용자를 생성하는 코드입니다.
예를 들어 PostgreSQL에는 다음처럼 저장됩니다.
user_id | user_name | location | preferences
------------------------------------------------
1 | 김현정 | 대한민국 | Python과 AI Agent 학습실제 서비스에서는 매번 새 사용자를 만들면 안 됩니다.
실제 서비스에서는 다음처럼 처리해야 합니다.
로그인한 사용자 확인
↓
기존 user_id 조회
↓
해당 user_id의 기억 검색이번에는 학습을 단순하게 하기 위해 매번 테스트 사용자를 생성합니다.
7. 가장 중요한 부분: run_once()
def run_once(conn, user_id, user_message):이 함수는 사용자의 질문 한 번을 처리하는 작은 Agent Loop입니다.
전체 흐름은 다음과 같습니다.
run_once()
│
├── search_memory()
│
├── build_memory_context()
│
├── call_model()
│
├── extract_memory()
│
├── save_memories()
│
└── return answer이 함수가 앞으로 기존 Agent Loop에 들어가게 됩니다.
8. 기존 기억 검색
memories = search_memory(
conn,
user_id=user_id,
limit=10
)예를 들어 PostgreSQL에 다음 기억이 저장되어 있다고 하겠습니다.
사용자는 Python과 AI Agent 개발에 관심이 있다.
사용자는 회사 업무용 AI Agent를 만들고 싶어 한다.
사용자는 코드 설명에서 전체 흐름을 먼저 듣는 것을 선호한다.그러면 memories에는 데이터베이스에서 가져온 행들이 들어옵니다.
현재 database.py의 get_memories()는 튜플 형태를 반환할 가능성이 높습니다.
[
(
1,
"interest",
"사용자는 Python과 AI Agent 개발에 관심이 있다.",
8,
datetime(...)
),
(
2,
"goal",
"사용자는 회사 업무용 AI Agent를 만들고 싶어 한다.",
9,
datetime(...)
)
]여기서 중요한 점은 현재 검색 방식이 아직 질문과 의미적으로 관련된 기억만 검색하는 방식은 아니라는 것입니다.
현재는 다음 기준으로 가져옵니다.
중요도 높은 순서
↓
최근 생성된 순서따라서 지금 단계에서는 “사용자의 전체 장기기억 중 상위 10개를 가져오는 방식”입니다.
9. build_memory_context()
memory_context = build_memory_context(memories)이 함수는 데이터베이스의 튜플을 모델이 이해하기 쉬운 문자열로 바꿉니다.
예를 들어:
[
(
1,
"interest",
"사용자는 Python과 AI Agent 개발에 관심이 있다.",
8,
datetime(...)
)
]이 데이터가 다음 문자열로 변환됩니다.
다음은 사용자의 장기기억입니다.
답변에 도움이 될 경우 자연스럽게 활용하세요.
- [interest] 사용자는 Python과 AI Agent 개발에 관심이 있다. (중요도: 8)즉, build_memory_context()는 일종의 번역기입니다.
PostgreSQL 데이터
↓
Python 튜플
↓
읽기 쉬운 텍스트
↓
LLM 입력10. 모델 호출
answer = call_model(
user_message=user_message,
memory_context=memory_context
)여기서 모델은 두 가지를 받습니다.
① 사용자의 현재 질문
② 사용자의 과거 장기기억예를 들어 모델 입력은 개념적으로 다음과 같습니다.
SYSTEM:
당신은 사용자의 장기기억을 참고하여 답변하는 AI Agent입니다.
SYSTEM:
다음은 사용자의 장기기억입니다.
- 사용자는 Python과 AI Agent 개발에 관심이 있다.
- 사용자는 회사 업무용 AI Agent를 만들고 싶어 한다.
- 사용자는 전체 흐름을 먼저 설명받는 것을 선호한다.
USER:
내가 요즘 무엇을 공부하고 있었지?모델은 현재 질문과 기억을 함께 참고하여 답변합니다.
11. 답변 이후 기억 추출
conversation = f"""
사용자: {user_message}
AI: {answer}
"""이 코드는 사용자 질문과 AI 답변을 하나의 대화문으로 합칩니다.
예를 들어:
사용자: 나는 PostgreSQL로 AI Agent 장기기억을 만들고 있어.
AI: PostgreSQL에 사용자별 기억을 저장하고 검색하는 구조를 만들고 있습니다.그리고 이 대화를 다음 함수에 전달합니다.
new_memories = extract_memory(conversation)그러면 LLM이 다음과 같은 판단을 합니다.
이 대화에서 앞으로도 도움이 될 정보가 있는가?있다면:
[
{
"memory_type": "project",
"content": "사용자는 PostgreSQL을 이용해 AI Agent의 장기기억 기능을 개발하고 있다.",
"importance": 9
}
]없다면:
[]12. 추출한 기억 저장
saved_ids = save_memories(
conn,
user_id=user_id,
memories=new_memories
)이 함수는 추출된 기억을 하나씩 PostgreSQL에 저장합니다.
new_memories
↓
save_memories()
↓
save_memory()
↓
database.save_memory()
↓
PostgreSQL INSERT예를 들어:
INSERT INTO memories
(
user_id,
memory_type,
content,
importance
)
VALUES
(
1,
'project',
'사용자는 PostgreSQL을 이용해 AI Agent의 장기기억 기능을 개발하고 있다.',
9
);13. 실행 방법
터미널에서 다음을 실행합니다.
python agent_with_memory.py예를 들어 첫 번째 실행에서:
사용자 질문:
나는 Python으로 AI Agent를 만들고 있어.그러면 Agent가 답변한 뒤 기억을 저장합니다.
🧠 새롭게 저장된 기억
저장된 Memory ID: [1, 2]그런데 현재 코드는 프로그램이 종료되면 다시 시작할 때 새로운 테스트 사용자를 만들 수 있습니다.
따라서 진짜 다음 실험에서는 사용자 ID를 고정해야 합니다.
14. 두 번째 실행에서 같은 사용자를 사용하기
현재 코드의 이 부분:
user_id = create_user(
conn,
user_name="김현정",
location="대한민국",
preferences="Python과 AI Agent 학습"
)은 실행할 때마다 새로운 사용자를 만들 수 있습니다.
따라서 장기기억 테스트를 위해서는 다음처럼 바꾸는 것이 좋습니다.
database.py에 사용자 조회 함수 활용
기존에 이미 다음 함수가 있을 것입니다.
get_user()이를 이용하면 됩니다.
개념적으로는 다음과 같습니다.
user = get_user(conn, user_name="김현정")사용자가 있으면:
user_id = user[0]없으면:
user_id = create_user(...)이를 함수로 만들면 다음과 같습니다.
def get_or_create_user(conn):
user = get_user(
conn,
user_name="김현정"
)
if user is not None:
return user[0]
return create_user(
conn,
user_name="김현정",
location="대한민국",
preferences="Python과 AI Agent 학습"
)그리고 main()에서는:
user_id = get_or_create_user(conn)이렇게 사용합니다. database.py 수정하기
# ============================================================
# database.py : PostgreSQL 장기 메모리 관리를 담당하는 모듈
# 기능
# 1. connect_db()
# 2. create_user()
# 3. get_user()
# 4. save_memory()
# 5. get_memories()
# 6. close_connection()
# ============================================================
import os
import psycopg
from dotenv import load_dotenv
# ============================================================
# 1. 환경설정
# ============================================================
load_dotenv()
# PostgreSQL 접속 정보
DB_HOST = os.getenv("DB_HOST", "localhost")
DB_PORT = os.getenv("DB_PORT", "5432")
DB_NAME = os.getenv("DB_NAME", "agent_db")
DB_USER = os.getenv("DB_USER", "agent_user")
DB_PASSWORD = os.getenv("DB_PASSWORD", "agent_password")
# ============================================================
# 2. DB 연결
# ============================================================
def connect_db():
"""
PostgreSQL 데이터베이스에 연결한다.
반환값: psycopg Connection 객체
"""
conn = psycopg.connect(
host = DB_HOST,
port = DB_PORT,
dbname = DB_NAME,
user = DB_USER,
password = DB_PASSWORD
)
print("☑️ PostgreSQL 연결 성공")
return conn
# ============================================================
# 3. 사용자 생성
# ============================================================
def create_user(conn, user_name, location=None, preferences=None):
"""
새로운 사용자를 생성한다.
매개변수:
conn :DB 연결 객체
user_name :사용자 이름
location :위치
preferences :사용자 선호
반환값: 생성된 user_id
"""
sql = """
INSERT INTO
user_profiles (user_name, location, preferences)
VALUES
(%s, %s, %s)
RETURNING
user_id;
"""
with conn.cursor() as cursor:
cursor.execute(sql, (user_name, location, preferences))
user_id = cursor.fetchone()[0]
# DB에 실제 반영
conn.commit()
print(f"☑️ 사용자 생성 완료 (user_id = {user_id})")
return user_id
# ============================================================
# 4. 사용자 조회
# ============================================================
def get_user(conn, user_id): #user_id로 사용자 조회
"""
user_id를 이용해서 사용자 정보를 조회한다.
반환값: 사용자 정보(tuple). 없으면 None
"""
sql = """
SELECT
user_id, user_name, location, preferences
FROM
user_profiles
WHERE
user_id = %s;
"""
with conn.cursor() as cursor:
cursor.execute(sql, (user_id,))
user = cursor.fetchone()
return user
### 장기기억 테스트용(추가(1)!!!)-user_name으로 사용자 조회
def get_user_by_name(conn, user_name):
"""
user_name을 이용해서 사용자 정보를 조회한다.
반환값: 사용자 정보(tuple). 없으면 None
"""
sql = """
SELECT
user_id, user_name, location, preferences
FROM
user_profiles
WHERE
user_name = %s;
"""
with conn.cursor() as cursor:
cursor.execute(sql, (user_name,))
user = cursor.fetchone()
return user
### 장기기억 테스트용(추가(2)!!!) 있으면 조회하고 없으면 생성
def get_or_create_user(conn, user_name, location=None, preferences=None):
"""
사용자가 존재하면 기존 user_id를 반환하고,
존재하지 않으면 새 사용자를 생성한 뒤 user_id를 반환한다.
"""
# --------------------------------------------------------
# 1. 사용자 이름으로 기존 사용자 조회
# --------------------------------------------------------
user = get_user_by_name(conn, user_name=user_name)
# --------------------------------------------------------
# 2. 사용자가 이미 존재하면 기존 user_id 반환
# --------------------------------------------------------
if user is not None:
return user[0]
# --------------------------------------------------------
# 3. 사용자가 없으면 새 사용자 생성
# --------------------------------------------------------
return create_user(
conn,
user_name=user_name,
location=location,
preferences=preferences,
)
# ============================================================
# 5. 기억 저장
# ============================================================
def save_memory(conn, user_id, memory_type, content, importance=5):
"""
사용자의 장기 기억을 저장한다.
매개변수:
conn: DB 연결 개체
user_id: 사용자 ID
memory_type: 기억 종류
content: 기억 내용
importance: 중요도(기본값 5)
반환값: 생성된 memory_id
"""
sql = """
INSERT INTO
memories (user_id, memory_type, content, importance)
VALUES
(%s, %s, %s, %s)
RETURNING
memory_id;
"""
with conn.cursor() as cursor:
cursor.execute(sql, (user_id, memory_type, content, importance))
memory_id = cursor.fetchone()[0]
#DB에 실제 반영
conn.commit()
print(f"🧠기억 저장 완료 (memory_id={memory_id})")
return memory_id
# ============================================================
# 6. 기억 조회
# ============================================================
def get_memories(conn, user_id, limit=10):
"""
특정 사용자의 장기 기억을 조회한다.
중요도가 높은 기억을 우선하고, 같은 중요도라면 최근 기억부터 가져온다.
반환값: 기억 목록
"""
sql = """
SELECT
memory_id, memory_type, content, importance, created_at
FROM
memories
WHERE
user_id = %s
ORDER BY
importance DESC, created_at DESC
LIMIT %s;
"""
with conn.cursor() as cursor:
cursor.execute(sql, (user_id, limit))
memories = cursor.fetchall()
return memories
# ============================================================
# 7. DB 연결 종료
# ============================================================
def close_connection(conn):
""" PostgreSQL 연결을 종료한다."""
if conn is not None:
conn.close()
print("📴 PostgreSQL 연결 종료")
15. get_or_create_user()의 의미
이 함수의 뜻은 다음과 같습니다.
사용자 조회
↓
존재하는가?
├── YES → 기존 user_id 사용
│
└── NO → 새 사용자 생성이 구조가 필요한 이유는 장기기억이 다음과 같이 연결되기 때문입니다.
user_id = 1
│
├── Memory 1
├── Memory 2
├── Memory 3
└── Memory 4사용자가 매번 새로 생성되면:
첫 번째 실행 → user_id 1
두 번째 실행 → user_id 2
세 번째 실행 → user_id 3이렇게 되어 기억이 분리됩니다.
반대로 같은 사용자 ID를 사용하면:
첫 번째 실행
↓
user_id 1
↓
Memory 1, 2
두 번째 실행
↓
user_id 1
↓
Memory 1, 2를 다시 검색이것이 장기기억의 핵심입니다.
16. 실제 장기기억 테스트 시나리오
이제 다음 순서로 테스트해보세요.
테스트 1: 기억 저장
프로그램 실행:
python agent_with_memory.py질문:
나는 요즘 Python으로 AI Agent를 만들고 있어.또는:
장기적으로 회사 업무에 사용할 수 있는 AI Agent를 만들고 싶어.또는:
코드를 설명할 때는 전체 흐름을 먼저 설명해 주면 이해하기 쉬워.이 대화가 끝나면 PostgreSQL에 기억이 저장되어야 합니다.
테스트 2: 기억 확인
프로그램을 종료한 뒤 다시 실행합니다.
python agent_with_memory.py질문:
내가 요즘 무엇을 공부하고 있었지?실행 로그에서 먼저 다음과 같은 내용이 보여야 합니다.
📚 현재 Agent가 참고하는 장기기억
- [interest] 사용자는 Python과 AI Agent 개발에 관심이 있다.
- [goal] 사용자는 회사 업무용 AI Agent를 만들고 싶어 한다.
- [preference] 사용자는 전체 흐름을 먼저 설명받는 것을 선호한다.그리고 답변에 이 기억이 반영되어야 합니다.
17. 이번 단계에서 반드시 확인할 것
이번에는 기능을 한꺼번에 많이 추가하지 말고, 다음 네 가지를 확인하는 것이 좋습니다.
확인 1. 기억이 저장되는가?
사용자 질문
↓
extract_memory()
↓
PostgreSQL INSERT확인 2. 프로그램을 종료해도 기억이 남는가?
프로그램 종료
↓
PostgreSQL 데이터 유지
↓
프로그램 재실행
↓
기존 기억 검색확인 3. Agent가 기억을 실제 답변에 사용하는가?
단순히 DB에서 읽어오는 것만으로는 부족합니다.
반드시 모델 입력에 포함되어야 합니다.
{
"role": "system",
"content": memory_context
}확인 4. 중요하지 않은 정보가 저장되지 않는가?
예를 들어:
오늘 점심으로 김치찌개를 먹었어.이런 정보는 일반적으로 장기기억에 저장하지 않는 것이 좋습니다.
반면:
나는 매운 음식을 잘 먹지 못해.이 정보는 앞으로 식당이나 음식 추천에 도움이 될 수 있으므로 저장할 가치가 있습니다.
즉, Memory Manager는 단순한 대화 저장소가 아니라:
대화 전체
↓
중요한 정보만 선별
↓
장기기억 저장의 역할을 해야 합니다.
18. 현재 단계의 한계
이번에 만든 구조는 장기기억의 기본 동작을 이해하기 위한 버전입니다.
아직 다음 문제들이 있습니다.
① 매번 같은 기억이 중복 저장될 수 있음
예를 들어:
사용자는 Python과 AI Agent에 관심이 있다.라는 기억이 매번 저장될 수 있습니다.
Memory 1: 사용자는 Python과 AI Agent에 관심이 있다.
Memory 2: 사용자는 Python과 AI Agent에 관심이 있다.
Memory 3: 사용자는 Python과 AI Agent에 관심이 있다.나중에는 중복 검사 기능이 필요합니다.
② 현재 검색은 의미 기반 검색이 아님
현재는:
중요도 높은 기억 10개를 가져옵니다.
예를 들어 사용자가:
내가 선호하는 코드 설명 방식은?이라고 질문해도 모든 종류의 기억 중 상위 10개를 가져옵니다.
나중에는 다음처럼 바꿔야 합니다.
질문:
내가 선호하는 코드 설명 방식은?
↓
검색어:
코드 설명 선호 방식
↓
관련 기억만 검색③ 기억의 수정과 삭제가 없음
사용자가 처음에는:
나는 Java를 주로 사용해.라고 했다가 나중에는:
앞으로는 Python을 주로 사용할 거야.라고 할 수 있습니다.
그러면 기존 기억을 그대로 두는 것이 아니라:
기존 기억 수정
또는
새 기억으로 대체해야 합니다.
④ 최근 대화와 장기기억이 아직 분리되지 않음
Agent에는 사실 세 종류의 기억이 필요합니다.
① 최근 대화 기억
② 요약 기억
③ 장기기억각각의 역할은 다릅니다.
기억 종류역할
| 최근 대화 | 현재 대화의 흐름 유지 |
| 요약 기억 | 오래된 대화의 핵심 내용 압축 |
| 장기기억 | 사용자 목표, 선호, 프로젝트 등 지속적인 정보 저장 |
구조는 다음과 같습니다.
┌── 최근 대화
사용자 질문 ────────┤
├── 요약 기억
│
└── 장기기억이 세 가지를 합쳐서 모델에 전달해야 실제 서비스 수준의 Agent가 됩니다.
19. 지금 단계에서의 최종 구조
현재 우리가 만들고 있는 구조는 다음과 같습니다.
┌─────────────────────────────┐
│ 사용자 질문 │
└──────────────┬──────────────┘
↓
┌─────────────────────────────┐
│ Agent Loop │
│ │
│ 1. 장기기억 검색 │
│ 2. 모델 호출 │
│ 3. 답변 생성 │
│ 4. 기억 추출 │
│ 5. 기억 저장 │
└──────────────┬──────────────┘
↓
┌─────────────────────────────┐
│ memory.py │
│ │
│ extract_memory() │
│ search_memory() │
│ build_memory_context() │
│ save_memories() │
└──────────────┬──────────────┘
↓
┌─────────────────────────────┐
│ database.py │
│ │
│ connect_db() │
│ get_user() │
│ save_memory() │
│ get_memories() │
└──────────────┬──────────────┘
↓
┌─────────────────────────────┐
│ PostgreSQL │
│ │
│ user_profiles │
│ memories │
└─────────────────────────────┘20. 이번 단계에서의 추천 진행 순서
저는 다음 순서로 진행하는 것이 가장 좋다고 생각합니다.
1단계
간단한 Memory Agent 실행
↓
2단계
같은 user_id로 기억 유지 확인
↓
3단계
기존 Agent Loop에 memory_context 연결
↓
4단계
Agent 답변 후 자동 기억 저장
↓
5단계
중복 기억 방지
↓
6단계
질문과 관련된 기억만 검색
↓
7단계
최근 대화 + 요약 기억 + 장기기억 통합
↓
8단계
Embedding + pgvector 기반 의미 검색지금은 1단계와 2단계를 먼저 완성하는 것이 좋습니다.
특히 이번 단계의 목표는 단순히 코드가 실행되는 것이 아니라 다음 문장이 실제로 성립하는 것입니다.
“첫 번째 실행에서 저장한 기억이 프로그램을 종료한 뒤에도 남아 있고, 두 번째 실행에서 Agent가 그 기억을 활용하여 답변한다.”
이것이 확인되면 비로소 “AI Agent에 장기기억이 연결되었다”고 볼 수 있습니다.
다음으로 저는 현재 사용 중인 기존 Agent 클래스에 이 Memory Layer를 실제로 연결하여, 최근 대화·요약 기억·PostgreSQL 장기기억이 함께 작동하는 통합 Agent 구조로 발전시키는 작업을 진행할 수 있습니다.
반응형