NLP(21)--任务型对话机器人_任务型智能问答

前言

仅记录学习过程，有问题欢迎讨论

问答系统

• 闲聊、任务型（帮我设闹钟）、回答型（Q&A）

任务型对话机器人：（帮我定火车票/多轮次）

• 领域识别（分类、匹配）+意图识别（分类、匹配）+槽填充（序列标注）
• 对话状态跟踪（DST–Dialogue state）：
  对于用户目标的达成状态跟踪
• 对话策略(DPO-- Dialogue policy)：
  反问获取信息，向用户确认，回答等
  • 基于模版：根据模版进行槽位填充（死板。但是快，可以拓展）
  • 基于神经网络:将语义槽信息和对话策略做成1-hot向量融入模型
• 按照策略进行回复

常见的评估方式：

• 模型层面：意图识别率，槽位填充率
• 应用层面：节点到达率，任务达成率，场景覆盖度
• 功能层面：新场景迁移效率，跨场景交互能力，人工复杂度

问答型：

1、基于Faq库 --文本匹配问题相似度

优势：

• 实施简单：构建FAQ问答库相对容易，只需收集常见问题及对应答案即可，不需要复杂的知识建模。
• 维护成本较低：对于新增或更新的问题，直接在库中增删改即可，操作简便。
• 适用快速响应：对于明确、结构化的问题，能够迅速提供准确答案。

劣势：

• 泛化能力弱：对于变化形式多样的提问方式可能无法有效应对，需要为相似问题创建多个条目。
• 理解局限性：缺乏对问题深层次理解和语境识别的能力，难以处理需要推理或理解复杂关系的问题。
• 扩展性差：随着问题数量增加，维护成本和难度会逐渐上升，且难以自然地扩展到新领域。

适用场景：

• 客户服务领域，如产品使用帮助、常见问题解答等，问题类型固定且范围明确。
  加粗样式简单信息查询，如营业时间、地址查询等。

2、基于知识图谱 – 文本转sql 去数据库查询数据

优势：

• 强大的语义理解：通过知识图谱，机器人能更好地理解问题的语义和上下文，支持更自然语言的交互。
• 推理能力强：可以进行一定程度的逻辑推理和关系链分析，回答需要综合多源信息的问题。
• 高度可扩展：知识图谱的结构化特性便于添加新知识，支持跨领域的知识整合和应用。
• 个性化服务：基于用户历史和上下文，提供更加个性化的信息和服务。

劣势：

• 构建复杂：构建知识图谱需要大量时间和资源，包括数据采集、清洗、实体识别、关系建模等。
  加粗样式维护成本高：知识图谱的持续更新和优化是一大挑战，需要不断监控和调整以保证数据的准确性和时效性。
  加粗样式技术门槛：相比FAQ系统，实现基于知识图谱的问答需要更高级的技术支持，如自然语言处理、图数据库等。

适用场景：

• 高级客服场景，处理复杂、需要推理的问题，如金融咨询、医疗诊断辅助。
• 智能助手，如个人助理、教育辅导等，需要根据用户需求灵活提供信息和服务。
  大规模信息整合查询，如企业内部知识管理、百科全书类应用。

3、基于文档（逐渐被LLM打击）

要求文档中有问题和答案
训练的输入为 ： 问题++文章
输出为 ：文章中答案的位置【start,end】(两次分类)

优势：

• 信息丰富性：可以直接利用现有文档资源，包括但不限于手册、报告、网页等，不需要额外构建专门的知识库或图谱，信息来源广泛。
• 动态更新：随文档内容的更新而自动获得最新信息，适合需要实时性信息的场景。
• 灵活性：对于长尾或非常规问题，如果文档中包含相关信息，机器人仍有可能给出答案，适应性强。

劣势：

• 准确性问题：由于直接从原始文档中抽取答案，没有经过专门的知识结构化处理，可能抓取到的信息不够精确或者上下文不完全匹配。
• 响应速度：相对于预编译好的FAQ或知识图谱，基于文档检索可能需要更多时间来查找和分析文档，影响用户体验。
• 理解深度有限：对于需要深入理解或逻辑推理的问题，单纯依赖文档检索可能难以提供满意答案，缺乏高级语义理解能力。

适用场景：

• 企业内部知识管理：员工可以通过机器人查询公司政策、操作指南、技术文档等，尤其是在文档体系庞大、更新频繁的环境中。
• 法律咨询、科研文献查询：用户需要查询具体的法律条款、学术论文内容时，机器人可以辅助检索相关文档段落。
• 新闻媒体、出版业：用于快速检索和引用过往新闻报道、文章内容，提升内容创作效率。

