Categories
程式開發

Serverless 實戰:如何結合NLP實現文本摘要和關鍵詞提取?


對文本進行自動摘要的提取和關鍵詞的提取,屬於自然語言處理的範疇。提取摘要的一個好處是可以讓閱讀者通過最少的信息判斷出這個文章對自己是否有意義或者價值,是否需要進行更加詳細的閱讀;而提取關鍵詞的好處是可以讓文章與文章之間產生關聯,同時也可以讓讀者通過關鍵詞快速定位到和該關鍵詞相關的文章內容。

文本摘要和關鍵詞提取都可以和傳統的CMS進行結合,通過對文章/新聞等發布功能進行改造,同步提取關鍵詞和摘要,放到HTML頁面中作為Description和Keyworks。這樣做在一定程度上有利於搜索引擎收錄,屬於SEO優化的範疇。

關鍵詞提取

關鍵詞提取的方法很多,但是最常見的應該就是tf-idf了。

通過jieba實現基於tf-idf關鍵詞提取的方法:

jieba.analyse.extract_tags(text, topK=5, withWeight=False, allowPOS=('n', 'vn', 'v'))

文本摘要

文本摘要的方法也有很多,如果從廣義上來劃分,包括提取式和生成式。其中提取式就是在文章中通過TextRank等算法,找出關鍵句然後進行拼裝,形成摘要,這種方法相對來說比較簡單,但是很難提取出真實的語義等;另一種方法是生成式,通過深度學習等方法,對文本語義進行提取再生成摘要。

如果簡單理解,提取式方式生成的摘要,所有句子來自原文,而生成式方法則是獨立生成的。

為了簡化難度,本文將採用提取式來實現文本摘要功能,通過SnowNLP第三方庫,實現基於TextRank的文本摘要功能。我們以《海底兩萬里》部分內容作為原文,進行摘要生成:

原文:

這些事件發生時,我剛從美國內布拉斯加州的貧瘠地區做完一項科考工作回來。我當時是巴黎自然史博物館的客座教授,法國政府派我參加這次考察活動。我在內布拉斯加州度過了半年時間,收集了許多珍貴資料,滿載而歸,3月底抵達紐約。我決定5月初動身回法國。於是,我就抓緊這段候船逗留時間,把收集到的礦物和動植物標本進行分類整理,可就在這時,斯科舍號出事了。
我對當時的街談巷議自然瞭如指掌,再說了,我怎能聽而不聞、無動於衷呢?我把美國和歐洲的各種報刊讀了又讀,但未能深入了解真相。神秘莫測,百思不得其解。我左思右想,搖擺於兩個極端之間,始終形不成一種見解。其中肯定有名堂,這是不容置疑的,如果有人表示懷疑,就請他們去摸一摸斯科舍號的傷口好了。
我到紐約時,這個問題正炒得沸反盈天。某些不學無術之徒提出設想,有說是浮動的小島,也有說是不可捉摸的暗礁,不過,這些個假設通通都被推翻了。很顯然,除非這暗礁腹部裝有機器,不然的話,它怎能如此快速地轉移呢?
同樣的道理,說它是一塊浮動的船體或是一堆大船殘片,這種假設也不能成立,理由仍然是移動速度太快。
那麼,問題只能有兩種解釋,人們各持己見,自然就分成觀點截然不同的兩派:一派說這是一個力大無比的怪物,另一派說這是一艘動力極強的“潛水船”。
哦,最後那種假設固然可以接受,但到歐美各國調查之後,也就難以自圓其說了。有哪個普通人會擁有如此強大動力的機械?這是不可能的。他在何地何時叫何人製造了這麼個龐然大物,而且如何能在建造中做到風聲不走漏呢?
看來,只有政府才有可能擁有這種破壞性的機器,在這個災難深重的時代,人們千方百計要增強戰爭武器威力,那就有這種可能,一個國家瞞著其他國家在試制這類駭人聽聞的武器。繼夏斯勃步槍之後有水雷,水雷之後有水下撞錘,然後魔道攀升反應,事態愈演愈烈。至少,我是這樣想的。

通過SnowNLP提供的算法:

from snownlp import SnowNLP

text = "上面的原文内容,此处省略"
s = SnowNLP(text)
print("。".join(s.summary(5)))

輸出結果:

自然就分成观点截然不同的两派:一派说这是一个力大无比的怪物。这种假设也不能成立。我到纽约时。说它是一块浮动的船体或是一堆大船残片。另一派说这是一艘动力极强的“潜水船”

初步來看,效果並不是很好,接下來我們自己計算句子權重,實現一個簡單的摘要功能,這個就需要jieba

