# الجلسة 5: بناء وكلاء مدعومين بالذكاء الاصطناعي بسرعة باستخدام Foundry Local

ملاحظة: تتطور قدرات الوكلاء في Foundry Local—تأكد من دعمها في أحدث ملاحظات الإصدار قبل تنفيذ الأنماط المتقدمة.

## نظرة عامة

استخدم Foundry Local لتطوير تطبيقات الوكلاء بسرعة: مطالبات النظام، التأسيس، وأنماط التنسيق. عندما يكون دعم الوكلاء متاحًا، يمكنك الاعتماد على استدعاء الوظائف المتوافق مع OpenAI أو استخدام Azure AI Agents على السحابة في التصاميم الهجينة.

> **🔄 محدث لـ SDK الحديث**: تم تحديث هذه الوحدة لتتوافق مع أنماط مستودع Microsoft Foundry-Local الأحدث وتطابق التنفيذ الشامل في `samples/05/`. الأمثلة الآن تستخدم `foundry-local-sdk` الحديث وعميل `OpenAI` بدلاً من الطلبات اليدوية.

**🏗️ أبرز معالم الهندسة المعمارية:**
- **وكلاء متخصصون**: وكلاء الاسترجاع، التفكير، والتنفيذ بقدرات مميزة
- **نمط التنسيق**: تنسيق سير العمل متعدد الوكلاء مع حلقات التغذية الراجعة
- **تكامل SDK الحديث**: يستخدم `FoundryLocalManager` وعميل OpenAI
- **جاهز للإنتاج**: يشمل معالجة الأخطاء، مراقبة الأداء، وفحوصات الصحة
- **أمثلة شاملة**: دفتر Jupyter تفاعلي مع ميزات متقدمة

**📁 التنفيذ المحلي:**
- `samples/05/multi_agent_orchestration.ipynb` - أمثلة تفاعلية ومعايير
- `samples/05/agents/specialists.py` - تنفيذ الوكلاء
- `samples/05/agents/coordinator.py` - منطق التنسيق

المراجع:
- وثائق Foundry Local: https://learn.microsoft.com/en-us/azure/ai-foundry/foundry-local/
- وكلاء Azure AI Foundry: https://learn.microsoft.com/en-us/azure/ai-services/agents/overview
- مثال استدعاء الوظائف (عينات Foundry Local): https://github.com/microsoft/Foundry-Local/tree/main/samples/python/functioncalling

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

## الجزء 1: مطالبات النظام والتأسيس

- تحديد أدوار صارمة، قيود، ومخططات المخرجات
- تأسيس الردود باستخدام بيانات محلية أو مؤسسية
- فرض مخرجات JSON لأتمتة العمليات اللاحقة

## الجزء 2: استدعاء الوظائف (نهج SDK الحديث)

```python
# tools.py
import json
from typing import List, Dict, Any

def get_weather(city: str) -> str:
    return f"Weather in {city}: Sunny, 25C"

# Modern tools format for OpenAI API
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a city",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "City name"}
                },
                "required": ["city"]
            }
        }
    }
]
```

```python
# agent.py
from foundry_local import FoundryLocalManager
from openai import OpenAI
import json
from tools import TOOLS, get_weather

# Initialize Foundry Local Manager
alias = "phi-4-mini"
manager = FoundryLocalManager(alias)

# Create OpenAI client using Foundry Local endpoint
client = OpenAI(
    base_url=manager.endpoint,
    api_key=manager.api_key
)

SYSTEM_PROMPT = "You are a helpful assistant. Use tools when needed."

def process_function_call(messages: List[Dict], tools: List[Dict]) -> str:
    """Process function calling with modern OpenAI API."""
    try:
        response = client.chat.completions.create(
            model=manager.get_model_info(alias).id,
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )
        
        message = response.choices[0].message
        
        if message.tool_calls:
            # Handle function calls
            messages.append(message)
            
            for tool_call in message.tool_calls:
                if tool_call.function.name == "get_weather":
                    args = json.loads(tool_call.function.arguments)
                    result = get_weather(args["city"])
                    
                    # Add function result to messages
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": result
                    })
            
            # Get final response
            final_response = client.chat.completions.create(
                model=manager.get_model_info(alias).id,
                messages=messages
            )
            return final_response.choices[0].message.content
        else:
            return message.content
            
    except Exception as e:
        return f"Error: {str(e)}"

# Example usage
messages = [
    {"role": "system", "content": SYSTEM_PROMPT},
    {"role": "user", "content": "What's the weather in Paris?"}
]

result = process_function_call(messages, TOOLS)
print(result)
```

