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