前言

「问答机器人」在我们日常生活中并不少见到 ：像是一些电商客服、智能问诊、技术支持等人工输入与沟通界面的场景下，机器人“智能”问答系统一定程度上可以在无需人力、不需要耗费终端用户心智去做知识库、商品搜索、科室选择等等的情况下实时给出问题答案。

问答机器人系统背后的技术有多重可能：

• 基于检索，全文搜索接近的问题
• 基于机器学习阅读理解
• 基于知识图谱（Knowledge-Based Question Answering system: KBQA）
• 其他

基于知识图谱构建问答系统在以下三个情况下很有优势：

• 对于领域类型是结构化数据场景：电商、医药、系统运维（微服务、服务器、事件）、产品支持系统等，其中作为问答系统的参考对象已经是结构化数据；
• 问题的解答过程涉及多跳查询，比如“姚明的妻子今年是本命年吗？”，“你们家的产品 A 和 A+ 的区别是什么？”；
• 为了解决其他需求（风控、推荐、管理），已经构建了图结构数据、知识图谱的情况。

为了方便读者最快速了解如何构建 KBQA 系统，我写了非常简陋的小 KBQA 项目，在本文中，我会带领大家从头到尾把它搭起来。

💡：这个小项目叫做 Siwi，它的代码就在 GitHub 上：github.com/wey-gu/nebula-siwi

Siwi 的发音是：/ˈsɪwi/ 或者叫：思二为 ，它是一个能解答 NBA 相关问题的机器人。

我们开始吧。

鸟瞰 TL;DR

KBQA 用一句话说就是把问题解析、转换成在知识图谱中的查询，查询得到结果之后进行筛选、翻译成结果（句子、卡片或者任何方便人理解的答案格式）。

💡：知识图谱的构建实际上是非常重要的过程，在本文中，我们专注在串起来 KBQA 系统的骨架，我们假设需求是基于一个已经有的图谱之上，为其增加一个 QA 系统。

「问题到图谱查询的转换」有不同的方法可以实现。

• 可以是对语义进行分析：理解问题的意图，针对不同意图匹配最可能的问题类型，从而构建这个类型问题的图谱查询，查得结果；
• 也可以是基于信息的抽取：从问题中抽取主要的实体，在图谱中获取实体的所有知识、关系条目（子图），再对结果根据问题中的约束条件匹配、排序选择结果。

💡：美团技术团队在这篇文章里分享了他们的真实世界实践，下图是美团结合了机器学习和 NLP 的方案。

而在 Siwi 里，我们一切从简，单独选择了语义分析这条路，它的特点是需要人为去标注或者编码一些问题类型的查询方式，但实际上在大多数场景下，尤其单一领域图谱的场景下反而是轻量却效果不差的方案，也是一个便于新手理解 KBQA 的合适的入门方式。

除了核心的问答部分，我还为 Siwi 增加了语音识别和语音回答（感谢浏览器接口标准的发展）的功能，于是，这个项目的结构和问答调用流程就是这样的了：一个语音问题自上而下分别经过三个部分：

• 基于网页的 Siwi Frontend 语音、文字问答界面
• Python Flask 实现的 Siwi Backend/API 系统
• Nebula Graph 开源分布式高性能图数据库之上的知识图谱

┌────────────────┬──────────────────────────────────────┐
│                │                                      │
│                │  Speech                              │
│     ┌──────────▼──────────┐                           │
│     │            Frontend │   Siwi, /ˈsɪwi/           │
│     │ Web_Speech_API      │   A PoC of                │
│     │                     │   Dialog System           │
│     │ Vue.JS              │   With Graph Database     │
│     │                     │   Backed Knowledge Graph  │
│     └──────────┬──────────┘                           │
│                │  Sentence                            │
│   ┌────────────┼──────────────────────────────┐       │
│   │            │              Backend         │       │
│   │ ┌──────────▼──────────┐                   │       │
│   │ │ Web API, Flask      │   ./app/          │       │
│   │ └──────────┬──────────┘                   │       │
│   │            │  Sentence    ./bot/          │       │
│   │ ┌──────────▼──────────┐                   │       │
│   │ │ Intent matching,    │   ./bot/classifier│       │
│   │ │ Symentic Processing │                   │       │
│   │ └──────────┬──────────┘                   │       │
│   │            │  Intent, Entities            │       │
│   │ ┌──────────▼──────────┐                   │       │
│   │ │ Intent Actor        │   ./bot/actions   │       │
│   └─┴──────────┬──────────┴───────────────────┘       │
│                │  Graph Query                         │
│     ┌──────────▼──────────┐                           │
│     │ Graph Database      │    Nebula Graph           │
│     └─────────────────────┘                           │
└───────────────────────────────────────────────────────┘

💡：图数据库相比于其他知识图谱存储系统来说，因为其设计专注于数据内的数据关系，非常擅长实时获取海量数据下实体之间的复杂关联关系。

Nebula Graph 的原生分布式设计和 share-nothing 架构使得它擅长于巨大数据量和高并发读写的场景，加上它的开源社区特别活跃，已经被国内很多团队用于支撑生产上的各种业务，这里有一些他们分享的选型、落地实践。

知识图谱

Siwi 构建于一个篮球相关的知识图谱之上，它其实是 Siwi 采用的开源分布式图数据库 Nebula Graph 社区的官方文档里的示例数据集。

在这个非常简单的图谱之中，只有两种点：

• player，球员
• team，球队

两种关系：

• serve 服役于（比如：姚明 -服役于-> 休斯顿火箭）
• follow 关注 （比如：姚明 -关注-> 奥尼尔）

💡：这个数据集在 Nebula 社区上有一个 在线体验 环境，任何人都无需登录，通过Nebula Graph Studio 可视化探索篮球图谱。

下图就是这个图谱的可视化探索截图，可以看到左边的中心节点勇士队（Warriors）有杜兰特（Durant）还有其他几个队员在其中服役（serve）；除了服役之外，还可以看到队员和队员之中也有关注（follow）的关系存在。

有了这个知识图谱，咱们接下来就在它之上搭一个简单的基于语法解析的 QA 系统吧😁。

Siwi-backend

