Macbook
4b86b2a660
- server-dev start/stop/deploy 및 Gitea push 자동 배포 - local-dev 로컬 개발 환경 Co-authored-by: Cursor <cursoragent@cursor.com>
284 lines
10 KiB
Python
284 lines
10 KiB
Python
"""
|
|
chat_history.py
|
|
────────────────────────────────────────────
|
|
MongoDB 대화 이력 관리 모듈
|
|
"""
|
|
import os
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import List, Dict, Optional
|
|
from pymongo import MongoClient, DESCENDING
|
|
from pymongo.errors import PyMongoError
|
|
|
|
from handlers.suggestion_handler import strip_suggestion_block
|
|
|
|
# 환경 변수
|
|
MONGO_HOST = os.getenv("MONGO_HOST", "localhost")
|
|
MONGO_PORT = int(os.getenv("MONGO_PORT", "27017"))
|
|
MONGO_USER = os.getenv("MONGO_USER", "exlink")
|
|
MONGO_PASSWORD = os.getenv("MONGO_PASSWORD", "!wkcproqkf1")
|
|
MONGO_DATABASE = os.getenv("MONGO_DATABASE", "chat_history")
|
|
MONGO_COLLECTION = os.getenv("MONGO_COLLECTION", "rag_conversations")
|
|
MONGO_TTL_DAYS = int(os.getenv("MONGO_TTL_DAYS", "30")) # 데이터 보관 기간 (일 단위, 기본값: 30일)
|
|
|
|
|
|
class ChatHistoryManager:
|
|
"""MongoDB 기반 대화 이력 관리"""
|
|
|
|
def __init__(self):
|
|
"""MongoDB 연결 초기화"""
|
|
try:
|
|
# MongoDB 연결 문자열 (데이터베이스별 인증)
|
|
# 연결: mongodb://localhost:27017/
|
|
# 인증: chat_history 데이터베이스에서 exlink/!wkcproqkf1
|
|
if MONGO_USER and MONGO_PASSWORD:
|
|
connection_string = f"mongodb://{MONGO_USER}:{MONGO_PASSWORD}@{MONGO_HOST}:{MONGO_PORT}/{MONGO_DATABASE}?authSource={MONGO_DATABASE}"
|
|
else:
|
|
connection_string = f"mongodb://{MONGO_HOST}:{MONGO_PORT}/"
|
|
|
|
# 연결 풀링 설정 (고트래픽 대응)
|
|
self.client = MongoClient(
|
|
connection_string,
|
|
serverSelectionTimeoutMS=5000,
|
|
maxPoolSize=50, # 최대 연결 수 (기본값: 100)
|
|
minPoolSize=10, # 최소 연결 수 (기본값: 0)
|
|
maxIdleTimeMS=45000, # 유휴 연결 유지 시간
|
|
)
|
|
self.db = self.client[MONGO_DATABASE]
|
|
self.collection = self.db[MONGO_COLLECTION]
|
|
|
|
# 인덱스 생성 (성능 최적화)
|
|
# background=True: 인덱스 생성 시 DB 블로킹 방지
|
|
self.collection.create_index(
|
|
[("bot_id", 1), ("timestamp", -1)], # -1 = DESCENDING
|
|
background=True,
|
|
name="idx_bot_timestamp"
|
|
)
|
|
|
|
# TTL 인덱스 생성 (자동 삭제)
|
|
ttl_seconds = MONGO_TTL_DAYS * 24 * 60 * 60 # 일 단위 → 초 단위 변환
|
|
self.collection.create_index(
|
|
[("timestamp", 1)],
|
|
expireAfterSeconds=ttl_seconds,
|
|
background=True,
|
|
name="idx_ttl"
|
|
)
|
|
|
|
print(f"[ChatHistory] MongoDB 연결 성공: {MONGO_HOST}:{MONGO_PORT}/{MONGO_DATABASE}")
|
|
print(f"[ChatHistory] TTL 설정: {MONGO_TTL_DAYS}일 ({ttl_seconds}초) 후 자동 삭제")
|
|
except PyMongoError as e:
|
|
print(f"[ChatHistory] MongoDB 연결 실패: {e}")
|
|
raise
|
|
|
|
def save_conversation(
|
|
self,
|
|
bot_id: Optional[str],
|
|
user_query: str,
|
|
ai_response: str,
|
|
matched_questions: List[str],
|
|
scores: List[float],
|
|
metadata: Optional[Dict] = None
|
|
) -> str:
|
|
"""
|
|
대화 기록 저장
|
|
|
|
Args:
|
|
bot_id: 봇 ID (없으면 None)
|
|
user_query: 사용자 질문
|
|
ai_response: AI 답변
|
|
matched_questions: 매칭된 질문 목록
|
|
scores: 매칭 점수
|
|
metadata: 추가 메타데이터
|
|
|
|
Returns:
|
|
저장된 문서의 ObjectId (문자열)
|
|
"""
|
|
try:
|
|
doc = {
|
|
"bot_id": bot_id, # None 허용
|
|
"user_query": user_query,
|
|
"ai_response": ai_response,
|
|
"matched_questions": matched_questions,
|
|
"scores": scores,
|
|
"metadata": metadata or {},
|
|
"timestamp": datetime.now(timezone.utc)
|
|
}
|
|
|
|
result = self.collection.insert_one(doc)
|
|
print(f"[ChatHistory] 대화 저장 완료: bot_id={bot_id}, id={result.inserted_id}")
|
|
return str(result.inserted_id)
|
|
|
|
except PyMongoError as e:
|
|
print(f"[ChatHistory] 저장 실패: {e}")
|
|
raise
|
|
|
|
def get_recent_history(
|
|
self,
|
|
bot_id: Optional[str],
|
|
hours: int = 24,
|
|
limit: int = 10
|
|
) -> List[Dict]:
|
|
"""
|
|
최근 대화 이력 조회 (24시간 이내)
|
|
|
|
Args:
|
|
bot_id: 봇 ID (None이면 전체 조회)
|
|
hours: 조회 시간 범위 (기본 24시간)
|
|
limit: 최대 조회 개수
|
|
|
|
Returns:
|
|
대화 이력 리스트 (최신순)
|
|
"""
|
|
try:
|
|
# 시간 필터 (UTC 기준)
|
|
cutoff_time = datetime.now(timezone.utc) - timedelta(hours=hours)
|
|
|
|
# 쿼리 구성 (인덱스 순서에 맞춤: bot_id → timestamp)
|
|
query = {}
|
|
if bot_id is not None:
|
|
query["bot_id"] = bot_id
|
|
query["timestamp"] = {"$gte": cutoff_time}
|
|
|
|
# 조회 (복합 인덱스 활용: bot_id + timestamp)
|
|
cursor = self.collection.find(query).sort("timestamp", DESCENDING).limit(limit)
|
|
|
|
# 결과 변환
|
|
history = []
|
|
for doc in cursor:
|
|
history.append({
|
|
"user_query": doc.get("user_query"),
|
|
"ai_response": doc.get("ai_response"),
|
|
"timestamp": doc.get("timestamp").isoformat() if doc.get("timestamp") else None
|
|
})
|
|
|
|
print(f"[ChatHistory] 이력 조회: bot_id={bot_id}, {len(history)}개")
|
|
return list(reversed(history)) # 시간순 정렬 (오래된 것 → 최신)
|
|
|
|
except PyMongoError as e:
|
|
print(f"[ChatHistory] 조회 실패: {e}")
|
|
return []
|
|
|
|
def get_context_for_llm(
|
|
self,
|
|
bot_id: Optional[str],
|
|
hours: int = 24,
|
|
max_conversations: int = 10
|
|
) -> str:
|
|
"""
|
|
LLM에 전달할 대화 컨텍스트 생성
|
|
|
|
Args:
|
|
bot_id: 봇 ID
|
|
hours: 조회 시간 범위
|
|
max_conversations: 최대 대화 개수
|
|
|
|
Returns:
|
|
포맷된 대화 이력 문자열
|
|
"""
|
|
history = self.get_recent_history(bot_id, hours, max_conversations)
|
|
|
|
if not history:
|
|
return ""
|
|
|
|
# 포맷팅 (추천 블록 제외 — LLM용 본문 전문, get_messages_for_llm과 동일)
|
|
context_parts = ["【이전 대화 이력】"]
|
|
for i, conv in enumerate(history, 1):
|
|
body = strip_suggestion_block(conv.get("ai_response"))
|
|
context_parts.append(
|
|
f"[대화 {i}]\n"
|
|
f"고객: {conv['user_query']}\n"
|
|
f"상담원: {body}"
|
|
)
|
|
|
|
return "\n\n".join(context_parts)
|
|
|
|
def get_messages_for_llm(
|
|
self,
|
|
bot_id: Optional[str],
|
|
hours: int = 24,
|
|
max_conversations: int = 10
|
|
) -> List[Dict[str, str]]:
|
|
"""
|
|
LLM messages format용 대화 이력 (표준 chat completion)
|
|
|
|
Args:
|
|
bot_id: 봇 ID
|
|
hours: 조회 시간 범위
|
|
max_conversations: 최대 대화 개수
|
|
|
|
Returns:
|
|
[{"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}]
|
|
"""
|
|
history = self.get_recent_history(bot_id, hours, max_conversations)
|
|
|
|
if not history:
|
|
return []
|
|
|
|
# messages 포맷으로 변환 (추천 블록 제외 — LLM용 본문만)
|
|
messages = []
|
|
for conv in history:
|
|
user_content = str(conv.get("user_query") or "").strip()
|
|
assistant_content = strip_suggestion_block(conv.get("ai_response")).strip()
|
|
if not user_content or not assistant_content:
|
|
continue
|
|
messages.append({"role": "user", "content": user_content})
|
|
messages.append({
|
|
"role": "assistant",
|
|
"content": assistant_content,
|
|
})
|
|
|
|
return messages
|
|
|
|
def cleanup_old_records(self, days: int = 7):
|
|
"""
|
|
오래된 기록 정리 (TTL 인덱스와 별개로 수동 정리)
|
|
|
|
Args:
|
|
days: 보관 일수
|
|
"""
|
|
try:
|
|
cutoff_time = datetime.now(timezone.utc) - timedelta(days=days)
|
|
result = self.collection.delete_many({"timestamp": {"$lt": cutoff_time}})
|
|
print(f"[ChatHistory] 정리 완료: {result.deleted_count}개 삭제")
|
|
except PyMongoError as e:
|
|
print(f"[ChatHistory] 정리 실패: {e}")
|
|
|
|
def __del__(self):
|
|
"""연결 종료"""
|
|
if hasattr(self, 'client'):
|
|
self.client.close()
|
|
|
|
|
|
# 싱글톤 인스턴스
|
|
_chat_history_manager = None
|
|
|
|
def get_chat_history_manager() -> ChatHistoryManager:
|
|
"""ChatHistoryManager 싱글톤 인스턴스 반환"""
|
|
global _chat_history_manager
|
|
if _chat_history_manager is None:
|
|
_chat_history_manager = ChatHistoryManager()
|
|
return _chat_history_manager
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# 테스트
|
|
manager = get_chat_history_manager()
|
|
|
|
# 저장 테스트
|
|
doc_id = manager.save_conversation(
|
|
user_id="test_user_123",
|
|
user_query="테스트 질문입니다",
|
|
ai_response="테스트 답변입니다",
|
|
matched_questions=["관련 질문 1", "관련 질문 2"],
|
|
scores=[0.95, 0.88]
|
|
)
|
|
print(f"저장된 ID: {doc_id}")
|
|
|
|
# 조회 테스트
|
|
history = manager.get_recent_history("test_user_123")
|
|
print(f"조회 결과: {len(history)}개")
|
|
|
|
# 컨텍스트 생성 테스트
|
|
context = manager.get_context_for_llm("test_user_123")
|
|
print(f"컨텍스트:\n{context}")
|
|
|