自定义组件流程.md · daiyizheng/rasa_demo

转载自https://zhuanlan.zhihu.com/p/231163197
# 自定义组件
rasa的nlu模块除了自身提供的实体抽取和意图识别模型外，还可以添加自定义模型，如自定义意图识别模块或者情感分析等模块。

## 数据结构
传入rasa_nlu的数据存储格式为Message对象，Message对象定义在rasa/nlu/training_data/message.py中。其中message.text中就是输入的文本。

## 处理流程
下图是文本传入rasa_nlu后的标准流程。
![](./images/1.jpg)

### tokenizer
rasa提供了5种分词（Tokenizers）如下：

| Tokenizer           | Requires           | Description                                                  |
| ------------------- | ------------------ | ------------------------------------------------------------ |
| WhitespaceTokenizer | /                  | 为每个以空格分隔的字符序列创建token。                        |
| MitieTokenizer      | 需要先配置MitieNLP | 使用Mitie进行分词，用MITIE tokenizer创建tokens，从而服务于 MITIEFeaturizer。 |
| SpacyTokenizer      | 需要先配置SpacyNLP | 使用Spacy进行分词，用Spacytokenizer创建tokens，从而服务于SpacyFeaturizer。 |
| ConveRTTokenizer    | /                  | 使用ConveRt进行分词，用ConveRT Tokenizer创建tokens，从而服务于ConveRTFeaturizer。 |
| JiebaTokenizer      | /                  | 使用Jieba作为 Tokenizer，对中文进行分词，用户的自定义字典文件可以通过特定的文件目录路径 dictionary_path自动加载。 |

分词后的结果会回传进Message对象中，key值为"tokens"，详见rasa/nlu/tokenizers/tokenizer.py的process方法。

###  **featurizer**

rasa提供了2类featurizer方法— dense_featurizer和sparse_featurizer。

dense_featurizer：

| Featurizer        | Requires                 | Description                                                  |
| ----------------- | ------------------------ | ------------------------------------------------------------ |
| MitieFeaturizer   | 需要先配置MitieNLP       | 使用MITIE featurizer为意图分类创建特征。                     |
| SpacyFeaturizer   | 需要先配置SpacyNLP       | 使用spacy featurizer为意图分类创建特征。                     |
| ConveRTFeaturizer | 需要配置ConveRTTokenizer | 使用ConveRT模型创建用户消息和响应，由于ConveRT模型仅在英语语料上训练，因此只有当训练数据是英语语言时才能使用这个featurizer。 |

sparse_featurizer：

| Featurizer             | Requires | Description                                                  |
| ---------------------- | -------- | ------------------------------------------------------------ |
| RegexFeaturizer        | /        | 为实体提取和意图分类创建特征。在训练期间，regex intent featurizer 以训练数据的格式创建一系列正则表达式列表。对于每个正则，都将设置一个特征，标记是否在输入中找到该表达式，然后将其输入到intent classifier / entity extractor 中以简化分类(假设分类器在训练阶段已经学习了该特征集合，该特征集合表示一定的意图) |
| CountVectorsFeaturizer | /        | 创建用户信息和标签(意图和响应)的词袋表征，用作意图分类器的输入，输入的意图特征以词袋表征 |

特征化后的结果会回传进Message对象中，key值为`text_dense_features"/"text_sparse_features"。

###  **intent classifier**

| Classifier                | Requires                | Description                                                  |
| ------------------------- | ----------------------- | ------------------------------------------------------------ |
| MitieIntentClassifier     | tokenizer 和 featurizer | 该分类器使用MITIE进行意图分类。底层分类器使用的是具有稀疏线性核的多类线性支持向量机 |
| SklearnIntentClassifier   | featurizer              | 该sklearn意图分类器训练一个线性支持向量机，该支持向量机通过网格搜索得到优化。除了其他分类器，它还提供没有“获胜”的标签的排名。spacy意图分类器需要在管道中的先加入一个featurizer。该featurizer创建用于分类的特征。 |
| EmbeddingIntentClassifier | featurizer              | 嵌入式意图分类器将用户输入和意图标签嵌入到同一空间中。Supervised embeddings通过最大化它们之间的相似性来训练。该算法基于StarSpace的。但是，在这个实现中，损失函数略有不同，添加了额外的隐藏层和dropout。该算法还提供了未“获胜”标签的相似度排序。在embedding intent classifier之前，需要在管道中加入一个featurizer。该featurizer创建用以embeddings的特征。建议使用CountVectorsFeaturizer，它可选的预处理有SpacyNLP和SpacyTokenizer。 |
| KeywordIntentClassifier   | /                       | 该分类器通过搜索关键字的消息来工作。默认情况下，匹配是大小写敏感的，只精确匹配地搜索用户消息中关键字。 |

classifier会从Message对象中获取tokenizer 和 featurizer作为分类模型的输入进行意图识别。

## **自定义方法**

由2中的处理流程可以看出，rasa提供的意图识别重点是featurizer，rasa本身提供了5种featurizer方法，除了`ConveRT`模型之外均可用与中文场景，其中MitieFeaturizer和SpacyFeaturizer需要特定的Tokenizer和对应的词向量。

我们也可以自定义自己的中文模型，先通过jieba分词（可以引入用户字典），然后用自己训练的词向量模型来featurize用户输入的text，最后再引入自定义的分类模型识别意图。

![](./images/2.jpg)

```python
class KeywordIntentClassifier(Component):
    """Intent classifier using simple keyword matching.

The classifier takes a list of keywords and associated intents as an input.
    A input sentence is checked for the keywords and the intent is returned.

"""

provides = [INTENT_ATTRIBUTE]

defaults = {"case_sensitive": True}

