# القسم 02: استدعاء الوظائف في نماذج اللغة الصغيرة (SLMs)

## جدول المحتويات
1. [ما هو استدعاء الوظائف؟](../../../Module06)
2. [كيف يعمل استدعاء الوظائف](../../../Module06)
3. [سيناريوهات التطبيق](../../../Module06)
4. [إعداد استدعاء الوظائف باستخدام Phi-4-mini و Ollama](../../../Module06)
5. [العمل مع استدعاء وظائف Qwen3](../../../Module06)
6. [التكامل المحلي مع Foundry](../../../Module06)
7. [أفضل الممارسات وحل المشكلات](../../../Module06)
8. [أمثلة متقدمة](../../../Module06)

## ما هو استدعاء الوظائف؟

استدعاء الوظائف هو قدرة قوية تتيح لنماذج اللغة الصغيرة (SLMs) التفاعل مع الأدوات الخارجية وواجهات البرمجة (APIs) والخدمات. بدلاً من أن تكون محدودة ببيانات التدريب الخاصة بها، يمكن للنماذج الآن:

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

هذه القدرة تحول نماذج اللغة الصغيرة من مولدات نصوص ثابتة إلى وكلاء ذكاء اصطناعي ديناميكيين يمكنهم تنفيذ مهام واقعية.

## كيف يعمل استدعاء الوظائف

عملية استدعاء الوظائف تتبع سير عمل منهجي:

### 1. تكامل الأدوات
- **الأدوات الخارجية**: يمكن للنماذج الاتصال بواجهات برمجة الطقس، قواعد البيانات، خدمات الويب، وأنظمة خارجية أخرى
- **تعريف الوظائف**: يتم تعريف كل أداة بمعايير محددة، تنسيقات الإدخال/الإخراج، ووصف
- **التوافق مع واجهات البرمجة**: يتم التكامل من خلال واجهات قياسية (REST APIs، SDKs، إلخ)

### 2. تعريف الوظائف
يتم تعريف الوظائف بثلاثة مكونات رئيسية:
```json
{
  "name": "function_name",
  "description": "Clear description of what the function does",
  "parameters": {
    "parameter_name": {
      "description": "What this parameter represents",
      "type": "data_type",
      "default": "default_value"
    }
  }
}
```

### 3. اكتشاف النوايا
- **معالجة اللغة الطبيعية**: يقوم النموذج بتحليل إدخال المستخدم لفهم النية
- **مطابقة الوظائف**: يحدد الوظيفة (الوظائف) المطلوبة لتلبية الطلب
- **استخراج المعلمات**: يحدد ويستخرج المعلمات المطلوبة من رسالة المستخدم

### 4. إنشاء إخراج JSON
يقوم النموذج بإنشاء JSON منظم يحتوي على:
- اسم الوظيفة المطلوب استدعاؤها
- المعلمات المطلوبة مع القيم المناسبة
- سياق التنفيذ والبيانات الوصفية

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

### 6. دمج الاستجابة
- **معالجة النتائج**: يتم إرجاع إخراج الوظيفة إلى النموذج
- **دمج السياق**: يقوم النموذج بدمج النتائج في استجابته
- **التواصل مع المستخدم**: يعرض المعلومات بطريقة طبيعية ومحادثية

## سيناريوهات التطبيق

### استرجاع البيانات
تحويل استفسارات اللغة الطبيعية إلى استدعاءات API منظمة:
- **"أظهر طلباتي الأخيرة"** → استعلام قاعدة بيانات باستخدام معرف المستخدم وفلاتر التاريخ
- **"ما حالة الطقس في طوكيو؟"** → استدعاء API الطقس مع معلمة الموقع
- **"ابحث عن رسائل البريد الإلكتروني من جون الأسبوع الماضي"** → استعلام خدمة البريد الإلكتروني مع المرسل وفلاتر التاريخ

### تنفيذ العمليات
تحويل طلبات المستخدم إلى استدعاءات وظائف محددة:
- **"جدول اجتماع غدًا الساعة 2 مساءً"** → تكامل API التقويم
- **"أرسل رسالة إلى الفريق"** → API منصة التواصل
- **"قم بإنشاء نسخة احتياطية لملفاتي"** → عملية نظام الملفات

### المهام الحسابية
التعامل مع العمليات الرياضية أو المنطقية المعقدة:
- **"احسب الفائدة المركبة على 10,000 دولار بنسبة 5% لمدة 10 سنوات"** → وظيفة حساب مالي
- **"حلل هذه البيانات للاتجاهات"** → أدوات التحليل الإحصائي
- **"قم بتحسين هذا المسار للتسليم"** → خوارزميات تحسين المسار

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

### تكامل واجهة المستخدم/تجربة المستخدم
تمكين تحديثات واجهة ديناميكية:
- **"أظهر بيانات المبيعات على لوحة التحكم"** → إنشاء وعرض الرسوم البيانية
- **"قم بتحديث الخريطة بالمواقع الجديدة"** → تكامل بيانات جغرافية
- **"قم بتحديث عرض المخزون"** → مزامنة البيانات في الوقت الحقيقي

## إعداد استدعاء الوظائف باستخدام Phi-4-mini و Ollama

يدعم Phi-4-mini من Microsoft استدعاء الوظائف الفردية والمتوازية من خلال Ollama. إليك كيفية إعداده:

### المتطلبات
- إصدار Ollama 0.5.13 أو أعلى
- نموذج Phi-4-mini (الموصى به: `phi4-mini:3.8b-fp16`)

### خطوات التثبيت

#### 1. تثبيت وتشغيل Phi-4-mini
```bash
# Download the model (if not already present)
ollama run phi4-mini:3.8b-fp16

# Verify the model is available
ollama list
```