┌────────────┼──────────────────────────────┐
│            │              Backend         │
│ ┌──────────▼──────────┐                   │
│ │ Web API, Flask      │   ./app/          │
│ └──────────┬──────────┘                   │
│            │  Sentence    ./bot/          │
│ ┌──────────▼──────────┐                   │
│ │ Intent matching,    │   ./bot/classifier│
│ │ Symentic Processing │                   │
│ └──────────┬──────────┘                   │
│            │  Intent, Entities            │
│ ┌──────────▼──────────┐                   │
│ │ Intent Actor        │   ./bot/actions   │
└─┴──────────┬──────────┴───────────────────┘
             │  Graph Query
  ┌──────────▼──────────┐
  │ Graph Database      │    Nebula Graph
  └─────────────────────┘

如上图的设计流程，Siwi 的后端部分需要接收问句，处理之后访问知识图谱（图数据库），然后将处理结果返回给用户。

接收 HTTP 请求(app)

对于请求，就简单地用 Flask 作为 web server 来接收 HTTP 的 POST 请求：

💡：还不熟悉 Flask 的同学，可以在 freeCodeCamp 上搜索一下，有一些不错的课程哈。

下边的代码就是告诉 Flask ：

1. 如果用户发过来 http://<server>/query 的 POST 请求，提的问题就在请求的 body 里的 question 的 Key 之下。
2. 取得问题之后，调用把请求传给  siwi_bot 的 query()，得到 answer 。

代码段：src/siwi/app/__init__.py

#...
from siwi.bot import bot

#...
@app.route("/query", methods=["POST"])
def query():
    request_data = request.get_json()
    question = request_data.get("question", "") # <----- 1.
    if question:
        answer = siwi_bot.query(
            request_data.get("question", ""))   # <----- 2.
    else:
        answer = "Sorry, what did you say?"
    return jsonify({"answer": answer})

接下来我们来实现 siwi_bot，真正处理提问的逻辑。

处理请求(bot)

│            │  Sentence    ./bot/          │
│ ┌──────────▼──────────┐                   │
│ │ Intent matching,    │   ./bot/classifier│
│ │ Symentic Processing │                   │
│ └──────────┬──────────┘                   │
│            │  Intent, Entities            │
│ ┌──────────▼──────────┐                   │
│ │ Intent Actor        │   ./bot/actions   │
└─┴──────────┬──────────┴───────────────────┘

前边提到过，KBQA 基本上是

a. 把问题解析、转换成在知识图谱中的查询

b. 查询得到结果之后进行筛选、翻译成结果

这里，我们把 a. 的逻辑放在 classifier 里，b. 的逻辑放在 actions(actor) 里。

a. HTTP 请求的问题句子 sentence 传过来，用 classifier 解析它的意图和句子实体

b. 用意图和句子实体构造 action，并链接图数据库执行，获取结果。

代码段：src/siwi/bot/bot/__init__.py

from siwi.bot.actions import SiwiActions
from siwi.bot.classifier import SiwiClassifier


class SiwiBot():
    def __init__(self, connection_pool) -> None:
        self.classifier = SiwiClassifier()
        self.actions = SiwiActions()
        self.connection_pool = connection_pool

    def query(self, sentence):
        intent = self.classifier.get(sentence) # <--- a.
        action = self.actions.get(intent)      # <--- b.
        return action.execute(self.connection_pool)

首先咱们来进一步实现一下 SiwiClassifier 吧。

语义解析(classifier)

classifier 需要在 get(sentence) 方法里将句子中的实体和句子的意图解析、分类出来。通常来说，这里是需要借助机器学习、NLP去分词、分类实现的，这里只是为了展示这个过程实际上只是各种 if/ else。

我们这里实现了三类意图的问题：

• 关系（A，B）：获得 A 和 B 在图谱中的关系路径，比如姚明和湖人队的关系是？
• 服役情况：比如乔纳森在哪里服役？
• 关注情况：比如邓肯关注了谁？

❓ 开放问题：

如果看教程的你觉得这几个问题太没意思了，这里留一个开放问题，你可以在 Siwi li 帮我们实现：「共同好友（A，B）获得 A 和 B 的一度共同好友」这个意图（或者更酷的其他句子）么？欢迎来 Github：github.com/wey-gu/nebula-siwi/ 提 PR 哦，看看谁先实现。

代码片段：src/siwi/bot/classfier/__init__.py

class SiwiClassifier():
    def get(self, sentence: str) -> dict:
        """
        Classify Sentences and Fill Slots.
        This should be done by NLP, here we fake one to demostrate
        the intent Actor --> Graph DB work flow.

        sentense:
          relation:
            - What is the relationship between Yao Ming and Lakers?
            - How does Tracy McGrady and Lakers connected?
          serving:
            - Which team had Jonathon Simmons served?
          friendship:
            - Whom does Tim Duncan follow?
            - Who are Tracy McGrady's friends?

        returns:
        {
            "entities": entities,
            "intents": intents
        }
        """
        entities = self.get_matched_entities(sentence)
        intents = self.get_matched_intents(sentence)
        return {
            "entities": entities,
            "intents": intents
        }

这里，我把匹配的规则（等价于 if else...）写在了 src/siwi/bot/test/data 之下的 YAML 文件里，这样增加 classifier 之中新的规则只需要更新这个文件就可以了：

意图识别(intent)

def load_entity_data(self) -> None:
    # load data from yaml files
    module_path = f"{ siwi.__path__[0] }/bot/test/data"
    #...
    with open(f"{ module_path }/intents.yaml", "r") as file:
        self.intents = yaml.safe_load(file)["intents"]

对于每一个意图来说：

• intents.<名字> 代表名字
• 名字之后的 action 代表后边在要实现的相应的 xxxAction 的类
• 比如 RelationshipAction 将是用来处理查询关系（A，B）这样的问题的 Action 类
• keywords 代表在句子之中匹配的关键词
• 比如问句里出现 serve，served，serving 的字眼的时候，将会匹配服役的问题

💡：写 if else 条件来对应意图是不容易的，因为不同意图不可能没有关键词相交的情况，我们的实现只是一个非常简陋、不严谨的方式。在实际场景下，训练模型去做匹配效果会更好，有意思的是，那些做的比较好的模型的输入和我们的 YAML 的格式是很类似的。