import re
import jieba.analyse
import jieba.posseg


class TextSummary:
    def __init__(self, text):
        self.text = text

    def splitSentence(self):
        sectionNum = 0
        self.sentences = []
        for eveSection in self.text.split("n"):
            if eveSection:
                sentenceNum = 0
                for eveSentence in re.split("!|。|?", eveSection):
                    if eveSentence:
                        mark = []
                        if sectionNum == 0:
                            mark.append("FIRSTSECTION")
                        if sentenceNum == 0:
                            mark.append("FIRSTSENTENCE")
                        self.sentences.append({
                            "text": eveSentence,
                            "pos": {
                                "x": sectionNum,
                                "y": sentenceNum,
                                "mark": mark
                            }
                        })
                        sentenceNum = sentenceNum + 1
                sectionNum = sectionNum + 1
                self.sentences[-1]["pos"]["mark"].append("LASTSENTENCE")
        for i in range(0, len(self.sentences)):
            if self.sentences[i]["pos"]["x"] == self.sentences[-1]["pos"]["x"]:
                self.sentences[i]["pos"]["mark"].append("LASTSECTION")

    def getKeywords(self):
        self.keywords = jieba.analyse.extract_tags(self.text, topK=20, withWeight=False, allowPOS=('n', 'vn', 'v'))

    def sentenceWeight(self):
        # 计算句子的位置权重
        for sentence in self.sentences:
            mark = sentence["pos"]["mark"]
            weightPos = 0
            if "FIRSTSECTION" in mark:
                weightPos = weightPos + 2
            if "FIRSTSENTENCE" in mark:
                weightPos = weightPos + 2
            if "LASTSENTENCE" in mark:
                weightPos = weightPos + 1
            if "LASTSECTION" in mark:
                weightPos = weightPos + 1
            sentence["weightPos"] = weightPos

        # 计算句子的线索词权重
        index = ["总之", "总而言之"]
        for sentence in self.sentences:
            sentence["weightCueWords"] = 0
            sentence["weightKeywords"] = 0
        for i in index:
            for sentence in self.sentences:
                if sentence["text"].find(i) >= 0:
                    sentence["weightCueWords"] = 1

        for keyword in self.keywords:
            for sentence in self.sentences:
                if sentence["text"].find(keyword) >= 0:
                    sentence["weightKeywords"] = sentence["weightKeywords"] + 1

        for sentence in self.sentences:
            sentence["weight"] = sentence["weightPos"] + 2 * sentence["weightCueWords"] + sentence["weightKeywords"]

    def getSummary(self, ratio=0.1):
        self.keywords = list()
        self.sentences = list()
        self.summary = list()

        # 调用方法,分别计算关键词、分句,计算权重
        self.getKeywords()
        self.splitSentence()
        self.sentenceWeight()

        # 对句子的权重值进行排序
        self.sentences = sorted(self.sentences, key=lambda k: k['weight'], reverse=True)

        # 根据排序结果,取排名占前ratio%的句子作为摘要
        for i in range(len(self.sentences)):
            if i < ratio * len(self.sentences):
                sentence = self.sentences[i]
                self.summary.append(sentence["text"])

        return self.summary

這段代碼主要是通過tf-idf實現關鍵詞提取,然後通過關鍵詞提取對句子盡心權重賦予,最後獲得到整體的結果,運行:

testSummary = TextSummary(text)
print("。".join(testSummary.getSummary()))

可以得到結果:

Building prefix dict from the default dictionary ...
Loading model from cache /var/folders/yb/wvy_7wm91mzd7cjg4444gvdjsglgs8/T/jieba.cache
Loading model cost 0.721 seconds.
Prefix dict has been built successfully.
看来,只有政府才有可能拥有这种破坏性的机器,在这个灾难深重的时代,人们千方百计要增强战争武器威力,那就有这种可能,一个国家瞒着其他国家在试制这类骇人听闻的武器。于是,我就抓紧这段候船逗留时间,把收集到的矿物和动植物标本进行分类整理,可就在这时,斯科舍号出事了。同样的道理,说它是一块浮动的船体或是一堆大船残片,这种假设也不能成立,理由仍然是移动速度太快

我們可以看到,整體效果要比剛才的好一些。

發布API

通過Serverless架構,將上面代碼進行整理,並發布。

代碼整理結果:

import re, json
import jieba.analyse
import jieba.posseg


