forked from lifelongeek/convai_sample_bot
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathCC.py
executable file
·121 lines (103 loc) · 4.43 KB
/
CC.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
import re
import chainer
import pickle
import numpy as np
import sys
#sys.path.append('../DSTC/ChatbotBaseline/egs/opensubs/tools') #bot.sh
sys.path.append('DSTC/ChatbotBaseline/egs/opensubs/tools') #main
from simpler_nlg import SimplerNLG
#import seq2seq_model
import pdb
from chainer import cuda
from nltk.tokenize import casual_tokenize
class CC:
def __init__(self, use_gpu=False, gpu=0, model_path='model/cc.opensub.bst', maxlen=20, beam=5, penalty=1, nbest=1):
print('initialize Chitchat module')
self.use_gpu = use_gpu
self.num_turn_history = 2 # 2 consecutive turns as history baseline
#pdb.set_trace()
if self.use_gpu and gpu >= 0:
cuda.check_cuda_available()
cuda.get_device_from_id(gpu).use()
self.xp = cuda.cupy
else:
#chainer.cuda.available = False
self.xp = np
# use chainer in testing mode
chainer.config.train = False
#pdb.set_trace()
# Prepare RNN model and load data
print('Loading model params from ' + model_path)
with open(model_path, 'rb') as f:
#pdb.set_trace()
self.vocab, self.model, _ = pickle.load(f)
#pdb.set_trace()
if self.use_gpu and gpu >= 0:
self.model.to_gpu()
# report data summary
print('vocabulary size = %d' % len(self.vocab))
self.vocablist = sorted(self.vocab.keys(), key=lambda s: self.vocab[s])
# generate sentences
self.unk = self.vocab['<unk>']
self.eos = self.vocab['<eos>']
self.state = None
self.maxlen = maxlen
self.beam = beam
self.penalty = penalty
self.nbest = nbest
def get_reply(self, history_context, history_reply, message=""):
# Note : history_context & history_reply : collections.deque
#print('cc0')
history_context_text = ""
if (len(history_context) >= self.num_turn_history):
#print(history_context[0])
#print(history_context[1])
for i in range(self.num_turn_history):
#print('history_context')
#print(len(history_context) + i - self.num_turn_history)
history_context_text += history_context[len(history_context) + i - self.num_turn_history] + " "
#print('cc01')
history_reply_text = ""
if (len(history_reply) >= self.num_turn_history):
for i in range(self.num_turn_history):
history_reply_text += history_reply[len(history_reply) + i - self.num_turn_history] + " "
# print('(CC) history_context = ' + history_context_text) # Test PASS
# print('(CC) history_reply = ' + history_reply_text) # Test PASS
# print('(CC) message = ' + message) # Test PASS
#print('cc02')
sentence = []
for token in casual_tokenize(message, preserve_case=False, reduce_len=True):
# make a space before apostrophe
token = re.sub(r'^([a-z]+)\'([a-z]+)$', '\\1 \'\\2', token)
for w in token.split():
sentence.append(self.vocab[w] if w in self.vocab else self.unk)
#print('cc1')
x_data = np.array(sentence, dtype=np.int32)
#print('cc2')
x = chainer.Variable(self.xp.asarray(x_data))
#print('cc3')
#self.state exists
"""
besthyps, self.state = self.model.generate(self.state, x, self.eos, self.eos, unk=self.unk,
maxlen=self.maxlen,
beam=self.beam,
penalty=self.penalty,
nbest=self.nbest)
"""
#self.state = None
besthyps, self.state = self.model.generate(None, x, self.eos, self.eos, unk=self.unk,
maxlen=self.maxlen,
beam=self.beam,
penalty=self.penalty,
nbest=self.nbest)
#print('cc4')
reply = []
#print('cc5')
for w in besthyps[0][0]:
if w != self.eos:
reply.append(self.vocablist[w])
#print('cc6')
return SimplerNLG.realise(reply)
if __name__ == "__main__":
cc = CC()
print(cc.get_reply([], [], "Hi."))