---
intents:
  fallback:
    action:
      FallbackAction
    keywords: []
  relationship:
    action:
      RelationshipAction
    keywords:
      - between
      - relation
      - relationship
      - related
      - connect
      - correlate
  serve:
    action:
      ServeAction
    keywords:
      - serve
      - served
      - serving
  friend:
    action:
      FollowAction
    keywords:
      - follows
      - followed
      - follow
      - friend
      - friends

实体识别(entity)

类似的，实体识别的部分本质上也是 if else，只不过这里利用到了**Aho–Corasick算法**来帮助搜索实体，在生产（非玩具）的情况下，应该用 NLP 里的分词的方法来做。

💡：大家可以去了解一下这个 AC自动机算法

def setup_entity_tree(self) -> None:
    self.entity_type_map.update({
        key: "player" for key in self.players.keys()
        })
    self.entity_type_map.update({
        key: "team" for key in self.teams.keys()
        })

    self.entity_tree = ahocorasick.Automaton()
    for index, entity in enumerate(self.entity_type_map.keys()):
        self.entity_tree.add_word(entity, (index, entity))
    self.entity_tree.make_automaton()

#...

def get_matched_entities(self, sentence: str) -> dict:
    """
    Consume a sentence to be matched with ahocorasick
    Returns a dict: {entity: entity_type}
    """
    _matched = []
    for item in self.entity_tree.iter(sentence):
        entities_matched.append(item[1][1])
    return {
        entity: self.entity_type_map[entity] for entity in _matched
    }

至此，我们的 SiwiClassifier.get(sentence) 已经能返回解析、分类出来的意图和实体了，这时候，它们会被传给 Actions 来让 siwi bot 知道如何去执行只是图谱的查询啦！

构造图谱查询(action)

还记得前边的 bot 代码里，最后一步，图谱查询的动作是这么被构造的：

action = self.actions.get(intent)

现在咱们就把它实现一下：

在前边提到过的 intents.yaml 里获取这个意图里配置的意图的类名称

导入相应的 Action 类

代码段：src/bot/actions/__init__.py

class SiwiActions():
    def __init__(self) -> None:
        self.intent_map = {}
        self.load_data()

    def load_data(self) -> None:
        # load data from yaml files
        module_path = f"{ siwi.__path__[0] }/bot/test/data"

        with open(f"{ module_path }/intents.yaml", "r") as file:
            self.intent_map = yaml.safe_load(file)["intents"]

    def get(self, intent: dict):
        """
        returns SiwiActionBase
        """
        if len(intent["intents"]) > 0:
            intent_name = intent["intents"][0]
        else:
            intent_name = "fallback"

        cls_name = self.intent_map.get(
            intent_name).get("action") #-------> 1.
        action_cls = getattr(          #-------> 2.
            importlib.import_module("siwi.bot.actions"), cls_name)
        action = action_cls(intent)
        return action

最后，我们来实现这个类吧，比如 RelationshipAction 对应的代码如下：

1. 根据提供的 A 和 B，构造并执行图数据库之中的 FIND PATH
2. 将 FIND PATH 的结果进行解析，通过 as_path() 方法的封装，获得 path 类型的数据，并处理一个句子返回给用户

💡：FIND PATH 就是字面意思的查找路径，这里有详细的解释哦。

class RelationshipAction(SiwiActionBase):
    """
    USE basketballplayer;
    FIND NOLOOP PATH
    FROM "player100" TO "team204" OVER * BIDIRECT UPTO 4 STEPS;
    """
    def __init__(self, intent):
        print(f"[DEBUG] RelationshipAction intent: { intent }")
        super().__init__(intent)
        try:
            self.entity_left, self.entity_right = intent["entities"]
            self.left_vid = self._vid(self.entity_left)
            self.right_vid = self._vid(self.entity_right)
        except Exception:
            print(
                f"[WARN] RelationshipAction entities recognition Failure "
                f"will fallback to FallbackAction, "
                f"intent: { intent }"
                )
            self.error = True

    def execute(self, connection_pool) -> str:
        self._error_check()
        query = (
            f'USE basketballplayer;'
            f'FIND NOLOOP PATH '
            f'FROM "{self.left_vid}" TO "{self.right_vid}" '
            f'OVER * BIDIRECT UPTO 4 STEPS;'
            )
        print(
            f"[DEBUG] query for RelationshipAction :\n\t{ query }"
            )
        with connection_pool.session_context("root", "nebula") as session:
            result = session.execute(query)        #--------------------> 1.

        if not result.is_succeeded():
            return (
                f"Something is wrong on Graph Database connection when query "
                f"{ query }"
                )

        if result.is_empty():
            return (
                f"There is no relationship between "
                f"{ self.entity_left } and { self.entity_right }"
                )
        path = result.row_values(0)[0].as_path()    #-------------------> 2.
        relationships = path.relationships()
        relations_str = self._name(
            relationships[0].start_vertex_id().as_string())
        for rel_index in range(path.length()):
            rel = relationships[rel_index]
            relations_str += (
                f" { rel.edge_name() }s "
                f"{ self._name(rel.end_vertex_id().as_string()) }")
        return (
            f"There are at least { result.row_size() } relations between "
            f"{ self.entity_left } and { self.entity_right }, "
            f"one relation path is: { relations_str }."
            )

至此，咱们就已经实现了后端的所有功能，我们可以把它启动起来试试了！

测试一下

启动图数据库

我们在 Nebula Graph 里建立（导入数据）一个篮球的知识图谱。

💡：在导入数据之前，请先部署一个 Nebula Graph 集群。最简便的部署方式是使用 Nebula-UP 这个小工具，只需要一行命令就能在 Linux 机器上同时启动一个 Nebula Graph 核心和可视化图探索工具 Nebula Graph Studio。如果你更愿意用 Docker 部署，请参考这个文档。

本文假设我们使用 Nebula-UP 来部署一个 Nebula Graph：

curl -fsSL nebula-up.siwei.io/install.sh | bash

之后，我们会看到这样的提示：

按照提示，我们可以通过这个命令进入到有 Nebula Console 的容器里：

💡：Nebula Console 是命令行访问 Nebula Graph 图数据库的客户端，支持 Linux，Windows 和 macOS，下载地址：这里

~/.nebula-up/console.sh

然后，在 # 的提示符下就表示我们进来了，我们在里边可以执行：

nebula-console -addr graphd -port 9669 -user root -p nebula

这样就表示我们连接上了 Nebula Graph 图数据库：

