كيف تبني AI Agent مستقلاً من الصفر: رحلة المهندس خلف الكواليس

# مثال مبسط على Perception Layer لوكيل إعلانات
import websockets
import asyncio
import json
from textblob import TextBlob

class AdCampaignAgentPerception:
 def __init__(self, campaign_id):
 self.campaign_id = campaign_id
 self.websocket_url = f"wss://api.social-platform.com/ws/campaigns/{campaign_id}"
 self.sentiment_history = []
 self.engagement_trend = []

 async def connect(self):
 async with websockets.connect(self.websocket_url) as ws:
 while True:
 data = await ws.recv()
 event = json.loads(data)
 
 # معالجة البيانات الحية
 if event['type'] == 'comment':
 sentiment = TextBlob(event['text']).sentiment.polarity
 self.sentiment_history.append(sentiment)
 # كشف تغيرات مفاجئة في المشاعر
 if len(self.sentiment_history) > 5 and abs(sentiment - sum(self.sentiment_history[-5:])/5) > 0.4:
 await self._handle_sentiment_shift(sentiment)
 
 # تحليل معدل التفاعل
 if event['type'] == 'engagement':
 self.engagement_trend.append(event['value'])
 if len(self.engagement_trend) > 10 and self.engagement_trend[-1] < sum(self.engagement_trend[-10:-5])/5 * 0.7:
 await self._handle_engagement_drop()

 async def _handle_sentiment_shift(self, sentiment):
 # هنا سيتم إرسال إشارة إلى Reasoning Engine
 print(f"ALERT: Sentiment shift detected! New sentiment: {sentiment}")
 # في الإنتاج، سيكون هذا Event يتم نشره في Message Queue
 
 async def _handle_engagement_drop(self):
 print("ALERT: Engagement drop detected!")
 # يمكن هنا تعديل ميزانية الحملة تلقائياً

# لاحظ كيف أن هذا الكود لا ينتظر البيانات بل يستمع إليها بشكل غير متزامن
# المشكلة الحقيقية هنا هي الـ Memory Leak في قائمة sentiment_history
# في الإنتاج، ستحتاج إلى آلية تنظيف أو قاعدة بيانات زمنية مثل TimescaleDB

# مثال مبسط على Reasoning Engine هجين
from prophet import Prophet
import pandas as pd
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

class HybridReasoningEngine:
 def __init__(self):
 self.llm = OpenAI(temperature=0.3)
 self.forecast_model = Prophet()
 self.rules = {
 'min_stock': 10,
 'reorder_quantity': 50,
 'max_price_drop': 0.2
 }

 def make_decision(self, product_data, historical_sales):
 # 1. تطبيق القواعد الثابتة
 if product_data['current_stock'] < self.rules['min_stock']:
 return {
 'action': 'reorder',
 'quantity': self.rules['reorder_quantity'],
 'reason': 'Stock below minimum threshold'
 }
 
 # 2. استخدام نموذج التنبؤ
 df = pd.DataFrame({
 'ds': pd.to_datetime(historical_sales['date']),
 'y': historical_sales['sales']
 })
 self.forecast_model.fit(df)
 future = self.forecast_model.make_future_dataframe(periods=7)
 forecast = self.forecast_model.predict(future)
 predicted_sales = forecast['yhat'].tail(7).sum()
 
 if predicted_sales < product_data['current_stock'] * 0.7:
 # 3. استخدام LLM لاتخاذ القرار الغامض
 prompt = PromptTemplate(
 input_variables=["product_name", "current_price", "predicted_sales", "current_stock"],
 template="""
 Product: {product_name}
 Current price: ${current_price}
 Predicted weekly sales: {predicted_sales} units
 Current stock: {current_stock} units
 
 Sales forecast shows a significant drop. Should we:
 1. Reduce price to increase demand
 2. Stop ordering this product
 3. Do nothing
 
 Consider that reducing price may increase sales but reduce profit margin.
 Provide only the number of the best option and a very brief reason.
 """
 )
 llm_resp self.llm(prompt.format(
 product_name=product_data['name'],
 current_price=product_data['price'],
 predicted_sales=predicted_sales,
 current_stock=product_data['current_stock']
 ))
 
 if "1" in llm_response:
 # حساب نسبة التخفيض المثلى
 price_drop = min(self.rules['max_price_drop'], 
 1 - (predicted_sales / (product_data['current_stock'] * 0.8)))
 return {
 'action': 'price_adjustment',
 'new_price': product_data['price'] * (1 - price_drop),
 'reason': f"Predicted sales drop. LLM suggested price reduction. {llm_response}"
 }
 elif "2" in llm_response:
 return {
 'action': 'discontinue',
 'reason': f"Predicted sales drop. LLM suggested discontinuation. {llm_response}"
 }
 
 return {'action': 'no_action', 'reason': 'No immediate action required'}

