自定义Graph Component:1.1-JiebaTokenizer具体实现

发布时间 2023-11-12 21:02:43作者: 扫地升

  JiebaTokenizer类继承自Tokenizer类,而Tokenizer类又继承自GraphComponent类,GraphComponent类继承自ABC类(抽象基类)。本文使用《使用ResponseSelector实现校园招聘FAQ机器人》中的例子,主要详解介绍JiebaTokenizer类中方法的具体实现。

0.JiebaTokenizer类中方法列表
  下面是JiebaTokenizer类中所有方法和属性列表,也包括从其它类中继承来的方法和属性,如下所示:   JiebaTokenizer类(默认参数)中本身方法执行顺序,如下所示:

JiebaTokenizer.supported_languages()
JiebaTokenizer.required_packages()
JiebaTokenizer.get_default_config()
JiebaTokenizer.create()
JiebaTokenizer.__init__()
JiebaTokenizer.train()
JiebaTokenizer.persist()
JiebaTokenizer.load()
JiebaTokenizer.tokenize()

  默认参数没有执行2个方法,如下所示:

_load_custom_dictionarypipeline(*)方法
_copy_files_dir_to_dir(*)方法

  接下来自然而然的问题是,如何在config.yml中给JiebaTokenizer自定义参数呢?可参考get_default_config()方法,如下所示:

def get_default_config() -> Dict[Text, Any]:
    return {
        # default don't load custom dictionary  # 默认不加载自定义字典
        "dictionary_path": None,  # 自定义字典的路径
        # Flag to check whether to split intents  # 检查是否拆分intent的标志
        "intent_tokenization_flag": False,
        # Symbol on which intent should be split  # intent应该拆分的符号
        "intent_split_symbol""_",
        # Regular expression to detect tokens  # 用于检测tokens的正则表达式
        "token_pattern": None,
        # Symbol on which prefix should be split  # 前缀应该拆分的符号
        "prefix_separator_symbol": None,
    }

1.supported_languages(*)方法
解析:支持的语言,即["zh"]。如下所示:

@staticmethod
def supported_languages() -> Optional[List[Text]]:
    """Supported languages (see parent class for full docstring)."""  # 支持的语言(请参阅父类的完整文档字符串)。
    print("JiebaTokenizer.supported_languages()")

    return ["zh"]

2.get_default_config(*)方法
解析:返回默认配置,如下所示:

@staticmethod
def get_default_config() -> Dict[Text, Any]:
    """Returns default config (see parent class for full docstring)."""  # 返回默认配置(请参阅父类的完整文档字符串)。
    print("JiebaTokenizer.get_default_config()")

    return {
        # default don't load custom dictionary  # 默认不加载自定义字典
        "dictionary_path": None,
        # Flag to check whether to split intents  # 检查是否拆分意图的标志
        "intent_tokenization_flag": False,
        # Symbol on which intent should be split  # 意图应该拆分的符号
        "intent_split_symbol""_",
        # Regular expression to detect tokens  # 用于检测tokens的正则表达式
        "token_pattern": None,
        # Symbol on which prefix should be split  # 前缀应该拆分的符号
        "prefix_separator_symbol": None,
    }

3.__init__(*)方法
解析:执行到create()方法的cls(config, model_storage, resource)时,实际调用的是def __init__()。如下所示:

def __init__(
    self, config: Dict[Text, Any], model_storage: ModelStorage, resource: Resource
) -> None:
    """Initialize the tokenizer."""  # 初始化标记器。
    print("JiebaTokenizer.__init__()")

    super().__init__(config)
    self._model_storage = model_storage
    self._resource = resource

4.create(*)方法
解析:创建一个新组件,如下所示:

@classmethod
def create(
    cls,
    config: Dict[Text, Any],
    model_storage: ModelStorage,
    resource: Resource,
    execution_context: ExecutionContext,
) -> JiebaTokenizer:
    """Creates a new component (see parent class for full docstring)."""  # 创建一个新组件(请参阅父类的完整文档字符串)。
    print("JiebaTokenizer.create()")

    # Path to the dictionaries on the local filesystem.
    dictionary_path = config["dictionary_path"]

    if dictionary_path is not None:
        cls._load_custom_dictionary(dictionary_path)
    return cls(config, model_storage, resource)

(1)config: Dict[Text, Any]

{
 'dictionary_path': None,
 'intent_split_symbol''_',
 'intent_tokenization_flag': False,
 'prefix_separator_symbol': None,
 'token_pattern': None
}

(2)model_storage: ModelStorage (3)resource: Resource

{
 name = 'train_JiebaTokenizer0'
 output_fingerprint = '318d7f231c4544dc9828e1a9d7dd1851'
}

(4)execution_context: ExecutionContext 其中,cls(config, model_storage, resource)实际调用的是def __init__()