/ # nebula-console -addr graphd -port 9669 -user root -p nebula
Welcome to Nebula Graph!

(root@nebula) [(none)]>

在这里，我们就可以通过 nGQL 去操作 Nebula Graph，不过我们先退出来，执行 exit：

(root@nebula) [(none)]> exit

Bye root!
Fri, 31 Dec 2021 04:11:28 UTC

我们在这个容器内把基于 nGQL 语句的数据下载下来：

/ # wget https://docs.nebula-graph.io/2.0/basketballplayer-2.X.ngql

然后通过 Nebula Console 的 -f <file_path> 把数据导入进去：

nebula-console -addr graphd -port 9669 -user root -p nebula -f basketballplayer-2.X.ngql

至此，我们就启动了一个 Nebula Graph 图数据库，还在里边加载了篮球的知识图谱！

💡：还记得前边我们提到的 在线体验 环境么？现在，我们可以在这个利用 Nebula-UP 部署了 Nebula 的环境里启动自己的 Nebula Studio 啦，按照上边 Nebula-UP 的提示：http://<本机IP>:7001 就是它的地址，然后大家可以参考文档和在线体验介绍去了解更多。

启动 Siwi-backend

大家可以直接 clone 我的代码：git clone https://github.com/wey-gu/nebula-siwi/

然后安装、启动 Siwi Backend：

cd nebula-siwi

# Install dependencies
python3 -m pip install -r src/requirements.txt

# Install siwi backend
python3 -m build

# Configure Nebula Graph Endpoint
export NG_ENDPOINTS=127.0.0.1:9669

# Run Backend API server
gunicorn --bind :5000 wsgi --workers 1 --threads 1 --timeout 60

启动之后，我们可以另外开窗口，通过 cURL 去发起问题给 backend，更多细节大家可以参考 GitHub 上的 README：

至此，我们已经写好了 QA 系统的重要的代码啦，大家是不是对一个 KBQA 的构成有了更清晰的概念了呢？

接下来，我们为它增加一个界面！

Siwi-frontend

聊天界面

我们利用 Vue Bot UI 这个可爱的机器人界面的 Vue 实现可以很容易构造一个

代码段：src/siwi/frontend/src/App.vue

<template>
  <div id="app">
    <VueBotUI
      :messages="msg"
      :options="botOptions"
      :bot-typing="locking"
      :input-disable="locking"
      @msg-send="msgSender"
    />
  </div>
</template>
<script>
import { VueBotUI } from 'vue-bot-ui'

注意到那个小飞机按钮了吧，它是发出问题请求的按键，我们要在按下它的时候对后端做出请求。

访问后端

这部分用到了Axios，它是浏览器里访问其他地址的 HTTP 客户端。

1. 在按下的时候，@msg-send="msgSender" 会触发 msgSender()
2. msgSender()去构造axios.post(this.apiEndpoint, { "question": data.text }) 的请求给 Siwi 的后端
3. 后端的结果被 push() 到界面的聊天消息里，渲染出来 this.msg.push()

代码段：src/siwi/frontend/src/App.vue

<template>
  <div id="app">
    <button id="mic_btn" @click="record = !record">
        {{record?'👂':'🎙️'}}      --------------------------> 1.
    </button>

    <vue-web-speech
      v-model="record"
      @results="onResults"       --------------------------> 1.
      @unrecognized="unrecognized"
    >
    </vue-web-speech>

...
    <vue-web-speech-synth
      v-model="agentSpeak"
      :voice="synthVoice"
      :text="synthText"
      @list-voices="listVoices"  --------------------------> 4.
    />

  </div>
</template>

<script>
import { VueBotUI } from 'vue-bot-ui'
import axios from "axios";

export default {
  name: 'App',
  components: {
    VueBotUI,
  },
  onResults (data) {             -------------------------> 2.
      this.results = data;
      this.locking = true;

      this.msg.push({
        agent: "user",
        type: "text",
        text: data[0],
      });

      this.locking = true;
      console.log(data[0]);
      axios.post(this.apiEndpoint, { "question": data[0] }).then((response) => {
        console.log(response.data);

        this.msg.push({
          agent: "bot",
          type: "text",
          text: response.data.answer,
        });

        this.synthText = response.data.answer;  ----------> 3.
        this.agentSpeak = true;
      });
      this.locking = false;
    },
  }
}
</script>

现在，我们已经有了一个图形界面的机器人啦，不过，更进一步，我们可以利用现代浏览器的接口，实现语音识别和机器人说话！

语音识别

我们借助于 Vue Web Speech, 这个语音 API 的 VueJS 的绑定，可以很容易在按下 🎙️ 的时候接收人的语音，并把语音转换成文字发出去，在回答被返回之后，它（还是他/她😁？）也会把回答的句子读出来给用户。

1. record 在 🎙️ 被按下之后，变成 👂
2. 触发 onResults() 监听
3. 把返回结果发给 this.synthText 合成器，准备读出
4. <vue-web-speech-synth> 把语音读出

代码段：src/siwi/frontend/src/App.vue

<template>
  <div id="app">
    <button id="mic_btn" @click="record = !record">
        {{record?'👂':'🎙️'}} -----------------------------> 1.
    </button>

    <vue-web-speech
      v-model="record"
      @results="onResults"   -----------------------------> 1.
      @unrecognized="unrecognized"
    >
    </vue-web-speech>

...
    <vue-web-speech-synth
      v-model="agentSpeak"
      :voice="synthVoice"
      :text="synthText"
      @list-voices="listVoices" ---------------------------> 4.
    />

  </div>
</template>

<script>
import { VueBotUI } from 'vue-bot-ui'
import axios from "axios";

export default {
  name: 'App',
  components: {
    VueBotUI,
  },
  onResults (data) {                 -------------------> 2.
      this.results = data;
      this.locking = true;

      this.msg.push({
        agent: "user",
        type: "text",
        text: data[0],
      });

      this.locking = true;
      console.log(data[0]);
      axios.post(this.apiEndpoint, { "question": data[0] }).then((response) => {
        console.log(response.data);

        this.msg.push({
          agent: "bot",
          type: "text",
          text: response.data.answer,
        });

        this.synthText = response.data.answer;  ----------------------> 3.
        this.agentSpeak = true;
      });
      this.locking = false;
    },
  }
}
</script>

总结

