Create chatbot.py

chatbot.py

@@ -0,0 +1,156 @@
+import re
+import json
+import numpy as np
+import requests
+from tensorflow.keras.models import load_model
+from tensorflow.keras.preprocessing.text import tokenizer_from_json
+from tensorflow.keras.preprocessing.sequence import pad_sequences
+
+# 토크나이저 불러오기
+def load_tokenizer(filename):
+    with open(filename, 'r', encoding='utf-8') as f:
+        return tokenizer_from_json(json.load(f))
+
+tokenizer_q = load_tokenizer('kossistant_q.json')
+tokenizer_a = load_tokenizer('kossistant_a.json')
+
+# 모델 및 파라미터 로드
+model = load_model('kossistant.h5', compile=False)
+max_len_q = model.input_shape[0][1]
+max_len_a = model.input_shape[1][1]
+index_to_word = {v: k for k, v in tokenizer_a.word_index.items()}
+index_to_word[0] = ''
+start_token = 'start'
+end_token = 'end'
+
+# 토큰 샘플링 함수
+def sample_from_top_p_top_k(prob_dist, top_p=0.85, top_k=40, temperature=0.8, repetition_penalty=1.4, generated_ids=[]):
+    logits = np.log(prob_dist + 1e-9) / temperature
+    for idx in generated_ids:
+        logits[idx] /= repetition_penalty
+    probs = np.exp(logits)
+    probs = probs / np.sum(probs)
+    top_k_indices = np.argsort(probs)[-top_k:]
+    top_k_probs = probs[top_k_indices]
+    sorted_indices = top_k_indices[np.argsort(top_k_probs)[::-1]]
+    sorted_probs = probs[sorted_indices]
+    cumulative_probs = np.cumsum(sorted_probs)
+    cutoff_index = np.searchsorted(cumulative_probs, top_p)
+    final_indices = sorted_indices[:cutoff_index + 1]
+    final_probs = probs[final_indices]
+    final_probs = final_probs / np.sum(final_probs)
+    return np.random.choice(final_indices, p=final_probs)
+
+# 디코딩
+def decode_sequence_custom(input_text, max_attempts=2):
+    input_seq = tokenizer_q.texts_to_sequences([input_text])
+    input_seq = pad_sequences(input_seq, maxlen=max_len_q, padding='post')
+
+    for _ in range(max_attempts + 1):
+        target_seq = tokenizer_a.texts_to_sequences([start_token])[0]
+        target_seq = pad_sequences([target_seq], maxlen=max_len_a, padding='post')
+
+        decoded_sentence = ''
+        generated_ids = []
+
+        for i in range(max_len_a):
+            predictions = model.predict([input_seq, target_seq], verbose=0)
+            prob_dist = predictions[0, i, :]
+            pred_id = sample_from_top_p_top_k(prob_dist, generated_ids=generated_ids)
+            generated_ids.append(pred_id)
+            pred_word = index_to_word.get(pred_id, '')
+            if pred_word == end_token:
+                break
+            decoded_sentence += pred_word + ' '
+            if i + 1 < max_len_a:
+                target_seq[0, i + 1] = pred_id
+
+        cleaned = re.sub(r'\b<end>\b', '', decoded_sentence)
+        cleaned = re.sub(r'\s+', ' ', cleaned)
+        if is_valid_response(cleaned):
+            return cleaned.strip()
+
+    return "죄송해요, 답변 생성에 실패했어요."
+
+def is_valid_response(response):
+    if len(response.strip()) < 2:
+        return False
+    if re.search(r'[ㄱ-ㅎㅏ-ㅣ]{3,}', response):
+        return False
+    if len(response.split()) < 2:
+        return False
+    if response.count(' ') < 2:
+        return False
+    if any(tok in response.lower() for tok in ['hello', 'this', 'ㅋㅋ']):
+        return False
+    return True
+
+def extract_main_query(text):
+    sentences = re.split(r'[.?!]\s*', text)
+    sentences = [s.strip() for s in sentences if s.strip()]
+    if not sentences:
+        return text
+    last = sentences[-1]
+    last = re.sub(r'[^가-힣a-zA-Z0-9 ]', '', last)
+    particles = ['이', '가', '은', '는', '을', '를', '의', '에서', '에게', '한테', '보다']
+    for p in particles:
+        last = re.sub(rf'\b(\w+){p}\b', r'\1', last)
+    return last.strip()
+
+def get_wikipedia_summary(query):
+    cleaned_query = extract_main_query(query)
+    url = f"https://ko.wikipedia.org/api/rest_v1/page/summary/{cleaned_query}"
+    res = requests.get(url)
+    if res.status_code == 200:
+        return res.json().get("extract", "요약 정보를 찾을 수 없습니다.")
+    else:
+        return "위키백과에서 정보를 가져올 수 없습니다."
+
+def simple_intent_classifier(text):
+    text = text.lower()
+    greet_keywords = ["안녕", "반가워", "이름", "누구", "소개", "어디서 왔", "정체", "몇 살", "너 뭐야"]
+    info_keywords = ["설명", "정보", "무엇", "뭐야", "어디", "누구", "왜", "어떻게", "종류", "개념"]
+    math_keywords = ["더하기", "빼기", "곱하기", "나누기", "루트", "제곱", "+", "-", "*", "/", "=", "^", "√", "계산", "몇이야", "얼마야"]
+
+    if any(kw in text for kw in greet_keywords):
+        return "인사"
+    elif any(kw in text for kw in info_keywords):
+        return "정보질문"
+    elif any(kw in text for kw in math_keywords):
+        return "수학질문"
+    else:
+        return "일상대화"
+
+def parse_math_question(text):
+    text = text.replace("곱하기", "*").replace("더하기", "+").replace("빼기", "-").replace("나누기", "/").replace("제곱", "*2")
+    text = re.sub(r'루트\s(\d+)', r'math.sqrt(\1)', text)
+    try:
+        result = eval(text)
+        return f"정답은 {result}입니다."
+    except:
+        return "계산할 수 없는 수식이에요. 다시 한번 확인해 주세요!"
+
+# 전체 응답 함수
+def respond(input_text):
+    intent = simple_intent_classifier(input_text)
+
+    if "/사용법" in input_text:
+        return "자유롭게 사용해주세요. 딱히 제약은 없습니다."
+
+    if "이름" in input_text:
+        return "제 이름은 kossistant입니다."
+
+    if "누구" in input_text:
+        return "저는 kossistant이라고 해요."
+
+    if intent == "수학질문":
+        return parse_math_question(input_text)
+
+    if intent == "정보질문":
+        keyword = re.sub(r"(에 대해|에 대한|에 대해서)?\s*(설명해줘|알려줘|뭐야|개념|정의|정보)?", "", input_text).strip()
+        if not keyword:
+            return "어떤 주제에 대해 궁금한가요?"
+        summary = get_wikipedia_summary(keyword)
+        return f"{summary}\n다른 궁금한 점 있으신가요?"
+
+    return decode_sequence_custom(input_text)