21f6869898
- EXAONE: 분류기+프롬프트엔지니어+직접응답 (JSON 출력) - 간단한 질문은 EXAONE이 직접 답변 (파이프라인 스킵) - 복잡한 질문은 AI 최적화 프롬프트로 Gemma에 전달 - 모호한 질문은 사용자에게 추가 질문 (clarify) - user별 최근 대화 기억 (최대 10개, 1시간 TTL) - ModelAdapter: messages 직접 전달 옵션 추가 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
53 lines
1.6 KiB
Python
53 lines
1.6 KiB
Python
"""Conversation — user별 최근 대화 기억."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import defaultdict
|
|
from dataclasses import dataclass, field
|
|
from time import time
|
|
|
|
MAX_HISTORY = 10 # user당 최근 대화 수
|
|
HISTORY_TTL = 3600.0 # 1시간 이후 대화 만료
|
|
|
|
|
|
@dataclass
|
|
class Message:
|
|
role: str # "user" | "assistant"
|
|
content: str
|
|
timestamp: float = field(default_factory=time)
|
|
|
|
|
|
class ConversationStore:
|
|
def __init__(self) -> None:
|
|
self._history: dict[str, list[Message]] = defaultdict(list)
|
|
|
|
def add(self, user_id: str, role: str, content: str) -> None:
|
|
msgs = self._history[user_id]
|
|
msgs.append(Message(role=role, content=content))
|
|
# 최대 개수 제한
|
|
if len(msgs) > MAX_HISTORY:
|
|
self._history[user_id] = msgs[-MAX_HISTORY:]
|
|
|
|
def get(self, user_id: str) -> list[Message]:
|
|
"""만료되지 않은 최근 대화 반환."""
|
|
now = time()
|
|
msgs = self._history.get(user_id, [])
|
|
# TTL 필터
|
|
fresh = [m for m in msgs if now - m.timestamp < HISTORY_TTL]
|
|
self._history[user_id] = fresh
|
|
return fresh
|
|
|
|
def format_for_prompt(self, user_id: str) -> str:
|
|
"""EXAONE에 전달할 대화 이력 포맷."""
|
|
msgs = self.get(user_id)
|
|
if not msgs:
|
|
return ""
|
|
lines = []
|
|
for m in msgs[-6:]: # 최근 6개만
|
|
prefix = "사용자" if m.role == "user" else "이드"
|
|
lines.append(f"{prefix}: {m.content}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
conversation_store = ConversationStore()
|