基於檢索的智能問答

基於檢索的智能問答。目前使用了簡單詞彙對比、詞性權重、詞向量3種類似度計算模式。輸入符合格式的QA文本文件便可馬上使用。
 

程序版本和依賴庫

使用 python3 運行
jieba 分詞使用的庫
gensim  詞向量使用的庫，若是使用詞向量vec模式，則須要載入

依賴的文件

若是使用詞向量vec模式，須要下載3個文件：Word60.model，Word60.model.syn0.npy，Word60.model.syn1neg.npy
下載地址：http://pan.baidu.com/s/1kURNutT 密碼：1tq1

QA文件

QA文件包含想要告知用戶的問答內容。
QA文件必須是UTF-8的無bom格式的文本文件。

註釋：註釋文字由#開頭。（整個一行都是註釋內容）

問答塊格式以下：
【問題】問題標題（能夠有1或多個，至少有1個。必須由"【問題】"開頭。）
答案內容（能夠有多行，必須緊跟着上面的【問題】，多行答案中間不能有空白的行。）
多個問答塊之間能夠用空白行分割
 

主程序qa.py

直接運行該文件，便可進行問答。你能夠載入本身的QA文件，請保證QA文件格式正確。
robot.answer(inputtxt,'simple_POS') 可得出輸入問題的返回答案。
simType參數有以下模式：
simple：簡單的對比相同詞彙數量，獲得句子類似度
simple_POS：簡單的對比相同詞彙數量,並對詞性乘以不一樣的權重，獲得句子類似度
vec：用詞向量計算類似度,並對詞性乘以不一樣的權重，獲得句子類似度
all：調試模式，把以上幾種模式的結果都顯示出來，方便對比和調試

utils.py

import logging
from os.path import join, dirname


POS_WEIGHT = {
    "Ag": 1,  # 形語素
    "a": 0.5,  # 形容詞
    "ad": 0.5,  # 副形詞
    "an": 1,  # 名形詞
    "b": 1,  # 區別詞
    "c": 0.2,  # 連詞
    "dg": 0.5,  # 副語素
    "d": 0.5,  # 副詞
    "e": 0.5,  # 嘆詞
    "f": 0.5,  # 方位詞
    "g": 0.5,  # 語素
    "h": 0.5,  # 前接成分
    "i": 0.5,  # 成語
    "j": 0.5,  # 簡稱略語
    "k": 0.5,  # 後接成分
    "l": 0.5,  # 習用語
    "m": 0.5,  # 數詞
    "Ng": 1,  # 名語素
    "n": 1,  # 名詞
    "nr": 1,  # 人名
    "ns": 1,  # 地名
    "nt": 1,  # 機構團體
    "nz": 1,  # 其餘專名
    "o": 0.5,  # 擬聲詞
    "p": 0.3,  # 介詞
    "q": 0.5,  # 量詞
    "r": 0.2,  # 代詞
    "s": 1,  # 處所詞
    "tg": 0.5,  # 時語素
    "t": 0.5,  # 時間詞
    "u": 0.5,  # 助詞
    "vg": 0.5,  # 動語素
    "v": 1,  # 動詞
    "vd": 1,  # 副動詞
    "vn": 1,  # 名動詞
    "w": 0.01,  # 標點符號
    "x": 0.5,  # 非語素字
    "y": 0.5,  # 語氣詞
    "z": 0.5,  # 狀態詞
    "un": 0.3  # 未知詞
}


def get_logger(name, logfile=None):
    """
    name: logger 的名稱，建議使用模塊名稱
    logfile: 日誌記錄文件，如無則輸出到標準輸出
    """
    formatter = logging.Formatter(
        '[%(levelname)1.1s %(asctime)s %(module)s:%(lineno)d] %(message)s',
        datefmt='%m/%d/%Y %I:%M:%S'
    )

    if not logfile:
        handler = logging.StreamHandler()
    else:
        handler = logging.FileHandler(logfile)

    handler.setFormatter(formatter)

    logger = logging.getLogger(name)
    logger.addHandler(handler)
    logger.setLevel(logging.DEBUG)

    return logger