至此，我们已经学会了搭建自己的第一个 KBQA：知识图谱驱动的问答系统。

回顾下它的代码结构：

• src/siwi 对应后端
• App 是 Flask API 处理的部分
• Bot 是处理请求、访问 Nebula Graph 的部分
• src/siwi_frontend 是前端

希望大家在这个简陋的基础之上，多多探索，做出来更加成熟的聊天机器人，欢迎你来给我邮件、留言告诉我呀，这里：https://siwei.io/about 有我的联系方式。

.
├── README.md
├── src
│   ├── siwi                        # Siwi-API Backend
│   │   ├── app                     # Web Server, take HTTP requests and calls Bot API
│   │   └── bot                     # Bot API
│   │       ├── actions             # Take Intent, Slots, Query Knowledge Graph here
│   │       ├── bot                 # Entrypoint of the Bot API
│   │       ├── classifier          # Symentic Parsing, Intent Matching, Slot Filling
│   │       └── test                # Example Data Source as equivalent/mocked module
│   └── siwi_frontend               # Browser End
│       ├── README.md
│       ├── package.json
│       └── src
│           ├── App.vue             # Listening to user and pass Questions to Siwi-API
│           └── main.js
└── wsgi.py

如果你很喜欢这样的小项目，欢迎来看看我之前的分享： 「从0-1：如何构建一个企业股权图谱系统？」哦。

💡：你知道吗，我其实借助于 Katacoda 已经为大家搭建了一个交互式体验 Siwi + Nebula 的部署的环境，如果您的网络条件够快（Katacoda服务器在国外），可以在这里点点鼠标就交互式体验它。

视频介绍： https://www.bilibili.com/video/BV1Rm4y1Q7B5

感谢用到的开源项目 ❤️

这个小项目里我们用到了好多开源的项目，非常感谢这些贡献者们的慷慨与无私，开源是不是很酷呢？

Backend

• KGQA on MedicalKG by Huanyong Liu
• Flask
• pyahocorasick created by Wojciech Muła
• PyYaml

Frontend

• VueJS for frontend framework
• Vue Bot UI, as a lovely bot UI in vue
• Vue Web Speech, for speech API vue wrapper
• Axios for browser http client
• Solarized for color scheme
• Vitesome for landing page design

Graph Database

• Nebula Graph 高性能、云原生的开源分布式图数据库

题图版权：Andy Kelly

无论是监管部门、企业还是个人，都有针对一个企业、法人做背景调查的需求，这些调查可以是法律诉讼、公开持股、企业任职等等多种多样的信息。这些背景信息可以辅助我们做商业上的重要决策，规避风险：比如根据公司的股权关系，了解是否存在利益冲突比如是否选择与一家公司进行商业往来。

在满足这样的关系分析需求的时候，我们往往面临一些挑战，比如：

1. 如何将这些数据的关联关系体现在系统之中？使得它们可以被挖掘、利用
2. 多种异构数据、数据源之间的关系可能随着业务的发展引申出更多的变化，在结构数据库中，这意味着 Schema 变更
3. 分析系统需要尽可能实时获取需要的查询结果，这通常涉及到多跳关系查询
4. 领域专家能否快速灵活、可视化获取分享信息

那么如何构建这样一个系统解决以上挑战呢？

数据存在哪里？

前提：数据集准备，为了更好的给大家演示解决这个问题，我写了一个轮子能随机生成股权结构相关的数据，生成的数据的例子在这里。

这里，我们有法人、公司的数据，更有公司与子公司之间的关系，公司持有公司股份，法人任职公司，法人持有公司股份和法人之间亲密度的关系数据。

数据存在哪里？这是一个关键的问题，这里我们剧透一下，答案是：图数据库。然后我们再简单解释一下为什么这样一个股权图谱系统跑在图数据库上是更好的。

在这样一个简单的数据模型之下，我们可以很直接的在关系型数据库中这么建模：

而这么建模的问题在于：这种逻辑关联的方式使得无论数据的关联关系查询表达、存储、还是引入新的关联关系都不是很高效。

• 查询表达不高效是因为关系型数据库是面向表结构设计的，这决定了关系查询要写嵌套的 JOIN。

这就是前边提到的挑战 1：能够表达，但是比较勉强，遇到稍微复杂的情况就变得很难。

• 存储不高效是因为表结构被设计的模式是面向数据记录，而非数据之间的关系：我们虽然习惯了将数据中实体（比如法人）和实体关联（比如持有股权 hold_sharing_relationship）以另外一个表中的记录来表达、存储起来，这逻辑上完全行得通，但是到了多跳、大量需要请求数据关系跳转的情况下，这样跨表 JOIN 的代价就成为了瓶颈。

这就是前边提到的挑战 3：无法应对多条查询的性能需要。

• 引入新的关联关系代价大，还是前边提到的，表结构下，用新的表来表达持有股权 hold_sharing_relationship这个关联关系是可行的，但是这非常不灵活、而且昂贵，它意味着我们在引入这个关系的时候限定了起点终点的类型，比如股权持有的关系可能是法人->公司，也可能是公司->公司，随着业务的演进，我们可能还需要引入政府->公司的新关系，而这些变化都需要做有不小代价的工作：改动 Schema。

这就是前边提到的挑战 2：无法应对业务上对数据关系上灵活多变的要求。

当一个通用系统无法满足不可忽视的具体需求的时候，一个新的系统就会诞生，这就是图数据库，针对这样的场景，图数据库很自然地特别针对关联关系场景去设计整个数据库：

• 面向关联关系表达的语义。（挑战 1）

如下表，我列举了一个等价的一跳查询在表结构数据库与图数据库中，查询语句的区别。大家应该可以看出“找到所有持有和 p_100 共同持有公司股份的人”这样的查询表达可以在图数据库如何自然表达，这仅仅是一条查询的区别，如果是多跳的话，他们的复杂度区分还会更明显一些。

• 将关联关系存储为物理连接，从而使得跳转查询代价最小。（挑战 3、2）

图数据之中，从点拓展（找到一个或者多个关系的另一头）出去的代价是非常小的，这因为图数据库是一个专有的系统，得益于它主要关心“图”结构的设计，查找确定的实体（比如和一个法人 A ）所有关联（可能是任职、亲戚、持有、等等关系）其他所有实体（公司、法人）这个查找的代价是 O(1) 的，因为它们在图数据库的数据机构里是真的链接在一起的。

