إضافة خط أنابيب جديد

اجعل Pipeline خاصاً بك من خلال إنشاء فئة فرعية منه وتنفيذ بعض الدوال. شارك الكود مع المجتمع على Hub وسجّل خط الأنابيب مع Transformers بحيث يتمكن الجميع من استخدامه بسرعة وسهولة.

ستأخذك هذه الدليل خطوة بخطوة في عملية إضافة خط أنابيب جديد إلى Transformers.

اختيارات التصميم

في الحد الأدنى، يكفي أن تزوّد Pipeline بمدخل مناسب للمهمة. وهذا هو المكان الذي ينبغي أن تبدأ منه عند تصميم خط أنابيبك.

قرر أنواع المدخلات التي يمكن لـ Pipeline قبولها. يمكن أن تكون نصوصًا، أو بايتات خام، أو قواميس، وما إلى ذلك. حاول إبقاء المدخلات بلغة بايثون النقية قدر الإمكان لأنها أكثر توافقًا. ثم قرر المخرجات التي يجب أن يعيدها Pipeline. مرة أخرى، إبقاء المخرجات بلغة بايثون هو الخيار الأبسط والأفضل لأنه أسهل في التعامل معه.

إن إبقاء المدخلات والمخرجات بسيطة، وقابلة للتسلسل بصيغة JSON في المثالي، يسهّل على المستخدمين تشغيل Pipeline الخاص بك دون الحاجة إلى تعلّم أنواع كائنات جديدة. كما أنه من الشائع دعم العديد من أنواع المدخلات المختلفة لسهولة استخدام أكبر. فعلى سبيل المثال، قبول ملف صوتي سواء من اسم الملف، أو رابط URL، أو بايتات خام يمنح المستخدم مرونة أكبر في كيفية تقديم البيانات الصوتية.

إنشاء خط أنابيب

بعد تحديد المدخلات والمخرجات، يمكنك البدء في تنفيذ Pipeline. يجب أن يرث خط أنابيبك من الفئة الأساسية Pipeline وأن يتضمن 4 طرق.

from transformers import Pipeline

class MyPipeline(Pipeline):
    def _sanitize_parameters(self, **kwargs):

    def preprocess(self, inputs, args=2):

    def _forward(self, model_inputs):

    def postprocess(self, model_outputs):

1. يقوم preprocess باستقبال المدخلات وتحويلها إلى صيغة الإدخال المناسبة للنموذج.

def preprocess(self, inputs, maybe_arg=2):
    model_input = Tensor(inputs["input_ids"])
    return {"model_input": model_input}

2. لا ينبغي استدعاء _forward مباشرةً. الطريقة المفضلة هي forward لأنها تتضمن تدابير وقائية للتأكد من أن كل شيء يعمل بشكل صحيح على الجهاز المتوقع. كل ما يتعلق بالنموذج يجب وضعه في _forward وكل ما عدا ذلك يجب وضعه في preprocess أو postprocess.

def _forward(self, model_inputs):
    outputs = self.model(**model_inputs)
    return outputs

3. postprocess يولّد المخرجات النهائية من مخرجات النموذج في _forward.

def postprocess(self, model_outputs, top_k=5):
    best_class = model_outputs["logits"].softmax(-1)
    return best_class

4. _sanitize_parameters يتيح للمستخدمين تمرير معاملات إضافية إلى Pipeline. وقد يحدث ذلك أثناء التهيئة أو عند استدعاء Pipeline. يعيد _sanitize_parameters ثلاثة قواميس (dicts) من معاملات الكلمات المفتاحية الإضافية التي تُمرَّر مباشرةً إلى preprocess و_forward وpostprocess. لا تضف أي شيء إذا لم يستدعِ المستخدم خط الأنابيب (pipeline) بمعاملات إضافية. وهذا يحافظ على المعاملات الافتراضية في تعريف الدالة، وهو أمر أكثر طبيعية دائمًا.

على سبيل المثال، أضف معامل top_k في postprocess لإرجاع أكثر 5 فئات احتمالاً. ثم في _sanitize_parameters، تحقق مما إذا كان المستخدم قد مرر top_k وأضفه إلى postprocess_kwargs.

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

الآن يمكن أن تُرجع خط المعالجة (pipeline) أكثر التسميات احتمالاً إذا اختار المستخدم ذلك.

from transformers import pipeline

pipeline = pipeline("my-task")
# returns 3 most likely labels
pipeline("This is the best meal I've ever had", top_k=3)
# returns 5 most likely labels by default
pipeline("This is the best meal I've ever had")

تسجيل خط أنابيب

سجّل المهمة الجديدة التي يدعمها خط أنابيبك في PIPELINE_REGISTRY. يحدد السجل:

