feat: Conversation sqlite 영구화 — 재시작에도 컨텍스트 유지

- db/database.py: conversation_messages 테이블 + save/load/cleanup 헬퍼
- conversation.py: write-through (memory + DB) + lazy load on first access
  - 메모리 캐시 1시간 TTL, DB 7일 보관
  - add/get/format_for_prompt가 async로 변경
- worker.py: 모든 conversation_store 호출에 await 추가
- main.py lifespan에 startup cleanup 호출 (7일 이상 정리)

서버 재시작 후 "방금 그거 더 자세히" 같은 후속 질문이 컨텍스트 유지

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Hyungi Ahn
2026-04-07 09:02:38 +09:00
parent 26ccdb0f5e
commit 6b36063010
4 changed files with 110 additions and 22 deletions

View File

@@ -1,4 +1,6 @@
"""aiosqlite DB — 요청/응답 로깅 및 메트릭.""" """aiosqlite DB — 요청/응답 로깅 + 대화 영구화."""
from time import time
import aiosqlite import aiosqlite
@@ -23,6 +25,17 @@ CREATE TABLE IF NOT EXISTS request_logs (
CREATE INDEX IF NOT EXISTS idx_logs_job ON request_logs(job_id); CREATE INDEX IF NOT EXISTS idx_logs_job ON request_logs(job_id);
CREATE INDEX IF NOT EXISTS idx_logs_created ON request_logs(created_at); CREATE INDEX IF NOT EXISTS idx_logs_created ON request_logs(created_at);
CREATE TABLE IF NOT EXISTS conversation_messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
role TEXT NOT NULL,
content TEXT NOT NULL,
created_at REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_conv_user ON conversation_messages(user_id, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_conv_created ON conversation_messages(created_at);
""" """
# Phase 1 → Phase 2 마이그레이션 (이미 존재하면 무시) # Phase 1 → Phase 2 마이그레이션 (이미 존재하면 무시)
@@ -55,6 +68,39 @@ async def log_request(job_id: str, message: str, model: str, created_at: float):
await db.commit() await db.commit()
async def save_conversation_message(user_id: str, role: str, content: str, created_at: float):
"""대화 메시지 저장."""
async with aiosqlite.connect(settings.db_path) as db:
await db.execute(
"INSERT INTO conversation_messages (user_id, role, content, created_at) VALUES (?, ?, ?, ?)",
(user_id, role, content, created_at),
)
await db.commit()
async def load_conversation_messages(user_id: str, since: float, limit: int = 40) -> list[dict]:
"""user_id의 최근 메시지 로드 (오래된 순)."""
async with aiosqlite.connect(settings.db_path) as db:
db.row_factory = aiosqlite.Row
cursor = await db.execute(
"SELECT role, content, created_at FROM conversation_messages "
"WHERE user_id=? AND created_at >= ? ORDER BY created_at DESC LIMIT ?",
(user_id, since, limit),
)
rows = await cursor.fetchall()
# 오래된 순으로 뒤집기
return [{"role": r["role"], "content": r["content"], "created_at": r["created_at"]} for r in reversed(rows)]
async def cleanup_old_conversations(days: int = 7) -> int:
"""N일 이상 오래된 대화 삭제. 삭제된 행 수 반환."""
cutoff = time() - (days * 86400)
async with aiosqlite.connect(settings.db_path) as db:
cursor = await db.execute("DELETE FROM conversation_messages WHERE created_at < ?", (cutoff,))
await db.commit()
return cursor.rowcount
async def log_completion( async def log_completion(
job_id: str, job_id: str,
status: str, status: str,

View File

@@ -24,6 +24,15 @@ logging.basicConfig(
@asynccontextmanager @asynccontextmanager
async def lifespan(app: FastAPI): async def lifespan(app: FastAPI):
await init_db() await init_db()
# 7일 이상 오래된 대화 정리
try:
from db.database import cleanup_old_conversations
deleted = await cleanup_old_conversations(days=7)
if deleted:
import logging
logging.getLogger(__name__).info("Cleaned up %d old conversation messages", deleted)
except Exception:
pass
backend_registry.init_from_settings(settings) backend_registry.init_from_settings(settings)
backend_registry.start_health_loop(settings.health_check_interval) backend_registry.start_health_loop(settings.health_check_interval)
jq_module.init_queue(settings.max_concurrent_jobs) jq_module.init_queue(settings.max_concurrent_jobs)

View File

@@ -1,14 +1,18 @@
"""Conversation — user별 최근 대화 기억.""" """Conversation — user별 최근 대화 기억 (메모리 캐시 + sqlite write-through)."""
from __future__ import annotations from __future__ import annotations
import logging
from collections import defaultdict from collections import defaultdict
from dataclasses import dataclass, field from dataclasses import dataclass, field
from time import time from time import time
MAX_HISTORY = 10 # user당 최근 대화 수 logger = logging.getLogger(__name__)
HISTORY_TTL = 3600.0 # 1시간 이후 대화 만료
MAX_HISTORY = 10 # 메모리 캐시 user당 최근 대화 수
HISTORY_TTL = 3600.0 # 1시간 이후 메모리 만료 (DB는 7일 보관)
DRAFT_TTL = 300.0 # pending_draft 5분 만료 DRAFT_TTL = 300.0 # pending_draft 5분 만료
DB_LOAD_LIMIT = 40 # DB lazy load 시 최대 메시지 수
@dataclass @dataclass
@@ -21,27 +25,56 @@ class Message:
class ConversationStore: class ConversationStore:
def __init__(self) -> None: def __init__(self) -> None:
self._history: dict[str, list[Message]] = defaultdict(list) self._history: dict[str, list[Message]] = defaultdict(list)
self._loaded_users: set[str] = set() # DB lazy load 완료한 user_id
self._pending_drafts: dict[str, tuple[dict, float]] = {} # user_id → (draft_data, created_at) self._pending_drafts: dict[str, tuple[dict, float]] = {} # user_id → (draft_data, created_at)
def add(self, user_id: str, role: str, content: str) -> None: async def add(self, user_id: str, role: str, content: str) -> None:
"""메모리 + DB write-through."""
from db.database import save_conversation_message
msg = Message(role=role, content=content)
# 메모리 캐시
msgs = self._history[user_id] msgs = self._history[user_id]
msgs.append(Message(role=role, content=content)) msgs.append(msg)
# 최대 개수 제한
if len(msgs) > MAX_HISTORY: if len(msgs) > MAX_HISTORY:
self._history[user_id] = msgs[-MAX_HISTORY:] self._history[user_id] = msgs[-MAX_HISTORY:]
def get(self, user_id: str) -> list[Message]: # DB write-through (실패해도 메모리에는 남음)
"""만료되지 않은 최근 대화 반환.""" try:
await save_conversation_message(user_id, role, content, msg.timestamp)
except Exception:
logger.warning("Failed to persist conversation for %s", user_id, exc_info=True)
async def get(self, user_id: str) -> list[Message]:
"""만료되지 않은 최근 대화. 메모리 비어있으면 DB lazy load."""
from db.database import load_conversation_messages
now = time() now = time()
# 메모리 캐시에서 fresh 한 것만
msgs = self._history.get(user_id, []) msgs = self._history.get(user_id, [])
# TTL 필터
fresh = [m for m in msgs if now - m.timestamp < HISTORY_TTL] fresh = [m for m in msgs if now - m.timestamp < HISTORY_TTL]
# 메모리 비어있고 DB lazy load 안 했으면 DB에서 가져옴
if not fresh and user_id not in self._loaded_users:
self._loaded_users.add(user_id)
since = now - HISTORY_TTL
try:
db_msgs = await load_conversation_messages(user_id, since, limit=DB_LOAD_LIMIT)
fresh = [
Message(role=m["role"], content=m["content"], timestamp=m["created_at"])
for m in db_msgs
]
if fresh:
logger.info("Loaded %d messages from DB for %s", len(fresh), user_id)
except Exception:
logger.warning("Failed to load conversation from DB for %s", user_id, exc_info=True)
self._history[user_id] = fresh self._history[user_id] = fresh
return fresh return fresh
def format_for_prompt(self, user_id: str) -> str: async def format_for_prompt(self, user_id: str) -> str:
"""EXAONE에 전달할 대화 이력 포맷.""" """EXAONE에 전달할 대화 이력 포맷."""
msgs = self.get(user_id) msgs = await self.get(user_id)
lines = [] lines = []
for m in msgs[-6:]: # 최근 6개만 for m in msgs[-6:]: # 최근 6개만
prefix = "사용자" if m.role == "user" else "이드" prefix = "사용자" if m.role == "user" else "이드"

View File

@@ -240,7 +240,7 @@ async def run(job: Job) -> None:
kst = timezone(timedelta(hours=9)) kst = timezone(timedelta(hours=9))
now_kst = datetime.now(kst) now_kst = datetime.now(kst)
now_str = now_kst.strftime("%Y년 %m월 %d%H:%M (%A) KST") now_str = now_kst.strftime("%Y년 %m월 %d%H:%M (%A) KST")
history = conversation_store.format_for_prompt(user_id) history = await conversation_store.format_for_prompt(user_id)
classify_input = f"[현재 시간]\n{now_str}\n\n" classify_input = f"[현재 시간]\n{now_str}\n\n"
if history: if history:
classify_input += f"[대화 이력]\n{history}\n\n" classify_input += f"[대화 이력]\n{history}\n\n"
@@ -268,7 +268,7 @@ async def run(job: Job) -> None:
logger.info("Job %s classified as '%s'", job.id, action) logger.info("Job %s classified as '%s'", job.id, action)
# 대화 기록: 사용자 메시지 # 대화 기록: 사용자 메시지
conversation_store.add(user_id, "user", job.message) await conversation_store.add(user_id, "user", job.message)
collected: list[str] = [] collected: list[str] = []
@@ -281,7 +281,7 @@ async def run(job: Job) -> None:
status_text = await _build_system_status(force_measure=True) status_text = await _build_system_status(force_measure=True)
collected.append(status_text) collected.append(status_text)
await state_stream.push(job.id, "result", {"content": status_text}) await state_stream.push(job.id, "result", {"content": status_text})
conversation_store.add(user_id, "assistant", status_text) await conversation_store.add(user_id, "assistant", status_text)
elif action == "tools": elif action == "tools":
# === TOOLS: 도구 실행 === # === TOOLS: 도구 실행 ===
@@ -299,7 +299,7 @@ async def run(job: Job) -> None:
response = "확인할 일정이 없습니다. 다시 요청해주세요." response = "확인할 일정이 없습니다. 다시 요청해주세요."
collected.append(response) collected.append(response)
await state_stream.push(job.id, "result", {"content": response}) await state_stream.push(job.id, "result", {"content": response})
conversation_store.add(user_id, "assistant", response) await conversation_store.add(user_id, "assistant", response)
else: else:
try: try:
result = await asyncio.wait_for(execute_tool(tool_name, operation, draft), timeout=TOOL_TIMEOUT) result = await asyncio.wait_for(execute_tool(tool_name, operation, draft), timeout=TOOL_TIMEOUT)
@@ -309,7 +309,7 @@ async def run(job: Job) -> None:
response = result.get("summary", "") if result["ok"] else result.get("error", "⚠️ 오류") response = result.get("summary", "") if result["ok"] else result.get("error", "⚠️ 오류")
collected.append(response) collected.append(response)
await state_stream.push(job.id, "result", {"content": response}) await state_stream.push(job.id, "result", {"content": response})
conversation_store.add(user_id, "assistant", response) await conversation_store.add(user_id, "assistant", response)
else: else:
# 일반 도구 실행 # 일반 도구 실행
try: try:
@@ -355,13 +355,13 @@ async def run(job: Job) -> None:
collected.append(response) collected.append(response)
await state_stream.push(job.id, "result", {"content": response}) await state_stream.push(job.id, "result", {"content": response})
conversation_store.add(user_id, "assistant", "".join(collected)) await conversation_store.add(user_id, "assistant", "".join(collected))
elif action == "clarify": elif action == "clarify":
# === CLARIFY: 추가 질문 === # === CLARIFY: 추가 질문 ===
collected.append(response_text) collected.append(response_text)
await state_stream.push(job.id, "result", {"content": response_text}) await state_stream.push(job.id, "result", {"content": response_text})
conversation_store.add(user_id, "assistant", response_text) await conversation_store.add(user_id, "assistant", response_text)
elif action == "route" and settings.pipeline_enabled and backend_registry.is_healthy("reasoner"): elif action == "route" and settings.pipeline_enabled and backend_registry.is_healthy("reasoner"):
# === ROUTE: Gemma reasoning === # === ROUTE: Gemma reasoning ===
@@ -396,7 +396,7 @@ async def run(job: Job) -> None:
reasoner_system = f"{backend_registry.reasoner.system_prompt}\n\n현재 시간: {now_kst} (한국 표준시)" reasoner_system = f"{backend_registry.reasoner.system_prompt}\n\n현재 시간: {now_kst} (한국 표준시)"
# 대화 이력을 OpenAI messages 형식으로 변환 (현재 user 메시지 포함됨) # 대화 이력을 OpenAI messages 형식으로 변환 (현재 user 메시지 포함됨)
history_msgs = conversation_store.get(user_id)[-10:] history_msgs = (await conversation_store.get(user_id))[-10:]
reasoner_messages = [{"role": "system", "content": reasoner_system}] reasoner_messages = [{"role": "system", "content": reasoner_system}]
for m in history_msgs: for m in history_msgs:
reasoner_messages.append({"role": m.role, "content": m.content}) reasoner_messages.append({"role": m.role, "content": m.content})
@@ -419,7 +419,7 @@ async def run(job: Job) -> None:
return return
if collected: if collected:
conversation_store.add(user_id, "assistant", "".join(collected)) await conversation_store.add(user_id, "assistant", "".join(collected))
else: else:
# === DIRECT: EXAONE 직접 응답 === # === DIRECT: EXAONE 직접 응답 ===
@@ -435,7 +435,7 @@ async def run(job: Job) -> None:
return return
if collected: if collected:
conversation_store.add(user_id, "assistant", "".join(collected)) await conversation_store.add(user_id, "assistant", "".join(collected))
# --- Complete --- # --- Complete ---
if not collected: if not collected: