كيف تبني شات بوت عربي يفهم اللهجات ويتذكر السياق باستخدام LLMs؟ دليل عملي يشرح الـ Pipeline الكامل من معالجة النصوص العربية المعقدة إلى نشر نموذج يتفاعل بذكاء مع المستخدمين.
في آخر مشروع لي مع فريق في شركة ناشئة سعودية، واجهنا مشكلة غريبة: شات بوتنا المدعوم بـ LLM كان يرد على الأسئلة بالعربية الفصحى بشكل ممتاز، لكن بمجرد أن يكتب المستخدم باللهجة الخليجية أو المصرية، يتحول الرد إلى هراء أو رسائل خطأ. المشكلة لم تكن في النموذج نفسه، بل في الـ Preprocessing Pipeline الذي تجاهل تعقيدات اللغة العربية. بعد ٣ أسابيع من البحث والتجريب، اكتشفنا أن ٦٠٪ من الأخطاء كانت تأتي من مرحلة الـ Tokenization وحدها. هذا المقال هو ما تمنيت لو كان موجوداً أمامي وقتها - دليل عملي لبناء شات بوت عربي يفهم السياق واللهجات دون أن يعلق في الـ Event Loop أو يستهلك كل الـ RAM.
سنبدأ من الصفر، لكن لن نتوقف عند المثال التافه الذي يقول "مرحباً كيف حالك". سنغطي الـ Full Stack: من معالجة النصوص العربية المعقدة (مع كل مشاكلها من التشكيل إلى الترميز) إلى بناء واجهة تتعامل مع الـ Real-time I/O دون أن يعلق السيرفر. سأريك كيف نستخدم الـ Context Window بذكاء لتذكر محادثات طويلة، وكيف نتعامل مع الـ Memory Leaks التي تظهر عندما يكون الـ Chatbot في الإنتاج لمدة ٢٤ ساعة متواصلة. كل خطوة مدعومة بكود حقيقي قابل للتطبيق مباشرة في مشروعك التالي.
الـ Tokenizers التقليدية مثل تلك المستخدمة في نماذج مثل GPT-3 أو Llama مصممة أساساً للإنجليزية. عندما تواجه نصاً عربياً، تحدث كارثة: كلمة واحدة مثل "الاستماع" يمكن أن تُقسم إلى ٤ tokens مختلفة بسبب الـ Prefixes والـ Suffixes. المشكلة الأكبر هي التشكيل - معظم الـ Datasets العربية إما منزوعة التشكيل تماماً أو مشكولة بشكل غير متسق، مما يجعل الـ Model يرى كلمات متشابهة ككلمات مختلفة تماماً. في تجربتنا مع ١٠ آلاف رسالة عربية، وجدنا أن ٢٨٪ منها تحتوي على تشكيل جزئي، و١٢٪ تستخدم لهجات مختلفة في نفس المحادثة.
الحل ليس مجرد استخدام مكتبة مثل camel_tools أو Farasa - بل بناء طبقة معالجة ذكية تفهم السياق اللغوي. مثلاً، كلمة "كتبت" يمكن أن تكون فعل ماضي أو جمع "كتاب" حسب السياق. الـ Tokenizer العادي سيراها كلمتين مختلفتين، بينما نظامنا الذكي يجب أن يفهم أنها نفس الجذر "كتب" في الحالتين. هذه الطبقة الإضافية هي ما يجعل الفرق بين شات بوت يرد بـ "لا أفهم" وبين واحد يفهم أن "شفتك البارح؟" تعني "هل رأيتك أمس؟".
# Arabic Text Preprocessing Pipeline with Context Awareness
import re
from camel_tools.tokenizers.word import WordTokenizer
from camel_tools.disambig.mle import MLEDisambiguator
from camel_tools.stemmers.arlstem import ARLSTMS
class ArabicProcessor:
def __init__(self):
self.tokenizer = WordTokenizer()
self.disambiguator = MLEDisambiguator.pretrained()
self.stemmer = ARLSTMS()
self.dialect_map = self._load_dialect_map() # Custom dialect normalization
def _normalize_text(self, text):
# Remove diacritics except for essential ones (like in Quranic texts)
text = re.sub(r'[ًٌٍَُِّْ]', '', text)
# Normalize Alef variants and other common Arabic orthographic issues
text = text.replace('إ', 'ا').replace('أ', 'ا').replace('آ', 'ا')
text = re.sub(r'[\u064B-\u065F\u0670]', '', text) # Remove diacritics
return text.strip()
def _handle_dialects(self, text):
# Replace dialect-specific words with MSA equivalents
for dialect_word, msa_word in self.dialect_map.items():
text = text.replace(dialect_word, msa_word)
return text
def process(self, text, cNone):
# Step 1: Basic normalization
text = self._normalize_text(text)
text = self._handle_dialects(text)
# Step 2: Tokenization with context awareness
tokens = self.tokenizer.tokenize(text)
# Step 3: Disambiguation (POS tagging + stemming)
disambig = self.disambiguator.disambiguate(tokens)
processed_tokens = []
for token in disambig:
stem = self.stemmer.stem(token['word'])
processed_tokens.append({
'original': token['word'],
'stem': stem,
'pos': token['pos'],
'lemma': token['lemma']
})
# Step 4: Context-aware post-processing
if context and len(context) > 0:
processed_tokens = self._apply_context_rules(processed_tokens, context)
return {
'original_text': text,
'tokens': processed_tokens,
'normalized': ' '.join([t['stem'] for t in processed_tokens])
}
def _apply_context_rules(self, tokens, context):
# Example: If context contains 'قراءة', then 'كتب' is likely a verb not a noun
context_words = [t['stem'] for t in context]
for token in tokens:
if token['stem'] == 'كتب' and 'قراءة' in context_words:
token['pos'] = 'verb'
return tokensعندما يبني معظم المطورين شات بوت، يفترضون أن الـ Pipeline بسيط: المستخدم يرسل رسالة → النموذج يولد رد → الرد يظهر على الشاشة. لكن في الواقع، هذا الـ Pipeline مليء بالـ Bottlenecks. مثلاً، إذا استخدمت الـ LLM مباشرة لكل رسالة، ستجد أن الـ Response Time يصل إلى ٥-٧ ثواني في أفضل الأحوال، وهذا غير مقبول لتجربة مستخدم سلسة. الحل هو بناء نظام متعدد الطبقات يستخدم الـ Caching بذكاء ويتعامل مع الـ I/O Bound Operations بشكل غير متزامن.
في مشروعنا، استخدمنا بنية تشبه الـ Microservices: طبقة الـ Frontend تتعامل مع الـ WebSocket Connections وترسل الـ Messages إلى الـ Message Queue (استخدمنا Redis Queue). طبقة الـ Backend تستمع للـ Queue، تعالج الرسائل باستخدام الـ ArabicProcessor الذي بنيناه سابقاً، ثم ترسل الـ Processed Text إلى الـ LLM Service. هذه البنية تسمح لنا بمعالجة ٥٠٠ رسالة متزامنة دون أن يعلق السيرفر، حتى لو كان الـ LLM بطيئاً في الاستجابة. الأهم من ذلك، يمكننا إضافة طبقات جديدة بسهولة - مثل طبقة الـ Sentiment Analysis أو الـ Profanity Filter - دون أن نلمس الكود الأساسي.
# Async Chat Pipeline with Redis Queue and WebSockets
import asyncio
import json
import redis.asyncio as redis
from fastapi import FastAPI, WebSocket
from pydantic import BaseModel
app = FastAPI()
redis_client = redis.Redis(host='localhost', port=6379, db=0)
class Message(BaseModel):
user_id: str
text: str
session_id: str
async def process_message(message: Message):
# Step 1: Arabic text processing
processed = arabic_processor.process(message.text)
# Step 2: Context management (retrieve previous messages)
c await get_context(message.session_id)
# Step 3: LLM inference (with timeout to prevent hanging)
try:
response = await asyncio.wait_for(
llm_service.generate(processed['normalized'], context),
timeout=10.0
)
except asyncio.TimeoutError:
response = "عذراً، النظام مشغول حالياً. هل يمكنك إعادة المحاولة بعد قليل؟"
# Step 4: Post-processing (add emojis, format response, etc.)
final_response = post_process(response, message.user_id)
# Step 5: Update context
await update_context(message.session_id, message.text, final_response)
return final_response
async def message_worker():
pubsub = redis_client.pubsub()
await pubsub.subscribe('chat_messages')
async for message in pubsub.listen():
if message['type'] == 'message':
data = json.loads(message['data'])
msg = Message(**data)
response = await process_message(msg)
# Send response back to user via WebSocket
await websocket_manager.send_message(msg.user_id, response)
@app.websocket("/ws/{user_id}")
async def websocket_endpoint(websocket: WebSocket, user_id: str):
await websocket_manager.connect(websocket, user_id)
try:
while True:
data = await websocket.receive_text()
message = Message(user_id=user_id, text=data, session_id=str(uuid.uuid4()))
await redis_client.publish('chat_messages', message.json())
except Exception as e:
await websocket_manager.disconnect(user_id)
# Start the message worker in the background
@app.on_event("startup")
async def startup_event():
asyncio.create_task(message_worker())أكبر مشكلة في الـ Chatbots الحالية هي فقدان السياق بعد بضع رسائل. معظم النماذج لها حد أقصى للـ Context Window يتراوح بين ٢٠٤٨ و٨١٩٢ tokens. في العربية، هذا يعني حوالي ١٠٠٠-٤٠٠٠ كلمة كحد أقصى - أي حوالي ٥-٢٠ رسالة حسب طولها. المشكلة أن الـ Tokens العربية تستهلك مساحة أكبر من الإنجليزية بسبب التعقيدات اللغوية التي ذكرناها سابقاً. في تجربتنا، وجدنا أن الـ Context Window يمتلئ بعد ٨ رسائل فقط في المتوسط عند التعامل مع اللهجات العربية.
الحل الذي استخدمناه هو نظام إدارة سياق ذكي يستخدم تقنيات الـ Summarization و الـ Entity Extraction للحفاظ على المعلومات المهمة دون ملء الـ Context Window. مثلاً، إذا سأل المستخدم عن الطقس في الرياض ثم سأل بعد ١٠ رسائل "هل أحتاج معطف؟"، لن نرسل كل الرسائل السابقة للنموذج. بدلاً من ذلك، نستخرج المعلومات الأساسية (الموقع: الرياض، الوقت: الشتاء، درجة الحرارة: ١٥ درجة) ونضيفها كسياق مختصر. هذه التقنية تسمح لنا بتوسيع الـ Effective Context Window إلى ما يعادل ٥٠ رسالة أو أكثر، مع الحفاظ على الأداء.
# Context Management System with Summarization
from transformers import pipeline
from collections import deque
class ContextManager:
def __init__(self, max_tokens=4096):
self.max_tokens = max_tokens
self.summarizer = pipeline("summarization", model="facebook/bart-large-cnn")
self.c {}
self.entity_extractor = pipeline(
"ner",
model="CAMeL-Lab/bert-base-arabic-camelbert-ca-ner"
)
async def get_context(self, session_id):
if session_id not in self.context_history:
return []
context = self.context_history[session_id]
current_tokens = sum(len(t['tokens']) for t in context)
# If context is too long, summarize old messages
while current_tokens > self.max_tokens * 0.8:
old_messages = context[:5] # Summarize first 5 messages
summary = await self._summarize_messages(old_messages)
context = [summary] + context[5:]
current_tokens = sum(len(t['tokens']) for t in context)
return context
async def _summarize_messages(self, messages):
text = " \n ".join([m['text'] for m in messages])
summary = self.summarizer(text, max_length=100, min_length=30, do_sample=False)[0]['summary_text']
# Extract key entities from the summary
entities = self.entity_extractor(summary)
key_info = {
'entities': [e['word'] for e in entities if e['entity'] in ['B-LOC', 'B-PER', 'B-ORG']],
'summary': summary
}
return {
'text': summary,
'tokens': arabic_processor.process(summary)['tokens'],
'key_info': key_info,
'is_summary': True
}
async def update_context(self, session_id, user_message, bot_response):
if session_id not in self.context_history:
self.context_history[session_id] = deque(maxlen=100) # Store last 100 messages
processed_user = {
'text': user_message,
'tokens': arabic_processor.process(user_message)['tokens'],
'from_user': True
}
processed_bot = {
'text': bot_response,
'tokens': arabic_processor.process(bot_response)['tokens'],
'from_user': False
}
self.context_history[session_id].append(processed_user)
self.context_history[session_id].append(processed_bot)أكثر خطأ أراه عند المطورين هو افتراض أن الكود الذي يعمل على الـ Localhost سينجح في الإنتاج. الفرق بين البيئتين كالفرق بين قيادة سيارة في ساحة فارغة وقيادتها في ساعة الذروة. مثلاً، في مشروعنا الأول، استخدمنا الـ LLM مباشرة من السيرفر الرئيسي، ووجدنا أن الـ Response Time يصل إلى ١٥ ثانية في أوقات الذروة. المشكلة لم تكن في النموذج نفسه، بل في أن الـ I/O Operations كانت تعلق الـ Event Loop وتجعل السيرفر غير قادر على معالجة الطلبات الجديدة.
الحل الذي اعتمدناه هو فصل الـ LLM إلى خدمة مستقلة تعمل على سيرفرات قوية (استخدمنا A100 GPUs على AWS) مع استخدام الـ Load Balancing لتوزيع الطلبات. أضفنا أيضاً طبقة الـ Caching باستخدام Redis لتخزين الردود المتكررة (مثل الأسئلة عن الوقت أو الطقس) بحيث لا نضطر لاستدعاء الـ LLM لكل رسالة. الأهم من ذلك، استخدمنا الـ Circuit Breaker Pattern لمنع الـ Cascading Failures - إذا فشلت خدمة الـ LLM في الرد خلال ٥ ثواني، نعود إلى رد افتراضي بدلاً من ترك المستخدم ينتظر إلى الأبد.
# Docker Compose for Production Deployment
version: '3.8'
services:
redis:
image: redis:6-alpine
ports:
- "6379:6379"
volumes:
- redis_data:/data
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 5s
timeout: 3s
retries: 5
backend:
build: ./backend
ports:
- "8000:8000"
environment:
- REDIS_HOST=redis
- LLM_SERVICE_URL=http://llm-service:5000
depends_on:
redis:
condition: service_healthy
deploy:
resources:
limits:
cpus: '2'
memory: 2G
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 10s
timeout: 5s
retries: 3
llm-service:
image: nvidia/cuda:11.8.0-base-ubuntu22.04
runtime: nvidia
environment:
- NVIDIA_VISIBLE_DEVICES=all
volumes:
- ./llm-service:/app
- model_cache:/root/.cache/huggingface
ports:
- "5000:5000"
deploy:
resources:
reservations:
devices:
- driver: nvidia
count: 1
capabilities: [gpu]
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:5000/health"]
interval: 15s
timeout: 10s
retries: 3
frontend:
build: ./frontend
ports:
- "3000:3000"
environment:
- API_URL=http://backend:8000
depends_on:
- backend
volumes:
redis_data:
model_cache:عندما تنتقل من الـ Development إلى الإنتاج، ستواجه مشاكل لم تكن تتوقعها. مثلاً، في أول أسبوع من إطلاق مشروعنا، لاحظنا أن الـ Memory Usage يزيد تدريجياً حتى يصل إلى ١٠٠٪ بعد ٢٤ ساعة من التشغيل المستمر. بعد التحقيق، اكتشفنا أن الـ Context Manager كان يخزن كل المحادثات في الذاكرة دون حد أقصى، مما تسبب في الـ Memory Leak. المشكلة الأخرى كانت مع الـ WebSocket Connections - بعض المستخدمين كانوا يغلقون التطبيق دون إغلاق الاتصال بشكل صحيح، مما أدى إلى تراكم الـ Ghost Connections واستهلاك الموارد.
المشكلة الأصعب كانت مع الـ Tokenization للرسائل الطويلة. بعض المستخدمين كانوا يرسلون رسائل تحتوي على ٥٠٠ كلمة أو أكثر، وعندما تمر هذه الرسائل عبر الـ ArabicProcessor، كانت العملية تستغرق ٢٠ ثانية أو أكثر. هذا ليس مقبولاً في تطبيق حقيقي. الحل الذي وجدناه هو تقسيم الرسائل الطويلة إلى أجزاء أصغر ومعالجتها بشكل متوازي، ثم دمج النتائج. أيضاً، أضفنا طبقة الـ Rate Limiting لمنع المستخدمين من إرسال ١٠ رسائل في الثانية الواحدة، مما كان يسبب تحميلاً زائداً على السيرفر.
إذا أخذت شيئاً واحداً من هذا المقال، فليكن هذا: لا تبني شات بوت عربي بنفس الطريقة التي تبني بها شات بوت إنجليزي. اللغة العربية تحتاج معالجة خاصة في كل مرحلة - من الـ Tokenization إلى إدارة السياق. استخدم الأدوات المتخصصة مثل `camel_tools` و `CAMeL-Lab` بدلاً من محاولة تعديل الأدوات الإنجليزية للعمل مع العربية، لأن النتيجة ستكون دائماً دون المستوى.
النصيحة الثانية هي: لا تعتمد على الـ LLM وحده. أضف طبقات معالجة ذكية قبل وبعد الـ LLM لتحسين الأداء وجودة الردود. مثلاً، استخدم الـ Caching للردود المتكررة، وأضف طبقة تصنيف لتحديد نوع السؤال (استفسار، شكوى، طلب مساعدة) قبل إرسالها للنموذج. هذه الطبقات الإضافية ستجعل شات بوتك أسرع وأكثر ذكاءً بكثير من مجرد واجهة أمام نموذج كبير.
وأخيراً، تذكر أن الـ Deployment ليس نهاية الرحلة - بل بداية مرحلة جديدة من التحسين المستمر. راقب أداء النظام باستمرار، واستمع لملاحظات المستخدمين، وكن مستعداً لإعادة تصميم أجزاء من الـ Pipeline عندما تكتشف مشاكل جديدة. الشات بوت الجيد ليس الذي يعمل عند إطلاقه، بل الذي يستمر في التحسن مع الوقت.