大家可以从下表的定量参考数据一窥图数据库在这种查询下的优势，这种优势在多跳高并发情况下的区别是“能”与”不能“作为线上系统的区别，是“实时”与“离线”的区别。

在面向关联关系的数据建模和数据结构之下，引入新的实体、关联关系的代价要小很多，还是前边提到的例子：
在 Nebula Graph 图数据中引入一个新的“政府机构”类型的实体，并增加政府机构->公司的“持有股份”的关联关系相比于在非图模型的数据库中的代价小很多。

• 建模符合直觉；图数据库有面向数据连接的数据可视化能力（挑战 4）

大家在下表第二列中可以对比我们本文中进行的股权分析数据在两种数据库之中的建模的区别，尤其是在关心关联关系的场景下，我们可以感受到属性图的模型建立是很符合人类大脑直觉的，而这和大脑之中神经元的结构可能也有一些关系。

图数据库中内置的可视化工具提供了一般用户便捷理解数据关系的能力，也给领域专家用户提供了表达请求复杂数据关系的直观接口。

表结构数据库与图数据库的总体比较：

综上，在本教程里，我们将利用图数据库来进行数据存储。

图数据建模

前面在讨论数据存在哪里的时候，我们已经揭示了在图数据库中建模的方式：本质上，在这张图中，将会有两种实体：

• 人
• 公司

四种关系：

• 人 –作为亲人–>人
• 人 –作为角色–> 公司
• 人 或者 公司 –持有股份–> 公司
• 公司 –作为子机构–> 公司

这里面，实体与关系本身都可以包含更多的信息，这些信息在图数据库里就是实体、关系自身的属性。如下图表示：

• 人的属性包括 name，age
• 公司的属性包括 name，location
• 持有股份 这个关系有属性 share(份额)
• 任职这个关系有属性 role，level

数据入库

本教程中，我们使用的图数据库叫做 Nebula Graph（星云图数据库），它是一个以 Apache 2.0 许可证开源的分布式图数据库。

Nebula Graph in Github: https://github.com/vesoft-inc/nebula

在向 Nebula Graph 导入数据的时候，关于如何选择工具，请参考这篇文档和这个视频。

这里，由于数据格式是 csv 文件并且利用单机的客户端资源就足够了，我们可以选择使用 nebula-importer 来完成这个工作。

提示：在导入数据之前，请先部署一个 Nebula Graph 集群，最简便的部署方式是使用 nebula-up 这个小工具，只需要一行命令就能在 Linux 机器上同时启动一个 Nebula Graph 核心和可视化图探索工具  Nebula Graph Studio。如果你更愿意用 Docker 部署，请参考这个文档。

本文假设我们使用 Nebula-UP 来部署：

curl -fsSL nebula-up.siwei.io/install.sh | bash

这里的数据是生成器生成的，你可以按需生成任意规模随机数据集，或者选择一份生成好了的数据在这里

有了这些数据，我们可以开始导入了。

$ pip install Faker==2.0.5 pydbgen==1.0.5
$ python3 data_generator.py
$ ls -l data
total 1688
-rw-r--r--  1 weyl  staff   23941 Jul 14 13:28 corp.csv
-rw-r--r--  1 weyl  staff    1277 Jul 14 13:26 corp_rel.csv
-rw-r--r--  1 weyl  staff    3048 Jul 14 13:26 corp_share.csv
-rw-r--r--  1 weyl  staff  211661 Jul 14 13:26 person.csv
-rw-r--r--  1 weyl  staff  179770 Jul 14 13:26 person_corp_role.csv
-rw-r--r--  1 weyl  staff  322965 Jul 14 13:26 person_corp_share.csv
-rw-r--r--  1 weyl  staff   17689 Jul 14 13:26 person_rel.csv

导入工具 nebula-importer 是一个 golang 的二进制文件，使用方式就是将导入的 Nebula Graph 连接信息、数据源中字段的含义的信息写进 YAML 格式的配置文件里，然后通过命令行调用它。可以参考文档或者它的 GitHub 仓库里的例子。

这里我已经写好了准备好了一份 nebula-importer 的配置文件，在数据生成器同一个 repo 之下的这里。

最后，只需要执行如下命令就可以开始数据导入了：

注意，在写本文的时候，nebula 的新版本是 2.6.1，这里对应的 nebula-importer 是 v2.6.0，如果您出现导入错误可能是版本不匹配，可以相应调整下边命令中的版本号。

git clone https://github.com/wey-gu/nebula-shareholding-example
cp -r data_sample /tmp/data
cp nebula-importer.yaml /tmp/data/
docker run --rm -ti \
    --network=nebula-docker-compose_nebula-net \
    -v /tmp/data:/root \
    vesoft/nebula-importer:v2.6.0 \
    --config /root/nebula-importer.yaml

你知道吗？TL;DR

实际上，这份 importer 的配置里帮我们做了 Nebula Graph 之中的图建模的操作，它们的指令在下边，我们不需要手动去执行了。

CREATE SPACE IF NOT EXISTS shareholding(partition_num=5, replica_factor=1, vid_type=FIXED_STRING(10));
USE shareholding;
CREATE TAG person(name string);
CREATE TAG corp(name string);
CREATE TAG INDEX person_name on person(name(20));
CREATE TAG INDEX corp_name on corp(name(20));
CREATE EDGE role_as(role string);
CREATE EDGE is_branch_of();
CREATE EDGE hold_share(share float);
CREATE EDGE reletive_with(degree int);

图库中查询数据

Tips: 你知道吗，你也可以无需部署安装，通过 Nebula-Playground 之中，找到股权穿透来在线访问同一份数据集。

我们可以借助  Nebula Graph Studio 来访问数据，访问我们部署 Nebula-UP 的服务器地址的 7001 端口就可以了：

假设服务器地址为 192.168.8.127，则有：

• Nebula Studio 地址：192.168.8.127:7001
• Nebula Graph 地址：192.168.8.127:9669
• 默认用户名：root
• 默认密码：nebula

访问 Nebula Studio：

选择图空间: Shareholding

之后，我们就可以在里边探索比如一个公司的三跳以内的股权穿透，具体的操作可以参考：股权穿透在线 Playground 的介绍：

构建一个图谱系统