تشغيل:
```powershell
# Ensure Foundry Local is running with a model
foundry model run phi-4-mini
python agent.py
```


## الجزء 3: تنسيق متعدد الوكلاء (النمط)

صمم منسقًا يوجه المهام إلى وكلاء متخصصين (الاسترجاع، التفكير، التنفيذ) باستخدام نقطة النهاية المتوافقة مع OpenAI في Foundry Local.

الخطوة 1) تعريف الوكلاء المتخصصين باستخدام SDK الحديث (راجع `samples/05/agents/specialists.py`)
```python
# agents/specialists.py
from foundry_local import FoundryLocalManager
from openai import OpenAI
from typing import List, Dict, Any

class FoundryClient:
    """Shared client for all specialist agents."""
    
    def __init__(self, model_alias: str = "phi-4-mini"):
        self.client = None
        self.model_name = None
        self.model_alias = model_alias
        self._initialize_client()
    
    def _initialize_client(self):
        """Initialize OpenAI client with Foundry Local."""
        try:
            manager = FoundryLocalManager(self.model_alias)
            model_info = manager.get_model_info(self.model_alias)
            
            self.client = OpenAI(
                base_url=manager.endpoint,
                api_key=manager.api_key
            )
            self.model_name = model_info.id
            print(f"✅ Foundry Local initialized with model: {self.model_name}")
        except Exception as e:
            print(f"❌ Error initializing Foundry Local: {e}")
            raise
    
    def chat(self, messages: List[Dict[str, str]], max_tokens: int = 300, temperature: float = 0.4) -> str:
        """Send chat completion request to the model."""
        try:
            response = self.client.chat.completions.create(
                model=self.model_name,
                messages=messages,
                max_tokens=max_tokens,
                temperature=temperature
            )
            return response.choices[0].message.content
        except Exception as e:
            return f"Error generating response: {str(e)}"

# Global client instance
_client = FoundryClient()

class RetrievalAgent:
    """Agent specialized in retrieving relevant information from knowledge sources."""
    
    SYSTEM = """You are a specialized retrieval agent. Your job is to extract and retrieve 
    the most relevant information from knowledge sources based on a given query. Focus on key facts, 
    data points, and contextual information that would be useful for decision-making."""
    
    def run(self, query: str) -> str:
        """Retrieve relevant information based on the query."""
        messages = [
            {"role": "system", "content": self.SYSTEM},
            {"role": "user", "content": f"Query: {query}\n\nRetrieve the most relevant key facts, data points, and contextual information that would help answer this query or support decision-making around it."}
        ]
        return _client.chat(messages)

class ReasoningAgent:
    """Agent specialized in step-by-step analysis and reasoning."""
    
    SYSTEM = """You are a specialized reasoning agent. Your job is to analyze inputs 
    step-by-step and produce structured, logical conclusions. Break down complex problems 
    into manageable parts and provide clear reasoning for your conclusions."""
    
    def run(self, context: str, question: str) -> str:
        """Analyze context and question to produce structured conclusions."""
        messages = [
            {"role": "system", "content": self.SYSTEM},
            {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}\n\nAnalyze this step-by-step and provide a structured, logical conclusion with clear reasoning."}
        ]
        return _client.chat(messages, max_tokens=400)

class ExecutionAgent:
    """Agent specialized in creating actionable execution plans."""
    
    SYSTEM = """You are a specialized execution agent. Your job is to transform decisions 
    and conclusions into concrete, actionable steps. Always format your response as valid JSON 
    with an array of action items. Each action should be specific, measurable, and achievable."""
    
    def run(self, decision: str) -> str:
        """Transform decision into actionable steps in JSON format."""
        messages = [
            {"role": "system", "content": self.SYSTEM},
            {"role": "user", "content": f"Decision/Conclusion:\n{decision}\n\nCreate 3-5 specific, actionable steps to implement this decision. Format as JSON with this structure:\n{{\"actions\": [{{\"step\": 1, \"description\": \"...\", \"priority\": \"high/medium/low\", \"timeline\": \"...\"}}]}}"}
        ]
        return _client.chat(messages, max_tokens=400, temperature=0.3)
```

