[처음 배우는 딥러닝 챗봇] Python을 이용한 딥러닝 챗봇 제작기 (6)
데이터베이스 제어 모듈 생성
지금까지의 과정들을 통해 우리는 화자로부터 문장을 입력받고, 이를 전처리, 의도 분류, 개체명 인식 작업을 거쳐 해석하는 단계까지 진행해왔다. 이제 해석된 결과를 기반으로 DB에서 적절한 답변을 검색하는 기능을 구현해보자.
데이터베이스 제어를 용이하게 하는 모듈을 먼저 구현한다. /utils 디렉터리에 Database.py 파일을 생성하여 다음과 같이 코드를 작성한다.
import pymysql
import pymysql.cursors
import logging
class Database:
'''
데이터베이스 제어
'''
def __init__(self, host, user, password, db_name, charset='utf8'):
self.host = host
self.user = user
self.password = password
self.charset = charset
self.db_name = db_name
self.conn = None
# DB 연결
def connect(self):
if self.conn is not None:
return
self.conn = pymysql.connect(
host=self.host,
user=self.user,
password=self.password,
db=self.db_name,
charset=self.charset
)
# DB 연결 닫기
def close(self):
if self.conn is None:
return
if not self.conn.open:
self.conn = None
return
self.conn.close()
self.conn = None
# SQL 구문 실행
def execute(self, sql):
last_row_id = -1
try:
with self.conn.cursor() as cursor:
cursor.execute(sql)
self.conn.commit()
last_row_id = cursor.lastrowid
# logging.debug("execute last_row_id : %d", last_row_id)
except Exception as ex:
logging.error(ex)
finally:
return last_row_id
# SELECT 구문 실행 후 단 1개의 데이터 ROW만 불러옴
def select_one(self, sql):
result = None
try:
with self.conn.cursor(pymysql.cursors.DictCursor) as cursor:
cursor.execute(sql)
result = cursor.fetchone()
except Exception as ex:
logging.error(ex)
finally:
return result
# SELECT 구문 실행 후 전체 데이터 ROW를 불러옴
def select_all(self, sql):
result = None
try:
with self.conn.cursor(pymysql.cursors.DictCursor) as cursor:
cursor.execute(sql)
result = cursor.fetchall()
except Exception as ex:
logging.error(ex)
finally:
return result
답변 검색 모듈 생성
전처리, 의도 분류, 개체명 인식 과정을 통해 얻은 해석 결과를 이용해 학습 DB에서 적절한 답변을 검색하는 기능을 구현한다. /utils 디렉터리에 FindAnswer.py 파일을 생성하여 다음과 같이 코드를 작성한다.
class FindAnswer:
def __init__(self, db):
self.db = db
# 검색 쿼리 생성
def _make_query(self, intent_name, ner_tags):
sql = "select * from chatbot_train_data"
if intent_name is not None and ner_tags is None:
sql = sql + " where intent='{}' ".format(intent_name)
elif intent_name is not None and ner_tags is not None:
where = " where intent='%s' " % intent_name
if len(ner_tags) > 0:
where += "and ("
for ne in ner_tags:
where += " ner like '%{}%' or ".format(ne)
where = where[:-3] + ')'
sql = sql + where
# 동일한 답변이 2개 이상인 경우 랜덤으로 선택
sql = sql + " order by rand() limit 1"
return sql
# 답변 검색
def search(self, intent_name, ner_tags):
# 의도명과 개체명으로 답변 검색
sql = self._make_query(intent_name, ner_tags)
answer = self.db.select_one(sql)
# 검색되는 답변이 없으면 의도명만 검색
if answer is None:
sql = self._make_query(intent_name, None)
answer = self.db.select_one(sql)
return answer['answer'], answer['answer_image']
# NER 태그를 실제 입력된 단어로 변환
def tag_to_word(self, ner_predicts, answer):
for word, tag in ner_predicts:
# 변환해야 하는 태그가 있는 경우 추가
if tag == 'B_FOOD':
answer = answer.replace(tag, word)
answer = answer.replace('{', '')
answer = answer.replace('}', '')
return answer
챗봇 엔진 동작
앞서 작성한 코드가 잘 작동하는지를 확인해보자. /test 디렉터리에 chatbot_test.py 파일을 생성하여 다음과 같이 코드를 작성한다.
from config.DatabaseConfig import *
from utils.Database import Database
from utils.Preprocess import Preprocess
# 전처리 객체 생성
p = Preprocess(word2index_dic='../train_tools/dict/chatbot_dict.bin',
userdic='../utils/user_dic.tsv')
# 질문/답변 학습 DB 연결 객체 생성
db = Database(
host=DB_HOST, user=DB_USER, password=DB_PASSWORD, db_name=DB_NAME
)
db.connect() # DB 연결
# 원문
query = input()
# 의도 파악
from models.intent.IntentModel import IntentModel
intent = IntentModel(model_name='../models/intent/intent_model.h5', preprocess=p)
predict = intent.predict_class(query)
intent_name = intent.labels[predict]
# 개체명 인식
from models.ner.NerModel import NerModel
ner = NerModel(model_name='../models/ner/ner_model.h5', preprocess=p)
predicts = ner.predict(query)
ner_tags = ner.predict_tags(query)
print("질문 : ", query)
print("=" * 40)
print("의도 파악 : ", intent_name)
print("개체명 인식 : ", predicts)
print("답변 검색에 필요한 NER 태그", ner_tags)
print("=" * 40)
# 답변 검색
from utils.FindAnswer import FindAnswer
try:
f = FindAnswer(db)
answer_text, answer_image = f.search(intent_name, ner_tags)
answer = f.tag_to_word(predicts, answer_text)
except:
answer = "죄송해요, 무슨 말인지 모르겠어요."
print("답변 : ", answer)
db.close() # DB 연결 끊음
다중 접속을 위한 TCP 소켓 서버
지금까지 만든 챗봇 엔진을 다양한 플랫폼에서 사용할 수 있게 하기 위해 서버 통신을 위한 기능을 구현해보자. 이 때 서버 통신용 프로토콜로 JSON 포멧을 이용한다. 구현은 클라이언트에서 서버 쪽으로 요청하는 프로토콜과 챗봇 엔진의 처리 결과를 클라이언트 쪽으로 응답하는 프로토콜 두 개를 구현한다.
지금까지의 예제에서는 한 번에 하나의 작업만 실행하는 싱글 스레드 방식을 이용했다. 그러나 실제 서비스에서는 다수의 서비스 요청에 응답할 수 있도록 멀티 스레드 방식을 이용해야 한다.
이 때 사용되는 TCP 소켓 서버를 관리하는 모듈을 구현한다. /utils 디렉터리에 BotServer.py 파일을 생성하여 다음과 같이 코드를 작성한다.
import socket
class BotServer:
def __init__(self, srv_port, listen_num):
self.port = srv_port
self.listen = listen_num
self.mySock = None
# sock 생성
def create_sock(self):
self.mySock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.mySock.bind(("0.0.0.0", int(self.port)))
self.mySock.listen(int(self.listen))
return self.mySock
# client 대기
def ready_for_client(self):
return self.mySock.accept()
# sock 반환
def get_sock(self):
return self.mySock
챗봇 서버 메인 프로그램
이제 챗봇 엔진 서버 프로그램을 구현한다. 이전에 만든 챗봇 엔진이 서버에서 작동할 수 있도록 BotServer 클래스와 멀티 스레드 모듈을 이용한다. 루트 디렉터리에 bot.py 파일을 생성하여 다음과 같이 코드를 작성한다.
import threading
import json
from config.DatabaseConfig import *
from utils.Database import Database
from utils.BotServer import BotServer
from utils.Preprocess import Preprocess
from models.intent.IntentModel import IntentModel
from models.ner.NerModel import NerModel
from utils.FindAnswer import FindAnswer
# 전처리 객체 생성
p = Preprocess(word2index_dic='train_tools/dict/chatbot_dict.bin',
userdic='utils/user_dic.tsv')
# 의도 파악 모델
intent = IntentModel(model_name='models/intent/intent_model.h5', preprocess=p)
# 개체명 인식 모델
ner = NerModel(model_name='models/ner/ner_model.h5', preprocess=p)
def to_client(conn, addr, params):
db = params['db']
try:
db.connect() # DB 연결
# 데이터 수신
read = conn.recv(2048) # 수신 데이터가 있을 때까지 블로킹
print("===========================")
print("Connection from: %s" % str(addr))
if read is None or not read:
# 클라이언트 연결이 끊어지거나 오류가 있는 경우
print("클라이언트 연결 끊어짐")
exit(0) # 스레드 강제 종료
# json 데이터로 변환
recv_json_data = json.loads(read.decode())
print("데이터 수신 : ", recv_json_data)
query = recv_json_data['Query']
# 의도 파악
intent_predict = intent.predict_class(query)
intent_name = intent.labels[intent_predict]
# 개체명 파악
ner_predicts = ner.predict(query)
ner_tags = ner.predict_tags(query)
# 답변 검색
try:
f = FindAnswer(db)
answer_text, answer_image = f.search(intent_name, ner_tags)
answer = f.tag_to_word(ner_predicts, answer_text)
except:
answer = "죄송해요 무슨 말인지 모르겠어요. 조금 더 공부할게요."
answer_image = None
send_json_data_str = {
"Query": query,
"Answer": answer,
"AnswerImageUrl": answer_image,
"Intent": intent_name,
"NER": str(ner_predicts)
}
message = json.dumps(send_json_data_str) # json 객체를 전송 가능한 문자열로 변환
conn.send(message.encode()) # 응답 전송
except Exception as ex:
print(ex)
finally:
if db is not None: # db 연결 끊기
db.close()
conn.close()
if __name__ == '__main__':
# 질문/답변 학습 db 연결 객체 생성
db = Database(
host=DB_HOST, user=DB_USER, password=DB_PASSWORD, db_name=DB_NAME
)
print("DB 접속")
# 봇 서버 동작
port = 5050
listen = 100
bot = BotServer(port, listen)
bot.create_sock()
print("bot start")
while True:
conn, addr = bot.ready_for_client()
params = {
"db": db
}
client = threading.Thread(target=to_client, args=(
conn, # 클라이언트 연결 소켓
addr, # 클라이언트 연결 주소 정보
params # 스레드 함수 파라미터
))
client.start() # 스레드 시작
챗봇 테스트 클라이언트 프로그램
앞서 구현한 챗봇 엔진 서버가 제대로 작동하는지 테스트해보자. /test 디렉터리에 chatbot_client_test.py 파일을 생성하여 다음과 같이 코드를 작성한다.
import socket
import json
# 챗봇 엔진 서버 접속 정보
host = "127.0.0.1" # 챗봇 엔진 서버 IP 주소
port = 5050 # 챗봇 엔진 서버 통신 포트
# 클라이언트 프로그램 시작
while True:
print("질문 : ")
query = input() # 질문 입력
if query == "exit":
exit(0)
print("=" * 40)
# 챗봇 엔진 서버 연결
mySocket = socket.socket()
mySocket.connect((host, port))
# 챗봇 엔진 질의 요청
json_data = {
'Query': query,
'BotType': "MyService"
}
message = json.dumps(json_data)
mySocket.send(message.encode())
# 챗봇 엔진 답변 출력
data = mySocket.recv(2048).decode()
ret_data = json.loads(data) # json 형태 문자열을 json 객체로 변환
print("답변 : ")
print(ret_data['Answer'])
print("\n")
# 챗봇 엔진 서버 연결 소켓 닫기
mySocket.close()
다음과 같이 결과가 나온다면 제대로 구현된 것이다. 학습 정도에 따라 완벽히 똑같은 결과가 나오지 않고 몇 개는 의도와 다르게 표현될 수 있다.
출처
조경래 님이 집필하신 처음 배우는 딥러닝 챗봇 교재를 참고하여 글을 작성하였다.
예제 또한 저자 분의 Github 에서 발췌하였다.
Leave a comment