这部分的代码开源在 GitHub 上：

https://github.com/wey-gu/nebula-corp-rel-search

本项目的 Demo 也在 PyCon China 2021 上的演讲中有过展示：视频地址

在此基础之上，我们可以构建一个提供给终端用户来使用的股权查询系统了，我们已经有了图数据库作为这个图谱的存储引擎，理论上，如果业务允许，我们可以直接使用或者封装 Nebula Graph Studio 来提供服务，这完全是可行也是合规的，不过，有一些情况下，我们需要自己去实现界面、或者我们需要封装出一个 API 给上游（多端）提供图谱查询的功能。

为此，我为大家写了一个简单的实例项目，提供这样的服务，他的架构也很直接：

• 前端接受用户要查询的穿透法人、公司，按需发请求给后端，并用 D3.js 将返回结果渲染为关系图
• 后端接受前端的 API 请求，将请求转换为 Graph DB 的查询，并返回前端期待的结果

  ┌───────────────┬───────────────┐
  │               │  Frontend     │
  │               │               │
  │    ┌──────────▼──────────┐    │
  │    │ Vue.JS              │    │
  │    │ D3.JS               │    │
  │    └──────────┬──────────┘    │
  │               │  Backend      │
  │    ┌──────────┴──────────┐    │
  │    │ Flask               │    │
  │    │ Nebula-Python       │    │
  │    └──────────┬──────────┘    │
  │               │  Graph Query  │
  │    ┌──────────▼──────────┐    │
  │    │ Graph Database      │    │
  │    └─────────────────────┘    │
  │                               │
  └───────────────────────────────┘

后端服务-->图数据库

详细的数据格式分析大家可以参考这里

查询语句

我们假设用户请求的实体是 c_132 ，那么请求 1 到 3 步的关系穿透的语法是：

MATCH p=(v)-[e:hold_share|:is_branch_of|:reletive_with|:role_as*1..3]-(v2) \
WHERE id(v) IN ["c_132"] RETURN p LIMIT 100

这里边 ()包裹的是图之中的点，而[] 包裹的则是点之间的关系：边，所以：

(v)-[e:hold_share|:is_branch_of|:reletive_with|:role_as*1..3]-(v2) 之中的：

(v)-[xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx]-(v2)应该比较好理解，意思是从 v 到v2 做拓展。

现在我们介绍中间[]包裹的部分，这里，它的语义是：经由四种类型的边（:之后的是边的类型，|代表或者）通过可变的跳数：*1..3 （一跳到三跳）。

所以，简单来说整理看开，我们的拓展的路径是：从点 v 开始，经由四种关系一到三跳拓展到点v2，返回整个拓展路径 p，限制 100 个路径结果，其中 v 是 c_132。

Nebula Python Client/ SDK

我们已经知道了查询语句的语法，那么就只需要在后端程序里根据请求、通过图数据库的客户端来发出查询请求，并处理返回结构就好了。在今天的例子中，我选择使用 Python 来实现后端的逻辑，所以我用了 Nebula-python 这个库，它是 Nebula 的 Python Client。

你知道么？截至到现在，Nebula 在 GitHub 上有 Java，GO，Python，C++，Spark，Flink，Rust（未GA），NodeJS（未GA） 的客户端支持，更多的语言的客户端也会慢慢被发布哦。

下边是一个 Python Client 执行一个查询并返回结果的例子，值得注意的是，在我实现这个代码的时候，Nebula Python 尚未支持返回 JSON （通过session.execute_json()）结果，如果你要实现自己的代码，我非常推荐试试 JSON 哈，就可以不用从对象中一点点取数据了，不过借助 iPython/IDLE 这种 REPL，快速了解返回对象的结构也没有那么麻烦。

$ python3 -m pip install nebula2-python==2.5.0 # 注意这里我引用旧的记录，它是 2.5.0，
$ ipython
In [1]: from nebula2.gclient.net import ConnectionPool
In [2]: from nebula2.Config import Config
In [3]: config = Config()
   ...: config.max_connection_pool_size = 10
   ...: # init connection pool
   ...: connection_pool = ConnectionPool()
   ...: # if the given servers are ok, return true, else return false
   ...: ok = connection_pool.init([('192.168.8.137', 9669)], config)
   ...: session = connection_pool.get_session('root', 'nebula')
[2021-10-13 13:44:24,242]:Get connection to ('192.168.8.137', 9669)

In [4]: resp = session.execute("use shareholding")
In [5]: query = '''
   ...: MATCH p=(v)-[e:hold_share|:is_branch_of|:reletive_with|:role_as*1..3]-(v2) \
   ...: WHERE id(v) IN ["c_132"] RETURN p LIMIT 100
   ...: '''
In [6]: resp = session.execute(query) # Note: after nebula graph 2.6.0, we could use execute_json as well

In [7]: resp.col_size()
Out[7]: 1

In [9]: resp.row_size()
Out[10]: 100

我们往下分析看看，我们知道这个请求本质上结果是路径，它有一个 .nodes() 方法和 .relationships()方法来获得路径上的点和边：

In [11]: p=resp.row_values(22)[0].as_path()

In [12]: p.nodes()
Out[12]:
[("c_132" :corp{name: "Chambers LLC"}),
 ("p_4000" :person{name: "Colton Bailey"})]

In [13]: p.relationships()
Out[13]: [("p_4000")-[:role_as@0{role: "Editorial assistant"}]->("c_132")]

对于边来说有这些方法 .edge_name(), .properties(), .start_vertex_id(), .end_vertex_id()，这里 edge_name 是获得边的类型。

In [14]: rel=p.relationships()[0]

In [15]: rel
Out[15]: ("p_4000")-[:role_as@0{role: "Editorial assistant"}]->("c_132")

In [16]: rel.edge_name()
Out[16]: 'role_as'

In [17]: rel.properties()
Out[17]: {'role': "Editorial assistant"}

In [18]: rel.start_vertex_id()
Out[18]: "p_4000"

In [19]: rel.end_vertex_id()
Out[19]: "c_132"

对于点来说，可以用到这些方法 .tags(), properties, get_id()，这里边 tags 是获得点的类型，它在 Nebula 里叫标签tag。

这些概念可以在文档里获得更详细的解释。

In [20]: node=p.nodes()[0]

In [21]: node.tags()
Out[21]: ['corp']