#### 2. إنشاء قالب ModelFile مخصص
نظرًا للقيود الحالية في قوالب Ollama الافتراضية، تحتاج إلى إنشاء ModelFile مخصص باستخدام القالب التالي:

```modelfile
TEMPLATE """
{{- if .Messages }}
{{- if or .System .Tools }}<|system|>
{{ if .System }}{{ .System }}
{{- end }}
In addition to plain text responses, you can chose to call one or more of the provided functions.
Use the following rule to decide when to call a function:
* if the response can be generated from your internal knowledge (e.g., as in the case of queries like "What is the capital of Poland?"), do so
* if you need external information that can be obtained by calling one or more of the provided functions, generate a function calls
If you decide to call functions:
* prefix function calls with functools marker (no closing marker required)
* all function calls should be generated in a single JSON list formatted as functools[{"name": [function name], "arguments": [function arguments as JSON]}, ...]
* follow the provided JSON schema. Do not hallucinate arguments or values. Do to blindly copy values from the provided samples
* respect the argument type formatting. E.g., if the type if number and format is float, write value 7 as 7.0
* make sure you pick the right functions that match the user intent
Available functions as JSON spec:
{{- if .Tools }}
{{ .Tools }}
{{- end }}<|end|>
{{- end }}
{{- range .Messages }}
{{- if ne .Role "system" }}<|{{ .Role }}|>
{{- if and .Content (eq .Role "tools") }}
{"result": {{ .Content }}}
{{- else if .Content }}
{{ .Content }}
{{- else if .ToolCalls }}
functools[
{{- range .ToolCalls }}{{ "{" }}"name": "{{ .Function.Name }}", "arguments": {{ .Function.Arguments }}{{ "}" }}
{{- end }}]
{{- end }}<|end|>
{{- end }}
{{- end }}<|assistant|>
{{ else }}
{{- if .System }}<|system|>
{{ .System }}<|end|>{{ end }}{{ if .Prompt }}<|user|>
{{ .Prompt }}<|end|>{{ end }}<|assistant|>
{{ end }}{{ .Response }}{{ if .Response }}<|user|>{{ end }}
"""
```

#### 3. إنشاء النموذج المخصص
```bash
# Save the template above as 'Modelfile' and run:
ollama create phi4-mini-fc:3.8b-fp16 -f ./Modelfile
```

### مثال على استدعاء وظيفة فردية

```python
import json
import requests

# Define the tool/function
tools = [
    {
        "name": "get_weather",
        "description": "Get current weather information for a location",
        "parameters": {
            "location": {
                "description": "The city or location name",
                "type": "str",
                "default": "New York"
            },
            "units": {
                "description": "Temperature units (celsius or fahrenheit)",
                "type": "str",
                "default": "celsius"
            }
        }
    }
]

# Create the message with system prompt including tools
messages = [
    {
        "role": "system",
        "content": "You are a helpful weather assistant",
        "tools": json.dumps(tools)
    },
    {
        "role": "user",
        "content": "What's the weather like in London today?"
    }
]

# Make request to Ollama API
response = requests.post(
    "http://localhost:11434/api/chat",
    json={
        "model": "phi4-mini-fc:3.8b-fp16",
        "messages": messages,
        "stream": False
    }
)

print(response.json())
```

### مثال على استدعاء وظائف متوازية

```python
import json
import requests

# Define multiple tools for parallel execution
AGENT_TOOLS = {
    "booking_flight": {
        "name": "booking_flight",
        "description": "Book a flight ticket",
        "parameters": {
            "departure": {
                "description": "Departure airport code",
                "type": "str"
            },
            "destination": {
                "description": "Destination airport code", 
                "type": "str"
            },
            "outbound_date": {
                "description": "Departure date (YYYY-MM-DD)",
                "type": "str"
            },
            "return_date": {
                "description": "Return date (YYYY-MM-DD)",
                "type": "str"
            }
        }
    },
    "booking_hotel": {
        "name": "booking_hotel",
        "description": "Book a hotel room",
        "parameters": {
            "city": {
                "description": "City name for hotel booking",
                "type": "str"
            },
            "check_in_date": {
                "description": "Check-in date (YYYY-MM-DD)",
                "type": "str"
            },
            "check_out_date": {
                "description": "Check-out date (YYYY-MM-DD)",
                "type": "str"
            }
        }
    }
}

SYSTEM_PROMPT = """
You are my travel agent with some tools available.
"""

messages = [
    {
        "role": "system",
        "content": SYSTEM_PROMPT,
        "tools": json.dumps(AGENT_TOOLS)
    },
    {
        "role": "user", 
        "content": "I need to travel from London to New York from March 21 2025 to March 27 2025. Please book both flight and hotel."
    }
]

# The model will generate parallel function calls
response = requests.post(
    "http://localhost:11434/api/chat",
    json={
        "model": "phi4-mini-fc:3.8b-fp16",
        "messages": messages,
        "stream": False
    }
)

print(response.json())
```

## العمل مع استدعاء وظائف Qwen3

يوفر Qwen3 قدرات متقدمة لاستدعاء الوظائف مع أداء ومرونة ممتازين. إليك كيفية تنفيذه:

### استخدام إطار عمل Qwen-Agent

يوفر Qwen-Agent إطار عمل عالي المستوى يبسط تنفيذ استدعاء الوظائف:

#### التثبيت
```bash
pip install -U "qwen-agent[gui,rag,code_interpreter,mcp]"
```

#### الإعداد الأساسي

