--39.답변검색.ipynb--

사전 필요모듈, 데이터:

Konlpy 설치

준비해둘 파일들

./utils/Preprocess.py

./config/DatabaseConfig.py

./config/GlobalParams.py

./models/ner/NerModel.py

./models/intent/IntentModel.py

./out 에 만들어둔 파일들

chatbot_dict.bin 이전에 만든 사전파일

*.h5

user_dic.tsv 사용자 사전

example.db <- 이전에 구축한 데이터베이스

답변검색

입력된 문장 -> 전처리 -> 의도분류 -> 개체명 인식

위와 같이 해석된 데이터를 기반으로 적절한 답변을 DB로 부터 검색하여 답변

base_path = r'/content/drive/MyDrive/dataset/chatbot'

데이터 베이스 제어 모듈

/utils/Database.py 생성

import sqlite3
import logging

class Database:
def init(self, db_file):
self.db_file = db_file
self.conn = None

DB 연결

def connect(self):
if self.conn != None:
return

self.conn = sqlite3.connect(self.db_file)
self.conn.row_factory = sqlite3.Row  # SELECT 결과를 Row 객체로 받게 함. => Row객체는 나중에 dict, tuple 형태로 변환 가능.
print("DB 연결 성공")

DB 연결 닫기

def close(self):
if self.conn is None:
return

self.conn.close()
self.conn = None
print("DB 연결 닫기 성공")

SQL 구문 실행 (INSERT, UPDATE, DELETE <- DML명령)

def execute(self, sql):
last_row_id = -1
try:
cursor = self.conn.cursor()
cursor.execute(sql)
self.conn.commit()
last_row_id = cursor.lastrowid
except Exception as ex:
logging.error(ex)

finally:
  return last_row_id

SELECT 구문 실행. 1개의 ROW만 불러오기

def select_one(self, sql):
result = None

try:
  cursor = self.conn.cursor()
  cursor.execute(sql)
  result = dict(cursor.fetchone())  # 한개의 Row 를 읽어와 dict로 변환
except Exception as ex:
  logging.error(ex)
finally:
  return result

SELECT 구문 실행. 전체 Row(들) 불러오기, dict 의 list 형태로 리턴

def select_all(self, sql):
results = None

try:
  cursor = self.conn.cursor()
  cursor.execute(sql)
  results = [dict(row) for row in cursor.fetchall()]
except Exception as ex:
  logging.error(ex)
finally:
  return results

from utils.Database import Database

테스트

db = Database('example.db')
db.connect()

results = db.select_all("SELECT * FROM chatbot_train_data")
print(results)

result = db.select_one("SELECT * FROM chatbot_train_data WHERE id = 1")
print(result)
print(result['id'], result['intent'])

db.close()

답변 검색 모듈

/utils/FindAnswer.py 생성

class FindAnswer :

Database 인스턴스 객체를 받아 생성

def init(self, db) :
self.db = db

답변검색

의도명(intent_name) 과 개체명 태그 리스트 (ner_tags) 를 이용해 질문의 답변을 검색

def search (self, intent_name, ner_tags) :

# 의도명, 개체명으로 답변 검색
sql = self._make_query(intent_name, ner_tags)
answer = self.db.select_one(sql)
return answer

# 검색되는 답변이 없었으면 의도명만 이용하여 답변 검색
if answer is None :
  sql = self._make_query(intent_name, None)
  answer = self.db.select_one(sql)
return answer['answer'], answer['answer_image']

검색 쿼리 생성

'의도명' 만 검색할지, 여러종류의 개체명 태그와 함께 검색할지 결정하는 '조건'을 만드는 간단한 함수

SELECT * FROM chatbot_train_data WHERE intent = '주문' ORDER BY random() LIMIT 1

SELECT * FROM chatbot_train_data WHERE intent = '주문' and ( ner LIKE '%B_FOOD%' ) ORDER BY random() LIMIT 1

SELECT * FROM chatbot_train_data WHERE intent = '예약' and ( ner LIKE '%B_FOOD%' OR ner LIKE '%B_DT%' ) ORDER BY random() LIMIT 1

