Macbook
4b86b2a660
- server-dev start/stop/deploy 및 Gitea push 자동 배포 - local-dev 로컬 개발 환경 Co-authored-by: Cursor <cursoragent@cursor.com>
246 lines
12 KiB
Python
246 lines
12 KiB
Python
"""
|
|
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
|