# وكلاء الذكاء الاصطناعي ونماذج اللغة الصغيرة: دليل شامل

## المقدمة

في هذا الدليل، سنستكشف وكلاء الذكاء الاصطناعي ونماذج اللغة الصغيرة (SLMs) واستراتيجيات تنفيذها المتقدمة في بيئات الحوسبة الطرفية. سنغطي المفاهيم الأساسية للذكاء الاصطناعي الوكالي، تقنيات تحسين SLM، استراتيجيات النشر العملية للأجهزة ذات الموارد المحدودة، وإطار عمل Microsoft Agent لبناء أنظمة وكلاء جاهزة للإنتاج.

يشهد مجال الذكاء الاصطناعي تحولًا جذريًا في عام 2025. بينما كان عام 2023 عام الروبوتات الحوارية وشهد عام 2024 ازدهارًا في المساعدين الذكيين، فإن عام 2025 ينتمي إلى وكلاء الذكاء الاصطناعي - أنظمة ذكية تفكر، تخطط، تستخدم الأدوات، وتنفذ المهام بأقل تدخل بشري، مدعومة بشكل متزايد بنماذج اللغة الصغيرة الفعالة. يظهر إطار عمل Microsoft Agent كحل رائد لبناء هذه الأنظمة الذكية بقدرات طرفية غير متصلة.

## أهداف التعلم

بنهاية هذا الدليل، ستكون قادرًا على:

- 🤖 فهم المفاهيم الأساسية لوكلاء الذكاء الاصطناعي وأنظمة الوكلاء
- 🔬 التعرف على مزايا نماذج اللغة الصغيرة مقارنة بنماذج اللغة الكبيرة في التطبيقات الوكالية
- 🚀 تعلم استراتيجيات نشر SLM المتقدمة في بيئات الحوسبة الطرفية
- 📱 تنفيذ وكلاء مدعومين بـ SLM للتطبيقات الواقعية
- 🏗️ بناء وكلاء جاهزين للإنتاج باستخدام إطار عمل Microsoft Agent
- 🌐 نشر وكلاء طرفيين غير متصلين مع تكامل LLM وSLM المحلي
- 🔧 دمج إطار عمل Microsoft Agent مع Foundry Local للنشر الطرفي

## فهم وكلاء الذكاء الاصطناعي: الأسس والتصنيفات

### التعريف والمفاهيم الأساسية

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

### إطار تصنيف الوكلاء

يساعد فهم حدود الوكلاء في اختيار الأنواع المناسبة من الوكلاء لسيناريوهات الحوسبة المختلفة:

- **🔬 وكلاء ردود الفعل البسيطة**: أنظمة تعتمد على القواعد وتستجيب للإدراكات الفورية (مثل منظمات الحرارة، الأتمتة الأساسية)
- **📱 وكلاء قائمون على النموذج**: أنظمة تحتفظ بحالة داخلية وذاكرة (مثل المكانس الروبوتية، أنظمة الملاحة)
- **⚖️ وكلاء قائمون على الأهداف**: أنظمة تخطط وتنفذ تسلسلات لتحقيق الأهداف (مثل مخططي الطرق، جداول المهام)
- **🧠 وكلاء التعلم**: أنظمة تكيفية تحسن الأداء بمرور الوقت (مثل أنظمة التوصيات، المساعدين الشخصيين)

### المزايا الرئيسية لوكلاء الذكاء الاصطناعي

تقدم وكلاء الذكاء الاصطناعي العديد من المزايا الأساسية التي تجعلها مثالية لتطبيقات الحوسبة الطرفية:

**الاستقلالية التشغيلية**: توفر الوكلاء تنفيذ المهام بشكل مستقل دون إشراف بشري مستمر، مما يجعلها مثالية للتطبيقات في الوقت الفعلي. تتطلب إشرافًا ضئيلًا مع الحفاظ على السلوك التكيفي، مما يتيح النشر على الأجهزة ذات الموارد المحدودة مع تقليل العبء التشغيلي.

**مرونة النشر**: تتيح هذه الأنظمة قدرات الذكاء الاصطناعي على الأجهزة دون الحاجة إلى الاتصال بالإنترنت، وتعزز الخصوصية والأمان من خلال المعالجة المحلية، ويمكن تخصيصها للتطبيقات الخاصة بالمجال، وهي مناسبة لمختلف بيئات الحوسبة الطرفية.

**فعالية التكلفة**: توفر أنظمة الوكلاء نشرًا فعالًا من حيث التكلفة مقارنة بالحلول السحابية، مع تقليل تكاليف التشغيل ومتطلبات النطاق الترددي المنخفض للتطبيقات الطرفية.

## استراتيجيات نماذج اللغة الصغيرة المتقدمة

### أساسيات SLM (نماذج اللغة الصغيرة)

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

**ميزات اكتشاف التنسيق**: تقدم SLMs دعمًا متقدمًا لمستويات التكميم المختلفة، التوافق عبر الأنظمة الأساسية، تحسين الأداء في الوقت الفعلي، وقدرات النشر الطرفي. يمكن للمستخدمين الوصول إلى خصوصية محسنة من خلال المعالجة المحلية ودعم WebGPU للنشر المستند إلى المتصفح.

**مجموعات مستويات التكميم**: تشمل تنسيقات SLM الشائعة Q4_K_M لضغط متوازن في التطبيقات المحمولة، سلسلة Q5_K_S للنشر الطرفي الذي يركز على الجودة، Q8_0 لدقة قريبة من الأصل على الأجهزة الطرفية القوية، وتنسيقات تجريبية مثل Q2_K للسيناريوهات ذات الموارد المنخفضة للغاية.

### GGUF (تنسيق GGML العام العالمي) لنشر SLM

يعمل GGUF كالتنسيق الأساسي لنشر SLMs المكممة على وحدات المعالجة المركزية والأجهزة الطرفية، وهو مُحسّن خصيصًا للتطبيقات الوكالية:

**ميزات محسّنة للوكلاء**: يوفر التنسيق موارد شاملة لتحويل SLMs ونشرها مع دعم محسّن لاستدعاء الأدوات، إنشاء المخرجات المهيكلة، والمحادثات متعددة الأدوار. يضمن التوافق عبر الأنظمة الأساسية سلوكًا متسقًا للوكلاء عبر الأجهزة الطرفية المختلفة.

**تحسين الأداء**: يتيح GGUF استخدامًا فعالًا للذاكرة لسير عمل الوكلاء، يدعم تحميل النماذج الديناميكي لأنظمة الوكلاء المتعددة، ويوفر استنتاجًا محسّنًا لتفاعلات الوكلاء في الوقت الفعلي.

### أطر SLM المحسّنة للطرف

#### تحسين Llama.cpp للوكلاء

يوفر Llama.cpp تقنيات تكميم متطورة مُحسّنة خصيصًا لنشر SLM الوكالي:

**تكميم خاص بالوكيل**: يدعم الإطار Q4_0 (الأمثل لنشر الوكلاء المحمول مع تقليل الحجم بنسبة 75%)، Q5_1 (جودة-ضغط متوازن لوكلاء الاستنتاج الطرفي)، وQ8_0 (جودة قريبة من الأصل لأنظمة الوكلاء الإنتاجية). تتيح التنسيقات المتقدمة وكلاء مضغوطين للغاية لسيناريوهات الطرف القصوى.

**فوائد التنفيذ**: يوفر الاستنتاج المحسّن لوحدة المعالجة المركزية مع تسريع SIMD تنفيذًا فعالًا للذاكرة للوكيل. يتيح التوافق عبر الأنظمة الأساسية عبر معماريات x86، ARM، وApple Silicon قدرات نشر عالمية للوكيل.

#### إطار Apple MLX لوكلاء SLM

يوفر Apple MLX تحسينًا أصليًا مصممًا خصيصًا للوكلاء المدعومين بـ SLM على أجهزة Apple Silicon:

**تحسين الوكلاء على Apple Silicon**: يستخدم الإطار بنية ذاكرة موحدة مع تكامل Metal Performance Shaders، دقة مختلطة تلقائية لاستنتاج الوكلاء، وعرض نطاق ذاكرة محسّن لأنظمة الوكلاء المتعددة. تظهر وكلاء SLM أداءً استثنائيًا على رقائق سلسلة M.

**ميزات التطوير**: دعم API لـ Python وSwift مع تحسينات خاصة بالوكيل، التمايز التلقائي لتعلم الوكلاء، والتكامل السلس مع أدوات تطوير Apple توفر بيئات تطوير شاملة للوكيل.

#### ONNX Runtime لوكلاء SLM عبر الأنظمة الأساسية

يوفر ONNX Runtime محرك استنتاج عالمي يتيح لوكلاء SLM العمل بشكل متسق عبر منصات الأجهزة وأنظمة التشغيل المتنوعة:

**النشر العالمي**: يضمن ONNX Runtime سلوكًا متسقًا لوكلاء SLM عبر منصات Windows، Linux، macOS، iOS، وAndroid. يتيح هذا التوافق عبر الأنظمة الأساسية للمطورين الكتابة مرة واحدة والنشر في كل مكان، مما يقلل بشكل كبير من عبء التطوير والصيانة للتطبيقات متعددة المنصات.

**خيارات تسريع الأجهزة**: يوفر الإطار موفري تنفيذ محسّنين لتكوينات الأجهزة المختلفة بما في ذلك وحدة المعالجة المركزية (Intel، AMD، ARM)، وحدة معالجة الرسومات (NVIDIA CUDA، AMD ROCm)، والمسرعات المتخصصة (Intel VPU، Qualcomm NPU). يمكن لوكلاء SLM الاستفادة تلقائيًا من أفضل الأجهزة المتاحة دون تغييرات في الكود.

**ميزات جاهزة للإنتاج**: يقدم ONNX Runtime ميزات على مستوى المؤسسات ضرورية لنشر الوكلاء الإنتاجي بما في ذلك تحسين الرسم البياني للاستنتاج الأسرع، إدارة الذاكرة للبيئات ذات الموارد المحدودة، وأدوات تحليل الأداء الشاملة. يدعم الإطار واجهات برمجة التطبيقات لـ Python وC++ للتكامل المرن.

## SLM مقابل LLM في أنظمة الوكلاء: مقارنة متقدمة

### مزايا SLM في تطبيقات الوكلاء

**الكفاءة التشغيلية**: توفر SLMs تخفيضًا في التكلفة بنسبة 10-30× مقارنة بـ LLMs لمهام الوكلاء، مما يتيح استجابات وكالية في الوقت الفعلي على نطاق واسع. تقدم أوقات استنتاج أسرع بسبب التعقيد الحسابي المنخفض، مما يجعلها مثالية لتطبيقات الوكلاء التفاعلية.

**قدرات النشر الطرفي**: تتيح SLMs تنفيذ الوكلاء على الأجهزة دون الاعتماد على الإنترنت، وتعزز الخصوصية من خلال معالجة الوكلاء المحلية، وتخصيصها لتطبيقات الوكلاء الخاصة بالمجال المناسبة لمختلف بيئات الحوسبة الطرفية.

**تحسين خاص بالوكيل**: تتفوق SLMs في استدعاء الأدوات، إنشاء المخرجات المهيكلة، وسير العمل الروتيني لاتخاذ القرارات التي تشكل 70-80% من مهام الوكلاء النموذجية.

### متى تستخدم SLMs مقابل LLMs في أنظمة الوكلاء

**مثالية لـ SLMs**:
- **مهام الوكلاء المتكررة**: إدخال البيانات، ملء النماذج، استدعاءات API الروتينية
- **تكامل الأدوات**: استعلامات قواعد البيانات، عمليات الملفات، التفاعلات النظامية
- **سير العمل المهيكل**: اتباع عمليات الوكلاء المحددة مسبقًا
- **وكلاء خاصون بالمجال**: خدمة العملاء، الجدولة، التحليل الأساسي
- **المعالجة المحلية**: عمليات الوكلاء الحساسة للخصوصية

**أفضل لـ LLMs**:
- **التفكير المعقد**: حل المشكلات الجديدة، التخطيط الاستراتيجي
- **المحادثات المفتوحة**: الدردشة العامة، المناقشات الإبداعية
- **مهام المعرفة الواسعة**: البحث الذي يتطلب معرفة عامة واسعة
- **المواقف الجديدة**: التعامل مع سيناريوهات الوكلاء الجديدة تمامًا

### بنية الوكلاء الهجينة

النهج الأمثل يجمع بين SLMs وLLMs في أنظمة وكالية غير متجانسة:

**تنسيق الوكلاء الذكي**:
1. **SLM كأولوية**: التعامل مع 70-80% من مهام الوكلاء الروتينية محليًا
2. **LLM عند الحاجة**: توجيه الاستفسارات المعقدة إلى النماذج الأكبر المستندة إلى السحابة
3. **SLMs المتخصصة**: نماذج صغيرة مختلفة لمجالات الوكلاء المختلفة
4. **تحسين التكلفة**: تقليل استدعاءات LLM المكلفة من خلال التوجيه الذكي

## استراتيجيات نشر وكلاء SLM الإنتاجية

### Foundry Local: تشغيل الذكاء الاصطناعي الطرفي على مستوى المؤسسات

