اكتشف كيف تبني AI Agent مستقلاً قادراً على اتخاذ القرارات والتنفيذ دون تدخل بشري، بخطوات عملية وكود حقيقي من الصفر إلى الإنتاج.
في عام ٢٠٢٣، أعلنت شركة Adept AI عن نموذجها الجديد ACT-1 الذي يستطيع التحكم في واجهات المستخدم وتنفيذ مهام معقدة عبر المتصفح دون تدخل بشري. بعد أشهر، أطلقت شركة Cognition AI منتجها Devin، أول مهندس برمجيات آلي مستقل قادر على كتابة الكود وتصحيح الأخطاء وإدارة المشاريع كاملة. هذه ليست مجرد تجارب أكاديمية، بل منتجات حقيقية تعمل اليوم في شركات مثل Scale AI وBrex. السؤال ليس هل يمكن بناء AI Agent مستقل، بل كيف تبنيه بشكل صحيح دون أن ينتهي بك الأمر مع وكيل يتصرف كروبوت عشوائي يحذف قاعدة البيانات أو يرسل رسائل بريد إلكتروني مزعجة لكل عملائك.
الحقيقة هي أن معظم المطورين الذين يحاولون بناء AI Agents مستقلين يفشلون ليس لأنهم لا يفهمون النماذج اللغوية الكبيرة، بل لأنهم يتجاهلون البنية التحتية اللازمة لتشغيل هذه الوكلاء بشكل موثوق. المشكلة ليست في الـ Prompt Engineering، بل في كيفية إدارة الـ State، التعامل مع الـ I/O Bound Operations، والتعامل مع الأخطاء التي لا يمكن التنبؤ بها. في هذا المقال، سأريك كيف تبني AI Agent مستقلاً من الصفر، خطوة بخطوة، مع التركيز على التفاصيل التي لا تجدها في الـ Tutorials العادية.
عندما بدأت العمل على أول AI Agent مستقل لي قبل عامين، كنت أعتقد أن التحدي الأكبر هو جعل النموذج اللغوي يفهم المهمة ويتخذ القرارات. لكن سرعان ما اكتشفت أن التحدي الحقيقي هو إدارة الـ State بشكل صحيح. معظم المطورين يستخدمون متغيرات بسيطة لتخزين حالة الوكيل، لكن هذا النهج ينهار عندما يكون لديك وكيل يعمل لفترات طويلة أو يتعامل مع مهام معقدة. مثلاً، إذا كان الوكيل يعمل على مهمة تتطلب عدة خطوات مثل تحليل البيانات ثم إرسال تقرير عبر البريد الإلكتروني، فإن فقدان الـ State يعني أن الوكيل قد ينسى ما فعله في الخطوة السابقة ويبدأ من الصفر.
مشكلة أخرى شائعة هي التعامل مع الـ Blocking Calls. عندما يرسل الوكيل طلباً إلى API خارجي أو ينتظر رداً من قاعدة بيانات، فإن الـ Event Loop يتوقف عن العمل، مما يجعل الوكيل غير قادر على الاستجابة لأي أحداث أخرى. في إحدى المشاريع التي عملت عليها، كان الوكيل يتوقف تماماً عند انتظار رد من واجهة برمجة تطبيقات خارجية، مما أدى إلى تجميد النظام بالكامل. الحل؟ استخدام الـ Async/Await بشكل صحيح وإدارة الـ Timeouts بدقة. لكن حتى هذا ليس كافياً إذا لم تكن لديك آلية لإعادة محاولة الطلبات الفاشلة أو التعامل مع الأخطاء غير المتوقعة.
لبناء AI Agent مستقل، تحتاج إلى بنية تحتية تتكون من عدة مكونات رئيسية: نظام إدارة الـ State، محرك اتخاذ القرارات، واجهة برمجة التطبيقات للتفاعل مع العالم الخارجي، وآلية للتعامل مع الأخطاء والاسترداد. لنبدأ بنظام إدارة الـ State. بدلاً من استخدام متغيرات بسيطة، يجب استخدام قاعدة بيانات خفيفة وسريعة مثل SQLite أو Redis لتخزين حالة الوكيل. هذا يسمح للوكيل باستعادة حالته في أي وقت، حتى بعد إعادة تشغيل النظام.
محرك اتخاذ القرارات هو الجزء الذي يستخدم النموذج اللغوي الكبير لاتخاذ القرارات بناءً على الـ State الحالي. هنا تأتي أهمية الـ Prompt Engineering. يجب تصميم الـ Prompt بحيث يحتوي على جميع المعلومات اللازمة لاتخاذ القرار، بما في ذلك السياق التاريخي والإجراءات المتاحة. مثلاً، إذا كان الوكيل مسؤولاً عن إدارة حسابات العملاء، فإن الـ Prompt يجب أن يحتوي على معلومات عن الحسابات الحالية، الإجراءات المسموح بها (مثل الإيداع والسحب)، والقيود المفروضة (مثل الحد الأقصى للسحب).
import sqlite3
from typing import Dict, List, Optional
class StateManager:
def __init__(self, db_path: str = ":memory:"):
self.c sqlite3.connect(db_path)
self._init_db()
def _init_db(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS agent_state (
key TEXT PRIMARY KEY,
value TEXT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
self.conn.commit()
def set_state(self, key: str, value: str) -> None:
self.conn.execute(
"INSERT OR REPLACE INTO agent_state (key, value) VALUES (?, ?)",
(key, value)
)
self.conn.commit()
def get_state(self, key: str) -> Optional[str]:
cursor = self.conn.execute("SELECT value FROM agent_state WHERE key = ?", (key,))
row = cursor.fetchone()
return row[0] if row else None
def get_all_states(self) -> Dict[str, str]:
cursor = self.conn.execute("SELECT key, value FROM agent_state")
return {row[0]: row[1] for row in cursor.fetchall()}
def clear_state(self) -> None:
self.conn.execute("DELETE FROM agent_state")
self.conn.commit()
# مثال على استخدام StateManager
state_manager = StateManager()
state_manager.set_state("current_task", "analyze_customer_data")
state_manager.set_state("last_action", "fetch_data_from_api")
print(state_manager.get_all_states())محرك اتخاذ القرارات هو قلب الـ AI Agent. هنا تأتي أهمية تصميم الـ Prompt بشكل صحيح. يجب أن يحتوي الـ Prompt على ثلاثة أجزاء رئيسية: السياق، الإجراءات المتاحة، والهدف. السياق يشمل المعلومات الحالية مثل حالة الوكيل والبيانات التي تم جمعها. الإجراءات المتاحة هي الخيارات التي يمكن للوكيل اتخاذها في هذه اللحظة. الهدف هو المهمة الرئيسية التي يحاول الوكيل تحقيقها.
في تجربتي، وجدت أن إضافة أمثلة قليلة (Few-Shot Learning) إلى الـ Prompt تحسن بشكل كبير من جودة القرارات التي يتخذها النموذج. مثلاً، إذا كان الوكيل مسؤولاً عن إدارة المخزون، يمكنك إضافة أمثلة على كيفية اتخاذ قرارات الشراء أو البيع بناءً على مستويات المخزون والطلب. لكن يجب توخي الحذر هنا، لأن إضافة أمثلة كثيرة قد يؤدي إلى زيادة حجم الـ Prompt بشكل كبير، مما يزيد من تكلفة الاستدعاءات ويبطئ الأداء.
from openai import OpenAI
import json
class DecisionEngine:
def __init__(self, api_key: str):
self.client = OpenAI(api_key=api_key)
def make_decision(self, state: Dict[str, str], available_actions: List[str], goal: str) -> str:
# بناء الـ Prompt
c "\n".join([f"{k}: {v}" for k, v in state.items()])
actions = ", ".join(available_actions)
prompt = f"""
You are an autonomous AI agent with the following state:
{context}
Available actions: {actions}
Goal: {goal}
Based on the current state and available actions, what is the best next action to achieve the goal?
Respond with ONLY the action name, nothing else.
Examples:
- If inventory is low and action is 'order_more', respond with 'order_more'
- If customer has a complaint and action is 'escalate_to_manager', respond with 'escalate_to_manager'
"""
response = self.client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.3,
max_tokens=10
)
decision = response.choices[0].message.content.strip()
if decision not in available_actions:
raise ValueError(f"Invalid decision: {decision}")
return decision
# مثال على استخدام DecisionEngine
decision_engine = DecisionEngine(api_key="your-api-key")
state = {"inventory_level": "low", "customer_complaint": "none"}
actions = ["order_more", "wait", "contact_supplier"]
goal = "maintain optimal inventory levels"
decision = decision_engine.make_decision(state, actions, goal)
print(f"Decision: {decision}")حتى أفضل الـ AI Agents يمكن أن تواجه أخطاء غير متوقعة. قد يفشل استدعاء API، أو قد يتوقف السيرفر عن الاستجابة، أو قد يتخذ النموذج قراراً غير منطقي. لذلك، يجب أن يكون لديك آلية للتعامل مع الأخطاء والاسترداد. في أحد المشاريع التي عملت عليها، كان الوكيل يرسل طلبات إلى واجهة برمجة تطبيقات خارجية، لكن هذه الواجهة كانت تفشل أحياناً دون سبب واضح. الحل كان استخدام آلية إعادة المحاولة مع تأخير متزايد (Exponential Backoff). هذا يعني أنه إذا فشل الطلب الأول، ينتظر الوكيل ثانية واحدة قبل المحاولة مرة أخرى، ثم ثانيتين، ثم أربع ثوانٍ، وهكذا.
بالإضافة إلى ذلك، يجب أن يكون لديك آلية لتسجيل الأخطاء والمراقبة. استخدم أدوات مثل Sentry أو Datadog لتتبع الأخطاء ومراقبة أداء الوكيل. هذا يسمح لك بتحديد الأنماط ومعرفة متى وأين تحدث الأخطاء. مثلاً، إذا لاحظت أن الوكيل يفشل دائماً عند التعامل مع طلبات معينة، يمكنك تعديل الـ Prompt أو إضافة منطق إضافي للتعامل مع هذه الحالات الخاصة.
import time
import logging
from typing import Callable, Any
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def retry_with_backoff(
func: Callable,
max_retries: int = 3,
initial_delay: float = 1.0,
backoff_factor: float = 2.0
) -> Any:
"""
تنفيذ دالة مع إعادة المحاولة باستخدام تأخير متزايد.
"""
retries = 0
delay = initial_delay
while retries < max_retries:
try:
return func()
except Exception as e:
retries += 1
if retries == max_retries:
logger.error(f"Failed after {max_retries} retries: {e}")
raise
logger.warning(f"Attempt {retries} failed, retrying in {delay} seconds: {e}")
time.sleep(delay)
delay *= backoff_factor
# مثال على استخدام retry_with_backoff
def fetch_data_from_api():
# محاكاة فشل عشوائي
import random
if random.random() < 0.7: # 70% فرصة للفشل
raise ConnectionError("API request failed")
return {"data": "success"}
try:
result = retry_with_backoff(fetch_data_from_api, max_retries=5)
print(f"Result: {result}")
except Exception as e:
print(f"Final error: {e}")الـ AI Agent ليس مجرد برنامج يعمل في عزلة، بل يحتاج إلى التفاعل مع الأنظمة الخارجية مثل قواعد البيانات، واجهات برمجة التطبيقات، والبريد الإلكتروني. هنا تأتي أهمية تصميم واجهة برمجة تطبيقات مرنة تمكن الوكيل من تنفيذ الإجراءات المطلوبة. مثلاً، إذا كان الوكيل مسؤولاً عن إرسال تقارير عبر البريد الإلكتروني، فيجب أن يكون لديه وصول إلى واجهة برمجة تطبيقات البريد الإلكتروني مثل SendGrid أو SMTP.
في أحد المشاريع، كان الوكيل يحتاج إلى التفاعل مع عدة أنظمة مختلفة، بما في ذلك قاعدة بيانات PostgreSQL، واجهة برمجة تطبيقات دفع، ونظام إدارة المحتوى. بدلاً من كتابة كود مخصص لكل نظام، قررت استخدام نمط الـ Adapter Pattern. هذا يعني أن كل نظام له واجهة موحدة تمكن الوكيل من التفاعل معه دون الحاجة إلى معرفة التفاصيل الداخلية لكل نظام. هذا يجعل الكود أكثر مرونة ويسهل إضافة أنظمة جديدة في المستقبل.
from abc import ABC, abstractmethod
import smtplib
from email.mime.text import MIMEText
import psycopg2
class ActionAdapter(ABC):
@abstractmethod
def execute(self, **kwargs) -> str:
pass
class EmailAdapter(ActionAdapter):
def __init__(self, smtp_server: str, smtp_port: int, username: str, password: str):
self.smtp_server = smtp_server
self.smtp_port = smtp_port
self.username = username
self.password = password
def execute(self, to: str, subject: str, body: str) -> str:
msg = MIMEText(body)
msg['Subject'] = subject
msg['From'] = self.username
msg['To'] = to
with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:
server.starttls()
server.login(self.username, self.password)
server.send_message(msg)
return f"Email sent to {to}"
class DatabaseAdapter(ActionAdapter):
def __init__(self, db_connection_string: str):
self.db_c db_connection_string
def execute(self, query: str, params: tuple = None) -> str:
with psycopg2.connect(self.db_connection_string) as conn:
with conn.cursor() as cursor:
cursor.execute(query, params or ())
if cursor.description: # إذا كان الاستعلام يرجع نتائج
results = cursor.fetchall()
return str(results)
conn.commit()
return "Query executed successfully"
# مثال على استخدام Adapter Pattern
email_adapter = EmailAdapter(
smtp_server="smtp.example.com",
smtp_port=587,
username="user@example.com",
password="password"
)
db_adapter = DatabaseAdapter(db_connection_string="postgresql://user:password@localhost:5432/db")
# تنفيذ إجراءات
print(email_adapter.execute(
to="customer@example.com",
subject="Monthly Report",
body="Here is your monthly report."
))
print(db_adapter.execute(
query="SELECT * FROM customers WHERE id = %s",
params=(1,)
))الاختبار هو جزء أساسي من بناء أي نظام، والـ AI Agents ليسوا استثناءً. لكن اختبار الوكلاء المستقلين يختلف عن اختبار البرامج التقليدية. أولاً، يجب أن تختبر كل مكون على حدة، مثل نظام إدارة الـ State، محرك اتخاذ القرارات، وواجهة برمجة التطبيقات. ثانياً، يجب أن تختبر التكامل بين هذه المكونات. ثالثاً، يجب أن تختبر النظام كاملاً في بيئة مشابهة للبيئة الحقيقية.
في تجربتي، وجدت أن استخدام الـ Unit Tests التقليدي لا يكفي. تحتاج إلى كتابة اختبارات تكامل واختبارات نهاية إلى نهاية (End-to-End Tests) للتأكد من أن الوكيل يعمل كما تتوقع في السيناريوهات الحقيقية. مثلاً، يمكنك كتابة اختبار يحاكي سيناريو حيث يطلب العميل منتجاً غير متوفر، ويتأكد من أن الوكيل يتخذ الإجراءات الصحيحة مثل طلب المنتج من المورد وإرسال رسالة تأكيد للعميل.
import unittest
from unittest.mock import patch, MagicMock
class TestDecisionEngine(unittest.TestCase):
def setUp(self):
self.decisi DecisionEngine(api_key="test-key")
@patch('openai.OpenAI')
def test_make_decision_valid_action(self, mock_openai):
# إعداد الاستجابة الوهمية
mock_response = MagicMock()
mock_response.choices = [MagicMock()]
mock_response.choices[0].message.content = "order_more"
mock_openai.return_value.chat.completions.create.return_value = mock_response
state = {"inventory_level": "low"}
actions = ["order_more", "wait"]
goal = "maintain optimal inventory levels"
decision = self.decision_engine.make_decision(state, actions, goal)
self.assertEqual(decision, "order_more")
@patch('openai.OpenAI')
def test_make_decision_invalid_action(self, mock_openai):
# إعداد استجابة وهمية بقرار غير صالح
mock_response = MagicMock()
mock_response.choices = [MagicMock()]
mock_response.choices[0].message.content = "invalid_action"
mock_openai.return_value.chat.completions.create.return_value = mock_response
state = {"inventory_level": "low"}
actions = ["order_more", "wait"]
goal = "maintain optimal inventory levels"
with self.assertRaises(ValueError):
self.decision_engine.make_decision(state, actions, goal)
class TestStateManager(unittest.TestCase):
def setUp(self):
self.state_manager = StateManager(db_path=":memory:")
def test_set_and_get_state(self):
self.state_manager.set_state("test_key", "test_value")
value = self.state_manager.get_state("test_key")
self.assertEqual(value, "test_value")
def test_get_nonexistent_state(self):
value = self.state_manager.get_state("nonexistent_key")
self.assertIsNone(value)
if __name__ == "__main__":
unittest.main()المراقبة هي الجزء الآخر من المعادلة. حتى إذا كان الوكيل يعمل بشكل مثالي في بيئة الاختبار، فقد تواجه مشاكل في الإنتاج. لذلك، يجب أن يكون لديك نظام مراقبة قوي يتتبع أداء الوكيل ويسجل جميع الإجراءات التي يتخذها. استخدم أدوات مثل Prometheus وGrafana لمراقبة أداء الوكيل، وELK Stack لتسجيل وتحليل السجلات. هذا يسمح لك باكتشاف المشاكل قبل أن تؤثر على المستخدمين.
إذا كنت تريد بناء AI Agent مستقل، ابدأ بتحديد المهمة بوضوح. لا تحاول بناء وكيل يقوم بكل شيء، بل ركز على مهمة محددة وقابلة للقياس. مثلاً، بدلاً من بناء وكيل لإدارة الشركة بالكامل، ابدأ بوكيل مسؤول عن إدارة المخزون أو الرد على استفسارات العملاء. بعد ذلك، قم ببناء البنية التحتية اللازمة: نظام إدارة الـ State، محرك اتخاذ القرارات، وواجهة برمجة التطبيقات للتفاعل مع العالم الخارجي. استخدم أدوات مثل SQLite لتخزين الـ State، OpenAI API لاتخاذ القرارات، وAdapter Pattern للتكامل مع الأنظمة الخارجية.
لا تنسَ التعامل مع الأخطاء والاسترداد. استخدم آليات إعادة المحاولة مع تأخير متزايد، وسجل جميع الأخطاء لمراقبتها لاحقاً. وأخيراً، اختبر الوكيل بشكل شامل في بيئة مشابهة للبيئة الحقيقية قبل نشره في الإنتاج. تذكر أن بناء AI Agent مستقل ليس مجرد مسألة برمجة، بل هو مسألة هندسة أنظمة. عليك التفكير في كيفية إدارة الـ State، التعامل مع الـ I/O Bound Operations، والتعامل مع الأخطاء غير المتوقعة. إذا فعلت ذلك بشكل صحيح، فستكون لديك أداة قوية يمكنها العمل بشكل مستقل وتحقيق نتائج مذهلة.
الذكاء الاصطناعي ليس مجرد تقنية، بل هو شريك في اتخاذ القرارات. لكن مثل أي شريك، يحتاج إلى البنية التحتية الصحيحة ليعمل بكفاءة.
— مهندس برمجيات سنيور في Adept AI