htlee
e797beaac6
- Ollama Docker(14b/32b) + Redis 컨텍스트 캐싱 + 대화 히스토리 - Python SSE 채팅 엔드포인트 + 사전 쿼리 + Tool Calling - 도메인 지식(해양법/어업협정/알고리즘) + DB 스키마 가이드 - Frontend SSE 스트리밍 + 타이머 + thinking 접기 + 확장 UI
91 lines
2.6 KiB
Python
91 lines
2.6 KiB
Python
"""Redis 캐시 유틸 — 분석 컨텍스트 + 대화 히스토리."""
|
|
|
|
import json
|
|
import logging
|
|
from typing import Optional
|
|
|
|
import redis
|
|
|
|
from config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_redis: Optional[redis.Redis] = None
|
|
|
|
|
|
def _get_redis() -> redis.Redis:
|
|
global _redis
|
|
if _redis is None:
|
|
_redis = redis.Redis(
|
|
host=settings.REDIS_HOST,
|
|
port=settings.REDIS_PORT,
|
|
password=settings.REDIS_PASSWORD or None,
|
|
decode_responses=True,
|
|
socket_connect_timeout=3,
|
|
)
|
|
return _redis
|
|
|
|
|
|
# ── 분석 컨텍스트 캐시 (전역, 5분 주기 갱신) ──
|
|
|
|
CONTEXT_KEY = 'kcg:chat:context'
|
|
CONTEXT_TTL = 360 # 6분 (5분 주기 + 1분 버퍼)
|
|
|
|
|
|
def cache_analysis_context(context_dict: dict):
|
|
"""스케줄러에서 분석 완료 후 호출 — Redis에 요약 데이터 캐싱."""
|
|
try:
|
|
r = _get_redis()
|
|
r.setex(CONTEXT_KEY, CONTEXT_TTL, json.dumps(context_dict, ensure_ascii=False, default=str))
|
|
logger.debug('cached analysis context (%d bytes)', len(json.dumps(context_dict)))
|
|
except Exception as e:
|
|
logger.warning('failed to cache analysis context: %s', e)
|
|
|
|
|
|
def get_cached_context() -> Optional[dict]:
|
|
"""Redis에서 캐시된 분석 컨텍스트 조회."""
|
|
try:
|
|
r = _get_redis()
|
|
data = r.get(CONTEXT_KEY)
|
|
return json.loads(data) if data else None
|
|
except Exception as e:
|
|
logger.warning('failed to read cached context: %s', e)
|
|
return None
|
|
|
|
|
|
# ── 대화 히스토리 (계정별, 24h TTL) ──
|
|
|
|
HISTORY_TTL = 86400 # 24시간
|
|
MAX_HISTORY = 50
|
|
|
|
|
|
def save_chat_history(user_id: str, messages: list[dict]):
|
|
"""대화 히스토리 저장 (최근 50개 메시지만 유지)."""
|
|
try:
|
|
r = _get_redis()
|
|
key = f'kcg:chat:history:{user_id}'
|
|
trimmed = messages[-MAX_HISTORY:]
|
|
r.setex(key, HISTORY_TTL, json.dumps(trimmed, ensure_ascii=False))
|
|
except Exception as e:
|
|
logger.warning('failed to save chat history for %s: %s', user_id, e)
|
|
|
|
|
|
def load_chat_history(user_id: str) -> list[dict]:
|
|
"""대화 히스토리 로드."""
|
|
try:
|
|
r = _get_redis()
|
|
data = r.get(f'kcg:chat:history:{user_id}')
|
|
return json.loads(data) if data else []
|
|
except Exception as e:
|
|
logger.warning('failed to load chat history for %s: %s', user_id, e)
|
|
return []
|
|
|
|
|
|
def clear_chat_history(user_id: str):
|
|
"""대화 히스토리 삭제."""
|
|
try:
|
|
r = _get_redis()
|
|
r.delete(f'kcg:chat:history:{user_id}')
|
|
except Exception as e:
|
|
logger.warning('failed to clear chat history for %s: %s', user_id, e)
|