الخطوة 2) بناء المنسق بميزات متقدمة
```python
# agents/coordinator.py
from .specialists import RetrievalAgent, ReasoningAgent, ExecutionAgent
from typing import Dict, Any
import time
import json

class Coordinator:
    """Multi-agent coordinator that orchestrates specialist agents to handle complex tasks."""
    
    def __init__(self):
        """Initialize the coordinator with specialist agents."""
        self.retrieval = RetrievalAgent()
        self.reasoning = ReasoningAgent()
        self.execution = ExecutionAgent()
    
    def handle(self, user_goal: str) -> Dict[str, Any]:
        """
        Orchestrate multiple agents to handle a complex user goal.
        
        Args:
            user_goal: The user's high-level goal or request
            
        Returns:
            Dictionary containing the goal, context, decision, and actions
        """
        print(f"🎯 **Coordinator:** Processing goal: {user_goal}")
        print("=" * 60)
        
        start_time = time.time()
        
        # Step 1: Retrieve relevant context
        print("📚 **Step 1:** Retrieving context...")
        context = self.retrieval.run(user_goal)
        print(f"   ✅ Context retrieved ({len(context)} chars)")
        
        # Step 2: Analyze and reason about the context
        print("🧠 **Step 2:** Analyzing and reasoning...")
        decision = self.reasoning.run(context, user_goal)
        print(f"   ✅ Analysis completed ({len(decision)} chars)")
        
        # Step 3: Create actionable execution plan
        print("⚡ **Step 3:** Creating execution plan...")
        actions = self.execution.run(decision)
        print(f"   ✅ Execution plan created ({len(actions)} chars)")
        
        end_time = time.time()
        processing_time = end_time - start_time
        
        result = {
            "goal": user_goal,
            "context": context,
            "decision": decision,
            "actions": actions,
            "agent_flow": ["retrieval", "reasoning", "execution"],
            "processing_time": processing_time,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
        }
        
        print(f"✅ **Coordination Complete** (⏱️ {processing_time:.2f}s)")
        return result
    
    def handle_with_feedback(self, user_goal: str, feedback_rounds: int = 1) -> Dict[str, Any]:
        """
        Handle a goal with multiple feedback rounds for refinement.
        
        Args:
            user_goal: The user's high-level goal or request
            feedback_rounds: Number of feedback rounds to perform
            
        Returns:
            Dictionary containing the refined result
        """
        result = self.handle(user_goal)
        
        for round_num in range(feedback_rounds):
            print(f"\n🔄 **Feedback Round {round_num + 1}:**")
            print("-" * 40)
            
            # Use reasoning agent to refine the execution plan
            refinement_prompt = f"""
            Original Goal: {user_goal}
            Current Decision: {result['decision']}
            Current Actions: {result['actions']}
            
            Review the above and suggest improvements or refinements to make the execution plan more effective.
            """
            
            refined_decision = self.reasoning.run(result['context'], refinement_prompt)
            refined_actions = self.execution.run(refined_decision)
            
            result['decision'] = refined_decision
            result['actions'] = refined_actions
            result['refinement_rounds'] = round_num + 1
            
            print(f"   ✅ Round {round_num + 1} refinement completed")
        
        return result

def main():
    """Main function demonstrating the multi-agent coordinator."""
    print("🤖 **Multi-Agent Coordinator Demo**")
    print("=" * 50)
    
    # Create coordinator
    coord = Coordinator()
    
    # Example goals
    example_goals = [
        "Create a plan to onboard 5 new customers this month",
        "Develop a strategy to improve team productivity by 20%",
        "Design a customer feedback collection system"
    ]
    
    # Process example with feedback
    goal = example_goals[0]
    print(f"🎯 **Processing Goal:** {goal}")
    print("-" * 50)
    
    try:
        # Basic processing
        result = coord.handle(goal)
        
        # With feedback refinement
        refined_result = coord.handle_with_feedback(goal, feedback_rounds=1)
        
        print("\n📊 **Final Result:**")
        print("=" * 50)
        print(f"**Goal:** {refined_result['goal']}")
        print(f"**Processing Time:** {refined_result['processing_time']:.2f}s")
        
        # Try to parse actions as JSON
        try:
            actions_json = json.loads(refined_result['actions'])
            print(f"\n**Formatted Actions:**")
            print(json.dumps(actions_json, indent=2))
        except (json.JSONDecodeError, TypeError):
            print(f"\n**Actions:** {refined_result['actions']}")
            
    except Exception as e:
        print(f"❌ **Error:** {e}")
        print("\nPlease ensure Foundry Local is running with a model loaded.")

if __name__ == "__main__":
    main()
```

