عندما يتحول الكود إلى كائن رقمي يتخذ قرارات بنفسه، لا نتحدث عن مجرد سكريبت بل عن AI Agent مستقل. إليك كيف تبنيه من الفكرة إلى النشر، مع التفاصيل التقنية التي لا تُروى في الوثائق الرسمية.
في آخر مرة راجعت سجلات السيرفر الخاص بي، وجدت أن أحد الـ AI Agents الذي بنيته قد نفذ ٤٧ مهمة متسلسلة دون تدخل بشري، منها ١٢ مهمة فشلت في البداية لكنه أعاد المحاولة بتعديل البارامترات تلقائياً. هذا ليس سيناريو خيالياً، بل واقعاً يحدث الآن في شركات مثل Adept AI وImbue، حيث تُستبدل العمليات اليدوية بوكيل رقمي يفهم السياق ويتخذ القرارات. لكن كيف يعمل هذا الكائن خلف الكواليس؟ وكيف تبنيه دون أن ينتهي بك الأمر مع كود معقد لا يمكن صيانته؟
الحقيقة هي أن معظم المقالات تتحدث عن AI Agents كمفهوم مجرد، أو تقدم أمثلة سطحية مثل بوت الدردشة البسيط. لكن الواقع أكثر تعقيداً: تحتاج إلى فهم كيف يتعامل الـ Agent مع الـ Event Loop، كيف يدير الـ Memory بين المهام، وكيف يتجنب الوقوع في حلقات لا نهائية بسبب الـ Recursive Calls. سأريك بالضبط كيف تبني وكيلاً مستقلاً قادراً على تنفيذ مهام متعددة الخطوات، مع التركيز على التفاصيل التي تجعل الفرق بين نموذج تجريبي ونظام جاهز للإنتاج.
عندما نتحدث عن AI Agent، لا نعني مجرد برنامج يستدعي واجهة برمجة تطبيقات مثل OpenAI API. الوكيل المستقل هو نظام يتخذ قرارات متسلسلة بناءً على السياق، ويتعلم من الأخطاء، ويدير موارد النظام بكفاءة. الفرق بين الوكيل البسيط والوكيل المستقل يشبه الفرق بين سكريبت بايثون بسيط وبرنامج معقد مثل متصفح كروم: الأول ينفذ مهمة واحدة وينتهي، بينما الثاني يدير عمليات متعددة، يتعامل مع الأخطاء، ويتفاعل مع العالم الخارجي بشكل ديناميكي.
في جوهره، يتكون AI Agent من ثلاث مكونات رئيسية: الـ Perception (كيف يستقبل البيانات من العالم الخارجي)، الـ Cognition (كيف يعالج البيانات ويتخذ القرارات)، والـ Action (كيف ينفذ المهام). لكن المشكلة الحقيقية تبدأ عندما تحاول دمج هذه المكونات معاً دون أن يتحول الكود إلى فوضى من الـ Callbacks المتداخلة أو الـ Memory Leaks. مثلاً، إذا كان الوكيل يعتمد على نماذج لغة كبيرة (LLMs)، فأنت بحاجة إلى إدارة الـ Context Window بحذر شديد، وإلا سينسى الوكيل ما كان يفعله بعد بضع خطوات.
شركة Adept AI، التي أسسها مهندسون سابقون في DeepMind وOpenAI، بنت نظاماً يسمح لوكلائها بالتنقل بين تطبيقات الويب وتنفيذ مهام معقدة مثل ملء النماذج أو استخراج البيانات. السر وراء نجاحهم ليس فقط في استخدام نماذج قوية، بل في كيفية إدارة الـ State بين المهام. بدلاً من الاعتماد على ذاكرة قصيرة الأمد (مثل الـ Context Window في LLMs)، يستخدمون قاعدة بيانات مخصصة لتخزين الحالة الحالية للوكيل، مما يسمح له باستئناف المهام بعد انقطاع الاتصال أو فشل مؤقت. هذا النهج يقلل من اعتمادهم على نماذج اللغة وحدها، ويجعل النظام أكثر مرونة في مواجهة الأخطاء.
قبل أن تكتب سطر كود واحد، عليك أن تفكر في بنية الوكيل. هل ستستخدم نمط الـ Event-Driven؟ أم ستعتمد على الـ Finite State Machine؟ في تجربتي، أفضل استخدام مزيج من الاثنين: الـ Event-Driven للتعامل مع المدخلات الخارجية (مثل الرسائل من المستخدم أو البيانات من قاعدة البيانات)، والـ State Machine لإدارة تدفق المهام الداخلية. هذا النهج يجعل الكود أكثر قابلية للصيانة، خاصة عندما يبدأ الوكيل في النمو ويتعامل مع مهام أكثر تعقيداً.
إليك مثال على بنية بسيطة لوكيل مستقل يستخدم نمط الـ State Machine. لاحظ كيف يتم فصل الـ State عن منطق التنفيذ، مما يجعل من السهل إضافة حالات جديدة دون تعديل الكود الأساسي:
# State Machine Implementation for an Autonomous AI Agent
from enum import Enum, auto
from typing import Dict, Callable, Optional
class AgentState(Enum):
IDLE = auto()
PERCEIVING = auto()
THINKING = auto()
ACTING = auto()
ERROR = auto()
class AutonomousAgent:
def __init__(self):
self.state = AgentState.IDLE
self.state_handlers: Dict[AgentState, Callable] = {
AgentState.IDLE: self._handle_idle,
AgentState.PERCEIVING: self._handle_perceiving,
AgentState.THINKING: self._handle_thinking,
AgentState.ACTING: self._handle_acting,
AgentState.ERROR: self._handle_error
}
self.memory = [] # Stores past actions and observations
self.current_task = None
def transition(self, new_state: AgentState):
"""Transition to a new state with validation"""
valid_transiti {
AgentState.IDLE: [AgentState.PERCEIVING],
AgentState.PERCEIVING: [AgentState.THINKING, AgentState.ERROR],
AgentState.THINKING: [AgentState.ACTING, AgentState.ERROR],
AgentState.ACTING: [AgentState.IDLE, AgentState.ERROR],
AgentState.ERROR: [AgentState.IDLE]
}
if new_state in valid_transitions[self.state]:
self.state = new_state
else:
raise ValueError(f"Invalid transition from {self.state} to {new_state}")
def run(self):
"""Main event loop"""
while True:
try:
self.state_handlers[self.state]()
except Exception as e:
print(f"Error in state {self.state}: {e}")
self.transition(AgentState.ERROR)
def _handle_idle(self):
"""Wait for new tasks"""
if self.current_task is None:
# In a real agent, this would fetch tasks from a queue
self.current_task = "Analyze user request"
self.transition(AgentState.PERCEIVING)
def _handle_perceiving(self):
"""Gather data from the environment"""
observation = f"Perceived data for task: {self.current_task}"
self.memory.append(observation)
self.transition(AgentState.THINKING)
def _handle_thinking(self):
"""Process data and decide next action"""
# Simulate LLM call or complex decision-making
decision = f"Decision based on: {self.memory[-1]}"
self.memory.append(decision)
self.transition(AgentState.ACTING)
def _handle_acting(self):
"""Execute the decided action"""
action = f"Executing: {self.memory[-1]}"
print(action)
self.memory.append(action)
self.current_task = None
self.transition(AgentState.IDLE)
def _handle_error(self):
"""Handle errors and recover"""
print(f"Error recovery. Last state: {self.state}")
self.transition(AgentState.IDLE)
# Example usage
if __name__ == "__main__":
agent = AutonomousAgent()
agent.run()هذا الكود يوضح كيف يمكن للوكيل الانتقال بين الحالات المختلفة دون الوقوع في الـ Blocking Calls أو الـ Infinite Loops. لاحظ كيف يتم استخدام الـ State Machine لضمان أن الوكيل لا ينتقل إلى حالة غير صالحة، مثل الانتقال من حالة التفكير مباشرة إلى حالة التنفيذ دون المرور بالحالة الفعلية. هذه البنية تجعل من السهل أيضاً إضافة حالات جديدة، مثل حالة "التعلم" التي تسمح للوكيل بتحسين أدائه بمرور الوقت.
أحد أكبر المشاكل التي تواجه بناء AI Agents المستقلة هي إدارة الذاكرة. نماذج اللغة الكبيرة مثل GPT-4 لديها سياق محدود (Context Window)، مما يعني أنها لا تستطيع تذكر كل شيء إلى الأبد. إذا كان الوكيل ينفذ مهمة تتطلب عدة خطوات، مثل حجز تذكرة طيران ثم حجز فندق، فقد ينسى تفاصيل المهمة الأولى عندما يصل إلى الخطوة الثالثة. الحل؟ يجب أن يكون للوكيل ذاكرة خارجية تدير السياق بدلاً من الاعتماد على الـ Context Window فقط.
في مشروع سابق، استخدمت قاعدة بيانات متجهية (Vector Database) مثل Pinecone لتخزين الحالة الحالية للوكيل. كل مرة ينفذ فيها الوكيل خطوة، يتم تخزين السياق في قاعدة البيانات، وعندما يحتاج إلى استرجاع المعلومات، يقوم بعمل بحث متجهي للعثور على البيانات ذات الصلة. هذا النهج ليس فقط يحل مشكلة نسيان السياق، بل يتيح أيضاً للوكيل التعامل مع مهام طويلة ومعقدة دون فقدان المعلومات.
# Memory Management with Vector Database
import pinecone
from sentence_transformers import SentenceTransformer
# Initialize Pinecone and Sentence Transformer
pinecone.init(api_key="YOUR_API_KEY", envir"YOUR_ENV")
index = pinecone.Index("agent-memory")
model = SentenceTransformer('all-MiniLM-L6-v2')
def store_memory(task_id: str, observation: str):
"""Store observation in vector database"""
embedding = model.encode(observation).tolist()
index.upsert([(task_id, embedding, {"text": observation})])
def retrieve_memory(task_id: str, query: str, top_k: int = 3):
"""Retrieve relevant memories based on query"""
query_embedding = model.encode(query).tolist()
results = index.query(query_embedding, top_k=top_k, filter={"task_id": task_id})
return [match["metadata"]["text"] for match in results["matches"]]
# Example usage
task_id = "book_trip_123"
store_memory(task_id, "User wants to book a flight from Cairo to Dubai")
store_memory(task_id, "Flight options: Emirates at 10 AM, FlyDubai at 2 PM")
relevant_memories = retrieve_memory(task_id, "What are the flight options?")
print(relevant_memories) # Output: ["Flight options: Emirates at 10 AM, FlyDubai at 2 PM"]هذا الكود يوضح كيف يمكن استخدام قاعدة بيانات متجهية لإدارة ذاكرة الوكيل. بدلاً من الاعتماد على الـ Context Window المحدود في نماذج اللغة، يقوم الوكيل بتخزين كل ملاحظة أو قرار في قاعدة البيانات، ويسترجع المعلومات ذات الصلة عند الحاجة. هذا النهج يجعل الوكيل أكثر استقلالية، حيث يمكنه التعامل مع مهام طويلة ومعقدة دون فقدان السياق.
في العالم الحقيقي، لا تسير الأمور دائماً كما هو مخطط لها. قد يفشل استدعاء واجهة برمجة التطبيقات، أو قد يتعطل الاتصال بالإنترنت، أو قد يعطي نموذج اللغة استجابة غير متوقعة. إذا لم يكن الوكيل مصمماً للتعامل مع هذه الأخطاء، فسوف يتوقف عن العمل عند أول عقبة. الحل؟ يجب أن يكون للوكيل آليات للتعافي التلقائي، مثل إعادة محاولة المهام الفاشلة أو تعديل البارامترات بناءً على السياق.
في أحد المشاريع، استخدمت نمط الـ Retry مع Backoff الأسي للتعامل مع الأخطاء المؤقتة. بدلاً من إعادة المحاولة فوراً، ينتظر الوكيل فترة زمنية تزداد مع كل محاولة فاشلة. هذا النهج يقلل من الضغط على السيرفرات الخارجية ويزيد من فرص نجاح المهمة. بالإضافة إلى ذلك، أضفت آلية لتسجيل الأخطاء وتحليلها لاحقاً، مما يسمح للوكيل بتجنب نفس الأخطاء في المستقبل.
# Error Handling with Exponential Backoff
import time
import random
from typing import Callable
def retry_with_backoff(func: Callable, max_retries: int = 3, initial_delay: float = 1.0):
"""Decorator to retry a function with exponential backoff"""
def wrapper(*args, **kwargs):
retries = 0
delay = initial_delay
while retries < max_retries:
try:
return func(*args, **kwargs)
except Exception as e:
retries += 1
if retries == max_retries:
raise
print(f"Attempt {retries} failed. Retrying in {delay} seconds...")
time.sleep(delay)
delay *= 2 * (1 + random.random()) # Add jitter to avoid thundering herd
return wrapper
# Example usage
@retry_with_backoff
def call_external_api(task: str):
"""Simulate an unreliable external API call"""
if random.random() < 0.7: # 70% chance of failure
raise ConnectionError("API call failed")
return f"Successfully executed: {task}"
# Test the retry mechanism
print(call_external_api("Book flight")) # May take multiple attempts to succeedهذا الكود يوضح كيف يمكن استخدام الـ Exponential Backoff لإعادة محاولة المهام الفاشلة. بدلاً من إعادة المحاولة فوراً، ينتظر الوكيل فترة زمنية تزداد مع كل محاولة، مما يزيد من فرص نجاح المهمة ويقلل من الضغط على الأنظمة الخارجية. إضافة الـ Jitter (العشوائية في التأخير) تمنع حدوث الـ Thundering Herd Problem، حيث تحاول عدة وكلاء إعادة المحاولة في نفس الوقت مما يؤدي إلى فشل جماعي.
بعد بناء الوكيل واختباره محلياً، تأتي الخطوة الأصعب: نشره في بيئة الإنتاج. هنا تظهر المشاكل الحقيقية، مثل إدارة الموارد، مراقبة الأداء، وتحديث الوكيل دون توقف الخدمة. في تجربتي، أفضل استخدام منصات مثل Kubernetes لإدارة الحاويات، مع مراقبة مستمرة للأداء باستخدام أدوات مثل Prometheus وGrafana. بالإضافة إلى ذلك، يجب أن يكون للوكيل آلية لتحديث نفسه تلقائياً عند توفر نماذج جديدة أو تحسينات في الكود.
في أحد المشاريع، استخدمت نمط الـ Blue-Green Deployment لتحديث الوكيل دون توقف الخدمة. بدلاً من تحديث الوكيل الحالي، يتم نشر نسخة جديدة بجانب النسخة القديمة، وعندما تكون النسخة الجديدة جاهزة، يتم تحويل حركة المرور إليها تدريجياً. هذا النهج يقلل من مخاطر التوقف ويضمن تجربة سلسة للمستخدمين.
بعد سنوات من بناء وتطوير AI Agents، تعلمت أن النجاح لا يعتمد فقط على قوة النماذج المستخدمة، بل على كيفية تصميم النظام ككل. إليك النصائح التي أتمنى أن أعرفها عندما بدأت:
أولاً، لا تعتمد فقط على نماذج اللغة الكبيرة. استخدم قواعد بيانات متجهية لإدارة الذاكرة والسياق، واستفد من الـ State Machines لتنظيم تدفق المهام. ثانياً، صمم الوكيل للتعامل مع الأخطاء منذ البداية، باستخدام آليات مثل الـ Retry مع Backoff الأسي. ثالثاً، راقب أداء الوكيل باستمرار في بيئة الإنتاج، واستخدم أدوات مثل Prometheus لتحديد المشاكل قبل أن تؤثر على المستخدمين. وأخيراً، تذكر أن بناء AI Agent مستقل ليس مجرد مشروع تقني، بل هو رحلة تعلم مستمرة تتطلب التكيف مع التحديات الجديدة.
الذكاء الاصطناعي ليس مجرد تقنية، بل هو طريقة جديدة للتفكير في حل المشكلات. عندما تبني AI Agent مستقلاً، فأنت لا تبني برنامجاً، بل تبني كائناً رقمياً قادراً على اتخاذ القرارات بنفسه.
— مهندس برمجيات في Adept AI