闲聊型：

• faq库
• 生成式：seq2seq模型做文本生成
  检索+生成
• LLm全黑盒处理

代码

提供一个任务型机器人demo
关键是思路，流程逻辑

"""
实现一个简单的订票小助手

"""
import json
import re

import pandas


class memory:
    def __init__(self):
        # 本轮对话命中的节点
        self.hit_node_id = {}
        # 当前对话可用节点
        self.available_node = {}
        # 节点路径
        self.node_path = ""
        # 模版路径
        self.template_path = ""
        # 还没填的槽位
        self.miss_slot = {}
        # 已经填的槽位
        self.fill_slot = {}
        # 对话策略
        self.policy = ""
        # 客户的提问
        self.query = ""
        # 每轮的回答
        self.answer = ""

    def __str__(self):
        # 打印所有字段信息
        return f"hit_node_id: {self.hit_node_id}, available_node: {self.available_node},\n " \
               f"node_path: {self.node_path}, template_path: {self.template_path}, \n" \
               f"miss_slot: {self.miss_slot}, fill_slot: {self.fill_slot},\n" \
               f" policy: {self.policy}, query: {self.query}, answer: {self.answer}\n"


class ticketAssistant:
    def __init__(self, memory):
        # 加载模版和节点数据
        self.memory = memory
        # 保存所有的node info
        self.node_list = {}  # node_id : node_info
        # 保存所有的slot info
        self.slot_list = {}  # slot : [query,value]
        self.load_data()

    def load_data(self):
        # 加载节点数据
        self.load_node_data()
        # 加载模版数据
        self.load_template_data()
        print(self.node_list)
        print(self.slot_list)

    def load_node_data(self):
        # 加载json数据
        with open(self.memory.node_path, "r", encoding="utf-8") as f:
            for node in json.load(f):
                self.node_list[node["id"]] = node
                # 如果包含node1
                if "node1" in node["id"]:
                    # 初始化默认在第一个节点
                    self.memory.available_node = [node["id"]]

        return

    def load_template_data(self):
        # 加载模版
        self.template = pandas.read_excel(self.memory.template_path)
        for index, row in self.template.iterrows():
            self.slot_list[row["slot"]] = [row["query"], row["values"]]
        return

    def run(self, query):
        self.memory.query = query

        # 意图识别 + 槽位填充
        self.nlu()
        # 对话状态
        self.dst()
        # 对话策略
        self.dpo()
        # 生成对话
        self.nlp()
        return self.memory

    def nlu(self):
        # 意图识别
        self.intention_recognition()
        # 槽位填充
        self.fit_slot()

    # 检查当前状态
    def dst(self):
        # 检查当前槽位状态是否还有没填充的
        # 获取命中节点
        hit_node_id = self.memory.hit_node_id
        # 获取槽位list
        slot_list = self.node_list[hit_node_id].get("slot", [])
        for slot in slot_list:
            if slot not in self.memory.fill_slot:
                self.memory.miss_slot = slot
                return
        # 槽位填充完毕
        self.memory.miss_slot = None
        return

    def dpo(self):
        # 根据槽位状态 选择对话策略
        if self.memory.miss_slot:
            self.memory.policy = "slot_filling"
            # 留在当前节点
            self.memory.available_node = [self.memory.hit_node_id]
        else:
            self.memory.policy = "dialogue_continue"
            # 去下个节点
            # 如果存在 childnode
            if self.node_list[self.memory.hit_node_id].get("childnode"):
                self.memory.available_node = self.node_list[self.memory.hit_node_id]["childnode"]
            else:
                # 没有childnode 对话结束
                self.memory.available_node = []
        return

    # 按照对话策略 生成返回用户的answer
    def nlp(self):
        # 槽位填充 询问获取具体槽位信息
        if self.memory.policy == "slot_filling":
            # 获取槽位信息
            slot = self.memory.miss_slot
            # 获取槽位query
            query, _ = self.slot_list[slot]
            # 生成回答
            self.memory.answer = query
        elif self.memory.policy == "dialogue_continue":
            # 获取命中节点的response
            hit_node_id = self.memory.hit_node_id
            response = self.node_list[hit_node_id]["response"]

            # 替换槽位
            slot_list = self.node_list[hit_node_id].get("slot", [])
            for slot in slot_list:
                response = response.replace(slot, self.memory.fill_slot[slot])
            self.memory.answer = response
            # 如果当前节点为回滚修改节点 需要跳转到对应节点
            if hit_node_id == "ticket-node5":
                # 对于命中节点的childnode，遍历每个slot，看改正的信息位于哪个slot
                # query_slot = self.memory.query.sub(0, self.memory.query.index("改正为"))
                # 截取 xx改正为xxx的后半部分

                query = self.memory.query
                query_slot_value = query[query.index("改为"):]
                for node in self.memory.available_node:
                    node_info = self.node_list[node]
                    slot_list = node_info.get("slot", [])
                    # 这里可以用文本匹配
                    for slot in slot_list:
                        slot = slot.replace("#", "")
                        # 回答如果命中了slot
                        if slot in self.memory.query:
                            # 修改槽位
                            self.memory.hit_node_id = node
                            self.memory.fill_slot[slot] = query_slot_value
                            self.memory.answer = node_info["response"]
                            self.memory.available_node = node_info["childnode"]
                            # 需要手动跳转到nlp
                            self.nlp()
                            return

        return

    def intention_recognition(self):
        hit_node_id = None
        hit_score = 0
        for node_id in self.memory.available_node:
            # 获取 node信息
            node_info = self.node_list[node_id]
            # 获取每个节点的intent 做匹配 返回最相似的node
            intent = node_info["intent"]
            # 计算相似度
            cal_score = self.cal_similarity(intent, self.memory.query)
            # 更新命中节点 更新命中分数
            if cal_score >= hit_score:
                hit_node_id = node_id
                hit_score = cal_score
        self.memory.hit_node_id = hit_node_id
        return

    # 使用jaccard相似度计算
    def cal_similarity(self, str1_list, str2):
        score = []
        # 可能有多个 intent
        for str1 in str1_list:
            score.append(len(set(str1) & set(str2)) / len(set(str1) | set(str2)))
        return max(score)

    # 槽位填充 根据用户的输入信息 填入槽位
    def fit_slot(self):
        # 获取命中节点
        hit_node_id = self.memory.hit_node_id
        # 获取槽位list
        slot_list = self.node_list[hit_node_id].get("slot", [])
        for slot in slot_list:
            # 不能同时命中出发和目的地
            if "出发城市" in self.memory.answer and slot == "#目的地#":
                continue
            if "目的城市" in self.memory.answer and slot == "#出发地#":
                continue
            # 对于每个槽，看用户输入信息是否包含该槽位信息
            # 获取槽的values信息
            _, values = self.slot_list[slot]
            # 搜索问题中是否含有想要的slot信息
            search_result = re.search(values, self.memory.query)
            # 如果搜到了 就保存 后面替换
            if search_result:
                self.memory.fill_slot[slot] = search_result.group()
            # else:
            # 如果没有搜到 记录下没搜到的slot
            # self.memory.miss_slot.append(slot)

        return


