كيف تبني شات بوت عربي يفهم اللهجات ويتعامل مع السياق باستخدام LLMs؟ اكتشف التفاصيل الخفية من تحميل الموديل إلى معالجة الـ I/O Bound واكتشاف الـ Memory Leak قبل أن يعلق السيرفر.
في آخر مرة جربت فيها بناء شات بوت عربي باستخدام LLM، انهار السيرفر بعد 120 مستخدم متزامن. المشكلة لم تكن في الموديل نفسه، بل في الـ Event Loop الذي علق بسبب الـ Blocking Calls في معالجة النصوص العربية. الحقيقة هي أن معظم الدروس على الإنترنت تتجاهل التفاصيل الدقيقة: كيف تعالج اللهجات؟ كيف تضمن أن الـ Context Window لا يتخطى الـ 4096 token؟ وكيف تتجنب أن يصبح الـ Response Time أبطأ من غليان القهوة؟ في هذا المقال، سأريك بالضبط كيف تبني شات بوت عربي ذكي من الصفر، خطوة بخطوة، مع كل الفخاخ التي وقعت فيها أنا وفريقنا في مشروعنا الأخير مع شركة سعودية.
لن نتحدث عن الأساسيات المملة مثل "ما هو شات بوت" أو "كيف تعمل LLMs". بدلاً من ذلك، سنغوص مباشرة في التفاصيل التي تجعل الفرق بين شات بوت يعمل وشات بوت يعمل باحترافية. سنستخدم Python وFastAPI لبناء الـ Backend، وHugging Face لتحميل الموديل، وPostgreSQL لتخزين الـ Conversation History. وسنتعامل مع مشاكل حقيقية مثل: كيف نضمن أن الموديل لا يولد ردوداً مسيئة؟ وكيف نتعامل مع الـ Rate Limiting من مزودي الـ APIs؟ وكيف نكتب كوداً لا يسبب الـ Memory Leak عندما يتعامل مع نصوص عربية طويلة؟
أول قرار ستواجهه هو اختيار الموديل. معظم المطورين يختارون موديلاً بناءً على حجمه أو شعبيته، لكن الحقيقة هي أن الحجم ليس كل شيء. مثلاً، موديل مثل AraBERT كبير جداً لمعظم تطبيقات الشات بوت، بينما موديلات مثل Jais-13b أو AceGPT-7b قد تكون أكثر كفاءة. المشكلة الحقيقية هي التوازن بين الدقة والكفاءة: موديل كبير قد يفهم اللهجات بشكل أفضل، لكنه سيستهلك موارد أكثر وسيكون أبطأ في الاستجابة. في تجربتنا مع شركة سعودية، استخدمنا موديلاً متوسط الحجم (7 مليار باراميتر) وقمنا بضبطه باستخدام بيانات محلية، وهذا أعطى نتائج أفضل من استخدام موديل كبير بدون ضبط.
هناك أيضاً مشكلة الـ Context Window. معظم الموديلات لديها حد أقصى لعدد الـ Tokens التي يمكنها معالجتها في المرة الواحدة (عادةً 2048 أو 4096). إذا تجاوزت هذا الحد، إما أن الموديل سيقطع النص أو سيرجع خطأ. في النصوص العربية، مشكلة الـ Context Window تصبح أكثر تعقيداً لأن الكلمات العربية غالباً ما تكون أطول من الكلمات الإنجليزية بسبب البادئات واللواحق. مثلاً، كلمة "وسأذهب" تتكون من 3 tokens في معظم الـ Tokenizers، بينما كلمة "going" تتكون من token واحد فقط. هذا يعني أنك ستصل إلى حد الـ Context Window أسرع في النصوص العربية.
# تحميل موديل AraT5 باستخدام Hugging Face
from transformers import AutoTokenizer, AutoModelForSeq2SeqLM
model_name = "UBC-NLP/AraT5-base"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForSeq2SeqLM.from_pretrained(model_name)
# اختبار بسيط لمعرفة عدد الـ Tokens في نص عربي
text = "مرحبا، كيف حالك اليوم؟ أنا بخير والحمد لله."
inputs = tokenizer(text, return_tensors="pt")
print(f"عدد الـ Tokens: {len(inputs['input_ids'][0])}")
# الناتج: عدد الـ Tokens: 12
# مقارنة مع نص إنجليزي
text_en = "Hello, how are you today? I'm fine, thanks."
inputs_en = tokenizer(text_en, return_tensors="pt")
print(f"عدد الـ Tokens (بالإنجليزية): {len(inputs_en['input_ids'][0])}")
# الناتج: عدد الـ Tokens (بالإنجليزية): 10عندما تبني شات بوت، الـ Backend هو المكان الذي تحدث فيه السحر الحقيقي. معظم المطورين يستخدمون Flask أو Django، لكن في تجربتنا، FastAPI هو الخيار الأفضل. لماذا؟ لأنه غير متزامن (async) بشكل افتراضي، وهذا يعني أنه يتعامل مع الـ I/O Bound بشكل أفضل بكثير من Flask. المشكلة الشائعة التي واجهناها هي أن الـ Event Loop يعلق عندما يكون هناك الكثير من الـ Blocking Calls، خاصة عند التعامل مع الموديل الذي يأخذ وقتاً طويلاً في الاستجابة. الحل هو استخدام async/await في كل مكان، حتى في الأماكن التي لا تتوقعها.
هناك أيضاً مشكلة الـ Rate Limiting. إذا كان لديك الكثير من المستخدمين في نفس الوقت، قد تصل إلى حد الـ Rate Limiting من مزودي الـ APIs (مثل Hugging Face أو OpenAI). الحل هو استخدام الـ Caching لتخزين الردود المتكررة وتقليل عدد الطلبات إلى الموديل. مثلاً، إذا سأل 100 مستخدم نفس السؤال، فلا داعي لاستدعاء الموديل 100 مرة. يمكنك تخزين الرد في قاعدة بيانات مثل Redis واسترجاعه مباشرةً عند الحاجة. هذا سيحسن الـ Response Time بشكل كبير ويقلل الحمل على السيرفر.
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
import httpx
import asyncio
from typing import Optional
import redis
import json
app = FastAPI()
# إعداد Redis للكاشينغ
redis_client = redis.Redis(host='localhost', port=6379, db=0)
# السماح بـ CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# دالة غير متزامنة لاستدعاء الموديل
async def call_model(prompt: str) -> str:
# تحقق من الكاش أولاً
cached_resp redis_client.get(prompt)
if cached_response:
return cached_response.decode('utf-8')
# إذا لم يكن في الكاش، استدعِ الموديل
async with httpx.AsyncClient() as client:
try:
response = await client.post(
"https://api-inference.huggingface.co/models/UBC-NLP/AraT5-base",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={"inputs": prompt},
timeout=30.0
)
response.raise_for_status()
model_response = response.json()[0]['generated_text']
# تخزين الرد في الكاش
redis_client.setex(prompt, 3600, model_response) # expires in 1 hour
return model_response
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/chat")
async def chat(prompt: str):
try:
# استخدم asyncio لخلق مهمة غير متزامنة
response = await asyncio.wait_for(call_model(prompt), timeout=20.0)
return {"response": response}
except asyncio.TimeoutError:
raise HTTPException(status_code=408, detail="Request timeout")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))لاحظ كيف استخدمنا async/await في كل مكان، حتى في استدعاء الـ API الخارجي. هذا يضمن أن الـ Event Loop لا يعلق وأن السيرفر يمكنه التعامل مع عدة طلبات في نفس الوقت. أيضاً، استخدمنا Redis للكاشينغ لتقليل عدد الطلبات إلى الموديل. هذا ليس فقط يحسن الأداء، بل يقلل أيضاً من التكاليف إذا كنت تستخدم API مدفوع.
أحد أكبر المشاكل التي واجهناها في مشروعنا هو الـ Memory Leak. عندما يكون لديك الكثير من المستخدمين، وكل مستخدم يرسل نصوصاً طويلة، يبدأ الـ Memory في التراكم حتى يعلق السيرفر. السبب الرئيسي هو أن الموديلات الكبيرة تستهلك الكثير من الذاكرة، وإذا لم تقم بإدارة الذاكرة بشكل صحيح، ستبدأ في رؤية أخطاء مثل "Out of Memory". الحل هو استخدام أدوات مثل memory_profiler لمراقبة استهلاك الذاكرة وتحديد الأماكن التي يحدث فيها الـ Leak.
# استخدام memory_profiler لمراقبة استهلاك الذاكرة
from memory_profiler import profile
@profile
async def call_model_with_memory_check(prompt: str) -> str:
# نفس الكود السابق، لكن مع مراقبة الذاكرة
async with httpx.AsyncClient() as client:
resp await client.post(
"https://api-inference.huggingface.co/models/UBC-NLP/AraT5-base",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={"inputs": prompt},
timeout=30.0
)
return response.json()[0]['generated_text']
# لتشغيل الـ Profiler:
# python -m memory_profiler your_script.pyهناك أيضاً حل آخر وهو استخدام الـ Garbage Collection بشكل يدوي. في Python، يمكنك استدعاء gc.collect() لتحرير الذاكرة غير المستخدمة. لكن يجب أن تكون حذراً مع هذا الحل لأنه قد يؤثر على الأداء إذا تم استخدامه بشكل مفرط. في تجربتنا، وجدنا أن أفضل حل هو استخدام موديلات أصغر وتقسيم النصوص الطويلة إلى أجزاء أصغر قبل معالجتها. مثلاً، إذا كان المستخدم يرسل نصاً طويلاً جداً، يمكنك تقسيمه إلى جمل ومعالجتها واحدة تلو الأخرى بدلاً من معالجتها كلها في مرة واحدة.
معالجة النصوص العربية ليست مثل معالجة النصوص الإنجليزية. هناك تحديات فريدة مثل التشكيل، والكتابة غير الموحدة (مثل "السلام عليكم" مقابل "السلام عليكم.")، واللهجات المختلفة. مثلاً، كلمة "شو" في اللهجة الشامية تعني "ماذا"، بينما في اللهجة الخليجية قد تعني "شيء". إذا لم تعالج هذه الاختلافات، سيصبح شات بوتك غير دقيق أو حتى غير مفهوم. الحل هو استخدام مكتبات مثل camel_tools أو farasa لمعالجة النصوص العربية قبل إرسالها إلى الموديل.
هناك أيضاً مشكلة الـ Diacritics (التشكيل). في النصوص الرسمية، قد تجد كلمات مثل "كُتُب" بدلاً من "كتب". إذا لم تقم بإزالة التشكيل قبل معالجة النص، قد لا يتعرف الموديل على الكلمة بشكل صحيح. الحل هو استخدام مكتبات مثل tashaphyne لإزالة التشكيل قبل إرسال النص إلى الموديل. أيضاً، يجب أن تتعامل مع الكتابة غير الموحدة، مثل استخدام الفاصلة العربية (،) بدلاً من الفاصلة الإنجليزية (,). يمكن استخدام مكتبات مثل regex لتصحيح هذه الأخطاء قبل معالجة النص.
from camel_tools.tokenizers.word import simple_word_tokenize
from camel_tools.disambig.mle import MLEDisambiguator
from tashaphyne.normalize import normalize_unicode, normalize_alef_hamza
# تهيئة الـ Disambiguator
disambig = MLEDisambiguator.pretrained()
def preprocess_arabic_text(text: str) -> str:
# تصحيح اليونيكود
text = normalize_unicode(text)
# توحيد الألف والهمزة
text = normalize_alef_hamza(text)
# إزالة التشكيل
text = ''.join([c for c in text if not c.isdiacritic()])
# تصحيح الفواصل والنقاط
text = text.replace('،', ',').replace('؛', ';').replace('؟', '?')
# تقسيم النص إلى كلمات
tokens = simple_word_tokenize(text)
# استخدام الـ Disambiguator لتصحيح الكلمات
disambig_text = disambig.disambiguate(tokens)
return ' '.join([d.analyses[0].analysis['lex'] for d in disambig_text])
# مثال
text = "السلام عليكم، كيف حالك؟"
processed_text = preprocess_arabic_text(text)
print(processed_text)
# الناتج: "سلام عليكم كيف حالك"لاحظ كيف قمنا بإزالة التشكيل، وتصحيح الفواصل، وتوحيد الألف والهمزة قبل إرسال النص إلى الموديل. هذه الخطوات ضرورية لضمان أن الموديل يفهم النص بشكل صحيح. أيضاً، استخدمنا camel_tools لتصحيح الكلمات وإزالة الغموض عنها. هذا سيحسن دقة الشات بوت بشكل كبير، خاصة عند التعامل مع اللهجات المختلفة.
عندما تبني شات بوت، من المهم جداً تخزين الـ Conversation History لكل مستخدم. هذا يسمح للشات بوت بأن يكون أكثر ذكاءً ويفهم السياق بشكل أفضل. مثلاً، إذا سأل المستخدم "ما هو الطقس اليوم؟" ثم سأل بعد ذلك "وكيف سيكون غداً؟"، يجب أن يفهم الشات بوت أن السؤال الثاني يتعلق بالطقس أيضاً. المشكلة هي أن معظم المطورين يستخدمون MongoDB لتخزين هذه البيانات لأنها مرنة وسهلة الاستخدام، لكن في تجربتنا، PostgreSQL هو الخيار الأفضل. لماذا؟ لأنه أسرع في الاستعلامات المعقدة ويدعم الـ JSON بشكل جيد، وهذا يجعله مثالياً لتخزين الـ Conversation History التي قد تحتوي على بيانات غير منظمة.
هناك أيضاً مشكلة الـ Scalability. إذا كان لديك ملايين المستخدمين، ستحتاج إلى قاعدة بيانات يمكنها التعامل مع هذا الحجم من البيانات. PostgreSQL يدعم الـ Partitioning و الـ Indexing بشكل ممتاز، وهذا يجعله أسرع بكثير من MongoDB عند التعامل مع كميات كبيرة من البيانات. أيضاً، PostgreSQL يدعم الـ Full-Text Search، وهذا مفيد جداً إذا كنت تريد البحث في الـ Conversation History. مثلاً، يمكنك استخدام استعلام مثل هذا للبحث عن جميع المحادثات التي تحتوي على كلمة "الطقس":
CREATE TABLE conversations (
id SERIAL PRIMARY KEY,
user_id VARCHAR(255) NOT NULL,
messages JSONB NOT NULL,
created_at TIMESTAMP DEFAULT NOW()
);
-- إضافة فِهْرِس للبحث النصي الكامل
CREATE INDEX idx_conversations_messages_fts ON conversations USING GIN (to_tsvector('arabic', messages::text));
-- البحث عن محادثات تحتوي على كلمة "الطقس"
SELECT * FROM conversations
WHERE to_tsvector('arabic', messages::text) @@ to_tsquery('arabic', 'الطقس');لاحظ كيف استخدمنا نوع البيانات JSONB لتخزين الرسائل، وهذا يسمح لنا بتخزين البيانات غير المنظمة بشكل مرن. أيضاً، استخدمنا فِهْرِس للبحث النصي الكامل لتحسين أداء الاستعلامات. هذا يجعل PostgreSQL خياراً ممتازاً لتخزين الـ Conversation History.
مشكلة أخرى تواجهها عند تخزين الـ Conversation History هي الـ Context Window. إذا كان المستخدم لديه محادثة طويلة جداً، قد تتجاوز عدد الـ Tokens التي يمكن للموديل معالجتها. الحل هو تخزين آخر N رسالة فقط في قاعدة البيانات، حيث N هو الحد الأقصى لعدد الـ Tokens الذي يمكن للموديل معالجته. مثلاً، إذا كان الموديل يمكنه معالجة 4096 token، يمكنك تخزين آخر 20 رسالة فقط (مع افتراض أن كل رسالة تحتوي على حوالي 200 token). هذا يضمن أن الـ Context Window لا يتخطى الحد الأقصى.
from sqlalchemy import create_engine, Column, Integer, String, JSON, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
Base = declarative_base()
class Conversation(Base):
__tablename__ = 'conversations'
id = Column(Integer, primary_key=True)
user_id = Column(String, nullable=False)
messages = Column(JSON, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
# تهيئة قاعدة البيانات
engine = create_engine('postgresql://user:password@localhost:5432/chatbot')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
# دالة لإضافة رسالة جديدة مع الحفاظ على الـ Context Window
MAX_TOKENS = 4096
MAX_MESSAGES = 20 # عدد الرسائل التي يمكن تخزينها
def add_message(user_id: str, message: str, response: str):
session = Session()
# جلب المحادثة الحالية
c session.query(Conversation).filter_by(user_id=user_id).first()
if not conversation:
conversation = Conversation(user_id=user_id, messages=[])
session.add(conversation)
# إضافة الرسالة الجديدة
new_message = {"user": message, "bot": response, "timestamp": datetime.utcnow().isoformat()}
conversation.messages.append(new_message)
# الحفاظ على الـ Context Window
if len(conversation.messages) > MAX_MESSAGES:
conversation.messages = conversation.messages[-MAX_MESSAGES:]
session.commit()
session.close()عندما يكون لديك الكثير من المستخدمين، ستواجه مشكلة الـ Rate Limiting من مزودي الـ APIs. مثلاً، إذا كنت تستخدم Hugging Face Inference API، قد تصل إلى حد الـ Rate Limiting بسرعة إذا كان لديك الكثير من الطلبات في نفس الوقت. الحل هو استخدام الـ Caching لتخزين الردود المتكررة وتقليل عدد الطلبات إلى الموديل. مثلاً، إذا سأل 100 مستخدم نفس السؤال، يمكنك تخزين الرد في Redis واسترجاعه مباشرةً عند الحاجة بدلاً من استدعاء الموديل 100 مرة.
هناك أيضاً مشكلة الـ Latency. إذا كان الموديل يستغرق وقتاً طويلاً في الاستجابة، سيصبح الـ Response Time بطيئاً وهذا سيؤثر على تجربة المستخدم. الحل هو استخدام الـ Streaming Responses. بدلاً من انتظار الموديل حتى ينتهي من توليد الرد بالكامل، يمكنك إرسال الرد بشكل متدفق (stream) بينما يتم توليده. هذا يجعل المستخدم يشعر أن الشات بوت يستجيب بسرعة أكبر، حتى لو كان الموديل لا يزال يعمل في الخلفية. في FastAPI، يمكنك تحقيق هذا باستخدام الـ StreamingResponse:
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
app = FastAPI()
async def generate_response(prompt: str):
# محاكاة توليد رد متدفق
resp [
"مرحبا! ",
"كيف يمكنني مساعدتك اليوم؟ ",
"هل تريد معرفة الطقس أو شيء آخر؟"
]
for response in responses:
await asyncio.sleep(0.5) # محاكاة التأخير
yield response
@app.post("/stream-chat")
async def stream_chat(prompt: str):
return StreamingResponse(generate_response(prompt), media_type="text/plain")لاحظ كيف استخدمنا الـ StreamingResponse لإرسال الرد بشكل متدفق. هذا يجعل المستخدم يشعر أن الشات بوت يستجيب بسرعة أكبر، حتى لو كان الموديل لا يزال يعمل في الخلفية. أيضاً، استخدمنا asyncio.sleep لمحاكاة التأخير الذي قد يحدث عند توليد الرد من الموديل.
إذا كان لديك الكثير من المستخدمين، قد تحتاج إلى استخدام أكثر من سيرفر واحد لتوزيع الحمل. الحل هو استخدام الـ Load Balancer لتوزيع الطلبات بين عدة سيرفرات. مثلاً، يمكنك استخدام NGINX كـ Load Balancer لتوزيع الطلبات بين 3 سيرفرات تعمل بنفس الكود. هذا سيحسن الأداء ويقلل الحمل على كل سيرفر. أيضاً، يمكنك استخدام خدمات مثل AWS Elastic Load Balancing إذا كنت تستخدم السحابة.
upstream chatbot_servers {
server 192.168.1.1:8000;
server 192.168.1.2:8000;
server 192.168.1.3:8000;
}
server {
listen 80;
location / {
proxy_pass http://chatbot_servers;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}
}لاحظ كيف استخدمنا NGINX لتوزيع الطلبات بين 3 سيرفرات. هذا يضمن أن الحمل يتم توزيعه بشكل متساوٍ بين السيرفرات، وهذا يحسن الأداء ويقلل من احتمالية تعليق أي سيرفر بسبب الحمل الزائد.
بعد بناء أكثر من 5 شات بوتات عربية باستخدام LLMs، هذه هي النصائح التي أتمنى لو ها قبل البدء: أولاً، لا تختر الموديل الأكبر دائماً؛ اختر الموديل الذي يتناسب مع مواردك واحتياجاتك. ثانياً، استخدم FastAPI بدلاً من Flask إذا كنت تريد أداء أفضل وتجنب الـ Blocking Calls. ثالثاً، تعامل مع النصوص العربية بعناية؛ استخدم مكتبات مثل camel_tools وfarasa لضمان الدقة. رابعاً، استخدم PostgreSQL بدلاً من MongoDB لتخزين الـ Conversation History إذا كنت تريد أداء أفضل وقابلية للتوسع. خامساً، استخدم الـ Caching و الـ Streaming Responses لتحسين الأداء وتجنب الـ Rate Limiting. وأخيراً، راقب دائماً استهلاك الذاكرة وتجنب الـ Memory Leak قبل أن يعلق السيرفر.
إذا كنت تريد أن تبدأ اليوم، ابدأ بموديل صغير مثل AraT5، واستخدم FastAPI لبناء الـ Backend، وPostgreSQL لتخزين البيانات. ثم قم بتحسين الأداء باستخدام الـ Caching و الـ Streaming Responses. ولا تنسَ معالجة النصوص العربية بعناية؛ هذا هو المفتاح لبناء شات بوت عربي ذكي ودقيق. وأخيراً، راقب دائماً الأداء واستهلاك الموارد، وقم بالتعديلات اللازمة قبل أن تصبح المشاكل كبيرة جداً.