الخطوات التالية

مبروك! لقد بنيت وكيل بحث فعال. إليك كيفية توسيعه ومواصلة رحلة التعلم.

تحسينات للتجربة

1. إضافة المزيد من الأدوات

# tools/scraper.py
import requests
from bs4 import BeautifulSoup

class WebScraperTool(BaseTool):
    name = "web_scraper"
    description = "استخراج المحتوى الكامل من URL محدد"

    def run(self, url: str) -> dict:
        try:
            response = requests.get(url, timeout=10)
            soup = BeautifulSoup(response.text, 'html.parser')

            # إزالة السكربتات والأنماط
            for tag in soup(['script', 'style', 'nav', 'footer']):
                tag.decompose()

            text = soup.get_text(separator=' ', strip=True)

            return {
                "success": True,
                "content": text[:Config.MAX_CONTENT_LENGTH],
                "url": url
            }
        except Exception as e:
            return {"success": False, "error": str(e)}

2. تنفيذ التخزين المؤقت

# utils/cache.py
import hashlib
import json
from pathlib import Path
from datetime import datetime, timedelta

class SearchCache:
    def __init__(self, cache_dir: str = ".cache", ttl_hours: int = 24):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)
        self.ttl = timedelta(hours=ttl_hours)

    def _get_key(self, query: str) -> str:
        return hashlib.md5(query.encode()).hexdigest()

    def get(self, query: str) -> dict | None:
        key = self._get_key(query)
        cache_file = self.cache_dir / f"{key}.json"

        if cache_file.exists():
            data = json.loads(cache_file.read_text())
            cached_time = datetime.fromisoformat(data["timestamp"])

            if datetime.now() - cached_time < self.ttl:
                return data["results"]

        return None

    def set(self, query: str, results: dict):
        key = self._get_key(query)
        cache_file = self.cache_dir / f"{key}.json"

        data = {
            "timestamp": datetime.now().isoformat(),
            "query": query,
            "results": results
        }
        cache_file.write_text(json.dumps(data))

3. إضافة الإخراج المتدفق

# agent.py (محسّن)
async def research_stream(self, topic: str):
    """بث تقدم البحث للمستخدم"""
    yield f"بدء البحث عن: {topic}\n"

    self.memory = ResearchMemory()
    iteration = 0

    async for step in self._research_loop_async(topic):
        iteration += 1
        yield f"\n[خطوة {iteration}] {step['type']}\n"

        if step['type'] == 'search':
            yield f"  جاري البحث: {step['query']}\n"
            yield f"  وُجد {len(step['results'])} نتائج\n"
        elif step['type'] == 'thinking':
            yield f"  {step['thought'][:100]}...\n"

    yield "\n[جاري تجميع التقرير...]\n"
    report = await self._synthesize_report_async(topic)

    yield "\n" + "="*50 + "\n"
    yield report

تحسينات البنية

الحالي	المحسّن
LLM واحد	LLM أساسي + احتياطي
تنفيذ متزامن	غير متزامن مع بحث متوازي
الذاكرة في RAM	مخزن متجهات دائم
بحث واحد	تجميع من مصادر متعددة

اعتبارات الإنتاج

المراقبة

# utils/monitoring.py
from dataclasses import dataclass
import time

@dataclass
class AgentMetrics:
    total_searches: int = 0
    total_tokens: int = 0
    avg_response_time: float = 0
    error_count: int = 0

class MetricsCollector:
    def __init__(self):
        self.metrics = AgentMetrics()
        self.response_times = []

    def record_search(self):
        self.metrics.total_searches += 1

    def record_tokens(self, count: int):
        self.metrics.total_tokens += count

    def record_response_time(self, duration: float):
        self.response_times.append(duration)
        self.metrics.avg_response_time = sum(self.response_times) / len(self.response_times)

    def record_error(self):
        self.metrics.error_count += 1

تحديد المعدل

# utils/rate_limiter.py
import time
from collections import deque

class RateLimiter:
    def __init__(self, max_requests: int, time_window: int):
        self.max_requests = max_requests
        self.time_window = time_window  # ثواني
        self.requests = deque()

    def acquire(self) -> bool:
        now = time.time()

        # إزالة الطلبات القديمة
        while self.requests and self.requests[0] < now - self.time_window:
            self.requests.popleft()

        if len(self.requests) < self.max_requests:
            self.requests.append(now)
            return True

        return False

    def wait(self):
        while not self.acquire():
            time.sleep(0.1)

مواصلة التعلم

مواضيع متقدمة

• أنظمة متعددة الوكلاء: بناء فرق من الوكلاء المتخصصين
• الإنسان في الحلقة: إضافة خطوات موافقة للإجراءات الحرجة
• الضبط الدقيق: تدريب نماذج لمجالات بحث محددة
• أطر التقييم: بناء تقييم جودة منهجي

الموارد

المورد	التركيز
وثائق LangChain	الغوص العميق في الإطار
LangSmith	مراقبة الإنتاج
كتاب Anthropic	أفضل ممارسات Claude
دليل OpenAI	تحسين GPT

أفكار مشاريع

1. وكيل خبير مجال: وكيل بحث متخصص لمجال معين (قانوني، طبي، تقني)
2. وكيل مقارنة: بحث يقارن خيارات متعددة
3. محلل اتجاهات: تتبع والإبلاغ عن المواضيع الناشئة
4. مدقق حقائق: التحقق من الادعاءات مقابل مصادر موثوقة

ملخص الدورة

لقد تعلمت:

• أنماط الوكلاء: ReAct، استخدام الأدوات، التخطيط، سير العمل متعدد الخطوات
• الأطر: LangChain، CrewAI، OpenAI Agents SDK
• أنظمة الذاكرة: إدارة السياق، RAG، الذاكرة قصيرة/طويلة المدى
• معالجة الأخطاء: التدهور السلس، التحقق، التصحيح
• بناء الوكلاء: التنفيذ الكامل من الإعداد إلى الاختبار

أكمل الاختبار النهائي للحصول على شارة الدورة! :::