def similarity(a, b, method='simple', pos_weight=None, embedding=None):
    """a 和 b 是同類型的可迭代對象，好比都是詞的 list"""
    if not a or not b:
        return 0

    pos_weight = pos_weight or POS_WEIGHT
    if method == 'simple':
        # 詞重疊率
        return len(set(a) & set(a)) / len(set(a))

    elif method == 'simple_pos':
        sim_weight = 0
        for word, pos in set(a):
            sim_weight += pos_weight.get(pos, 1) if word in b else 0

        total_weight = sum(pos_weight.get(pos, 1) for _, pos in set(a))
        return sim_weight / total_weight if total_weight > 0 else 0

    elif method == 'vec' and embedding:
        # 詞向量+詞性權重
        sim_weight = 0
        total_weight = 0
        for word, pos in a:
            if word not in embedding.wv.index2word:
                continue

            # 詞性權重
            cur_weight = pos_weight.get(pos, 1)
            # 最大的詞向量類似度
            max_word_sim = max(embedding.similarity(bword, word) for bword in b)
            # 詞性權重*最大的詞向量類似度
            sim_weight += cur_weight * max_word_sim
            # 詞性權重之和
            total_weight += cur_weight

        # 返回 詞性權重*最大的詞向量類似度/詞性權重之和
        return sim_weight / total_weight if total_weight > 0 else 0

qs_a.txt

【問題】我己簽約怎麼沒有放款？
【問題】已經簽約何時放款
【問題】簽約成功何時放款
【問題】你好，我昨天4.20簽約的，款怎麼一直沒有到？
【問題】請問簽約了要多久放款
【問題】簽約後，還須要等多長時間
【問題】簽約後多久下款
【問題】我想問一下，簽約到放款要多久
【問題】何時放款
簽約以後總部會對您的合同進行最後一個環節審覈，審覈都經過纔會放款。簽約後審覈的時效爲1-3個工做日左右


【問題】提早還款
【問題】我要提早還款
【問題】申請提早還款
【問題】我想了解提早還款
【問題】如何提早還款
【問題】提早還款怎麼辦 我想提早還款，應該怎麼操做
提早還款1.還款日前三個工做日與客戶經理聯繫2.利息截止到當期，服務費減免 您要辦理提早結清，您提早三個工做日聯繫門店，在您還款日先後辦理不了


【問題】你好初審額度已經出面籤也簽了還須要等多久
【問題】審批結果要多久？
【問題】提交申請了多久審覈
【問題】請問審覈須要多久？
【問題】審覈通常要幾天
【問題】撒時候放款
【問題】審覈總共有幾個環節
客戶審批流程須要通過三個環節：第一環節材料審覈（資料齊全，符合標準，且不須要實地徵信）時效須要3個工做日左右；第二環節面審(簽署合同)，時效1個工做日左右；第三環節合同審覈，時效1-3個工做日左右。

qa.py

import os
import time
import logging
from collections import deque

import jieba
import jieba.posseg as pseg

from utils import get_logger
from utils import similarity

jieba.dt.tmp_dir = "./"
jieba.default_logger.setLevel(logging.ERROR)
logger = get_logger('qa', logfile="qa.log")


class Repository(object):
    """
    知識庫類
    a是答案（必須是1給）, q是問題（1個或多個）
    用以存放處理以後的知識庫形式
    """
    def __init__(self, q):
        self.q = [q]
        self.a = ""
        self.sim = 0
        self.q_vec = []
        self.q_word = []

    def __str__(self):
        return 'q=' + str(self.q) + '\na=' + str(self.a) + '\nq_word=' + str(self.q_word) + '\nq_vec=' + str(self.q_vec)