class NLPAttr:
    def __init__(self, text):
        self.text = text

    def splitSentence(self):
        sectionNum = 0
        self.sentences = []
        for eveSection in self.text.split("n"):
            if eveSection:
                sentenceNum = 0
                for eveSentence in re.split("!|。|?", eveSection):
                    if eveSentence:
                        mark = []
                        if sectionNum == 0:
                            mark.append("FIRSTSECTION")
                        if sentenceNum == 0:
                            mark.append("FIRSTSENTENCE")
                        self.sentences.append({
                            "text": eveSentence,
                            "pos": {
                                "x": sectionNum,
                                "y": sentenceNum,
                                "mark": mark
                            }
                        })
                        sentenceNum = sentenceNum + 1
                sectionNum = sectionNum + 1
                self.sentences[-1]["pos"]["mark"].append("LASTSENTENCE")
        for i in range(0, len(self.sentences)):
            if self.sentences[i]["pos"]["x"] == self.sentences[-1]["pos"]["x"]:
                self.sentences[i]["pos"]["mark"].append("LASTSECTION")

    def getKeywords(self):
        self.keywords = jieba.analyse.extract_tags(self.text, topK=20, withWeight=False, allowPOS=('n', 'vn', 'v'))
        return self.keywords

    def sentenceWeight(self):
        # 计算句子的位置权重
        for sentence in self.sentences:
            mark = sentence["pos"]["mark"]
            weightPos = 0
            if "FIRSTSECTION" in mark:
                weightPos = weightPos + 2
            if "FIRSTSENTENCE" in mark:
                weightPos = weightPos + 2
            if "LASTSENTENCE" in mark:
                weightPos = weightPos + 1
            if "LASTSECTION" in mark:
                weightPos = weightPos + 1
            sentence["weightPos"] = weightPos

        # 计算句子的线索词权重
        index = ["总之", "总而言之"]
        for sentence in self.sentences:
            sentence["weightCueWords"] = 0
            sentence["weightKeywords"] = 0
        for i in index:
            for sentence in self.sentences:
                if sentence["text"].find(i) >= 0:
                    sentence["weightCueWords"] = 1

        for keyword in self.keywords:
            for sentence in self.sentences:
                if sentence["text"].find(keyword) >= 0:
                    sentence["weightKeywords"] = sentence["weightKeywords"] + 1

        for sentence in self.sentences:
            sentence["weight"] = sentence["weightPos"] + 2 * sentence["weightCueWords"] + sentence["weightKeywords"]

    def getSummary(self, ratio=0.1):
        self.keywords = list()
        self.sentences = list()
        self.summary = list()

        # 调用方法,分别计算关键词、分句,计算权重
        self.getKeywords()
        self.splitSentence()
        self.sentenceWeight()

        # 对句子的权重值进行排序
        self.sentences = sorted(self.sentences, key=lambda k: k['weight'], reverse=True)

        # 根据排序结果,取排名占前ratio%的句子作为摘要
        for i in range(len(self.sentences)):
            if i < ratio * len(self.sentences):
                sentence = self.sentences[i]
                self.summary.append(sentence["text"])

        return self.summary


def main_handler(event, context):
    nlp = NLPAttr(json.loads(event['body'])['text'])
    return {
        "keywords": nlp.getKeywords(),
        "summary": "。".join(nlp.getSummary())
    }

編寫項目serverless.yaml文件:

nlpDemo:
  component: "@serverless/tencent-scf"
  inputs:
    name: nlpDemo
    codeUri: ./
    handler: index.main_handler
    runtime: Python3.6
    region: ap-guangzhou
    description: 文本摘要/关键词功能
    memorySize: 256
    timeout: 10
    events:
      - apigw:
          name: nlpDemo_apigw_service
          parameters:
            protocols:
              - http
            serviceName: serverless
            description: 文本摘要/关键词功能
            environment: release
            endpoints:
              - path: /nlp
                method: ANY

由於項目中使用了jieba,所以在安裝的時候推薦在CentOS系統下與對應的Python版本下安裝,也可以使用我之前為了方便做的一個依賴工具:

Serverless 實戰:如何結合NLP實現文本摘要和關鍵詞提取? 1

通過sls --debug進行部署:

Serverless 實戰:如何結合NLP實現文本摘要和關鍵詞提取? 2

部署完成,可以通過PostMan進行簡單的測試:

Serverless 實戰:如何結合NLP實現文本摘要和關鍵詞提取? 3

從上圖可以看到,我們已經按照預期輸出了目標結果。至此,文本摘要/關鍵詞提取的API已經部署完成。

總結

相對來說,通過Serveless架構做API是非常容易和方便的,可實現API的插拔行,組件化,希望本文能夠給讀者更多的思路和啟發。