# المشكلة الحقيقية هنا هي الـ Latency في اتخاذ القرار
# في الإنتاج، ستحتاج إلى Caching للنتائج المتوقعة لتجنب إعادة الحساب
# كما أن LLM بطيء جداً - ربما ستحتاج إلى نموذج أصغر مثل DistilBERT
# أو استخدام Vector Database لتسريع استرجاع السياق المشابه

# نظام ذاكرة متعدد الطبقات لAI Agent
from datetime import datetime, timedelta
from sentence_transformers import SentenceTransformer
import numpy as np
import faiss

class MultiLayerMemorySystem:
 def __init__(self):
 # Working Memory - ذاكرة مؤقتة للبيانات الحية
 self.working_memory = []
 self.working_memory_max_size = 100
 
 # Long-Term Memory - قاعدة بيانات للذكريات المهمة
 self.l []
 
 # نموذج لتوليد Embeddings
 self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
 
 # FAISS index للبحث السريع في الذكريات
 self.memory_index = faiss.IndexFlatL2(384) # 384 هو حجم Embedding
 
 # عتبة لترقية الذكريات من Working إلى Long-Term
 self.promotion_threshold = 3
 self.promotion_window = timedelta(days=30)
 
 # تتبع تكرار الأحداث
 self.event_counter = {}

 def add_to_working_memory(self, event):
 """إضافة حدث إلى الذاكرة الفورية"""
 # تنظيف الذاكرة إذا تجاوزت الحد الأقصى
 if len(self.working_memory) >= self.working_memory_max_size:
 self.working_memory.pop(0)
 
 # إضافة الحدث الجديد
 self.working_memory.append({
 'event': event,
 'timestamp': datetime.now(),
 'count': 1
 })
 
 # تحديث عداد الأحداث
 event_key = str(event) # في الإنتاج، استخدم مفتاحاً أفضل
 self.event_counter[event_key] = self.event_counter.get(event_key, 0) + 1
 
 # التحقق من ترقية الحدث إلى الذاكرة طويلة المدى
 if self.event_counter[event_key] >= self.promotion_threshold:
 recent_events = [e for e in self.working_memory 
 if str(e['event']) == event_key 
 and (datetime.now() - e['timestamp']) <= self.promotion_window]
 
 if len(recent_events) >= self.promotion_threshold:
 self._promote_to_long_term_memory(event_key)

 def _promote_to_long_term_memory(self, event_key):
 """ترقية حدث إلى الذاكرة طويلة المدى"""
 # توليد Embedding للحدث
 embedding = self.embedding_model.encode([event_key])[0]
 
 # إضافة إلى قاعدة البيانات والـ FAISS index
 memory_id = len(self.long_term_memory)
 self.long_term_memory.append({
 'event': event_key,
 'embedding': embedding,
 'timestamp': datetime.now()
 })
 
 self.memory_index.add(np.array([embedding]))
 
 # إعادة تعيين العداد
 self.event_counter[event_key] = 0

 def retrieve_similar_memories(self, query, k=3):
 """استرجاع الذكريات المشابهة للسياق الحالي"""
 query_embedding = self.embedding_model.encode([query])[0]
 distances, indices = self.memory_index.search(np.array([query_embedding]), k)
 
 return [self.long_term_memory[i] for i in indices[0]]

