الاستدلال عبر خادم ويب

الخادم الويب هو نظام ينتظر الطلبات ويقدمها عند وصولها. وهذا يعني أنه يمكنك استخدام Pipeline كمحرك استنتاج على خادم الويب، إذ يمكنك استخدام مُكرِّر (بشكل مشابه لكيفية التكرار على مجموعة بيانات) للتعامل مع كل طلب وارد.

تصميم خادم ويب باستخدام Pipeline أمر فريد من نوعه، وذلك لأنهما مختلفان جوهرياً. فخدمات الويب تعمل بتعدد المهام (متعددة الخيوط، غير متزامنة، إلخ) للتعامل مع طلبات متعددة بشكل متزامن. في المقابل، لم يُصمَّم Pipeline والنموذج الأساسي الذي يعمل تحته لدعم التوازي، نظراً لاستهلاكه الكبير للذاكرة. ومن الأفضل تخصيص جميع الموارد المتاحة لـ Pipeline أثناء تشغيله، أو عند تنفيذ مهمة تتطلب جهداً حاسوبياً كبيراً.

تُبيّن هذه الدليل كيفية تجاوز هذا الاختلاف من خلال استخدام خادم ويب للتعامل مع الحمل الأخف المتمثل في استلام الطلبات وإرسالها، مع الاعتماد على خيط واحد للتعامل مع الحمل الأثقل المتمثل في تشغيل Pipeline.

إنشاء خادم

Starlette هو إطار عمل خفيف الوزن لبناء خوادم الويب. يمكنك استخدام أي إطار عمل آخر ترغب به، لكن قد تحتاج إلى إجراء بعض التعديلات على الكود أدناه.

قبل أن تبدأ، تأكد من تثبيت Starlette وuvicorn.

!pip install starlette uvicorn

الآن يمكنك إنشاء خادم ويب بسيط في ملف server.py. المفتاح هو تحميل النموذج مرة واحدة فقط لمنع النسخ غير الضرورية منه من استهلاك الذاكرة.

أنشئ خط أنابيب لملء الرمز المقنّع [MASK].

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
from transformers import pipeline
import asyncio

async def homepage(request):
    payload = await request.body()
    string = payload.decode("utf-8")
    response_q = asyncio.Queue()
    await request.app.model_queue.put((string, response_q))
    output = await response_q.get()
    return JSONResponse(output)

async def server_loop(q):
    pipe = pipeline(task="fill-mask",model="google-bert/bert-base-uncased")
    while True:
        (string, response_q) = await q.get()
        out = pipe(string)
        await response_q.put(out)

app = Starlette(
    routes=[
        Route("/", homepage, methods=["POST"]),
    ],
)

@app.on_event("startup")
async def startup_event():
    q = asyncio.Queue()
    app.model_queue = q
    asyncio.create_task(server_loop(q))

ابدأ الخادم باستخدام الأمر التالي.

uvicorn server:app

استعلم عن الخادم باستخدام طلب POST.

curl -X POST -d "Paris is the [MASK] of France." http://localhost:8000/

يجب أن يعيد هذا المخرجات أدناه.

[{'score': 0.9969332218170166,
  'token': 3007,
  'token_str': 'capital',
  'sequence': 'paris is the capital of france.'},
 {'score': 0.0005914849461987615,
  'token': 2540,
  'token_str': 'heart',
  'sequence': 'paris is the heart of france.'},
 {'score': 0.00043787318281829357,
  'token': 2415,
  'token_str': 'center',
  'sequence': 'paris is the center of france.'},
 {'score': 0.0003378340043127537,
  'token': 2803,
  'token_str': 'centre',
  'sequence': 'paris is the centre of france.'},
 {'score': 0.00026995912776328623,
  'token': 2103,
  'token_str': 'city',
  'sequence': 'paris is the city of france.'}]

تسلسل الطلبات

يمكن استخدام آلية الطابور في الخادم لبعض التطبيقات المثيرة للاهتمام، مثل الدفع الديناميكي. يقوم الدفع الديناميكي بتجميع عدة طلبات أولاً قبل معالجتها باستخدام Pipeline.

المثال التالي مكتوب بلغة زائفة (Pseudocode) من أجل وضوح القراءة وليس من أجل الأداء، وبشكل خاص، ستلاحظ أن:

1. لا يوجد حد لحجم الدفعة.

2. يتم إعادة ضبط مهلة الانتظار timeout عند كل عملية استعلام من الطابور، مما قد يؤدي إلى انتظار أطول بكثير من قيمة timeout قبل معالجة الطلب. كما سيؤخر هذا الأمر أول طلب استدلال (Inference) بنفس مقدار الوقت. ينتظر خادم الويب دائمًا 1 مللي ثانية حتى لو كان الطابور فارغًا، وهو أمر غير فعّال لأن هذا الوقت يمكن استخدامه لبدء الاستدلال. ومع ذلك، قد يكون من المنطقي إذا كان التجميع (Batching) أمرًا أساسيًا في سيناريو استخدامك.

   من الأفضل أن يكون لدينا مهلة واحدة تبلغ 1 مللي ثانية، بدلاً من إعادة ضبطها عند كل عملية جلب، كما هو موضح أدناه.

async def server_loop(q):
    pipe = pipeline(task="fill-mask", model="google-bert/bert-base-uncased")
    while True:
        (string, rq) = await q.get()
        strings = []
        queues = []
        strings.append(string)
        queues.append(rq)
        while True:
            try:
                (string, rq) = await asyncio.wait_for(q.get(), timeout=1)
            except asyncio.exceptions.TimeoutError:
                break
            strings.append(string)
            queues.append(rq)
        outs = pipe(strings, batch_size=len(strings))
        for rq, out in zip(queues, outs):
            await rq.put(out)

فحص الأخطاء

هناك العديد من الأمور التي قد تسوء في بيئة الإنتاج. فقد تنفد الذاكرة، أو تنفد المساحة، أو يفشل تحميل النموذج، أو تكون إعدادات النموذج غير صحيحة، أو تكون الاستعلامات غير صحيحة، وغيرها الكثير.

يُفيد إضافة عبارات try...except في إرجاع هذه الأخطاء إلى المستخدم لأغراض تصحيح الأخطاء. لكن تذكّر أن هذا قد يشكّل خطرًا أمنيًا إذا كان عليك عدم الكشف عن معلومات معينة.

قطع الدائرة

حاول إرجاع خطأ 503 أو 504 عندما يكون الخادم مُحمَّلاً بشكل زائد، بدلاً من إجبار المستخدم على الانتظار إلى ما لا نهاية.

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

حجب الخيط الرئيسي

لا يدعم PyTorch العمليات غير المتزامنة، لذا فإن الحسابات ستُعرقل تشغيل الخيط الرئيسي.

لذلك، من الأفضل تشغيل PyTorch في خيط أو عملية منفصلة تمامًا. وعندما يكون زمن استنتاج طلب واحد طويلًا بشكل خاص (أكثر من ثانية واحدة)، تزداد أهمية ذلك، لأن ذلك يعني أن كل استعلام أثناء الاستنتاج يجب أن ينتظر ثانية واحدة قبل أن يستلم خطأً حتى.

التجميع الديناميكي

يمكن أن يكون الدفع الديناميكي (Dynamic batching) فعالاً للغاية عند استخدامه في الإعداد الصحيح، لكنه ليس ضرورياً عندما تمرر طلباً واحداً في كل مرة (انظر الاستدلال الدفعي لمزيد من التفاصيل).