Qwen

سلسلة نماذج اللغة المتقدمة

أولاً: ما هو نموذج Qwen2.5-1.5B وما مدى «خفّته»؟

يُعتبر نموذج Qwen2.5-1.5B من تطوير Alibaba Cloud خطوة ثورية في عالم النماذج الصغيرة (SLMs)، حيث يجمع بين الكفاءة العالية واستهلاك الموارد المنخفض جداً، مما يجعله المحرك المثالي لـ ONYX Engine.

حجم المعاملات (Parameters)

يحتوي على 1.5 مليار باراميتر. هذا الرقم يعتبر 'نقطة ذهبية' لأنه صغير بما يكفي ليعمل على أجهزة المستخدمين العادية، وكبير بما يكفي ليحتفظ بقدرات منطقية عالية.

الأداء واللغة العربية

تم تدريب النموذج على مجموعات بيانات ضخمة، مما منحه تفوقاً ملحوظاً في فهم التعليمات البرمجية والتعامل مع اللغة العربية بطلاقة، متفوقاً على نماذج أكبر منه حجماً.

متطلبات التشغيل (Efficiency)

بفضل تقنية float16، يستهلك النموذج حوالي 3-4 جيجابايت فقط من الـ VRAM. يمكن تشغيله بسلاسة على كروت شاشة اقتصادية أو حتى على المعالج (CPU) بذاكرة 8 جيجا.

ثانياً: شرح كود السيرفر (FastAPI) سطراً بسطر

1. استيراد المكتبات (Dependencies)

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer
import torch
from threading import Thread

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer
import torch
from threading import Thread

نستورد FastAPI لبناء السيرفر، وStreamingResponse للسماح بإرسال الردود كدفق مستمر. نستخدم Thread لضمان تشغيل عملية التوليد في خلفية النظام دون حظر (Block) استجابة السيرفر للطلبات الأخرى.

2. تهيئة المعالج (Tokenizer)

app = FastAPI(title="ONYX Engine - Qwen2.5-1.5B")
model_id = "Qwen/Qwen2.5-1.5B-Instruct"

tokenizer = AutoTokenizer.from_pretrained(model_id, use_fast=True)
dtype = torch.float16 if torch.cuda.is_available() else torch.float32

app = FastAPI(title="ONYX Engine - Qwen2.5-1.5B")
model_id = "Qwen/Qwen2.5-1.5B-Instruct"

tokenizer = AutoTokenizer.from_pretrained(model_id, use_fast=True)
dtype = torch.float16 if torch.cuda.is_available() else torch.float32

الـ Tokenizer هو المسؤول عن تحويل كلماتنا إلى متجهات رقمية. نحدد نوع البيانات (dtype) بناءً على توفر كرت الشاشة (CUDA)؛ الـ float16 يقلل استهلاك الذاكرة للنصف مع الحفاظ على الدقة.

3. تحميل النموذج للذاكرة

model = AutoModelForCausalLM.from_pretrained(
    model_id,
    dtype=dtype,
    device_map="auto"
)

model = AutoModelForCausalLM.from_pretrained(
    model_id,
    dtype=dtype,
    device_map="auto"
)

هنا يتم تحميل وزن النموذج. خاصية device_map='auto' هي سحر برمجيات Hugging Face، حيث تقوم تلقائياً بتوزيع أجزاء النموذج بين الـ GPU والـ RAM بناءً على المساحة المتوفرة.

4. نقطة فحص الحالة (Health Check)

@app.get("/")
async def root():
    return {"status": "ONYX Engine is Running"}

@app.get("/")
async def root():
    return {"status": "ONYX Engine is Running"}

مسار بسيط للتأكد من أن السيرفر يعمل بشكل صحيح واستجابة API سليمة قبل إرسال طلبات المعالجة الثقيلة.

5. استقبال طلبات التوليد

@app.post("/predict")
async def generate(request: Request):
    data = await request.json()
    messages = data.get("messages", [])

@app.post("/predict")
async def generate(request: Request):
    data = await request.json()
    messages = data.get("messages", [])

نستخدم POST لاستقبال مصفوفة الرسائل (التي تحتوي على تاريخ المحادثة) من طرف العميل (Frontend) بصيغة JSON.

6. تحويل التنسيق (Chat Template)

model_inputs = tokenizer.apply_chat_template(
    messages,
    tokenize=True,
    add_generation_prompt=True,
    return_tensors="pt"
).to(model.device)

model_inputs = tokenizer.apply_chat_template(
    messages,
    tokenize=True,
    add_generation_prompt=True,
    return_tensors="pt"
).to(model.device)

كل نموذج له طريقة في فهم المحادثة (User vs Assistant). الـ apply_chat_template يضيف الرموز الخاصة بالنموذج تلقائياً ثم ينقل البيانات إلى نفس الجهاز (Device) الموجود عليه النموذج.

7. إعداد البث التدريجي (Streaming)

streamer = TextIteratorStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True)

generation_kwargs = dict(
    **model_inputs,
    streamer=streamer,
    max_new_tokens=512,
    do_sample=True,
    temperature=0.7,
    top_p=0.9,
)

streamer = TextIteratorStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True)

generation_kwargs = dict(
    **model_inputs,
    streamer=streamer,
    max_new_tokens=512,
    do_sample=True,
    temperature=0.7,
    top_p=0.9,
)

الـ streamer هو المراقب الذي يمسك الكلمات فور خروجها. نضبط خيارات التوليد مثل الـ Temperature (درجة الإبداع/العشوائية) والـ Top-p للتحكم في جودة النص.

8. إطلاق عملية التوليد

thread = Thread(target=model.generate, kwargs=generation_kwargs)
thread.start()

thread = Thread(target=model.generate, kwargs=generation_kwargs)
thread.start()

نبدأ التوليد في 'خيط برمي' منفصل. هذا يسمح للدالة الأساسية بالانتقال فوراً للخطوة التالية وبدء إرسال الكلمات للعميل بينما النموذج لا يزال يعمل.

9. إرسال الرد للعميل

def generate_chunks():
    for new_text in streamer:
        if new_text:
            yield new_text

return StreamingResponse(generate_chunks(), media_type="text/plain")

def generate_chunks():
    for new_text in streamer:
        if new_text:
            yield new_text

return StreamingResponse(generate_chunks(), media_type="text/plain")

الدالة الموّلدة (Generator) تقوم بعمل yield لكل كلمة جديدة تصل من الـ streamer. الـ StreamingResponse يبقي الاتصال مفتوحاً ويرسل الكلمات فوراً لتظهر لدى المستخدم كأنها تكتب حياً.

Author: ONYX (2026)

Base Model: Qwen2.5-1.5B