الخطوة 3) التحقق من Foundry Local وتشغيل العينات
```powershell
REM Confirm the local endpoint and model are available
foundry model list
foundry model run phi-4-mini
curl http://localhost:8000/v1/models

REM Run the coordinator from Module08 directory
cd Module08
python -m samples.05.agents.coordinator

REM Or explore the comprehensive Jupyter notebook
jupyter notebook samples/05/multi_agent_orchestration.ipynb
```


> **📚 مراجع العينات المحلية:**
> - **التنفيذ الرئيسي**: `samples/05/agents/specialists.py` و `samples/05/agents/coordinator.py`
> - **أمثلة شاملة**: `samples/05/multi_agent_orchestration.ipynb`
> - **تعليمات الإعداد**: `samples/05/README.md`
> 
> **🔗 عينات Foundry Local ذات الصلة:**
> - [مثال استدعاء الوظائف](https://github.com/microsoft/Foundry-Local/tree/main/samples/python/functioncalling)
> - [مرحبًا Foundry Local](https://github.com/microsoft/Foundry-Local/tree/main/samples/python/hello-foundry-local)

الإرشادات:
- تنفيذ إعادة المحاولات والمهلات بين الوكلاء
- إضافة مخزن صغير في الذاكرة (dict) لحالة المحادثة/الخيط
- تقديم تحديد معدل عند ربط عدة مكالمات

## الجزء 4: الرصد والسلامة

تتبع المطالبات، الردود، والأخطاء محليًا، مع فرض نظافة البيانات في مجموعة الوكلاء الخاصة بك.

الخطوة 1) تسجيل الطلبات بشكل خفيف (اختياري)

ملاحظة: المساعد التالي غير مضمن افتراضيًا. قم بإنشاء `infra/obs.py` إذا كنت تريد تسجيل JSON محلي للتجارب.
```python
# infra/obs.py
import time, json, os
from datetime import datetime

LOG_DIR = os.getenv("FOUNDRY_AGENT_LOG_DIR", "./agent_logs")
os.makedirs(LOG_DIR, exist_ok=True)

def log_event(kind: str, payload: dict):
    ts = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
    path = os.path.join(LOG_DIR, f"{ts}_{kind}.json")
    with open(path, "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)
```

دمج التسجيل في الوكلاء (اختياري):
```python
# in agents/specialists.py after receiving content
from infra.obs import log_event
# ... inside chat(...)
resp = r.json()
log_event("chat_request", {"endpoint": f"{BASE_URL}/v1/chat/completions"})
log_event("chat_response", resp)
return resp["choices"][0]["message"]["content"]
```

الخطوة 2) التحقق من التوافر والصحة الأساسية عبر CLI
```powershell
REM Ensure Foundry Local is running a model
foundry model list
foundry model run phi-4-mini

REM Validate the OpenAI-compatible endpoint
curl http://localhost:8000/v1/models
```

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

