迅速掌握Python中的Hook鉤子函數-Python教學-PHP中文網

首頁

後端開發

Python教學

迅速掌握Python中的Hook鉤子函數

coldplay.xixi

Dec 11, 2020 pm 05:11 PM

hookpython

Python教學欄位介紹Python中的Hook鉤子函數

迅速掌握Python中的Hook鉤子函數

大量免費學習推薦，敬請訪問python教學(影片)

1. 什麼是Hook

常常會聽到鉤子函數(hook function)這個概念，最近在看目標偵測開源框架mmdetection，裡面也出現大量Hook的程式設計方式，那到底什麼是hook？ hook的作用是什麼？

what is hook ？鉤子hook，顧名思義，可以理解是一個掛鉤，作用是有需要的時候掛一個東西上去。具體的解釋是：鉤子函數是把我們自己實現的hook函數在某個時刻掛接到目標掛載點。
hook函數的作用舉個例子，hook的概念在windows桌面軟體開發很常見，特別是各種事件觸發的機制; 例如C 的MFC程式中，要監聽滑鼠左鍵按下的時間，MFC提供了一個onLeftKeyDown的鉤子函數。很顯然，MFC框架並沒有為我們實現onLeftKeyDown具體的操作，只是為我們提供一個鉤子，當我們需要處理的時候，只要去重寫這個函數，把我們需要操作掛載在這個鉤子裡，如果我們不掛載，MFC事件觸發機制中執行的就是空的操作。

從上面可知

hook函數是程式中預先定義好的函數，這個函數處於原始程式流程當中（揭露一個鉤子出來）
我們需要再在有流程中鉤子定義的函數塊中實現某個具體的細節，需要把我們的實現，掛接或註冊（register）到鉤子裡，使得hook函數對目標可用
hook 是一種程式設計機制，和具體的語言沒有直接的關係
##如果從設計模式上看，hook模式是模板方法的擴展
鉤子只有註冊的時候，才會使用，所以原有程式的流程中，沒有註冊或掛載時，執行的是空（即沒有執行任何操作）

本文用python來解釋hook的實作方式，並展示在開源專案中hook的應用案例。 hook函數和我們常聽到另一個名稱：回呼函數（callback function）功能是類似的，可以按照同種模式來理解。

迅速掌握Python中的Hook鉤子函數

2. hook實作範例

據我所知，hook函數最常使用在某種流程處理當中。這個流程往往有很多步驟。 hook函數常常掛載在這些步驟中，為增加額外的一些操作，提供彈性。

下面舉一個簡單的例子，這個例子的目的是實作一個通用往佇列中插入內容的功能。流程步驟有2個

需要再插入佇列前，對資料進行篩選
input_filter_fn
插入佇列
insert_queue

class ContentStash(object):
    """
    content stash for online operation
    pipeline is
    1. input_filter: filter some contents, no use to user
    2. insert_queue(redis or other broker): insert useful content to queue
    """

    def __init__(self):
        self.input_filter_fn = None
        self.broker = []

    def register_input_filter_hook(self, input_filter_fn):
        """
        register input filter function, parameter is content dict
        Args:
            input_filter_fn: input filter function

        Returns:

        """
        self.input_filter_fn = input_filter_fn

    def insert_queue(self, content):
        """
        insert content to queue
        Args:
            content: dict

        Returns:

        """
        self.broker.append(content)

    def input_pipeline(self, content, use=False):
        """
        pipeline of input for content stash
        Args:
            use: is use, defaul False
            content: dict

        Returns:

        """
        if not use:
            return

        # input filter
        if self.input_filter_fn:
            _filter = self.input_filter_fn(content)
            
        # insert to queue
        if not _filter:
            self.insert_queue(content)



