Agent 2.0 exdev 서버 배포 스택
- server-dev start/stop/deploy 및 Gitea push 자동 배포 - local-dev 로컬 개발 환경 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Macbook
@@ -0,0 +1,33 @@
|
||||
"""
|
||||
RAG 챗봇 핸들러 모듈
|
||||
──────────────────
|
||||
기능별로 분리된 핸들러 컴포넌트
|
||||
"""
|
||||
|
||||
from .config import Config
|
||||
from .query_rewriter import QueryRewriter
|
||||
from .search_handler import SearchHandler
|
||||
from .prompt_builder import PromptBuilder
|
||||
from .llm_handler import LLMHandler
|
||||
from .response_handler import ResponseHandler
|
||||
from .intent_detector import IntentDetector, Intent
|
||||
from .greeting_handler import GreetingHandler
|
||||
from .emotion_detector import EmotionDetector, Emotion
|
||||
from .emotion_handler import EmotionHandler
|
||||
from .suggestion_handler import SuggestionHandler
|
||||
|
||||
__all__ = [
|
||||
"Config",
|
||||
"QueryRewriter",
|
||||
"SearchHandler",
|
||||
"PromptBuilder",
|
||||
"LLMHandler",
|
||||
"ResponseHandler",
|
||||
"IntentDetector",
|
||||
"Intent",
|
||||
"GreetingHandler",
|
||||
"EmotionDetector",
|
||||
"Emotion",
|
||||
"EmotionHandler",
|
||||
"SuggestionHandler"
|
||||
]
|
||||
@@ -0,0 +1,92 @@
|
||||
"""
|
||||
설정 관리 모듈
|
||||
────────────
|
||||
환경 변수 기반 설정 관리
|
||||
"""
|
||||
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class Config:
|
||||
"""RAG 시스템 설정"""
|
||||
|
||||
# 벡터 검색 설정
|
||||
top_k: int
|
||||
threshold: float
|
||||
threshold_rewrite: float
|
||||
hybrid_search_enabled: bool
|
||||
sparse_top_k: int
|
||||
hybrid_merge_top_k: int
|
||||
|
||||
# 재랭킹 설정
|
||||
rerank_candidates: int
|
||||
rerank_batch_size: int
|
||||
top_n_for_llm: int
|
||||
# 리랭커 1위 점수 미만 → 제안 문구·Full Context Rewriting (Qwen3-Reranker sigmoid 0~1)
|
||||
low_confidence_threshold: float
|
||||
# low 이상이면 medium, 이상이면 high (MongoDB answer_confidence)
|
||||
high_confidence_threshold: float
|
||||
|
||||
# LLM 설정
|
||||
llm_max_tokens: int
|
||||
|
||||
# Query Rewriting 설정
|
||||
query_rewrite_enabled: bool
|
||||
|
||||
# 대화 이력 설정
|
||||
chat_history_enabled: bool
|
||||
chat_history_limit: int
|
||||
chat_history_hours: int
|
||||
chat_history_always_include: bool
|
||||
|
||||
@classmethod
|
||||
def from_env(cls, chat_history_enabled: bool = False) -> "Config":
|
||||
"""환경 변수에서 설정 로드"""
|
||||
return cls(
|
||||
# 벡터 검색
|
||||
top_k=int(os.getenv("FAISS_TOP_K", "30")),
|
||||
threshold=float(os.getenv("FAISS_THRESHOLD", "0.55")),
|
||||
threshold_rewrite=float(os.getenv("FAISS_THRESHOLD_REWRITE", "0.50")),
|
||||
hybrid_search_enabled=os.getenv("HYBRID_SEARCH_ENABLED", "false").lower() == "true",
|
||||
sparse_top_k=int(os.getenv("SPARSE_TOP_K", "30")),
|
||||
hybrid_merge_top_k=int(os.getenv("HYBRID_MERGE_TOP_K", "40")),
|
||||
|
||||
# 재랭킹
|
||||
rerank_candidates=int(os.getenv("RERANK_CANDIDATES", "20")),
|
||||
rerank_batch_size=int(os.getenv("RERANK_BATCH_SIZE", "16")),
|
||||
top_n_for_llm=int(os.getenv("TOP_N_FOR_LLM", "5")),
|
||||
low_confidence_threshold=float(os.getenv("LOW_CONFIDENCE_THRESHOLD", "0.65")),
|
||||
high_confidence_threshold=float(os.getenv("HIGH_CONFIDENCE_THRESHOLD", "0.75")),
|
||||
|
||||
# LLM
|
||||
llm_max_tokens=int(os.getenv("LLM_MAX_TOKENS", "2048")),
|
||||
|
||||
# Query Rewriting
|
||||
query_rewrite_enabled=os.getenv("QUERY_REWRITE_ENABLED", "true").lower() == "true",
|
||||
|
||||
# 대화 이력
|
||||
chat_history_enabled=chat_history_enabled,
|
||||
chat_history_limit=int(os.getenv("CHAT_HISTORY_LIMIT", "10")),
|
||||
chat_history_hours=int(os.getenv("CHAT_HISTORY_HOURS", "24")),
|
||||
chat_history_always_include=os.getenv("CHAT_HISTORY_ALWAYS_INCLUDE", "true").lower() == "true"
|
||||
)
|
||||
|
||||
def print_summary(self):
|
||||
"""설정 요약 출력"""
|
||||
print(f"[Config] Query Rewriting: {'활성화' if self.query_rewrite_enabled else '비활성화'}")
|
||||
print(f"[Config] FAISS Threshold: 1차={self.threshold}, 2차(Rewrite)={self.threshold_rewrite}")
|
||||
print(
|
||||
f"[Config] Hybrid Search: {'활성화' if self.hybrid_search_enabled else '비활성화'} "
|
||||
f"(sparse_top_k={self.sparse_top_k}, merge_top_k={self.hybrid_merge_top_k})"
|
||||
)
|
||||
print(
|
||||
f"[Config] Reranker 신뢰도: low<{self.low_confidence_threshold}, "
|
||||
f"high>={self.high_confidence_threshold}"
|
||||
)
|
||||
print(f"[Config] 대화 이력: {'활성화' if self.chat_history_enabled else '비활성화'}")
|
||||
if self.chat_history_enabled:
|
||||
print(f"[Config] - 최근 {self.chat_history_limit}개 / {self.chat_history_hours}시간")
|
||||
print(f"[Config] - 정상 질문 포함: {'예' if self.chat_history_always_include else '아니오'}")
|
||||
@@ -0,0 +1,169 @@
|
||||
"""
|
||||
감정 분석 모듈
|
||||
──────────────
|
||||
사용자 질문의 감정 상태를 세밀하게 분석
|
||||
"""
|
||||
|
||||
from typing import Dict, List, Optional
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass
|
||||
class Emotion:
|
||||
"""감정 분석 결과"""
|
||||
primary: str # 주요 감정: angry, frustrated, satisfied, confused, worried, neutral
|
||||
intensity: float # 강도: 0.0 (약함) ~ 1.0 (강함)
|
||||
confidence: float # 신뢰도: 0.0 ~ 1.0
|
||||
matched_keywords: List[str] # 매칭된 키워드
|
||||
|
||||
def needs_empathy(self) -> bool:
|
||||
"""공감이 필요한 감정인지 (부정적 감정)"""
|
||||
return self.primary in ["angry", "frustrated", "worried"]
|
||||
|
||||
def is_positive(self) -> bool:
|
||||
"""긍정적 감정인지"""
|
||||
return self.primary == "satisfied"
|
||||
|
||||
def needs_clarification(self) -> bool:
|
||||
"""명확한 설명이 필요한지 (혼란)"""
|
||||
return self.primary == "confused"
|
||||
|
||||
|
||||
class EmotionDetector:
|
||||
"""사용자 감정 분석기"""
|
||||
|
||||
# 감정별 키워드 및 강도
|
||||
EMOTION_PATTERNS = {
|
||||
"angry": {
|
||||
"keywords": {
|
||||
# (키워드, 강도)
|
||||
"화나": 0.9, "화남": 0.9, "분노": 1.0, "열받": 0.9,
|
||||
"짜증": 0.7, "짜증나": 0.8, "짜증남": 0.8,
|
||||
"불만": 0.6, "불쾌": 0.7, "기분 나쁘": 0.6,
|
||||
"최악": 0.9, "엉망": 0.7, "터무니": 0.8,
|
||||
"너무해": 0.8, "심각": 0.7, "문제": 0.5,
|
||||
"불친절": 0.7, "무례": 0.8, "실망": 0.6
|
||||
},
|
||||
"description": "화남/분노"
|
||||
},
|
||||
"frustrated": {
|
||||
"keywords": {
|
||||
"답답": 0.8, "막막": 0.7, "곤란": 0.6,
|
||||
"어렵": 0.5, "힘들": 0.6, "난감": 0.7,
|
||||
"복잡": 0.5, "이해가 안": 0.6, "잘 모르": 0.5,
|
||||
"왜 안": 0.6, "계속": 0.4, "여전": 0.5,
|
||||
"해결이 안": 0.7, "안 되": 0.6
|
||||
},
|
||||
"description": "답답함/막막함"
|
||||
},
|
||||
"satisfied": {
|
||||
"keywords": {
|
||||
"감사": 0.8, "고마워": 0.8, "고맙": 0.8,
|
||||
"좋": 0.7, "훌륭": 0.9, "최고": 0.9,
|
||||
"도움": 0.7, "해결": 0.8, "완벽": 0.9,
|
||||
"잘": 0.6, "쉽": 0.6, "편리": 0.7,
|
||||
"만족": 0.9, "훌륭": 0.9
|
||||
},
|
||||
"description": "만족/긍정"
|
||||
},
|
||||
"confused": {
|
||||
"keywords": {
|
||||
"모르겠": 0.8, "헷갈": 0.9, "혼란": 0.9,
|
||||
"무슨": 0.6, "뭐가": 0.6,
|
||||
"이해가": 0.6, "뜻이": 0.5, "의미": 0.5,
|
||||
"차이": 0.5, "구별": 0.6, "잘 안": 0.6
|
||||
},
|
||||
"description": "혼란/이해 부족"
|
||||
},
|
||||
"worried": {
|
||||
"keywords": {
|
||||
"걱정": 0.8, "불안": 0.9, "염려": 0.7,
|
||||
"겁나": 0.8, "무섭": 0.7, "두렵": 0.7,
|
||||
"조심": 0.5, "주의": 0.5, "위험": 0.6,
|
||||
"문제가": 0.6, "괜찮": 0.5
|
||||
},
|
||||
"description": "걱정/불안"
|
||||
}
|
||||
}
|
||||
|
||||
def detect(self, query: str, ts: Optional[str] = None) -> Emotion:
|
||||
"""감정 분석
|
||||
|
||||
Args:
|
||||
query: 사용자 질문
|
||||
ts: 타임스탬프 (로깅용)
|
||||
|
||||
Returns:
|
||||
Emotion 객체
|
||||
"""
|
||||
query_lower = query.lower().strip()
|
||||
|
||||
# 빈 문자열 체크
|
||||
if not query_lower:
|
||||
return Emotion(
|
||||
primary="neutral",
|
||||
intensity=0.0,
|
||||
confidence=1.0,
|
||||
matched_keywords=[]
|
||||
)
|
||||
|
||||
# 각 감정별 점수 계산
|
||||
emotion_scores = {}
|
||||
|
||||
for emotion_name, config in self.EMOTION_PATTERNS.items():
|
||||
matched = []
|
||||
total_intensity = 0.0
|
||||
|
||||
for keyword, intensity in config["keywords"].items():
|
||||
if keyword in query_lower:
|
||||
matched.append(keyword)
|
||||
total_intensity += intensity
|
||||
|
||||
if matched:
|
||||
# 평균 강도 계산
|
||||
avg_intensity = total_intensity / len(matched)
|
||||
# 매칭 개수와 강도를 결합한 점수
|
||||
score = len(matched) * avg_intensity
|
||||
|
||||
emotion_scores[emotion_name] = {
|
||||
"score": score,
|
||||
"intensity": min(1.0, avg_intensity),
|
||||
"matched": matched
|
||||
}
|
||||
|
||||
# 매칭된 감정이 없으면 중립
|
||||
if not emotion_scores:
|
||||
return Emotion(
|
||||
primary="neutral",
|
||||
intensity=0.0,
|
||||
confidence=1.0,
|
||||
matched_keywords=[]
|
||||
)
|
||||
|
||||
# 가장 높은 점수의 감정 선택
|
||||
best_emotion = max(emotion_scores.items(), key=lambda x: x[1]["score"])
|
||||
emotion_name = best_emotion[0]
|
||||
emotion_data = best_emotion[1]
|
||||
|
||||
# 신뢰도 계산 (매칭 개수와 강도 기반)
|
||||
confidence = min(0.95, 0.6 + (emotion_data["score"] * 0.1))
|
||||
|
||||
result = Emotion(
|
||||
primary=emotion_name,
|
||||
intensity=emotion_data["intensity"],
|
||||
confidence=confidence,
|
||||
matched_keywords=emotion_data["matched"]
|
||||
)
|
||||
|
||||
if ts:
|
||||
print(f"[EmotionDetector] {ts} 감정 분석: {result.primary} "
|
||||
f"(강도: {result.intensity:.2f}, 신뢰도: {result.confidence:.2f}, "
|
||||
f"키워드: {result.matched_keywords})")
|
||||
|
||||
return result
|
||||
|
||||
def get_emotion_summary(self, emotion: Emotion) -> str:
|
||||
"""감정 요약 문자열"""
|
||||
description = self.EMOTION_PATTERNS.get(emotion.primary, {}).get("description", emotion.primary)
|
||||
intensity_label = "강함" if emotion.intensity > 0.7 else "중간" if emotion.intensity > 0.4 else "약함"
|
||||
return f"{description} ({intensity_label})"
|
||||
@@ -0,0 +1,171 @@
|
||||
"""
|
||||
감정별 공감 응답 핸들러
|
||||
──────────────────────
|
||||
감정에 따른 공감 메시지 및 프롬프트 커스터마이징
|
||||
"""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class EmotionHandler:
|
||||
"""감정별 공감 응답 생성기"""
|
||||
|
||||
# 감정별 공감 프리픽스
|
||||
EMPATHY_PREFIXES = {
|
||||
"angry": [
|
||||
"고객님의 불편한 경험에 대해 진심으로 사과드립니다. 😔",
|
||||
"불편을 드려 정말 죄송합니다. 😔",
|
||||
"고객님의 화나신 마음을 충분히 이해합니다. 😔"
|
||||
],
|
||||
"frustrated": [
|
||||
"답답하셨겠습니다. 😓",
|
||||
"막막하셨을 것 같습니다. 😓",
|
||||
"불편하셨겠네요. 저희가 더 명확히 안내드리겠습니다. 😓"
|
||||
],
|
||||
"satisfied": [
|
||||
"도움이 되셨다니 정말 기쁩니다! 😊",
|
||||
"만족하셨다니 다행입니다! 😊",
|
||||
"고객님께 도움을 드릴 수 있어 기쁩니다! 😊"
|
||||
],
|
||||
"confused": [
|
||||
"이해하기 어려우셨군요. 제가 더 명확히 설명드리겠습니다. 🤔",
|
||||
"헷갈리셨을 것 같네요. 차근차근 설명드리겠습니다. 🤔",
|
||||
"복잡하게 느껴지셨나봅니다. 쉽게 풀어서 설명드릴게요. 🤔"
|
||||
],
|
||||
"worried": [
|
||||
"걱정되시는 부분이 있으시군요. 안심하셔도 됩니다. 😌",
|
||||
"염려하지 않으셔도 괜찮습니다. 자세히 안내드릴게요. 😌",
|
||||
"불안하셨겠습니다. 정확한 정보를 드리겠습니다. 😌"
|
||||
],
|
||||
"neutral": []
|
||||
}
|
||||
|
||||
# 감정별 시스템 프롬프트 추가 지시사항
|
||||
EMOTION_INSTRUCTIONS = {
|
||||
"angry": (
|
||||
"\n【감정 고려사항 - 화난 고객】\n"
|
||||
"⚠️ 고객이 매우 화가 난 상태입니다:\n"
|
||||
"1. 먼저 진심으로 사과하세요.\n"
|
||||
"2. 고객의 감정을 인정하고 공감하세요.\n"
|
||||
"3. 구체적인 해결책을 빠르게 제시하세요.\n"
|
||||
"4. 추가 불편을 드리지 않도록 명확하고 간결하게 답변하세요.\n"
|
||||
"5. 필요시 상담원 연결이나 콜센터 안내를 제안하세요."
|
||||
),
|
||||
"frustrated": (
|
||||
"\n【감정 고려사항 - 답답한 고객】\n"
|
||||
"💡 고객이 문제 해결에 어려움을 겪고 있습니다:\n"
|
||||
"1. 고객의 답답함에 공감하세요.\n"
|
||||
"2. 복잡한 설명보다는 단계별로 쉽게 설명하세요.\n"
|
||||
"3. 즉시 실행 가능한 해결 방법을 제시하세요.\n"
|
||||
"4. 추가 도움이 필요한지 물어보세요."
|
||||
),
|
||||
"satisfied": (
|
||||
"\n【감정 고려사항 - 만족한 고객】\n"
|
||||
"😊 고객이 긍정적인 상태입니다:\n"
|
||||
"1. 긍정적인 톤을 유지하세요.\n"
|
||||
"2. 추가로 도움이 될 만한 정보를 제안하세요.\n"
|
||||
"3. 다른 질문이 있는지 자연스럽게 물어보세요."
|
||||
),
|
||||
"confused": (
|
||||
"\n【감정 고려사항 - 혼란스러운 고객】\n"
|
||||
"🤔 고객이 개념 이해에 어려움을 겪고 있습니다:\n"
|
||||
"1. 전문 용어를 피하고 쉬운 말로 설명하세요.\n"
|
||||
"2. 예시를 들어 설명하세요.\n"
|
||||
"3. 단계를 나눠서 차근차근 설명하세요.\n"
|
||||
"4. 추가 질문을 환영하는 분위기를 만드세요."
|
||||
),
|
||||
"worried": (
|
||||
"\n【감정 고려사항 - 걱정하는 고객】\n"
|
||||
"😌 고객이 불안해하거나 걱정하고 있습니다:\n"
|
||||
"1. 먼저 안심시키세요.\n"
|
||||
"2. 정확하고 신뢰할 수 있는 정보를 제공하세요.\n"
|
||||
"3. 예방 방법이나 주의사항을 함께 안내하세요.\n"
|
||||
"4. 문제가 없음을 명확히 전달하세요."
|
||||
),
|
||||
"neutral": ""
|
||||
}
|
||||
|
||||
def get_empathy_prefix(self, emotion_name: str, intensity: float) -> Optional[str]:
|
||||
"""감정에 맞는 공감 프리픽스 반환
|
||||
|
||||
Args:
|
||||
emotion_name: 감정 이름
|
||||
intensity: 감정 강도 (0.0 ~ 1.0)
|
||||
|
||||
Returns:
|
||||
공감 메시지 또는 None
|
||||
"""
|
||||
prefixes = self.EMPATHY_PREFIXES.get(emotion_name, [])
|
||||
|
||||
if not prefixes:
|
||||
return None
|
||||
|
||||
# 강도에 따라 선택
|
||||
if intensity > 0.7:
|
||||
# 강한 감정 → 첫 번째 (가장 강한 공감)
|
||||
return prefixes[0]
|
||||
elif intensity > 0.4:
|
||||
# 중간 감정 → 두 번째
|
||||
return prefixes[min(1, len(prefixes) - 1)]
|
||||
else:
|
||||
# 약한 감정 → 마지막 (가장 약한 공감)
|
||||
return prefixes[-1]
|
||||
|
||||
def get_emotion_instruction(self, emotion_name: str) -> str:
|
||||
"""감정별 시스템 프롬프트 추가 지시사항"""
|
||||
return self.EMOTION_INSTRUCTIONS.get(emotion_name, "")
|
||||
|
||||
def should_add_empathy(self, emotion_name: str, intensity: float, confidence: float) -> bool:
|
||||
"""공감 메시지를 추가해야 하는지 판단
|
||||
|
||||
Args:
|
||||
emotion_name: 감정 이름
|
||||
intensity: 감정 강도
|
||||
confidence: 신뢰도
|
||||
|
||||
Returns:
|
||||
공감 메시지 추가 여부
|
||||
"""
|
||||
# 중립 감정은 공감 불필요
|
||||
if emotion_name == "neutral":
|
||||
return False
|
||||
|
||||
# 신뢰도가 낮으면 공감 추가 안함
|
||||
if confidence < 0.6:
|
||||
return False
|
||||
|
||||
# 부정적 감정 (angry, frustrated, worried)은 강도 상관없이 공감
|
||||
if emotion_name in ["angry", "frustrated", "worried"]:
|
||||
return True
|
||||
|
||||
# 긍정적/혼란 감정은 강도가 충분히 높을 때만
|
||||
return intensity > 0.5
|
||||
|
||||
def enhance_answer_with_empathy(
|
||||
self,
|
||||
answer: str,
|
||||
emotion_name: str,
|
||||
intensity: float,
|
||||
confidence: float
|
||||
) -> str:
|
||||
"""답변에 공감 메시지 추가
|
||||
|
||||
Args:
|
||||
answer: 원본 답변
|
||||
emotion_name: 감정 이름
|
||||
intensity: 감정 강도
|
||||
confidence: 신뢰도
|
||||
|
||||
Returns:
|
||||
공감 메시지가 추가된 답변
|
||||
"""
|
||||
if not self.should_add_empathy(emotion_name, intensity, confidence):
|
||||
return answer
|
||||
|
||||
empathy_prefix = self.get_empathy_prefix(emotion_name, intensity)
|
||||
|
||||
if not empathy_prefix:
|
||||
return answer
|
||||
|
||||
# 공감 메시지를 답변 앞에 추가
|
||||
return f"{empathy_prefix}\n\n{answer}"
|
||||
@@ -0,0 +1,132 @@
|
||||
"""
|
||||
인사/종료 응답 핸들러
|
||||
──────────────────────
|
||||
인사, 종료 의도에 대한 특별 응답 생성
|
||||
"""
|
||||
|
||||
from typing import List, Dict, Any, Optional
|
||||
|
||||
|
||||
class GreetingHandler:
|
||||
"""인사 및 종료 응답 생성기"""
|
||||
|
||||
# 퀵 리플라이 템플릿
|
||||
GREETING_QUICK_REPLIES = [
|
||||
"통행료 조회",
|
||||
"하이패스 문의",
|
||||
"환불 신청",
|
||||
"휴게소 안내"
|
||||
]
|
||||
|
||||
FAREWELL_QUICK_REPLIES = [
|
||||
"추가 문의하기",
|
||||
"처음으로",
|
||||
"상담 종료"
|
||||
]
|
||||
|
||||
def generate_greeting_response(
|
||||
self,
|
||||
query: str,
|
||||
matched_keywords: List[str],
|
||||
bot_id: Optional[str] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""인사 응답 생성
|
||||
|
||||
Args:
|
||||
query: 원본 질문
|
||||
matched_keywords: 매칭된 키워드
|
||||
bot_id: 봇 ID
|
||||
|
||||
Returns:
|
||||
응답 딕셔너리 (answer, quick_replies 포함)
|
||||
"""
|
||||
answer = (
|
||||
"안녕하세요! 한국도로공사 채팅상담 챗봇입니다. 😊\n\n"
|
||||
"고속도로 이용과 관련하여 궁금하신 점을 편하게 물어보세요.\n\n"
|
||||
"📌 자주 묻는 질문\n"
|
||||
"• 통행료 조회 및 환불\n"
|
||||
"• 하이패스 발급 및 사용법\n"
|
||||
"• 휴게소 및 편의시설 안내\n"
|
||||
"• 고속도로 소음/환경 민원\n\n"
|
||||
"무엇을 도와드릴까요?"
|
||||
)
|
||||
|
||||
return {
|
||||
"answer": answer,
|
||||
"matched_questions": [],
|
||||
"scores": [],
|
||||
"num_references": 0,
|
||||
"botId": bot_id,
|
||||
"intent": "greeting",
|
||||
"quick_replies": self.GREETING_QUICK_REPLIES,
|
||||
"rerank_info": {
|
||||
"used": False,
|
||||
"detail": "Greeting intent detected"
|
||||
}
|
||||
}
|
||||
|
||||
def generate_farewell_response(
|
||||
self,
|
||||
query: str,
|
||||
matched_keywords: List[str],
|
||||
bot_id: Optional[str] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""종료 인사 응답 생성"""
|
||||
# 감사 표현이 있는지 확인
|
||||
thanks_keywords = ["감사", "고마워", "고맙", "도움"]
|
||||
has_thanks = any(kw in query for kw in thanks_keywords)
|
||||
|
||||
if has_thanks:
|
||||
answer = (
|
||||
"도움이 되셨다니 기쁩니다! 😊\n\n"
|
||||
"한국도로공사를 이용해 주셔서 감사합니다.\n"
|
||||
"궁금하신 점이 더 있으시면 언제든지 다시 찾아주세요.\n\n"
|
||||
"안전운전 하세요! 🚗"
|
||||
)
|
||||
else:
|
||||
answer = (
|
||||
"상담을 종료하시겠습니까?\n\n"
|
||||
"추가로 궁금하신 사항이 있으시면\n"
|
||||
"언제든지 질문해 주세요.\n\n"
|
||||
"한국도로공사를 이용해 주셔서 감사합니다. 😊"
|
||||
)
|
||||
|
||||
return {
|
||||
"answer": answer,
|
||||
"matched_questions": [],
|
||||
"scores": [],
|
||||
"num_references": 0,
|
||||
"botId": bot_id,
|
||||
"intent": "farewell",
|
||||
"quick_replies": self.FAREWELL_QUICK_REPLIES,
|
||||
"rerank_info": {
|
||||
"used": False,
|
||||
"detail": "Farewell intent detected"
|
||||
}
|
||||
}
|
||||
|
||||
def generate_response(
|
||||
self,
|
||||
intent_name: str,
|
||||
query: str,
|
||||
matched_keywords: List[str],
|
||||
bot_id: Optional[str] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""의도에 따른 응답 생성 (greeting, farewell만)
|
||||
|
||||
Args:
|
||||
intent_name: "greeting" 또는 "farewell"
|
||||
query: 원본 질문
|
||||
matched_keywords: 매칭된 키워드
|
||||
bot_id: 봇 ID
|
||||
|
||||
Returns:
|
||||
응답 딕셔너리
|
||||
"""
|
||||
if intent_name == "greeting":
|
||||
return self.generate_greeting_response(query, matched_keywords, bot_id)
|
||||
elif intent_name == "farewell":
|
||||
return self.generate_farewell_response(query, matched_keywords, bot_id)
|
||||
else:
|
||||
# complaint는 더 이상 여기서 처리하지 않음 (RAG 파이프라인으로)
|
||||
raise ValueError(f"Unknown intent: {intent_name}")
|
||||
@@ -0,0 +1,115 @@
|
||||
"""
|
||||
의도 감지 모듈
|
||||
──────────────
|
||||
사용자 질문의 의도를 분류 (인사, 종료, 불만, 일반 등)
|
||||
"""
|
||||
|
||||
from typing import Dict, List, Optional
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass
|
||||
class Intent:
|
||||
"""의도 분류 결과"""
|
||||
name: str # "greeting", "farewell", "general"
|
||||
confidence: float # 0.0 ~ 1.0
|
||||
matched_keywords: List[str] # 매칭된 키워드
|
||||
|
||||
def is_special(self) -> bool:
|
||||
"""특별 처리가 필요한 의도인지 (인사, 종료만)"""
|
||||
return self.name in ["greeting", "farewell"]
|
||||
|
||||
|
||||
class IntentDetector:
|
||||
"""사용자 의도 감지기"""
|
||||
|
||||
# 의도별 키워드 패턴
|
||||
INTENT_PATTERNS = {
|
||||
"greeting": {
|
||||
"keywords": [
|
||||
"안녕", "안녕하세요", "안녕하십니까",
|
||||
"처음", "반가", "반갑습니다",
|
||||
"hi", "hello", "hey",
|
||||
"처음 뵙겠습니다", "처음입니다"
|
||||
],
|
||||
"priority": 1 # 우선순위 (낮을수록 높음)
|
||||
},
|
||||
"farewell": {
|
||||
"keywords": [
|
||||
"감사", "고마워", "고맙습니다", "감사합니다",
|
||||
"잘됐", "해결", "알겠", "알았",
|
||||
"끝", "종료", "그만",
|
||||
"bye", "goodbye", "끝내", "닫기",
|
||||
"도움 됐", "도움됐", "충분"
|
||||
],
|
||||
"priority": 2
|
||||
}
|
||||
# complaint 제거: 감정 분석(EmotionDetector)으로만 처리
|
||||
}
|
||||
|
||||
def detect(self, query: str, ts: Optional[str] = None) -> Intent:
|
||||
"""의도 감지
|
||||
|
||||
Args:
|
||||
query: 사용자 질문
|
||||
ts: 타임스탬프 (로깅용)
|
||||
|
||||
Returns:
|
||||
Intent 객체
|
||||
"""
|
||||
query_lower = query.lower().strip()
|
||||
|
||||
# 빈 문자열 체크
|
||||
if not query_lower:
|
||||
return Intent(name="general", confidence=1.0, matched_keywords=[])
|
||||
|
||||
# 각 의도별 매칭 점수 계산
|
||||
intent_scores = {}
|
||||
|
||||
for intent_name, config in self.INTENT_PATTERNS.items():
|
||||
matched = []
|
||||
for keyword in config["keywords"]:
|
||||
if keyword.lower() in query_lower:
|
||||
matched.append(keyword)
|
||||
|
||||
if matched:
|
||||
# 매칭된 키워드 개수와 우선순위를 고려한 점수
|
||||
score = len(matched) * (1.0 / config["priority"])
|
||||
intent_scores[intent_name] = {
|
||||
"score": score,
|
||||
"matched": matched,
|
||||
"priority": config["priority"]
|
||||
}
|
||||
|
||||
# 매칭된 의도가 없으면 일반 질문
|
||||
if not intent_scores:
|
||||
return Intent(name="general", confidence=1.0, matched_keywords=[])
|
||||
|
||||
# 가장 높은 점수의 의도 선택
|
||||
best_intent = max(intent_scores.items(), key=lambda x: x[1]["score"])
|
||||
intent_name = best_intent[0]
|
||||
intent_data = best_intent[1]
|
||||
|
||||
# 신뢰도 계산 (0.7 ~ 0.95)
|
||||
confidence = min(0.95, 0.7 + (intent_data["score"] * 0.1))
|
||||
|
||||
result = Intent(
|
||||
name=intent_name,
|
||||
confidence=confidence,
|
||||
matched_keywords=intent_data["matched"]
|
||||
)
|
||||
|
||||
if ts:
|
||||
print(f"[IntentDetector] {ts} 의도 감지: {result.name} (신뢰도: {result.confidence:.2f}, 키워드: {result.matched_keywords})")
|
||||
|
||||
return result
|
||||
|
||||
def is_greeting(self, query: str) -> bool:
|
||||
"""인사 여부 판단 (간단한 헬퍼)"""
|
||||
intent = self.detect(query)
|
||||
return intent.name == "greeting"
|
||||
|
||||
def is_farewell(self, query: str) -> bool:
|
||||
"""종료 인사 여부 판단"""
|
||||
intent = self.detect(query)
|
||||
return intent.name == "farewell"
|
||||
@@ -0,0 +1,132 @@
|
||||
"""
|
||||
LLM 핸들러 모듈
|
||||
─────────────
|
||||
LLM 답변 생성 및 후처리
|
||||
"""
|
||||
|
||||
import re
|
||||
from typing import Optional, List, Dict
|
||||
|
||||
|
||||
class LLMHandler:
|
||||
"""LLM 답변 생성 핸들러"""
|
||||
|
||||
def __init__(self, llm_client, config):
|
||||
self.llm_client = llm_client
|
||||
self.config = config
|
||||
|
||||
def generate_answer(
|
||||
self,
|
||||
system_prompt: str,
|
||||
user_prompt: str,
|
||||
ts: str,
|
||||
max_tokens: Optional[int] = None,
|
||||
temperature: float = 0.3
|
||||
) -> str:
|
||||
"""LLM 답변 생성 (기존 방식 - 하위 호환성)
|
||||
|
||||
Args:
|
||||
system_prompt: 시스템 프롬프트
|
||||
user_prompt: 사용자 프롬프트
|
||||
ts: 타임스탬프 (로깅용)
|
||||
max_tokens: 최대 토큰 수 (기본값: config에서 가져옴)
|
||||
temperature: 온도 파라미터
|
||||
|
||||
Returns:
|
||||
생성된 답변 텍스트
|
||||
"""
|
||||
if max_tokens is None:
|
||||
max_tokens = self.config.llm_max_tokens
|
||||
|
||||
try:
|
||||
print(f"[LLMHandler] {ts} LLM 답변 생성 중... (max_tokens={max_tokens})")
|
||||
|
||||
response = self.llm_client.chat_completion(
|
||||
messages=[
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": user_prompt}
|
||||
],
|
||||
max_tokens=max_tokens,
|
||||
temperature=temperature
|
||||
)
|
||||
|
||||
# <think> 태그 제거
|
||||
formatted = self._remove_think_tags(response)
|
||||
|
||||
print(f"[LLMHandler] {ts} LLM 답변 생성 완료 (길이: {len(formatted)}자)")
|
||||
return formatted
|
||||
|
||||
except Exception as e:
|
||||
print(f"[LLMHandler] {ts} ❌ LLM 답변 생성 실패: {e}")
|
||||
raise
|
||||
|
||||
def generate_answer_from_messages(
|
||||
self,
|
||||
messages: List[Dict[str, str]],
|
||||
ts: str,
|
||||
max_tokens: Optional[int] = None,
|
||||
temperature: float = 0.3
|
||||
) -> str:
|
||||
"""LLM 답변 생성 (Messages Format - 표준 방식)
|
||||
|
||||
Args:
|
||||
messages: [{"role": "system|user|assistant", "content": "..."}]
|
||||
ts: 타임스탬프 (로깅용)
|
||||
max_tokens: 최대 토큰 수
|
||||
temperature: 온도 파라미터
|
||||
|
||||
Returns:
|
||||
생성된 답변 텍스트
|
||||
"""
|
||||
if max_tokens is None:
|
||||
max_tokens = self.config.llm_max_tokens
|
||||
|
||||
try:
|
||||
print(f"[LLMHandler] {ts} LLM 답변 생성 중... (messages={len(messages)}개, max_tokens={max_tokens})")
|
||||
|
||||
response = self.llm_client.chat_completion(
|
||||
messages=messages,
|
||||
max_tokens=max_tokens,
|
||||
temperature=temperature
|
||||
)
|
||||
|
||||
# <think> 태그 제거
|
||||
formatted = self._remove_think_tags(response)
|
||||
|
||||
print(f"[LLMHandler] {ts} LLM 답변 생성 완료 (길이: {len(formatted)}자)")
|
||||
return formatted
|
||||
|
||||
except Exception as e:
|
||||
print(f"[LLMHandler] {ts} ❌ LLM 답변 생성 실패: {e}")
|
||||
raise
|
||||
|
||||
def generate_fallback_answer(self, best_match: dict) -> str:
|
||||
"""폴백 답변 생성 (LLM 실패 시)
|
||||
|
||||
Args:
|
||||
best_match: 최상위 매칭 결과
|
||||
|
||||
Returns:
|
||||
폴백 답변
|
||||
"""
|
||||
return (
|
||||
"고객님의 질문은 다음과 같다고 생각됩니다.\n\n"
|
||||
f"{best_match['q']}\n\n"
|
||||
"이에 대한 답변을 드리겠습니다.\n\n"
|
||||
f"{best_match['a']}"
|
||||
)
|
||||
|
||||
def generate_default_guidance(self) -> str:
|
||||
"""기본 질문 유도 답변 (LLM 실패 시)"""
|
||||
return (
|
||||
"죄송합니다. 고객님의 질문과 관련된 정보를 찾을 수 없습니다.\n\n"
|
||||
"한국도로공사는 고속도로 이용과 관련된 상담을 제공하고 있습니다.\n"
|
||||
"통행료, Hi-pass, 환불, 소음 민원, 시설물 이용 등에 대해 궁금하신 사항이 있으시다면 질문해 주세요.\n\n"
|
||||
"또는 한국도로공사 콜센터(1588-2504)로 문의하시면 자세한 안내를 받으실 수 있습니다."
|
||||
)
|
||||
|
||||
def _remove_think_tags(self, text: str) -> str:
|
||||
"""<think> 태그 제거"""
|
||||
if "<think>" in text and "</think>" in text:
|
||||
return re.sub(r'<think>.*?</think>\s*', '', text, flags=re.DOTALL).strip()
|
||||
return text
|
||||
@@ -0,0 +1,399 @@
|
||||
"""
|
||||
프롬프트 빌더 모듈
|
||||
────────────────
|
||||
LLM 프롬프트 생성 로직
|
||||
"""
|
||||
|
||||
from typing import List, Dict, Any, Optional
|
||||
|
||||
|
||||
class PromptBuilder:
|
||||
"""LLM 프롬프트 생성기"""
|
||||
|
||||
# 시스템 프롬프트 (일반 답변)
|
||||
SYSTEM_PROMPT = (
|
||||
"당신은 한국도로공사 채팅상담 챗봇입니다.\n"
|
||||
"한국도로공사의 고객 문의에 답변하는 전문 상담원으로서 행동하세요.\n"
|
||||
"\n"
|
||||
"【대화 이력 활용】\n"
|
||||
"1. 이전 대화는 현재 질문에 지시어가 있거나 주제가 명확히 이어질 때만 참고하세요.\n"
|
||||
"2. '그럼', '그거', '그건', '아까', '방금', '그때' 등의 지시어가 있다면:\n"
|
||||
" - 이전 대화에서 언급된 주제를 파악하세요.\n"
|
||||
" - 해당 주제와 현재 질문을 연결하여 답변하세요.\n"
|
||||
"3. 이전 대화와 현재 질문이 같은 주제라면 자연스럽게 이어서 답변하세요.\n"
|
||||
"4. 현재 질문이 인사, 감사, 종료, 새 주제라면 이전 대화와 억지로 연결하지 말고 현재 질문만 기준으로 답변하세요.\n"
|
||||
"5. 이전 대화가 없거나 무관하다면 현재 질문만 기준으로 답변하세요.\n"
|
||||
"\n"
|
||||
"【답변 작성 원칙】\n"
|
||||
"1. 제공된 참고자료 중 고객의 상황에 일반적으로 적용 가능한 내용만 사용하세요.\n"
|
||||
"2. 특정 개인, 특정 지역, 특수한 상황에만 해당하는 내용은 제외하세요.\n"
|
||||
"3. 일반적인 정책, 절차, 규정에 관한 내용을 우선적으로 활용하세요.\n"
|
||||
"4. 참고자료가 고객의 질문과 직접적으로 관련이 없거나 특수 사례만 있다면,\n"
|
||||
" '관련 정보를 정확히 안내드리기 어렵습니다. 한국도로공사 콜센터(1588-2504)로 문의해 주시면 자세히 안내드리겠습니다.'라고 답변하세요.\n"
|
||||
"5. 참고자료에 없는 내용은 추측하지 마세요.\n"
|
||||
"\n"
|
||||
"【특수 케이스 식별 기준 - 다음 내용이 포함된 참고자료는 제외】\n"
|
||||
"❌ 특정 인명, 차량번호, 계좌번호, 주민등록번호 등 개인정보\n"
|
||||
"❌ '00아파트', '00지역 주민만', '특정 구간 한정' 등 특정 지역 한정\n"
|
||||
"❌ '2023년 특별 이벤트', '한시적 조치', '임시 운영' 등 기간 한정\n"
|
||||
"❌ '귀하의 경우', '고객님만', '해당 건에 한해' 등 개별 맞춤 답변\n"
|
||||
"❌ '예외적으로', '특별히', '이번 건에 한해서만' 등 특수 조건\n"
|
||||
"❌ 과거 민원 처리 결과나 개별 사례의 구체적 내용\n"
|
||||
"\n"
|
||||
"【용어 및 형식】\n"
|
||||
"- 답변 시 'KEC' 대신 '한국도로공사'라는 명칭을 사용하세요.\n"
|
||||
"- 답변은 한국어로 작성하고, 존댓말을 사용하세요.\n"
|
||||
"- 간결하고 명확하게 답변하세요.\n"
|
||||
"- 카카오톡 챗봇 응답이므로 3~4문장 이내로 답변하세요.\n"
|
||||
"- 인사말과 마무리 인사는 생략하세요.\n"
|
||||
"- 최종 답변에서 '참고자료', '제공된 자료', '주어진 자료', '자료를 확인한 결과' 같은 내부 근거 표현을 사용하지 마세요.\n"
|
||||
"- 개인정보나 특수 사례가 포함된 참고자료는 절대 언급하지 마세요."
|
||||
)
|
||||
|
||||
DOMAIN_DATA_PRIORITY_PROMPT = (
|
||||
"\n\n"
|
||||
"【DB 조회 결과 우선 원칙】\n"
|
||||
"1. 【DB 조회 결과】가 제공된 경우, 이 데이터는 시스템이 실시간으로 조회한 정확한 결과입니다.\n"
|
||||
"2. DB 조회 결과가 고객 질문에 답할 수 있으면, 참고자료 유무와 관계없이 DB 조회 결과를 최우선으로 사용하세요.\n"
|
||||
"3. 참고자료는 DB 조회 결과를 보완하는 용도로만 사용하고, DB 조회 결과와 충돌하면 DB 조회 결과를 따르세요.\n"
|
||||
"4. DB 조회 결과에 없는 세부 내용은 추측하지 말고, 제공된 값만 자연스럽게 설명하세요."
|
||||
)
|
||||
|
||||
def _format_reference(self, index: int, ref: Dict[str, Any], score: Any = None) -> str:
|
||||
"""참고자료를 LLM 프롬프트용 텍스트로 변환"""
|
||||
source = ref.get("source") or "unknown"
|
||||
category = ref.get("category")
|
||||
url = ref.get("url")
|
||||
|
||||
lines = [
|
||||
f"[참고자료 {index}]",
|
||||
f"출처: {source}",
|
||||
]
|
||||
if category:
|
||||
lines.append(f"분류: {category}")
|
||||
if score is not None:
|
||||
lines.append(f"점수: {score}")
|
||||
lines.extend([
|
||||
f"질문: {ref['q']}",
|
||||
f"답변: {ref['a']}",
|
||||
])
|
||||
if url:
|
||||
lines.append(f"공식 FAQ URL: {url}")
|
||||
lines.append("주의: 이 URL은 고객에게 상세 원문 확인 링크로 제공할 수 있습니다.")
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
# 시스템 프롬프트 (질문 유도)
|
||||
GUIDANCE_SYSTEM_PROMPT = (
|
||||
"당신은 한국도로공사 채팅상담 챗봇입니다.\n"
|
||||
"한국도로공사 관련 문의를 돕는 챗봇이라는 역할 안에서만 답변하세요.\n"
|
||||
"\n"
|
||||
"【상황】\n"
|
||||
"고객 질문에 대해 검색 가능한 DB/벡터 자료에서 정확한 근거를 찾지 못한 상황입니다.\n"
|
||||
"\n"
|
||||
"【중요: 절대 금지 사항】\n"
|
||||
"❌ 한국도로공사 업무 정보, 제도, 요금, 정책, 운영 현황을 임의로 만들어내지 마세요.\n"
|
||||
"❌ 추측하거나 가정하여 구체적인 정보를 답변하지 마세요.\n"
|
||||
"❌ 새로운 질문을 추천하거나 다른 질문을 유도하지 마세요.\n"
|
||||
"❌ '참고자료', '제공된 자료', '주어진 자료' 같은 내부 표현을 사용하지 마세요.\n"
|
||||
"\n"
|
||||
"【답변 원칙】\n"
|
||||
"1. '넌 누구야?', '자기소개해줘', '뭐 하는 챗봇이야?' 같은 챗봇 정체성/가벼운 질문에는 짧게 답변하세요.\n"
|
||||
"2. 업무 정보에 대한 질문인데 정확한 근거가 없으면, 현재 확인 가능한 정보가 없다고 안내하고 한국도로공사 콜센터(1588-2504)로 문의하도록 안내하세요.\n"
|
||||
"3. 이전 대화는 현재 질문에 지시어가 있거나 주제가 명확히 이어질 때만 참고하세요.\n"
|
||||
"4. 현재 질문이 인사, 감사, 종료, 새 주제이거나 이전 대화와 무관한 경우:\n"
|
||||
" - 이전 대화와 억지로 연결하지 말고 현재 질문만 기준으로 답변하세요.\n"
|
||||
"\n"
|
||||
"【용어 및 형식】\n"
|
||||
"- 정중하고 친절한 톤을 유지하세요.\n"
|
||||
"- 답변은 한국어로 작성하고, 존댓말을 사용하세요.\n"
|
||||
"- 간결하게 2~4문장으로 답변하세요.\n"
|
||||
"- 인사말과 마무리 인사는 생략하세요.\n"
|
||||
"- 콜센터 안내 시: 한국도로공사 콜센터(1588-2504)"
|
||||
)
|
||||
|
||||
def build_answer_prompt_messages(
|
||||
self,
|
||||
original_query: str,
|
||||
rewritten_query: Optional[str],
|
||||
references: List[Dict[str, Any]],
|
||||
scores: List[float],
|
||||
conversation_history: List[Dict[str, str]] = None,
|
||||
emotion_instruction: Optional[str] = None,
|
||||
emotion_name: Optional[str] = None,
|
||||
domain_data: Optional[Dict[str, Any]] = None
|
||||
) -> List[Dict[str, str]]:
|
||||
"""일반 답변 프롬프트 생성 (Messages Format)
|
||||
|
||||
Args:
|
||||
original_query: 원본 질문
|
||||
rewritten_query: 재작성된 질문 (없으면 None)
|
||||
references: 참고자료 리스트
|
||||
scores: 참고자료 점수
|
||||
conversation_history: 대화 이력 (messages format)
|
||||
emotion_instruction: 감정별 추가 지시사항
|
||||
emotion_name: 감정 이름 (angry, frustrated, worried 등)
|
||||
domain_data: chatbotApi 도메인 서비스 DB 조회 결과 (있으면 프롬프트에 포함)
|
||||
|
||||
Returns:
|
||||
messages 리스트
|
||||
"""
|
||||
has_domain_data = self._has_usable_domain_data(domain_data)
|
||||
|
||||
# 시스템 프롬프트에 domainData/감정 지시사항 추가
|
||||
system_prompt = self.SYSTEM_PROMPT
|
||||
if has_domain_data:
|
||||
system_prompt = system_prompt + self.DOMAIN_DATA_PRIORITY_PROMPT
|
||||
if emotion_instruction:
|
||||
system_prompt = system_prompt + emotion_instruction
|
||||
|
||||
messages = [
|
||||
{"role": "system", "content": system_prompt}
|
||||
]
|
||||
|
||||
# 대화 이력 추가 (messages format)
|
||||
if conversation_history:
|
||||
messages.extend(conversation_history)
|
||||
|
||||
# 현재 질문 구성
|
||||
current_query_parts = []
|
||||
|
||||
# 감정 정보 추가 (부정적 감정인 경우)
|
||||
if emotion_name and emotion_name in ["angry", "frustrated", "worried"]:
|
||||
emotion_labels = {
|
||||
"angry": "화난/불만",
|
||||
"frustrated": "답답한/짜증난",
|
||||
"worried": "걱정되는/불안한"
|
||||
}
|
||||
current_query_parts.extend([
|
||||
f"【⚠️ 고객 감정 상태: {emotion_labels.get(emotion_name, emotion_name)}】",
|
||||
"고객이 부정적인 감정을 느끼고 있습니다.",
|
||||
"이전 대화나 현재 상황에서 무엇이 불편했는지 파악하고,",
|
||||
"그 점에 대해 구체적으로 공감한 후 실질적인 해결책을 제시하세요.",
|
||||
""
|
||||
])
|
||||
|
||||
if rewritten_query:
|
||||
# Query Rewriting이 적용된 경우
|
||||
current_query_parts.extend([
|
||||
"【원본 질문】",
|
||||
original_query,
|
||||
"",
|
||||
"【맥락 기반 재작성 질문】",
|
||||
rewritten_query,
|
||||
"(이전 대화를 참고하여 재작성된 질문입니다)",
|
||||
""
|
||||
])
|
||||
else:
|
||||
# Query Rewriting이 없는 경우 원본 질문 명시
|
||||
current_query_parts.extend([
|
||||
"【고객 질문】",
|
||||
original_query,
|
||||
""
|
||||
])
|
||||
|
||||
# DB 조회 결과 추가 (domain_data가 있을 때만, 참고자료보다 앞에 위치)
|
||||
if has_domain_data:
|
||||
summary = domain_data.get("llmSummary")
|
||||
if summary is not None and str(summary).strip():
|
||||
current_query_parts.extend([
|
||||
"【DB 조회 결과】",
|
||||
"(아래는 시스템 요약(llmSummary)입니다. 참고자료보다 우선하여 활용하세요.)",
|
||||
str(summary).strip(),
|
||||
""
|
||||
])
|
||||
else:
|
||||
db_parts = []
|
||||
for key, value in domain_data.items():
|
||||
if key not in ("status", "statusMsg", "errorMsg") and value is not None:
|
||||
db_parts.append(f"- {key}: {value}")
|
||||
if db_parts:
|
||||
current_query_parts.extend([
|
||||
"【DB 조회 결과】",
|
||||
"(아래는 시스템에서 조회한 정확한 데이터입니다. 참고자료보다 우선하여 활용하세요.)",
|
||||
"\n".join(db_parts),
|
||||
""
|
||||
])
|
||||
|
||||
# 참고자료 추가
|
||||
context_parts = []
|
||||
for i, (ref, score) in enumerate(zip(references, scores), 1):
|
||||
context_parts.append(self._format_reference(i, ref, score))
|
||||
|
||||
if context_parts:
|
||||
current_query_parts.extend([
|
||||
"【참고자료】",
|
||||
"\n".join(context_parts),
|
||||
"",
|
||||
"【답변 지시사항】"
|
||||
])
|
||||
else:
|
||||
current_query_parts.extend([
|
||||
"【참고자료】",
|
||||
"제공된 참고자료가 없습니다.",
|
||||
"",
|
||||
"【답변 지시사항】"
|
||||
])
|
||||
if emotion_name in ["angry", "frustrated", "worried"]:
|
||||
current_query_parts.append(
|
||||
f"⚠️ 고객이 {emotion_labels.get(emotion_name, emotion_name)} 상태입니다. "
|
||||
"답변 시작 부분에 무엇이 불편했는지 구체적으로 언급하며 공감하고, "
|
||||
"그 후 명확한 해결 방법을 제시하세요."
|
||||
)
|
||||
if rewritten_query:
|
||||
current_query_parts.append("- 원본 질문은 간단하지만, 재작성된 질문의 의도를 파악하여 답변하세요.")
|
||||
current_query_parts.append("- 답변 시에는 사용자가 실제로 물어본 질문에 대해 자연스럽게 답변하세요.")
|
||||
if conversation_history:
|
||||
current_query_parts.append("- 이전 대화는 현재 질문에 지시어가 있거나 주제가 명확히 이어질 때만 참고하세요.")
|
||||
current_query_parts.append("- 현재 질문이 인사, 감사, 종료, 새 주제라면 이전 대화와 억지로 연결하지 말고 현재 질문만 기준으로 답변하세요.")
|
||||
|
||||
if has_domain_data:
|
||||
current_query_parts.append("- 【DB 조회 결과】를 최우선 근거로 고객의 질문에 답변해주세요.")
|
||||
current_query_parts.append("- 참고자료가 없거나 질문과 무관해도, DB 조회 결과가 질문에 답할 수 있으면 콜센터 안내로 대체하지 마세요.")
|
||||
current_query_parts.append("- 참고자료는 DB 조회 결과를 보완할 때만 사용하고, DB 조회 결과와 충돌하면 DB 조회 결과를 따르세요.")
|
||||
if domain_data and domain_data.get("msgMap"):
|
||||
current_query_parts.append("- 휴게소 메뉴 목록은 전체를 모두 나열하지 말고, 대표 메뉴 3~5개만 짧게 언급한 뒤 상세 목록은 시스템 응답의 메뉴 목록을 확인하도록 안내하세요.")
|
||||
current_query_parts.append("- 메뉴가 많거나 방향/매장이 여러 개인 경우에도 답변 본문은 5문장 이내로 간결하게 작성하세요.")
|
||||
current_query_parts.append("- 공식 FAQ URL이 있는 참고자료가 관련 있다면, 답변 본문에는 간단히 안내하고 시스템 응답의 링크 버튼으로 상세 확인을 유도할 수 있습니다.")
|
||||
else:
|
||||
current_query_parts.append("- 위 참고자료를 바탕으로 고객의 질문에 답변해주세요.")
|
||||
current_query_parts.append("- 참고자료가 질문과 무관하거나 특수 사례만 있다면 콜센터로 안내하세요.")
|
||||
current_query_parts.append("- 공식 FAQ URL이 있는 참고자료가 관련 있다면, 답변 본문에는 간단히 안내하고 시스템 응답의 링크 버튼으로 상세 확인을 유도할 수 있습니다.")
|
||||
current_query_parts.append("- 카카오톡 챗봇 응답이므로 3~4문장 이내로 답변하세요.")
|
||||
current_query_parts.append("- 인사말과 마무리 인사는 생략하세요.")
|
||||
current_query_parts.append("- 최종 답변에서 '참고자료', '제공된 자료', '주어진 자료', '자료를 확인한 결과' 같은 내부 근거 표현을 사용하지 마세요.")
|
||||
|
||||
# 현재 질문 추가
|
||||
messages.append({
|
||||
"role": "user",
|
||||
"content": "\n".join(current_query_parts)
|
||||
})
|
||||
|
||||
return messages
|
||||
|
||||
def _has_usable_domain_data(self, domain_data: Optional[Dict[str, Any]]) -> bool:
|
||||
if not domain_data:
|
||||
return False
|
||||
status = domain_data.get("status")
|
||||
if status is False:
|
||||
return False
|
||||
if isinstance(status, str) and status.lower() == "false":
|
||||
return False
|
||||
return True
|
||||
|
||||
def build_answer_prompt(
|
||||
self,
|
||||
original_query: str,
|
||||
rewritten_query: Optional[str],
|
||||
references: List[Dict[str, Any]],
|
||||
scores: List[float],
|
||||
conversation_context: Optional[str] = None,
|
||||
emotion_instruction: Optional[str] = None
|
||||
) -> tuple[str, str]:
|
||||
"""일반 답변 프롬프트 생성
|
||||
|
||||
Args:
|
||||
original_query: 원본 질문
|
||||
rewritten_query: 재작성된 질문 (없으면 None)
|
||||
references: 참고자료 리스트
|
||||
scores: 참고자료 점수
|
||||
conversation_context: 대화 이력 텍스트
|
||||
emotion_instruction: 감정별 추가 지시사항
|
||||
|
||||
Returns:
|
||||
(system_prompt, user_prompt)
|
||||
"""
|
||||
# 시스템 프롬프트에 감정 지시사항 추가
|
||||
system_prompt = self.SYSTEM_PROMPT
|
||||
if emotion_instruction:
|
||||
system_prompt = system_prompt + emotion_instruction
|
||||
|
||||
user_prompt_parts = []
|
||||
|
||||
# 이전 대화 이력
|
||||
if conversation_context:
|
||||
user_prompt_parts.append("【이전 대화 이력】")
|
||||
user_prompt_parts.append("(현재 질문과 명확히 이어지는 경우에만 참고하세요)")
|
||||
user_prompt_parts.append(conversation_context)
|
||||
user_prompt_parts.append("")
|
||||
|
||||
# 질문 (Query Rewriting 적용 여부에 따라)
|
||||
if rewritten_query:
|
||||
user_prompt_parts.extend([
|
||||
"【사용자의 현재 질문】",
|
||||
original_query,
|
||||
"",
|
||||
"【맥락을 고려한 완전한 질문】",
|
||||
rewritten_query,
|
||||
"(이전 대화를 참고하여 재작성된 질문입니다. 이 질문에 대해 답변해주세요.)",
|
||||
"",
|
||||
])
|
||||
else:
|
||||
user_prompt_parts.extend([
|
||||
"【현재 질문】",
|
||||
f"고객 질문: {original_query}",
|
||||
"",
|
||||
])
|
||||
|
||||
# 참고자료
|
||||
context_parts = []
|
||||
for i, (ref, score) in enumerate(zip(references, scores), 1):
|
||||
context_parts.append(self._format_reference(i, ref, score))
|
||||
|
||||
user_prompt_parts.extend([
|
||||
"【참고자료】",
|
||||
"\n".join(context_parts),
|
||||
"",
|
||||
"【지시사항】"
|
||||
])
|
||||
|
||||
# 추가 지시사항
|
||||
if rewritten_query:
|
||||
user_prompt_parts.append("- 사용자의 현재 질문은 간단하지만, 재작성된 완전한 질문의 의도를 파악하여 답변하세요.")
|
||||
user_prompt_parts.append("- 답변 시에는 사용자가 실제로 물어본 현재 질문에 대해 자연스럽게 답변하세요.")
|
||||
if conversation_context:
|
||||
user_prompt_parts.append("- 이전 대화는 현재 질문에 지시어가 있거나 주제가 명확히 이어질 때만 참고하세요.")
|
||||
user_prompt_parts.append("- 현재 질문이 인사, 감사, 종료, 새 주제라면 이전 대화와 억지로 연결하지 말고 현재 질문만 기준으로 답변하세요.")
|
||||
|
||||
user_prompt_parts.append("- 위 참고자료를 바탕으로 고객의 질문에 답변해주세요.")
|
||||
user_prompt_parts.append("- 참고자료가 질문과 무관하거나 특수 사례만 있다면 콜센터로 안내하세요.")
|
||||
user_prompt_parts.append("- 공식 FAQ URL이 있는 참고자료가 관련 있다면, 답변 본문에는 간단히 안내하고 시스템 응답의 링크 버튼으로 상세 확인을 유도할 수 있습니다.")
|
||||
user_prompt_parts.append("- 최종 답변에서 '참고자료', '제공된 자료', '주어진 자료', '자료를 확인한 결과' 같은 내부 근거 표현을 사용하지 마세요.")
|
||||
|
||||
return system_prompt, "\n".join(user_prompt_parts)
|
||||
|
||||
def build_guidance_prompt(
|
||||
self,
|
||||
original_query: str,
|
||||
conversation_context: Optional[str] = None
|
||||
) -> tuple[str, str]:
|
||||
"""질문 유도 프롬프트 생성
|
||||
|
||||
Args:
|
||||
original_query: 원본 질문
|
||||
conversation_context: 대화 이력 텍스트
|
||||
|
||||
Returns:
|
||||
(system_prompt, user_prompt)
|
||||
"""
|
||||
user_prompt_parts = []
|
||||
|
||||
# 이전 대화 이력
|
||||
if conversation_context:
|
||||
user_prompt_parts.append(conversation_context)
|
||||
user_prompt_parts.append("")
|
||||
|
||||
# 현재 상황
|
||||
user_prompt_parts.extend([
|
||||
"【현재 상황】",
|
||||
f"고객 질문: {original_query}",
|
||||
"",
|
||||
"위 질문에 대해 검색 가능한 DB/벡터 자료에서 정확한 근거를 찾지 못했습니다.",
|
||||
"",
|
||||
"【지시사항】",
|
||||
"- 챗봇 정체성이나 가벼운 질문이면 한국도로공사 채팅상담 챗봇이라는 역할 안에서 짧게 답변하세요.",
|
||||
"- 업무 정보에 대한 질문이면 현재 정확한 정보를 확인하기 어렵다고 안내하고 콜센터로 연결하세요.",
|
||||
"- 임의로 새로운 정보를 만들어내지 마세요.",
|
||||
"- 새로운 질문을 추천하거나 다른 질문을 유도하지 마세요."
|
||||
])
|
||||
|
||||
return self.GUIDANCE_SYSTEM_PROMPT, "\n".join(user_prompt_parts)
|
||||
@@ -0,0 +1,245 @@
|
||||
"""
|
||||
Query Rewriting 모듈
|
||||
──────────────────
|
||||
대화 이력 기반 질문 재작성
|
||||
"""
|
||||
|
||||
import re
|
||||
from typing import Optional, List, Dict, Any
|
||||
|
||||
from handlers.suggestion_handler import strip_suggestion_block
|
||||
|
||||
|
||||
class QueryRewriter:
|
||||
"""대화 이력을 활용한 질문 재작성"""
|
||||
|
||||
NO_REWRITE_TOKEN = "__NO_REWRITE__"
|
||||
|
||||
CONTEXT_CUE_KEYWORDS = [
|
||||
"그럼", "그거", "그건", "그게", "그걸", "그곳", "거기",
|
||||
"아까", "방금", "이어서", "계속", "위 내용", "위에",
|
||||
"앞에서", "앞서", "이전", "그 요금", "그 휴게소", "해당"
|
||||
]
|
||||
FOLLOW_UP_PATTERNS = [
|
||||
r".*(얼마|몇\s*원|가격|비용|요금).*",
|
||||
r".*(어디|어디서|위치|장소).*",
|
||||
r".*(어떻게|방법|절차|신청|구매|사\?|사요|사나요|살|등록|해지|취소).*",
|
||||
r".*(가능|돼|되나|되나요|필요|있어|없어).*",
|
||||
r".*(언제|몇\s*시|시간|기간).*",
|
||||
r".*(왜|이유).*"
|
||||
]
|
||||
|
||||
def __init__(self, llm_client, embed_client):
|
||||
self.llm_client = llm_client
|
||||
self.embed_client = embed_client
|
||||
|
||||
self.system_prompt = (
|
||||
"당신은 질문 재작성 전문가입니다.\n"
|
||||
"현재 질문이 이전 대화와 자연스럽게 이어지는 후속 질문인지 먼저 판단하세요.\n"
|
||||
"이어지는 후속 질문이면 이전 대화를 참고하여 완전한 질문으로 재작성하세요.\n"
|
||||
f"이어지지 않는 새 질문이면 정확히 {self.NO_REWRITE_TOKEN}만 출력하세요.\n"
|
||||
"\n"
|
||||
"【재작성 원칙】\n"
|
||||
"1. '그럼', '그거', '그건', '아까', '이거' 등의 지시어를 구체적인 명사로 교체하세요.\n"
|
||||
"2. '얼마야?', '어디서 사?', '어떻게 해?'처럼 짧은 후속 질문은 이전 대화 주제와 연결하세요.\n"
|
||||
f"3. 현재 질문이 인사, 감사, 종료, 새 주제이면 이전 대화와 억지로 연결하지 말고 {self.NO_REWRITE_TOKEN}만 출력하세요.\n"
|
||||
"4. 후속 질문으로 판단한 경우에만 질문의 의도를 유지하면서 완전한 문장으로 만드세요.\n"
|
||||
f"5. 출력은 재작성된 질문 한 문장 또는 {self.NO_REWRITE_TOKEN}만 허용됩니다.\n"
|
||||
"\n"
|
||||
"【예시】\n"
|
||||
"이전 대화: '하이패스 단말기가 뭔가요?'\n"
|
||||
"현재 질문: '그럼 어디서 사나요?'\n"
|
||||
"재작성: '하이패스 단말기는 어디서 구매할 수 있나요?'\n"
|
||||
"이전 대화: '하이패스 단말기에 대해 알려줘'\n"
|
||||
"현재 질문: '얼마야?'\n"
|
||||
"재작성: '하이패스 단말기 가격은 얼마인가요?'\n"
|
||||
"이전 대화: '통행요금 조회해줘'\n"
|
||||
"현재 질문: '안녕'\n"
|
||||
f"재작성: {self.NO_REWRITE_TOKEN}\n"
|
||||
"이전 대화: '하이패스 단말기에 대해 알려줘'\n"
|
||||
"현재 질문: '동김천 휴게소 메뉴 알려줘'\n"
|
||||
f"재작성: {self.NO_REWRITE_TOKEN}\n"
|
||||
)
|
||||
|
||||
def format_history(self, history: List[Dict[str, Any]]) -> str:
|
||||
"""대화 이력을 텍스트로 포맷팅 (질문 + 답변 전문, 추천 블록 제외)"""
|
||||
lines = []
|
||||
for h in history:
|
||||
body = strip_suggestion_block(h.get("ai_response"))
|
||||
lines.append(f"고객: {h['user_query']}\n상담원: {body}")
|
||||
return "\n".join(lines)
|
||||
|
||||
def has_context_cue(self, query: str) -> bool:
|
||||
"""현재 질문이 이전 대화 참조가 필요한 후속 질문인지 판단"""
|
||||
text = (query or "").strip()
|
||||
normalized = text.replace(" ", "").lower()
|
||||
if any(keyword.replace(" ", "").lower() in normalized for keyword in self.CONTEXT_CUE_KEYWORDS):
|
||||
return True
|
||||
|
||||
compact = re.sub(r"\s+", "", text)
|
||||
is_short_question = len(compact) <= 20
|
||||
if not is_short_question:
|
||||
return False
|
||||
return any(re.match(pattern, text) for pattern in self.FOLLOW_UP_PATTERNS)
|
||||
|
||||
def _postprocess_rewrite(self, rewritten_query: str, original_query: str, ts: str, label: str) -> Optional[str]:
|
||||
rewritten_query = self._remove_think_tags(rewritten_query).strip()
|
||||
if not rewritten_query:
|
||||
return None
|
||||
if self.NO_REWRITE_TOKEN in rewritten_query:
|
||||
print(f"[QueryRewriter] {ts} ⏭️ {label} 생략: 이전 대화와 이어지지 않음")
|
||||
return None
|
||||
if rewritten_query == original_query:
|
||||
print(f"[QueryRewriter] {ts} ⏭️ {label} 생략: 재작성 결과가 원문과 동일")
|
||||
return None
|
||||
return rewritten_query
|
||||
|
||||
def rewrite_query_with_full_context(
|
||||
self,
|
||||
original_query: str,
|
||||
history: List[Dict[str, Any]],
|
||||
ts: str
|
||||
) -> Optional[str]:
|
||||
"""질문 재작성 (이전 질문 + 답변 + 현재 질문 모두 활용)
|
||||
|
||||
Args:
|
||||
original_query: 원본 질문
|
||||
history: 대화 이력 (질문 + 답변)
|
||||
ts: 타임스탬프 (로깅용)
|
||||
|
||||
Returns:
|
||||
재작성된 질문 또는 None (실패 시)
|
||||
"""
|
||||
if not history:
|
||||
return None
|
||||
|
||||
try:
|
||||
print(f"[QueryRewriter] {ts} 🔄 Full Context Rewriting 시도 (이력 {len(history)}개)")
|
||||
|
||||
# 대화 이력 포맷팅 (질문 + 답변)
|
||||
history_text = self.format_history(history)
|
||||
|
||||
# 확장된 프롬프트 (답변 포함한 맥락 활용)
|
||||
cue_hint = "있음" if self.has_context_cue(original_query) else "없음"
|
||||
enhanced_system_prompt = (
|
||||
"당신은 질문 재작성 전문가입니다.\n"
|
||||
"현재 질문이 이전 대화와 자연스럽게 이어지는 후속 질문인지 먼저 판단하세요.\n"
|
||||
"이어지는 후속 질문이면 이전 대화 내용(질문과 답변 모두)을 참고하여 완전한 질문으로 재작성하세요.\n"
|
||||
f"이어지지 않는 새 질문이면 정확히 {self.NO_REWRITE_TOKEN}만 출력하세요.\n"
|
||||
"\n"
|
||||
"【재작성 원칙】\n"
|
||||
"1. 후속 질문으로 판단한 경우에만 이전 답변에서 설명된 개념이나 용어를 활용하세요.\n"
|
||||
"2. '그럼', '그거', '그건', '아까', '이거' 등의 지시어를 구체적인 명사로 교체하세요.\n"
|
||||
"3. '얼마야?', '어디서 사?', '어떻게 해?'처럼 짧은 후속 질문은 이전 대화 주제와 연결하세요.\n"
|
||||
f"4. 현재 질문이 인사, 감사, 종료, 새 주제이면 이전 대화와 억지로 연결하지 말고 {self.NO_REWRITE_TOKEN}만 출력하세요.\n"
|
||||
"5. 후속 질문으로 판단한 경우에만 질문의 의도를 유지하면서 완전한 문장으로 만드세요.\n"
|
||||
f"6. 출력은 재작성된 질문 한 문장 또는 {self.NO_REWRITE_TOKEN}만 허용됩니다.\n"
|
||||
"\n"
|
||||
"【예시】\n"
|
||||
"이전 대화:\n"
|
||||
"고객: '하이패스가 뭐야?'\n"
|
||||
"상담원: '하이패스는 전자식 통행료 결제 시스템입니다. 단말기를 차량에 부착하면...'\n"
|
||||
"\n"
|
||||
"현재 질문: '그럼 어디서 사?'\n"
|
||||
"재작성: '하이패스 단말기는 어디서 구매할 수 있나요?'\n"
|
||||
"이전 대화:\n"
|
||||
"고객: '하이패스 단말기에 대해 알려줘'\n"
|
||||
"상담원: '하이패스 단말기는 차량에 부착해 통행료를 자동 결제하는 장치입니다...'\n"
|
||||
"\n"
|
||||
"현재 질문: '얼마야?'\n"
|
||||
"재작성: '하이패스 단말기 가격은 얼마인가요?'\n"
|
||||
"이전 대화:\n"
|
||||
"고객: '통행요금 조회해줘'\n"
|
||||
"상담원: '출발 IC와 도착 IC를 알려주세요...'\n"
|
||||
"\n"
|
||||
"현재 질문: '안녕'\n"
|
||||
f"재작성: {self.NO_REWRITE_TOKEN}\n"
|
||||
)
|
||||
|
||||
user_prompt = (
|
||||
f"【이전 대화】\n{history_text}\n\n"
|
||||
f"【현재 질문】\n{original_query}\n\n"
|
||||
f"【후속 질문 힌트】\n패턴 기반 후속 질문 후보: {cue_hint}\n\n"
|
||||
f"이전 대화와 자연스럽게 이어지는 후속 질문이면 완전한 질문으로 재작성하고, 이어지지 않으면 {self.NO_REWRITE_TOKEN}만 출력하세요."
|
||||
)
|
||||
|
||||
# LLM 호출
|
||||
rewritten_query = self.llm_client.chat_completion(
|
||||
messages=[
|
||||
{"role": "system", "content": enhanced_system_prompt},
|
||||
{"role": "user", "content": user_prompt}
|
||||
],
|
||||
max_tokens=1000,
|
||||
temperature=0.1
|
||||
).strip()
|
||||
|
||||
rewritten_query = self._postprocess_rewrite(
|
||||
rewritten_query, original_query, ts, "Full Context Rewriting"
|
||||
)
|
||||
if not rewritten_query:
|
||||
return None
|
||||
|
||||
print(f"[QueryRewriter] {ts} ✅ Full Context Rewriting 완료: '{original_query}' → '{rewritten_query}'")
|
||||
return rewritten_query
|
||||
|
||||
except Exception as e:
|
||||
print(f"[QueryRewriter] {ts} ❌ Full Context Rewriting 실패: {e}")
|
||||
return None
|
||||
|
||||
def rewrite_query(self, original_query: str, history: List[Dict[str, Any]], ts: str) -> Optional[str]:
|
||||
"""질문 재작성
|
||||
|
||||
Args:
|
||||
original_query: 원본 질문
|
||||
history: 대화 이력
|
||||
ts: 타임스탬프 (로깅용)
|
||||
|
||||
Returns:
|
||||
재작성된 질문 또는 None (실패 시)
|
||||
"""
|
||||
if not history:
|
||||
return None
|
||||
|
||||
try:
|
||||
print(f"[QueryRewriter] {ts} 🔄 Query Rewriting 시도 (대화 이력 {len(history)}개)")
|
||||
|
||||
# 대화 이력 포맷팅
|
||||
history_text = self.format_history(history)
|
||||
cue_hint = "있음" if self.has_context_cue(original_query) else "없음"
|
||||
|
||||
# 프롬프트 구성
|
||||
user_prompt = (
|
||||
f"【이전 대화】\n{history_text}\n\n"
|
||||
f"【현재 질문】\n{original_query}\n\n"
|
||||
f"【후속 질문 힌트】\n패턴 기반 후속 질문 후보: {cue_hint}\n\n"
|
||||
f"이전 대화와 자연스럽게 이어지는 후속 질문이면 완전한 질문으로 재작성하고, 이어지지 않으면 {self.NO_REWRITE_TOKEN}만 출력하세요."
|
||||
)
|
||||
|
||||
# LLM 호출
|
||||
rewritten_query = self.llm_client.chat_completion(
|
||||
messages=[
|
||||
{"role": "system", "content": self.system_prompt},
|
||||
{"role": "user", "content": user_prompt}
|
||||
],
|
||||
max_tokens=1000,
|
||||
temperature=0.1
|
||||
).strip()
|
||||
|
||||
rewritten_query = self._postprocess_rewrite(
|
||||
rewritten_query, original_query, ts, "Query Rewriting"
|
||||
)
|
||||
if not rewritten_query:
|
||||
return None
|
||||
|
||||
print(f"[QueryRewriter] {ts} ✅ Query Rewriting 완료: '{original_query}' → '{rewritten_query}'")
|
||||
return rewritten_query
|
||||
|
||||
except Exception as e:
|
||||
print(f"[QueryRewriter] {ts} ❌ Query Rewriting 실패: {e}")
|
||||
return None
|
||||
|
||||
def _remove_think_tags(self, text: str) -> str:
|
||||
"""<think> 태그 제거"""
|
||||
if "<think>" in text and "</think>" in text:
|
||||
return re.sub(r'<think>.*?</think>\s*', '', text, flags=re.DOTALL).strip()
|
||||
return text
|
||||
@@ -0,0 +1,202 @@
|
||||
"""
|
||||
응답 핸들러 모듈
|
||||
──────────────
|
||||
응답 생성, 저장, 로깅 처리
|
||||
"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import List, Dict, Any, Optional
|
||||
|
||||
|
||||
class ResponseHandler:
|
||||
"""응답 처리 및 저장 핸들러"""
|
||||
|
||||
def __init__(self, data_dir: Path, chat_manager=None, config=None):
|
||||
self.data_dir = data_dir
|
||||
self.chat_manager = chat_manager
|
||||
self.config = config
|
||||
|
||||
def build_references(self, top_results: List[Dict], scores: List[float]) -> List[Dict[str, Any]]:
|
||||
"""클라이언트 응답/로그용 참고자료 메타데이터 구성"""
|
||||
references = []
|
||||
for idx, result in enumerate(top_results):
|
||||
references.append({
|
||||
"question": result.get("q"),
|
||||
"answer": result.get("a"),
|
||||
"score": scores[idx] if idx < len(scores) else None,
|
||||
"category": result.get("category"),
|
||||
"source": result.get("source"),
|
||||
"source_id": result.get("source_id"),
|
||||
"quality": result.get("quality"),
|
||||
"url": result.get("url"),
|
||||
})
|
||||
return references
|
||||
|
||||
def extract_faq_urls(self, top_results: List[Dict]) -> List[str]:
|
||||
"""상위 참고자료에서 중복 없는 URL 목록 추출"""
|
||||
urls = []
|
||||
seen = set()
|
||||
for result in top_results:
|
||||
url = result.get("url")
|
||||
if url and url not in seen:
|
||||
urls.append(url)
|
||||
seen.add(url)
|
||||
return urls
|
||||
|
||||
def log_failed_query(self, query: str, ts: str):
|
||||
"""실패한 질문 로깅"""
|
||||
try:
|
||||
fail_path = self.data_dir / "qa_failed.jsonl"
|
||||
fail_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with fail_path.open("a", encoding="utf-8") as fout:
|
||||
rec = {"q": query, "ts": ts}
|
||||
json.dump(rec, fout, ensure_ascii=False)
|
||||
fout.write("\n")
|
||||
except Exception as e:
|
||||
print(f"[ResponseHandler] 실패 로깅 오류: {e}")
|
||||
|
||||
def log_success(
|
||||
self,
|
||||
original_query: str,
|
||||
rewritten_query: Optional[str],
|
||||
top_results: List[Dict],
|
||||
scores: List[float],
|
||||
answer: str,
|
||||
ts: str
|
||||
):
|
||||
"""성공한 질문 로깅"""
|
||||
try:
|
||||
succ_path = self.data_dir / "qa_successed.jsonl"
|
||||
succ_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with succ_path.open("a", encoding="utf-8") as fout:
|
||||
rec = {
|
||||
"user_q": original_query,
|
||||
"top_matches": [{"q": r["q"], "score": s} for r, s in zip(top_results, scores)],
|
||||
"llm_answer": answer,
|
||||
"ts": ts
|
||||
}
|
||||
if rewritten_query:
|
||||
rec["rewritten_q"] = rewritten_query
|
||||
json.dump(rec, fout, ensure_ascii=False)
|
||||
fout.write("\n")
|
||||
except Exception as e:
|
||||
print(f"[ResponseHandler] 성공 로깅 오류: {e}")
|
||||
|
||||
def save_to_mongodb(
|
||||
self,
|
||||
bot_id: Optional[str],
|
||||
user_query: str,
|
||||
ai_response: str,
|
||||
matched_questions: List[str],
|
||||
scores: List[float],
|
||||
metadata: Dict[str, Any],
|
||||
ts: str
|
||||
) -> bool:
|
||||
"""MongoDB에 대화 저장
|
||||
|
||||
Returns:
|
||||
저장 성공 여부
|
||||
"""
|
||||
if not self.chat_manager:
|
||||
print(f"[ResponseHandler] {ts} ⏭️ MongoDB 비활성화 → 저장 스킵")
|
||||
return False
|
||||
|
||||
try:
|
||||
print(f"[ResponseHandler] {ts} ✍️ MongoDB 저장 중... (bot_id={bot_id})")
|
||||
result_id = self.chat_manager.save_conversation(
|
||||
bot_id=bot_id,
|
||||
user_query=user_query,
|
||||
ai_response=ai_response,
|
||||
matched_questions=matched_questions,
|
||||
scores=scores,
|
||||
metadata=metadata
|
||||
)
|
||||
print(f"[ResponseHandler] {ts} ✅ MongoDB 저장 성공! doc_id={result_id}")
|
||||
return True
|
||||
except Exception as e:
|
||||
import traceback
|
||||
print(f"[ResponseHandler] {ts} ❌ MongoDB 저장 실패!")
|
||||
print(f"[ResponseHandler] {ts} 오류: {e}")
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
def print_console_log(
|
||||
self,
|
||||
original_query: str,
|
||||
rewritten_query: Optional[str],
|
||||
top_results: List[Dict],
|
||||
scores: List[float],
|
||||
answer: str,
|
||||
ts: str
|
||||
):
|
||||
"""콘솔 로그 출력"""
|
||||
try:
|
||||
preview = answer.replace("\n", " ")[:120]
|
||||
top_match = top_results[0]["q"] if top_results else "N/A"
|
||||
top_score = scores[0] if scores else None
|
||||
|
||||
query_log = f"'{original_query}'"
|
||||
if rewritten_query:
|
||||
query_log = f"'{original_query}' → '{rewritten_query}'"
|
||||
|
||||
print(f"[ResponseHandler] {ts} query={query_log}")
|
||||
print(f"[ResponseHandler] {ts} top_match='{top_match}' score={top_score}")
|
||||
print(f"[ResponseHandler] {ts} answer_preview='{preview}'")
|
||||
except Exception as e:
|
||||
print(f"[ResponseHandler] 콘솔 로그 오류: {e}")
|
||||
|
||||
def build_response(
|
||||
self,
|
||||
answer: str,
|
||||
matched_questions: List[str],
|
||||
scores: List[float],
|
||||
bot_id: Optional[str],
|
||||
rerank_info: Dict[str, Any],
|
||||
references: List[Dict[str, Any]] = None,
|
||||
faq_urls: List[str] = None,
|
||||
faiss_scores: List[float] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""최종 응답 생성"""
|
||||
if faiss_scores is None:
|
||||
faiss_scores = []
|
||||
if references is None:
|
||||
references = []
|
||||
if faq_urls is None:
|
||||
faq_urls = []
|
||||
|
||||
return {
|
||||
"answer": answer,
|
||||
"matched_questions": matched_questions,
|
||||
"scores": scores,
|
||||
"num_references": len(matched_questions),
|
||||
"botId": bot_id,
|
||||
"references": references,
|
||||
"faq_urls": faq_urls,
|
||||
"rerank_info": rerank_info
|
||||
}
|
||||
|
||||
def build_no_match_response(
|
||||
self,
|
||||
answer: str,
|
||||
bot_id: Optional[str],
|
||||
top_k: int
|
||||
) -> Dict[str, Any]:
|
||||
"""매칭 실패 응답 생성"""
|
||||
return {
|
||||
"answer": answer,
|
||||
"matched_questions": [],
|
||||
"scores": [],
|
||||
"num_references": 0,
|
||||
"botId": bot_id,
|
||||
"references": [],
|
||||
"faq_urls": [],
|
||||
"rerank_info": {
|
||||
"used": False,
|
||||
"faiss_top_k": top_k,
|
||||
"rerank_top_n": 0,
|
||||
"faiss_scores": [],
|
||||
"rerank_scores": [],
|
||||
"detail": "No matching documents found (threshold not met)"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,165 @@
|
||||
"""
|
||||
검색 핸들러 모듈
|
||||
──────────────
|
||||
벡터 검색 및 재랭킹 처리
|
||||
"""
|
||||
|
||||
import numpy as np
|
||||
from typing import List, Dict, Any, Optional, Tuple
|
||||
|
||||
|
||||
def _format_rerank_document(candidate: Dict[str, Any]) -> str:
|
||||
"""리랭커용 FAQ passage — 질문과 답변 전체."""
|
||||
q = str(candidate.get("q") or candidate.get("question") or "").strip()
|
||||
a = str(candidate.get("a") or candidate.get("answer") or "").strip()
|
||||
if q and a:
|
||||
return f"질문: {q}\n답변: {a}"
|
||||
return q or a
|
||||
|
||||
|
||||
class SearchHandler:
|
||||
"""벡터 검색 및 재랭킹 핸들러"""
|
||||
|
||||
def __init__(self, vector_store, embed_client, rerank_client, config):
|
||||
self.vector_store = vector_store
|
||||
self.embed_client = embed_client
|
||||
self.rerank_client = rerank_client
|
||||
self.config = config
|
||||
self.last_search_info: Dict[str, Any] = {}
|
||||
|
||||
def embed_query(self, query: str, ts: str) -> Optional[np.ndarray]:
|
||||
"""질문 임베딩
|
||||
|
||||
Args:
|
||||
query: 검색 질문
|
||||
ts: 타임스탬프 (로깅용)
|
||||
|
||||
Returns:
|
||||
임베딩 벡터 또는 None (실패 시)
|
||||
"""
|
||||
self._last_query_text = query or ""
|
||||
try:
|
||||
embeddings = self.embed_client.embed([query], normalize=True, is_query=True)
|
||||
return np.array(embeddings[0], dtype="float32")
|
||||
except Exception as e:
|
||||
print(f"[SearchHandler] {ts} ❌ 임베딩 실패: {e}")
|
||||
return None
|
||||
|
||||
def search(self, query_vec: np.ndarray, threshold: float, ts: str) -> List[Dict[str, Any]]:
|
||||
"""벡터 검색
|
||||
|
||||
Args:
|
||||
query_vec: 질문 임베딩 벡터
|
||||
threshold: 유사도 임계값
|
||||
ts: 타임스탬프 (로깅용)
|
||||
|
||||
Returns:
|
||||
검색 결과 리스트
|
||||
"""
|
||||
try:
|
||||
if (
|
||||
getattr(self.config, "hybrid_search_enabled", False)
|
||||
and hasattr(self.vector_store, "hybrid_search")
|
||||
):
|
||||
results = self.vector_store.hybrid_search(
|
||||
query_vec,
|
||||
getattr(self, "_last_query_text", ""),
|
||||
top_k=self.config.top_k,
|
||||
threshold=threshold,
|
||||
sparse_top_k=getattr(self.config, "sparse_top_k", 30),
|
||||
merge_top_k=getattr(self.config, "hybrid_merge_top_k", 40),
|
||||
)
|
||||
self.last_search_info = {
|
||||
"searchMode": "hybrid_sparse",
|
||||
"candidateCount": len(results),
|
||||
}
|
||||
else:
|
||||
results = self.vector_store.search(
|
||||
query_vec,
|
||||
top_k=self.config.top_k,
|
||||
threshold=threshold
|
||||
)
|
||||
self.last_search_info = {
|
||||
"searchMode": "vector",
|
||||
"candidateCount": len(results),
|
||||
}
|
||||
return results
|
||||
except Exception as e:
|
||||
print(f"[SearchHandler] {ts} ❌ 벡터 검색 실패: {e}")
|
||||
self.last_search_info = {"searchMode": "error", "error": str(e)}
|
||||
return []
|
||||
|
||||
def rerank(self, query: str, candidates: List[Dict[str, Any]], ts: str) -> Tuple[List[Dict[str, Any]], List[float], bool, Optional[Dict]]:
|
||||
"""재랭킹
|
||||
|
||||
Args:
|
||||
query: 검색 질문
|
||||
candidates: 후보 문서 리스트
|
||||
ts: 타임스탬프 (로깅용)
|
||||
|
||||
Returns:
|
||||
(상위 N개 결과, 점수 리스트, 재랭킹 사용 여부, 재랭킹 정보)
|
||||
"""
|
||||
if not candidates:
|
||||
print(f"[SearchHandler] {ts} 재랭킹 건너뜀: 후보 문서 없음")
|
||||
return [], [], False, {"detail": "No candidates to rerank"}
|
||||
|
||||
# 재랭킹 후보 제한
|
||||
max_rerank = min(self.config.rerank_candidates, len(candidates))
|
||||
rerank_candidates = candidates[:max_rerank]
|
||||
|
||||
print(f"[SearchHandler] {ts} 재랭킹 시작 (candidates={len(candidates)} → 상위 {max_rerank}개)")
|
||||
|
||||
try:
|
||||
# TEI Reranker API 호출 (질문+답변 전체를 passage로 전달)
|
||||
documents = [_format_rerank_document(c) for c in rerank_candidates]
|
||||
print(f"[SearchHandler] {ts} 리랭커 호출: query={query[:50]}..., documents={len(documents)}개 (Q+A)")
|
||||
|
||||
rerank_results = self.rerank_client.rerank(
|
||||
query=query,
|
||||
documents=documents,
|
||||
return_documents=False
|
||||
)
|
||||
|
||||
print(f"[SearchHandler] {ts} 리랭커 응답: {len(rerank_results) if rerank_results else 0}개 결과")
|
||||
|
||||
if not rerank_results:
|
||||
print(f"[SearchHandler] {ts} 재랭킹 결과 없음 → FAISS 상위 결과 사용")
|
||||
top_n_results = rerank_candidates[:self.config.top_n_for_llm]
|
||||
top_scores = [None] * len(top_n_results)
|
||||
return top_n_results, top_scores, False, None
|
||||
|
||||
# score 기준 정렬 후 상위 N개
|
||||
sorted_results = sorted(rerank_results, key=lambda x: x["score"], reverse=True)
|
||||
top_n_indices = [r["index"] for r in sorted_results[:self.config.top_n_for_llm]]
|
||||
top_n_results = [rerank_candidates[idx] for idx in top_n_indices]
|
||||
top_scores = [
|
||||
round(sorted_results[i]["score"], 4) if i < len(sorted_results) else None
|
||||
for i in range(len(top_n_results))
|
||||
]
|
||||
|
||||
# 순위 변화 계산
|
||||
rank_changes = [idx - i for i, idx in enumerate(top_n_indices)]
|
||||
|
||||
def _question_of(candidate: Dict[str, Any]) -> str:
|
||||
return str(candidate.get("q") or candidate.get("question") or "")
|
||||
|
||||
rerank_info = {
|
||||
"used": True,
|
||||
"original_top_question": _question_of(rerank_candidates[0])[:50] + "...",
|
||||
"reranked_top_question": _question_of(top_n_results[0])[:50] + "...",
|
||||
"rank_changes": rank_changes
|
||||
}
|
||||
|
||||
print(f"[SearchHandler] {ts} 재랭킹 완료: 상위 {len(top_n_results)}개, 최고 점수={top_scores[0]}")
|
||||
if rank_changes[0] != 0:
|
||||
print(f"[SearchHandler] {ts} 순위 변화: FAISS #{top_n_indices[0]+1} → Rerank #1")
|
||||
|
||||
return top_n_results, top_scores, True, rerank_info
|
||||
|
||||
except Exception as e:
|
||||
print(f"[SearchHandler] {ts} ❌ 재랭킹 실패: {e}")
|
||||
print(f"[SearchHandler] {ts} FAISS 상위 결과로 대체")
|
||||
top_n_results = candidates[:self.config.top_n_for_llm]
|
||||
top_scores = [None] * len(top_n_results)
|
||||
return top_n_results, top_scores, False, None
|
||||
@@ -0,0 +1,193 @@
|
||||
"""
|
||||
제안 핸들러 모듈
|
||||
──────────────
|
||||
낮은 신뢰도 시 대안 질문 제안
|
||||
"""
|
||||
|
||||
from typing import List, Dict, Any, Optional
|
||||
|
||||
|
||||
SUGGESTION_BLOCK_MARKER = "💡 혹시 이런 것을 찾으셨나요?"
|
||||
SUGGESTION_BLOCK_MARKER_ALT = "혹시 이런 것을 찾으셨나요?"
|
||||
|
||||
|
||||
def strip_suggestion_block(text: Optional[str]) -> str:
|
||||
"""LLM 이력용: 추천 블록(및 LLM 모방 블록) 제거, FAQ 본문만 반환"""
|
||||
if not text:
|
||||
return text or ""
|
||||
|
||||
indices = []
|
||||
for marker in (SUGGESTION_BLOCK_MARKER, SUGGESTION_BLOCK_MARKER_ALT):
|
||||
idx = text.find(marker)
|
||||
if idx != -1:
|
||||
indices.append(idx)
|
||||
|
||||
if not indices:
|
||||
return text
|
||||
|
||||
return text[: min(indices)].rstrip()
|
||||
|
||||
|
||||
def has_suggestion_block(text: Optional[str]) -> bool:
|
||||
if not text:
|
||||
return False
|
||||
return any(marker in text for marker in (SUGGESTION_BLOCK_MARKER, SUGGESTION_BLOCK_MARKER_ALT))
|
||||
|
||||
|
||||
class SuggestionHandler:
|
||||
"""낮은 신뢰도 답변에 대한 대안 질문 제안"""
|
||||
|
||||
# 신뢰도 임계값 (기본값 — Config.from_env / LOW_CONFIDENCE_THRESHOLD 로 override)
|
||||
DEFAULT_LOW_CONFIDENCE_THRESHOLD = 0.65
|
||||
DEFAULT_HIGH_CONFIDENCE_THRESHOLD = 0.75
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
low_confidence_threshold: float = DEFAULT_LOW_CONFIDENCE_THRESHOLD,
|
||||
high_confidence_threshold: float = DEFAULT_HIGH_CONFIDENCE_THRESHOLD,
|
||||
):
|
||||
"""
|
||||
Args:
|
||||
low_confidence_threshold: 낮은 신뢰도 기준 (기본 0.65, env LOW_CONFIDENCE_THRESHOLD)
|
||||
high_confidence_threshold: 높은 신뢰도 기준 (기본 0.75, env HIGH_CONFIDENCE_THRESHOLD)
|
||||
"""
|
||||
self.low_confidence_threshold = low_confidence_threshold
|
||||
self.high_confidence_threshold = high_confidence_threshold
|
||||
|
||||
def is_low_confidence(self, top_score: Optional[float]) -> bool:
|
||||
"""신뢰도가 낮은지 판단
|
||||
|
||||
Args:
|
||||
top_score: 최상위 검색 결과의 점수
|
||||
|
||||
Returns:
|
||||
낮은 신뢰도 여부
|
||||
"""
|
||||
if top_score is None:
|
||||
return True
|
||||
return top_score < self.low_confidence_threshold
|
||||
|
||||
def generate_suggestions(
|
||||
self,
|
||||
top_results: List[Dict[str, Any]],
|
||||
top_scores: List[float],
|
||||
max_suggestions: int = 3
|
||||
) -> Optional[str]:
|
||||
"""대안 질문 제안 생성
|
||||
|
||||
Args:
|
||||
top_results: 검색 결과 리스트
|
||||
top_scores: 검색 점수 리스트
|
||||
max_suggestions: 최대 제안 개수 (기본 3개)
|
||||
|
||||
Returns:
|
||||
제안 문자열 또는 None
|
||||
"""
|
||||
if not top_results or not top_scores:
|
||||
return None
|
||||
|
||||
# 최상위 점수가 충분히 높으면 제안 불필요
|
||||
if not self.is_low_confidence(top_scores[0]):
|
||||
return None
|
||||
|
||||
# 1위 포함 상위 max_suggestions개 질문 제안 (번호 1부터)
|
||||
suggestion_count = min(max_suggestions, len(top_results))
|
||||
|
||||
if suggestion_count <= 0:
|
||||
return None
|
||||
|
||||
suggestions = []
|
||||
for i in range(suggestion_count):
|
||||
question = top_results[i]["q"]
|
||||
score = top_scores[i] if i < len(top_scores) else None
|
||||
suggestions.append({
|
||||
"index": i + 1,
|
||||
"question": question,
|
||||
"score": score,
|
||||
})
|
||||
|
||||
if not suggestions:
|
||||
return None
|
||||
|
||||
# 제안 텍스트 생성
|
||||
suggestion_text = f"\n\n{SUGGESTION_BLOCK_MARKER}\n"
|
||||
|
||||
for suggestion in suggestions:
|
||||
# 점수가 있으면 신뢰도 표시 (선택적)
|
||||
# suggestion_text += f"{suggestion['index']}. {suggestion['question']} (유사도: {suggestion['score']:.2f})\n"
|
||||
suggestion_text += f"{suggestion['index']}. {suggestion['question']}\n"
|
||||
|
||||
suggestion_text += "\n위 질문 중 하나를 선택하시면 정확한 답변을 드리겠습니다."
|
||||
|
||||
return suggestion_text
|
||||
|
||||
def enhance_answer_with_suggestions(
|
||||
self,
|
||||
answer: str,
|
||||
top_results: List[Dict[str, Any]],
|
||||
top_scores: List[float],
|
||||
max_suggestions: int = 3
|
||||
) -> str:
|
||||
"""답변에 제안 추가
|
||||
|
||||
Args:
|
||||
answer: 원본 답변
|
||||
top_results: 검색 결과
|
||||
top_scores: 검색 점수
|
||||
max_suggestions: 최대 제안 개수
|
||||
|
||||
Returns:
|
||||
제안이 추가된 답변 (또는 원본)
|
||||
"""
|
||||
if has_suggestion_block(answer):
|
||||
return answer
|
||||
|
||||
suggestions = self.generate_suggestions(top_results, top_scores, max_suggestions)
|
||||
|
||||
if suggestions:
|
||||
return answer + suggestions
|
||||
|
||||
return answer
|
||||
|
||||
def get_confidence_level(self, score: Optional[float]) -> str:
|
||||
"""점수를 신뢰도 레벨로 변환
|
||||
|
||||
Args:
|
||||
score: 검색 점수
|
||||
|
||||
Returns:
|
||||
"high", "medium", "low" 중 하나
|
||||
"""
|
||||
if score is None:
|
||||
return "low"
|
||||
|
||||
if score >= self.high_confidence_threshold:
|
||||
return "high"
|
||||
elif score >= self.low_confidence_threshold:
|
||||
return "medium"
|
||||
else:
|
||||
return "low"
|
||||
|
||||
def should_show_confidence_warning(self, top_score: Optional[float]) -> bool:
|
||||
"""신뢰도 경고를 표시해야 하는지
|
||||
|
||||
Args:
|
||||
top_score: 최상위 점수
|
||||
|
||||
Returns:
|
||||
경고 표시 여부
|
||||
"""
|
||||
if top_score is None:
|
||||
return True
|
||||
|
||||
# 매우 낮은 신뢰도 (0.50 미만)
|
||||
return top_score < 0.50
|
||||
|
||||
def generate_confidence_warning(self) -> str:
|
||||
"""신뢰도 경고 메시지 생성"""
|
||||
return (
|
||||
"\n\n⚠️ **정확도 안내**\n"
|
||||
"질문과 정확히 일치하는 정보를 찾기 어려웠습니다.\n"
|
||||
"더 구체적으로 질문하시거나, 아래 옵션을 참고해 주세요.\n"
|
||||
"정확한 답변이 필요하시면 한국도로공사 콜센터(1588-2504)로 문의해 주세요."
|
||||
)
|
||||
Reference in New Issue
Block a user