239 lines
10 KiB
Markdown
239 lines
10 KiB
Markdown
<!--
|
||
Copyright 2020 The HuggingFace Team. All rights reserved.
|
||
|
||
Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
|
||
the License. You may obtain a copy of the License at
|
||
|
||
http://www.apache.org/licenses/LICENSE-2.0
|
||
|
||
Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
|
||
an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
|
||
|
||
⚠️ Note that this file is in Markdown but contain specific syntax for our doc-builder (similar to MDX) that may not be
|
||
rendered properly in your Markdown viewer.
|
||
-->
|
||
|
||
# 如何创建自定义流水线？
|
||
|
||
在本指南中，我们将演示如何创建一个自定义流水线并分享到 [Hub](https://hf.co/models)，或将其添加到 🤗 Transformers 库中。
|
||
|
||
首先，你需要决定流水线将能够接受的原始条目。它可以是字符串、原始字节、字典或任何看起来最可能是期望的输入。
|
||
尽量保持输入为纯 Python 语言，因为这样可以更容易地实现兼容性（甚至通过 JSON 在其他语言之间）。
|
||
这些将是流水线 (`preprocess`) 的 `inputs`。
|
||
|
||
然后定义 `outputs`。与 `inputs` 相同的策略。越简单越好。这些将是 `postprocess` 方法的输出。
|
||
|
||
首先继承基类 `Pipeline`，其中包含实现 `preprocess`、`_forward`、`postprocess` 和 `_sanitize_parameters` 所需的 4 个方法。
|
||
|
||
```python
|
||
from transformers import Pipeline
|
||
|
||
|
||
class MyPipeline(Pipeline):
|
||
def _sanitize_parameters(self, **kwargs):
|
||
preprocess_kwargs = {}
|
||
if "maybe_arg" in kwargs:
|
||
preprocess_kwargs["maybe_arg"] = kwargs["maybe_arg"]
|
||
return preprocess_kwargs, {}, {}
|
||
|
||
def preprocess(self, inputs, maybe_arg=2):
|
||
model_input = Tensor(inputs["input_ids"])
|
||
return {"model_input": model_input}
|
||
|
||
def _forward(self, model_inputs):
|
||
# model_inputs == {"model_input": model_input}
|
||
outputs = self.model(**model_inputs)
|
||
# Maybe {"logits": Tensor(...)}
|
||
return outputs
|
||
|
||
def postprocess(self, model_outputs):
|
||
best_class = model_outputs["logits"].softmax(-1)
|
||
return best_class
|
||
```
|
||
|
||
这种分解的结构旨在为 CPU/GPU 提供相对无缝的支持，同时支持在不同线程上对 CPU 进行预处理/后处理。
|
||
|
||
`preprocess` 将接受最初定义的输入，并将其转换为可供模型输入的内容。它可能包含更多信息，通常是一个 `Dict`。
|
||
|
||
`_forward` 是实现细节，不应直接调用。`forward` 是首选的调用方法，因为它包含保障措施，以确保一切都在预期的设备上运作。
|
||
如果任何内容与实际模型相关，它应该属于 `_forward` 方法，其他内容应该在 preprocess/postprocess 中。
|
||
|
||
`postprocess` 方法将接受 `_forward` 的输出，并将其转换为之前确定的最终输出。
|
||
|
||
`_sanitize_parameters` 存在是为了允许用户在任何时候传递任何参数，无论是在初始化时 `pipeline(...., maybe_arg=4)`
|
||
还是在调用时 `pipe = pipeline(...); output = pipe(...., maybe_arg=4)`。
|
||
|
||
`_sanitize_parameters` 的返回值是将直接传递给 `preprocess`、`_forward` 和 `postprocess` 的 3 个关键字参数字典。
|
||
如果调用方没有使用任何额外参数调用，则不要填写任何内容。这样可以保留函数定义中的默认参数，这总是更"自然"的。
|
||
|
||
在分类任务中，一个经典的例子是在后处理中使用 `top_k` 参数。
|
||
|
||
```python
|
||
>>> pipe = pipeline("my-new-task")
|
||
>>> pipe("This is a test")
|
||
[{"label": "1-star", "score": 0.8}, {"label": "2-star", "score": 0.1}, {"label": "3-star", "score": 0.05}
|
||
{"label": "4-star", "score": 0.025}, {"label": "5-star", "score": 0.025}]
|
||
|
||
>>> pipe("This is a test", top_k=2)
|
||
[{"label": "1-star", "score": 0.8}, {"label": "2-star", "score": 0.1}]
|
||
```
|
||
|
||
为了实现这一点，我们将更新我们的 `postprocess` 方法，将默认参数设置为 `5`，
|
||
并编辑 `_sanitize_parameters` 方法，以允许这个新参数。
|
||
|
||
```python
|
||
def postprocess(self, model_outputs, top_k=5):
|
||
best_class = model_outputs["logits"].softmax(-1)
|
||
# Add logic to handle top_k
|
||
return best_class
|
||
|
||
|
||
def _sanitize_parameters(self, **kwargs):
|
||
preprocess_kwargs = {}
|
||
if "maybe_arg" in kwargs:
|
||
preprocess_kwargs["maybe_arg"] = kwargs["maybe_arg"]
|
||
|
||
postprocess_kwargs = {}
|
||
if "top_k" in kwargs:
|
||
postprocess_kwargs["top_k"] = kwargs["top_k"]
|
||
return preprocess_kwargs, {}, postprocess_kwargs
|
||
```
|
||
|
||
尽量保持简单输入/输出，最好是可 JSON 序列化的，因为这样可以使流水线的使用非常简单，而不需要用户了解新的对象类型。
|
||
通常也相对常见地支持许多不同类型的参数以便使用（例如音频文件，可以是文件名、URL 或纯字节）。
|
||
|
||
## 将其添加到支持的任务列表中
|
||
|
||
要将你的 `new-task` 注册到支持的任务列表中，你需要将其添加到 `PIPELINE_REGISTRY` 中：
|
||
|
||
```python
|
||
from transformers.pipelines import PIPELINE_REGISTRY
|
||
|
||
PIPELINE_REGISTRY.register_pipeline(
|
||
"new-task",
|
||
pipeline_class=MyPipeline,
|
||
pt_model=AutoModelForSequenceClassification,
|
||
)
|
||
```
|
||
|
||
如果需要，你可以指定一个默认模型，此时它应该带有一个特定的修订版本（可以是分支名称或提交哈希，这里我们使用了 `"abcdef"`），以及类型：
|
||
|
||
```python
|
||
PIPELINE_REGISTRY.register_pipeline(
|
||
"new-task",
|
||
pipeline_class=MyPipeline,
|
||
pt_model=AutoModelForSequenceClassification,
|
||
default={"pt": ("user/awesome_model", "abcdef")},
|
||
type="text", # current support type: text, audio, image, multimodal
|
||
)
|
||
```
|
||
|
||
## 在 Hub 上分享你的流水线
|
||
|
||
要在 Hub 上分享你的自定义流水线，你只需要将 `Pipeline` 子类的自定义代码保存在一个 Python 文件中。
|
||
例如，假设我们想使用一个自定义流水线进行句对分类，如下所示：
|
||
|
||
```py
|
||
import numpy as np
|
||
|
||
from transformers import Pipeline
|
||
|
||
|
||
def softmax(outputs):
|
||
maxes = np.max(outputs, axis=-1, keepdims=True)
|
||
shifted_exp = np.exp(outputs - maxes)
|
||
return shifted_exp / shifted_exp.sum(axis=-1, keepdims=True)
|
||
|
||
|
||
class PairClassificationPipeline(Pipeline):
|
||
def _sanitize_parameters(self, **kwargs):
|
||
preprocess_kwargs = {}
|
||
if "second_text" in kwargs:
|
||
preprocess_kwargs["second_text"] = kwargs["second_text"]
|
||
return preprocess_kwargs, {}, {}
|
||
|
||
def preprocess(self, text, second_text=None):
|
||
return self.tokenizer(text, text_pair=second_text, return_tensors=self.framework)
|
||
|
||
def _forward(self, model_inputs):
|
||
return self.model(**model_inputs)
|
||
|
||
def postprocess(self, model_outputs):
|
||
logits = model_outputs.logits[0].numpy()
|
||
probabilities = softmax(logits)
|
||
|
||
best_class = np.argmax(probabilities)
|
||
label = self.model.config.id2label[best_class]
|
||
score = probabilities[best_class].item()
|
||
logits = logits.tolist()
|
||
return {"label": label, "score": score, "logits": logits}
|
||
```
|
||
|
||
这个实现与框架无关，适用于 PyTorch 和 TensorFlow 模型。如果我们将其保存在一个名为
|
||
`pair_classification.py` 的文件中，然后我们可以像这样导入并注册它：
|
||
|
||
```py
|
||
from pair_classification import PairClassificationPipeline
|
||
from transformers.pipelines import PIPELINE_REGISTRY
|
||
from transformers import AutoModelForSequenceClassification, TFAutoModelForSequenceClassification
|
||
|
||
PIPELINE_REGISTRY.register_pipeline(
|
||
"pair-classification",
|
||
pipeline_class=PairClassificationPipeline,
|
||
pt_model=AutoModelForSequenceClassification,
|
||
tf_model=TFAutoModelForSequenceClassification,
|
||
)
|
||
```
|
||
|
||
完成这些步骤后，我们可以将其与预训练模型一起使用。例如，`sgugger/finetuned-bert-mrpc`
|
||
已经在 MRPC 数据集上进行了微调，用于将句子对分类为是释义或不是释义。
|
||
|
||
```py
|
||
from transformers import pipeline
|
||
|
||
classifier = pipeline("pair-classification", model="sgugger/finetuned-bert-mrpc")
|
||
```
|
||
|
||
然后，我们可以通过在 `Repository` 中使用 `save_pretrained` 方法将其分享到 Hub 上：
|
||
|
||
```py
|
||
from huggingface_hub import Repository
|
||
|
||
repo = Repository("test-dynamic-pipeline", clone_from="{your_username}/test-dynamic-pipeline")
|
||
classifier.save_pretrained("test-dynamic-pipeline")
|
||
repo.push_to_hub()
|
||
```
|
||
|
||
这将会复制包含你定义的 `PairClassificationPipeline` 的文件到文件夹 `"test-dynamic-pipeline"` 中，
|
||
同时保存流水线的模型和分词器，然后将所有内容推送到仓库 `{your_username}/test-dynamic-pipeline` 中。
|
||
之后，只要提供选项 `trust_remote_code=True`，任何人都可以使用它：
|
||
|
||
```py
|
||
from transformers import pipeline
|
||
|
||
classifier = pipeline(model="{your_username}/test-dynamic-pipeline", trust_remote_code=True)
|
||
```
|
||
|
||
## 将流水线添加到 🤗 Transformers
|
||
|
||
如果你想将你的流水线贡献给 🤗 Transformers，你需要在 `pipelines` 子模块中添加一个新模块，
|
||
其中包含你的流水线的代码，然后将其添加到 `pipelines/__init__.py` 中定义的任务列表中。
|
||
|
||
然后，你需要添加测试。创建一个新文件 `tests/test_pipelines_MY_PIPELINE.py`，其中包含其他测试的示例。
|
||
|
||
`run_pipeline_test` 函数将非常通用，并在每种可能的架构上运行小型随机模型，如 `model_mapping` 和 `tf_model_mapping` 所定义。
|
||
|
||
这对于测试未来的兼容性非常重要，这意味着如果有人为 `XXXForQuestionAnswering` 添加了一个新模型，
|
||
流水线测试将尝试在其上运行。由于模型是随机的，所以不可能检查实际值，这就是为什么有一个帮助函数 `ANY`，它只是尝试匹配流水线的输出类型。
|
||
|
||
你还 **需要** 实现 2（最好是 4）个测试。
|
||
|
||
- `test_small_model_pt`：为这个流水线定义一个小型模型（结果是否合理并不重要），并测试流水线的输出。
|
||
结果应该与 `test_small_model_tf` 的结果相同。
|
||
- `test_small_model_tf`：为这个流水线定义一个小型模型（结果是否合理并不重要），并测试流水线的输出。
|
||
结果应该与 `test_small_model_pt` 的结果相同。
|
||
- `test_large_model_pt`（可选）：在一个真实的流水线上测试流水线，结果应该是有意义的。
|
||
这些测试速度较慢，应该被如此标记。这里的目标是展示流水线，并确保在未来的发布中没有漂移。
|
||
- `test_large_model_tf`（可选）：在一个真实的流水线上测试流水线，结果应该是有意义的。
|
||
这些测试速度较慢，应该被如此标记。这里的目标是展示流水线，并确保在未来的发布中没有漂移。
|