def __init__(
        self,
        component_config: Optional[Dict[Text, Any]] = None,
        intent_keyword_map: Optional[Dict] = None,
    ):

super(KeywordIntentClassifier, self).__init__(component_config)

self.case_sensitive = self.component_config.get("case_sensitive")
        self.intent_keyword_map = intent_keyword_map or {}

def train(
        self,
        training_data: "TrainingData",
        cfg: Optional["RasaNLUModelConfig"] = None,
        **kwargs: Any,
    ) -> None:

duplicate_examples = set()
        for ex in training_data.training_examples:
            if (
                ex.text in self.intent_keyword_map.keys()
                and ex.get(INTENT_ATTRIBUTE) != self.intent_keyword_map[ex.text]
            ):
                duplicate_examples.add(ex.text)
                raise_warning(
                    f"Keyword '{ex.text}' is a keyword to trigger intent "
                    f"'{self.intent_keyword_map[ex.text]}' and also "
                    f"intent '{ex.get(INTENT_ATTRIBUTE)}', it will be removed "
                    f"from the list of keywords for both of them. "
                    f"Remove (one of) the duplicates from the training data.",
                    docs=DOCS_URL_COMPONENTS + "#keyword-intent-classifier",
                )
            else:
                self.intent_keyword_map[ex.text] = ex.get(INTENT_ATTRIBUTE)
        for keyword in duplicate_examples:
            self.intent_keyword_map.pop(keyword)
            logger.debug(
                f"Removed '{keyword}' from the list of keywords because it was "
                "a keyword for more than one intent."
            )

self._validate_keyword_map()

def _validate_keyword_map(self) -> None:
        re_flag = 0 if self.case_sensitive else re.IGNORECASE

ambiguous_mappings = []
        for keyword1, intent1 in self.intent_keyword_map.items():
            for keyword2, intent2 in self.intent_keyword_map.items():
                if (
                    re.search(r"\b" + keyword1 + r"\b", keyword2, flags=re_flag)
                    and intent1 != intent2
                ):
                    ambiguous_mappings.append((intent1, keyword1))
                    raise_warning(
                        f"Keyword '{keyword1}' is a keyword of intent '{intent1}', "
                        f"but also a substring of '{keyword2}', which is a "
                        f"keyword of intent '{intent2}."
                        f" '{keyword1}' will be removed from the list of keywords.\n"
                        f"Remove (one of) the conflicting keywords from the"
                        f" training data.",
                        docs=DOCS_URL_COMPONENTS + "#keyword-intent-classifier",
                    )
        for intent, keyword in ambiguous_mappings:
            self.intent_keyword_map.pop(keyword)
            logger.debug(
                f"Removed keyword '{keyword}' from intent '{intent}' because it matched a "
                "keyword of another intent."
            )

def process(self, message: Message, **kwargs: Any) -> None:
        intent_name = self._map_keyword_to_intent(message.text)

confidence = 0.0 if intent_name is None else 1.0
        intent = {"name": intent_name, "confidence": confidence}

if message.get(INTENT_ATTRIBUTE) is None or intent is not None:
            message.set(INTENT_ATTRIBUTE, intent, add_to_output=True)

def _map_keyword_to_intent(self, text: Text) -> Optional[Text]:
        re_flag = 0 if self.case_sensitive else re.IGNORECASE

for keyword, intent in self.intent_keyword_map.items():
            if re.search(r"\b" + keyword + r"\b", text, flags=re_flag):
                logger.debug(
                    f"KeywordClassifier matched keyword '{keyword}' to"
                    f" intent '{intent}'."
                )
                return intent

logger.debug("KeywordClassifier did not find any keywords in the message.")
        return None

def persist(self, file_name: Text, model_dir: Text) -> Dict[Text, Any]:
        """Persist this model into the passed directory.

Return the metadata necessary to load the model again.
        """

file_name = file_name + ".json"
        keyword_file = os.path.join(model_dir, file_name)
        utils.write_json_to_file(keyword_file, self.intent_keyword_map)

return {"file": file_name}

@classmethod
    def load(
        cls,
        meta: Dict[Text, Any],
        model_dir: Optional[Text] = None,
        model_metadata: "Metadata" = None,
        cached_component: Optional["KeywordIntentClassifier"] = None,
        **kwargs: Any,
    ) -> "KeywordIntentClassifier":

if model_dir and meta.get("file"):
            file_name = meta.get("file")
            keyword_file = os.path.join(model_dir, file_name)
            if os.path.exists(keyword_file):
                intent_keyword_map = utils.read_json_file(keyword_file)
            else:
                raise_warning(
                    f"Failed to load key word file for `IntentKeywordClassifier`, "
                    f"maybe {keyword_file} does not exist?",
                )
                intent_keyword_map = None
            return cls(meta, intent_keyword_map)
        else:
            raise Exception(
                f"Failed to load keyword intent classifier model. "
                f"Path {os.path.abspath(meta.get('file'))} doesn't exist."
            )
```
从上面的代码可以看到，整个KeywordIntentClassifier由4部分组成，train、process、persist和load。其中，train是模型训练部分，process是模型推理部分，persist是模型持久化部分，load是加载训练好的模型。一开始的provide定义的是返回值提供什么字段，default是模型的一些自定义参数，比如模型的epochs和batch_size等，这里定义的是这些参数的默认值，在config.yml中自定义的pipeline可以传入其他值，如下所示：
![](./images/3.png)

daiyizheng/rasa_demo

简介

发行版

贡献者

近期动态

daiyizheng/rasa_demo .gitee-modal { width: 500px !important; }

简介

发行版

贡献者

近期动态

搜索帮助

daiyizheng/rasa_demo