if __name__ == '__main__':
    memory = memory()
    memory.node_path = r"scenario/scenario-ticket-order.json"
    memory.template_path = r"scenario/slot_fitting_templet.xlsx"

    assistant = ticketAssistant(memory)
    while True:
        query = input()
        memory = assistant.run(query)
        print(memory)
        print(memory.answer)
        if "出票成功" in memory.answer:
            break

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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240

scenario-ticket-order.json

[
{
"id":"ticket-node1",
"intent":["我想要订车票"],
"slot":["#车票类型#", "#出发地#", "#目的地#" , "#日期#"],
"action":[""],
"response":"请您确认信息:订购时间为：#日期# ,从 #出发地# 到 #目的地# 的#车票类型#",
"childnode":["ticket-node2", "ticket-node3", "ticket-node4"]
},
{
"id":"ticket-node2",
"intent":["我要帮别人购买车票","我要帮其他人购买车票"],
"slot":["#名字#", "#身份证#"],
"response":"请您确认信息:乘坐人为#名字#，身份证为#身份证#",
"childnode":["ticket-node3","ticket-node4"]
},

{
"id":"ticket-node3",
"intent":["确认","正确"],
"response":"好的，正在帮你出票，出票成功！！"
},
{
"id":"ticket-node4",
"intent":["有信息错误","有地方不对","错误","不正确"],
"childnode":["ticket-node5"],
"response":"请问什么信息不对呢？请你重新输入该信息,格式如：xx改为xxx"
},
{
"id":"ticket-node5",
"intent":["xx改正为xx"],
"childnode":["ticket-node1","ticket-node2"],
"response":"好的，正在帮您更正中==="
}
]
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