مثال مساعد التنقيح:
```python
# infra/redact.py
import re
EMAIL_RE = re.compile(r"[\w\.-]+@[\w\.-]+")
PHONE_RE = re.compile(r"\+?\d[\d\s\-]{7,}\d")

def sanitize(text: str) -> str:
    text = EMAIL_RE.sub("[REDACTED_EMAIL]", text)
    text = PHONE_RE.sub("[REDACTED_PHONE]", text)
    return text
```

الاستخدام في الوكلاء:
```python
from infra.redact import sanitize
# user_goal = sanitize(user_goal)
# context = sanitize(context)
```

الخطوة 4) قواطع الدائرة ومعالجة الأخطاء
- قم بتغليف كل مكالمة وكيل بـ try/except و backoff أسي
- قم بقطع الدائرة في حالة الفشل المتكرر

```python
import time

def with_retry(func, retries=3, base_delay=0.5):
    for i in range(retries):
        try:
            return func()
        except Exception as e:
            if i == retries - 1:
                raise
            time.sleep(base_delay * (2 ** i))
```

الخطوة 5) سجل تدقيق محلي وتصدير
- تخزين سجلات JSON تحت `./agent_logs`
- ضغط وتدوير السجلات بشكل دوري
- تصدير ملخصات للمراجعات (العدادات، متوسط زمن الاستجابة، معدلات الأخطاء)

الخطوة 6) التحقق المتقاطع مع وثائق Microsoft Learn
- يقدم Foundry Local واجهة برمجية متوافقة مع OpenAI (تم التحقق منها باستخدام `curl /v1/models`)
- استخدم `foundry model run <name>` لتأكيد توفر النموذج
- اتبع الإرشادات الرسمية لتكامل العميل وتطبيقات العينات (Open WebUI/how-tos)

المراجع:
- **وثائق Foundry Local**: https://learn.microsoft.com/en-us/azure/ai-foundry/foundry-local/
- **وكلاء Azure AI**: https://learn.microsoft.com/en-us/azure/ai-services/agents/overview
- **عينات محلية**:
  - تنسيق متعدد الوكلاء: `Module08/samples/05/multi_agent_orchestration.ipynb`
  - تنفيذ الوكلاء: `Module08/samples/05/agents/`
  - README العينة: `Module08/samples/05/README.md`
- **عينات Microsoft الرسمية**:
  - [استدعاء الوظائف](https://github.com/microsoft/Foundry-Local/tree/main/samples/python/functioncalling)
  - [مرحبًا Foundry Local](https://github.com/microsoft/Foundry-Local/tree/main/samples/python/hello-foundry-local)
  - [Foundry Local Python SDK](https://github.com/microsoft/Foundry-Local/tree/main/sdk/python)
- **أمثلة التكامل**: https://learn.microsoft.com/en-us/azure/ai-foundry/foundry-local/how-to/how-to-chat-application-with-open-web-ui

## الخطوات التالية
- استكشاف وكلاء Azure AI للتنسيق المستضاف على السحابة
- إضافة موصلات مؤسسية (Microsoft Graph، البحث، قواعد البيانات)

---