이미지 로딩 중...
AI Generated
2025. 11. 8. · 3 Views
Python 챗봇 개발 5편 - 대화 컨텍스트 관리
챗봇이 이전 대화를 기억하고 자연스러운 대화를 이어갈 수 있도록 컨텍스트를 관리하는 방법을 배웁니다. 메모리 관리, 세션 처리, 컨텍스트 윈도우 최적화까지 실무에 필요한 모든 기법을 다룹니다.
목차
- 대화 컨텍스트란 - 챗봇이 대화를 기억하는 방법
- 세션 기반 컨텍스트 관리 - 사용자별로 대화 분리하기
- 컨텍스트 윈도우 최적화 - 토큰 제한 극복하기
- 중요 정보 영구 저장 - 핵심 컨텍스트 추출하기
- 컨텍스트 압축과 요약 - 오래된 대화 효율적으로 관리하기
- 멀티턴 대화 플로우 관리 - 상태 기반 대화 처리하기
- 컨텍스트 우선순위 관리 - 중요한 메시지 우선 보존하기
- 컨텍스트 공유와 동기화 - 멀티 채널 대화 통합하기
1. 대화 컨텍스트란 - 챗봇이 대화를 기억하는 방법
시작하며
여러분이 챗봇을 개발할 때 이런 상황을 겪어본 적 있나요? 사용자가 "그거 어떻게 해?"라고 물었는데, 챗봇이 "무엇을 말씀하시는 건가요?"라고 되묻는 상황 말이에요.
이런 문제는 챗봇이 이전 대화 내용을 기억하지 못하기 때문에 발생합니다. 사람은 자연스럽게 "그거", "그것", "아까 말한 것"처럼 앞의 내용을 참조하며 대화하지만, 컨텍스트 관리가 없는 챗봇은 매번 새로운 대화처럼 처리합니다.
바로 이럴 때 필요한 것이 대화 컨텍스트 관리입니다. 이전 대화 내용을 저장하고 참조하여, 마치 사람처럼 자연스러운 대화 흐름을 만들 수 있습니다.
개요
간단히 말해서, 대화 컨텍스트는 챗봇이 기억해야 할 이전 대화 내용의 집합입니다. 이 개념이 필요한 이유는 실제 사용자들이 챗봇과 대화할 때 단발성 질문만 하는 것이 아니기 때문입니다.
"날씨 알려줘" 다음에 "내일은?"이라고 물을 수 있고, "Python 함수 알려줘" 다음에 "예제 좀 더 줘"라고 요청할 수 있습니다. 예를 들어, 고객 상담 챗봇에서 "주문 조회"를 요청한 후 "취소하고 싶어요"라고 말했을 때, 챗봇이 어떤 주문을 취소하려는지 알아야 합니다.
기존에는 매 요청마다 모든 정보를 다시 입력해야 했다면, 이제는 컨텍스트를 통해 자연스러운 대화가 가능합니다. 컨텍스트의 핵심 특징은 첫째, 대화 이력을 순서대로 저장한다는 것, 둘째, 필요할 때 이전 정보를 참조할 수 있다는 것, 셋째, 메모리 효율을 위해 오래된 대화는 제거할 수 있다는 것입니다.
이러한 특징들이 챗봇을 단순한 질의응답 시스템에서 대화형 에이전트로 진화시킵니다.
코드 예제
# 대화 컨텍스트를 저장하는 기본 클래스
class ConversationContext:
def __init__(self, max_history=10):
# 대화 이력을 저장할 리스트
self.messages = []
# 최대 저장할 메시지 수
self.max_history = max_history
def add_message(self, role, content):
# 새 메시지를 추가 (role: 'user' 또는 'assistant')
self.messages.append({"role": role, "content": content})
# 오래된 메시지 제거 (메모리 관리)
if len(self.messages) > self.max_history:
self.messages = self.messages[-self.max_history:]
def get_context(self):
# 전체 대화 컨텍스트 반환
return self.messages
설명
이것이 하는 일: ConversationContext 클래스는 챗봇의 대화 이력을 저장하고 관리하는 컨테이너 역할을 합니다. 마치 사람의 단기 기억처럼, 최근 대화 내용을 일정 개수만큼 보관합니다.
첫 번째로, __init__ 메서드에서 빈 리스트 messages와 최대 이력 개수 max_history를 초기화합니다. 이렇게 하는 이유는 무한정 대화를 저장하면 메모리 문제가 발생하기 때문입니다.
실무에서는 보통 10-20개 정도의 최근 대화만 유지합니다. 그 다음으로, add_message 메서드가 실행되면서 새로운 메시지를 딕셔너리 형태로 리스트에 추가합니다.
내부에서는 role(역할)과 content(내용)를 구분하여 저장하는데, 이는 OpenAI API나 대부분의 LLM API 형식과 일치합니다. 메시지 추가 후에는 자동으로 오래된 메시지를 제거하는 로직이 실행됩니다.
마지막으로, get_context 메서드가 전체 대화 이력을 반환하여 최종적으로 API 호출 시 함께 전달할 수 있습니다. 이를 통해 LLM은 이전 대화 맥락을 이해하고 적절한 답변을 생성합니다.
여러분이 이 코드를 사용하면 챗봇이 "그것", "아까", "더"와 같은 참조 표현을 이해할 수 있게 됩니다. 실무에서의 이점은 첫째, 자연스러운 대화 흐름을 만들 수 있고, 둘째, 사용자가 매번 전체 문맥을 설명할 필요가 없으며, 셋째, 메모리를 효율적으로 관리할 수 있다는 것입니다.
실전 팁
💡 max_history는 사용하는 LLM의 토큰 제한과 평균 메시지 길이를 고려해서 설정하세요. GPT-3.5는 4K 토큰이므로 보통 10-15개가 적당하지만, GPT-4는 8K 이상이므로 20-30개도 가능합니다.
💡 messages 리스트를 직접 수정하지 말고 반드시 add_message 메서드를 통해 추가하세요. 이렇게 해야 자동 정리 로직이 작동하고 나중에 유효성 검사 로직을 추가하기도 쉽습니다.
💡 중요한 대화는 별도로 저장하세요. 예를 들어 사용자 이름, 선호도 같은 정보는 max_history에 관계없이 영구 저장소에 보관해야 합니다.
💡 개발 단계에서는 get_context()로 반환되는 내용을 로그로 출력해보세요. 어떤 컨텍스트가 LLM에 전달되는지 확인하면 디버깅이 훨씬 쉬워집니다.
2. 세션 기반 컨텍스트 관리 - 사용자별로 대화 분리하기
시작하며
여러분이 실제 서비스용 챗봇을 만들 때 이런 문제를 만나게 됩니다. A 사용자가 "Python 코드 보여줘"라고 물었는데, B 사용자가 동시에 "JavaScript 예제 줘"라고 물으면 어떻게 될까요?
이런 문제는 다중 사용자 환경에서 컨텍스트가 섞이기 때문에 발생합니다. 웹 서비스에서는 수십, 수백 명의 사용자가 동시에 챗봇을 사용하는데, 각자의 대화 내용이 뒤섞이면 안 됩니다.
A의 대화가 B에게 보이거나, 더 심각하게는 A의 개인정보가 B에게 노출될 수도 있습니다. 바로 이럴 때 필요한 것이 세션 기반 컨텍스트 관리입니다.
각 사용자별로 독립된 세션 ID를 부여하고, 세션별로 별도의 컨텍스트를 유지하여 완벽하게 격리된 대화 환경을 만들 수 있습니다.
개요
간단히 말해서, 세션 기반 컨텍스트 관리는 사용자별로 독립된 대화 공간을 만드는 기술입니다. 이 개념이 필요한 이유는 실제 프로덕션 환경에서는 항상 여러 사용자가 동시에 접속하기 때문입니다.
Flask나 FastAPI로 챗봇 API를 만들면, 동시에 여러 요청이 들어옵니다. 예를 들어, 고객 상담 챗봇에서 100명의 고객이 동시에 문의하면, 각각의 대화 컨텍스트를 완벽하게 분리해야 합니다.
기존에는 전역 변수 하나에 모든 대화를 저장하려다가 데이터가 뒤섞였다면, 이제는 세션 ID를 키로 하는 딕셔너리 구조로 각 사용자의 컨텍스트를 독립적으로 관리할 수 있습니다. 세션 관리의 핵심 특징은 첫째, 각 사용자에게 고유한 세션 ID를 부여한다는 것, 둘째, 세션 ID별로 완전히 독립된 컨텍스트를 유지한다는 것, 셋째, 일정 시간 비활성 상태인 세션은 자동으로 정리할 수 있다는 것입니다.
이러한 특징들이 실제 서비스 가능한 챗봇의 필수 요소입니다.
코드 예제
import uuid
from datetime import datetime, timedelta
class SessionManager:
def __init__(self, session_timeout=30):
# 세션 ID를 키로 하는 컨텍스트 저장소
self.sessions = {}
# 세션 마지막 활동 시간 추적
self.last_activity = {}
# 세션 타임아웃 (분 단위)
self.session_timeout = session_timeout
def create_session(self):
# 고유한 세션 ID 생성
session_id = str(uuid.uuid4())
self.sessions[session_id] = ConversationContext()
self.last_activity[session_id] = datetime.now()
return session_id
def get_session(self, session_id):
# 세션 활동 시간 갱신
self.last_activity[session_id] = datetime.now()
return self.sessions.get(session_id)
def cleanup_inactive_sessions(self):
# 오래된 세션 제거 (메모리 관리)
current_time = datetime.now()
timeout = timedelta(minutes=self.session_timeout)
inactive = [sid for sid, last_time in self.last_activity.items()
if current_time - last_time > timeout]
for sid in inactive:
del self.sessions[sid]
del self.last_activity[sid]
설명
이것이 하는 일: SessionManager 클래스는 여러 사용자의 대화 컨텍스트를 동시에 관리하는 중앙 관리자 역할을 합니다. 마치 호텔의 프런트 데스크처럼, 각 손님(사용자)에게 방 키(세션 ID)를 주고 방(컨텍스트)을 관리합니다.
첫 번째로, __init__ 메서드에서 세 개의 핵심 데이터 구조를 초기화합니다. sessions 딕셔너리는 세션 ID를 키로, ConversationContext 객체를 값으로 저장하며, last_activity는 각 세션의 마지막 활동 시간을 추적합니다.
이렇게 분리하는 이유는 세션 데이터와 메타데이터를 구분하여 관리 효율을 높이기 위함입니다. 그 다음으로, create_session 메서드가 실행되면서 UUID를 사용해 고유한 세션 ID를 생성합니다.
내부에서는 새로운 ConversationContext 객체를 생성하여 딕셔너리에 저장하고, 현재 시간을 기록합니다. UUID를 사용하는 이유는 수백만 개의 세션이 생성되어도 절대 중복되지 않는 ID를 보장하기 위함입니다.
get_session 메서드는 세션 ID로 컨텍스트를 조회하면서 동시에 활동 시간을 갱신합니다. 이는 매우 중요한데, 활동 시간이 갱신되지 않으면 사용 중인 세션도 정리될 수 있기 때문입니다.
마지막으로, cleanup_inactive_sessions 메서드가 오래 사용되지 않은 세션들을 찾아 삭제합니다. 최종적으로 메모리 누수를 방지하고 시스템 리소스를 효율적으로 관리할 수 있습니다.
여러분이 이 코드를 사용하면 수천 명의 사용자가 동시에 챗봇을 사용해도 각자의 대화가 완벽하게 분리됩니다. 실무에서의 이점은 첫째, 데이터 격리로 보안과 프라이버시가 보장되고, 둘째, 자동 세션 정리로 메모리 관리가 쉬우며, 셋째, 확장 가능한 구조로 대규모 서비스에도 적용 가능하다는 것입니다.
실전 팁
💡 세션 타임아웃은 서비스 특성에 맞게 조절하세요. 고객 상담은 30분, 일반 챗봇은 60분, 장시간 작업이 필요한 경우 2-3시간이 적당합니다. 너무 짧으면 사용자 불편, 너무 길면 메모리 낭비입니다.
💡 프로덕션 환경에서는 cleanup_inactive_sessions()를 백그라운드 스레드나 스케줄러로 주기적으로 실행하세요. APScheduler나 Celery Beat를 사용하면 5-10분마다 자동 정리가 가능합니다.
💡 Redis나 Memcached 같은 외부 캐시를 사용하면 서버가 재시작되어도 세션이 유지되고, 여러 서버 간 세션 공유도 가능합니다. sessions 딕셔너리 대신 Redis 클라이언트를 사용하도록 수정하면 됩니다.
💡 세션 ID는 반드시 클라이언트 쿠키나 헤더에 저장하여 요청마다 전달받으세요. 웹에서는 쿠키, 모바일 앱에서는 로컬 스토리지, API에서는 Authorization 헤더가 일반적입니다.
💡 디버깅 시에는 현재 활성 세션 수와 각 세션의 메시지 개수를 모니터링하세요. 갑자기 세션 수가 증가하면 정리 로직 문제나 메모리 누수를 의심해볼 수 있습니다.
3. 컨텍스트 윈도우 최적화 - 토큰 제한 극복하기
시작하며
여러분이 챗봇을 운영하다 보면 이런 에러를 만나게 됩니다. "maximum context length exceeded" 또는 "token limit exceeded".
사용자와 긴 대화를 나누다가 갑자기 챗봇이 응답을 거부하는 상황이죠. 이런 문제는 대부분의 LLM API가 입력 토큰 수에 제한이 있기 때문에 발생합니다.
GPT-3.5는 4,096 토큰, GPT-4는 8,192 토큰(또는 32K) 같은 제한이 있는데, 대화가 길어지면 이 한계에 도달합니다. 한글 기준으로 약 2-3글자가 1토큰이므로, 생각보다 빨리 제한에 걸립니다.
바로 이럴 때 필요한 것이 컨텍스트 윈도우 최적화입니다. 불필요한 내용은 제거하고 중요한 정보만 남겨서, 토큰 제한 내에서 최대한 많은 대화 맥락을 유지할 수 있습니다.
개요
간단히 말해서, 컨텍스트 윈도우 최적화는 제한된 토큰 안에서 가장 중요한 대화 내용만 선별하여 전달하는 기술입니다. 이 개념이 필요한 이유는 실무에서 토큰은 곧 비용이고 성능이기 때문입니다.
OpenAI API는 토큰당 과금되므로, 불필요한 토큰을 줄이면 비용이 절감됩니다. 또한 토큰이 적을수록 응답 속도도 빨라집니다.
예를 들어, 기술 상담 챗봇에서 10분 전에 나눈 인사말은 필요 없지만, 2분 전에 언급한 에러 메시지는 반드시 포함되어야 합니다. 기존에는 모든 대화를 무조건 포함시키다가 토큰 한계에 걸렸다면, 이제는 요약, 우선순위 부여, 슬라이딩 윈도우 같은 기법으로 효율적인 컨텍스트 관리가 가능합니다.
컨텍스트 윈도우 최적화의 핵심 특징은 첫째, 토큰 수를 실시간으로 계산하고 모니터링한다는 것, 둘째, 중요도에 따라 메시지를 선택적으로 유지한다는 것, 셋째, 오래된 대화는 요약하여 압축한다는 것입니다. 이러한 특징들이 안정적이고 비용 효율적인 챗봇 서비스를 가능하게 합니다.
코드 예제
import tiktoken
class TokenOptimizedContext(ConversationContext):
def __init__(self, max_tokens=3000, model="gpt-3.5-turbo"):
super().__init__()
# 최대 허용 토큰 수 (여유 공간 확보)
self.max_tokens = max_tokens
# 토큰 계산을 위한 인코더
self.encoder = tiktoken.encoding_for_model(model)
def count_tokens(self, text):
# 텍스트의 토큰 수 계산
return len(self.encoder.encode(text))
def get_optimized_context(self):
# 토큰 제한 내에서 최적화된 컨텍스트 반환
total_tokens = 0
optimized = []
# 최신 메시지부터 역순으로 추가
for msg in reversed(self.messages):
msg_tokens = self.count_tokens(msg["content"])
if total_tokens + msg_tokens <= self.max_tokens:
optimized.insert(0, msg) # 원래 순서 유지
total_tokens += msg_tokens
else:
break # 토큰 한계 도달
return optimized
설명
이것이 하는 일: TokenOptimizedContext 클래스는 ConversationContext를 상속받아 토큰 제한을 고려한 스마트한 컨텍스트 관리를 수행합니다. 마치 짐을 쌀 때 중요한 것부터 가방에 넣는 것처럼, 최신 대화부터 우선적으로 포함시킵니다.
첫 번째로, __init__ 메서드에서 tiktoken 라이브러리의 인코더를 초기화합니다. tiktoken은 OpenAI가 공식적으로 제공하는 토큰 계산 라이브러리로, 실제 API가 사용하는 것과 동일한 방식으로 토큰을 계산합니다.
이렇게 하는 이유는 정확한 토큰 수를 알아야 API 에러를 사전에 방지할 수 있기 때문입니다. 그 다음으로, count_tokens 메서드가 텍스트를 인코딩하여 토큰 수를 계산합니다.
내부에서는 문자열을 토큰 ID 리스트로 변환하고 길이를 반환하는데, 이는 단순 글자 수 계산보다 훨씬 정확합니다. 예를 들어 "안녕하세요"는 5글자지만 토큰으로는 2-3개일 수 있습니다.
get_optimized_context 메서드는 최신 메시지부터 역순으로 순회하면서 토큰을 누적 계산합니다. 토큰 합계가 max_tokens를 초과하지 않는 선에서 메시지를 선택하고, 원래 순서를 유지하기 위해 insert(0, msg)로 앞에 삽입합니다.
마지막으로, 이 메서드는 토큰 제한 내에서 최대한 많은 최신 대화를 포함한 최적화된 리스트를 반환하여, 최종적으로 API 호출 시 에러 없이 가장 관련성 높은 컨텍스트를 전달할 수 있습니다. 여러분이 이 코드를 사용하면 긴 대화에서도 절대 토큰 제한 에러가 발생하지 않습니다.
실무에서의 이점은 첫째, API 비용을 정확하게 예측하고 제어할 수 있고, 둘째, 응답 속도가 향상되며, 셋째, 최신 대화 우선 정책으로 대화 품질이 유지된다는 것입니다.
실전 팁
💡 max_tokens는 모델의 전체 한계보다 20-30% 여유있게 설정하세요. GPT-3.5(4K)면 3000, GPT-4(8K)면 6000 정도가 안전합니다. 나머지는 시스템 프롬프트와 응답 생성에 사용됩니다.
💡 tiktoken 설치는 pip install tiktoken으로 간단합니다. 오프라인 환경이라면 대략적인 계산식(한글 1글자 ≈ 0.4토큰, 영어 1단어 ≈ 1.3토큰)을 사용할 수도 있지만 정확도가 떨어집니다.
💡 시스템 메시지나 중요한 지침은 별도로 관리하여 항상 포함시키세요. 사용자 대화만 슬라이딩 윈도우로 관리하고, "당신은 친절한 상담원입니다" 같은 시스템 프롬프트는 고정으로 유지해야 합니다.
💡 토큰 사용량을 로그로 남기면 최적화 포인트를 찾을 수 있습니다. "평균 대화당 1500토큰 사용" 같은 데이터로 max_tokens를 더 정밀하게 조정할 수 있습니다.
💡 고급 기법으로는 오래된 대화를 GPT로 요약한 후 앞부분에 추가하는 방법이 있습니다. "이전 대화 요약: 사용자가 Python 설치 방법을 물어봄"처럼 압축하면 토큰을 크게 절약할 수 있습니다.
4. 중요 정보 영구 저장 - 핵심 컨텍스트 추출하기
시작하며
여러분이 챗봇으로 고객 상담을 하다 보면 이런 불편함을 느끼게 됩니다. 고객이 처음에 "내 이름은 김철수입니다"라고 소개했는데, 30분 후에 다시 물어보면 챗봇이 이름을 기억하지 못하는 상황이죠.
이런 문제는 슬라이딩 윈도우 방식의 컨텍스트 관리가 시간이 지나면 오래된 정보를 버리기 때문에 발생합니다. 사용자 이름, 선호도, 계정 정보처럼 대화 내내 기억해야 할 정보까지 함께 삭제되는 것입니다.
바로 이럴 때 필요한 것이 중요 정보 영구 저장 기능입니다. 대화에서 핵심 정보를 자동으로 추출하여 별도로 저장하고, 매 요청마다 이 정보를 컨텍스트에 포함시켜 마치 챗봇이 장기 기억을 가진 것처럼 동작하게 만들 수 있습니다.
개요
간단히 말해서, 중요 정보 영구 저장은 대화에서 핵심 팩트를 추출하여 장기 메모리에 보관하는 기술입니다. 이 개념이 필요한 이유는 실제 사용자 경험 측면에서 챗봇이 중요한 정보를 기억하지 못하면 신뢰도가 크게 떨어지기 때문입니다.
은행 챗봇이 계좌번호를 잊어버리거나, 쇼핑 챗봇이 배송지 주소를 기억하지 못하면 사용자는 답답함을 느낍니다. 예를 들어, 기술 지원 챗봇에서 "Python 3.9를 사용 중입니다"라는 정보는 세션 내내 유지되어야 정확한 답변이 가능합니다.
기존에는 모든 정보를 일시적 컨텍스트에만 저장했다면, 이제는 엔티티 추출과 키-값 저장소를 활용하여 중요 정보를 영구 보관할 수 있습니다. 영구 저장의 핵심 특징은 첫째, 대화에서 자동으로 핵심 엔티티(이름, 날짜, 장소 등)를 추출한다는 것, 둘째, 추출된 정보를 구조화하여 저장한다는 것, 셋째, 저장된 정보를 매 대화마다 컨텍스트에 자동 주입한다는 것입니다.
이러한 특징들이 챗봇을 단순 대화 도구에서 지능형 어시스턴트로 업그레이드시킵니다.
코드 예제
import re
class PersistentMemoryContext(TokenOptimizedContext):
def __init__(self, max_tokens=3000, model="gpt-3.5-turbo"):
super().__init__(max_tokens, model)
# 영구 저장할 핵심 정보 (키-값 저장소)
self.persistent_facts = {}
def extract_and_save_facts(self, user_message):
# 간단한 패턴 매칭으로 핵심 정보 추출
# 실무에서는 NER(Named Entity Recognition) 사용 권장
name_pattern = r"(?:제|내)\s*이름은\s*([가-힣]+)"
name_match = re.search(name_pattern, user_message)
if name_match:
self.persistent_facts["user_name"] = name_match.group(1)
# 이메일 추출
email_pattern = r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}"
email_match = re.search(email_pattern, user_message)
if email_match:
self.persistent_facts["email"] = email_match.group(0)
def get_context_with_facts(self):
# 영구 정보를 시스템 메시지로 추가
context = self.get_optimized_context()
if self.persistent_facts:
facts_text = "핵심 정보: " + ", ".join(
f"{k}={v}" for k, v in self.persistent_facts.items()
)
# 시스템 메시지로 앞에 추가
context.insert(0, {"role": "system", "content": facts_text})
return context
설명
이것이 하는 일: PersistentMemoryContext 클래스는 TokenOptimizedContext를 확장하여 장기 기억 기능을 추가합니다. 마치 사람이 중요한 정보를 메모장에 적어두듯, 핵심 팩트를 별도 딕셔너리에 저장합니다.
첫 번째로, __init__ 메서드에서 persistent_facts라는 빈 딕셔너리를 초기화합니다. 이 딕셔너리는 세션이 살아있는 동안 계속 유지되며, 필요하다면 데이터베이스나 Redis에 영구 저장할 수도 있습니다.
이렇게 분리하는 이유는 일시적 대화 내용과 장기 보관 정보를 명확히 구분하기 위함입니다. 그 다음으로, extract_and_save_facts 메서드가 실행되면서 정규식을 사용해 사용자 메시지에서 패턴을 찾습니다.
내부에서는 "내 이름은 홍길동"이나 "제 이름은 김철수" 같은 패턴을 매칭하여 이름을 추출하고, 이메일 주소도 표준 패턴으로 찾아냅니다. 실무에서는 spaCy나 KoNLPy 같은 NER 라이브러리를 사용하면 훨씬 정확하게 추출할 수 있습니다.
get_context_with_facts 메서드는 최적화된 컨텍스트를 가져온 후, persistent_facts에 저장된 정보를 "핵심 정보: user_name=홍길동, email=hong@example.com" 형태의 시스템 메시지로 만들어 컨텍스트 맨 앞에 삽입합니다. 마지막으로, 이 통합된 컨텍스트가 LLM에 전달되어 최종적으로 챗봇이 사용자의 개인 정보를 지속적으로 기억하면서 답변할 수 있게 됩니다.
여러분이 이 코드를 사용하면 사용자가 한 번만 정보를 제공하면 세션 내내 기억합니다. 실무에서의 이점은 첫째, 사용자 경험이 크게 개선되고, 둘째, 개인화된 응답이 가능하며, 셋째, 중요 정보를 구조화하여 저장하므로 나중에 분석이나 CRM 연동도 쉽다는 것입니다.
실전 팁
💡 정규식은 간단한 경우에만 사용하고, 실무에서는 spaCy의 NER이나 OpenAI의 함수 호출 기능으로 구조화된 정보를 추출하세요. "이름, 이메일, 전화번호를 JSON으로 추출해줘"라고 요청하면 더 정확합니다.
💡 persistent_facts를 데이터베이스에 저장하면 세션이 끝난 후에도 정보가 유지됩니다. SQLite나 PostgreSQL에 session_id를 키로 저장하거나, Redis에 TTL을 길게 설정하면 됩니다.
💡 민감한 개인정보(주민번호, 카드번호 등)는 절대 평문으로 저장하지 마세요. 암호화하거나 해싱하고, GDPR/개인정보보호법을 준수해야 합니다.
💡 사용자가 정보 수정을 요청할 수 있으므로 업데이트 로직도 필요합니다. "이메일을 변경하고 싶어요"라는 말을 감지하여 기존 값을 덮어쓰도록 구현하세요.
💡 추출된 정보의 신뢰도를 함께 저장하면 좋습니다. {"user_name": "홍길동", "confidence": 0.95} 형태로 저장하여, 낮은 신뢰도면 사용자에게 재확인을 요청할 수 있습니다.
5. 컨텍스트 압축과 요약 - 오래된 대화 효율적으로 관리하기
시작하며
여러분이 장시간 챗봇과 대화를 이어가다 보면 이런 딜레마에 빠집니다. 오래된 대화도 중요한 맥락이 있어서 완전히 버리기는 아깝지만, 그대로 두면 토큰이 너무 많이 소비되는 상황이죠.
이런 문제는 단순 슬라이딩 윈도우 방식이 오래된 대화를 무조건 삭제하기 때문에 발생합니다. 예를 들어 20분 전에 나눈 기술 토론 내용이 현재 질문과 관련이 있을 수 있는데, 이미 컨텍스트에서 제거되었다면 챗봇은 중복 설명을 하거나 일관성 없는 답변을 하게 됩니다.
바로 이럴 때 필요한 것이 컨텍스트 압축과 요약 기능입니다. 오래된 대화를 완전히 버리는 대신 핵심만 추출하여 요약본으로 변환하고, 이를 컨텍스트 앞부분에 배치하여 토큰은 절약하면서도 대화 맥락은 유지할 수 있습니다.
개요
간단히 말해서, 컨텍스트 압축은 긴 대화를 짧은 요약문으로 변환하여 토큰 효율을 극대화하는 기술입니다. 이 개념이 필요한 이유는 실무에서 긴 컨설팅 세션이나 기술 지원 대화처럼 1-2시간 이상 이어지는 대화가 많기 때문입니다.
전체 대화를 유지하면 수만 토큰이 들지만, 핵심만 요약하면 수백 토큰으로 줄어듭니다. 예를 들어, 기술 지원 챗봇에서 1시간 동안 디버깅한 내용을 "사용자는 Django 3.2에서 마이그레이션 오류 발생, settings.py의 DATABASE 설정 수정으로 해결"처럼 압축할 수 있습니다.
기존에는 오래된 대화를 통째로 버려서 맥락이 단절되었다면, 이제는 LLM을 활용한 지능형 요약으로 핵심 정보는 보존하면서도 토큰을 90% 이상 절약할 수 있습니다. 컨텍스트 압축의 핵심 특징은 첫째, 일정 개수 이상의 오래된 메시지를 자동으로 감지한다는 것, 둘째, LLM을 사용해 대화를 간결한 요약으로 변환한다는 것, 셋째, 요약본을 시스템 메시지로 컨텍스트 앞에 배치한다는 것입니다.
이러한 특징들이 장시간 대화에서도 일관성 있고 비용 효율적인 챗봇을 가능하게 합니다.
코드 예제
class CompressedContext(PersistentMemoryContext):
def __init__(self, max_tokens=3000, compress_threshold=15, model="gpt-3.5-turbo"):
super().__init__(max_tokens, model)
# 압축을 시작할 메시지 개수 임계값
self.compress_threshold = compress_threshold
# 요약된 이전 대화
self.compressed_summary = ""
def compress_old_messages(self, llm_client):
# 메시지가 임계값 초과 시 압축 실행
if len(self.messages) <= self.compress_threshold:
return
# 오래된 절반의 메시지를 요약 대상으로 선택
split_point = len(self.messages) // 2
old_messages = self.messages[:split_point]
# LLM에게 요약 요청
messages_text = "\n".join(
f"{m['role']}: {m['content']}" for m in old_messages
)
summary_prompt = f"다음 대화를 핵심 내용만 3-5문장으로 요약해주세요:\n\n{messages_text}"
# 실제 LLM 호출 (예시: OpenAI API)
# response = llm_client.chat.completions.create(...)
# summary = response.choices[0].message.content
summary = "[요약] " + "대화 핵심 요약본" # 예시
# 기존 요약에 추가
self.compressed_summary += "\n" + summary if self.compressed_summary else summary
# 요약된 메시지 제거
self.messages = self.messages[split_point:]
def get_full_context(self):
# 압축된 요약 + 영구 정보 + 최근 대화
context = self.get_context_with_facts()
if self.compressed_summary:
context.insert(0, {"role": "system", "content": self.compressed_summary})
return context
설명
이것이 하는 일: CompressedContext 클래스는 PersistentMemoryContext를 기반으로 자동 압축 기능을 추가합니다. 마치 문서를 ZIP 파일로 압축하듯, 긴 대화를 짧은 요약문으로 변환하여 저장 공간(토큰)을 절약합니다.
첫 번째로, __init__ 메서드에서 압축 임계값(compress_threshold)과 요약 저장 변수(compressed_summary)를 초기화합니다. 임계값은 보통 10-20개가 적당한데, 이렇게 설정하는 이유는 너무 자주 압축하면 LLM API 호출 비용이 늘어나고, 너무 늦게 압축하면 토큰이 이미 초과될 수 있기 때문입니다.
그 다음으로, compress_old_messages 메서드가 실행되면서 메시지 개수를 체크합니다. 내부에서는 임계값을 초과하면 전체 메시지의 절반(오래된 부분)을 선택하고, 이를 "user: 안녕하세요\nassistant: 반갑습니다" 같은 텍스트 형식으로 변환합니다.
그리고 LLM에게 "핵심만 3-5문장으로 요약해달라"는 프롬프트와 함께 전달하여 압축된 요약본을 받아옵니다. 요약이 완료되면 기존 compressed_summary에 새 요약을 추가하고, 원본 메시지는 리스트에서 제거합니다.
이렇게 하면 메시지 개수와 토큰 수가 절반으로 줄어들지만, 핵심 정보는 요약본에 남아있습니다. 마지막으로, get_full_context 메서드가 압축된 요약, 영구 저장된 팩트, 최근 대화를 순서대로 조합하여 최종적으로 완전한 컨텍스트를 만들어 LLM에 전달합니다.
여러분이 이 코드를 사용하면 몇 시간 동안 이어지는 대화도 일관성 있게 처리할 수 있습니다. 실무에서의 이점은 첫째, 토큰 비용이 크게 절감되고(50-90%), 둘째, 긴 대화에서도 맥락이 유지되며, 셋째, 자동화되어 있어 개발자가 신경 쓸 필요가 없다는 것입니다.
실전 팁
💡 압축 타이밍은 비동기로 처리하세요. 사용자가 메시지를 보낸 직후가 아니라, 백그라운드 작업이나 다음 요청 전에 압축하면 응답 지연을 방지할 수 있습니다.
💡 요약 프롬프트를 도메인에 맞게 커스터마이즈하세요. 기술 지원이면 "문제점과 해결책을 중심으로", 쇼핑이면 "제품과 고객 요구사항을 중심으로" 요약하도록 지시하면 더 유용한 요약이 나옵니다.
💡 압축 비용을 고려하세요. GPT-4로 요약하면 품질은 좋지만 비용이 높으므로, GPT-3.5나 더 저렴한 모델로 요약하는 것도 좋은 전략입니다. 요약은 창의성이 덜 필요하므로 저렴한 모델로도 충분합니다.
💡 압축 전후 토큰 수를 로그로 남기면 효과를 측정할 수 있습니다. "압축 전: 5000 토큰 → 압축 후: 800 토큰 (84% 절감)" 같은 데이터로 임계값을 최적화할 수 있습니다.
💡 중요한 대화는 압축하지 않도록 플래그를 추가하세요. 특정 메시지에 {"important": true} 같은 메타데이터를 붙여서, 압축 대상에서 제외하거나 별도로 보존할 수 있습니다.
6. 멀티턴 대화 플로우 관리 - 상태 기반 대화 처리하기
시작하며
여러분이 예약 시스템이나 설문조사 챗봇을 만들 때 이런 어려움을 겪게 됩니다. "예약하고 싶어요" → "날짜를 알려주세요" → "내일이요" → "시간은요?" 같은 단계적 대화를 어떻게 관리할지 막막한 상황이죠.
이런 문제는 단순히 대화 내용만 저장하는 것으로는 부족하기 때문에 발생합니다. 현재 대화가 어떤 단계에 있는지, 사용자에게서 무엇을 받아야 하는지, 모든 정보가 모였는지 등을 추적해야 하는데, 이는 컨텍스트만으로는 처리하기 어렵습니다.
바로 이럴 때 필요한 것이 상태 기반 대화 플로우 관리입니다. 대화를 여러 상태(state)로 나누고, 각 상태에서 필요한 정보를 수집하며, 조건이 충족되면 다음 상태로 전환하는 구조화된 접근 방식입니다.
개요
간단히 말해서, 멀티턴 대화 플로우 관리는 복잡한 대화를 단계별 상태로 나누어 체계적으로 처리하는 기술입니다. 이 개념이 필요한 이유는 실무에서 단순 질의응답을 넘어 복잡한 업무 프로세스를 챗봇으로 처리하는 경우가 많기 때문입니다.
호텔 예약, 배송 조회, 계좌 개설 같은 작업은 여러 정보를 순차적으로 수집해야 합니다. 예를 들어, 음식 주문 챗봇은 "메뉴 선택 → 수량 입력 → 주소 확인 → 결제 수단 선택" 순서로 진행되며, 각 단계를 명확히 관리해야 합니다.
기존에는 if-else 조건문으로 복잡하게 처리하다가 코드가 스파게티처럼 얽혔다면, 이제는 상태 머신 패턴으로 각 단계를 명확히 정의하고 전환 규칙을 설정할 수 있습니다. 상태 기반 플로우의 핵심 특징은 첫째, 현재 대화 상태를 명시적으로 추적한다는 것, 둘째, 각 상태에서 필요한 정보와 다음 상태 조건을 정의한다는 것, 셋째, 수집된 정보를 구조화하여 저장한다는 것입니다.
이러한 특징들이 복잡한 비즈니스 로직을 챗봇으로 구현할 수 있게 해줍니다.
코드 예제
from enum import Enum
class ReservationState(Enum):
INITIAL = "initial"
COLLECTING_DATE = "collecting_date"
COLLECTING_TIME = "collecting_time"
COLLECTING_NAME = "collecting_name"
CONFIRMING = "confirming"
COMPLETED = "completed"
class StatefulContext(CompressedContext):
def __init__(self, max_tokens=3000, model="gpt-3.5-turbo"):
super().__init__(max_tokens, model=model)
# 현재 대화 상태
self.current_state = ReservationState.INITIAL
# 수집 중인 정보
self.collected_data = {}
def transition_to(self, next_state):
# 상태 전환 (로그 출력 권장)
print(f"상태 전환: {self.current_state} → {next_state}")
self.current_state = next_state
def update_collected_data(self, key, value):
# 수집된 정보 저장 및 다음 상태 결정
self.collected_data[key] = value
# 상태별 전환 로직
if self.current_state == ReservationState.COLLECTING_DATE:
self.transition_to(ReservationState.COLLECTING_TIME)
elif self.current_state == ReservationState.COLLECTING_TIME:
self.transition_to(ReservationState.COLLECTING_NAME)
elif self.current_state == ReservationState.COLLECTING_NAME:
self.transition_to(ReservationState.CONFIRMING)
def get_next_prompt(self):
# 현재 상태에 따른 안내 메시지 반환
prompts = {
ReservationState.COLLECTING_DATE: "예약 날짜를 알려주세요.",
ReservationState.COLLECTING_TIME: "예약 시간을 알려주세요.",
ReservationState.COLLECTING_NAME: "성함을 알려주세요.",
ReservationState.CONFIRMING: f"확인: {self.collected_data}. 맞나요?"
}
return prompts.get(self.current_state, "무엇을 도와드릴까요?")
설명
이것이 하는 일: StatefulContext 클래스는 CompressedContext를 확장하여 대화 상태 머신을 구현합니다. 마치 게임의 스테이지처럼, 각 대화 단계를 명확한 상태로 정의하고 순차적으로 진행합니다.
첫 번째로, ReservationState Enum에서 가능한 모든 대화 상태를 정의합니다. Enum을 사용하는 이유는 오타를 방지하고, IDE의 자동완성을 활용하며, 가능한 상태를 한눈에 파악할 수 있기 때문입니다.
이렇게 하면 "collecting_date"를 "collecting_dat"로 잘못 쓰는 실수를 컴파일 단계에서 잡을 수 있습니다. 그 다음으로, __init__ 메서드가 현재 상태(current_state)와 수집된 데이터(collected_data)를 초기화합니다.
내부에서는 INITIAL 상태로 시작하며, 빈 딕셔너리에 사용자가 제공하는 정보를 점진적으로 채워나갑니다. update_collected_data 메서드는 정보를 저장하면서 동시에 상태 전환 로직을 실행합니다.
예를 들어 날짜를 입력받으면 자동으로 시간 입력 상태로 전환되고, 시간을 입력받으면 이름 입력 상태로 넘어갑니다. 이는 명확한 순서가 있는 대화 플로우를 강제합니다.
마지막으로, get_next_prompt 메서드가 현재 상태에 맞는 안내 메시지를 반환하여, 최종적으로 사용자에게 다음에 무엇을 입력해야 하는지 명확하게 알려줍니다. 여러분이 이 코드를 사용하면 복잡한 다단계 대화도 체계적으로 관리할 수 있습니다.
실무에서의 이점은 첫째, 코드 가독성과 유지보수성이 크게 향상되고, 둘째, 각 단계별 유효성 검사를 쉽게 추가할 수 있으며, 셋째, 사용자가 중간에 이탈해도 현재 상태를 저장했다가 나중에 이어서 진행할 수 있다는 것입니다.
실전 팁
💡 상태 전환 조건을 별도 메서드로 분리하세요. can_transition(from_state, to_state) 같은 메서드로 전환 가능 여부를 검증하면, 잘못된 순서로 진행되는 것을 방지할 수 있습니다.
💡 각 상태에서 수집할 데이터의 유효성을 검사하세요. 날짜 형식이 맞는지, 시간이 영업시간 내인지 등을 체크하고, 잘못된 입력이면 같은 상태를 유지하며 재입력을 요청하세요.
💡 사용자가 "처음부터 다시" 같은 명령으로 상태를 초기화할 수 있게 하세요. reset() 메서드를 만들어 current_state와 collected_data를 초기값으로 되돌리면 됩니다.
💡 상태와 데이터를 데이터베이스나 Redis에 저장하면 세션이 끊겨도 이어서 진행할 수 있습니다. 사용자가 앱을 껐다가 다시 켜도 "아까 예약하시던 중이었죠? 이어서 진행할까요?"라고 물을 수 있습니다.
💡 복잡한 플로우는 상태 다이어그램을 그려서 설계하세요. INITIAL → COLLECTING_DATE → ... → COMPLETED 같은 흐름을 시각화하면 빠진 경로나 순환 참조를 발견하기 쉽습니다.
7. 컨텍스트 우선순위 관리 - 중요한 메시지 우선 보존하기
시작하며
여러분이 토큰 제한 때문에 오래된 메시지를 삭제해야 할 때 이런 고민을 하게 됩니다. 모든 메시지가 똑같이 중요하지 않은데, 단순히 시간 순서로만 삭제하는 게 맞을까?
이런 문제는 기존 슬라이딩 윈도우 방식이 메시지의 중요도를 고려하지 않기 때문에 발생합니다. 예를 들어 5분 전의 잡담은 별로 중요하지 않지만, 20분 전에 사용자가 제공한 에러 로그는 현재 문제 해결에 핵심적일 수 있습니다.
시간만 보고 삭제하면 중요한 정보를 잃게 됩니다. 바로 이럴 때 필요한 것이 컨텍스트 우선순위 관리입니다.
각 메시지에 중요도 점수를 부여하고, 토큰 제한에 도달했을 때 중요도가 낮은 메시지부터 제거하여 핵심 정보는 최대한 오래 보존할 수 있습니다.
개요
간단히 말해서, 컨텍스트 우선순위 관리는 메시지의 중요도를 평가하여 제한된 토큰 안에서 가장 가치 있는 정보를 선별하는 기술입니다. 이 개념이 필요한 이유는 실무에서 모든 대화 내용이 동일한 가치를 갖지 않기 때문입니다.
인사말, 확인 응답, 잡담은 맥락 이해에 거의 기여하지 않지만, 에러 메시지, 코드 예제, 구체적인 요구사항은 매우 중요합니다. 예를 들어, 기술 지원 챗봇에서 "안녕하세요", "네 알겠습니다" 같은 메시지보다 "ImportError: No module named 'pandas'" 같은 에러 메시지가 훨씬 중요합니다.
기존에는 FIFO(First In First Out) 방식으로 오래된 것부터 무조건 삭제했다면, 이제는 중요도 점수를 기준으로 가치가 낮은 메시지부터 선택적으로 제거할 수 있습니다. 우선순위 관리의 핵심 특징은 첫째, 메시지 내용을 분석하여 자동으로 중요도를 계산한다는 것, 둘째, 키워드, 길이, 역할(user/assistant) 등 여러 요소를 고려한다는 것, 셋째, 토큰 제한 시 중요도가 낮은 것부터 제거한다는 것입니다.
이러한 특징들이 제한된 컨텍스트에서 최대한의 정보를 보존하게 해줍니다.
코드 예제
import re
class PriorityContext(StatefulContext):
def __init__(self, max_tokens=3000, model="gpt-3.5-turbo"):
super().__init__(max_tokens, model=model)
# 각 메시지의 중요도 점수 저장
self.message_priorities = []
def calculate_priority(self, message):
# 메시지 중요도 계산 (0-100 점수)
score = 50 # 기본 점수
content = message["content"].lower()
# 중요 키워드 포함 시 가산점
important_keywords = ["error", "에러", "문제", "exception", "코드", "예제", "help"]
for keyword in important_keywords:
if keyword in content:
score += 10
# 길이 기반 점수 (긴 메시지는 보통 중요)
if len(content) > 100:
score += 15
# 사용자 메시지는 더 중요
if message["role"] == "user":
score += 10
# 인사말이나 짧은 확인은 감점
greetings = ["안녕", "hello", "hi", "감사", "thank"]
if any(g in content for g in greetings) and len(content) < 20:
score -= 20
return min(100, max(0, score)) # 0-100 범위로 제한
def add_message(self, role, content):
# 메시지 추가 시 우선순위도 함께 저장
message = {"role": role, "content": content}
priority = self.calculate_priority(message)
self.messages.append(message)
self.message_priorities.append(priority)
def get_priority_optimized_context(self):
# 우선순위 기반으로 메시지 선택
if not self.messages:
return []
# 메시지와 우선순위를 쌍으로 묶기
paired = list(zip(self.messages, self.message_priorities))
# 우선순위 내림차순 정렬
paired.sort(key=lambda x: x[1], reverse=True)
# 토큰 제한 내에서 중요한 것부터 선택
selected = []
total_tokens = 0
for msg, priority in paired:
tokens = self.count_tokens(msg["content"])
if total_tokens + tokens <= self.max_tokens:
selected.append(msg)
total_tokens += tokens
# 시간 순서로 재정렬 (대화 흐름 유지)
original_order = {id(m): i for i, m in enumerate(self.messages)}
selected.sort(key=lambda m: original_order.get(id(m), 0))
return selected
설명
이것이 하는 일: PriorityContext 클래스는 StatefulContext를 확장하여 지능형 메시지 필터링 기능을 추가합니다. 마치 뉴스 편집자가 중요한 기사를 1면에 배치하듯, 중요한 메시지를 우선적으로 보존합니다.
첫 번째로, calculate_priority 메서드에서 다양한 휴리스틱을 사용해 메시지 중요도를 계산합니다. 기본 50점에서 시작하여 "error", "코드" 같은 중요 키워드가 있으면 가산점을 주고, "안녕" 같은 인사말은 감점합니다.
이렇게 하는 이유는 완벽한 AI 분류기를 사용하면 비용이 높으므로, 간단한 규칙 기반으로도 상당히 정확한 우선순위를 매길 수 있기 때문입니다. 그 다음으로, add_message 메서드가 실행되면서 메시지를 저장할 때 동시에 우선순위 점수도 계산하여 별도 리스트에 저장합니다.
내부에서는 messages와 message_priorities 두 리스트의 인덱스를 일치시켜 관리하며, 나중에 쌍으로 묶어서 처리할 수 있게 합니다. get_priority_optimized_context 메서드는 메시지와 우선순위를 zip으로 묶은 후 우선순위 내림차순으로 정렬합니다.
그리고 중요한 메시지부터 토큰 제한까지 선택하는데, 이때 중요한 것은 선택 후 다시 원래 시간 순서로 재정렬한다는 점입니다. 대화의 자연스러운 흐름을 유지하기 위함입니다.
마지막으로, 이 메서드는 중요도와 시간 순서를 모두 고려한 최적화된 컨텍스트를 반환하여, 최종적으로 LLM이 가장 관련성 높은 정보를 받아 정확한 답변을 생성할 수 있게 합니다. 여러분이 이 코드를 사용하면 토큰 제한이 있어도 중요한 정보는 절대 잃어버리지 않습니다.
실무에서의 이점은 첫째, 핵심 정보 보존율이 크게 향상되고, 둘째, 답변 품질이 일관되게 유지되며, 셋째, 규칙을 도메인에 맞게 커스터마이즈할 수 있다는 것입니다.
실전 팁
💡 도메인별로 중요 키워드를 다르게 설정하세요. 의료 챗봇이면 "증상", "약", "처방", 쇼핑 챗봇이면 "주문", "배송", "환불"을 중요하게 취급해야 합니다.
💡 LLM을 사용해 더 정교한 중요도 계산이 가능합니다. "이 메시지가 대화 이해에 얼마나 중요한지 0-100으로 평가해줘"라고 물어보면 규칙보다 정확하지만, API 비용이 추가됩니다.
💡 사용자 피드백을 반영하세요. 사용자가 "아까 말한 거 기억 안 나?" 같은 불만을 표현하면, 해당 메시지의 우선순위를 높여야 한다는 신호입니다.
💡 최근 메시지에 시간 기반 가산점을 주는 것도 좋은 전략입니다. score += max(0, 20 - (현재시간 - 메시지시간).minutes) 식으로 최근 메시지일수록 추가 점수를 줍니다.
💡 우선순위 점수를 시각화하여 디버깅하세요. "메시지: '안녕하세요' (우선순위: 30)", "메시지: 'ImportError 발생' (우선순위: 85)" 같이 출력하면 점수 매기기 로직을 개선하기 쉽습니다.
8. 컨텍스트 공유와 동기화 - 멀티 채널 대화 통합하기
시작하며
여러분이 멀티 채널 서비스를 운영할 때 이런 문제를 겪게 됩니다. 사용자가 웹에서 챗봇과 대화하다가 모바일 앱으로 이동했는데, 챗봇이 처음부터 다시 물어보는 상황이죠.
이런 문제는 각 채널(웹, 모바일, 메신저)이 독립된 세션과 컨텍스트를 사용하기 때문에 발생합니다. 사용자 입장에서는 같은 서비스인데, 채널을 바꿀 때마다 대화를 처음부터 시작해야 하면 매우 불편합니다.
특히 요즘처럼 사용자들이 여러 디바이스를 오가며 사용하는 환경에서는 심각한 UX 문제입니다. 바로 이럴 때 필요한 것이 컨텍스트 공유와 동기화 기능입니다.
중앙 저장소에 컨텍스트를 저장하고, 모든 채널이 동일한 컨텍스트를 참조하도록 하여 사용자가 어디에서 접속하든 자연스럽게 이어지는 대화를 제공할 수 있습니다.
개요
간단히 말해서, 컨텍스트 공유는 여러 채널 간에 대화 내용을 실시간으로 동기화하여 일관된 사용자 경험을 제공하는 기술입니다. 이 개념이 필요한 이유는 실무에서 옴니채널 전략이 표준이 되었기 때문입니다.
사용자는 출근길에는 모바일, 사무실에서는 웹, 집에서는 태블릿을 사용합니다. 각 채널마다 대화를 처음부터 시작하면 사용자는 같은 정보를 여러 번 입력해야 합니다.
예를 들어, 고객이 모바일로 주문 문의를 시작했다가 PC로 이어서 결제하려고 할 때, 챗봇이 이전 대화를 기억하지 못하면 주문 과정이 중단됩니다. 기존에는 각 채널이 로컬 메모리나 세션 저장소를 사용해서 컨텍스트가 격리되었다면, 이제는 Redis나 데이터베이스 같은 중앙 저장소로 사용자 ID 기반 컨텍스트 공유가 가능합니다.
컨텍스트 공유의 핵심 특징은 첫째, 사용자 ID를 기준으로 컨텍스트를 식별한다는 것, 둘째, 중앙 저장소에서 실시간으로 읽기/쓰기한다는 것, 셋째, 동시 접속 시 충돌을 방지하는 동기화 메커니즘이 있다는 것입니다. 이러한 특징들이 진정한 옴니채널 챗봇 경험을 가능하게 합니다.
코드 예제
import json
import redis
class SharedContext(PriorityContext):
def __init__(self, user_id, redis_client, max_tokens=3000, model="gpt-3.5-turbo"):
super().__init__(max_tokens, model=model)
# 사용자 고유 ID
self.user_id = user_id
# Redis 클라이언트
self.redis = redis_client
# Redis 키 (사용자별 컨텍스트 저장)
self.redis_key = f"context:{user_id}"
# 저장소에서 컨텍스트 로드
self.load_from_storage()
def load_from_storage(self):
# Redis에서 컨텍스트 복원
data = self.redis.get(self.redis_key)
if data:
loaded = json.loads(data)
self.messages = loaded.get("messages", [])
self.message_priorities = loaded.get("priorities", [])
self.persistent_facts = loaded.get("facts", {})
self.current_state = loaded.get("state", "initial")
def save_to_storage(self):
# 컨텍스트를 Redis에 저장
data = {
"messages": self.messages,
"priorities": self.message_priorities,
"facts": self.persistent_facts,
"state": self.current_state
}
# JSON으로 직렬화 후 저장 (TTL 24시간)
self.redis.setex(self.redis_key, 86400, json.dumps(data, ensure_ascii=False))
def add_message(self, role, content):
# 메시지 추가 후 즉시 저장
super().add_message(role, content)
self.save_to_storage()
def sync(self):
# 다른 채널의 변경사항 가져오기
self.load_from_storage()
설명
이것이 하는 일: SharedContext 클래스는 PriorityContext를 확장하여 Redis 기반 중앙 저장소 연동 기능을 추가합니다. 마치 클라우드 동기화처럼, 모든 디바이스가 같은 대화 내용을 공유합니다.
첫 번째로, __init__ 메서드에서 user_id와 redis_client를 받아 초기화하고, "context:사용자ID" 형태의 Redis 키를 생성합니다. 이렇게 하는 이유는 수백만 명의 사용자 컨텍스트를 구분하여 저장하고, Redis의 빠른 키-값 조회를 활용하기 위함입니다.
초기화 마지막에 load_from_storage()를 호출하여 기존 대화가 있으면 복원합니다. 그 다음으로, load_from_storage 메서드가 실행되면서 Redis에서 JSON 문자열을 가져와 파싱합니다.
내부에서는 messages, priorities, facts, state 등 모든 컨텍스트 데이터를 딕셔너리로 복원하여 클래스 속성에 할당합니다. 데이터가 없으면 빈 상태로 시작하므로 신규 사용자도 문제없이 처리됩니다.
save_to_storage 메서드는 현재 컨텍스트 상태를 딕셔너리로 만들고 JSON으로 직렬화한 후 Redis에 저장합니다. setex를 사용해 TTL 24시간을 설정하여, 오래 사용하지 않은 컨텍스트는 자동으로 만료됩니다.
ensure_ascii=False 옵션으로 한글도 정상적으로 저장됩니다. add_message를 오버라이드하여 메시지 추가 후 자동으로 save_to_storage()를 호출합니다.
이렇게 하면 개발자가 수동으로 저장을 신경 쓸 필요 없이, 모든 변경사항이 즉시 Redis에 반영됩니다. 마지막으로, sync 메서드로 다른 채널의 변경사항을 수동으로 가져올 수 있어, 최종적으로 멀티 채널 환경에서 실시간에 가까운 동기화가 가능합니다.
여러분이 이 코드를 사용하면 웹에서 시작한 대화를 모바일 앱에서 바로 이어갈 수 있습니다. 실무에서의 이점은 첫째, 사용자 경험이 극적으로 개선되고, 둘째, 서버 재시작 시에도 대화가 보존되며, 셋째, 여러 서버 인스턴스 간 세션 공유가 자동으로 해결된다는 것입니다.
실전 팁
💡 Redis 연결 풀을 사용하세요. redis.ConnectionPool()로 풀을 만들고 재사용하면, 매번 연결을 맺고 끊는 오버헤드를 제거할 수 있습니다. 성능이 10배 이상 향상됩니다.
💡 저장 빈도를 조절하여 Redis 부하를 줄이세요. 매 메시지마다 저장하는 대신, 3-5개 메시지마다 또는 일정 시간 간격으로 배치 저장하면 Redis 트래픽이 크게 감소합니다.
💡 동시 접속 시 충돌을 방지하려면 Redis의 WATCH/MULTI/EXEC 트랜잭션을 사용하세요. 또는 버전 번호를 추가하여 낙관적 잠금(Optimistic Locking)을 구현할 수 있습니다.
💡 JSON 대신 MessagePack을 사용하면 저장 크기가 30-50% 줄어듭니다. import msgpack으로 설치하고, msgpack.packb()와 msgpack.unpackb()로 직렬화하면 됩니다.
💡 중요한 대화는 Redis(캐시)와 PostgreSQL(영구 저장소) 모두에 저장하세요. Redis는 빠른 접근용, DB는 장기 보관 및 분석용으로 사용하면 안정성과 성능을 모두 확보할 수 있습니다.