# المشكلة الحقيقية هنا هي الـ Memory Leak المحتملة
# في الإنتاج، ستحتاج إلى:
# 1. آلية تنظيف للذكريات القديمة
# 2. ضغط للEmbeddings لتقليل حجم قاعدة البيانات
# 3. نظام نسخ احتياطي للذاكرة طويلة المدى
# 4. آلية للتعامل مع الذكريات المتضاربة (مثل حدثين بنفس السياق ولكن بنتائج مختلفة)

# نظام تنفيذ مرن مع Retry وFallbacks
import time
import random
import requests
from typing import List, Dict, Callable

class ActionExecutor:
 def __init__(self):
 self.acti []
 self.fallback_strategies = {
 'reorder': self._get_fallback_suppliers,
 'price_adjustment': self._get_fallback_pricing_strategies
 }
 self.retry_delays = [0.5, 1, 5, 30] # ثوانٍ

 async def execute_action(self, action: Dict, context: Dict) -> bool:
 """تنفيذ فعل مع Retry وFallbacks"""
 action_type = action['action']
 max_retries = len(self.retry_delays)
 attempt = 0
 
 while attempt <= max_retries:
 try:
 # محاولة التنفيذ
 success = await self._try_execute(action, context)
 
 if success:
 self._log_action(action, context, True)
 return True
 
 # إذا فشلت المحاولة الأخيرة، جرب Fallback
 if attempt == max_retries:
 fallback_actions = self.fallback_strategies.get(action_type, lambda _: [])(context)
 for fallback_action in fallback_actions:
 fallback_success = await self.execute_action(fallback_action, context)
 if fallback_success:
 return True
 
 # إذا فشلت كل الفallbacks، أرسل إشعاراً
 self._notify_human(action, context)
 return False
 
 # انتظار قبل إعادة المحاولة
 delay = self.retry_delays[attempt] * (1 + random.uniform(-0.1, 0.1))
 time.sleep(delay)
 attempt += 1
 
 except Exception as e:
 self._log_action(action, context, False, str(e))
 if attempt == max_retries:
 self._notify_human(action, context, str(e))
 return False
 delay = self.retry_delays[attempt] * (1 + random.uniform(-0.1, 0.1))
 time.sleep(delay)
 attempt += 1
 
 return False

 async def _try_execute(self, action: Dict, context: Dict) -> bool:
 """محاولة تنفيذ الفعل"""
 if action['action'] == 'reorder':
 # مثال على استدعاء API للمورد
 response = requests.post(
 "https://api.supplier.com/orders",
 json={
 "product_id": context['product_id'],
 "quantity": action['quantity'],
 "expected_delivery": context['expected_delivery']
 },
 timeout=10
 )
 return response.status_code == 200
 
 elif action['action'] == 'price_adjustment':
 response = requests.patch(
 f"https://api.ecommerce.com/products/{context['product_id']}",
 json={"price": action['new_price']},
 timeout=10
 )
 return response.status_code == 200
 
 return False

 def _get_fallback_suppliers(self, context: Dict) -> List[Dict]:
 """الحصول على موردين بديلين"""
 # في الإنتاج، هذه البيانات ستأتي من قاعدة بيانات
 fallback_suppliers = [
 {"supplier_id": "supplier_b", "price_multiplier": 1.1},
 {"supplier_id": "supplier_c", "price_multiplier": 1.2}
 ]
 
 return [{
 'action': 'reorder',
 'quantity': context['reorder_quantity'],
 'supplier': supplier['supplier_id'],
 'price_multiplier': supplier['price_multiplier']
 } for supplier in fallback_suppliers]

 def _get_fallback_pricing_strategies(self, context: Dict) -> List[Dict]:
 """الحصول على استراتيجيات تسعير بديلة"""
 current_price = context['current_price']
 return [
 {
 'action': 'price_adjustment',
 'new_price': current_price * 0.9 # تخفيض 10%
 },
 {
 'action': 'price_adjustment',
 'new_price': current_price * 0.85 # تخفيض 15%
 }
 ]

 def _log_action(self, action: Dict, context: Dict, success: bool, error: str = None):
 """تسجيل محاولة التنفيذ"""
 self.action_history.append({
 'action': action,
 'context': context,
 'success': success,
 'error': error,
 'timestamp': datetime.now()
 })
 
 # في الإنتاج، سيتم إرسال هذا إلى قاعدة بيانات أو نظام مراقبة

 def _notify_human(self, action: Dict, context: Dict, error: str = None):
 """إرسال إشعار للإنسان"""
 message = f"AI Agent failed to execute action: {action['action']}"
 if error:
 message += f"\nError: {error}"
 message += f"\nContext: {context}"
 
 # في الإنتاج، سيتم إرسال هذا عبر Slack أو Email
 print(f"HUMAN NOTIFICATION: {message}")

