النشر للإنتاج
توسيع الاستدلال المحلي
3 دقيقة للقراءة
تعامل مع مستخدمين متعددين وحركة مرور عالية بتوسيع بنيتك التحتية المحلية لنماذج اللغة مع موازنة الحمل والنسخ المتماثلة.
استراتيجيات التوسع
┌─────────────────────────────────────────────────────────────────┐
│ استراتيجيات توسيع LLM المحلي │
├─────────────────────────────────────────────────────────────────┤
│ │
│ الاستراتيجية │ الوصف │ الأفضل لـ │
│ ──────────────────│───────────────────────│───────────────── │
│ رأسي │ GPU أكبر/ذاكرة أكثر │ خادم واحد │
│ أفقي │ نسخ متماثلة متعددة │ توافر عالي │
│ توجيه النموذج │ نماذج/GPUs مختلفة │ أحمال مختلطة │
│ طوابير الطلبات │ معالجة غير متزامنة │ أحمال الدفعات │
│ التخزين المؤقت │ خزّن الاستعلامات الشائعة│ أسئلة متكررة │
│ │
└─────────────────────────────────────────────────────────────────┘
موازنة الحمل مع NGINX
# nginx.conf
upstream ollama_backend {
# موازنة حمل round-robin عبر النسخ
server ollama1:11434;
server ollama2:11434;
server ollama3:11434;
# اختياري: استراتيجية أقل اتصالات
# least_conn;
# فحوصات الصحة
keepalive 32;
}
server {
listen 80;
server_name llm.example.com;
location / {
proxy_pass http://ollama_backend;
proxy_http_version 1.1;
proxy_set_header Connection "";
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
# دعم البث
proxy_buffering off;
proxy_cache off;
chunked_transfer_encoding on;
# مهلات للاستدلال طويل المدى
proxy_connect_timeout 300s;
proxy_send_timeout 300s;
proxy_read_timeout 300s;
}
}
Docker Compose مع النسخ المتماثلة
# docker-compose.scale.yml
version: '3.8'
services:
ollama:
image: ollama/ollama:latest
volumes:
- ollama_shared:/root/.ollama
deploy:
replicas: 3
resources:
reservations:
devices:
- driver: nvidia
count: 1
capabilities: [gpu]
environment:
- OLLAMA_KEEP_ALIVE=24h
nginx:
image: nginx:alpine
ports:
- "80:80"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf:ro
depends_on:
- ollama
volumes:
ollama_shared:
# وسّع لـ 3 نسخ
docker compose -f docker-compose.scale.yml up -d --scale ollama=3
طابور الطلبات مع Redis
import redis
import json
import ollama
from typing import Optional
import threading
import time
class OllamaQueue:
"""معالجة طلبات قائمة على الطوابير لتحديد المعدل."""
def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url)
self.queue_name = "ollama_requests"
self.results_prefix = "ollama_result:"
def submit(self, request_id: str, model: str, prompt: str) -> str:
"""أرسل طلب للطابور."""
request = {
"id": request_id,
"model": model,
"prompt": prompt,
"timestamp": time.time()
}
self.redis.rpush(self.queue_name, json.dumps(request))
return request_id
def get_result(self, request_id: str, timeout: int = 300) -> Optional[dict]:
"""انتظر واسترجع النتيجة."""
key = f"{self.results_prefix}{request_id}"
for _ in range(timeout):
result = self.redis.get(key)
if result:
self.redis.delete(key)
return json.loads(result)
time.sleep(1)
return None
def process_queue(self):
"""عامل يعالج الطلبات المطوّبة."""
while True:
# سحب مانع من الطابور
_, request_json = self.redis.blpop(self.queue_name)
request = json.loads(request_json)
try:
# عالج مع Ollama
response = ollama.generate(
model=request["model"],
prompt=request["prompt"]
)
result = {
"success": True,
"response": response["response"]
}
except Exception as e:
result = {
"success": False,
"error": str(e)
}
# خزّن النتيجة
key = f"{self.results_prefix}{request['id']}"
self.redis.setex(key, 3600, json.dumps(result))
# ابدأ العامل في الخلفية
queue = OllamaQueue()
worker = threading.Thread(target=queue.process_queue, daemon=True)
worker.start()
# أرسل طلبات
request_id = queue.submit("req-001", "llama3.2", "ما هو الذكاء الاصطناعي؟")
result = queue.get_result(request_id)
print(result)
تخزين الردود مؤقتاً
import hashlib
import json
import redis
import ollama
from typing import Optional
class CachedOllama:
"""عميل Ollama مع تخزين الردود مؤقتاً."""
def __init__(self, redis_url: str = "redis://localhost:6379", ttl: int = 3600):
self.redis = redis.from_url(redis_url)
self.ttl = ttl
def _cache_key(self, model: str, prompt: str, options: dict) -> str:
"""ولّد مفتاح التخزين من بارامترات الطلب."""
content = json.dumps({
"model": model,
"prompt": prompt,
"options": options
}, sort_keys=True)
return f"ollama_cache:{hashlib.sha256(content.encode()).hexdigest()}"
def generate(self, model: str, prompt: str, **options) -> dict:
"""ولّد مع التخزين المؤقت."""
cache_key = self._cache_key(model, prompt, options)
# تحقق من التخزين
cached = self.redis.get(cache_key)
if cached:
return json.loads(cached)
# ولّد رد جديد
response = ollama.generate(model=model, prompt=prompt, **options)
# خزّن النتيجة
self.redis.setex(cache_key, self.ttl, json.dumps(response))
return response
def cache_stats(self) -> dict:
"""احصل على إحصائيات التخزين."""
keys = self.redis.keys("ollama_cache:*")
return {
"cached_responses": len(keys),
"memory_usage_bytes": sum(
self.redis.memory_usage(k) or 0 for k in keys
)
}
# الاستخدام
client = CachedOllama()
# الاستدعاء الأول - يولّد ويخزّن
response1 = client.generate("llama3.2", "ما هو Python؟")
# الاستدعاء الثاني - يرجع الرد المخزّن فوراً
response2 = client.generate("llama3.2", "ما هو Python؟")
print(client.cache_stats())
التوجيه المبني على النموذج
import ollama
from typing import Literal
class ModelRouter:
"""وجّه الطلبات لنماذج مناسبة بناءً على المهمة."""
def __init__(self):
self.model_map = {
"code": "deepseek-coder:6.7b",
"chat": "llama3.2",
"embedding": "nomic-embed-text",
"fast": "phi3:mini",
}
def classify_task(self, prompt: str) -> str:
"""صنّف نوع المهمة من الطلب."""
prompt_lower = prompt.lower()
if any(kw in prompt_lower for kw in ["code", "function", "debug", "program", "كود", "دالة"]):
return "code"
if any(kw in prompt_lower for kw in ["embed", "vector", "similarity", "تضمين", "متجه"]):
return "embedding"
if len(prompt) < 50:
return "fast"
return "chat"
def generate(self, prompt: str, task_type: str = None) -> dict:
"""وجّه الطلب للنموذج المناسب."""
if task_type is None:
task_type = self.classify_task(prompt)
model = self.model_map.get(task_type, "llama3.2")
if task_type == "embedding":
return ollama.embed(model=model, input=prompt)
else:
return ollama.generate(model=model, prompt=prompt)
# الاستخدام
router = ModelRouter()
# يوجّه تلقائياً لنموذج الكود
code_response = router.generate("اكتب دالة Python لترتيب قائمة")
# يوجّه تلقائياً للنموذج السريع
quick_response = router.generate("ما هو 2+2؟")
# حدد المهمة صراحةً
chat_response = router.generate("أخبرني عن الذكاء الاصطناعي", task_type="chat")
مراقبة الصحة
import ollama
import time
import threading
from dataclasses import dataclass
from typing import Dict, List
@dataclass
class ServerHealth:
url: str
healthy: bool
latency_ms: float
last_check: float
class HealthMonitor:
"""راقب صحة نسخ Ollama المتعددة."""
def __init__(self, servers: List[str], check_interval: int = 30):
self.servers = servers
self.check_interval = check_interval
self.health: Dict[str, ServerHealth] = {}
self._start_monitoring()
def _check_server(self, url: str) -> ServerHealth:
"""تحقق من صحة خادم واحد."""
try:
client = ollama.Client(host=url)
start = time.time()
client.list() # فحص صحة بسيط
latency = (time.time() - start) * 1000
return ServerHealth(
url=url,
healthy=True,
latency_ms=latency,
last_check=time.time()
)
except Exception:
return ServerHealth(
url=url,
healthy=False,
latency_ms=-1,
last_check=time.time()
)
def _monitor_loop(self):
"""حلقة مراقبة مستمرة."""
while True:
for server in self.servers:
self.health[server] = self._check_server(server)
time.sleep(self.check_interval)
def _start_monitoring(self):
"""ابدأ مسار مراقبة خلفي."""
thread = threading.Thread(target=self._monitor_loop, daemon=True)
thread.start()
def get_healthy_servers(self) -> List[str]:
"""احصل على قائمة الخوادم السليمة."""
return [
url for url, health in self.health.items()
if health.healthy
]
def get_fastest_server(self) -> str:
"""احصل على الخادم بأقل تأخير."""
healthy = [h for h in self.health.values() if h.healthy]
if not healthy:
raise Exception("لا توجد خوادم سليمة متاحة")
return min(healthy, key=lambda h: h.latency_ms).url
# الاستخدام
monitor = HealthMonitor([
"http://ollama1:11434",
"http://ollama2:11434",
"http://ollama3:11434"
])
# انتظر فحص الصحة الأولي
time.sleep(2)
# استخدم أسرع خادم سليم
fastest = monitor.get_fastest_server()
client = ollama.Client(host=fastest)
response = client.generate(model="llama3.2", prompt="مرحباً!")
مصفوفة قرار التوسع
| المستخدمين | الاستراتيجية |
|---|---|
| 1-5 | نسخة Ollama واحدة |
| 5-20 | 2-3 نسخ مع NGINX |
| 20-100 | GPUs متعددة + تخزين مؤقت |
| 100+ | فكّر في vLLM + Kubernetes |
التوسع يضمن أن بنيتك التحتية المحلية لنماذج اللغة تتعامل مع حمل الإنتاج. في الدرس الأخير، سنناقش الخطوات التالية لتطوير مهاراتك. :::