• فئة نموذج Pytorch المدعومة مع pt_model
• نموذج افتراضي يجب أن يأتي من مراجعة محددة (فرع، أو تجزئة التزامن) حيث يعمل النموذج كما هو متوقع مع default
• المدخل المتوقع من نوع type

from transformers.pipelines import PIPELINE_REGISTRY
from transformers import AutoModelForSequenceClassification

PIPELINE_REGISTRY.register_pipeline(
    "new-task",
    pipeline_class=MyPipeline,
    pt_model=AutoModelForSequenceClassification,
    default={"pt": ("user/awesome-model", "branch-name")},
    type="text",
)

شارك خط أنابيبك

شارك خط أنابيبك مع المجتمع على Hub أو يمكنك إضافته مباشرة إلى Transformers.

يكون رفع كود خط الأنابيب الخاص بك إلى Hub أسرع لأنه لا يتطلب مراجعة من فريق Transformers. قد يكون إضافة خط الأنابيب إلى Transformers أبطأ لأنه يتطلب مراجعة، كما تحتاج إلى إضافة اختبارات للتأكد من أن Pipeline الخاص بك يعمل.

الرفع إلى المستودع

أضف كود خط الأنابيب الخاص بك إلى Hub في ملف Python.

على سبيل المثال، قد يبدو خط أنابيب مخصص لتصنيف أزواج الجمل كما هو موضح في الكود أدناه.

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}

احفظ الكود في ملف باسم pair_classification.py، واستورده وسجّله كما هو موضح أدناه.

from pair_classification import PairClassificationPipeline
from transformers.pipelines import PIPELINE_REGISTRY
from transformers import AutoModelForSequenceClassification

PIPELINE_REGISTRY.register_pipeline(
    "pair-classification",
    pipeline_class=PairClassificationPipeline,
    pt_model=AutoModelForSequenceClassification,
)

تُسجّل الدالة register_pipeline تفاصيل خط المعالجة (نوع المهمة، فئة خط المعالجة، الخوادم المدعومة) في ملف config.json الخاص بالنماذج.

  "custom_pipelines": {
    "pair-classification": {
      "impl": "pair_classification.PairClassificationPipeline",
      "pt": [
        "AutoModelForSequenceClassification"
      ],
    }
  },

استدعِ push_to_hub() لدفع خط المعالجة إلى Hub. يتم نسخ ملف Python الذي يحتوي على الكود إلى Hub، كما يتم حفظ نموذج خط المعالجة ومُجزّئ الرموز ودفعهما إلى Hub. يجب أن يكون خط المعالجة متوفرًا الآن على Hub ضمن نطاقك.

from transformers import pipeline

pipeline = pipeline(task="pair-classification", model="sgugger/finetuned-bert-mrpc")
pipeline.push_to_hub("pair-classification-pipeline")

للاستخدام خط الأنابيب، أضف trust_remote_code=True عند تحميل خط الأنابيب.

from transformers import pipeline

pipeline = pipeline(task="pair-classification", trust_remote_code=True)

إضافة إلى Transformers

يتطلب إضافة خط أنابيب مخصص إلى Transformers إضافة اختبارات للتأكد من أن كل شيء يعمل كما هو متوقع، وطلب مراجعة من فريق Transformers.

أضف كود خط الأنابيب الخاص بك كوحدة جديدة إلى وحدة pipelines الفرعية، وأضفه إلى قائمة المهام المحددة في pipelines/init.py.

بعد ذلك، أضف اختبارًا جديدًا للخط الأنابيب في transformers/tests/pipelines. يمكنك الاطلاع على الاختبارات الأخرى كمثال على كيفية اختبار خط أنابيبك.

يجب أن تكون الدالة run_pipeline_test عامة جداً وتعمل على النماذج المعرّفة في model_mapping. وهذا أمر مهم لاختبار التوافق المستقبلي مع النماذج الجديدة.

ستلاحظ أيضاً استخدام ANY في جميع أنحاء دالة run_pipeline_test. النماذج عشوائية، لذا لا يمكنك التحقق من القيم الفعلية. يتيح استخدام ANY للمختبر مطابقة مخرجات نوع الخطّ الأنابيب بدلاً من ذلك.

أخيرًا، يجب عليك أيضًا تنفيذ الاختبارات الأربعة التالية.

1. test_small_model_pt، استخدم نموذجًا صغيرًا لهذه الأنابيب للتأكد من إرجاعها المخرجات الصحيحة. لا يجب أن تكون النتائج ذات معنى. يجب أن تُرجع كل أنبوب النتيجة نفسها.
2. test_large_model_pt، استخدم نموذجًا واقعيًا لهذه الأنابيب للتأكد من إرجاعها نتائج ذات معنى. هذه الاختبارات بطيئة ويجب وضع علامة عليها على أنها بطيئة.