def make_query(self, intent_name, ner_tags) : # 로 시작하는 함수는 보통 class 내부에서 쓰이는 함수라고 보면 됨.
sql = "SELECT * FROM chatbot_train_data"
if intent_name != None and ner_tags == None :
sql += f" WHERE intent = '{intent_name }'"

elif intent_name != None and ner_tags != None :
  where = f' WHERE intent = "{intent_name}"'
  if len(ner_tags) > 0 :
    where += ' and ('
    for ne in ner_tags :
      where += f" ner LIKE '%{ne}%' OR "
    where = where[: -3] + ')'  # 마지막 'OR '을 없애기 위해
  sql += where

# 동일한 답변이 복수개인 경우, 랜덤으로 한개 선택
sql += " ORDER BY random() LIMIT 1"
return sql

④ NER 태그를 실제 입력된 단어로 변환

예를 들어 '자장면 주문할께요' 라는 텍스트가 챗본 엔진에 입력되었다고 합시다.

그러면 챗봇 엔진은 '자장면'을 'B_FOOD 객체명'으로 인식합니다.

이때 검색된 답변이 '{B_FOOD} 주문 처리 완료 되었습니다 주문해주셔서 감사합니다' 라고 한다면,

답변 내용속 '{B_FOOD}' 를 '자장면' 으로 치환해 주는 함수입니다.

치환해야 하는 태그가 더 존재한다면 치환 규칙을 추가하면 됩니다.

def tag_to_word(self, ner_predicts, answer) :
for word, tag in ner_predicts :

  # 변환해야 하는 태그가 있는 경우 치환
  if tag in ['B_FOOD', 'B_DT', 'B_TI'] :
    answer = answer.replace(f'{{{tag}}}', word)

return answer

from utils.FindAnswer import FindAnswer

챗봇 엔진 동작 테스트

!pip install konlpy

from config.DatabaseConfig import *
from utils.Database import Database
from utils.Preprocess import Preprocess

import tensorflow as tf
from tensorflow.keras.models import Model, load_model
from tensorflow.keras import preprocessing
import os

실행마다 동일한 결과를 얻기 위해 keras 에 랜덤시드 사용

tf.keras.utils.set_random_seed(42)
tf.config.experimental.enable_op_determinism()

DB_FILE

전처리 객체 생성

p = Preprocess(word2index_dic= os.path.join(base_path, 'out', 'chatbot_dict.bin'),
userdic = os.path.join(base_path, 'user_dic.tsv'))

DB 연결 객체 생성

db = Database(db_file = DB_FILE)
db.connect()

의도분류 모델

from models.intent.IntentModel import IntentModel
intent= IntentModel(model_name = os.path.join(base_path, 'out', 'intent_model.h5'), preprocess=p)

개체명 인식 모델

from models.ner.NerModel import NerModel
ner = NerModel(model_name=os.path.join(base_path, 'out', 'ner_model.h5'), preprocess=p)

def chat(query) :
predict = intent.predict_class(query)
intent_name = intent.labels[predict]

predicts = ner.predict(query)
ner_tags = ner.predict_tags(query)

print('질문: ', query)
print("=" * 100)
print("의도 파악: ", intent_name)
print("개체명 인식: ", predicts)
print("답변 검색에 필요한 NER 태그: ", ner_tags)

queries = [
"오전에 탕수육 10개 주문합니다",
"화자의 질문 의도를 파악합니다",
"안녕하세요",
"자장면 주문할께요"
]

for q in queries :
chat(q)

답변검색

from utils.FindAnswer import FindAnswer

def chatbot(query) :
predict = intent.predict_class(query)
intent_name = intent.labels[predict]

predicts = ner.predict(query)
ner_tags = ner.predict_tags(query)

print('질문: ', query)
print("=" * 100)
print("의도 파악: ", intent_name)
print("개체명 인식: ", predicts)
print("답변 검색에 필요한 NER 태그: ", ner_tags)

try :
f = FindAnswer(db)
answertext, = f.search(intent_name, ner_tags)
answer = f.tag_to_word(predicts, answer_text)
except :
answer = "죄송해요 무슨 말인지 모르곘어요"
return answer

chatbot(queries[0])

chatbot("짬뽕 2개 주문합니다")

chatbot("수고하셨습니다")