5.required_packages(*)方法
解析:此组件运行所需的任何额外python依赖项,即["jieba"]。如下所示:

@staticmethod
def required_packages() -> List[Text]:
    """Any extra python dependencies required for this component to run."""  # 此组件运行所需的任何额外python依赖项。
    print("JiebaTokenizer.required_packages()")

    return ["jieba"]

6._load_custom_dictionary(*)方法
解析:从模型存储加载自定义字典,如下所示:

@staticmethod
def _load_custom_dictionary(path: Text) -> None:
    """Load all the custom dictionaries stored in the path.  # 加载存储在路径中的所有自定义字典。
    More information about the dictionaries file format can be found in the documentation of jieba. https://github.com/fxsjy/jieba#load-dictionary
    "
""
    print("JiebaTokenizer._load_custom_dictionary()")
    import jieba

    jieba_userdicts = glob.glob(f"{path}/*")  # 获取路径下的所有文件。
    for jieba_userdict in jieba_userdicts:  # 遍历所有文件。
        logger.info(f"Loading Jieba User Dictionary at {jieba_userdict}")  # 加载结巴用户字典。
        jieba.load_userdict(jieba_userdict)  # 加载用户字典。

7.train(*)方法
解析:将字典复制到模型存储,如下所示:

def train(self, training_data: TrainingData) -> Resource:
    """Copies the dictionary to the model storage."""
    print("JiebaTokenizer.train()")

    self.persist()  # 持久化。
    return self._resource

  其中,返回的self._resource内容如下所示:

8.tokenize(*)方法(重点)
解析:对传入消息的提供属性的文本进行tokenize,如下所示:

def tokenize(self, message: Message, attribute: Text) -> List[Token]:
    """Tokenizes the text of the provided attribute of the incoming message."""
    print("JiebaTokenizer.tokenize()")

    import jieba

    text = message.get(attribute)  # 获取消息的属性

    tokenized = jieba.tokenize(text)  # 对文本进行标记化
    tokens = [Token(word, start) for (word, start, end) in tokenized]  # 生成标记

    return self._apply_token_pattern(tokens)

  其中,message.data内容为{'intent': 'goodbye', 'text': '拜拜'}。其它字段具体数值,如下所示:

9.load(*)方法
解析:从模型存储加载自定义字典,如下所示:

@classmethod
def load(
    cls,
    config: Dict[Text, Any],
    model_storage: ModelStorage,
    resource: Resource,
    execution_context: ExecutionContext,
    **kwargs: Any,
) -> JiebaTokenizer:
    """Loads a custom dictionary from model storage."""  # 从模型存储加载自定义字典。
    print("JiebaTokenizer.load()")

    dictionary_path = config["dictionary_path"]

    # If a custom dictionary path is in the config we know that it should have been saved to the model storage.  # 如果配置中有自定义字典路径,我们知道它应该已保存到模型存储中。
    if dictionary_path is not None:
        try:
            with model_storage.read_from(resource) as resource_directory:
                cls._load_custom_dictionary(str(resource_directory))
        except ValueError:
            logger.debug(
                f"Failed to load {cls.__name__} from model storage. "
                f"Resource '{resource.name}' doesn't exist."
            )
    return cls(config, model_storage, resource)

10._copy_files_dir_to_dir(*)方法
解析:执行persist(*)方法时会调用该方法,如下所示:

@staticmethod
def _copy_files_dir_to_dir(input_dir: Text, output_dir: Text) -> None:
    print("JiebaTokenizer._copy_files_dir_to_dir()")

    # make sure target path exists  # 确保目标路径存在
    if not os.path.exists(output_dir):
        os.makedirs(output_dir)

    target_file_list = glob.glob(f"{input_dir}/*")
    for target_file in target_file_list:
        shutil.copy2(target_file, output_dir)

11.persist(*)方法
解析:持久化自定义字典,如下所示:

def persist(self) -> None:
    """Persist the custom dictionaries."""
    print("JiebaTokenizer.persist()")

    dictionary_path = self._config["dictionary_path"]
    if dictionary_path is not None:
        with self._model_storage.write_to(self._resource) as resource_directory:
            self._copy_files_dir_to_dir(dictionary_path, str(resource_directory))

12._model_storage属性
解析:用来初始化JiebaTokenizer类的属性,详见构造函数。

13._resource属性
解析:用来初始化JiebaTokenizer类的属性,详见构造函数。

参考文献:
[1]https://github.com/RasaHQ/rasa/blob/main/rasa/nlu/tokenizers/jieba_tokenizer.py
[2]使用ResponseSelector实现校园招聘FAQ机器人:https://mp.weixin.qq.com/s/ZG3mBPvkAfaRcjmXq7zVLA