class QA(object):
    def __init__(self, zhishitxt, lastTxtLen=10, usedVec=False):
        # usedVec 若是是True 在初始化時會解析詞向量，加快計算句子類似度的速度
        self.lastTxt = deque([], lastTxtLen)
        self.zhishitxt = zhishitxt
        self.usedVec = usedVec
        self.reload()

    def load_qa(self):
        print('問答知識庫開始載入')
        self.zhishiku = []
        with open(self.zhishitxt, encoding='utf-8') as f:
            txt = f.readlines()
            abovetxt = 0  # 上一行的種類： 0空白/註釋  1答案   2問題
            for t in txt:  # 讀取FAQ文本文件
                t = t.strip()
                if not t or t.startswith('#'):
                    abovetxt = 0
                elif abovetxt != 2:
                    if t.startswith('【問題】'):  # 輸入第一個問題
                        self.zhishiku.append(Repository(t[4:]))
                        abovetxt = 2
                    else:  # 輸入答案文本（非第一行的）
                        self.zhishiku[-1].a += '\n' + t
                        abovetxt = 1
                else:
                    if t.startswith('【問題】'):  # 輸入問題（非第一行的）
                        self.zhishiku[-1].q.append(t[4:])
                        abovetxt = 2
                    else:  # 輸入答案文本
                        self.zhishiku[-1].a += t
                        abovetxt = 1

        for t in self.zhishiku:
            for question in t.q:
                t.q_word.append(set(jieba.cut(question)))

    def load_embedding(self):
        from gensim.models import Word2Vec
        # 若是不存在詞向量文件，則不使用詞向量
        if not os.path.exists('Word60.model'):
            self.vecModel = None
            return

        # 載入60維的詞向量(Word60.model，Word60.model.syn0.npy，Word60.model.syn1neg.npy）
        self.vecModel = Word2Vec.load('Word60.model')
        for t in self.zhishiku:
            t.q_vec = []
            for question in t.q_word:
                t.q_vec.append({t for t in question if t in self.vecModel.wv.index2word})

    def reload(self):
        self.load_qa()
        self.load_embedding()

        print('問答知識庫載入完畢')

    def maxSimTxt(self, intxt, simCondision=0.1, simType='simple'):
        """
        找出知識庫裏的和輸入句子類似度最高的句子
        simType=simple, simple_POS, vec
        """
        self.lastTxt.append(intxt)
        if simType not in ('simple', 'simple_pos', 'vec'):
            return 'error:  maxSimTxt的simType類型不存在: {}'.format(simType)

        # 若是沒有加載詞向量，那麼降級成 simple_pos 方法
        embedding = self.vecModel
        if simType == 'vec' and not embedding:
            simType = 'vec'

        for t in self.zhishiku:
            questions = t.q_vec if simType == 'vec' else t.q_word
            in_vec = jieba.lcut(intxt) if simType == 'simple' else pseg.lcut(intxt)

            t.sim = max(similarity(in_vec, question, method=simType, embedding=embedding) for question in questions)
        maxSim = max(self.zhishiku, key=lambda x: x.sim)
        logger.info('maxSim=' + format(maxSim.sim, '.0%'))

        if maxSim.sim < simCondision:
            return '抱歉，我沒有理解您的意思。請您詢問有關業務的話題。'

        return maxSim.a

    def answer(self, intxt, simType='simple'):
        """simType=simple, simple_POS, vec, all"""
        if not intxt:
            return ''

        if simType == 'all':  # 用於測試不一樣類型方法的準確度，返回空文本
            for method in ('simple', 'simple_pos', 'vec'):
                outtext = 'method:\t' + self.maxSim(intxt, simType=method)
                print(outtext)

            return ''
        else:
            outtxt = self.maxSimTxt(intxt, simType=simType)
            # 輸出回覆內容，並計入日誌
        return outtxt


if __name__ == '__main__':
    robot = QA('qs_a.txt', usedVec=True)
    while True:
        # simType=simple, simple_pos, vec, all
        print('回覆：' + robot.answer(input('輸入：'), 'vec') + '\n')

詞向量文件下載地址

連接：https://pan.baidu.com/s/1c7V91VcWbHPBFIfmtWGb2g 密碼：mgps

知識庫形式

q=['我己簽約怎麼沒有放款？', '已經簽約何時放款', '簽約成功何時放款', '你好，我昨天4.20簽約的，款怎麼一直沒有到？', '請問簽約了要多久放款', '簽約後，還須要等多長時間', '簽約後多久下款', '我想問一下，簽約到放款要多久', '何時放款']
a=簽約以後總部會對您的合同進行最後一個環節審覈，審覈都經過纔會放款。簽約後審覈的時效爲1-3個工做日左右
q_word=[{'簽約', '我己', '放款', '怎麼', '沒有', '？'}, {'簽約', '已經', '什麼', '放款', '時候'}, {'簽約', '成功', '什麼', '放款', '時候'}, {'一直', '簽約', '你好', '的', '4.20', '，', '怎麼', '沒有', '？', '我', '到', '昨天', '款'}, {'簽約', '要', '請問', '放款', '了', '多久'}, {'簽約', '等', '後', '須要', '，', '多長時間', '還'}, {'簽約', '下款', '多久', '後'}, {'簽約', '要', '多久', '，', '放款', '問', '想', '我', '到', '一下'}, {'什麼', '放款', '時候'}]
q_vec=[{'簽約', '我己', '放款', '怎麼', '沒有', '？'}, {'簽約', '已經', '什麼', '放款', '時候'}, {'簽約', '成功', '什麼', '放款', '時候'}, {'一直', '簽約', '你好', '的', '4.20', '，', '怎麼', '沒有', '？', '我', '到', '昨天', '款'}, {'簽約', '要', '請問', '放款', '了', '多久'}, {'簽約', '等', '後', '須要', '，', '多長時間', '還'}, {'簽約', '下款', '多久', '後'}, {'簽約', '要', '，', '放款', '問', '想', '一下', '我', '到', '多久'}, {'什麼', '放款', '時候'}]

• 造成知識庫
• 將問題分詞
• pseg.lcut 分詞帶詞性
• 帶詞性權重的詞重疊率
• 詞性權重
• 詞向量類似度
• 詞性權重*最大的詞向量類似度/詞性權重之和