يعمل Foundry Local (https://github.com/microsoft/foundry-local) كحل رئيسي من Microsoft لنشر نماذج اللغة الصغيرة في بيئات الطرف الإنتاجية. يوفر بيئة تشغيل كاملة مصممة خصيصًا للوكلاء المدعومين بـ SLM مع ميزات على مستوى المؤسسات وقدرات تكامل سلسة.

**الهيكل الأساسي والميزات**:
- **واجهة برمجة تطبيقات متوافقة مع OpenAI**: توافق كامل مع SDK OpenAI وتكاملات إطار عمل الوكلاء
- **تحسين الأجهزة التلقائي**: اختيار ذكي لمتغيرات النموذج بناءً على الأجهزة المتاحة (CUDA GPU، Qualcomm NPU، CPU)
- **إدارة النماذج**: تنزيل تلقائي، تخزين مؤقت، وإدارة دورة حياة نماذج SLM
- **اكتشاف الخدمة**: اكتشاف الخدمة بدون تكوين لإطارات عمل الوكلاء
- **تحسين الموارد**: إدارة ذكية للذاكرة وكفاءة الطاقة للنشر الطرفي

#### التثبيت والإعداد

**التثبيت عبر الأنظمة الأساسية**:
```bash
# Windows (recommended)
winget install Microsoft.FoundryLocal

# macOS
brew tap microsoft/foundrylocal
brew install foundrylocal

# Linux (manual installation)
wget https://github.com/microsoft/foundry-local/releases/latest/download/foundry-local-linux.tar.gz
tar -xzf foundry-local-linux.tar.gz
sudo mv foundry-local /usr/local/bin/
```

**البدء السريع لتطوير الوكلاء**:
```bash
# Start service with automatic model loading
foundry model run phi-4-mini

# Verify service status and endpoint
foundry service status

# List available models
foundry model ls

# Test API endpoint
curl http://localhost:<port>/v1/models
```

#### تكامل إطار عمل الوكلاء

**تكامل SDK Foundry Local**:
```python
from foundry_local import FoundryLocalManager
from microsoft_agent_framework import Agent, Config
import openai

# Initialize Foundry Local with automatic service management
manager = FoundryLocalManager("phi-4-mini")

# Configure OpenAI client for local inference
client = openai.OpenAI(
    base_url=manager.endpoint,
    api_key=manager.api_key  # Auto-generated for local usage
)

# Create agent with Foundry Local backend
agent_config = Config(
    name="production-agent",
    model_provider="foundry-local",
    model_id=manager.get_model_info("phi-4-mini").id,
    endpoint=manager.endpoint,
    api_key=manager.api_key
)

agent = Agent(config=agent_config)
```

**اختيار النموذج التلقائي وتحسين الأجهزة**:
```python
# Foundry Local automatically selects optimal model variant
models_by_use_case = {
    "lightweight_routing": "qwen2.5-0.5b",      # 500MB, ultra-fast
    "general_conversation": "phi-4-mini",       # 2.4GB, balanced
    "complex_reasoning": "phi-4",               # 7GB, high-capability
    "code_assistance": "qwen2.5-coder-0.5b"    # 500MB, code-optimized
}

# Foundry Local handles hardware detection and quantization
for use_case, model_alias in models_by_use_case.items():
    manager = FoundryLocalManager(model_alias)
    print(f"{use_case}: {manager.get_model_info(model_alias).variant_selected}")
    # Output examples:
    # lightweight_routing: qwen2.5-0.5b-instruct-q4_k_m.gguf (CPU optimized)
    # general_conversation: phi-4-mini-instruct-cuda-q5_k_m.gguf (GPU accelerated)
```

#### أنماط النشر الإنتاجي

**إعداد الإنتاج لوكيل واحد**:
```python
import asyncio
from foundry_local import FoundryLocalManager
from microsoft_agent_framework import Agent, Config, Tool

class ProductionAgentService:
    def __init__(self, model_alias="phi-4-mini"):
        self.foundry = FoundryLocalManager(model_alias)
        self.agent = self._create_agent()
        
    def _create_agent(self):
        config = Config(
            name="production-customer-service",
            model_provider="foundry-local",
            model_id=self.foundry.get_model_info().id,
            endpoint=self.foundry.endpoint,
            api_key=self.foundry.api_key,
            max_tokens=512,
            temperature=0.1,
            timeout=30.0
        )
        
        agent = Agent(config=config)
        
        # Add production tools
        @agent.tool
        def lookup_customer(customer_id: str) -> dict:
            """Look up customer information from local database."""
            return self.local_db.get_customer(customer_id)
            
        @agent.tool
        def create_ticket(issue: str, priority: str = "medium") -> str:
            """Create a support ticket."""
            ticket_id = self.ticketing_system.create(issue, priority)
            return f"Created ticket {ticket_id}"
            
        return agent
    
    async def process_request(self, user_input: str) -> str:
        """Process user request with error handling and monitoring."""
        try:
            response = await self.agent.chat_async(user_input)
            self.log_interaction(user_input, response, "success")
            return response
        except Exception as e:
            self.log_interaction(user_input, str(e), "error")
            return "I'm experiencing technical difficulties. Please try again."
    
    def health_check(self) -> dict:
        """Check service health for monitoring."""
        return {
            "foundry_status": self.foundry.health_check(),
            "model_loaded": self.foundry.is_model_loaded(),
            "endpoint": self.foundry.endpoint,
            "memory_usage": self.foundry.get_memory_usage()
        }

# Production usage
service = ProductionAgentService("phi-4-mini")
response = await service.process_request("I need help with my order #12345")
```

**تنسيق الإنتاج لوكلاء متعددين**:
```python
from foundry_local import FoundryLocalManager
from microsoft_agent_framework import AgentOrchestrator, Agent, Config

class MultiAgentProductionSystem:
    def __init__(self):
        self.agents = self._initialize_agents()
        self.orchestrator = AgentOrchestrator(list(self.agents.values()))
        
    def _initialize_agents(self):
        agents = {}
        
        # Lightweight routing agent
        routing_foundry = FoundryLocalManager("qwen2.5-0.5b")
        agents["router"] = Agent(Config(
            name="request-router",
            model_provider="foundry-local",
            endpoint=routing_foundry.endpoint,
            api_key=routing_foundry.api_key,
            role="Route user requests to appropriate specialized agents"
        ))
        
        # Customer service agent
        service_foundry = FoundryLocalManager("phi-4-mini")
        agents["customer_service"] = Agent(Config(
            name="customer-service",
            model_provider="foundry-local",
            endpoint=service_foundry.endpoint,
            api_key=service_foundry.api_key,
            role="Handle customer service inquiries and support requests"
        ))
        
        # Technical support agent
        tech_foundry = FoundryLocalManager("qwen2.5-coder-0.5b")
        agents["technical"] = Agent(Config(
            name="technical-support",
            model_provider="foundry-local",
            endpoint=tech_foundry.endpoint,
            api_key=tech_foundry.api_key,
            role="Provide technical assistance and troubleshooting"
        ))
        
        return agents
    
    async def process_request(self, user_input: str) -> str:
        """Route and process user requests through appropriate agents."""
        # Route request to appropriate agent
        routing_result = await self.agents["router"].chat_async(
            f"Classify this request and route to customer_service or technical: {user_input}"
        )
        
        # Determine target agent based on routing
        target_agent = "customer_service" if "customer" in routing_result.lower() else "technical"
        
        # Process with specialized agent
        response = await self.agents[target_agent].chat_async(user_input)
        
        return response

# Production deployment
system = MultiAgentProductionSystem()
response = await system.process_request("My application keeps crashing")
```

#### ميزات المؤسسات والمراقبة

**مراقبة الصحة وقابلية الملاحظة**:
```python
from foundry_local import FoundryLocalManager
import asyncio
import logging

class FoundryMonitoringService:
    def __init__(self):
        self.managers = {}
        self.metrics = []
        
    def add_model(self, alias: str) -> FoundryLocalManager:
        """Add a model to monitoring."""
        manager = FoundryLocalManager(alias)
        self.managers[alias] = manager
        return manager
    
    async def collect_metrics(self):
        """Collect performance metrics from all Foundry Local instances."""
        metrics = {
            "timestamp": time.time(),
            "models": {}
        }
        
        for alias, manager in self.managers.items():
            try:
                model_metrics = {
                    "status": "healthy" if manager.health_check() else "unhealthy",
                    "memory_usage": manager.get_memory_usage(),
                    "inference_count": manager.get_inference_count(),
                    "average_latency": manager.get_average_latency(),
                    "error_rate": manager.get_error_rate()
                }
                metrics["models"][alias] = model_metrics
            except Exception as e:
                logging.error(f"Failed to collect metrics for {alias}: {e}")
                metrics["models"][alias] = {"status": "error", "error": str(e)}
        
        self.metrics.append(metrics)
        return metrics
    
    def get_health_status(self) -> dict:
        """Get overall system health status."""
        healthy_models = 0
        total_models = len(self.managers)
        
        for alias, manager in self.managers.items():
            if manager.health_check():
                healthy_models += 1
        
        return {
            "overall_status": "healthy" if healthy_models == total_models else "degraded",
            "healthy_models": healthy_models,
            "total_models": total_models,
            "health_percentage": (healthy_models / total_models) * 100 if total_models > 0 else 0
        }

# Production monitoring setup
monitor = FoundryMonitoringService()
monitor.add_model("phi-4-mini")
monitor.add_model("qwen2.5-0.5b")

# Continuous monitoring
async def monitoring_loop():
    while True:
        metrics = await monitor.collect_metrics()
        health = monitor.get_health_status()
        
        if health["health_percentage"] < 100:
            logging.warning(f"System health degraded: {health}")
        
        await asyncio.sleep(30)  # Collect metrics every 30 seconds
```

**إدارة الموارد والتوسع التلقائي**:
```python
class FoundryResourceManager:
    def __init__(self):
        self.model_instances = {}
        self.resource_limits = {
            "max_memory_gb": 8,
            "max_concurrent_models": 3,
            "cpu_threshold": 80
        }
    
    def auto_scale_models(self, demand_metrics: dict):
        """Automatically scale models based on demand."""
        current_memory = self.get_total_memory_usage()
        
        # Scale down if memory usage is high
        if current_memory > self.resource_limits["max_memory_gb"] * 0.8:
            self.scale_down_idle_models()
        
        # Scale up if demand is high and resources allow
        for model_alias, demand in demand_metrics.items():
            if demand > 0.8 and len(self.model_instances) < self.resource_limits["max_concurrent_models"]:
                self.load_model_instance(model_alias)
    
    def load_model_instance(self, alias: str) -> FoundryLocalManager:
        """Load a new model instance if resources allow."""
        if alias not in self.model_instances:
            try:
                manager = FoundryLocalManager(alias)
                self.model_instances[alias] = manager
                logging.info(f"Loaded model instance: {alias}")
                return manager
            except Exception as e:
                logging.error(f"Failed to load model {alias}: {e}")
                return None
        return self.model_instances[alias]
    
    def scale_down_idle_models(self):
        """Remove idle model instances to free resources."""
        idle_models = []
        
        for alias, manager in self.model_instances.items():
            if manager.get_idle_time() > 300:  # 5 minutes idle
                idle_models.append(alias)
        
        for alias in idle_models:
            self.model_instances[alias].shutdown()
            del self.model_instances[alias]
            logging.info(f"Scaled down idle model: {alias}")
```

#### التكوين والتحسين المتقدم

**تكوين النموذج المخصص**:
```python
# Advanced Foundry Local configuration for production
from foundry_local import FoundryLocalManager, ModelConfig

# Custom configuration for specific use cases
config = ModelConfig(
    alias="phi-4-mini",
    quantization="Q5_K_M",  # Specific quantization level
    context_length=4096,    # Extended context for complex agents
    batch_size=1,          # Optimized for single-user agents
    threads=4,             # CPU thread optimization
    gpu_layers=32,         # GPU acceleration layers
    memory_lock=True,      # Lock model in memory for consistent performance
    numa=True              # NUMA optimization for multi-socket systems
)

manager = FoundryLocalManager(config=config)
```

**قائمة التحقق لنشر الإنتاج**:

✅ **تكوين الخدمة**:
- تكوين أسماء مستعارة مناسبة للنماذج لحالات الاستخدام
- تعيين حدود الموارد وعتبات المراقبة
- تمكين فحوصات الصحة وجمع المقاييس
- تكوين إعادة التشغيل التلقائي والتجاوز

✅ **إعداد الأمان**:
- تمكين الوصول إلى واجهة برمجة التطبيقات المحلية فقط (بدون تعرض خارجي)
- تكوين إدارة مفاتيح واجهة برمجة التطبيقات المناسبة
- إعداد تسجيل التدقيق لتفاعلات الوكلاء
- تنفيذ تحديد المعدل للاستخدام الإنتاجي

✅ **تحسين الأداء**:
- اختبار أداء النموذج تحت الحمل المتوقع
- تكوين مستويات التكميم المناسبة
- إعداد استراتيجيات التخزين المؤقت والتسخين للنموذج
- مراقبة أنماط استخدام الذاكرة ووحدة المعالجة المركزية

✅ **اختبار التكامل**:
- اختبار تكامل إطار عمل الوكلاء
- التحقق من قدرات التشغيل غير المتصل
- اختبار سيناريوهات التجاوز والاسترداد
- التحقق من سير عمل الوكلاء من البداية إلى النهاية

### Ollama: نشر وكلاء SLM المبسط

### Ollama: نشر وكلاء SLM الموجه نحو المجتمع

يوفر Ollama نهجًا مدفوعًا بالمجتمع لنشر وكلاء SLM مع التركيز على البساطة، النظام البيئي الواسع للنماذج، وسير العمل الصديق للمطورين. بينما يركز Foundry Local على الميزات على مستوى المؤسسات، يتفوق Ollama في النماذج الأولية السريعة، الوصول إلى نماذج المجتمع، وسيناريوهات النشر المبسطة.

**الهيكل الأساسي والميزات**:
- **واجهة برمجة تطبيقات متوافقة مع OpenAI**: توافق كامل مع واجهة REST API لتكامل إطار عمل الوكلاء السلس
- **مكتبة نماذج واسعة**: الوصول إلى مئات النماذج المساهمة من المجتمع والرسمية
- **إدارة النماذج البسيطة**: تثبيت النموذج وتبديله بأمر واحد
- **دعم عبر الأنظمة الأساسية**: دعم أصلي عبر Windows، macOS، وLinux
- **تحسين الموارد**: التكميم التلقائي واكتشاف الأجهزة

#### التثبيت والإعداد

**التثبيت عبر الأنظمة الأساسية**:
```bash
# Windows
winget install Ollama.Ollama

# macOS
brew install ollama

# Linux
curl -fsSL https://ollama.com/install.sh | sh

# Docker deployment
docker run -d -v ollama:/root/.ollama -p 11434:11434 --name ollama ollama/ollama
```

**البدء السريع لتطوير الوكلاء**:
```bash
# Start Ollama service
ollama serve

# Pull and run models for agent development
ollama pull phi3.5:3.8b-mini-instruct-q4_K_M    # Microsoft Phi-3.5 Mini
ollama pull qwen2.5:0.5b-instruct-q4_K_M        # Qwen2.5 0.5B
ollama pull llama3.2:1b-instruct-q4_K_M         # Llama 3.2 1B

# Test model availability
ollama list

# Test API endpoint
curl http://localhost:11434/api/generate -d '{
  "model": "phi3.5:3.8b-mini-instruct-q4_K_M",
  "prompt": "Hello, how can I help you today?"
}'
```

#### تكامل إطار عمل الوكلاء

**Ollama مع إطار عمل Microsoft Agent**:
```python
from microsoft_agent_framework import Agent, Config
import openai
import requests
import json

class OllamaManager:
    def __init__(self, model_name: str, base_url: str = "http://localhost:11434"):
        self.model_name = model_name
        self.base_url = base_url
        self.api_url = f"{base_url}/api"
        self.openai_url = f"{base_url}/v1"
        
    def ensure_model_available(self) -> bool:
        """Ensure the model is pulled and available."""
        try:
            response = requests.post(f"{self.api_url}/pull", 
                json={"name": self.model_name})
            return response.status_code == 200
        except Exception as e:
            print(f"Failed to pull model {self.model_name}: {e}")
            return False
    
    def get_openai_client(self) -> openai.OpenAI:
        """Get OpenAI-compatible client for Ollama."""
        return openai.OpenAI(
            base_url=self.openai_url,
            api_key="ollama",  # Ollama doesn't require real API key
        )
    
    def health_check(self) -> bool:
        """Check if Ollama service is running."""
        try:
            response = requests.get(f"{self.base_url}/api/tags")
            return response.status_code == 200
        except:
            return False

# Initialize Ollama for agent development
ollama_manager = OllamaManager("phi3.5:3.8b-mini-instruct-q4_K_M")
ollama_manager.ensure_model_available()

# Configure agent with Ollama backend
agent_config = Config(
    name="ollama-agent",
    model_provider="ollama",
    model_id="phi3.5:3.8b-mini-instruct-q4_K_M",
    endpoint=ollama_manager.openai_url,
    api_key="ollama"
)

agent = Agent(config=agent_config)
```

**إعداد وكيل متعدد النماذج مع Ollama**:
```python
class OllamaMultiModelManager:
    def __init__(self):
        self.models = {
            "lightweight": "qwen2.5:0.5b-instruct-q4_K_M",      # 350MB
            "balanced": "phi3.5:3.8b-mini-instruct-q4_K_M",     # 2.3GB  
            "capable": "llama3.2:3b-instruct-q4_K_M",           # 1.9GB
            "coding": "codellama:7b-code-q4_K_M"                # 4.1GB
        }
        self.base_url = "http://localhost:11434"
        self.clients = {}
        self._initialize_models()
    
    def _initialize_models(self):
        """Pull all required models and create clients."""
        for category, model_name in self.models.items():
            # Pull model if not available
            self._pull_model(model_name)
            
            # Create OpenAI client for each model
            self.clients[category] = openai.OpenAI(
                base_url=f"{self.base_url}/v1",
                api_key="ollama"
            )
    
    def _pull_model(self, model_name: str):
        """Pull model if not already available."""
        try:
            response = requests.post(f"{self.base_url}/api/pull", 
                json={"name": model_name})
            if response.status_code == 200:
                print(f"Model {model_name} ready")
        except Exception as e:
            print(f"Failed to pull {model_name}: {e}")
    
    def get_agent_for_task(self, task_type: str) -> Agent:
        """Get appropriate agent based on task complexity."""
        model_category = self._classify_task(task_type)
        model_name = self.models[model_category]
        
        config = Config(
            name=f"ollama-{model_category}-agent",
            model_provider="ollama",
            model_id=model_name,
            endpoint=f"{self.base_url}/v1",
            api_key="ollama"
        )
        
        return Agent(config=config)
    
    def _classify_task(self, task_type: str) -> str:
        """Classify task to appropriate model category."""
        if any(keyword in task_type.lower() for keyword in ["simple", "route", "classify"]):
            return "lightweight"
        elif any(keyword in task_type.lower() for keyword in ["code", "programming", "debug"]):
            return "coding"
        elif any(keyword in task_type.lower() for keyword in ["complex", "analysis", "research"]):
            return "capable"
        else:
            return "balanced"

# Usage example
manager = OllamaMultiModelManager()

# Get appropriate agents for different tasks
routing_agent = manager.get_agent_for_task("simple routing")
coding_agent = manager.get_agent_for_task("code debugging")
analysis_agent = manager.get_agent_for_task("complex analysis")
```

#### أنماط النشر الإنتاجي

**خدمة الإنتاج مع Ollama**:
```python
import asyncio
import logging
from typing import Dict, Optional
from microsoft_agent_framework import Agent, Config
import requests
import openai

class OllamaProductionService:
    def __init__(self, models_config: Dict[str, str]):
        self.models_config = models_config
        self.base_url = "http://localhost:11434"
        self.agents = {}
        self.metrics = {
            "requests_processed": 0,
            "errors": 0,
            "model_usage": {model: 0 for model in models_config.keys()}
        }
        self._initialize_production_agents()
    
    def _initialize_production_agents(self):
        """Initialize production agents with health checks."""
        for agent_type, model_name in self.models_config.items():
            try:
                # Ensure model is available
                self._ensure_model_ready(model_name)
                
                # Create production agent
                config = Config(
                    name=f"production-{agent_type}",
                    model_provider="ollama",
                    model_id=model_name,
                    endpoint=f"{self.base_url}/v1",
                    api_key="ollama",
                    max_tokens=512,
                    temperature=0.1,
                    timeout=30.0
                )
                
                agent = Agent(config=config)
                
                # Add production tools based on agent type
                self._add_production_tools(agent, agent_type)
                
                self.agents[agent_type] = agent
                logging.info(f"Initialized {agent_type} agent with model {model_name}")
                
            except Exception as e:
                logging.error(f"Failed to initialize {agent_type} agent: {e}")
    
    def _ensure_model_ready(self, model_name: str):
        """Ensure model is pulled and ready for use."""
        try:
            # Check if model exists
            response = requests.get(f"{self.base_url}/api/tags")
            models = response.json().get('models', [])
            
            model_exists = any(model['name'] == model_name for model in models)
            
            if not model_exists:
                logging.info(f"Pulling model {model_name}...")
                pull_response = requests.post(f"{self.base_url}/api/pull", 
                    json={"name": model_name})
                
                if pull_response.status_code != 200:
                    raise Exception(f"Failed to pull model {model_name}")
                    
        except Exception as e:
            raise Exception(f"Model setup failed for {model_name}: {e}")
    
    def _add_production_tools(self, agent: Agent, agent_type: str):
        """Add tools based on agent type."""
        if agent_type == "customer_service":
            @agent.tool
            def lookup_customer(customer_id: str) -> dict:
                """Look up customer information."""
                # Simulate database lookup
                return {"customer_id": customer_id, "status": "active", "tier": "premium"}
            
            @agent.tool
            def create_support_ticket(issue: str, priority: str = "medium") -> str:
                """Create a support ticket."""
                ticket_id = f"TICK-{hash(issue) % 10000:04d}"
                return f"Created ticket {ticket_id} with priority {priority}"
        
        elif agent_type == "technical_support":
            @agent.tool
            def run_diagnostics(system_info: str) -> dict:
                """Run system diagnostics."""
                return {"status": "healthy", "issues": [], "recommendations": []}
            
            @agent.tool
            def access_knowledge_base(query: str) -> str:
                """Search technical knowledge base."""
                return f"Knowledge base results for: {query}"
    
    async def process_request(self, request: str, agent_type: str = "customer_service") -> dict:
        """Process user request with monitoring and error handling."""
        start_time = time.time()
        
        try:
            if agent_type not in self.agents:
                raise ValueError(f"Agent type {agent_type} not available")
            
            agent = self.agents[agent_type]
            response = await agent.chat_async(request)
            
            # Update metrics
            self.metrics["requests_processed"] += 1
            self.metrics["model_usage"][agent_type] += 1
            
            processing_time = time.time() - start_time
            
            self._log_interaction(request, response, "success", processing_time, agent_type)
            
            return {
                "response": response,
                "status": "success",
                "processing_time": processing_time,
                "agent_type": agent_type
            }
            
        except Exception as e:
            self.metrics["errors"] += 1
            processing_time = time.time() - start_time
            
            self._log_interaction(request, str(e), "error", processing_time, agent_type)
            
            return {
                "response": "I'm experiencing technical difficulties. Please try again.",
                "status": "error",
                "error": str(e),
                "processing_time": processing_time
            }
    
    def _log_interaction(self, request: str, response: str, status: str, 
                        processing_time: float, agent_type: str):
        """Log interaction for monitoring and analysis."""
        logging.info(f"Agent: {agent_type}, Status: {status}, Time: {processing_time:.2f}s")
        
        # In production, this would write to a proper logging system
        log_entry = {
            "timestamp": time.time(),
            "agent_type": agent_type,
            "request_length": len(request),
            "response_length": len(response),
            "status": status,
            "processing_time": processing_time
        }
    
    def get_health_status(self) -> dict:
        """Get service health status."""
        try:
            # Check Ollama service health
            response = requests.get(f"{self.base_url}/api/tags", timeout=5)
            ollama_healthy = response.status_code == 200
            
            # Check model availability
            available_models = []
            if ollama_healthy:
                models = response.json().get('models', [])
                available_models = [model['name'] for model in models]
            
            return {
                "service_status": "healthy" if ollama_healthy else "unhealthy",
                "ollama_endpoint": self.base_url,
                "available_models": available_models,
                "active_agents": list(self.agents.keys()),
                "metrics": self.metrics,
                "timestamp": time.time()
            }
            
        except Exception as e:
            return {
                "service_status": "error",
                "error": str(e),
                "timestamp": time.time()
            }

# Production deployment example
production_models = {
    "customer_service": "phi3.5:3.8b-mini-instruct-q4_K_M",
    "technical_support": "llama3.2:3b-instruct-q4_K_M",
    "routing": "qwen2.5:0.5b-instruct-q4_K_M"
}

service = OllamaProductionService(production_models)

# Process requests
result = await service.process_request(
    "I need help with my account settings", 
    "customer_service"
)
print(result)
```

#### ميزات المؤسسات والمراقبة

**مراقبة Ollama وقابلية الملاحظة**:
```python
import time
import asyncio
import requests
from typing import Dict, List

class OllamaMonitoringService:
    def __init__(self, base_url: str = "http://localhost:11434"):
        self.base_url = base_url
        self.metrics_history = []
        self.alert_thresholds = {
            "response_time_ms": 2000,
            "error_rate_percent": 5,
            "memory_usage_percent": 85
        }
    
    async def collect_metrics(self) -> dict:
        """Collect comprehensive metrics from Ollama service."""
        metrics = {
            "timestamp": time.time(),
            "service_status": "unknown",
            "models": {},
            "performance": {},
            "resources": {}
        }
        
        try:
            # Check service health
            health_response = requests.get(f"{self.base_url}/api/tags", timeout=5)
            metrics["service_status"] = "healthy" if health_response.status_code == 200 else "unhealthy"
            
            if metrics["service_status"] == "healthy":
                # Get model information
                models_data = health_response.json().get('models', [])
                for model in models_data:
                    model_name = model['name']
                    metrics["models"][model_name] = {
                        "size_gb": model.get('size', 0) / (1024**3),
                        "modified": model.get('modified_at', ''),
                        "digest": model.get('digest', '')[:12]  # Short digest
                    }
                
                # Test inference performance
                start_time = time.time()
                test_response = requests.post(f"{self.base_url}/api/generate", 
                    json={
                        "model": list(metrics["models"].keys())[0] if metrics["models"] else "",
                        "prompt": "Hello",
                        "stream": False
                    }, timeout=10)
                
                if test_response.status_code == 200:
                    inference_time = (time.time() - start_time) * 1000
                    metrics["performance"] = {
                        "inference_time_ms": inference_time,
                        "tokens_per_second": self._calculate_tokens_per_second(test_response.json()),
                        "last_successful_inference": time.time()
                    }
            
        except Exception as e:
            metrics["service_status"] = "error"
            metrics["error"] = str(e)
        
        self.metrics_history.append(metrics)
        
        # Keep only last 100 metrics entries
        if len(self.metrics_history) > 100:
            self.metrics_history = self.metrics_history[-100:]
        
        return metrics
    
    def _calculate_tokens_per_second(self, response_data: dict) -> float:
        """Calculate approximate tokens per second from response."""
        try:
            # Estimate tokens (rough approximation)
            response_text = response_data.get('response', '')
            estimated_tokens = len(response_text.split())
            
            # Get timing info if available
            eval_duration = response_data.get('eval_duration', 0)
            if eval_duration > 0:
                # Convert nanoseconds to seconds
                duration_seconds = eval_duration / 1e9
                return estimated_tokens / duration_seconds if duration_seconds > 0 else 0
        except:
            pass
        return 0
    
    def check_alerts(self, current_metrics: dict) -> List[dict]:
        """Check current metrics against alert thresholds."""
        alerts = []
        
        # Check response time
        if current_metrics.get('performance', {}).get('inference_time_ms', 0) > self.alert_thresholds['response_time_ms']:
            alerts.append({
                "type": "performance",
                "message": f"High response time: {current_metrics['performance']['inference_time_ms']:.0f}ms",
                "severity": "warning"
            })
        
        # Check service status
        if current_metrics.get('service_status') != 'healthy':
            alerts.append({
                "type": "availability",
                "message": f"Service unhealthy: {current_metrics.get('error', 'Unknown error')}",
                "severity": "critical"
            })
        
        return alerts
    
    def get_performance_summary(self, minutes: int = 60) -> dict:
        """Get performance summary for the last N minutes."""
        cutoff_time = time.time() - (minutes * 60)
        recent_metrics = [m for m in self.metrics_history if m['timestamp'] > cutoff_time]
        
        if not recent_metrics:
            return {"error": "No recent metrics available"}
        
        # Calculate averages
        response_times = [m.get('performance', {}).get('inference_time_ms', 0) 
                         for m in recent_metrics if m.get('performance')]
        
        healthy_checks = sum(1 for m in recent_metrics if m.get('service_status') == 'healthy')
        uptime_percent = (healthy_checks / len(recent_metrics)) * 100 if recent_metrics else 0
        
        return {
            "period_minutes": minutes,
            "total_checks": len(recent_metrics),
            "uptime_percent": uptime_percent,
            "avg_response_time_ms": sum(response_times) / len(response_times) if response_times else 0,
            "max_response_time_ms": max(response_times) if response_times else 0,
            "min_response_time_ms": min(response_times) if response_times else 0
        }

# Production monitoring setup
monitor = OllamaMonitoringService()

async def monitoring_loop():
    """Continuous monitoring loop."""
    while True:
        try:
            metrics = await monitor.collect_metrics()
            alerts = monitor.check_alerts(metrics)
            
            if alerts:
                for alert in alerts:
                    logging.warning(f"ALERT: {alert['message']} (Severity: {alert['severity']})")
            
            # Log performance summary every 10 minutes
            if int(time.time()) % 600 == 0:  # Every 10 minutes
                summary = monitor.get_performance_summary(10)
                logging.info(f"Performance Summary: {summary}")
            
        except Exception as e:
            logging.error(f"Monitoring error: {e}")
        
        await asyncio.sleep(30)  # Check every 30 seconds

# Start monitoring
# asyncio.create_task(monitoring_loop())
```

#### التكوين والتحسين المتقدم

**إدارة النماذج المخصصة مع Ollama**:
```python
class OllamaModelManager:
    def __init__(self, base_url: str = "http://localhost:11434"):
        self.base_url = base_url
        self.model_catalog = {
            # Lightweight models for fast responses
            "ultra_light": [
                "qwen2.5:0.5b-instruct-q4_K_M",
                "tinyllama:1.1b-chat-q4_K_M"
            ],
            # Balanced models for general use
            "balanced": [
                "phi3.5:3.8b-mini-instruct-q4_K_M",
                "llama3.2:3b-instruct-q4_K_M"
            ],
            # Specialized models for specific tasks
            "code_specialist": [
                "codellama:7b-code-q4_K_M",
                "codegemma:7b-code-q4_K_M"
            ],
            # High capability models
            "high_capability": [
                "llama3.1:8b-instruct-q4_K_M",
                "qwen2.5:7b-instruct-q4_K_M"
            ]
        }
    
    def setup_production_models(self, categories: List[str]) -> dict:
        """Set up models for production use."""
        setup_results = {}
        
        for category in categories:
            if category not in self.model_catalog:
                setup_results[category] = {"status": "error", "message": "Unknown category"}
                continue
            
            models = self.model_catalog[category]
            category_results = []
            
            for model in models:
                try:
                    # Pull model
                    response = requests.post(f"{self.base_url}/api/pull", 
                        json={"name": model})
                    
                    if response.status_code == 200:
                        category_results.append({"model": model, "status": "ready"})
                    else:
                        category_results.append({"model": model, "status": "failed"})
                        
                except Exception as e:
                    category_results.append({"model": model, "status": "error", "error": str(e)})
            
            setup_results[category] = category_results
        
        return setup_results
    
    def optimize_for_hardware(self) -> dict:
        """Recommend optimal models based on available hardware."""
        # This would typically check actual hardware specs
        # For demo purposes, we'll simulate hardware detection
        
        recommendations = {
            "low_resource": {
                "models": ["qwen2.5:0.5b-instruct-q4_K_M"],
                "max_concurrent": 1,
                "memory_usage": "< 1GB"
            },
            "medium_resource": {
                "models": ["phi3.5:3.8b-mini-instruct-q4_K_M", "llama3.2:3b-instruct-q4_K_M"],
                "max_concurrent": 2,
                "memory_usage": "2-4GB"
            },
            "high_resource": {
                "models": ["llama3.1:8b-instruct-q4_K_M", "codellama:7b-code-q4_K_M"],
                "max_concurrent": 3,
                "memory_usage": "6-12GB"
            }
        }
        
        return recommendations

# Production model setup
model_manager = OllamaModelManager()
setup_results = model_manager.setup_production_models(["balanced", "ultra_light"])
print(f"Model setup results: {setup_results}")
```

**قائمة التحقق لنشر الإنتاج مع Ollama**:

✅ **تكوين الخدمة**:
- تثبيت خدمة Ollama مع تكامل النظام المناسب
- تكوين النماذج لحالات استخدام الوكلاء المحددة
- إعداد نصوص بدء التشغيل وإدارة الخدمة المناسبة
- اختبار تحميل النموذج وتوافر واجهة برمجة التطبيقات

✅ **إدارة النماذج**:
- تنزيل النماذج المطلوبة والتحقق من سلامتها
- إعداد إجراءات تحديث وتدوير النماذج
- تكوين التخزين المؤقت للنماذج وتحسين التخزين
- اختبار أداء النموذج تحت الحمل المتوقع

✅ **إعداد الأمان**:
- تكوين قواعد جدار الحماية للوصول المحلي فقط
- إعداد ضوابط الوصول إلى واجهة برمجة التطبيقات وتحديد المعدل
- تنفيذ تسجيل التدقيق لتفاعلات الوكلاء
- تكوين تخزين النموذج الآمن والوصول

✅ **تحسين الأداء**:
- قياس أداء النماذج لحالات الاستخدام المتوقعة
- تكوين تسريع الأجهزة المناسب
- إعداد استراتيجيات تسخين وتخزين النموذج
- مراقبة استخدام الموارد ومقاييس الأداء

✅ **اختبار التكامل**:
- اختبار تكامل إطار عمل Microsoft Agent  
- التحقق من قدرات التشغيل دون اتصال  
- اختبار سيناريوهات الفشل ومعالجة الأخطاء  
- التحقق من سير العمل الكامل للوكيل  

**مقارنة مع Foundry Local**:

| الميزة | Foundry Local | Ollama |
|--------|---------------|--------|
| **حالة الاستخدام المستهدفة** | الإنتاج المؤسسي | التطوير والمجتمع |
| **نظام النماذج** | مختار بواسطة Microsoft | مجتمع واسع النطاق |
| **تحسين الأجهزة** | تلقائي (CUDA/NPU/CPU) | إعداد يدوي |
| **ميزات المؤسسة** | مراقبة وأمان مدمجان | أدوات المجتمع |
| **تعقيد النشر** | بسيط (winget install) | بسيط (curl install) |
| **توافق API** | OpenAI + إضافات | معيار OpenAI |
| **الدعم** | رسمي من Microsoft | مدفوع من المجتمع |
| **الأفضل لـ** | وكلاء الإنتاج | النماذج الأولية والبحث |

**متى تختار Ollama**:
- **التطوير والنماذج الأولية**: تجربة سريعة مع نماذج مختلفة  
- **نماذج المجتمع**: الوصول إلى أحدث النماذج المقدمة من المجتمع  
- **الاستخدام التعليمي**: تعلم وتدريس تطوير وكلاء الذكاء الاصطناعي  
- **مشاريع البحث**: البحث الأكاديمي الذي يتطلب الوصول إلى نماذج متنوعة  
- **النماذج المخصصة**: بناء واختبار نماذج مخصصة محسّنة  

### VLLM: استنتاج وكيل SLM عالي الأداء

يوفر VLLM (استنتاج نموذج لغة كبير جدًا) محرك استنتاج عالي الإنتاجية وفعال من حيث الذاكرة، مُحسّن خصيصًا لنشر SLM في الإنتاج على نطاق واسع. بينما يركز Foundry Local على سهولة الاستخدام ويؤكد Ollama على نماذج المجتمع، يتفوق VLLM في سيناريوهات الأداء العالي التي تتطلب أقصى إنتاجية واستخدامًا فعالًا للموارد.

**الهيكل الأساسي والميزات**:
- **PagedAttention**: إدارة ذاكرة ثورية لحساب الانتباه بكفاءة  
- **Dynamic Batching**: تجميع الطلبات الذكي لتحقيق الإنتاجية المثلى  
- **تحسين GPU**: دعم CUDA المتقدم وتوازي المصفوفات  
- **توافق OpenAI**: توافق كامل مع API للتكامل السلس  
- **Speculative Decoding**: تقنيات تسريع الاستنتاج المتقدمة  
- **دعم التكميم**: INT4، INT8، وFP16 لتوفير الذاكرة  

#### التثبيت والإعداد

**خيارات التثبيت**:
```bash
# Standard installation
pip install vllm

# With additional dependencies for agent frameworks
pip install vllm[agent] openai

# Docker deployment for production
docker pull vllm/vllm-openai:latest

# From source for latest features
git clone https://github.com/vllm-project/vllm.git
cd vllm
pip install -e .
```
  
**البدء السريع لتطوير الوكلاء**:
```bash
# Start VLLM server with SLM model
python -m vllm.entrypoints.openai.api_server \
    --model microsoft/Phi-3.5-mini-instruct \
    --trust-remote-code \
    --max-model-len 4096 \
    --gpu-memory-utilization 0.8

# Alternative: Start with Qwen2.5 for lightweight agents
python -m vllm.entrypoints.openai.api_server \
    --model Qwen/Qwen2.5-0.5B-Instruct \
    --trust-remote-code \
    --max-model-len 2048 \
    --tensor-parallel-size 1

# Test API endpoint
curl http://localhost:8000/v1/models

# Test chat completion
curl http://localhost:8000/v1/chat/completions \
    -H "Content-Type: application/json" \
    -d '{
        "model": "microsoft/Phi-3.5-mini-instruct",
        "messages": [{"role": "user", "content": "Hello!"}]
    }'
```
  
#### تكامل إطار عمل الوكيل

**VLLM مع إطار عمل Microsoft Agent**:
```python
from microsoft_agent_framework import Agent, Config
import openai
import subprocess
import time
import requests
from typing import Optional, Dict, Any

class VLLMManager:
    def __init__(self, model_name: str, 
                 host: str = "localhost", 
                 port: int = 8000,
                 gpu_memory_utilization: float = 0.8,
                 max_model_len: int = 4096):
        self.model_name = model_name
        self.host = host
        self.port = port
        self.base_url = f"http://{host}:{port}"
        self.gpu_memory_utilization = gpu_memory_utilization
        self.max_model_len = max_model_len
        self.process = None
        self.client = None
        
    def start_server(self) -> bool:
        """Start VLLM server with optimized settings for agents."""
        try:
            cmd = [
                "python", "-m", "vllm.entrypoints.openai.api_server",
                "--model", self.model_name,
                "--host", self.host,
                "--port", str(self.port),
                "--gpu-memory-utilization", str(self.gpu_memory_utilization),
                "--max-model-len", str(self.max_model_len),
                "--trust-remote-code",
                "--disable-log-requests",  # Reduce logging for agents
                "--served-model-name", self.get_served_model_name()
            ]
            
            self.process = subprocess.Popen(cmd, 
                stdout=subprocess.PIPE, 
                stderr=subprocess.PIPE)
            
            # Wait for server to start
            max_retries = 30
            for _ in range(max_retries):
                if self.health_check():
                    self.client = openai.OpenAI(base_url=f"{self.base_url}/v1")
                    return True
                time.sleep(2)
                
            return False
            
        except Exception as e:
            print(f"Failed to start VLLM server: {e}")
            return False
    
    def get_served_model_name(self) -> str:
        """Get a clean model name for serving."""
        return self.model_name.replace("/", "--")
    
    def health_check(self) -> bool:
        """Check if VLLM server is healthy."""
        try:
            response = requests.get(f"{self.base_url}/health", timeout=5)
            return response.status_code == 200
        except:
            return False
    
    def get_openai_client(self) -> openai.OpenAI:
        """Get OpenAI-compatible client for VLLM."""
        if not self.client:
            self.client = openai.OpenAI(base_url=f"{self.base_url}/v1")
        return self.client
    
    def get_model_info(self) -> Dict[str, Any]:
        """Get model information and statistics."""
        try:
            response = requests.get(f"{self.base_url}/v1/models")
            if response.status_code == 200:
                return response.json()
        except:
            pass
        return {}
    
    def shutdown(self):
        """Shutdown VLLM server."""
        if self.process:
            self.process.terminate()
            self.process.wait()

# Initialize VLLM for high-performance agents
vllm_manager = VLLMManager("microsoft/Phi-3.5-mini-instruct")
if vllm_manager.start_server():
    print("VLLM server started successfully")
    
    # Configure agent with VLLM backend
    agent_config = Config(
        name="vllm-performance-agent",
        model_provider="vllm",
        model_id=vllm_manager.get_served_model_name(),
        endpoint=f"{vllm_manager.base_url}/v1",
        api_key="none"  # VLLM doesn't require API key
    )
    
    agent = Agent(config=agent_config)
else:
    print("Failed to start VLLM server")
```
  
**إعداد متعدد الوكلاء عالي الإنتاجية**:
```python
import asyncio
from concurrent.futures import ThreadPoolExecutor
from microsoft_agent_framework import Agent, Config
import openai

class VLLMHighThroughputManager:
    def __init__(self):
        self.model_configs = {
            "lightweight": {
                "model": "Qwen/Qwen2.5-0.5B-Instruct",
                "port": 8000,
                "max_model_len": 2048,
                "gpu_memory_utilization": 0.3
            },
            "balanced": {
                "model": "microsoft/Phi-3.5-mini-instruct",
                "port": 8001,
                "max_model_len": 4096,
                "gpu_memory_utilization": 0.5
            },
            "capable": {
                "model": "meta-llama/Llama-3.2-3B-Instruct",
                "port": 8002,
                "max_model_len": 8192,
                "gpu_memory_utilization": 0.7
            }
        }
        self.managers = {}
        self.agents = {}
        self.client_pool = {}
        
    async def initialize_all_models(self):
        """Initialize all VLLM models in parallel."""
        initialization_tasks = []
        
        for category, config in self.model_configs.items():
            task = self._initialize_model(category, config)
            initialization_tasks.append(task)
        
        results = await asyncio.gather(*initialization_tasks, return_exceptions=True)
        
        successful_inits = 0
        for i, result in enumerate(results):
            category = list(self.model_configs.keys())[i]
            if isinstance(result, Exception):
                print(f"Failed to initialize {category}: {result}")
            else:
                successful_inits += 1
                print(f"Successfully initialized {category} model")
        
        return successful_inits
    
    async def _initialize_model(self, category: str, config: Dict[str, Any]):
        """Initialize a single VLLM model instance."""
        manager = VLLMManager(
            model_name=config["model"],
            port=config["port"],
            max_model_len=config["max_model_len"],
            gpu_memory_utilization=config["gpu_memory_utilization"]
        )
        
        # Start server in thread to avoid blocking
        loop = asyncio.get_event_loop()
        with ThreadPoolExecutor() as executor:
            success = await loop.run_in_executor(executor, manager.start_server)
        
        if success:
            self.managers[category] = manager
            
            # Create agent
            agent_config = Config(
                name=f"vllm-{category}-agent",
                model_provider="vllm",
                model_id=manager.get_served_model_name(),
                endpoint=f"{manager.base_url}/v1",
                api_key="none"
            )
            
            self.agents[category] = Agent(config=agent_config)
            
            # Create client pool for high throughput
            self.client_pool[category] = [
                openai.OpenAI(base_url=f"{manager.base_url}/v1")
                for _ in range(5)  # 5 clients per model for parallelism
            ]
            
            return True
        else:
            raise Exception(f"Failed to start VLLM server for {category}")
    
    def get_optimal_agent(self, request_complexity: str, current_load: Dict[str, int]) -> str:
        """Select optimal agent based on request complexity and current load."""
        complexity_mapping = {
            "simple": "lightweight",
            "moderate": "balanced", 
            "complex": "capable"
        }
        
        preferred_category = complexity_mapping.get(request_complexity, "balanced")
        
        # Check if preferred agent is available and not overloaded
        if (preferred_category in self.agents and 
            current_load.get(preferred_category, 0) < 10):  # Max 10 concurrent per agent
            return preferred_category
        
        # Fallback to least loaded available agent
        available_agents = [(cat, load) for cat, load in current_load.items() 
                          if cat in self.agents and load < 10]
        
        if available_agents:
            return min(available_agents, key=lambda x: x[1])[0]
        
        return "balanced"  # Default fallback
    
    async def process_batch_requests(self, requests: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Process multiple requests in parallel for maximum throughput."""
        current_load = {cat: 0 for cat in self.agents.keys()}
        tasks = []
        
        for request in requests:
            # Determine optimal agent
            complexity = request.get("complexity", "moderate")
            agent_category = self.get_optimal_agent(complexity, current_load)
            current_load[agent_category] += 1
            
            # Create processing task
            task = self._process_single_request(request, agent_category)
            tasks.append(task)
        
        # Process all requests in parallel
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Format results
        formatted_results = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                formatted_results.append({
                    "request_id": requests[i].get("id", i),
                    "status": "error",
                    "error": str(result)
                })
            else:
                formatted_results.append(result)
        
        return formatted_results
    
    async def _process_single_request(self, request: Dict[str, Any], agent_category: str) -> Dict[str, Any]:
        """Process a single request with the specified agent."""
        start_time = time.time()
        
        try:
            agent = self.agents[agent_category]
            response = await agent.chat_async(request["message"])
            
            processing_time = time.time() - start_time
            
            return {
                "request_id": request.get("id"),
                "status": "success",
                "response": response,
                "agent_used": agent_category,
                "processing_time": processing_time
            }
            
        except Exception as e:
            return {
                "request_id": request.get("id"),
                "status": "error",
                "error": str(e),
                "agent_used": agent_category,
                "processing_time": time.time() - start_time
            }

# High-throughput usage example
throughput_manager = VLLMHighThroughputManager()

# Initialize all models
initialized_count = await throughput_manager.initialize_all_models()
print(f"Initialized {initialized_count} models")

# Process batch requests
batch_requests = [
    {"id": 1, "message": "Simple question", "complexity": "simple"},
    {"id": 2, "message": "Complex analysis needed", "complexity": "complex"},
    {"id": 3, "message": "Moderate difficulty task", "complexity": "moderate"}
]

results = await throughput_manager.process_batch_requests(batch_requests)
for result in results:
    print(f"Request {result['request_id']}: {result['status']} in {result.get('processing_time', 0):.2f}s")
```
  
#### أنماط نشر الإنتاج

**خدمة إنتاج VLLM للمؤسسات**:
```python
import asyncio
import logging
import time
from typing import Dict, List, Optional
from dataclasses import dataclass
from microsoft_agent_framework import Agent, Config
import uvicorn
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel

@dataclass
class VLLMServerConfig:
    model_name: str
    port: int
    gpu_memory_utilization: float
    max_model_len: int
    tensor_parallel_size: int = 1
    quantization: Optional[str] = None

class AgentRequest(BaseModel):
    message: str
    agent_type: str = "general"
    priority: str = "normal"
    timeout: int = 30

class VLLMProductionService:
    def __init__(self, server_configs: Dict[str, VLLMServerConfig]):
        self.server_configs = server_configs
        self.managers = {}
        self.agents = {}
        self.metrics = {
            "requests_processed": 0,
            "requests_failed": 0,
            "total_processing_time": 0,
            "agent_usage": {name: 0 for name in server_configs.keys()},
            "throughput_per_minute": 0
        }
        self.request_queue = asyncio.Queue(maxsize=1000)
        self.processing_workers = []
        self.app = FastAPI(title="VLLM Agent Service")
        self._setup_routes()
        
    async def initialize_production_environment(self):
        """Initialize all VLLM servers for production."""
        logging.info("Initializing VLLM production environment...")
        
        initialization_tasks = []
        for name, config in self.server_configs.items():
            task = self._initialize_server(name, config)
            initialization_tasks.append(task)
        
        results = await asyncio.gather(*initialization_tasks, return_exceptions=True)
        
        successful_servers = 0
        for i, result in enumerate(results):
            server_name = list(self.server_configs.keys())[i]
            if isinstance(result, Exception):
                logging.error(f"Failed to initialize {server_name}: {result}")
            else:
                successful_servers += 1
                logging.info(f"Successfully initialized {server_name}")
        
        if successful_servers == 0:
            raise Exception("No VLLM servers could be initialized")
        
        # Start processing workers
        self.processing_workers = [
            asyncio.create_task(self._processing_worker(i))
            for i in range(min(4, successful_servers))  # 4 workers max
        ]
        
        logging.info(f"Production environment ready with {successful_servers} servers")
        return successful_servers
    
    async def _initialize_server(self, name: str, config: VLLMServerConfig):
        """Initialize a single VLLM server."""
        manager = VLLMManager(
            model_name=config.model_name,
            port=config.port,
            gpu_memory_utilization=config.gpu_memory_utilization,
            max_model_len=config.max_model_len
        )
        
        # Add quantization if specified
        if config.quantization:
            # This would be added to the manager's start command
            pass
        
        success = manager.start_server()
        if success:
            self.managers[name] = manager
            
            # Create production agent
            agent_config = Config(
                name=f"vllm-production-{name}",
                model_provider="vllm",
                model_id=manager.get_served_model_name(),
                endpoint=f"{manager.base_url}/v1",
                api_key="none",
                timeout=30.0
            )
            
            agent = Agent(config=agent_config)
            
            # Add production tools
            self._add_production_tools(agent, name)
            
            self.agents[name] = agent
            return True
        else:
            raise Exception(f"Failed to start VLLM server for {name}")
    
    def _add_production_tools(self, agent: Agent, server_type: str):
        """Add production tools based on server type."""
        if server_type == "customer_service":
            @agent.tool
            def escalate_to_human(issue: str, customer_id: str) -> str:
                """Escalate complex issues to human agents."""
                return f"Escalated issue for customer {customer_id}: {issue}"
            
            @agent.tool
            def lookup_order_status(order_id: str) -> dict:
                """Look up order status from production database."""
                # Production database lookup
                return {"order_id": order_id, "status": "shipped", "eta": "2 days"}
        
        elif server_type == "technical_support":
            @agent.tool
            def run_system_diagnostics(system_id: str) -> dict:
                """Run comprehensive system diagnostics."""
                return {"system_id": system_id, "status": "healthy", "issues": []}
            
            @agent.tool
            def create_incident_report(description: str, severity: str) -> str:
                """Create incident report in production system."""
                incident_id = f"INC-{hash(description) % 100000:05d}"
                return f"Created incident {incident_id} with severity {severity}"
    
    def _setup_routes(self):
        """Set up FastAPI routes for production service."""
        @self.app.post("/chat")
        async def chat_endpoint(request: AgentRequest, background_tasks: BackgroundTasks):
            try:
                # Add request to queue
                await self.request_queue.put({
                    "request": request,
                    "timestamp": time.time(),
                    "future": asyncio.Future()
                })
                
                # Wait for processing (with timeout)
                result = await asyncio.wait_for(
                    self._wait_for_result(request),
                    timeout=request.timeout
                )
                
                return result
                
            except asyncio.TimeoutError:
                raise HTTPException(status_code=408, detail="Request timeout")
            except Exception as e:
                raise HTTPException(status_code=500, detail=str(e))
        
        @self.app.get("/health")
        async def health_endpoint():
            return await self.get_health_status()
        
        @self.app.get("/metrics")
        async def metrics_endpoint():
            return self.get_production_metrics()
    
    async def _processing_worker(self, worker_id: int):
        """Background worker for processing agent requests."""
        logging.info(f"Starting processing worker {worker_id}")
        
        while True:
            try:
                # Get request from queue
                queue_item = await self.request_queue.get()
                request_data = queue_item["request"]
                request_future = queue_item["future"]
                
                # Select appropriate agent
                agent_name = self._select_agent(request_data.agent_type)
                
                if agent_name not in self.agents:
                    request_future.set_exception(Exception(f"Agent {agent_name} not available"))
                    continue
                
                # Process request
                start_time = time.time()
                try:
                    agent = self.agents[agent_name]
                    response = await agent.chat_async(request_data.message)
                    
                    processing_time = time.time() - start_time
                    
                    # Update metrics
                    self.metrics["requests_processed"] += 1
                    self.metrics["total_processing_time"] += processing_time
                    self.metrics["agent_usage"][agent_name] += 1
                    
                    result = {
                        "response": response,
                        "agent_used": agent_name,
                        "processing_time": processing_time,
                        "worker_id": worker_id
                    }
                    
                    request_future.set_result(result)
                    
                except Exception as e:
                    self.metrics["requests_failed"] += 1
                    request_future.set_exception(e)
                
                finally:
                    self.request_queue.task_done()
                    
            except Exception as e:
                logging.error(f"Worker {worker_id} error: {e}")
                await asyncio.sleep(1)
    
    def _select_agent(self, agent_type: str) -> str:
        """Select appropriate agent based on request type."""
        agent_mapping = {
            "customer_service": "customer_service",
            "technical": "technical_support",
            "general": "general_purpose"
        }
        
        return agent_mapping.get(agent_type, "general_purpose")
    
    async def _wait_for_result(self, request: AgentRequest):
        """Wait for request processing to complete."""
        # This is simplified - in production you'd track futures properly
        await asyncio.sleep(0.1)  # Placeholder
        return {"response": "Processed", "status": "success"}
    
    async def get_health_status(self) -> dict:
        """Get comprehensive health status of all services."""
        health_status = {
            "overall_status": "healthy",
            "servers": {},
            "queue_size": self.request_queue.qsize(),
            "active_workers": len([w for w in self.processing_workers if not w.done()]),
            "timestamp": time.time()
        }
        
        unhealthy_servers = 0
        for name, manager in self.managers.items():
            try:
                is_healthy = manager.health_check()
                health_status["servers"][name] = {
                    "status": "healthy" if is_healthy else "unhealthy",
                    "endpoint": manager.base_url,
                    "model": manager.model_name
                }
                if not is_healthy:
                    unhealthy_servers += 1
            except Exception as e:
                health_status["servers"][name] = {
                    "status": "error",
                    "error": str(e)
                }
                unhealthy_servers += 1
        
        if unhealthy_servers > 0:
            health_status["overall_status"] = "degraded" if unhealthy_servers < len(self.managers) else "unhealthy"
        
        return health_status
    
    def get_production_metrics(self) -> dict:
        """Get production performance metrics."""
        total_requests = self.metrics["requests_processed"] + self.metrics["requests_failed"]
        avg_processing_time = (
            self.metrics["total_processing_time"] / self.metrics["requests_processed"]
            if self.metrics["requests_processed"] > 0 else 0
        )
        
        success_rate = (
            self.metrics["requests_processed"] / total_requests * 100
            if total_requests > 0 else 0
        )
        
        return {
            "total_requests": total_requests,
            "successful_requests": self.metrics["requests_processed"],
            "failed_requests": self.metrics["requests_failed"],
            "success_rate_percent": success_rate,
            "average_processing_time_seconds": avg_processing_time,
            "agent_usage_distribution": self.metrics["agent_usage"],
            "queue_size": self.request_queue.qsize()
        }
    
    async def start_production_server(self, host: str = "0.0.0.0", port: int = 8080):
        """Start the production FastAPI server."""
        config = uvicorn.Config(
            self.app,
            host=host,
            port=port,
            log_level="info",
            workers=1  # Single worker for simplicity
        )
        server = uvicorn.Server(config)
        await server.serve()

# Production deployment example
production_configs = {
    "customer_service": VLLMServerConfig(
        model_name="microsoft/Phi-3.5-mini-instruct",
        port=8000,
        gpu_memory_utilization=0.4,
        max_model_len=4096
    ),
    "technical_support": VLLMServerConfig(
        model_name="meta-llama/Llama-3.2-3B-Instruct",
        port=8001,
        gpu_memory_utilization=0.6,
        max_model_len=8192
    ),
    "general_purpose": VLLMServerConfig(
        model_name="Qwen/Qwen2.5-1.5B-Instruct",
        port=8002,
        gpu_memory_utilization=0.3,
        max_model_len=2048
    )
}

production_service = VLLMProductionService(production_configs)

# Initialize and start production service
# await production_service.initialize_production_environment()
# await production_service.start_production_server()
```
  
#### ميزات المؤسسة والمراقبة

**مراقبة أداء VLLM المتقدمة**:
```python
import psutil
import nvidia_ml_py3 as nvml
from dataclasses import dataclass
from typing import List, Dict, Optional
import json
import asyncio

@dataclass
class PerformanceMetrics:
    timestamp: float
    requests_per_second: float
    average_latency_ms: float
    gpu_utilization_percent: float
    gpu_memory_used_gb: float
    cpu_utilization_percent: float
    memory_used_gb: float
    queue_length: int
    active_requests: int

class VLLMAdvancedMonitoring:
    def __init__(self, vllm_managers: Dict[str, VLLMManager]):
        self.managers = vllm_managers
        self.metrics_history = []
        self.alert_thresholds = {
            "gpu_utilization_max": 95,
            "gpu_memory_max_gb": 10,
            "latency_max_ms": 3000,
            "queue_length_max": 50,
            "error_rate_max_percent": 10
        }
        
        # Initialize NVIDIA ML for GPU monitoring
        try:
            nvml.nvmlInit()
            self.gpu_monitoring_available = True
            self.gpu_count = nvml.nvmlDeviceGetCount()
        except:
            self.gpu_monitoring_available = False
            self.gpu_count = 0
    
    async def collect_comprehensive_metrics(self) -> Dict[str, PerformanceMetrics]:
        """Collect detailed performance metrics for all VLLM instances."""
        all_metrics = {}
        
        for name, manager in self.managers.items():
            try:
                metrics = await self._collect_single_instance_metrics(name, manager)
                all_metrics[name] = metrics
            except Exception as e:
                logging.error(f"Failed to collect metrics for {name}: {e}")
                # Create error metrics
                all_metrics[name] = PerformanceMetrics(
                    timestamp=time.time(),
                    requests_per_second=0,
                    average_latency_ms=0,
                    gpu_utilization_percent=0,
                    gpu_memory_used_gb=0,
                    cpu_utilization_percent=0,
                    memory_used_gb=0,
                    queue_length=0,
                    active_requests=0
                )
        
        return all_metrics
    
    async def _collect_single_instance_metrics(self, name: str, manager: VLLMManager) -> PerformanceMetrics:
        """Collect metrics for a single VLLM instance."""
        timestamp = time.time()
        
        # Get VLLM-specific metrics via API
        vllm_stats = await self._get_vllm_stats(manager)
        
        # Get system metrics
        cpu_percent = psutil.cpu_percent(interval=0.1)
        memory_info = psutil.virtual_memory()
        memory_used_gb = memory_info.used / (1024**3)
        
        # Get GPU metrics if available
        gpu_utilization = 0
        gpu_memory_used = 0
        
        if self.gpu_monitoring_available and self.gpu_count > 0:
            try:
                # Assuming first GPU for simplicity
                handle = nvml.nvmlDeviceGetHandleByIndex(0)
                gpu_util = nvml.nvmlDeviceGetUtilizationRates(handle)
                gpu_utilization = gpu_util.gpu
                
                gpu_mem = nvml.nvmlDeviceGetMemoryInfo(handle)
                gpu_memory_used = gpu_mem.used / (1024**3)
                
            except Exception as e:
                logging.warning(f"GPU monitoring failed: {e}")
        
        return PerformanceMetrics(
            timestamp=timestamp,
            requests_per_second=vllm_stats.get("requests_per_second", 0),
            average_latency_ms=vllm_stats.get("average_latency_ms", 0),
            gpu_utilization_percent=gpu_utilization,
            gpu_memory_used_gb=gpu_memory_used,
            cpu_utilization_percent=cpu_percent,
            memory_used_gb=memory_used_gb,
            queue_length=vllm_stats.get("queue_length", 0),
            active_requests=vllm_stats.get("active_requests", 0)
        )
    
    async def _get_vllm_stats(self, manager: VLLMManager) -> dict:
        """Get VLLM-specific statistics via API calls."""
        try:
            # Test inference to measure latency
            start_time = time.time()
            client = manager.get_openai_client()
            
            response = await asyncio.wait_for(
                asyncio.to_thread(
                    client.chat.completions.create,
                    model=manager.get_served_model_name(),
                    messages=[{"role": "user", "content": "ping"}],
                    max_tokens=1
                ),
                timeout=5.0
            )
            
            latency_ms = (time.time() - start_time) * 1000
            
            return {
                "average_latency_ms": latency_ms,
                "requests_per_second": 1000 / latency_ms if latency_ms > 0 else 0,
                "queue_length": 0,  # Would need to be exposed by VLLM
                "active_requests": 1  # Approximation
            }
            
        except Exception as e:
            logging.warning(f"Failed to get VLLM stats: {e}")
            return {
                "average_latency_ms": 0,
                "requests_per_second": 0,
                "queue_length": 0,
                "active_requests": 0
            }
    
    def generate_performance_report(self, time_window_minutes: int = 60) -> dict:
        """Generate comprehensive performance report."""
        cutoff_time = time.time() - (time_window_minutes * 60)
        recent_metrics = [
            metrics for metrics in self.metrics_history
            if any(m.timestamp > cutoff_time for m in metrics.values())
        ]
        
        if not recent_metrics:
            return {"error": "No recent metrics available"}
        
        report = {
            "time_window_minutes": time_window_minutes,
            "total_samples": len(recent_metrics),
            "instances": {}
        }
        
        # Analyze each instance
        for instance_name in self.managers.keys():
            instance_metrics = [
                metrics[instance_name] for metrics in recent_metrics
                if instance_name in metrics
            ]
            
            if instance_metrics:
                report["instances"][instance_name] = {
                    "avg_latency_ms": sum(m.average_latency_ms for m in instance_metrics) / len(instance_metrics),
                    "max_latency_ms": max(m.average_latency_ms for m in instance_metrics),
                    "avg_gpu_utilization": sum(m.gpu_utilization_percent for m in instance_metrics) / len(instance_metrics),
                    "avg_requests_per_second": sum(m.requests_per_second for m in instance_metrics) / len(instance_metrics),
                    "max_queue_length": max(m.queue_length for m in instance_metrics),
                    "availability_percent": (len(instance_metrics) / len(recent_metrics)) * 100
                }
        
        return report
    
    async def auto_scaling_recommendations(self) -> List[dict]:
        """Generate auto-scaling recommendations based on performance metrics."""
        recommendations = []
        
        if not self.metrics_history:
            return recommendations
        
        latest_metrics = self.metrics_history[-1]
        
        for instance_name, metrics in latest_metrics.items():
            # High latency recommendation
            if metrics.average_latency_ms > self.alert_thresholds["latency_max_ms"]:
                recommendations.append({
                    "instance": instance_name,
                    "type": "scale_up",
                    "reason": f"High latency: {metrics.average_latency_ms:.0f}ms",
                    "suggestion": "Consider adding tensor parallelism or increasing GPU memory"
                })
            
            # High GPU utilization recommendation
            if metrics.gpu_utilization_percent > self.alert_thresholds["gpu_utilization_max"]:
                recommendations.append({
                    "instance": instance_name,
                    "type": "scale_out",
                    "reason": f"High GPU utilization: {metrics.gpu_utilization_percent:.1f}%",
                    "suggestion": "Consider adding additional GPU instances"
                })
            
            # Low utilization recommendation
            if (metrics.gpu_utilization_percent < 20 and 
                metrics.requests_per_second < 1):
                recommendations.append({
                    "instance": instance_name,
                    "type": "scale_down",
                    "reason": f"Low utilization: {metrics.gpu_utilization_percent:.1f}% GPU, {metrics.requests_per_second:.1f} RPS",
                    "suggestion": "Consider consolidating workloads or reducing resources"
                })
        
        return recommendations

# Advanced monitoring setup
monitoring = VLLMAdvancedMonitoring({
    "customer_service": vllm_manager,
    # Add other managers as needed
})

async def advanced_monitoring_loop():
    """Advanced monitoring with auto-scaling recommendations."""
    while True:
        try:
            # Collect metrics
            metrics = await monitoring.collect_comprehensive_metrics()
            monitoring.metrics_history.append(metrics)
            
            # Keep only last 1000 entries
            if len(monitoring.metrics_history) > 1000:
                monitoring.metrics_history = monitoring.metrics_history[-1000:]
            
            # Generate recommendations every 5 minutes
            if len(monitoring.metrics_history) % 10 == 0:  # Every 10th collection (5 minutes if collecting every 30s)
                recommendations = await monitoring.auto_scaling_recommendations()
                
                if recommendations:
                    logging.info(f"Auto-scaling recommendations: {recommendations}")
            
            # Generate performance report every hour
            if len(monitoring.metrics_history) % 120 == 0:  # Every 120th collection (1 hour)
                report = monitoring.generate_performance_report(60)
                logging.info(f"Performance report: {json.dumps(report, indent=2)}")
            
        except Exception as e:
            logging.error(f"Advanced monitoring error: {e}")
        
        await asyncio.sleep(30)  # Collect metrics every 30 seconds

# Start advanced monitoring
# asyncio.create_task(advanced_monitoring_loop())
```
  
#### التكوين والتحسين المتقدم

**قوالب تكوين إنتاج VLLM**:
```python
from enum import Enum
from typing import Dict, Any

class DeploymentScenario(Enum):
    DEVELOPMENT = "development"
    STAGING = "staging"
    PRODUCTION_LOW = "production_low"
    PRODUCTION_HIGH = "production_high"
    ENTERPRISE = "enterprise"

class VLLMConfigTemplates:
    """Production-ready VLLM configuration templates."""
    
    @staticmethod
    def get_config_template(scenario: DeploymentScenario) -> Dict[str, Any]:
        """Get optimized configuration for deployment scenario."""
        
        templates = {
            DeploymentScenario.DEVELOPMENT: {
                "gpu_memory_utilization": 0.6,
                "max_model_len": 2048,
                "tensor_parallel_size": 1,
                "pipeline_parallel_size": 1,
                "quantization": None,
                "enable_prefix_caching": False,
                "max_num_seqs": 32,
                "max_num_batched_tokens": 2048
            },
            
            DeploymentScenario.STAGING: {
                "gpu_memory_utilization": 0.8,
                "max_model_len": 4096,
                "tensor_parallel_size": 1,
                "pipeline_parallel_size": 1,
                "quantization": "awq",
                "enable_prefix_caching": True,
                "max_num_seqs": 64,
                "max_num_batched_tokens": 4096
            },
            
            DeploymentScenario.PRODUCTION_LOW: {
                "gpu_memory_utilization": 0.85,
                "max_model_len": 4096,
                "tensor_parallel_size": 1,
                "pipeline_parallel_size": 1,
                "quantization": "awq",
                "enable_prefix_caching": True,
                "max_num_seqs": 128,
                "max_num_batched_tokens": 8192,
                "enable_chunked_prefill": True
            },
            
            DeploymentScenario.PRODUCTION_HIGH: {
                "gpu_memory_utilization": 0.9,
                "max_model_len": 8192,
                "tensor_parallel_size": 2,
                "pipeline_parallel_size": 1,
                "quantization": "awq",
                "enable_prefix_caching": True,
                "max_num_seqs": 256,
                "max_num_batched_tokens": 16384,
                "enable_chunked_prefill": True,
                "speculative_model": "small_draft_model"
            },
            
            DeploymentScenario.ENTERPRISE: {
                "gpu_memory_utilization": 0.95,
                "max_model_len": 16384,
                "tensor_parallel_size": 4,
                "pipeline_parallel_size": 2,
                "quantization": "awq",
                "enable_prefix_caching": True,
                "max_num_seqs": 512,
                "max_num_batched_tokens": 32768,
                "enable_chunked_prefill": True,
                "speculative_model": "optimized_draft_model",
                "guided_decoding_backend": "outlines"
            }
        }
        
        return templates[scenario]
    
    @staticmethod
    def generate_vllm_command(model_name: str, 
                             scenario: DeploymentScenario,
                             port: int = 8000,
                             host: str = "0.0.0.0") -> List[str]:
        """Generate optimized VLLM command for deployment scenario."""
        
        config = VLLMConfigTemplates.get_config_template(scenario)
        
        cmd = [
            "python", "-m", "vllm.entrypoints.openai.api_server",
            "--model", model_name,
            "--host", host,
            "--port", str(port),
            "--gpu-memory-utilization", str(config["gpu_memory_utilization"]),
            "--max-model-len", str(config["max_model_len"]),
            "--tensor-parallel-size", str(config["tensor_parallel_size"]),
            "--max-num-seqs", str(config["max_num_seqs"]),
            "--max-num-batched-tokens", str(config["max_num_batched_tokens"]),
            "--trust-remote-code",
            "--disable-log-requests"
        ]
        
        # Add optional parameters
        if config.get("quantization"):
            cmd.extend(["--quantization", config["quantization"]])
        
        if config.get("enable_prefix_caching"):
            cmd.append("--enable-prefix-caching")
        
        if config.get("enable_chunked_prefill"):
            cmd.append("--enable-chunked-prefill")
        
        if config.get("pipeline_parallel_size", 1) > 1:
            cmd.extend(["--pipeline-parallel-size", str(config["pipeline_parallel_size"])])
        
        if config.get("speculative_model"):
            cmd.extend(["--speculative-model", config["speculative_model"]])
        
        return cmd

# Usage examples
dev_cmd = VLLMConfigTemplates.generate_vllm_command(
    "microsoft/Phi-3.5-mini-instruct",
    DeploymentScenario.DEVELOPMENT,
    port=8000
)

prod_cmd = VLLMConfigTemplates.generate_vllm_command(
    "microsoft/Phi-3.5-mini-instruct",
    DeploymentScenario.PRODUCTION_HIGH,
    port=8001
)

print(f"Development command: {' '.join(dev_cmd)}")
print(f"Production command: {' '.join(prod_cmd)}")
```
  
**قائمة التحقق لنشر إنتاج VLLM**:

✅ **تحسين الأجهزة**:
- إعداد توازي المصفوفات للأجهزة متعددة GPU  
- تمكين التكميم (AWQ/GPTQ) لتوفير الذاكرة  
- ضبط استخدام ذاكرة GPU الأمثل (85-95%)  
- إعداد أحجام دفعات مناسبة للإنتاجية  

✅ **ضبط الأداء**:
- تمكين التخزين المؤقت للمقدمات للاستفسارات المتكررة  
- إعداد التعبئة المجزأة للتسلسلات الطويلة  
- إعداد فك التشفير التخميني للاستنتاج الأسرع  
- تحسين max_num_seqs بناءً على الأجهزة  

✅ **ميزات الإنتاج**:
- إعداد مراقبة الصحة وجمع المقاييس  
- تكوين إعادة التشغيل التلقائي والتجاوز  
- تنفيذ قوائم انتظار الطلبات وتوازن التحميل  
- إعداد تسجيل شامل وتنبيهات  

✅ **الأمان والموثوقية**:
- إعداد قواعد الجدار الناري وعناصر التحكم في الوصول  
- إعداد تحديد معدل API والمصادقة  
- تنفيذ الإغلاق المنظم والتنظيف  
- إعداد النسخ الاحتياطي واستعادة الكوارث  

✅ **اختبار التكامل**:
- اختبار تكامل إطار عمل Microsoft Agent  
- التحقق من سيناريوهات الإنتاجية العالية  
- اختبار إجراءات الفشل والاسترداد  
- قياس الأداء تحت الحمل  

**مقارنة مع الحلول الأخرى**:

| الميزة | VLLM | Foundry Local | Ollama |
|--------|------|---------------|--------|
| **حالة الاستخدام المستهدفة** | الإنتاجية العالية | سهولة الاستخدام المؤسسية | التطوير والمجتمع |
| **الأداء** | أقصى إنتاجية | متوازن | جيد |
| **كفاءة الذاكرة** | تحسين PagedAttention | تحسين تلقائي | قياسي |
| **تعقيد الإعداد** | عالي (معلمات كثيرة) | منخفض (تلقائي) | منخفض (بسيط) |
| **القابلية للتوسع** | ممتاز (توازي المصفوفات/الأنابيب) | جيد | محدود |
| **التكميم** | متقدم (AWQ، GPTQ، FP8) | تلقائي | قياسي GGUF |
| **ميزات المؤسسة** | تحتاج إلى تنفيذ مخصص | مدمج | أدوات المجتمع |
| **الأفضل لـ** | وكلاء الإنتاج على نطاق واسع | الإنتاج المؤسسي | التطوير |

**متى تختار VLLM**:
- **متطلبات الإنتاجية العالية**: معالجة مئات الطلبات في الثانية  
- **النشر واسع النطاق**: نشر متعدد GPU ومتعدد العقد  
- **الأداء الحرج**: أوقات استجابة أقل من الثانية على نطاق واسع  
- **التحسين المتقدم**: الحاجة إلى التكميم والتجميع المخصص  
- **كفاءة الموارد**: الاستخدام الأمثل للأجهزة GPU المكلفة  

## تطبيقات وكلاء SLM في العالم الحقيقي

### وكلاء خدمة العملاء SLM
- **قدرات SLM**: البحث عن الحسابات، إعادة تعيين كلمات المرور، التحقق من حالة الطلبات  
- **فوائد التكلفة**: تقليل تكاليف الاستنتاج بمقدار 10 أضعاف مقارنة بوكلاء LLM  
- **الأداء**: أوقات استجابة أسرع مع جودة متسقة للاستفسارات الروتينية  

### وكلاء عمليات الأعمال SLM
- **وكلاء معالجة الفواتير**: استخراج البيانات، التحقق من المعلومات، التوجيه للموافقة  
- **وكلاء إدارة البريد الإلكتروني**: التصنيف، تحديد الأولويات، صياغة الردود تلقائيًا  
- **وكلاء الجدولة**: تنسيق الاجتماعات، إدارة الجداول، إرسال التذكيرات  

### مساعدات رقمية شخصية SLM
- **وكلاء إدارة المهام**: إنشاء، تحديث، تنظيم قوائم المهام بكفاءة  
- **وكلاء جمع المعلومات**: البحث عن المواضيع، تلخيص النتائج محليًا  
- **وكلاء التواصل**: صياغة رسائل البريد الإلكتروني، الرسائل، منشورات وسائل التواصل الاجتماعي بشكل خاص  

### وكلاء التداول والمالية SLM
- **وكلاء مراقبة السوق**: تتبع الأسعار، تحديد الاتجاهات في الوقت الفعلي  
- **وكلاء إنشاء التقارير**: إنشاء ملخصات يومية/أسبوعية تلقائيًا  
- **وكلاء تقييم المخاطر**: تقييم مواقف المحافظ باستخدام البيانات المحلية  

### وكلاء دعم الرعاية الصحية SLM
- **وكلاء جدولة المرضى**: تنسيق المواعيد، إرسال التذكيرات التلقائية  
- **وكلاء التوثيق**: إنشاء ملخصات وتقارير طبية محليًا  
- **وكلاء إدارة الوصفات الطبية**: تتبع التجديدات، التحقق من التفاعلات بشكل خاص  

## إطار عمل Microsoft Agent: تطوير وكلاء جاهزين للإنتاج

### نظرة عامة والهيكل

يوفر إطار عمل Microsoft Agent منصة شاملة على مستوى المؤسسات لبناء ونشر وإدارة وكلاء الذكاء الاصطناعي الذين يمكنهم العمل في بيئات السحابة والحافة دون اتصال. تم تصميم الإطار خصيصًا للعمل بسلاسة مع نماذج اللغة الصغيرة وسيناريوهات الحوسبة الطرفية، مما يجعله مثاليًا للنشر الذي يركز على الخصوصية والموارد المحدودة.

**مكونات الإطار الأساسية**:
- **وقت تشغيل الوكيل**: بيئة تنفيذ خفيفة الوزن محسّنة للأجهزة الطرفية  
- **نظام تكامل الأدوات**: بنية إضافية قابلة للتوسيع لتوصيل الخدمات الخارجية وواجهات برمجة التطبيقات  
- **إدارة الحالة**: ذاكرة الوكيل المستمرة ومعالجة السياق عبر الجلسات  
- **طبقة الأمان**: عناصر تحكم أمان مدمجة للنشر المؤسسي  
- **محرك التنسيق**: تنسيق متعدد الوكلاء وإدارة سير العمل  

### الميزات الرئيسية للنشر الطرفي

**هيكلية أولاً دون اتصال**: تم تصميم إطار عمل Microsoft Agent بمبادئ أولاً دون اتصال، مما يتيح للوكلاء العمل بفعالية دون اتصال دائم بالإنترنت. يشمل ذلك استنتاج النموذج المحلي، قواعد المعرفة المخزنة مؤقتًا، تنفيذ الأدوات دون اتصال، والتدهور التدريجي عند عدم توفر خدمات السحابة.

**تحسين الموارد**: يوفر الإطار إدارة موارد ذكية مع تحسين تلقائي للذاكرة لنماذج SLM، توازن تحميل CPU/GPU للأجهزة الطرفية، اختيار نموذج تكيفي بناءً على الموارد المتاحة، وأنماط استنتاج موفرة للطاقة للنشر المحمول.

**الأمان والخصوصية**: تشمل ميزات الأمان على مستوى المؤسسات معالجة البيانات المحلية للحفاظ على الخصوصية، قنوات اتصال مشفرة للوكيل، عناصر تحكم وصول قائمة على الأدوار لقدرات الوكيل، وتسجيل التدقيق لمتطلبات الامتثال.

### التكامل مع Foundry Local

يتكامل إطار عمل Microsoft Agent بسلاسة مع Foundry Local لتوفير حل AI طرفي كامل:

**اكتشاف النموذج التلقائي**: يكتشف الإطار تلقائيًا ويتصل بحالات Foundry Local، ويكتشف نماذج SLM المتاحة، ويختار النماذج المثلى بناءً على متطلبات الوكيل وقدرات الأجهزة.

**تحميل النموذج الديناميكي**: يمكن للوكلاء تحميل نماذج SLM مختلفة ديناميكيًا لمهام محددة، مما يتيح أنظمة وكلاء متعددة النماذج حيث تتعامل النماذج المختلفة مع أنواع مختلفة من الطلبات، والتجاوز التلقائي بين النماذج بناءً على التوافر والأداء.

**تحسين الأداء**: تقلل آليات التخزين المؤقت المدمجة من أوقات تحميل النموذج، وتحسن تجميع الاتصال مكالمات API إلى Foundry Local، وتحسن التجميع الذكي الإنتاجية لطلبات الوكلاء المتعددة.

### بناء الوكلاء باستخدام إطار عمل Microsoft Agent

#### تعريف الوكيل والتكوين

```python
from microsoft_agent_framework import Agent, Tool, Config
from foundry_local import FoundryLocalManager

# Configure agent with Foundry Local integration
config = Config(
    name="customer-service-agent",
    model_provider="foundry-local",
    model_alias="phi-4-mini",
    max_tokens=512,
    temperature=0.1,
    offline_mode=True
)

# Initialize Foundry Local connection
foundry = FoundryLocalManager("phi-4-mini")

# Create agent instance
agent = Agent(
    config=config,
    model_endpoint=foundry.endpoint,
    api_key=foundry.api_key
)
```
  
#### تكامل الأدوات لسيناريوهات الحافة

```python
# Define tools for offline operation
@agent.tool
def lookup_customer_info(customer_id: str) -> dict:
    """Look up customer information from local database."""
    # Local database query - works offline
    return local_db.get_customer(customer_id)

@agent.tool
def create_support_ticket(issue: str, priority: str) -> str:
    """Create a support ticket in local system."""
    # Local ticket creation with sync when online
    ticket_id = local_system.create_ticket(issue, priority)
    return f"Ticket {ticket_id} created successfully"

@agent.tool
def schedule_callback(customer_id: str, preferred_time: str) -> str:
    """Schedule a callback for the customer."""
    # Local scheduling with calendar integration
    return local_calendar.schedule(customer_id, preferred_time)
```
  
#### تنسيق متعدد الوكلاء

```python
from microsoft_agent_framework import AgentOrchestrator

# Create specialized agents for different domains
scheduling_agent = Agent(
    config=Config(
        name="scheduling-agent",
        model_alias="qwen2.5-0.5b",  # Lightweight for simple tasks
        specialized_for="scheduling"
    )
)

technical_support_agent = Agent(
    config=Config(
        name="technical-agent",
        model_alias="phi-4-mini",  # More capable for complex issues
        specialized_for="technical_support"
    )
)

# Orchestrate multiple agents
orchestrator = AgentOrchestrator([
    scheduling_agent,
    technical_support_agent
])

# Route requests based on intent
result = orchestrator.process_request(
    "I need to schedule a callback for a technical issue",
    routing_strategy="intent-based"
)
```
  
### أنماط نشر الحافة المتقدمة

#### هيكلية الوكيل الهرمية

**مجموعات الوكلاء المحلية**: نشر وكلاء SLM متخصصين متعددين على الأجهزة الطرفية، كل منهم محسّن لمهام محددة. استخدم نماذج خفيفة مثل Qwen2.5-0.5B للتوجيه والجدولة البسيطة، ونماذج متوسطة مثل Phi-4-Mini لخدمة العملاء والتوثيق، ونماذج أكبر للتفكير المعقد عند توفر الموارد.

**التنسيق بين الحافة والسحابة**: تنفيذ أنماط التصعيد الذكية حيث يتعامل الوكلاء المحليون مع المهام الروتينية، توفر الوكلاء السحابيون التفكير المعقد عند توفر الاتصال، ويحافظ التسليم السلس بين معالجة الحافة والسحابة على الاستمرارية.

#### تكوينات النشر

**نشر جهاز واحد**:
```yaml
deployment:
  type: single-device
  hardware: edge-device
  models:
    - alias: "phi-4-mini"
      primary: true
      tasks: ["conversation", "reasoning"]
    - alias: "qwen2.5-0.5b"
      secondary: true
      tasks: ["routing", "classification"]
  agents:
    - name: "primary-agent"
      model: "phi-4-mini"
      tools: ["database", "calendar", "email"]
```
  
**نشر طرفي موزع**:
```yaml
deployment:
  type: distributed-edge
  nodes:
    - id: "edge-1"
      agents: ["customer-service", "scheduling"]
      models: ["phi-4-mini"]
    - id: "edge-2"
      agents: ["technical-support", "documentation"]
      models: ["qwen2.5-coder-0.5b"]
  coordination:
    load_balancing: true
    failover: automatic
```
  
### تحسين الأداء لوكلاء الحافة

#### استراتيجيات اختيار النموذج

**تعيين النموذج بناءً على المهمة**: يتيح إطار عمل Microsoft Agent اختيار النموذج الذكي بناءً على تعقيد المهمة والمتطلبات:

- **المهام البسيطة** (الأسئلة والأجوبة، التوجيه): Qwen2.5-0.5B (500MB، <100ms استجابة)  
- **المهام المتوسطة** (خدمة العملاء، الجدولة): Phi-4-Mini (2.4GB، 200-500ms استجابة)  
- **المهام المعقدة** (التحليل الفني، التخطيط): Phi-4 (7GB، 1-3s استجابة عند توفر الموارد)  

**التبديل الديناميكي للنماذج**: يمكن للوكلاء التبديل بين النماذج بناءً على الحمل الحالي للنظام، تقييم تعقيد المهمة، مستويات أولوية المستخدم، وموارد الأجهزة المتاحة.

#### إدارة الذاكرة والموارد

```python
# Configure resource constraints for edge deployment
resource_config = ResourceConfig(
    max_memory_usage="4GB",
    max_concurrent_agents=3,
    model_cache_size="2GB",
    auto_unload_idle_models=True,
    power_management=True
)

agent = Agent(
    config=config,
    resource_limits=resource_config
)
```
  
### أنماط التكامل المؤسسي

#### الأمان والامتثال

**معالجة البيانات المحلية**: تتم معالجة جميع بيانات الوكيل محليًا، مما يضمن عدم مغادرة البيانات الحساسة للجهاز الطرفي. يشمل ذلك حماية معلومات العملاء، الامتثال لـ HIPAA لوكلاء الرعاية الصحية، أمان البيانات المالية لوكلاء البنوك، والامتثال لـ GDPR للنشر الأوروبي.

**التحكم في الوصول**: تتحكم الأذونات القائمة على الأدوار في الأدوات التي يمكن للوكلاء الوصول إليها، مصادقة المستخدم لتفاعلات الوكيل، ومسارات التدقيق لجميع إجراءات وقرارات الوكيل.

#### المراقبة والملاحظة

```python
from microsoft_agent_framework import AgentMonitor

# Set up monitoring for edge agents
monitor = AgentMonitor(
    metrics=["response_time", "success_rate", "resource_usage"],
    alerts=[
        {"metric": "response_time", "threshold": "2s", "action": "scale_down_model"},
        {"metric": "memory_usage", "threshold": "80%", "action": "unload_idle_agents"}
    ],
    local_storage=True  # Store metrics locally for offline operation
)

agent.add_monitor(monitor)
```
  
### أمثلة على التنفيذ في العالم الحقيقي

#### نظام وكيل التجزئة الطرفي

```python
# Retail kiosk agent for in-store customer assistance
retail_agent = Agent(
    config=Config(
        name="retail-assistant",
        model_alias="phi-4-mini",
        context="You are a helpful retail assistant in an electronics store."
    )
)

@retail_agent.tool
def check_inventory(product_sku: str) -> dict:
    """Check local inventory for a product."""
    return local_inventory.lookup(product_sku)

@retail_agent.tool
def find_alternatives(product_category: str) -> list:
    """Find alternative products in the same category."""
    return local_catalog.find_similar(product_category)

@retail_agent.tool
def create_price_quote(items: list) -> dict:
    """Generate a price quote for multiple items."""
    return pricing_engine.calculate_quote(items)
```
  
#### وكيل دعم الرعاية الصحية

```python
# HIPAA-compliant patient support agent
healthcare_agent = Agent(
    config=Config(
        name="patient-support",
        model_alias="phi-4-mini",
        privacy_mode=True,  # Enhanced privacy for healthcare
        compliance=["HIPAA"]
    )
)

@healthcare_agent.tool
def check_appointment_availability(provider_id: str, date_range: str) -> list:
    """Check appointment slots with healthcare provider."""
    return local_scheduling.get_availability(provider_id, date_range)

@healthcare_agent.tool
def access_patient_portal(patient_id: str, auth_token: str) -> dict:
    """Secure access to patient information."""
    if security.validate_token(auth_token):
        return patient_portal.get_summary(patient_id)
    return {"error": "Authentication failed"}
```
  
### أفضل الممارسات لإطار عمل Microsoft Agent

#### إرشادات التطوير

1. **ابدأ ببساطة**: ابدأ بسيناريوهات الوكيل الفردي قبل بناء أنظمة متعددة الوكلاء المعقدة  
2. **اختيار النموذج المناسب**: اختر أصغر نموذج يلبي متطلبات الدقة الخاصة بك  
3. **تصميم الأدوات**: قم بإنشاء أدوات مركزة ذات غرض واحد بدلاً من أدوات متعددة الوظائف المعقدة  
4. **معالجة الأخطاء**: تنفيذ التدهور التدريجي للسيناريوهات دون اتصال وأعطال النموذج  
5. **الاختبار**: اختبار الوكلاء بشكل مكثف في ظروف دون اتصال وبيئات محدودة الموارد  

#### أفضل ممارسات النشر

1. **النشر التدريجي**: النشر لمجموعات مستخدمين صغيرة في البداية، مراقبة مقاييس الأداء عن كثب  
2. **مراقبة الموارد**: إعداد تنبيهات للذاكرة، وحدة المعالجة المركزية، وحدود وقت الاستجابة  
3. **استراتيجيات التراجع**: دائمًا وجود خطط احتياطية لفشل النموذج أو استنفاد الموارد  
4. **الأمان أولاً**: تنفيذ عناصر التحكم الأمنية من البداية، وليس كفكرة لاحقة  
5. **التوثيق**: الحفاظ على توثيق واضح لقدرات الوكيل وحدوده  

### خارطة الطريق المستقبلية والتكامل

يستمر إطار عمل Microsoft Agent في التطور مع تحسين SLM، أدوات نشر الحافة المحسّنة، إدارة الموارد المحسّنة للبيئات المحدودة، وتوسيع نظام الأدوات البيئي لسيناريوهات المؤسسات الشائعة.

**الميزات القادمة**:
- **AutoML لتحسين الوكلاء**: تحسين تلقائي لنماذج SLM لمهام الوكلاء المحددة  
- **شبكات الحافة الشبكية**: التنسيق بين نشرات الوكلاء الطرفية المتعددة  
- **القياس عن بعد المتقدم**: مراقبة وتحليلات محسّنة لأداء الوكلاء  
- **منشئ الوكلاء المرئي**: أدوات تطوير الوكلاء منخفضة/بدون كود  

## أفضل الممارسات لتنفيذ وكلاء SLM

### إرشادات اختيار SLM للوكلاء

عند اختيار SLM لنشر الوكلاء، ضع في اعتبارك العوامل التالية:

**اعتبارات حجم النموذج**: اختر النماذج فائقة الضغط مثل Q2_K لتطبيقات الوكلاء المحمولة للغاية، والنماذج المتوازنة مثل Q4_K_M لسيناريوهات الوكلاء العامة، والنماذج ذات الدقة العالية مثل Q8_0 للتطبيقات الحرجة للجودة.

**توافق استخدام الوكيل**: طابق قدرات SLM مع متطلبات الوكيل المحددة، مع مراعاة عوامل مثل الحفاظ على الدقة لقرارات الوكيل، سرعة الاستنتاج لتفاعلات الوكيل في الوقت الفعلي، قيود الذاكرة لنشر الوكلاء الطرفيين، ومتطلبات التشغيل دون اتصال للوكلاء الذين يركزون على الخصوصية.

### اختيار استراتيجية التحسين لوكلاء SLM

**نهج التكميم للوكلاء**: اختر مستويات التكميم المناسبة بناءً على متطلبات جودة الوكيل وقيود الأجهزة. ضع في اعتبارك Q4_0 لأقصى ضغط في الوكلاء المحمولين، Q5_1 للجودة-الضغط المتوازن في الوكلاء العامين، وQ8_0 للجودة القريبة من الأصل في تطبيقات الوكلاء الحرجة.
**اختيار الإطار لتوزيع الوكلاء**: اختر أطر تحسين بناءً على الأجهزة المستهدفة ومتطلبات الوكلاء. استخدم Llama.cpp لتوزيع الوكلاء المحسنين للمعالجات، Apple MLX لتطبيقات الوكلاء على أجهزة Apple Silicon، وONNX لتوافق الوكلاء عبر المنصات.

## تحويل الوكلاء SLM واستخداماتهم العملية

### سيناريوهات توزيع الوكلاء في العالم الحقيقي

**تطبيقات الوكلاء على الأجهزة المحمولة**: تنسيقات Q4_K ممتازة لتطبيقات الوكلاء على الهواتف الذكية بذاكرة محدودة، بينما يوفر Q8_0 أداءً متوازنًا لأنظمة الوكلاء على الأجهزة اللوحية. تنسيقات Q5_K تقدم جودة عالية لوكلاء الإنتاجية على الأجهزة المحمولة.

**الحوسبة المكتبية ووكلاء الحافة**: يوفر Q5_K أداءً مثاليًا لتطبيقات الوكلاء على أجهزة الكمبيوتر المكتبية، بينما يقدم Q8_0 استنتاجًا عالي الجودة لبيئات الوكلاء على محطات العمل، ويتيح Q4_K معالجة فعالة على أجهزة الحافة.

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

### معايير أداء وكلاء SLM

**سرعة استنتاج الوكلاء**: يحقق Q4_K أسرع أوقات استجابة للوكلاء على معالجات الهواتف المحمولة، بينما يوفر Q5_K نسبة متوازنة بين السرعة والجودة لتطبيقات الوكلاء العامة، ويقدم Q8_0 جودة فائقة للمهام المعقدة للوكلاء، وتوفر التنسيقات التجريبية أعلى معدل إنتاجية للأجهزة المتخصصة للوكلاء.

**متطلبات ذاكرة الوكلاء**: تتراوح مستويات التكميم للوكلاء من Q2_K (أقل من 500 ميجابايت لنماذج الوكلاء الصغيرة) إلى Q8_0 (حوالي 50% من الحجم الأصلي)، مع تحقيق التكوينات التجريبية أقصى ضغط للبيئات التي تعاني من قيود الموارد.

## التحديات والاعتبارات لوكلاء SLM

### التوازن بين الأداء في أنظمة الوكلاء

يتطلب توزيع وكلاء SLM النظر بعناية في التوازن بين حجم النموذج، سرعة استجابة الوكلاء، وجودة المخرجات. بينما يوفر Q4_K سرعة وكفاءة استثنائية للوكلاء على الأجهزة المحمولة، يقدم Q8_0 جودة فائقة للمهام المعقدة للوكلاء. ويحقق Q5_K توازنًا مناسبًا لمعظم تطبيقات الوكلاء العامة.

### توافق الأجهزة لوكلاء SLM

تختلف قدرات أجهزة الحافة لتوزيع وكلاء SLM. يعمل Q4_K بكفاءة على المعالجات الأساسية للوكلاء البسيطين، بينما يتطلب Q5_K موارد حسابية معتدلة لأداء متوازن للوكلاء، ويستفيد Q8_0 من الأجهزة عالية الأداء لقدرات الوكلاء المتقدمة.

### الأمن والخصوصية في أنظمة وكلاء SLM

بينما تتيح وكلاء SLM المعالجة المحلية لتعزيز الخصوصية، يجب تنفيذ تدابير أمنية مناسبة لحماية نماذج الوكلاء والبيانات في بيئات الحافة. هذا مهم بشكل خاص عند توزيع تنسيقات الوكلاء عالية الدقة في بيئات المؤسسات أو تنسيقات الوكلاء المضغوطة في التطبيقات التي تتعامل مع بيانات حساسة.

## الاتجاهات المستقبلية في تطوير وكلاء SLM

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

**توقعات السوق لوكلاء SLM**: وفقًا لأبحاث حديثة، يمكن أن تقضي الأتمتة المدعومة بالوكلاء على 40-60% من المهام الإدراكية المتكررة في سير العمل المؤسسي بحلول عام 2027، مع قيادة SLM لهذا التحول بفضل كفاءتها من حيث التكلفة ومرونة التوزيع.

**اتجاهات التكنولوجيا في وكلاء SLM**:
- **وكلاء SLM المتخصصون**: نماذج مخصصة مدربة لمهام وكلاء معينة وصناعات محددة
- **الحوسبة على الحافة للوكلاء**: تحسين قدرات الوكلاء على الأجهزة مع تعزيز الخصوصية وتقليل التأخير
- **تنسيق الوكلاء**: تحسين التنسيق بين عدة وكلاء SLM مع التوجيه الديناميكي وتوزيع الحمل
- **الديمقراطية**: مرونة SLM تمكن مشاركة أوسع في تطوير الوكلاء عبر المؤسسات

## البدء مع وكلاء SLM

### الخطوة 1: إعداد بيئة Microsoft Agent Framework

**تثبيت التبعيات**:
```bash
# Install Microsoft Agent Framework
pip install microsoft-agent-framework

# Install Foundry Local SDK for edge deployment
pip install foundry-local-sdk

# Install additional dependencies for edge agents
pip install openai asyncio
```

**تهيئة Foundry Local**:
```bash
# Start Foundry Local service
foundry service start

# Load default model for agent development
foundry model run phi-4-mini
```

### الخطوة 2: اختيار SLM لتطبيقات الوكلاء
خيارات شائعة لـ Microsoft Agent Framework:
- **Microsoft Phi-4 Mini (3.8B)**: ممتاز لمهام الوكلاء العامة مع أداء متوازن
- **Qwen2.5-0.5B (0.5B)**: فعال للغاية للوكلاء البسيطين في التوجيه والتصنيف
- **Qwen2.5-Coder-0.5B (0.5B)**: متخصص لمهام الوكلاء المتعلقة بالبرمجة
- **Phi-4 (7B)**: استنتاج متقدم للسيناريوهات المعقدة على الحافة عند توفر الموارد

### الخطوة 3: إنشاء أول وكيل باستخدام Microsoft Agent Framework

**إعداد الوكيل الأساسي**:
```python
from microsoft_agent_framework import Agent, Config
from foundry_local import FoundryLocalManager

# Initialize Foundry Local connection
foundry = FoundryLocalManager("phi-4-mini")

# Create agent configuration
config = Config(
    name="my-first-agent",
    model_provider="foundry-local",
    model_alias="phi-4-mini",
    offline_mode=True
)

# Create and configure agent
agent = Agent(
    config=config,
    model_endpoint=foundry.endpoint,
    api_key=foundry.api_key
)

# Define a simple tool
@agent.tool
def get_current_time() -> str:
    """Get the current time."""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

# Test the agent
response = agent.chat("What time is it?")
print(response)
```

### الخطوة 4: تحديد نطاق الوكيل ومتطلباته
ابدأ بتطبيقات وكلاء مركزة ومحددة جيدًا باستخدام Microsoft Agent Framework:
- **وكلاء المجال الواحد**: خدمة العملاء أو الجدولة أو البحث
- **أهداف الوكيل الواضحة**: أهداف محددة وقابلة للقياس لأداء الوكلاء
- **دمج الأدوات المحدود**: 3-5 أدوات كحد أقصى لتوزيع الوكلاء الأولي
- **حدود الوكيل المحددة**: مسارات تصعيد واضحة للسيناريوهات المعقدة
- **تصميم يركز على الحافة**: إعطاء الأولوية للوظائف غير المتصلة والمعالجة المحلية

### الخطوة 5: تنفيذ التوزيع على الحافة باستخدام Microsoft Agent Framework

**تكوين الموارد**:
```python
from microsoft_agent_framework import ResourceConfig

# Configure for edge deployment
resource_config = ResourceConfig(
    max_memory_usage="2GB",
    max_concurrent_agents=2,
    model_cache_size="1GB",
    auto_unload_idle_models=True,
    power_management=True
)

agent = Agent(
    config=config,
    resource_limits=resource_config
)
```

**توزيع تدابير السلامة لوكلاء الحافة**:
- **التحقق المحلي من المدخلات**: فحص الطلبات دون الاعتماد على السحابة
- **تصفية المخرجات غير المتصلة**: ضمان استيفاء الردود لمعايير الجودة محليًا
- **ضوابط أمان الحافة**: تنفيذ الأمان دون الحاجة إلى الاتصال بالإنترنت
- **المراقبة المحلية**: تتبع الأداء والإبلاغ عن المشكلات باستخدام قياس الأداء على الحافة

### الخطوة 6: قياس وتحسين أداء وكلاء الحافة
- **معدلات إكمال مهام الوكلاء**: مراقبة معدلات النجاح في السيناريوهات غير المتصلة
- **أوقات استجابة الوكلاء**: ضمان أوقات استجابة أقل من الثانية لتوزيع الحافة
- **استخدام الموارد**: تتبع استخدام الذاكرة، وحدة المعالجة المركزية، والبطارية على أجهزة الحافة
- **كفاءة التكلفة**: مقارنة تكاليف التوزيع على الحافة بالبدائل القائمة على السحابة
- **موثوقية غير متصلة**: قياس أداء الوكلاء أثناء انقطاع الشبكة

## النقاط الرئيسية لتطبيق وكلاء SLM

1. **SLMs كافية للوكلاء**: بالنسبة لمعظم مهام الوكلاء، النماذج الصغيرة تقدم أداءً مشابهًا للنماذج الكبيرة مع تقديم مزايا كبيرة
2. **كفاءة التكلفة في الوكلاء**: تشغيل وكلاء SLM أرخص بـ 10-30 مرة، مما يجعلها اقتصادية للتوزيع الواسع
3. **التخصص يعمل للوكلاء**: غالبًا ما تتفوق SLMs المخصصة على LLMs العامة في تطبيقات الوكلاء المحددة
4. **هيكل الوكلاء الهجين**: استخدام SLMs للمهام الروتينية للوكلاء، وLLMs للتفكير المعقد عند الضرورة
5. **Microsoft Agent Framework يتيح التوزيع الإنتاجي**: يوفر أدوات على مستوى المؤسسات لبناء، توزيع، وإدارة وكلاء الحافة
6. **مبادئ التصميم التي تركز على الحافة**: وكلاء قادرون على العمل دون اتصال مع معالجة محلية لضمان الخصوصية والموثوقية
7. **تكامل Foundry Local**: اتصال سلس بين Microsoft Agent Framework واستنتاج النموذج المحلي
8. **المستقبل هو وكلاء SLM**: نماذج اللغة الصغيرة مع أطر الإنتاج هي مستقبل الذكاء الاصطناعي الوكيل، مما يتيح توزيع الوكلاء بشكل ديمقراطي وفعال

## المراجع والقراءة الإضافية

### الأوراق البحثية والمنشورات الأساسية

#### وكلاء الذكاء الاصطناعي وأنظمة الوكلاء
- **"Language Agents as Optimizable Graphs"** (2024) - بحث أساسي حول تصميم الوكلاء واستراتيجيات التحسين
  - المؤلفون: Wenyue Hua, Lishan Yang, وآخرون
  - الرابط: https://arxiv.org/abs/2402.16823
  - الأفكار الرئيسية: تصميم الوكلاء القائم على الرسوم البيانية واستراتيجيات التحسين

- **"The Rise and Potential of Large Language Model Based Agents"** (2023)
  - المؤلفون: Zhiheng Xi, Wenxiang Chen, وآخرون
  - الرابط: https://arxiv.org/abs/2309.07864
  - الأفكار الرئيسية: استعراض شامل لقدرات وتطبيقات الوكلاء القائمة على LLM

- **"Cognitive Architectures for Language Agents"** (2024)
  - المؤلفون: Theodore Sumers, Shunyu Yao, وآخرون
  - الرابط: https://arxiv.org/abs/2309.02427
  - الأفكار الرئيسية: أطر معرفية لتصميم الوكلاء الذكيين

#### نماذج اللغة الصغيرة والتحسين
- **"Phi-3 Technical Report: A Highly Capable Language Model Locally on Your Phone"** (2024)
  - المؤلفون: فريق Microsoft Research
  - الرابط: https://arxiv.org/abs/2404.14219
  - الأفكار الرئيسية: مبادئ تصميم SLM واستراتيجيات التوزيع على الأجهزة المحمولة

- **"Qwen2.5 Technical Report"** (2024)
  - المؤلفون: فريق Alibaba Cloud
  - الرابط: https://arxiv.org/abs/2407.10671
  - الأفكار الرئيسية: تقنيات تدريب SLM المتقدمة وتحسين الأداء

- **"TinyLlama: An Open-Source Small Language Model"** (2024)
  - المؤلفون: Peiyuan Zhang, Guangtao Zeng, وآخرون
  - الرابط: https://arxiv.org/abs/2401.02385
  - الأفكار الرئيسية: تصميم نموذج مضغوط للغاية وكفاءة التدريب

### الوثائق الرسمية والأطر

#### Microsoft Agent Framework
- **الوثائق الرسمية**: https://docs.microsoft.com/en-us/azure/ai-services/agents/
- **مستودع GitHub**: https://github.com/microsoft/agent-framework

#### Foundry Local
- **المستودع الأساسي**: https://github.com/microsoft/foundry-local
- **الوثائق**: https://github.com/microsoft/foundry-local/blob/main/docs/README.md


#### VLLM
- **المستودع الرئيسي**: https://github.com/vllm-project/vllm
- **الوثائق**: https://docs.vllm.ai/


#### Ollama
- **الموقع الرسمي**: https://ollama.ai/
- **مستودع GitHub**: https://github.com/ollama/ollama

### أطر تحسين النماذج

#### Llama.cpp
- **المستودع**: https://github.com/ggml-org/llama.cpp


#### Microsoft Olive
- **الوثائق**: https://microsoft.github.io/Olive/
- **مستودع GitHub**: https://github.com/microsoft/Olive

#### OpenVINO
- **الموقع الرسمي**: https://docs.openvino.ai/

#### Apple MLX
- **المستودع**: https://github.com/ml-explore/mlx

### تقارير الصناعة وتحليل السوق

#### أبحاث سوق وكلاء الذكاء الاصطناعي
- **"The State of AI Agents 2025"** - معهد McKinsey العالمي
  - الرابط: https://www.mckinsey.com/capabilities/mckinsey-digital/our-insights/ai-agents-2025
  - الأفكار الرئيسية: اتجاهات السوق وأنماط التبني المؤسسي

#### المعايير الفنية

- **"Edge AI Inference Benchmarks"** - MLPerf
  - الرابط: https://mlcommons.org/en/inference-edge/
  - الأفكار الرئيسية: مقاييس الأداء الموحدة لتوزيع الحافة

### المعايير والمواصفات

#### تنسيقات النماذج والمعايير
- **ONNX (Open Neural Network Exchange)**: https://onnx.ai/
  - تنسيق نموذج عبر المنصات للتشغيل البيني
- **GGUF Specification**: https://github.com/ggerganov/ggml/blob/master/docs/gguf.md
  - تنسيق نموذج مكمم لاستنتاج المعالجات
- **OpenAI API Specification**: https://platform.openai.com/docs/api-reference
  - تنسيق API قياسي لتكامل نماذج اللغة

#### الأمن والامتثال
- **NIST AI Risk Management Framework**: https://www.nist.gov/itl/ai-risk-management-framework
- **ISO/IEC 23053:2022 - AI Systems**: إطار لأنظمة الذكاء الاصطناعي والسلامة
- **IEEE Standards for AI**: https://standards.ieee.org/industry-connections/ai/

يمثل التحول نحو وكلاء SLM تغييرًا جذريًا في كيفية تعاملنا مع توزيع الذكاء الاصطناعي. يوفر Microsoft Agent Framework، جنبًا إلى جنب مع المنصات المحلية ونماذج اللغة الصغيرة الفعالة، حلاً كاملاً لبناء وكلاء جاهزين للإنتاج يعملون بفعالية في بيئات الحافة. من خلال التركيز على الكفاءة، التخصص، والفائدة العملية، يجعل هذا النظام التقني وكلاء الذكاء الاصطناعي أكثر سهولة، اقتصادية، وفعالية للتطبيقات الواقعية عبر كل صناعة وبيئة حافة.

مع تقدمنا ​​في عام 2025، فإن الجمع بين النماذج الصغيرة الأكثر قدرة، الأطر المتقدمة للوكلاء مثل Microsoft Agent Framework، ومنصات التوزيع القوية على الحافة سيفتح إمكانيات جديدة للأنظمة المستقلة التي يمكنها العمل بكفاءة على أجهزة الحافة مع الحفاظ على الخصوصية، تقليل التكاليف، وتقديم تجارب مستخدم استثنائية.

**الخطوات التالية للتنفيذ**:
1. **استكشاف استدعاء الوظائف**: تعلم كيفية تعامل SLMs مع تكامل الأدوات والمخرجات المهيكلة
2. **إتقان بروتوكول سياق النموذج (MCP)**: فهم أنماط الاتصال المتقدمة للوكلاء
3. **بناء وكلاء الإنتاج**: استخدام Microsoft Agent Framework للتوزيعات على مستوى المؤسسات
4. **التحسين للحافة**: تطبيق تقنيات تحسين متقدمة للبيئات ذات الموارد المحدودة

## ➡️ ما التالي

- [02: استدعاء الوظائف في نماذج اللغة الصغيرة (SLMs)](./02.FunctionCalling.md)

---

**إخلاء المسؤولية**:  
تم ترجمة هذا المستند باستخدام خدمة الترجمة بالذكاء الاصطناعي [Co-op Translator](https://github.com/Azure/co-op-translator). بينما نسعى لتحقيق الدقة، يرجى العلم أن الترجمات الآلية قد تحتوي على أخطاء أو عدم دقة. يجب اعتبار المستند الأصلي بلغته الأصلية المصدر الموثوق. للحصول على معلومات حاسمة، يُوصى بالترجمة البشرية الاحترافية. نحن غير مسؤولين عن أي سوء فهم أو تفسيرات خاطئة ناتجة عن استخدام هذه الترجمة.