In [22]: node.properties('corp')
Out[22]: {'name': "Chambers LLC"}

In [23]: node.get_id()
Out[23]: "c_132"

前端渲染点边为图

详细的分析大家也可以参考这里

为了方便实现，我们采用了 Vue.js 和 vue-network-d3（D3 的 Vue Binding）。

通过 vue-network-d3 的抽象，能看出来喂给他这样的数据，就可以把点边信息渲染成很好看的图

nodes: [
        {"id": "c_132", "name": "Chambers LLC", "tag": "corp"},
        {"id": "p_4000", "name": "Colton Bailey", "tag": "person"}],
relationships: [
        {"source": "p_4000", "target": "c_132", "properties": { "role": "Editorial assistant" }, "edge": "role_as"}]

前端<--后端

详细信息可以参考这里

我们从 D3 的初步研究上可以知道，后端只需要返回如下的 JSON 格式数据就好了

Nodes:

[{"id": "c_132", "name": "Chambers LLC", "tag": "corp"},
 {"id": "p_4000", "name": "Colton Bailey", "tag": "person"}]

Relationships:

[{"source": "p_4000", "target": "c_132", "properties": { "role": "Editorial assistant" }, "edge": "role_as"},
 {"source": "p_1039", "target": "c_132", "properties": { "share": "3.0" }, "edge": "hold_share"}]

于是，，结合前边我们用 iPython 分析 Python 返回结果看，这个逻辑大概是：

def make_graph_response(resp) -> dict:
    nodes, relationships = list(), list()
    for row_index in range(resp.row_size()):
        path = resp.row_values(row_index)[0].as_path()
        _nodes = [
            {
                "id": node.get_id(), "tag": node.tags()[0],
                "name": node.properties(node.tags()[0]).get("name", "")
                }
                for node in path.nodes()
        ]
        nodes.extend(_nodes)
        _relationships = [
            {
                "source": rel.start_vertex_id(),
                "target": rel.end_vertex_id(),
                "properties": rel.properties(),
                "edge": rel.edge_name()
                }
                for rel in path.relationships()
        ]
        relationships.extend(_relationships)
    return {"nodes": nodes, "relationships": relationships}

前端到后端的通信是 HTTP ，所以我们可以借助 Flask，把这个函数封装成一个 RESTful API：

前端程序通过 HTTP POST 到 /api

参考这里

from flask import Flask, jsonify, request



app = Flask(__name__)


@app.route("/")
def root():
    return "Hey There?"


@app.route("/api", methods=["POST"])
def api():
    request_data = request.get_json()
    entity = request_data.get("entity", "")
    if entity:
        resp = query_shareholding(entity)
        data = make_graph_response(resp)
    else:
        data = dict() # tbd
    return jsonify(data)


def parse_nebula_graphd_endpoint():
    ng_endpoints_str = os.environ.get(
        'NG_ENDPOINTS', '127.0.0.1:9669,').split(",")
    ng_endpoints = []
    for endpoint in ng_endpoints_str:
        if endpoint:
            parts = endpoint.split(":")  # we dont consider IPv6 now
            ng_endpoints.append((parts[0], int(parts[1])))
    return ng_endpoints

def query_shareholding(entity):
    query_string = (
        f"USE shareholding; "
        f"MATCH p=(v)-[e:hold_share|:is_branch_of|:reletive_with|:role_as*1..3]-(v2) "
        f"WHERE id(v) IN ['{ entity }'] RETURN p LIMIT 100"
    )
    session = connection_pool.get_session('root', 'nebula')
    resp = session.execute(query_string)
    return resp

这个请求的结果则是前边前端期待的 JSON，像这样：

curl --header "Content-Type: application/json" \
     --request POST \
     --data '{"entity": "c_132"}' \
     http://192.168.10.14:5000/api | jq

{
  "nodes": [
    {
      "id": "c_132",
      "name": "\"Chambers LLC\"",
      "tag": "corp"
    },
    {
      "id": "c_245",
      "name": "\"Thompson-King\"",
      "tag": "corp"
    },
    {
      "id": "c_132",
      "name": "\"Chambers LLC\"",
      "tag": "corp"
    },
...
    }
  ],
  "relationships": [
    {
      "edge": "hold_share",
      "properties": "{'share': 0.0}",
      "source": "c_245",
      "target": "c_132"
    {
      "edge": "hold_share",
      "properties": "{'share': 9.0}",
      "source": "p_1767",
      "target": "c_132"
    },
    {
      "edge": "hold_share",
      "properties": "{'share': 11.0}",
      "source": "p_1997",
      "target": "c_132"
    },
...
    },
    {
      "edge": "reletive_with",
      "properties": "{'degree': 51}",
      "source": "p_7283",
      "target": "p_4723"
    }
  ]
}

放到一起

项目的代码都在 GitHub 上，最后其实只有一两百行的代码，把所有东西拼起来之后的代码是：

├── README.md         # You could find Design Logs here
├── corp-rel-backend
│   └── app.py        # Flask App to handle Requst and calls GDB
├── corp-rel-frontend
│   └── src
│       ├── App.vue
│       └── main.js   # Vue App to call Flask App and Renders Graph
└── requirements.txt

最终效果

我们做出来了一个简陋但是足够具有参考性的小系统，它接受一个用户输入的实体的 ID，再回车之后：

• 前端程序把请求发给后端
• 后端拼接 Nebula Graph 的查询语句，通过 Nebula Python 客户端请求 Nebula Graph
• Nebula Graph 接受请求做出穿透查询，返回结构给后端
• 后端将结果构建成前端 D3 接受的格式，传给前端
• 前端接收到图结构的数据，渲染股权穿透的数据如下：

总结

现在，我们知道得益于图数据库的设计，在它上边构建一个方便的股权分析系统非常自然、高效，我们或者利用图数据库的图探索可视化能力、或者自己搭建，可以为用户提供非常高效、直观的多跳股权穿透分析。

如果你想了解更多关于分布式图数据库的知识，欢迎关注 Nebula Graph 这个开源项目，它已经被国内很多团队、公司认可选为图时代数据技术存储层的利器，大家可以访问这里，或者这里，了解更多相关的分享和文章。

未来，我会给大家分享更多图数据库相关的文章、视频和开源示例项目思路分享和教程。

题图版权：fabioha