```python
import os
from qwen_agent.agents import Assistant

# Configure the LLM
llm_cfg = {
    'model': 'Qwen3-8B',
    # Option 1: Use Alibaba Model Studio
    'model_type': 'qwen_dashscope',
    'api_key': os.getenv('DASHSCOPE_API_KEY'),
    
    # Option 2: Use local deployment
    # 'model_server': 'http://localhost:8000/v1',
    # 'api_key': 'EMPTY',
    
    # Optional configuration for thinking mode
    'generate_cfg': {
        'thought_in_content': True,  # Include reasoning in response
    }
}

# Define tools using MCP (Model Context Protocol)
tools = [
    {
        'mcpServers': {
            'time': {
                'command': 'uvx',
                'args': ['mcp-server-time', '--local-timezone=Asia/Shanghai']
            },
            'fetch': {
                'command': 'uvx', 
                'args': ['mcp-server-fetch']
            }
        }
    },
    'code_interpreter',  # Built-in code execution tool
]

# Create the assistant
bot = Assistant(llm=llm_cfg, function_list=tools)

# Example usage
messages = [
    {
        'role': 'user', 
        'content': 'What time is it now? Also, fetch the latest news from https://example.com/news'
    }
]

# Generate response with function calling
for response in bot.run(messages=messages):
    print(response)
```

### تنفيذ وظائف مخصصة

يمكنك أيضًا تعريف وظائف مخصصة لـ Qwen3:

```python
import json
from qwen_agent.tools.base import BaseTool

class WeatherTool(BaseTool):
    description = 'Get weather information for a specific location'
    parameters = [
        {
            'name': 'location',
            'type': 'string', 
            'description': 'City or location name',
            'required': True
        },
        {
            'name': 'units',
            'type': 'string',
            'description': 'Temperature units (celsius or fahrenheit)',
            'required': False,
            'default': 'celsius'
        }
    ]
    
    def call(self, params: str, **kwargs) -> str:
        """Execute the weather lookup"""
        params_dict = json.loads(params)
        location = params_dict.get('location')
        units = params_dict.get('units', 'celsius')
        
        # Simulate weather API call
        weather_data = {
            'location': location,
            'temperature': '22°C' if units == 'celsius' else '72°F',
            'condition': 'Partly cloudy',
            'humidity': '65%'
        }
        
        return json.dumps(weather_data)

# Use the custom tool
tools = [WeatherTool()]
bot = Assistant(llm=llm_cfg, function_list=tools)

messages = [{'role': 'user', 'content': 'What\'s the weather in Tokyo?'}]
response = bot.run(messages=messages)
print(list(response)[-1])
```

### ميزات Qwen3 المتقدمة

#### التحكم في وضع التفكير
يدعم Qwen3 التبديل الديناميكي بين وضع التفكير وغير التفكير:

```python
# Enable thinking mode for complex reasoning
messages = [
    {
        'role': 'user',
        'content': '/think Solve this complex math problem: If a train travels 120 km in 1.5 hours, and another train travels 200 km in 2.5 hours, which train is faster and by how much?'
    }
]

# Disable thinking mode for simple queries
messages = [
    {
        'role': 'user', 
        'content': '/no_think What is the capital of France?'
    }
]
```

#### استدعاء وظائف متعددة الخطوات
يتفوق Qwen3 في ربط استدعاءات وظائف متعددة:

```python
# Complex workflow example
messages = [
    {
        'role': 'user',
        'content': '''
        I need to prepare for a business meeting:
        1. Check my calendar for conflicts tomorrow
        2. Get weather forecast for the meeting location (San Francisco)
        3. Find recent news about the client company (TechCorp)
        4. Calculate travel time from my office to their headquarters
        '''
    }
]

# Qwen3 will automatically determine the sequence of function calls needed
```

## التكامل المحلي مع Foundry

يوفر Foundry Local من Microsoft واجهة API متوافقة مع OpenAI لتشغيل النماذج محليًا مع تحسين الخصوصية والأداء.

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