# المشكلة الحقيقية هنا هي الـ Blocking I/O في استدعاءات API
# في الإنتاج، ستحتاج إلى:
# 1. استخدام async/await مع aiohttp بدلاً من requests
# 2. نظام Circuit Breaker لمنع تحميل السيرفر في حالة الفشل
# 3. آلية لاختبار صحة الـ APIs قبل الاستخدام
# 4. نظام لتتبع زمن الاستجابة وتحسينه

# مثال على ملف Kubernetes Deployment لAI Agent
apiVersion: apps/v1
kind: Deployment
metadata:
 name: ai-agent
spec:
 replicas: 3
 selector:
 matchLabels:
 app: ai-agent
 template:
 metadata:
 labels:
 app: ai-agent
 spec:
 containers:
 - name: agent
 image: registry.nouvil.net/ai-agent:latest
 env:
 - name: PERCEPTION_WS_URL
 value: "wss://api.social-platform.com/ws"
 - name: MEMORY_DB_URL
 value: "postgresql://user:pass@timescaledb:5432/memory"
 - name: VECTOR_DB_URL
 value: "http://weaviate:8080"
 - name: MESSAGE_QUEUE_URL
 value: "kafka:9092"
 resources:
 limits:
 cpu: "1"
 memory: "2Gi"
 requests:
 cpu: "500m"
 memory: "1Gi"
 livenessProbe:
 httpGet:
 path: /health
 port: 8000
 initialDelaySeconds: 30
 periodSeconds: 10
 readinessProbe:
 httpGet:
 path: /ready
 port: 8000
 initialDelaySeconds: 5
 periodSeconds: 5
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
 name: ai-agent-hpa
spec:
 scaleTargetRef:
 apiVersion: apps/v1
 kind: Deployment
 name: ai-agent
 minReplicas: 2
 maxReplicas: 20
 metrics:
 - type: Resource
 resource:
 name: cpu
 target:
 type: Utilization
 averageUtilization: 70
 - type: Resource
 resource:
 name: memory
 target:
 type: Utilization
 averageUtilization: 80

# المشاكل الحقيقية هنا:
# 1. الـ Rolling Updates قد تسبب توقفاً مؤقتاً في الخدمة
# الحل: استخدام Blue-Green Deployment
# 2. الـ Liveness Probe قد يقتل الـ Pod بسبب بطء مؤقت
# الحل: ضبط الـ initialDelaySeconds بشكل مناسب
# 3. الـ Auto-scaling قد يسبب تكاليف غير متوقعة
# الحل: إضافة حدود قصوى للـ Replicas بناءً على الميزانية
# 4. الـ Secrets يجب ألا تكون في ملف YAML
# الحل: استخدام Kubernetes Secrets أو HashiCorp Vault