# test
## 实现一个你所需要的钩子实现：比如如果content 包含time就过滤掉，否则插入队列
def input_filter_hook(content):
    """
    test input filter hook
    Args:
        content: dict

    Returns: None or content

    """
    if content.get(&#39;time&#39;) is None:
        return
    else:
        return content


# 原有程序
content = {&#39;filename&#39;: &#39;test.jpg&#39;, &#39;b64_file&#39;: "#test", &#39;data&#39;: {"result": "cat", "probility": 0.9}}
content_stash = ContentStash(&#39;audit&#39;, work_dir=&#39;&#39;)

# 挂上钩子函数， 可以有各种不同钩子函数的实现，但是要主要函数输入输出必须保持原有程序中一致，比如这里是content
content_stash.register_input_filter_hook(input_filter_hook)

# 执行流程
content_stash.input_pipeline(content)

3. hook在開源框架中的應用

3.1 keras

在深度學習訓練流程中，hook函數體現的淋漓盡致。

一個訓練過程（不包括資料準備），會輪詢多次訓練集，每次稱為一個epoch，每個epoch又分成多個batch來訓練。流程先後拆解成：

開始訓練
#訓練一個epoch前

訓練一個epoch後

我們要保存下訓練的模型，在結束訓練時用最好的模型執行下測試集的效果等等。 keras中是透過各種回呼函數來實現鉤子hook功能的。這裡放一個callback的父類，定制時只要繼承這個父類，實作你過關注的鉤子就可以了。

@keras_export(&#39;keras.callbacks.Callback&#39;)
class Callback(object):
  """Abstract base class used to build new callbacks.

  Attributes:
      params: Dict. Training parameters
          (eg. verbosity, batch size, number of epochs...).
      model: Instance of `keras.models.Model`.
          Reference of the model being trained.

  The `logs` dictionary that callback methods
  take as argument will contain keys for quantities relevant to
  the current batch or epoch (see method-specific docstrings).
  """

  def __init__(self):
    self.validation_data = None  # pylint: disable=g-missing-from-attributes
    self.model = None
    # Whether this Callback should only run on the chief worker in a
    # Multi-Worker setting.
    # TODO(omalleyt): Make this attr public once solution is stable.
    self._chief_worker_only = None
    self._supports_tf_logs = False

  def set_params(self, params):
    self.params = params

  def set_model(self, model):
    self.model = model

  @doc_controls.for_subclass_implementers
  @generic_utils.default
  def on_batch_begin(self, batch, logs=None):
    """A backwards compatibility alias for `on_train_batch_begin`."""

  @doc_controls.for_subclass_implementers
  @generic_utils.default
  def on_batch_end(self, batch, logs=None):
    """A backwards compatibility alias for `on_train_batch_end`."""

  @doc_controls.for_subclass_implementers
  def on_epoch_begin(self, epoch, logs=None):
    """Called at the start of an epoch.

    Subclasses should override for any actions to run. This function should only
    be called during TRAIN mode.

    Arguments:
        epoch: Integer, index of epoch.
        logs: Dict. Currently no data is passed to this argument for this method
          but that may change in the future.
    """

  @doc_controls.for_subclass_implementers
  def on_epoch_end(self, epoch, logs=None):
    """Called at the end of an epoch.

    Subclasses should override for any actions to run. This function should only
    be called during TRAIN mode.

    Arguments:
        epoch: Integer, index of epoch.
        logs: Dict, metric results for this training epoch, and for the
          validation epoch if validation is performed. Validation result keys
          are prefixed with `val_`.
    """

  @doc_controls.for_subclass_implementers
  @generic_utils.default
  def on_train_batch_begin(self, batch, logs=None):
    """Called at the beginning of a training batch in `fit` methods.

    Subclasses should override for any actions to run.

    Arguments:
        batch: Integer, index of batch within the current epoch.
        logs: Dict, contains the return value of `model.train_step`. Typically,
          the values of the `Model`&#39;s metrics are returned.  Example:
          `{&#39;loss&#39;: 0.2, &#39;accuracy&#39;: 0.7}`.
    """
    # For backwards compatibility.
    self.on_batch_begin(batch, logs=logs)

  @doc_controls.for_subclass_implementers
  @generic_utils.default
  def on_train_batch_end(self, batch, logs=None):
    """Called at the end of a training batch in `fit` methods.

    Subclasses should override for any actions to run.

    Arguments:
        batch: Integer, index of batch within the current epoch.
        logs: Dict. Aggregated metric results up until this batch.
    """
    # For backwards compatibility.
    self.on_batch_end(batch, logs=logs)

  @doc_controls.for_subclass_implementers
  @generic_utils.default
  def on_test_batch_begin(self, batch, logs=None):
    """Called at the beginning of a batch in `evaluate` methods.

    Also called at the beginning of a validation batch in the `fit`
    methods, if validation data is provided.

    Subclasses should override for any actions to run.

    Arguments:
        batch: Integer, index of batch within the current epoch.
        logs: Dict, contains the return value of `model.test_step`. Typically,
          the values of the `Model`&#39;s metrics are returned.  Example:
          `{&#39;loss&#39;: 0.2, &#39;accuracy&#39;: 0.7}`.
    """

  @doc_controls.for_subclass_implementers
  @generic_utils.default
  def on_test_batch_end(self, batch, logs=None):
    """Called at the end of a batch in `evaluate` methods.

    Also called at the end of a validation batch in the `fit`
    methods, if validation data is provided.

    Subclasses should override for any actions to run.

    Arguments:
        batch: Integer, index of batch within the current epoch.
        logs: Dict. Aggregated metric results up until this batch.
    """

  @doc_controls.for_subclass_implementers
  @generic_utils.default
  def on_predict_batch_begin(self, batch, logs=None):
    """Called at the beginning of a batch in `predict` methods.

    Subclasses should override for any actions to run.

    Arguments:
        batch: Integer, index of batch within the current epoch.
        logs: Dict, contains the return value of `model.predict_step`,
          it typically returns a dict with a key &#39;outputs&#39; containing
          the model&#39;s outputs.
    """

  @doc_controls.for_subclass_implementers
  @generic_utils.default
  def on_predict_batch_end(self, batch, logs=None):
    """Called at the end of a batch in `predict` methods.

    Subclasses should override for any actions to run.

    Arguments:
        batch: Integer, index of batch within the current epoch.
        logs: Dict. Aggregated metric results up until this batch.
    """

  @doc_controls.for_subclass_implementers
  def on_train_begin(self, logs=None):
    """Called at the beginning of training.

    Subclasses should override for any actions to run.

    Arguments:
        logs: Dict. Currently no data is passed to this argument for this method
          but that may change in the future.
    """

  @doc_controls.for_subclass_implementers
  def on_train_end(self, logs=None):
    """Called at the end of training.

    Subclasses should override for any actions to run.

    Arguments:
        logs: Dict. Currently the output of the last call to `on_epoch_end()`
          is passed to this argument for this method but that may change in
          the future.
    """

  @doc_controls.for_subclass_implementers
  def on_test_begin(self, logs=None):
    """Called at the beginning of evaluation or validation.

    Subclasses should override for any actions to run.

    Arguments:
        logs: Dict. Currently no data is passed to this argument for this method
          but that may change in the future.
    """

  @doc_controls.for_subclass_implementers
  def on_test_end(self, logs=None):
    """Called at the end of evaluation or validation.

    Subclasses should override for any actions to run.

    Arguments:
        logs: Dict. Currently the output of the last call to
          `on_test_batch_end()` is passed to this argument for this method
          but that may change in the future.
    """

  @doc_controls.for_subclass_implementers
  def on_predict_begin(self, logs=None):
    """Called at the beginning of prediction.

    Subclasses should override for any actions to run.

    Arguments:
        logs: Dict. Currently no data is passed to this argument for this method
          but that may change in the future.
    """

  @doc_controls.for_subclass_implementers
  def on_predict_end(self, logs=None):
    """Called at the end of prediction.

    Subclasses should override for any actions to run.

    Arguments:
        logs: Dict. Currently no data is passed to this argument for this method
          but that may change in the future.
    """

  def _implements_train_batch_hooks(self):
    """Determines if this Callback should be called for each train batch."""
    return (not generic_utils.is_default(self.on_batch_begin) or
            not generic_utils.is_default(self.on_batch_end) or
            not generic_utils.is_default(self.on_train_batch_begin) or
            not generic_utils.is_default(self.on_train_batch_end))

這些鉤子的原始程式是在模型訓練流程中的

keras原始碼位置： tensorflow\python\keras\engine\training.py

#部分摘錄如下(## I am hook)：

# Container that configures and calls `tf.keras.Callback`s.
      if not isinstance(callbacks, callbacks_module.CallbackList):
        callbacks = callbacks_module.CallbackList(
            callbacks,
            add_history=True,
            add_progbar=verbose != 0,
            model=self,
            verbose=verbose,
            epochs=epochs,
            steps=data_handler.inferred_steps)

      ## I am hook
      callbacks.on_train_begin()
      training_logs = None
      # Handle fault-tolerance for multi-worker.
      # TODO(omalleyt): Fix the ordering issues that mean this has to
      # happen after `callbacks.on_train_begin`.
      data_handler._initial_epoch = (  # pylint: disable=protected-access
          self._maybe_load_initial_epoch_from_ckpt(initial_epoch))
      for epoch, iterator in data_handler.enumerate_epochs():
        self.reset_metrics()
        callbacks.on_epoch_begin(epoch)
        with data_handler.catch_stop_iteration():
          for step in data_handler.steps():
            with trace.Trace(
                &#39;TraceContext&#39;,
                graph_type=&#39;train&#39;,
                epoch_num=epoch,
                step_num=step,
                batch_size=batch_size):
              ## I am hook
              callbacks.on_train_batch_begin(step)
              tmp_logs = train_function(iterator)
              if data_handler.should_sync:
                context.async_wait()
              logs = tmp_logs  # No error, now safe to assign to logs.
              end_step = step + data_handler.step_increment
              callbacks.on_train_batch_end(end_step, logs)
        epoch_logs = copy.copy(logs)

        # Run validation.

        ## I am hook
        callbacks.on_epoch_end(epoch, epoch_logs)

3.2 mmdetection

mmdetection是一個目標偵測的開源框架，整合了許多不同的目標偵測深度學習演算法（pytorch版），如faster-rcnn, fpn, retianet等。裡面也大量使用了hook，暴露給應用實現流程中具體部分。

詳見

https://github.com/open-mmlab/mmdetection

<p>这里看一个训练的调用例子(摘录)（<code>https://github.com/open-mmlab/mmdetection/blob/5d592154cca589c5113e8aadc8798bbc73630d98/mmdet/apis/train.py）

def train_detector(model,
                   dataset,
                   cfg,
                   distributed=False,
                   validate=False,
                   timestamp=None,
                   meta=None):
    logger = get_root_logger(cfg.log_level)

    # prepare data loaders

    # put model on gpus

    # build runner
    optimizer = build_optimizer(model, cfg.optimizer)
    runner = EpochBasedRunner(
        model,
        optimizer=optimizer,
        work_dir=cfg.work_dir,
        logger=logger,
        meta=meta)
    # an ugly workaround to make .log and .log.json filenames the same
    runner.timestamp = timestamp

    # fp16 setting
    # register hooks
    runner.register_training_hooks(cfg.lr_config, optimizer_config,
                                   cfg.checkpoint_config, cfg.log_config,
                                   cfg.get(&#39;momentum_config&#39;, None))
    if distributed:
        runner.register_hook(DistSamplerSeedHook())

    # register eval hooks
    if validate:
        # Support batch_size > 1 in validation
        eval_cfg = cfg.get(&#39;evaluation&#39;, {})
        eval_hook = DistEvalHook if distributed else EvalHook
        runner.register_hook(eval_hook(val_dataloader, **eval_cfg))

    # user-defined hooks
    if cfg.get(&#39;custom_hooks&#39;, None):
        custom_hooks = cfg.custom_hooks
        assert isinstance(custom_hooks, list), \
            f&#39;custom_hooks expect list type, but got {type(custom_hooks)}&#39;
        for hook_cfg in cfg.custom_hooks:
            assert isinstance(hook_cfg, dict), \
                &#39;Each item in custom_hooks expects dict type, but got &#39; \
                f&#39;{type(hook_cfg)}&#39;
            hook_cfg = hook_cfg.copy()
            priority = hook_cfg.pop(&#39;priority&#39;, &#39;NORMAL&#39;)
            hook = build_from_cfg(hook_cfg, HOOKS)
            runner.register_hook(hook, priority=priority)

4. 总结

本文介绍了hook的概念和应用，并给出了python的实现细则。希望对比有帮助。总结如下：

hook函数是流程中预定义好的一个步骤，没有实现
挂载或者注册时，流程执行就会执行这个钩子函数
回调函数和hook函数功能上是一致的
hook设计方式带来灵活性，如果流程中有一个步骤，你想让调用方来实现，你可以用hook函数

相关免费学习推荐：php编程（视频）

以上是迅速掌握Python中的Hook鉤子函數的詳細內容。更多資訊請關注PHP中文網其他相關文章！

陳述

本文轉載於：CSDN。如有侵權，請聯絡admin@php.cn刪除

Python vs.C：申請和用例Apr 12, 2025 am 12:01 AM

Python适合数据科学、Web开发和自动化任务，而C 适用于系统编程、游戏开发和嵌入式系统。Python以简洁和强大的生态系统著称，C 则以高性能和底层控制能力闻名。

2小時的Python計劃：一種現實的方法Apr 11, 2025 am 12:04 AM

2小時內可以學會Python的基本編程概念和技能。 1.學習變量和數據類型，2.掌握控制流（條件語句和循環），3.理解函數的定義和使用，4.通過簡單示例和代碼片段快速上手Python編程。

Python：探索其主要應用程序Apr 10, 2025 am 09:41 AM

Python在web開發、數據科學、機器學習、自動化和腳本編寫等領域有廣泛應用。 1)在web開發中，Django和Flask框架簡化了開發過程。 2)數據科學和機器學習領域，NumPy、Pandas、Scikit-learn和TensorFlow庫提供了強大支持。 3)自動化和腳本編寫方面，Python適用於自動化測試和系統管理等任務。