이 글은 「온프레미스 AI Assistant 아키텍처 — Qwen3·Qwen3-VL 14일 설계」 시리즈 11일차로, LLM 기반 Assistant의 메모리 아키텍처를 단기·장기·세션 세 계층으로 나누어 실전 설계하는 방법을 다룹니다.
어제 10화에서는 Graph RAG와 멀티모달 RAG로 단순 벡터 검색의 한계를 넘는 고급 검색 아키텍처를 다뤘습니다. 오늘은 RAG가 ‘외부 지식’을 가져온다면, 그 지식과 대화 흐름을 어떻게 기억하고, 언제 꺼내 쓰고, 어디에 저장할 것인가 — 메모리 아키텍처를 설계합니다.
오늘의 핵심 3가지
- 3계층 메모리 분류 — 단기(Working Memory), 세션(Session Memory), 장기(Long-term Memory)를 명확히 분리하고 각각의 저장소·TTL·인출 전략을 다르게 설계한다.
- 메모리 스키마와 인출 파이프라인 — 사용자 프로파일·선호·대화 요약·사실 기억을 구조화된 스키마로 관리하고, 컨텍스트 조립 시 관련 메모리만 선별 주입한다.
- 온프레미스 프라이버시 정책 — 금융IT·규제 산업에서 대화 기록·사용자 프로파일의 보존 기간·접근 통제·감사 로그를 어떻게 설계하는가.
왜 메모리인가 — RAG만으로는 부족한 이유
8화에서 다룬 컨텍스트 윈도우는 ‘현재 대화’의 범위입니다. 9~10화의 RAG는 ‘외부 문서’에서 지식을 가져옵니다. 그런데 실제 AI Assistant를 써 보면 이 두 가지로 해결되지 않는 영역이 있습니다.
- “지난주에 물어봤던 환율 기준 기억나?” — 과거 대화에 대한 참조
- “나는 항상 표로 정리해 달라고 했잖아” — 사용자 선호
- “아까 1번 방안으로 하기로 했으니 그걸로 진행해” — 세션 내 맥락 유지
- “내가 담당하는 시스템은 Java 17 + Spring Boot 3.x야” — 사용자 프로파일
이 모든 것을 매번 프롬프트에 수동으로 넣을 수는 없습니다. 시스템이 자동으로 기억하고, 적절한 시점에 꺼내 쓰는 구조 — 이것이 메모리 아키텍처입니다.
메모리 3계층 분류
인간의 기억 체계에서 영감을 받아, LLM 기반 Assistant의 메모리를 세 계층으로 분류합니다. 각 계층은 수명(lifetime), 용량(capacity), 인출 비용(retrieval cost)이 다릅니다.
1계층: 단기 메모리 (Working Memory)
현재 진행 중인 대화의 최근 N턴을 그대로 보관합니다. Qwen3의 컨텍스트 윈도우(32,768 토큰 기본, 131,072 확장 가능) 안에 직접 들어가는 메시지 이력입니다.
| 속성 | 값 |
|---|---|
| 수명 | 현재 요청 처리 동안 |
| 저장소 | 인메모리 (Python 리스트/deque) |
| 용량 | 컨텍스트 윈도우의 40~60% (8화 프롬프트 예산 참조) |
| 인출 비용 | O(1) — 그대로 프롬프트에 붙임 |
| 갱신 전략 | 슬라이딩 윈도우 + 요약 압축 |
8화에서 설계한 컨텍스트 예산에서 conversation_history 슬롯이 바로 이 단기 메모리입니다. 핵심은 윈도우가 넘칠 때 어떻게 처리하느냐입니다.
2계층: 세션 메모리 (Session Memory)
하나의 대화 세션(session) 전체를 포괄합니다. 단기 메모리에서 밀려난 오래된 턴도 세션이 살아 있는 동안은 접근 가능합니다.
| 속성 | 값 |
|---|---|
| 수명 | 세션 시작 ~ 세션 종료 (또는 TTL 만료) |
| 저장소 | Redis / SQLite / PostgreSQL |
| 용량 | 세션당 수십~수백 턴 (수 MB) |
| 인출 비용 | O(log N) — 요약 조회 또는 키워드 검색 |
| 갱신 전략 | 턴 추가 시 자동 요약 + 핵심 사실 추출 |
3계층: 장기 메모리 (Long-term Memory)
세션을 넘어 영속적으로 보존되는 기억입니다. 사용자 프로파일, 선호도, 과거 대화에서 추출된 사실(fact), 학습된 패턴 등이 여기 들어갑니다.
| 속성 | 값 |
|---|---|
| 수명 | 명시적 삭제 또는 보존 정책 만료까지 |
| 저장소 | PostgreSQL + Qdrant (벡터 인덱스) |
| 용량 | 사용자당 수천 건 (수십 MB~GB) |
| 인출 비용 | O(log N) ~ O(N) — 의미 검색 + 필터링 |
| 갱신 전략 | 세션 종료 시 배치 추출 + 주기적 통합/정리 |
전체 메모리 아키텍처 다이어그램 — 3계층 구조 한눈에 보기
세 계층이 어떻게 연결되는지 전체 흐름을 봅시다.
graph TB
subgraph "사용자 요청 수신"
A[사용자 메시지] --> B[컨텍스트 어셈블러]
end
subgraph "단기 메모리 (Working Memory)"
C[대화 버퍼
최근 N턴 raw messages]
D[슬라이딩 윈도우
토큰 예산 관리]
C --> D
end
subgraph "세션 메모리 (Session Memory)"
E[세션 스토어
Redis / SQLite]
F[턴 요약기
Qwen3-8B]
G[세션 컨텍스트
누적 요약 + 핵심 사실]
E --> F --> G
end
subgraph "장기 메모리 (Long-term Memory)"
H[사용자 프로파일
PostgreSQL]
I[사실 저장소
Fact Store]
J[대화 아카이브
벡터 인덱스 / Qdrant]
K[메모리 인출기
Semantic Search + Filter]
H --> K
I --> K
J --> K
end
B --> D
B --> G
B --> K
subgraph "프롬프트 조립"
L[시스템 프롬프트]
M[메모리 컨텍스트 블록]
N[RAG 검색 결과]
O[현재 대화]
P[최종 프롬프트]
L --> P
M --> P
N --> P
O --> P
end
D --> O
G --> M
K --> M
P --> Q[Qwen3 추론]
Q --> R[응답 생성]
R --> S[메모리 업데이트
비동기]
S --> C
S --> E
S --> I
단기 메모리 — 슬라이딩 윈도우와 요약 압축
슬라이딩 윈도우 전략
가장 단순한 접근은 최근 K턴만 유지하는 것입니다. 하지만 턴 수 기반 윈도우는 문제가 있습니다 — 코드 블록이 포함된 긴 턴 하나가 토큰 예산 대부분을 잡아먹을 수 있습니다. 따라서 토큰 수 기반 윈도우를 사용합니다.
"""
단기 메모리 — 토큰 기반 슬라이딩 윈도우 + 요약 압축
Qwen3 토크나이저 기준 토큰 카운팅
"""
from __future__ import annotations
import hashlib
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Any
import tiktoken # 또는 transformers.AutoTokenizer
@dataclass
class Message:
role: str # "user" | "assistant" | "system"
content: str
timestamp: float = field(default_factory=time.time)
token_count: int = 0
metadata: dict[str, Any] = field(default_factory=dict)
def __post_init__(self) -> None:
if self.token_count == 0:
self.token_count = WorkingMemory.count_tokens(self.content)
@dataclass
class ConversationSummary:
"""밀려난 턴들의 요약."""
text: str
turn_range: tuple[int, int] # (start_turn, end_turn)
token_count: int = 0
created_at: float = field(default_factory=time.time)
def __post_init__(self) -> None:
if self.token_count == 0:
self.token_count = WorkingMemory.count_tokens(self.text)
class WorkingMemory:
"""토큰 예산 기반 슬라이딩 윈도우 + 요약 압축."""
# Qwen3 토크나이저 근사 — 실제 운영에서는 정확한 토크나이저 사용
_CHARS_PER_TOKEN = 3.5 # 한국어 기준 보수적 추정
def __init__(
self,
max_tokens: int = 12_000, # 컨텍스트 예산의 conversation 슬롯
summary_threshold: float = 0.8, # 80% 차면 요약 트리거
min_recent_turns: int = 4, # 최소 유지 턴 수
) -> None:
self.max_tokens = max_tokens
self.summary_threshold = summary_threshold
self.min_recent_turns = min_recent_turns
self._messages: deque[Message] = deque()
self._summaries: list[ConversationSummary] = []
self._total_tokens: int = 0
self._turn_counter: int = 0
@staticmethod
def count_tokens(text: str) -> int:
"""토큰 수 근사 계산. 프로덕션에서는 실제 토크나이저 사용."""
return max(1, int(len(text) / WorkingMemory._CHARS_PER_TOKEN))
@property
def current_tokens(self) -> int:
summary_tokens = sum(s.token_count for s in self._summaries)
return self._total_tokens + summary_tokens
def add_message(self, role: str, content: str, **metadata: Any) -> None:
"""새 메시지 추가. 예산 초과 시 자동 압축."""
msg = Message(role=role, content=content, metadata=metadata)
self._messages.append(msg)
self._total_tokens += msg.token_count
self._turn_counter += 1
if self.current_tokens > self.max_tokens * self.summary_threshold:
self._compress()
def _compress(self) -> None:
"""오래된 턴을 요약으로 압축."""
if len(self._messages) <= self.min_recent_turns:
return
# 최근 min_recent_turns를 제외한 나머지를 요약 대상으로
to_summarize: list[Message] = []
while len(self._messages) > self.min_recent_turns:
msg = self._messages.popleft()
self._total_tokens -= msg.token_count
to_summarize.append(msg)
if to_summarize:
summary_text = self._generate_summary(to_summarize)
turn_start = self._turn_counter - len(self._messages) - len(to_summarize)
turn_end = turn_start + len(to_summarize) - 1
summary = ConversationSummary(
text=summary_text,
turn_range=(turn_start, turn_end),
)
self._summaries.append(summary)
def _generate_summary(self, messages: list[Message]) -> str:
"""
메시지 목록을 요약.
실제 구현에서는 Qwen3-8B를 호출하여 요약 생성.
여기서는 인터페이스만 정의.
"""
# TODO: Qwen3-8B 요약 호출
# prompt = SUMMARY_PROMPT.format(conversation=...)
# summary = await qwen3_8b.generate(prompt)
lines = []
for m in messages:
preview = m.content[:200]
lines.append(f"[{m.role}] {preview}")
return "이전 대화 요약: " + " | ".join(lines)
def get_context_messages(self) -> list[dict[str, str]]:
"""프롬프트에 주입할 메시지 목록 반환."""
result: list[dict[str, str]] = []
# 1) 누적 요약을 하나의 system 메시지로
if self._summaries:
combined = "\n\n".join(s.text for s in self._summaries)
result.append({
"role": "system",
"content": f"[이전 대화 요약]\n{combined}",
})
# 2) 최근 raw 메시지
for msg in self._messages:
result.append({"role": msg.role, "content": msg.content})
return result
def get_stats(self) -> dict[str, Any]:
"""디버깅용 통계."""
return {
"recent_turns": len(self._messages),
"summary_count": len(self._summaries),
"total_message_tokens": self._total_tokens,
"total_with_summaries": self.current_tokens,
"budget_usage": f"{self.current_tokens / self.max_tokens:.1%}",
}
요약 프롬프트 설계
단기 메모리에서 밀려나는 턴을 요약할 때, Qwen3-8B를 경량 요약기로 사용합니다. 요약 품질이 전체 대화의 맥락 유지를 결정하므로, 프롬프트 설계가 중요합니다.
CONVERSATION_SUMMARY_PROMPT = """다음 대화 내용을 간결하게 요약해 주세요.
## 요약 규칙
1. 사용자가 요청한 작업과 그 결과를 중심으로 요약
2. 합의된 결정사항은 반드시 포함
3. 구체적인 수치, 이름, 설정값은 보존
4. 감정적 표현이나 인사말은 생략
5. 3~5문장 이내로 작성
## 기존 누적 요약
{previous_summary}
## 새로 요약할 대화
{new_messages}
## 통합 요약 (기존 요약 + 새 대화를 하나로):
"""
여기서 핵심은 점진적 요약(incremental summarization)입니다. 매번 전체를 다시 요약하는 것이 아니라, 기존 요약에 새 내용을 통합합니다. Qwen3-8B 호출 비용을 최소화하면서도 맥락 손실을 줄이는 균형점입니다.
요약 시점 전략 비교
| 전략 | 트리거 | 장점 | 단점 |
|---|---|---|---|
| 토큰 임계치 | 예산의 80% 초과 | 예산 초과 방지 확실 | 긴 메시지 하나로 갑자기 트리거될 수 있음 |
| 턴 수 기반 | 매 10턴마다 | 예측 가능한 주기 | 토큰 예산과 무관하게 동작 |
| 하이브리드 | 10턴 또는 80% 중 먼저 도달 | 두 장점 결합 | 구현 복잡도 소폭 증가 |
권장은 하이브리드입니다. 일반 대화에서는 턴 수 기반으로 규칙적으로 요약하고, 코드 블록이나 긴 문서가 포함된 턴에서는 토큰 임계치가 안전망 역할을 합니다.
세션 메모리 — 대화 세션의 전체 맥락
세션이란 무엇인가
세션은 사용자가 하나의 ‘대화 주제’를 다루는 연속된 상호작용 단위입니다. 웹 채팅의 ‘새 대화’ 버튼을 누르면 새 세션이 시작됩니다. 세션 메모리의 역할은 두 가지입니다.
- 단기 메모리의 백업 저장소 — 슬라이딩 윈도우에서 밀려난 턴의 원문과 요약을 보존
- 세션 수준 메타데이터 — 이 세션에서 합의된 결정, 사용된 도구, 참조된 문서 목록 등
세션 스토어 스키마
-- PostgreSQL 세션 메모리 스키마
-- 온프레미스 환경: 자체 호스팅 PostgreSQL (NAS 또는 전용 서버)
CREATE TABLE sessions (
session_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
user_id VARCHAR(128) NOT NULL,
title TEXT, -- 자동 생성 또는 사용자 지정
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
expires_at TIMESTAMPTZ, -- TTL 기반 만료
status VARCHAR(16) NOT NULL DEFAULT 'active',
-- 'active' | 'archived' | 'expired'
metadata JSONB DEFAULT '{}'::jsonb
);
CREATE TABLE session_turns (
turn_id BIGSERIAL PRIMARY KEY,
session_id UUID NOT NULL REFERENCES sessions(session_id),
turn_number INTEGER NOT NULL,
role VARCHAR(16) NOT NULL, -- 'user' | 'assistant' | 'system' | 'tool'
content TEXT NOT NULL,
token_count INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
metadata JSONB DEFAULT '{}'::jsonb,
-- tool_calls, model_id, latency_ms 등
UNIQUE (session_id, turn_number)
);
CREATE TABLE session_summaries (
summary_id BIGSERIAL PRIMARY KEY,
session_id UUID NOT NULL REFERENCES sessions(session_id),
summary_text TEXT NOT NULL,
turn_range_start INTEGER NOT NULL,
turn_range_end INTEGER NOT NULL,
token_count INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- 세션 내 추출된 핵심 사실 (결정사항, 합의 등)
CREATE TABLE session_facts (
fact_id BIGSERIAL PRIMARY KEY,
session_id UUID NOT NULL REFERENCES sessions(session_id),
fact_type VARCHAR(32) NOT NULL,
-- 'decision' | 'preference' | 'context' | 'action_item'
content TEXT NOT NULL,
source_turn INTEGER, -- 어느 턴에서 추출됐는가
confidence REAL DEFAULT 1.0,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- 인덱스
CREATE INDEX idx_sessions_user ON sessions(user_id, status);
CREATE INDEX idx_turns_session ON session_turns(session_id, turn_number);
CREATE INDEX idx_summaries_session ON session_summaries(session_id);
CREATE INDEX idx_facts_session ON session_facts(session_id, fact_type);
CREATE INDEX idx_sessions_expires ON sessions(expires_at)
WHERE status = 'active';
세션 컨텍스트 조립
세션 메모리에서 현재 요청에 필요한 컨텍스트를 조립하는 과정입니다. 모든 턴을 다 넣을 수 없으므로, 계층적 인출을 수행합니다.
"""
세션 메모리 인출기 — 계층적 컨텍스트 조립
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
@dataclass
class SessionContext:
"""세션에서 인출한 컨텍스트."""
session_id: str
session_title: str | None
cumulative_summary: str # 누적 요약
key_facts: list[dict[str, str]] # 핵심 사실 목록
recent_turns: list[dict[str, str]] # 최근 N턴 원문
total_tokens: int
def to_prompt_block(self) -> str:
"""프롬프트에 주입할 텍스트 블록 생성."""
parts: list[str] = []
if self.cumulative_summary:
parts.append(f"[세션 요약]\n{self.cumulative_summary}")
if self.key_facts:
facts_text = "\n".join(
f"- [{f['type']}] {f['content']}" for f in self.key_facts
)
parts.append(f"[이 세션의 핵심 사실]\n{facts_text}")
return "\n\n".join(parts)
class SessionMemoryRetriever:
"""세션 메모리에서 컨텍스트를 인출."""
def __init__(
self,
db_pool: Any, # asyncpg.Pool
max_context_tokens: int = 4_000, # 세션 컨텍스트 토큰 예산
) -> None:
self._pool = db_pool
self._max_tokens = max_context_tokens
async def retrieve(self, session_id: str) -> SessionContext:
"""세션 메모리에서 컨텍스트 인출."""
async with self._pool.acquire() as conn:
# 1) 세션 메타데이터
session = await conn.fetchrow(
"SELECT title, metadata FROM sessions WHERE session_id = $1",
session_id,
)
# 2) 가장 최신 누적 요약
latest_summary = await conn.fetchrow(
"""SELECT summary_text, token_count
FROM session_summaries
WHERE session_id = $1
ORDER BY turn_range_end DESC
LIMIT 1""",
session_id,
)
# 3) 핵심 사실 (결정사항 우선)
facts = await conn.fetch(
"""SELECT fact_type, content, confidence
FROM session_facts
WHERE session_id = $1
ORDER BY
CASE fact_type
WHEN 'decision' THEN 1
WHEN 'action_item' THEN 2
WHEN 'preference' THEN 3
ELSE 4
END,
created_at DESC
LIMIT 20""",
session_id,
)
# 4) 토큰 예산에 맞춰 조립
summary_text = latest_summary["summary_text"] if latest_summary else ""
key_facts = [
{"type": f["fact_type"], "content": f["content"]}
for f in facts
]
return SessionContext(
session_id=session_id,
session_title=session["title"] if session else None,
cumulative_summary=summary_text,
key_facts=key_facts,
recent_turns=[], # WorkingMemory가 관리
total_tokens=0, # 실제 토큰 카운팅 필요
)
핵심 사실 자동 추출
세션 메모리의 가치는 핵심 사실 추출에 있습니다. 매 턴이 끝날 때마다 Qwen3-8B를 비동기로 호출하여, 대화에서 중요한 사실을 추출합니다.
FACT_EXTRACTION_PROMPT = """다음 대화 턴에서 기억해야 할 핵심 사실을 추출해 주세요.
## 추출 기준
- decision: 사용자가 결정한 사항 (예: "A 방안으로 진행", "Python 3.12 사용")
- preference: 사용자의 선호 (예: "표 형식으로 정리해 줘", "영어로 답변해 줘")
- context: 작업 배경 정보 (예: "우리 시스템은 Spring Boot 3.x", "대상 사용자는 200명")
- action_item: 해야 할 일 (예: "내일까지 보고서 초안 작성")
## 규칙
1. 사실이 없으면 빈 배열 [] 반환
2. 이전에 추출된 사실과 모순되면 새 사실로 대체 (최신 우선)
3. JSON 배열로 반환: [{"type": "...", "content": "..."}]
## 이 세션에서 이미 추출된 사실
{existing_facts}
## 새 대화 턴
사용자: {user_message}
어시스턴트: {assistant_response}
## 추출된 사실 (JSON):
"""
추출 빈도를 매 턴마다 하면 Qwen3-8B 호출이 많아집니다. 운영 부하를 줄이려면 3턴마다 배치 추출하거나, 사용자 메시지에 결정/선호를 나타내는 키워드(“~로 하자”, “~해 줘”, “~으로 결정”)가 포함된 경우에만 트리거하는 조건부 추출도 효과적입니다.
장기 메모리 — 세션을 넘는 기억
장기 메모리의 구성 요소
장기 메모리는 세 가지 하위 저장소로 구성됩니다.
| 저장소 | 내용 | 저장 형태 | 인출 방법 |
|---|---|---|---|
| 사용자 프로파일 | 이름, 역할, 기술 스택, 커뮤니케이션 선호 | 구조화 (JSON/RDB) | 사용자 ID로 직접 조회 |
| 사실 저장소 (Fact Store) | 과거 대화에서 추출된 영속 사실 | 반구조화 (RDB + 벡터) | 의미 검색 + 타입 필터 |
| 대화 아카이브 | 종료된 세션의 요약 + 임베딩 | 벡터 인덱스 (Qdrant) | 의미 검색 |
사용자 프로파일 스키마
-- 사용자 프로파일 — 장기 메모리의 핵심
CREATE TABLE user_profiles (
user_id VARCHAR(128) PRIMARY KEY,
display_name VARCHAR(256),
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
-- 구조화된 선호도
preferences JSONB NOT NULL DEFAULT '{}'::jsonb,
/*
preferences 예시:
{
"language": "ko",
"response_format": "concise", -- "concise" | "detailed" | "structured"
"code_style": "type_annotated",
"timezone": "Asia/Seoul",
"expertise_level": "senior",
"domains": ["backend", "devops", "fintech"]
}
*/
-- 기술 프로파일 (개발자 사용 시)
tech_context JSONB DEFAULT '{}'::jsonb,
/*
tech_context 예시:
{
"primary_language": "Java",
"frameworks": ["Spring Boot 3.x", "MyBatis"],
"infra": ["Kubernetes", "AWS EKS"],
"conventions": ["Google Java Style", "REST API"]
}
*/
-- 통계
total_sessions INTEGER DEFAULT 0,
total_turns INTEGER DEFAULT 0,
last_active_at TIMESTAMPTZ
);
-- 장기 사실 저장소
CREATE TABLE long_term_facts (
fact_id BIGSERIAL PRIMARY KEY,
user_id VARCHAR(128) NOT NULL REFERENCES user_profiles(user_id),
category VARCHAR(32) NOT NULL,
-- 'skill' | 'project' | 'preference' | 'context' | 'relationship'
content TEXT NOT NULL,
source_session UUID, -- 어느 세션에서 왔는가
confidence REAL DEFAULT 1.0, -- 0.0 ~ 1.0
access_count INTEGER DEFAULT 0, -- 인출 빈도 (중요도 지표)
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
expires_at TIMESTAMPTZ, -- 시간 제한 사실
embedding VECTOR(1024) -- bge-m3 임베딩 (pgvector)
);
CREATE INDEX idx_ltf_user ON long_term_facts(user_id, category);
CREATE INDEX idx_ltf_embedding ON long_term_facts
USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);
-- 대화 아카이브 (종료된 세션 요약)
CREATE TABLE conversation_archive (
archive_id BIGSERIAL PRIMARY KEY,
user_id VARCHAR(128) NOT NULL,
session_id UUID NOT NULL,
title TEXT,
summary TEXT NOT NULL,
key_topics TEXT[] DEFAULT '{}', -- 주요 토픽 태그
turn_count INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
embedding VECTOR(1024) -- 요약의 bge-m3 임베딩
);
CREATE INDEX idx_archive_user ON conversation_archive(user_id, created_at DESC);
CREATE INDEX idx_archive_embedding ON conversation_archive
USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);
장기 메모리 인출 파이프라인
장기 메모리 인출은 9화의 RAG 파이프라인과 구조적으로 유사하지만, 검색 대상이 외부 문서가 아닌 사용자 자신의 기억이라는 점이 다릅니다.
"""
장기 메모리 인출기 — 사용자 프로파일 + 사실 검색 + 대화 아카이브
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
@dataclass
class LongTermContext:
"""장기 메모리에서 인출한 컨텍스트."""
user_profile: dict[str, Any]
relevant_facts: list[dict[str, Any]]
related_sessions: list[dict[str, Any]]
total_tokens: int
def to_prompt_block(self) -> str:
parts: list[str] = []
# 사용자 프로파일 (항상 포함)
prefs = self.user_profile.get("preferences", {})
tech = self.user_profile.get("tech_context", {})
if prefs or tech:
profile_lines = []
if prefs.get("response_format"):
profile_lines.append(
f"응답 스타일: {prefs['response_format']}"
)
if prefs.get("expertise_level"):
profile_lines.append(
f"전문성 수준: {prefs['expertise_level']}"
)
if tech.get("primary_language"):
profile_lines.append(
f"주 언어: {tech['primary_language']}"
)
if tech.get("frameworks"):
profile_lines.append(
f"프레임워크: {', '.join(tech['frameworks'])}"
)
if profile_lines:
parts.append(
"[사용자 프로파일]\n" + "\n".join(profile_lines)
)
# 관련 사실
if self.relevant_facts:
facts_text = "\n".join(
f"- {f['content']}" for f in self.relevant_facts
)
parts.append(f"[기억된 사실]\n{facts_text}")
# 관련 과거 세션
if self.related_sessions:
sessions_text = "\n".join(
f"- [{s['title']}] {s['summary'][:150]}"
for s in self.related_sessions
)
parts.append(f"[관련 과거 대화]\n{sessions_text}")
return "\n\n".join(parts)
class LongTermMemoryRetriever:
"""장기 메모리 인출."""
def __init__(
self,
db_pool: Any, # asyncpg.Pool (pgvector 확장)
embedding_fn: Any, # bge-m3 임베딩 함수
max_facts: int = 10,
max_sessions: int = 3,
similarity_threshold: float = 0.35,
) -> None:
self._pool = db_pool
self._embed = embedding_fn
self._max_facts = max_facts
self._max_sessions = max_sessions
self._threshold = similarity_threshold
async def retrieve(
self,
user_id: str,
query: str,
) -> LongTermContext:
"""현재 쿼리와 관련된 장기 메모리 인출."""
query_embedding = await self._embed(query)
async with self._pool.acquire() as conn:
# 1) 사용자 프로파일 (항상)
profile = await conn.fetchrow(
"""SELECT preferences, tech_context
FROM user_profiles WHERE user_id = $1""",
user_id,
)
# 2) 의미 검색으로 관련 사실 인출
facts = await conn.fetch(
"""SELECT content, category, confidence,
1 - (embedding <=> $1::vector) AS similarity
FROM long_term_facts
WHERE user_id = $2
AND 1 - (embedding <=> $1::vector) > $3
ORDER BY similarity DESC
LIMIT $4""",
query_embedding,
user_id,
self._threshold,
self._max_facts,
)
# 3) 관련 과거 세션 검색
sessions = await conn.fetch(
"""SELECT title, summary,
1 - (embedding <=> $1::vector) AS similarity
FROM conversation_archive
WHERE user_id = $2
AND 1 - (embedding <=> $1::vector) > $3
ORDER BY similarity DESC
LIMIT $4""",
query_embedding,
user_id,
self._threshold,
self._max_sessions,
)
# 사실 인출 시 access_count 증가 (중요도 학습)
if facts:
fact_ids = [f["fact_id"] for f in facts if "fact_id" in f]
if fact_ids:
await conn.execute(
"""UPDATE long_term_facts
SET access_count = access_count + 1
WHERE fact_id = ANY($1)""",
fact_ids,
)
return LongTermContext(
user_profile={
"preferences": profile["preferences"] if profile else {},
"tech_context": profile["tech_context"] if profile else {},
},
relevant_facts=[dict(f) for f in facts],
related_sessions=[dict(s) for s in sessions],
total_tokens=0, # 실제 토큰 카운팅 필요
)
메모리 통합 — 컨텍스트 어셈블러
8화에서 설계한 컨텍스트 조립 파이프라인에 메모리 계층을 통합합니다. 세 계층의 메모리가 프롬프트의 어디에 들어가는지가 핵심입니다.
graph LR
subgraph "프롬프트 구조 (토큰 예산 배분)"
A["시스템 프롬프트
~1,500 토큰"]
B["사용자 프로파일
(장기 메모리)
~500 토큰"]
C["세션 요약 + 핵심 사실
(세션 메모리)
~2,000 토큰"]
D["RAG 검색 결과
(외부 지식)
~4,000 토큰"]
E["관련 과거 대화
(장기 메모리)
~1,000 토큰"]
F["이전 대화 요약
(단기 메모리 압축)
~1,500 토큰"]
G["최근 대화
(단기 메모리 원문)
~4,000 토큰"]
H["현재 사용자 메시지
~500 토큰"]
A --> B --> C --> D --> E --> F --> G --> H
end
"""
컨텍스트 어셈블러 — 3계층 메모리 + RAG 통합
8화의 프롬프트 파이프라인에 메모리를 결합
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
@dataclass
class AssembledContext:
"""조립된 최종 프롬프트 컨텍스트."""
messages: list[dict[str, str]]
total_tokens: int
budget_usage: dict[str, int] # 슬롯별 토큰 사용량
class ContextAssembler:
"""3계층 메모리 + RAG를 통합하여 프롬프트 조립."""
# 토큰 예산 배분 (Qwen3-30B-A3B, 32K 컨텍스트 기준)
BUDGET = {
"system_prompt": 1_500,
"user_profile": 500,
"session_context": 2_000,
"rag_results": 4_000,
"long_term_memory": 1_000,
"conversation_summary": 1_500,
"recent_turns": 4_000,
"current_message": 500,
"response_reserve": 3_000, # 응답 생성용 예약
}
# 합계: 18,000 토큰 — 32K 윈도우의 ~56%
def __init__(
self,
system_prompt: str,
working_memory: Any, # WorkingMemory 인스턴스
session_retriever: Any, # SessionMemoryRetriever
longterm_retriever: Any, # LongTermMemoryRetriever
rag_retriever: Any, # RAG 검색기 (9~10화)
) -> None:
self._system_prompt = system_prompt
self._working = working_memory
self._session = session_retriever
self._longterm = longterm_retriever
self._rag = rag_retriever
async def assemble(
self,
user_id: str,
session_id: str,
current_message: str,
) -> AssembledContext:
"""모든 소스에서 컨텍스트를 수집하여 프롬프트 조립."""
budget_usage: dict[str, int] = {}
system_parts: list[str] = [self._system_prompt]
# 1) 장기 메모리 — 사용자 프로파일 + 관련 사실
lt_ctx = await self._longterm.retrieve(user_id, current_message)
profile_block = lt_ctx.to_prompt_block()
if profile_block:
system_parts.append(profile_block)
# 2) 세션 메모리 — 세션 요약 + 핵심 사실
session_ctx = await self._session.retrieve(session_id)
session_block = session_ctx.to_prompt_block()
if session_block:
system_parts.append(session_block)
# 3) RAG 검색 결과 (9~10화 파이프라인)
rag_results = await self._rag.search(current_message)
if rag_results:
rag_block = self._format_rag(rag_results)
system_parts.append(rag_block)
# 4) 시스템 메시지 조립
messages: list[dict[str, str]] = [
{"role": "system", "content": "\n\n".join(system_parts)}
]
# 5) 단기 메모리 — 요약 + 최근 턴
context_messages = self._working.get_context_messages()
messages.extend(context_messages)
# 6) 현재 메시지
messages.append({"role": "user", "content": current_message})
return AssembledContext(
messages=messages,
total_tokens=0, # 실제 카운팅 필요
budget_usage=budget_usage,
)
def _format_rag(self, results: list[Any]) -> str:
"""RAG 결과를 프롬프트 블록으로 포맷."""
chunks = []
for r in results[:5]: # 상위 5건
chunks.append(
f"[출처: {r.get('source', '알 수 없음')}]\n{r.get('content', '')}"
)
return "[참고 자료]\n" + "\n---\n".join(chunks)
메모리 생명주기 관리
세션 종료 시 장기 메모리 승격
세션이 종료되면 세션 메모리에서 장기 메모리로 중요한 정보를 ‘승격(promote)’합니다. 이 과정을 자동화합니다.
"""
세션 종료 시 장기 메모리 승격 워커
비동기 백그라운드 태스크로 실행
"""
from __future__ import annotations
from typing import Any
class MemoryPromotionWorker:
"""세션 → 장기 메모리 승격."""
def __init__(
self,
db_pool: Any,
embedding_fn: Any,
summarizer_fn: Any, # Qwen3-8B 요약 함수
) -> None:
self._pool = db_pool
self._embed = embedding_fn
self._summarize = summarizer_fn
async def promote_session(self, session_id: str, user_id: str) -> None:
"""세션 종료 시 호출. 장기 메모리로 승격."""
async with self._pool.acquire() as conn:
# 1) 세션 전체 요약 생성
turns = await conn.fetch(
"""SELECT role, content FROM session_turns
WHERE session_id = $1 ORDER BY turn_number""",
session_id,
)
if not turns:
return
full_conversation = "\n".join(
f"[{t['role']}] {t['content'][:500]}" for t in turns
)
session_summary = await self._summarize(full_conversation)
summary_embedding = await self._embed(session_summary)
# 2) 대화 아카이브에 저장
session_meta = await conn.fetchrow(
"SELECT title FROM sessions WHERE session_id = $1",
session_id,
)
await conn.execute(
"""INSERT INTO conversation_archive
(user_id, session_id, title, summary, turn_count, embedding)
VALUES ($1, $2, $3, $4, $5, $6)""",
user_id,
session_id,
session_meta["title"] if session_meta else None,
session_summary,
len(turns),
summary_embedding,
)
# 3) 세션 사실 중 confidence 높은 것을 장기 사실로 승격
session_facts = await conn.fetch(
"""SELECT fact_type, content, confidence
FROM session_facts
WHERE session_id = $1 AND confidence >= 0.7""",
session_id,
)
for fact in session_facts:
fact_embedding = await self._embed(fact["content"])
# 기존 유사 사실이 있으면 업데이트, 없으면 새로 생성
existing = await conn.fetchrow(
"""SELECT fact_id, content, confidence
FROM long_term_facts
WHERE user_id = $1
AND 1 - (embedding <=> $2::vector) > 0.85
ORDER BY 1 - (embedding <=> $2::vector) DESC
LIMIT 1""",
user_id,
fact_embedding,
)
if existing:
# 기존 사실 업데이트 (최신 우선)
await conn.execute(
"""UPDATE long_term_facts
SET content = $1,
confidence = $2,
updated_at = now(),
source_session = $3
WHERE fact_id = $4""",
fact["content"],
max(fact["confidence"], existing["confidence"]),
session_id,
existing["fact_id"],
)
else:
# 새 사실 생성
await conn.execute(
"""INSERT INTO long_term_facts
(user_id, category, content, source_session,
confidence, embedding)
VALUES ($1, $2, $3, $4, $5, $6)""",
user_id,
fact["fact_type"],
fact["content"],
session_id,
fact["confidence"],
fact_embedding,
)
# 4) 사용자 프로파일 통계 갱신
await conn.execute(
"""UPDATE user_profiles
SET total_sessions = total_sessions + 1,
total_turns = total_turns + $1,
last_active_at = now(),
updated_at = now()
WHERE user_id = $2""",
len(turns),
user_id,
)
# 5) 세션 상태를 archived로 변경
await conn.execute(
"""UPDATE sessions
SET status = 'archived', updated_at = now()
WHERE session_id = $1""",
session_id,
)
장기 메모리 정리 정책
장기 메모리가 무한 증가하면 인출 성능이 떨어지고 저장 비용이 늘어납니다. 주기적 정리가 필요합니다.
"""
장기 메모리 정리 — 주기적 배치 작업
cron 또는 APScheduler로 매일 실행
"""
from __future__ import annotations
from typing import Any
class MemoryCleanupJob:
"""장기 메모리 정리 정책."""
def __init__(self, db_pool: Any) -> None:
self._pool = db_pool
async def run(self) -> dict[str, int]:
"""정리 실행. 삭제된 건수 반환."""
stats: dict[str, int] = {}
async with self._pool.acquire() as conn:
# 1) 만료된 사실 삭제
result = await conn.execute(
"""DELETE FROM long_term_facts
WHERE expires_at IS NOT NULL
AND expires_at < now()"""
)
stats["expired_facts"] = int(result.split()[-1])
# 2) 저신뢰 + 미사용 사실 정리
# confidence < 0.3 이고 90일간 한 번도 인출되지 않은 사실
result = await conn.execute(
"""DELETE FROM long_term_facts
WHERE confidence < 0.3
AND access_count = 0
AND updated_at < now() - INTERVAL '90 days'"""
)
stats["low_confidence_unused"] = int(result.split()[-1])
# 3) 오래된 대화 아카이브 요약만 유지
# 1년 이상 된 아카이브는 요약만 남기고 상세 턴 삭제
result = await conn.execute(
"""DELETE FROM session_turns
WHERE session_id IN (
SELECT session_id FROM sessions
WHERE status = 'archived'
AND updated_at < now() - INTERVAL '365 days'
)"""
)
stats["old_turns_deleted"] = int(result.split()[-1])
# 4) 만료된 세션 정리
result = await conn.execute(
"""UPDATE sessions
SET status = 'expired'
WHERE status = 'active'
AND expires_at IS NOT NULL
AND expires_at < now()"""
)
stats["expired_sessions"] = int(result.split()[-1])
return stats
프라이버시와 데이터 보존 정책 — 금융IT 관점
온프레미스 AI Assistant의 메모리에는 사용자의 업무 대화, 의사결정 기록, 기술적 맥락이 누적됩니다. 금융IT·규제 산업에서는 이 데이터의 관리가 특히 중요합니다.
데이터 분류와 보존 기간
| 데이터 유형 | 민감도 | 보존 기간 | 접근 통제 |
|---|---|---|---|
| 대화 원문 (session_turns) | 높음 | 활성 세션 + 90일 아카이브 | 본인 + 관리자 |
| 대화 요약 (session_summaries) | 중간 | 365일 | 본인 + 관리자 |
| 사용자 프로파일 | 중간 | 계정 존속 기간 | 본인 + 시스템 |
| 추출된 사실 (facts) | 중간~높음 | 명시적 삭제 또는 정책 만료 | 본인 + 시스템 |
| 대화 아카이브 요약 | 낮음~중간 | 365일 | 본인 |
| 임베딩 벡터 | 낮음 (역복원 어려움) | 원본과 동일 | 시스템 |
규제 환경 설계 원칙
# memory_policy.yaml — 메모리 보존·접근 정책 설정
# 금융IT 환경 예시 (A 금융사 내부 가이드라인 기반 익명화)
retention:
# 대화 원문 보존
conversation_raw:
active_session: "unlimited" # 세션 활성 중
after_archive: "90d" # 아카이브 후 90일 보존
audit_log: "5y" # 감사 로그는 5년 (금융 규제)
# 요약·사실
summaries: "365d"
facts:
default: "365d"
with_pii: "90d" # PII 포함 사실은 90일
decision_record: "3y" # 의사결정 기록은 3년
# 사용자 프로파일
user_profile:
active: "unlimited"
after_deactivation: "90d" # 계정 비활성 후 90일
privacy:
# PII 자동 탐지 및 마스킹
pii_detection:
enabled: true
patterns:
- type: "resident_number" # 주민등록번호
action: "mask" # 저장 전 마스킹
- type: "phone_number"
action: "mask"
- type: "email"
action: "hash" # 해시로 대체
- type: "account_number" # 계좌번호
action: "mask"
- type: "card_number" # 카드번호
action: "redact" # 완전 제거
# 사용자 데이터 삭제 권한 (잊힐 권리)
right_to_forget:
enabled: true
scope: "all" # 해당 사용자의 모든 메모리
audit_trail: true # 삭제 이력은 감사 로그에 기록
access_control:
# 메모리 접근 권한
own_data: "read_write_delete" # 본인 데이터
admin: "read_only" # 관리자는 읽기만
system: "read_write" # 시스템(인출기)
audit: "read_only" # 감사 목적
encryption:
at_rest:
enabled: true
algorithm: "AES-256-GCM"
key_management: "local_hsm" # 온프레미스 HSM 또는 소프트웨어 키 관리
in_transit:
tls_version: "1.3"
internal_network: "required" # 내부 네트워크도 TLS 필수
audit:
# 감사 로그 — 메모리 접근/수정 이력
log_access: true # 메모리 인출 시 로깅
log_modification: true # 메모리 수정/삭제 시 로깅
log_retention: "5y" # 감사 로그 5년 보존
tamper_proof: true # 변조 방지 (append-only 로그)
PII 마스킹 적용 시점
PII(개인식별정보) 마스킹은 메모리 저장 시점에 수행합니다. 인출 시점이 아닙니다. 이유는 명확합니다 — 저장된 상태에서 DB 백업·덤프·복제 시 PII가 노출되는 것을 원천 차단해야 하기 때문입니다.
"""
PII 탐지 및 마스킹 — 메모리 저장 전 파이프라인
"""
from __future__ import annotations
import re
from dataclasses import dataclass
@dataclass
class PIIDetection:
pii_type: str
original: str
masked: str
start: int
end: int
class PIIMasker:
"""한국 금융IT 환경 PII 탐지·마스킹."""
# 패턴 정의 (한국 기준)
PATTERNS: list[tuple[str, str, str]] = [
# (pii_type, regex_pattern, mask_strategy)
(
"resident_number",
r"\b(\d{6})-?(\d{7})\b",
"mask_resident",
),
(
"phone_number",
r"\b(01[016789])-?(\d{3,4})-?(\d{4})\b",
"mask_phone",
),
(
"card_number",
r"\b(\d{4})-?(\d{4})-?(\d{4})-?(\d{4})\b",
"mask_card",
),
(
"account_number",
r"\b(\d{3,4})-(\d{2,6})-(\d{4,8})\b",
"mask_account",
),
(
"email",
r"\b([a-zA-Z0-9._%+-]+)@([a-zA-Z0-9.-]+\.[a-zA-Z]{2,})\b",
"mask_email",
),
]
def mask(self, text: str) -> tuple[str, list[PIIDetection]]:
"""텍스트에서 PII를 탐지하고 마스킹. (마스킹된 텍스트, 탐지 목록) 반환."""
detections: list[PIIDetection] = []
masked_text = text
for pii_type, pattern, strategy in self.PATTERNS:
for match in re.finditer(pattern, text):
original = match.group()
masked = getattr(self, f"_{strategy}")(match)
masked_text = masked_text.replace(original, masked, 1)
detections.append(PIIDetection(
pii_type=pii_type,
original="[REDACTED]", # 원본은 로그에도 남기지 않음
masked=masked,
start=match.start(),
end=match.end(),
))
return masked_text, detections
@staticmethod
def _mask_resident(match: re.Match[str]) -> str:
return f"{match.group(1)}-*******"
@staticmethod
def _mask_phone(match: re.Match[str]) -> str:
return f"{match.group(1)}-****-{match.group(3)}"
@staticmethod
def _mask_card(match: re.Match[str]) -> str:
return f"{match.group(1)}-****-****-{match.group(4)}"
@staticmethod
def _mask_account(match: re.Match[str]) -> str:
return f"***-**-{match.group(3)[-4:]}"
@staticmethod
def _mask_email(match: re.Match[str]) -> str:
local = match.group(1)
domain = match.group(2)
masked_local = local[0] + "***" if len(local) > 1 else "***"
return f"{masked_local}@{domain}"
잊힐 권리 (Right to Forget) 구현
사용자가 자신의 메모리 삭제를 요청하면, 모든 계층에서 해당 사용자의 데이터를 제거합니다. 금융 규제상 삭제 이력 자체는 감사 로그에 남겨야 합니다.
async def forget_user(
db_pool: Any,
user_id: str,
reason: str = "user_request",
) -> dict[str, int]:
"""사용자의 모든 메모리를 삭제. 감사 로그는 보존."""
stats: dict[str, int] = {}
async with db_pool.acquire() as conn:
async with conn.transaction():
# 1) 장기 사실 삭제
r = await conn.execute(
"DELETE FROM long_term_facts WHERE user_id = $1", user_id
)
stats["facts_deleted"] = int(r.split()[-1])
# 2) 대화 아카이브 삭제
r = await conn.execute(
"DELETE FROM conversation_archive WHERE user_id = $1", user_id
)
stats["archives_deleted"] = int(r.split()[-1])
# 3) 세션 데이터 삭제 (턴 → 사실 → 요약 → 세션)
session_ids = await conn.fetch(
"SELECT session_id FROM sessions WHERE user_id = $1", user_id
)
sids = [s["session_id"] for s in session_ids]
if sids:
for table in ("session_facts", "session_summaries", "session_turns"):
await conn.execute(
f"DELETE FROM {table} WHERE session_id = ANY($1)", sids # noqa: S608
)
r = await conn.execute(
"DELETE FROM sessions WHERE user_id = $1", user_id
)
stats["sessions_deleted"] = int(r.split()[-1])
# 4) 사용자 프로파일 삭제
await conn.execute(
"DELETE FROM user_profiles WHERE user_id = $1", user_id
)
stats["profile_deleted"] = 1
# 5) 감사 로그 기록 (삭제 자체의 기록)
await conn.execute(
"""INSERT INTO audit_log
(action, target_user, reason, details, created_at)
VALUES ('forget_user', $1, $2, $3::jsonb, now())""",
user_id,
reason,
str(stats), # JSON으로 변환 필요
)
return stats
인출 전략 — 언제, 무엇을 꺼내는가
메모리가 있어도 인출 전략이 잘못되면 무용지물입니다. 매 요청마다 모든 메모리를 검색하는 것은 비효율적이고, 관련 없는 기억이 프롬프트에 들어가면 오히려 응답 품질이 떨어집니다.
계층별 인출 트리거
| 메모리 계층 | 인출 트리거 | 인출 방법 | 비용 |
|---|---|---|---|
| 단기 (Working) | 매 요청 | 전량 주입 (슬라이딩 윈도우 범위) | 낮음 (인메모리) |
| 세션 (Session) | 매 요청 | 최신 요약 + 핵심 사실 (캐시 가능) | 낮음 (1회 DB 조회) |
| 장기 프로파일 | 세션 시작 시 1회 | 사용자 ID 조회 → 세션 캐시 | 매우 낮음 |
| 장기 사실 | 조건부 (관련성 높을 때) | 의미 검색 (bge-m3) | 중간 (임베딩 + 벡터 검색) |
| 대화 아카이브 | 조건부 (과거 참조 시) | 의미 검색 (bge-m3) | 중간 |
조건부 인출 — 언제 장기 메모리를 검색할 것인가
장기 메모리 검색은 bge-m3 임베딩 + Qdrant/pgvector 조회가 필요하므로, 매 요청마다 하면 지연이 누적됩니다. 다음 조건 중 하나라도 만족할 때만 트리거합니다.
"""
장기 메모리 인출 트리거 판별기
"""
from __future__ import annotations
import re
class MemoryTriggerDetector:
"""사용자 메시지를 분석하여 장기 메모리 인출 필요 여부 판단."""
# 과거 참조 패턴 (한국어)
PAST_REFERENCE_PATTERNS: list[re.Pattern[str]] = [
re.compile(r"(지난번|저번|예전|이전|과거|그때|전에|아까)"),
re.compile(r"(기억|remember|recall)"),
re.compile(r"(했었|했던|말했|물어봤|알려줬)"),
re.compile(r"(다시|또|반복).*해"),
re.compile(r"(항상|늘|매번|보통).*하"),
]
# 프로파일 참조 패턴
PROFILE_PATTERNS: list[re.Pattern[str]] = [
re.compile(r"(내가|나는|우리|저는).*쓰"),
re.compile(r"(우리\s*(팀|시스템|서버|환경))"),
re.compile(r"(선호|스타일|방식|좋아하는)"),
]
def should_retrieve_long_term(
self,
user_message: str,
session_turn_count: int,
) -> tuple[bool, str]:
"""
장기 메모리 인출 여부와 이유 반환.
Returns: (should_retrieve, reason)
"""
# 1) 과거 참조 키워드 탐지
for pattern in self.PAST_REFERENCE_PATTERNS:
if pattern.search(user_message):
return True, "past_reference"
# 2) 프로파일 참조 탐지
for pattern in self.PROFILE_PATTERNS:
if pattern.search(user_message):
return True, "profile_reference"
# 3) 세션 첫 번째 턴 (맥락 파악 필요)
if session_turn_count == 0:
return True, "session_start"
# 4) 주기적 인출 (매 10턴마다 — 드리프트 방지)
if session_turn_count % 10 == 0:
return True, "periodic_refresh"
return False, "not_needed"
이 트리거 기반 접근은 Qwen3-8B로 의도 분류하는 것보다 훨씬 가볍습니다. 정규식 기반이므로 지연이 0에 가깝고, 약간의 false positive가 있어도 불필요한 컨텍스트는 프롬프트 예산 관리에서 걸러집니다.
메모리 일관성 문제와 해결
사실 충돌 해결
장기 메모리에 "Java 17을 사용한다"는 사실이 있는데, 새 세션에서 "Java 21로 마이그레이션했다"고 하면 어떻게 할까요? 최신 우선(recency-first) 원칙을 적용합니다.
async def resolve_fact_conflict(
db_pool: Any,
user_id: str,
new_fact: str,
new_embedding: list[float],
category: str,
session_id: str,
) -> str:
"""
새 사실과 충돌하는 기존 사실이 있으면 업데이트.
Returns: "created" | "updated" | "skipped"
"""
async with db_pool.acquire() as conn:
# 유사도 0.85 이상인 기존 사실 검색
existing = await conn.fetchrow(
"""SELECT fact_id, content, confidence,
1 - (embedding <=> $1::vector) AS similarity
FROM long_term_facts
WHERE user_id = $2 AND category = $3
AND 1 - (embedding <=> $1::vector) > 0.85
ORDER BY similarity DESC
LIMIT 1""",
new_embedding, user_id, category,
)
if existing:
if existing["content"] == new_fact:
return "skipped" # 동일 내용 — 무시
# 기존 사실을 새 사실로 대체 (최신 우선)
await conn.execute(
"""UPDATE long_term_facts
SET content = $1,
embedding = $2,
source_session = $3,
confidence = GREATEST(confidence, 0.8),
updated_at = now()
WHERE fact_id = $4""",
new_fact, new_embedding, session_id, existing["fact_id"],
)
return "updated"
else:
# 새 사실 생성
await conn.execute(
"""INSERT INTO long_term_facts
(user_id, category, content, source_session,
confidence, embedding)
VALUES ($1, $2, $3, $4, 0.8, $5)""",
user_id, category, new_fact, session_id, new_embedding,
)
return "created"
메모리 드리프트 방지
점진적 요약이 반복되면 원래 의미에서 조금씩 벗어나는 의미 드리프트(semantic drift)가 발생할 수 있습니다. 이를 방지하는 방법:
- 앵커 사실(anchor facts) — 핵심 사실은 요약에 포함시키지 않고 별도 저장소에 원문 보존
- 주기적 재요약 — 누적 요약이 10회 이상 업데이트되면 원본 턴에서 다시 요약 생성
- 요약 체인 길이 제한 — "요약의 요약의 요약"은 3단계까지만 허용, 이후는 원본 턴 참조
운영 설정 — 환경별 메모리 프로파일
# memory_config.yaml — 메모리 아키텍처 설정
# Qwen3 기반 온프레미스 AI Assistant
# --- 단기 메모리 ---
working_memory:
max_tokens: 12000 # 컨텍스트 예산 중 대화 이력 슬롯
summary_threshold: 0.8 # 80% 차면 요약 트리거
min_recent_turns: 4 # 최소 유지 턴 수
summarizer:
model: "qwen3-8b" # 요약 전용 경량 모델
endpoint: "http://gateway:8000/v1"
max_summary_tokens: 500 # 요약 최대 길이
# --- 세션 메모리 ---
session_memory:
store:
backend: "postgresql" # "postgresql" | "sqlite" | "redis"
dsn: "${BRIDGE_SESSION_DSN}"
ttl:
active: "24h" # 활성 세션 기본 TTL
idle_timeout: "2h" # 유휴 시 만료
fact_extraction:
enabled: true
trigger: "every_3_turns" # "every_turn" | "every_3_turns" | "conditional"
model: "qwen3-8b"
min_confidence: 0.5
max_context_tokens: 2000 # 세션 컨텍스트 토큰 예산
# --- 장기 메모리 ---
long_term_memory:
store:
backend: "postgresql" # pgvector 확장 필요
dsn: "${MEMORY_PG_DSN}"
embedding:
model: "bge-m3"
endpoint: "http://embedding:8080/v1"
dimension: 1024
retrieval:
max_facts: 10
max_sessions: 3
similarity_threshold: 0.35
trigger: "conditional" # "always" | "conditional"
cleanup:
schedule: "0 3 * * *" # 매일 새벽 3시
low_confidence_days: 90
archive_retention_days: 365
# --- 프라이버시 ---
privacy:
pii_masking:
enabled: true
apply_at: "storage" # "storage" | "retrieval" | "both"
right_to_forget: true
encryption_at_rest: true
# --- 성능 ---
performance:
# 프로파일 캐시 (세션 시작 시 로드, 세션 동안 유지)
profile_cache_ttl: "session"
# 사실 검색 결과 캐시 (같은 쿼리 반복 방지)
fact_cache_ttl: "60s"
# 요약 생성 비동기 (응답 지연에 영향 없음)
async_summarization: true
성능 측정 기준점
메모리 아키텍처의 각 구성 요소가 추론 지연에 미치는 영향을 측정합니다. 아래는 PostgreSQL(NAS 호스팅, Ryzen 5 NAS, SSD) + Qwen3-8B(Mac Studio M2 Ultra, MLX) 기준 측정값입니다.
| 연산 | 지연 (p50) | 지연 (p99) | 비고 |
|---|---|---|---|
| 단기 메모리 조회 | <1ms | <1ms | 인메모리, 무시 가능 |
| 세션 컨텍스트 조회 | 3ms | 12ms | PostgreSQL 인덱스 조회 |
| 사용자 프로파일 조회 | 2ms | 5ms | 세션 시작 시 1회, 이후 캐시 |
| 장기 사실 벡터 검색 | 15ms | 45ms | pgvector IVFFlat, 사용자당 ~1000건 |
| 대화 아카이브 검색 | 12ms | 35ms | pgvector, 사용자당 ~200건 |
| bge-m3 임베딩 생성 | 25ms | 60ms | 쿼리 1건, Mac Studio MLX |
| Qwen3-8B 요약 생성 | 800ms | 2,500ms | 비동기 — 응답 지연에 불포함 |
| Qwen3-8B 사실 추출 | 600ms | 1,800ms | 비동기 — 응답 지연에 불포함 |
| 전체 메모리 인출 (조건부) | ~50ms | ~120ms | 프로파일(캐시) + 사실 + 아카이브 |
| 전체 메모리 인출 (단순) | ~5ms | ~15ms | 캐시 프로파일 + 세션 요약만 |
핵심 관찰: 조건부 인출(장기 메모리 벡터 검색 포함)도 p99 기준 120ms로, Qwen3의 추론 시간(수 초)에 비하면 무시 가능한 수준입니다. 요약·사실 추출은 비동기이므로 사용자 체감 지연에 영향 없습니다.
운영 함정 (Pitfall) 미니 코너
함정: "모든 것을 기억하면 좋겠지" 증후군
장기 메모리에 사실을 너무 적극적으로 저장하면 예상치 못한 부작용이 생깁니다. 한 프로젝트에서 매 턴마다 사실 추출을 활성화했더니, 사용자당 평균 3,000개의 사실이 누적되었습니다. 결과적으로:
- 벡터 검색 정확도 하락 — 유사한 사실이 너무 많아 관련성 높은 것이 묻힘
- 모순 증가 — "Java 17 사용" → "Java 21 마이그레이션" → "일부 서비스는 아직 Java 17" 같은 부분 모순이 누적
- 프롬프트 오염 — 현재 맥락과 무관한 과거 사실이 주입되어 응답이 엉뚱한 방향으로 감
해결책: confidence 임계치를 0.7 이상으로 올리고, access_count가 0인 채 90일 경과한 사실은 자동 삭제합니다. 사실의 양보다 질과 최신성이 중요합니다. "기억하지 않는 것"도 메모리 설계의 핵심입니다.
마무리 — 기억이 있는 Assistant
오늘 설계한 3계층 메모리 아키텍처를 정리합니다.
- 단기 메모리: 토큰 예산 기반 슬라이딩 윈도우 + Qwen3-8B 점진적 요약으로 현재 대화의 맥락을 유지
- 세션 메모리: PostgreSQL에 턴·요약·핵심 사실을 구조화 저장, 세션 수준의 결정사항과 맥락을 보존
- 장기 메모리: 사용자 프로파일 + 사실 저장소 + 대화 아카이브를 pgvector 의미 검색으로 인출, 조건부 트리거로 불필요한 검색 최소화
- 프라이버시: PII 마스킹, 잊힐 권리, 보존 기간 정책, 감사 로그 — 금융IT 규제 환경에서도 운영 가능한 설계
8화의 프롬프트 아키텍처, 9~10화의 RAG 파이프라인, 그리고 오늘의 메모리 아키텍처를 결합하면, Qwen3 기반 Assistant는 "외부 지식을 검색하고, 과거를 기억하며, 사용자를 이해하는" 시스템이 됩니다.
내일 12화에서는 이 모든 것 위에 에이전트 레이어를 올립니다 — Qwen3의 function calling과 MCP 기반 사내 시스템 연동, ReAct·Plan-Execute 패턴, 그리고 Qwen3-VL GUI 에이전트까지. 기억하고 검색하는 것을 넘어, 실제로 행동하는 Assistant를 만듭니다.
📚 시리즈: 온프레미스 AI Assistant 아키텍처 — Qwen3·Qwen3-VL 14일 설계 (총 14화 중 11화)
◀ 이전 10화 (다음 차수는 아직 게시되지 않았습니다)
◀ 이전 10화 (다음 차수는 아직 게시되지 않았습니다)
참고 자료
- Memory management — Wikipedia — 컴퓨팅 시스템의 메모리 관리 개념과 계층 구조에 대한 개괄
- A Survey on the Memory Mechanism of Large Language Model based Agents (arXiv 2404.13501) — LLM 에이전트의 메모리 메커니즘을 단기·장기·검색 증강 관점에서 분류·정리한 서베이 논문
[…] [온프레미스 AI Assistant 아키텍처 — Qwen3·Qwen3-VL 14일 설계] 11/14화: LLM 메모리… […]
[…] 온프레미스 AI Assistant 아키텍처 — Qwen3·Qwen3-VL 14일 설계 (총 14화 중 12화)◀ 이전 11화 (다음 차수는 아직 게시되지 […]