عندما يتوقف الـ AI Agent عن انتظار الأوامر ويبدأ باتخاذ القرارات بنفسه، يتحول من أداة إلى شريك برمجي. سنفكك معاً كيف تبني هذا الكائن الرقمي المستقل، خطوة بخطوة، من الذاكرة إلى المعالج، مع كل الفخاخ التي واجهتها في الإنتاج.
في أحد أيام الجمعة الصيفية، كان السيرفر الخاص بنا يعالج ٣٧ ألف طلب في الدقيقة، وكلها تعتمد على وكيل ذكاء اصطناعي واحد يُدعى "نافي". المشكلة؟ نافي كان يتجمد تماماً عندما يواجه مهمة غير متوقعة، مثل تحليل sentiment في نص عربي مختلط بالإنجليزية والفرنسية. لم يكن الخطأ في النموذج اللغوي، بل في الطريقة التي صممنا بها الـ Event Loop الخاص به. عندما اكتشفنا أن ٦٨٪ من الـ I/O Bound Tasks كانت تُعالج بشكل متزامن بدلاً من غير المتزامن، أدركنا أن بناء AI Agent مستقل ليس مجرد تدريب نموذج، بل هو هندسة نظام كامل يتفاعل مع العالم الحقيقي دون تدخل بشري.
الـ AI Agent المستقل ليس مجرد سكربت ينفذ أوامر، بل هو كائن رقمي يفهم السياق، يتخذ قرارات، ويتعلم من أخطائه. الفرق بينه وبين الـ Chatbot التقليدي كالفرق بين روبوت مصنع يتحرك على سكك حديدية وبين إنسان يمشي في شارع مزدحم. الأول يتبع تعليمات محددة مسبقاً، والثاني يتفاعل مع البيئة، يتجنب العوائق، ويغير مساره بناءً على ما يراه ويسمعه. في هذا المقال، سنبني معاً وكيل ذكاء اصطناعي مستقل من الصفر، وسنتعمق في التفاصيل التي لا تذكرها الدورات التعليمية: كيف يدير الذاكرة؟ كيف يتجنب الـ Blocking Calls؟ وكيف يتعامل مع الـ Memory Leaks التي تظهر بعد ساعات من التشغيل المتواصل؟
قبل أن تكتب سطر كود واحد، عليك أن تفهم أن الـ AI Agent المستقل هو في الأساس نظام موزع مصغر. لديه ثلاثة مكونات رئيسية تعمل معاً في تناغم: الدماغ (النموذج اللغوي)، الجهاز العصبي (نظام اتخاذ القرارات)، والأطراف (الأدوات الخارجية مثل قواعد البيانات وAPIs). المشكلة الأكبر التي واجهتها في أول مشروع لي كانت في التواصل بين هذه المكونات. مثلاً، عندما يرسل الدماغ أمراً إلى الأطراف لتنفيذ مهمة ما، قد يستغرق الأمر ثوانٍ طويلة، وخلال هذه الثواني، يكون الجهاز العصبي في حالة انتظار، مما يسبب تجميداً كاملاً للوكيل. الحل؟ استخدام نمط الـ Actor Model الذي يسمح لكل مكون بالعمل بشكل مستقل وإرسال الرسائل إلى الآخرين دون انتظار الرد الفوري.
هناك أيضاً مفهوم الـ Context Window الذي لا يُعطى الاهتمام الكافي. معظم المطورين يعتقدون أن الـ Context هو مجرد نص يدخل للنموذج، لكن الحقيقة هي أن الـ Context هو ذاكرة قصيرة الأمد للوكيل. إذا كان حجم الـ Context Window صغيراً جداً، سيفقد الوكيل السياق بين المهام وسيكرر الأخطاء. وإذا كان كبيراً جداً، سيصبح بطيئاً جداً بسبب الـ Attention Mechanism في النماذج اللغوية. في تجربتي مع نموذج Llama 3، وجدنا أن الحجم الأمثل للـ Context Window هو ٨ آلاف توكين، لكن هذا يختلف بناءً على نوع المهام. مثلاً، إذا كان الوكيل يتعامل مع نصوص طويلة مثل التقارير المالية، فقد تحتاج إلى نافذة أكبر، لكن هذا سيزيد من تكلفة الـ Inference بشكل كبير.
# مثال على تنفيذ Actor Model باستخدام مكتبة Ray
import ray
from typing import Dict, Any
@ray.remote
class Brain:
def __init__(self, model_name: str):
# تحميل النموذج اللغوي هنا
self.model = load_model(model_name)
def process(self, input_text: str) -> Dict[str, Any]:
# معالجة النص باستخدام النموذج
resp self.model.generate(input_text)
return {"response": response, "confidence": 0.95}
@ray.remote
class NervousSystem:
def __init__(self):
self.brain = Brain.remote("llama3")
self.tools = {}
def decide(self, context: Dict[str, Any]) -> str:
# اتخاذ قرار بناءً على السياق
decision = self._analyze_context(context)
if decision == "use_tool":
return self._execute_tool(context)
else:
return ray.get(self.brain.process.remote(context["input"]))
def _analyze_context(self, context: Dict[str, Any]) -> str:
# منطق بسيط لتحليل السياق
if "tool_required" in context:
return "use_tool"
return "process_text"
def _execute_tool(self, context: Dict[str, Any]) -> str:
tool_name = context["tool_required"]
return f"Executing {tool_name} with params: {context.get('params', {})}"
# تشغيل النظام
ray.init()
nervous_system = NervousSystem.remote()
result = ray.get(nervous_system.decide.remote({
"input": "ما هو سعر سهم شركة آبل اليوم؟",
"tool_required": "stock_api",
"params": {"symbol": "AAPL"}
}))
print(result)الـ AI Agent المستقل بدون ذاكرة طويلة الأمد هو مثل إنسان يعاني من فقدان الذاكرة قصير الأمد. يمكنه إجراء محادثة قصيرة، لكن بعد دقائق قليلة، ينسى كل شيء. في مشروعنا الأخير مع شركة تجزئة، كان الوكيل بحاجة إلى تذكر تفضيلات العملاء من المحادثات السابقة لتقديم توصيات مخصصة. المشكلة؟ تخزين كل محادثة في قاعدة بيانات تقليدية كان يسبب بطءاً شديداً في الاستجابة، خاصة عندما يصل عدد العملاء إلى عشرات الآلاف. الحل الذي توصلنا إليه كان استخدام قاعدة بيانات متجهية مثل Weaviate أو Pinecone لتخزين الـ Embeddings الخاصة بالمحادثات. بهذه الطريقة، يمكن للوكيل استرجاع المعلومات ذات الصلة بسرعة باستخدام بحث المتجهات، بدلاً من البحث النصي التقليدي الذي يستهلك الكثير من الموارد.
لكن تخزين الـ Embeddings ليس كافياً. عليك أيضاً تصميم نظام يسمح للوكيل بتحديث ذاكرته بناءً على التجارب الجديدة. مثلاً، إذا اكتشف الوكيل أن العميل يفضل المنتجات العضوية، يجب أن يُحدث الـ Embedding الخاص بهذا العميل ليعكس هذا التفضيل. هذا يتطلب نظاماً مستمراً لإعادة التدريب الجزئي للـ Embeddings، وهو ما يُعرف بـ Online Learning. في تجربتنا، استخدمنا مكتبة FAISS من فيسبوك لبناء فهرس متجهي سريع، وقمنا بتحديثه كل ساعة باستخدام بيانات جديدة. النتيجة؟ انخفض وقت الاستجابة من ١.٢ ثانية إلى ٨٠ مللي ثانية، وزادت دقة التوصيات بنسبة ٣٧٪.
# مثال على استخدام Weaviate لتخزين واسترجاع الذاكرة الطويلة الأمد
import weaviate
from sentence_transformers import SentenceTransformer
# تحميل نموذج Embedding
model = SentenceTransformer('all-MiniLM-L6-v2')
# الاتصال بقاعدة بيانات Weaviate
client = weaviate.Client("http://localhost:8080")
# تعريف schema للذاكرة
schema = {
"classes": [{
"class": "ConversationMemory",
"properties": [{
"name": "content",
"dataType": ["text"]
}, {
"name": "user_id",
"dataType": ["string"]
}, {
"name": "timestamp",
"dataType": ["date"]
}]
}]
}
# إنشاء schema إذا لم يكن موجوداً
if not client.schema.exists("ConversationMemory"):
client.schema.create(schema)
# وظيفة لإضافة ذاكرة جديدة
def add_memory(user_id: str, content: str):
embedding = model.encode(content).tolist()
data_object = {
"content": content,
"user_id": user_id
}
client.data_object.create(
data_object=data_object,
class_name="ConversationMemory",
vector=embedding
)
# وظيفة لاسترجاع ذكريات ذات صلة
def retrieve_memories(user_id: str, query: str, limit: int = 3):
query_embedding = model.encode(query).tolist()
result = (
client.query
.get("ConversationMemory", ["content", "user_id"])
.with_near_vector({"vector": query_embedding})
.with_where({
"path": ["user_id"],
"operator": "Equal",
"valueString": user_id
})
.with_limit(limit)
.do()
)
return result["data"]["Get"]["ConversationMemory"]
# مثال على الاستخدام
add_memory("user123", "العميل يفضل المنتجات العضوية ويكره الألوان الزاهية")
memories = retrieve_memories("user123", "ما هي تفضيلات هذا العميل؟")
print(memories)إذا كان الـ AI Agent الخاص بك بطيئاً، فالسبب غالباً هو الـ Blocking Calls. في أحد المشاريع، كان الوكيل يستغرق ٤ ثوانٍ كاملة لتنفيذ مهمة بسيطة مثل جلب بيانات الطقس من API خارجي. بعد تحليل الكود، اكتشفنا أن الـ Event Loop كان يتوقف تماماً أثناء انتظار الرد من الـ API. الحل؟ استخدام الـ Non-Blocking I/O مع مكتبة مثل aiohttp بدلاً من requests التقليدية. الفرق كان مذهلاً: انخفض وقت الاستجابة إلى ٣٠٠ مللي ثانية فقط. لكن هذا ليس كل شيء. عليك أيضاً أن تفهم كيف يعمل الـ Event Loop في بايثون وكيف يتعامل مع الـ Coroutines. مثلاً، إذا استخدمت await داخل حلقة تكرار بدون ضبط الوقت، قد ينتهي بك الأمر إلى تجميد الوكيل تماماً. الحل هو استخدام asyncio.wait_for لفرض timeout على الـ Coroutines، أو استخدام مكتبات مثل tenacity لإعادة المحاولة تلقائياً في حالة الفشل.
هناك أيضاً مشكلة الـ Backpressure التي تحدث عندما يتلقى الوكيل طلبات بمعدل أسرع من قدرته على معالجتها. مثلاً، إذا كان الوكيل يتلقى ١٠٠٠ طلب في الثانية لكنه يستطيع معالجة ٥٠٠ طلب فقط، ستبدأ الطلبات تتراكم في الذاكرة، مما يسبب زيادة في استخدام الـ RAM وقد يؤدي إلى تحطم النظام. الحل الذي استخدمناه هو تطبيق نمط الـ Rate Limiting باستخدام مكتبة مثل slowapi، بالإضافة إلى استخدام قوائم انتظار مثل Redis لتخزين الطلبات الزائدة ومعالجتها لاحقاً. بهذه الطريقة، يمكن للوكيل الحفاظ على أدائه حتى تحت الضغط الشديد دون فقدان البيانات.
# مثال على تنفيذ Non-Blocking I/O مع Rate Limiting
import aiohttp
import asyncio
from slowapi import Limiter
from slowapi.util import get_remote_address
from fastapi import FastAPI, Request
app = FastAPI()
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
# وظيفة غير متزامنة لجلب البيانات من API خارجي
async def fetch_weather_data(city: str):
url = f"https://api.weatherapi.com/v1/current.json?key=YOUR_KEY&q={city}"
async with aiohttp.ClientSession() as session:
try:
async with session.get(url, timeout=5) as response:
if response.status == 200:
return await response.json()
else:
return {"error": "Failed to fetch data"}
except asyncio.TimeoutError:
return {"error": "Request timed out"}
except Exception as e:
return {"error": str(e)}
# نقطة نهاية مع Rate Limiting
@app.get("/weather/{city}")
@limiter.limit("5/minute")
async def get_weather(request: Request, city: str):
weather_data = await fetch_weather_data(city)
return weather_data
# تشغيل السيرفر
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)الـ Memory Leaks هي القاتل الصامت للـ AI Agents. في أحد المشاريع، كان الوكيل يعمل بشكل مثالي في أول ٢٤ ساعة، لكن بعد ٣ أيام، كان يستخدم ١٢ جيجابايت من الـ RAM وكان بطيئاً جداً. بعد تحليل باستخدام أدوات مثل tracemalloc وmemory-profiler، اكتشفنا أن المشكلة كانت في الـ Caching. كنا نخزن كل استجابة من النموذج اللغوي في ذاكرة التخزين المؤقت دون حد أقصى للحجم، مما تسبب في تراكم البيانات في الذاكرة. الحل؟ استخدام مكتبة مثل cachetools مع حد أقصى للحجم ووقت انتهاء الصلاحية. بهذه الطريقة، يتم حذف البيانات القديمة تلقائياً عندما تصل الذاكرة إلى الحد المحدد.
هناك أيضاً مشكلة الـ Circular References التي تحدث عندما يكون هناك مراجع دائرية بين الكائنات في الذاكرة. مثلاً، إذا كان لديك كائن A يشير إلى كائن B، وكائن B يشير إلى كائن A، فلن يتم تحرير الذاكرة حتى لو لم تعد هناك حاجة إلى هذه الكائنات. في بايثون، يمكنك استخدام أداة مثل gc.collect() لفرض جمع القمامة، لكن هذا ليس حلاً مثالياً. الحل الأفضل هو تجنب الـ Circular References من الأساس باستخدام أنماط تصميم مثل الـ WeakRef التي تسمح لك بإنشاء مراجع ضعيفة لا تمنع جمع القمامة. في تجربتنا، استخدام WeakRef قلل من استخدام الذاكرة بنسبة ٢٣٪ بعد ٤٨ ساعة من التشغيل المتواصل.
# مثال على تجنب Memory Leaks باستخدام cachetools وWeakRef
from cachetools import TTLCache
from weakref import WeakValueDictionary
import tracemalloc
# تفعيل تتبع الذاكرة
tracemalloc.start()
# استخدام TTLCache مع حد أقصى للحجم ووقت انتهاء الصلاحية
cache = TTLCache(maxsize=1000, ttl=3600)
# استخدام WeakValueDictionary لتجنب Circular References
weak_cache = WeakValueDictionary()
class AgentMemory:
def __init__(self):
self.data = {}
def add_data(self, key: str, value: str):
# تخزين البيانات في الذاكرة العادية
self.data[key] = value
# تخزين نسخة ضعيفة في الـ Weak Cache
weak_cache[key] = value
# تخزين نسخة في الـ TTLCache
cache[key] = value
def get_data(self, key: str):
# محاولة الحصول على البيانات من الـ TTLCache أولاً
if key in cache:
return cache[key]
# إذا لم تكن موجودة، حاول الحصول عليها من الذاكرة العادية
return self.data.get(key)
# مثال على الاستخدام
memory = AgentMemory()
memory.add_data("user123_pref", "يحب القهوة السوداء بدون سكر")
# طباعة استخدام الذاكرة الحالي
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')
print("[ Top 5 Memory Usage ]")
for stat in top_stats[:5]:
print(stat)الـ AI Agent المستقل الحقيقي لا يتوقف عن التعلم. في شركة نتفليكس، يستخدمون نظاماً يُدعى "Bandits" لتحسين توصيات الأفلام بناءً على تفاعل المستخدمين. الفكرة بسيطة: بدلاً من استخدام نموذج ثابت، يقوم النظام بتجربة عدة خيارات (مثل عرض أفلام مختلفة) ويراقب أي منها يحصل على تفاعل أفضل (مثل النقرات أو المشاهدات الكاملة). ثم يقوم بتحديث النموذج بناءً على هذه النتائج. في مشروعنا، طبقنا نفس المبدأ على الوكيل الخاص بنا. بدلاً من استخدام نموذج لغوي ثابت، قمنا بدمج نظام تعلم مستمر يُدعى "Reinforcement Learning from Human Feedback" (RLHF). الفكرة هي أن الوكيل يقوم بتجربة عدة ردود على نفس السؤال، ثم يختار الرد الذي يحصل على أعلى تقييم من المستخدمين. مع مرور الوقت، يتعلم الوكيل أي الردود هي الأفضل ويبدأ في استخدامها تلقائياً.
لكن التعلم المستمر ليس سهلاً كما يبدو. هناك مشكلة تُدعى "Catastrophic Forgetting" حيث ينسى النموذج المعلومات القديمة عندما يتعلم معلومات جديدة. مثلاً، إذا كان الوكيل قد تعلم كيفية الرد على أسئلة عن الطقس، ثم تعلم كيفية الرد على أسئلة عن الأسهم، قد ينسى كيفية الرد على أسئلة الطقس. الحل الذي استخدمناه هو استخدام تقنية تُدعى "Elastic Weight Consolidation" (EWC) التي تحافظ على المعلومات القديمة أثناء تعلم المعلومات الجديدة. بالإضافة إلى ذلك، استخدمنا نظاماً يُدعى "Experience Replay" حيث نخزن أمثلة من المهام القديمة ونعيد استخدامها أثناء التدريب الجديد. بهذه الطريقة، يحافظ الوكيل على معرفته السابقة بينما يتعلم مهارات جديدة.
# مثال مبسط على تنفيذ RLHF باستخدام مكتبة TRL من Hugging Face
from transformers import AutoTokenizer, AutoModelForCausalLM
from trl import PPOTrainer, PPOConfig, AutoModelForCausalLMWithValueHead
from datasets import load_dataset
# تحميل النموذج واللغة
model_name = "gpt2"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLMWithValueHead.from_pretrained(model_name)
# تحميل بيانات التدريب
dataset = load_dataset("imdb", split="train")
dataset = dataset.rename_columns({"text": "query", "label": "score"})
def tokenize(sample):
sample["input_ids"] = tokenizer.encode(sample["query"])
sample["query"] = tokenizer.decode(sample["input_ids"])
return sample
dataset = dataset.map(tokenize, batched=False)
# إعداد PPO Trainer
c PPOConfig(
model_name=model_name,
learning_rate=1e-5,
batch_size=32,
mini_batch_size=4,
gradient_accumulation_steps=1,
optimize_cuda_cache=True,
early_stopping=True,
target_kl=0.1,
ppo_epochs=4,
)
trainer = PPOTrainer(
config=config,
model=model,
ref_model=None,
tokenizer=tokenizer,
dataset=dataset,
)
# تدريب النموذج باستخدام RLHF
for epoch, batch in enumerate(trainer.dataloader):
query_tensors = batch["input_ids"]
# الحصول على ردود النموذج
response_tensors = trainer.generate(
query_tensors,
return_prompt=False,
**generation_kwargs
)
batch["response"] = tokenizer.batch_decode(response_tensors, skip_special_tokens=True)
# حساب المكافآت (في السيناريو الحقيقي، هذه تأتي من تقييم المستخدمين)
rewards = [torch.tensor(1.0) for _ in batch["response"]]
# تدريب النموذج باستخدام PPO
stats = trainer.step(query_tensors, response_tensors, rewards)
trainer.log_stats(stats, batch, rewards)
if epoch % 10 == 0:
print(f"Epoch {epoch} completed")أولاً، لا تبني الـ AI Agent الخاص بك على افتراض أن العالم مثالي. دائماً افترض أن الـ APIs الخارجية ستفشل، وأن المستخدمين سيدخلون مدخلات غير متوقعة، وأن السيرفر سيصبح بطيئاً فجأة. استخدم دوماً أنماط تصميم مثل الـ Circuit Breaker وRetry مع Backoff الأسي. مثلاً، إذا كان الوكيل يعتمد على API خارجي، استخدم مكتبة مثل tenacity لإعادة المحاولة تلقائياً مع زيادة الوقت بين المحاولات تدريجياً. بهذه الطريقة، إذا فشل الـ API مؤقتاً، لن يتوقف الوكيل عن العمل.
ثانياً، راقب الوكيل كما تراقب طفلاً صغيراً. استخدم أدوات مثل Prometheus وGrafana لمراقبة استخدام الذاكرة، وقت الاستجابة، وعدد الأخطاء. في تجربتي، معظم المشاكل التي واجهناها كان من الممكن اكتشافها مبكراً لو كنا نراقب النظام بشكل صحيح. مثلاً، إذا لاحظت أن وقت الاستجابة بدأ في الزيادة تدريجياً، قد يكون هذا علامة على وجود Memory Leak أو مشكلة في الـ Event Loop. استخدم أيضاً أدوات مثل Sentry لتتبع الأخطاء في الوقت الفعلي وإرسال تنبيهات عندما تحدث مشاكل.
ثالثاً، لا تخف من إعادة بناء الوكيل من الصفر إذا لزم الأمر. في أحد المشاريع، قضينا ٦ أشهر في بناء وكيل معقد جداً، لكن عندما بدأنا في توسيع نطاقه، اكتشفنا أن البنية الأساسية التي اخترناها لا تتناسب مع حجم البيانات الجديد. بدلاً من محاولة إصلاح النظام القديم، قررنا إعادة بنائه من الصفر باستخدام تقنيات جديدة مثل الـ Serverless Functions وKubernetes. النتيجة؟ انخفض وقت الاستجابة بنسبة ٦٠٪، وزادت قابلية التوسع بشكل كبير. الدرس المستفاد: أحياناً، إعادة البناء أسهل وأسرع من محاولة إصلاح نظام معقد.