#### Windows
قم بتنزيل المثبت من [صفحة إصدارات Foundry Local](https://github.com/microsoft/Foundry-Local/releases) واتبع تعليمات التثبيت.

#### macOS
```bash
brew tap microsoft/foundrylocal
brew install foundrylocal
```

#### الاستخدام الأساسي

```python
import openai
from foundry_local import FoundryLocalManager

# Initialize with model alias
alias = "phi-3.5-mini"  # Or any supported model
manager = FoundryLocalManager(alias)

# Create OpenAI client pointing to local endpoint
client = openai.OpenAI(
    base_url=manager.endpoint,
    api_key=manager.api_key
)

# Define functions for the model
functions = [
    {
        "name": "calculate_tax",
        "description": "Calculate tax amount based on income and rate",
        "parameters": {
            "type": "object",
            "properties": {
                "income": {
                    "type": "number",
                    "description": "Annual income amount"
                },
                "tax_rate": {
                    "type": "number", 
                    "description": "Tax rate as decimal (e.g., 0.25 for 25%)"
                }
            },
            "required": ["income", "tax_rate"]
        }
    }
]

# Make function calling request
response = client.chat.completions.create(
    model=manager.model_info.id,
    messages=[
        {
            "role": "user",
            "content": "Calculate the tax for someone earning $75,000 with a 22% tax rate"
        }
    ],
    functions=functions,
    function_call="auto"
)

print(response.choices[0].message.content)
```

### ميزات Foundry Local المتقدمة

#### إدارة النماذج
```bash
# List available models
foundry model list

# Download specific model
foundry model download phi-3.5-mini

# Run model interactively
foundry model run phi-3.5-mini

# Remove model from cache
foundry model remove phi-3.5-mini

# Delete all cached models
foundry model remove "*"
```

#### تحسين الأداء
يقوم Foundry Local تلقائيًا باختيار أفضل نموذج متوافق مع جهازك:
- **CUDA GPU**: تنزيل نماذج محسنة لـ GPU
- **Qualcomm NPU**: استخدام نماذج مسرعة بواسطة NPU
- **CPU فقط**: اختيار نماذج محسنة لـ CPU

## أفضل الممارسات وحل المشكلات

### أفضل الممارسات لتعريف الوظائف

#### 1. تسمية واضحة وواصفة
```python
# Good
{
    "name": "get_stock_price",
    "description": "Retrieve current stock price for a given symbol"
}

# Avoid
{
    "name": "get_data", 
    "description": "Gets data"
}
```

#### 2. تعريف شامل للمعلمات
```python
{
    "name": "send_email",
    "description": "Send an email message to specified recipients",
    "parameters": {
        "to": {
            "type": "array",
            "items": {"type": "string"},
            "description": "List of recipient email addresses",
            "required": True
        },
        "subject": {
            "type": "string",
            "description": "Email subject line",
            "required": True
        },
        "body": {
            "type": "string", 
            "description": "Email message content",
            "required": True
        },
        "priority": {
            "type": "string",
            "enum": ["low", "normal", "high"],
            "description": "Email priority level",
            "default": "normal",
            "required": False
        }
    }
}
```

#### 3. التحقق من الإدخال ومعالجة الأخطاء
```python
def execute_function(function_name, parameters):
    try:
        # Validate required parameters
        if function_name == "send_email":
            if not parameters.get("to") or not parameters.get("subject"):
                return {"error": "Missing required parameters: to, subject"}
            
            # Validate email format
            email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
            for email in parameters["to"]:
                if not re.match(email_pattern, email):
                    return {"error": f"Invalid email format: {email}"}
        
        # Execute function logic
        result = perform_actual_function(function_name, parameters)
        return {"success": True, "data": result}
        
    except Exception as e:
        return {"error": str(e)}
```

### المشكلات الشائعة والحلول

#### المشكلة 1: الوظيفة لا يتم استدعاؤها
**الأعراض**: النموذج يستجيب بنص بدلاً من استدعاء الوظيفة

**الحلول**:
1. **تحقق من وصف الوظيفة**: تأكد من أنه يتطابق بوضوح مع نية المستخدم
2. **تحقق من تعريف المعلمات**: تأكد من تعريف جميع المعلمات المطلوبة بشكل صحيح
3. **راجع موجه النظام**: قم بتضمين تعليمات واضحة حول متى يتم استخدام الوظائف
4. **اختبر مع طلبات صريحة**: جرب "يرجى استخدام وظيفة الطقس للحصول على بيانات لندن"

#### المشكلة 2: معلمات غير صحيحة
**الأعراض**: الوظيفة يتم استدعاؤها بمعلمات خاطئة أو مفقودة

**الحلول**:
1. **أضف أمثلة للمعلمات**: قم بتضمين قيم نموذجية في وصف المعلمات
2. **استخدم قيود enum**: قم بتقييد قيم المعلمات إلى خيارات محددة عند الإمكان
3. **قم بتنفيذ قيم احتياطية**: قدم قيم افتراضية منطقية للمعلمات الاختيارية

```python
{
    "name": "book_restaurant",
    "parameters": {
        "cuisine": {
            "type": "string",
            "enum": ["italian", "chinese", "mexican", "american", "french"],
            "description": "Type of cuisine (example: 'italian' for Italian food)"
        },
        "party_size": {
            "type": "integer",
            "minimum": 1,
            "maximum": 20,
            "description": "Number of people (example: 4 for a family of four)"
        }
    }
}
```

#### المشكلة 3: فشل استدعاء الوظائف المتوازية
**الأعراض**: يتم تنفيذ وظيفة واحدة فقط عندما يجب تشغيل وظائف متعددة

**الحلول**:
1. **تحقق من دعم النموذج**: تأكد من أن النموذج يدعم استدعاء الوظائف المتوازية
2. **قم بتحديث موجه النظام**: قم بتضمين "بعض الأدوات" أو "أدوات متعددة" في رسالة النظام
3. **استخدم إصدارات النموذج المناسبة**: يوصى باستخدام Phi-4-mini:3.8b-fp16 لـ Ollama

#### المشكلة 4: مشكلات القالب مع Ollama
**الأعراض**: استدعاء الوظائف لا يعمل مع إعداد Ollama الافتراضي

**الحلول**:
1. **استخدم ModelFile مخصص**: قم بتطبيق القالب المصحح المقدم في هذا الدليل
2. **قم بتحديث Ollama**: تأكد من أنك تستخدم الإصدار 0.5.13 أو أعلى
3. **تحقق من تكميم النموذج**: مستويات التكميم الأعلى (Q8_0، fp16) تعمل بشكل أفضل من الإصدارات ذات التكميم العالي

### تحسين الأداء

#### 1. تصميم وظائف فعال
- **حافظ على تركيز الوظائف**: يجب أن يكون لكل وظيفة غرض واحد وواضح
- **قلل من التبعيات الخارجية**: قلل من استدعاءات API وطلبات الشبكة حيثما أمكن
- **قم بتخزين النتائج مؤقتًا**: احتفظ بالبيانات المطلوبة بشكل متكرر لتحسين أوقات الاستجابة

#### 2. العمليات المجمعة وغير المتزامنة
```python
import asyncio
import aiohttp

async def batch_function_calls(function_calls):
    """Execute multiple function calls concurrently"""
    async with aiohttp.ClientSession() as session:
        tasks = []
        for call in function_calls:
            if call["name"] == "fetch_url":
                task = fetch_url_async(session, call["parameters"]["url"])
                tasks.append(task)
        
        results = await asyncio.gather(*tasks)
        return results

async def fetch_url_async(session, url):
    async with session.get(url) as response:
        return await response.text()
```

#### 3. إدارة الموارد
- **إعادة استخدام الاتصالات**: استخدم اتصالات قاعدة البيانات وواجهات البرمجة بشكل متكرر
- **تحديد معدل الاستخدام**: قم بتطبيق تحديد معدل مناسب لواجهات البرمجة الخارجية
- **معالجة انتهاء المهلة**: قم بتعيين مهلات معقولة لجميع الاستدعاءات الخارجية

## أمثلة متقدمة

### نظام تعاون متعدد الوكلاء

```python
import json
from typing import List, Dict
from qwen_agent.agents import Assistant

class MultiAgentSystem:
    def __init__(self):
        # Research Agent
        self.research_agent = Assistant(
            llm={'model': 'Qwen3-8B', 'model_server': 'http://localhost:8000/v1'},
            function_list=[
                {'mcpServers': {'search': {'command': 'uvx', 'args': ['mcp-server-search']}}},
                {'mcpServers': {'fetch': {'command': 'uvx', 'args': ['mcp-server-fetch']}}}
            ]
        )
        
        # Analysis Agent
        self.analysis_agent = Assistant(
            llm={'model': 'Qwen3-8B', 'model_server': 'http://localhost:8000/v1'},
            function_list=['code_interpreter']
        )
        
        # Communication Agent
        self.comm_agent = Assistant(
            llm={'model': 'Qwen3-8B', 'model_server': 'http://localhost:8000/v1'},
            function_list=[self.create_email_tool(), self.create_slack_tool()]
        )
    
    def create_email_tool(self):
        """Custom email sending tool"""
        class EmailTool:
            name = "send_email"
            description = "Send email to specified recipients"
            parameters = {
                "to": {"type": "string", "description": "Recipient email"},
                "subject": {"type": "string", "description": "Email subject"},
                "body": {"type": "string", "description": "Email content"}
            }
            
            def call(self, params):
                # Implement actual email sending logic
                return f"Email sent successfully to {params['to']}"
        
        return EmailTool()
    
    def create_slack_tool(self):
        """Custom Slack messaging tool"""  
        class SlackTool:
            name = "send_slack"
            description = "Send message to Slack channel"
            parameters = {
                "channel": {"type": "string", "description": "Slack channel"},
                "message": {"type": "string", "description": "Message content"}
            }
            
            def call(self, params):
                # Implement actual Slack API call
                return f"Message sent to {params['channel']}"
        
        return SlackTool()
    
    async def process_complex_request(self, user_request: str):
        """Process complex multi-step requests using multiple agents"""
        
        # Step 1: Research phase
        research_prompt = f"Research the following topic and gather relevant information: {user_request}"
        research_results = []
        for response in self.research_agent.run([{'role': 'user', 'content': research_prompt}]):
            research_results.append(response)
        
        # Step 2: Analysis phase
        analysis_prompt = f"Analyze the following research data and provide insights: {research_results[-1]}"
        analysis_results = []
        for response in self.analysis_agent.run([{'role': 'user', 'content': analysis_prompt}]):
            analysis_results.append(response)
        
        # Step 3: Communication phase
        comm_prompt = f"Create a summary report and send it via email: {analysis_results[-1]}"
        comm_results = []
        for response in self.comm_agent.run([{'role': 'user', 'content': comm_prompt}]):
            comm_results.append(response)
        
        return {
            'research': research_results[-1],
            'analysis': analysis_results[-1], 
            'communication': comm_results[-1]
        }

# Usage example
async def main():
    system = MultiAgentSystem()
    
    request = """
    Analyze the impact of remote work on productivity in tech companies. 
    Research recent studies, analyze the data, and send a summary to our team.
    """
    
    results = await system.process_complex_request(request)
    print("Multi-agent processing complete:", results)

# Run the example
# asyncio.run(main())
```

### نظام اختيار الأدوات الديناميكي

```python
class DynamicToolSelector:
    def __init__(self):
        self.available_tools = {
            'weather': {
                'description': 'Get weather information',
                'domains': ['weather', 'temperature', 'forecast', 'climate'],
                'function': self.get_weather
            },
            'calculator': {
                'description': 'Perform mathematical calculations',
                'domains': ['math', 'calculate', 'compute', 'arithmetic'],
                'function': self.calculate
            },
            'web_search': {
                'description': 'Search the internet for information',
                'domains': ['search', 'find', 'lookup', 'research'],
                'function': self.web_search
            },
            'file_manager': {
                'description': 'Manage files and directories',
                'domains': ['file', 'directory', 'save', 'load', 'delete'],
                'function': self.manage_files
            }
        }
    
    def analyze_intent(self, user_input: str) -> List[str]:
        """Analyze user input to determine which tools might be needed"""
        user_words = user_input.lower().split()
        relevant_tools = []
        
        for tool_name, tool_info in self.available_tools.items():
            for domain in tool_info['domains']:
                if domain in user_words:
                    relevant_tools.append(tool_name)
                    break
        
        return relevant_tools
    
    def get_tool_definitions(self, tool_names: List[str]) -> List[Dict]:
        """Generate function definitions for selected tools"""
        definitions = []
        
        for tool_name in tool_names:
            if tool_name == 'weather':
                definitions.append({
                    'name': 'get_weather',
                    'description': 'Get current weather information',
                    'parameters': {
                        'location': {'type': 'string', 'description': 'City or location name'},
                        'units': {'type': 'string', 'enum': ['celsius', 'fahrenheit'], 'default': 'celsius'}
                    }
                })
            elif tool_name == 'calculator':
                definitions.append({
                    'name': 'calculate',
                    'description': 'Perform mathematical calculations',
                    'parameters': {
                        'expression': {'type': 'string', 'description': 'Mathematical expression to evaluate'},
                        'precision': {'type': 'integer', 'default': 2, 'description': 'Decimal places for result'}
                    }
                })
            # Add more tool definitions as needed
        
        return definitions
    
    def get_weather(self, location: str, units: str = 'celsius') -> Dict:
        """Mock weather function"""
        return {
            'location': location,
            'temperature': '22°C' if units == 'celsius' else '72°F',
            'condition': 'Sunny',
            'humidity': '60%'
        }
    
    def calculate(self, expression: str, precision: int = 2) -> Dict:
        """Safe mathematical calculation"""
        try:
            # Simple evaluation for demo - in production, use a proper math parser
            import math
            allowed_names = {
                k: v for k, v in math.__dict__.items() if not k.startswith("__")
            }
            allowed_names.update({"abs": abs, "round": round})
            
            result = eval(expression, {"__builtins__": {}}, allowed_names)
            return {
                'expression': expression,
                'result': round(float(result), precision),
                'success': True
            }
        except Exception as e:
            return {
                'expression': expression,
                'error': str(e),
                'success': False
            }
    
    def web_search(self, query: str, max_results: int = 5) -> Dict:
        """Mock web search function"""
        return {
            'query': query,
            'results': [
                {'title': f'Result {i+1} for {query}', 'url': f'https://example{i+1}.com'}
                for i in range(max_results)
            ]
        }
    
    def manage_files(self, action: str, file_path: str, content: str = None) -> Dict:
        """Mock file management function"""
        return {
            'action': action,
            'file_path': file_path,
            'success': True,
            'message': f'Successfully {action}ed file: {file_path}'
        }

# Usage example
def smart_assistant_with_dynamic_tools():
    selector = DynamicToolSelector()
    
    user_requests = [
        "What's the weather like in New York and calculate 15% tip on $50?",
        "Search for recent AI developments and save the results to a file",
        "Calculate the area of a circle with radius 10 and check weather in Tokyo"
    ]
    
    for request in user_requests:
        print(f"\nUser Request: {request}")
        
        # Analyze which tools might be needed
        relevant_tools = selector.analyze_intent(request)
        print(f"Relevant Tools: {relevant_tools}")
        
        # Get function definitions for the LLM
        tool_definitions = selector.get_tool_definitions(relevant_tools)
        print(f"Tool Definitions: {len(tool_definitions)} functions available")
        
        # In a real implementation, you would pass these to your LLM
        # The LLM would then decide which functions to call and with what parameters

### Enterprise Integration Example

```python
import asyncio
import json
from typing import Dict, List, Any
from dataclasses import dataclass
from datetime import datetime

@dataclass
class FunctionResult:
    """تنسيق قياسي لنتائج جميع استدعاءات الوظائف"""
    success: bool
    data: Any = None
    error: str = None
    execution_time: float = 0.0
    timestamp: datetime = None

class EnterpriseAIAgent:
    """وكيل ذكاء اصطناعي جاهز للإنتاج مع قدرات شاملة لاستدعاء الوظائف"""
    
    def __init__(self, config: Dict):
        self.config = config
        self.functions = {}
        self.audit_log = []
        self.rate_limiters = {}
        
        # تهيئة الوظائف الأساسية للأعمال
        self._register_core_functions()
    
    def _register_core_functions(self):
        """تسجيل جميع الوظائف المتاحة للأعمال"""
        
        # وظائف إدارة العملاء
        self.register_function(
            name="get_customer_info",
            description="استرجاع معلومات العميل من نظام إدارة العملاء",
            parameters={
                "customer_id": {"type": "string", "required": True},
                "include_history": {"type": "boolean", "default": False}
            },
            handler=self._get_customer_info,
            rate_limit=100  # استدعاءات في الدقيقة
        )
        
        # وظائف المبيعات
        self.register_function(
            name="create_sales_opportunity",
            description="إنشاء فرصة مبيعات جديدة",
            parameters={
                "customer_id": {"type": "string", "required": True},
                "product_id": {"type": "string", "required": True},
                "estimated_value": {"type": "number", "required": True},
                "expected_close_date": {"type": "string", "required": True}
            },
            handler=self._create_sales_opportunity,
            rate_limit=50
        )
        
        # وظائف التحليل
        self.register_function(
            name="generate_sales_report",
            description="إنشاء تقرير أداء المبيعات",
            parameters={
                "period": {"type": "string", "enum": ["daily", "weekly", "monthly", "quarterly"]},
                "region": {"type": "string", "required": False},
                "product_category": {"type": "string", "required": False}
            },
            handler=self._generate_sales_report,
            rate_limit=10
        )
        
        # وظائف الإشعارات
        self.register_function(
            name="send_notification",
            description="إرسال إشعار إلى أعضاء الفريق",
            parameters={
                "recipients": {"type": "array", "items": {"type": "string"}},
                "message": {"type": "string", "required": True},
                "priority": {"type": "string", "enum": ["low", "medium", "high"], "default": "medium"},
                "channel": {"type": "string", "enum": ["email", "slack", "teams"], "default": "email"}
            },
            handler=self._send_notification,
            rate_limit=200
        )
    
    def register_function(self, name: str, description: str, parameters: Dict, 
                         handler: callable, rate_limit: int = 60):
        """تسجيل وظيفة جديدة مع الوكيل"""
        self.functions[name] = {
            'description': description,
            'parameters': parameters,
            'handler': handler,
            'rate_limit': rate_limit,
            'call_count': 0,
            'last_reset': datetime.now()
        }
    
    async def execute_function(self, function_name: str, parameters: Dict) -
عذرًا، لم تقدم محتوى Markdown للترجمة. يرجى تقديم النص الذي ترغب في ترجمته وسأقوم بمساعدتك.
"""تنفيذ وظيفة مع معالجة شاملة للأخطاء وتسجيل الأحداث"""
وقت البدء = datetime.now()

حاول:
    # التحقق من وجود الوظيفة
    إذا لم تكن function_name موجودة في self.functions:
        return FunctionResult(
            success=False,
            error=f"الوظيفة '{function_name}' غير موجودة",
            timestamp=وقت البدء
        )
    
    # التحقق من حدود المعدل
    إذا لم تكن self._check_rate_limit(function_name):
        return FunctionResult(
            success=False,
            error=f"تم تجاوز حدود المعدل للوظيفة '{function_name}'",
            timestamp=وقت البدء
        )
    
    # التحقق من صحة المعلمات
    validation_result = self._validate_parameters(function_name, parameters)
    إذا لم يكن validation_result.success:
        return validation_result
    
    # تنفيذ الوظيفة
    func_info = self.functions[function_name]
    handler = func_info['handler']
    
    إذا كان asyncio.iscoroutinefunction(handler):
        result_data = await handler(**parameters)
    آخر:
        result_data = handler(**parameters)
    
    execution_time = (datetime.now() - وقت البدء).total_seconds()
    
    result = FunctionResult(
        success=True,
        data=result_data,
        execution_time=execution_time,
        timestamp=وقت البدء
    )
    
    # تسجيل التنفيذ الناجح
    self._log_function_call(function_name, parameters, result)
    
    return result
    
ما عدا Exception as e:
    execution_time = (datetime.now() - وقت البدء).total_seconds()
    result = FunctionResult(
        success=False,
        error=str(e),
        execution_time=execution_time,
        timestamp=وقت البدء
    )
    
    # تسجيل التنفيذ الفاشل
    self._log_function_call(function_name, parameters, result)
    
    return result

def _check_rate_limit(self, function_name: str) -> bool:
    """التحقق مما إذا كانت المكالمة ضمن حدود المعدل"""
    func_info = self.functions[function_name]
    الآن = datetime.now()
    
    # إعادة تعيين العداد إذا مرت دقيقة
    إذا كان (الآن - func_info['last_reset']).seconds >= 60:
        func_info['call_count'] = 0
        func_info['last_reset'] = الآن
    
    # التحقق مما إذا كان ضمن الحد
    إذا كان func_info['call_count'] >= func_info['rate_limit']:
        return False
    
    func_info['call_count'] += 1
    return True

def _validate_parameters(self, function_name: str, parameters: Dict) -> FunctionResult:
    """التحقق من صحة معلمات الوظيفة"""
    func_params = self.functions[function_name]['parameters']
    
    # التحقق من المعلمات المطلوبة
    لكل param_name, param_info في func_params.items():
        إذا كان param_info.get('required', False) و param_name ليس في parameters:
            return FunctionResult(
                success=False,
                error=f"المعلمة المطلوبة مفقودة: {param_name}"
            )
    
    # التحقق من أنواع المعلمات والقيود
    لكل param_name, value في parameters.items():
        إذا كان param_name في func_params:
            param_info = func_params[param_name]
            
            # التحقق من النوع
            expected_type = param_info.get('type')
            إذا كان expected_type == 'string' و ليس isinstance(value, str):
                return FunctionResult(
                    success=False,
                    error=f"المعلمة '{param_name}' يجب أن تكون نصية"
                )
            elif expected_type == 'number' و ليس isinstance(value, (int, float)):
                return FunctionResult(
                    success=False,
                    error=f"المعلمة '{param_name}' يجب أن تكون رقمية"
                )
            elif expected_type == 'boolean' و ليس isinstance(value, bool):
                return FunctionResult(
                    success=False,
                    error=f"المعلمة '{param_name}' يجب أن تكون منطقية"
                )
            
            # التحقق من القيم المسموح بها
            إذا كانت 'enum' في param_info و value ليس في param_info['enum']:
                return FunctionResult(
                    success=False,
                    error=f"المعلمة '{param_name}' يجب أن تكون واحدة من: {param_info['enum']}"
                )
    
    return FunctionResult(success=True)

def _log_function_call(self, function_name: str, parameters: Dict, result: FunctionResult):
    """تسجيل مكالمة الوظيفة لأغراض التدقيق"""
    log_entry = {
        'timestamp': result.timestamp.isoformat(),
        'function_name': function_name,
        'parameters': parameters,
        'success': result.success,
        'execution_time': result.execution_time,
        'error': result.error إذا لم يكن result.success آخر None
    }
    
    self.audit_log.append(log_entry)
    
    # اختيارياً الكتابة إلى نظام تسجيل خارجي
    إذا كانت self.config.get('enable_external_logging', False):
        self._write_to_external_log(log_entry)

def _write_to_external_log(self, log_entry: Dict):
    """كتابة سجل إلى نظام تسجيل خارجي"""
    # التنفيذ يعتمد على بنية التسجيل الخاصة بك
    # على سبيل المثال، الإرسال إلى ELK stack، CloudWatch، إلخ.
    pass

# تنفيذات الوظائف التجارية
async def _get_customer_info(self, customer_id: str, include_history: bool = False) -> Dict:
    """استرجاع معلومات العميل من نظام إدارة علاقات العملاء"""
    # محاكاة استدعاء قاعدة البيانات/واجهة برمجة التطبيقات
    await asyncio.sleep(0.1)  # محاكاة تأخير الشبكة
    
    customer_data = {
        'customer_id': customer_id,
        'name': 'John Doe',
        'email': 'john.doe@example.com',
        'phone': '+1-555-0123',
        'status': 'active',
        'tier': 'premium'
    }
    
    إذا كان include_history:
        customer_data['purchase_history'] = [
            {'date': '2024-01-15', 'product': 'Product A', 'amount': 1500},
            {'date': '2024-03-22', 'product': 'Product B', 'amount': 2300}
        ]
    
    return customer_data

async def _create_sales_opportunity(self, customer_id: str, product_id: str, 
                                  estimated_value: float, expected_close_date: str) -> Dict:
    """إنشاء فرصة مبيعات جديدة"""
    # محاكاة استدعاء واجهة برمجة التطبيقات لنظام إدارة علاقات العملاء
    await asyncio.sleep(0.2)
    
    opportunity_id = f"OPP-{datetime.now().strftime('%Y%m%d%H%M%S')}"
    
    return {
        'opportunity_id': opportunity_id,
        'customer_id': customer_id,
        'product_id': product_id,
        'estimated_value': estimated_value,
        'expected_close_date': expected_close_date,
        'status': 'open',
        'created_date': datetime.now().isoformat()
    }

async def _generate_sales_report(self, period: str, region: str = None, 
                               product_category: str = None) -> Dict:
    """إنشاء تقرير مبيعات شامل"""
    # محاكاة تجميع البيانات
    await asyncio.sleep(0.5)
    
    return {
        'report_id': f"RPT-{datetime.now().strftime('%Y%m%d%H%M%S')}",
        'period': period,
        'region': region,
        'product_category': product_category,
        'total_sales': 125000.00,
        'total_opportunities': 45,
        'conversion_rate': 0.67,
        'top_products': [
            {'product_id': 'PROD-001', 'sales': 45000},
            {'product_id': 'PROD-002', 'sales': 32000}
        ],
        'generated_at': datetime.now().isoformat()
    }

async def _send_notification(self, recipients: List[str], message: str, 
                           priority: str = 'medium', channel: str = 'email') -> Dict:
    """إرسال إشعار عبر القناة المحددة"""
    # محاكاة استدعاء خدمة الإشعارات
    await asyncio.sleep(0.1)
    
    notification_id = f"NOTIF-{datetime.now().strftime('%Y%m%d%H%M%S')}"
    
    return {
        'notification_id': notification_id,
        'recipients': recipients,
        'channel': channel,
        'priority': priority,
        'status': 'sent',
        'sent_at': datetime.now().isoformat()
    }

def get_function_definitions(self) -> List[Dict]:
    """الحصول على تعريفات الوظائف المتوافقة مع OpenAI لجميع الوظائف المسجلة"""
    definitions = []
    
    لكل func_name, func_info في self.functions.items():
        definition = {
            'name': func_name,
            'description': func_info['description'],
            'parameters': {
                'type': 'object',
                'properties': {},
                'required': []
            }
        }
        
        لكل param_name, param_info في func_info['parameters'].items():
            definition['parameters']['properties'][param_name] = {
                'type': param_info['type'],
                'description': param_info.get('description', '')
            }
            
            إذا كانت 'enum' في param_info:
                definition['parameters']['properties'][param_name]['enum'] = param_info['enum']
            
            إذا كانت 'default' في param_info:
                definition['parameters']['properties'][param_name]['default'] = param_info['default']
            
            إذا كان param_info.get('required', False):
                definition['parameters']['required'].append(param_name)
        
        definitions.append(definition)
    
    return definitions

# مثال استخدام للتكامل المؤسسي
async def enterprise_demo():
    """عرض قدرات وكيل الذكاء الاصطناعي المؤسسي"""
    
    config = {
        'enable_external_logging': True,
        'max_concurrent_functions': 10,
        'default_timeout': 30
    }
    
    agent = EnterpriseAIAgent(config)
    
    # المثال الأول: معالجة استفسارات العملاء
    print("=== معالجة استفسارات العملاء ===")
    
    # الحصول على معلومات العميل
    result = await agent.execute_function(
        'get_customer_info',
        {'customer_id': 'CUST-12345', 'include_history': True}
    )
    
    إذا كان result.success:
        print(f"تم استرجاع معلومات العميل: {result.data['name']}")
        print(f"وقت التنفيذ: {result.execution_time:.3f}s")
    
    # المثال الثاني: إنشاء فرصة مبيعات
    print("\n=== إنشاء فرصة مبيعات ===")
    
    result = await agent.execute_function(
        'create_sales_opportunity',
        {
            'customer_id': 'CUST-12345',
            'product_id': 'PROD-001',
            'estimated_value': 15000.0,
            'expected_close_date': '2025-09-30'
        }
    )
    
    إذا كان result.success:
        print(f"تم إنشاء الفرصة: {result.data['opportunity_id']}")
    
    # المثال الثالث: العمليات الدفعية
    print("\n=== العمليات الدفعية ===")
    
    tasks = [
        agent.execute_function('generate_sales_report', {'period': 'monthly'}),
        agent.execute_function('send_notification', {
            'recipients': ['manager@company.com'],
            'message': 'تم إنشاء فرصة جديدة',
            'priority': 'high',
            'channel': 'email'
        })
    ]
    
    results = await asyncio.gather(*tasks)
    
    لكل i, result في enumerate(results):
        إذا كان result.success:
            print(f"تم إكمال المهمة {i+1} بنجاح")
        آخر:
            print(f"فشلت المهمة {i+1}: {result.error}")
    
    # عرض سجل التدقيق
    print(f"\n=== سجل التدقيق ({len(agent.audit_log)} إدخالات) ===")
    لكل entry في agent.audit_log[-3:]:  # عرض آخر 3 إدخالات
        print(f"{entry['timestamp']}: {entry['function_name']} - {'SUCCESS' إذا كان entry['success'] آخر 'FAILED'}")

# تشغيل العرض المؤسسي
# asyncio.run(enterprise_demo())
- **نماذج Phi-4**: [مجموعة Hugging Face](https://huggingface.co/collections/microsoft/phi-4-677e9380e514feb5577a40e4)
- **وثائق Qwen3**: [الوثائق الرسمية لـ Qwen](https://qwen.readthedocs.io/)
- **Ollama**: [الموقع الرسمي](https://ollama.com/)
- **Foundry Local**: [مستودع GitHub](https://github.com/microsoft/Foundry-Local)
- **أفضل الممارسات لاستدعاء الوظائف**: [دليل Hugging Face](https://huggingface.co/docs/hugs/en/guides/function-calling)

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

## ➡️ ما الخطوة التالية

- [03: دمج بروتوكول سياق النموذج (MCP)](./03.IntroduceMCP.md)

---

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