الأنماط الناشئة

بعد آلاف عمليات نشر الوكلاء الإنتاجية في 2025، ظهرت أنماط واضحة. هذه ليست نظرية—إنها مختبرة في المعارك.

النمط 1: نقاط تحقق الإنسان في الحلقة

أفضل الوكلاء يعرفون متى يتوقفون ويسألون:

class HumanCheckpointAgent:
    def __init__(self):
        self.checkpoint_rules = [
            {"condition": "cost_exceeds", "threshold": 10.0},
            {"condition": "destructive_action", "types": ["delete", "deploy", "publish"]},
            {"condition": "confidence_below", "threshold": 0.7},
            {"condition": "external_api", "domains": ["production", "payment"]}
        ]

    async def execute_with_checkpoints(self, task: str) -> dict:
        """تنفيذ مع نقاط تحقق بشرية عند النقاط الحرجة."""
        plan = await self.create_plan(task)

        for step in plan["steps"]:
            checkpoint_needed = self.check_rules(step)

            if checkpoint_needed:
                approval = await self.request_human_approval(step)
                if not approval["approved"]:
                    return {"status": "cancelled", "reason": approval.get("reason")}

            await self.execute_step(step)

        return {"status": "completed"}

    def check_rules(self, step: dict) -> bool:
        """تحقق إذا أي قاعدة نقطة تحقق تنطبق."""
        for rule in self.checkpoint_rules:
            if rule["condition"] == "destructive_action":
                if step.get("action_type") in rule["types"]:
                    return True
            elif rule["condition"] == "confidence_below":
                if step.get("confidence", 1.0) < rule["threshold"]:
                    return True
        return False

النمط 2: التدهور السلس

عندما تفشل القدرات، تراجع بذكاء:

class GracefulAgent:
    def __init__(self):
        self.capability_fallbacks = {
            "web_search": ["cached_search", "ask_user"],
            "code_execution": ["static_analysis", "explanation_only"],
            "file_write": ["suggest_changes", "clipboard_copy"],
            "api_call": ["mock_response", "manual_instruction"]
        }

    async def execute_with_fallback(self, capability: str, task: dict) -> dict:
        """جرب القدرة مع سلسلة تراجع."""
        fallbacks = [capability] + self.capability_fallbacks.get(capability, [])

        for attempt_capability in fallbacks:
            try:
                result = await self.try_capability(attempt_capability, task)
                if result["success"]:
                    return {
                        "result": result,
                        "used_fallback": attempt_capability != capability,
                        "capability_used": attempt_capability
                    }
            except Exception as e:
                self.log(f"القدرة {attempt_capability} فشلت: {e}")
                continue

        return {
            "success": False,
            "error": f"جميع البدائل استنفدت لـ {capability}"
        }

النمط 3: ضغط السياق

أبقِ السياق صغيراً لكن مفيداً:

class ContextCompressor:
    def __init__(self, max_tokens: int = 4000):
        self.max_tokens = max_tokens

    async def compress_context(self, full_context: str) -> str:
        """ضغط السياق بذكاء ليناسب الحدود."""
        current_tokens = self.count_tokens(full_context)

        if current_tokens <= self.max_tokens:
            return full_context

        # الاستراتيجية 1: تلخيص الرسائل القديمة
        compressed = await self.summarize_history(full_context)

        # الاستراتيجية 2: الاحتفاظ بكتل الكود ذات الصلة فقط
        if self.count_tokens(compressed) > self.max_tokens:
            compressed = await self.extract_relevant_code(compressed)

        # الاستراتيجية 3: إزالة التفسيرات المطولة
        if self.count_tokens(compressed) > self.max_tokens:
            compressed = await self.strip_explanations(compressed)

        return compressed

    async def summarize_history(self, context: str) -> str:
        """تلخيص الأجزاء القديمة من المحادثة."""
        response = await llm.chat(
            model="claude-3-haiku-20240307",  # ملخص سريع
            messages=[{
                "role": "user",
                "content": f"""لخص تاريخ المحادثة هذا، محتفظاً بـ:
- القرارات الرئيسية المتخذة
- حالة المهمة الحالية
- مسارات وأسماء الملفات المهمة
- رسائل الخطأ المواجهة

السياق الكامل:
{context[:10000]}

أرجع ملخصاً موجزاً."""
            }]
        )
        return response.content

النمط 4: التنفيذ التخميني

قم بالعمل المحتمل الحاجة أثناء الانتظار:

class SpeculativeAgent:
    def __init__(self):
        self.speculation_cache = {}

    async def execute_speculatively(self, task: str) -> dict:
        """ابدأ المهام التالية المحتملة مبكراً."""

        # بدء المهمة الرئيسية
        main_task = asyncio.create_task(self.execute_task(task))

        # توقع وبدء الخطوات التالية المحتملة
        predictions = await self.predict_followups(task)
        speculative_tasks = [
            asyncio.create_task(self.speculative_execute(p))
            for p in predictions[:3]  # أعلى 3 توقعات
        ]

        # انتظار المهمة الرئيسية
        main_result = await main_task

        # تحقق إذا كان أي تخمين مفيداً
        for pred, spec_task in zip(predictions[:3], speculative_tasks):
            result = await spec_task
            self.speculation_cache[pred["task"]] = result

        return main_result

    async def predict_followups(self, task: str) -> list[dict]:
        """توقع المهام التالية المحتملة."""
        response = await llm.chat(
            model="claude-3-haiku-20240307",
            messages=[{
                "role": "user",
                "content": f"""بالنظر لهذه المهمة، توقع 3 مهام تالية محتملة:
المهمة: {task}

أرجع مصفوفة JSON: [{{"task": "...", "probability": 0.X}}]"""
            }]
        )
        return json.loads(response.content)

النمط 5: تخزين نتائج الأدوات مؤقتاً

لا تكرر العمليات المكلفة:

class CachingToolExecutor:
    def __init__(self, cache_ttl: int = 300):
        self.cache = {}
        self.cache_ttl = cache_ttl

    def get_cache_key(self, tool: str, args: dict) -> str:
        """توليد مفتاح كاش لاستدعاء الأداة."""
        # تخزين الأدوات الحتمية فقط
        cacheable = ["read_file", "search", "list_directory", "get_config"]
        if tool not in cacheable:
            return None
        return f"{tool}:{hashlib.md5(json.dumps(args, sort_keys=True).encode()).hexdigest()}"

    async def execute_tool(self, tool: str, args: dict) -> dict:
        """تنفيذ الأداة مع التخزين المؤقت."""
        cache_key = self.get_cache_key(tool, args)

        if cache_key and cache_key in self.cache:
            entry = self.cache[cache_key]
            if time.time() - entry["timestamp"] < self.cache_ttl:
                return {"result": entry["result"], "cached": True}

        result = await self.actual_execute(tool, args)

        if cache_key:
            self.cache[cache_key] = {
                "result": result,
                "timestamp": time.time()
            }

        return {"result": result, "cached": False}

النمط 6: التنفيذ المتوازي للأدوات

شغّل الأدوات المستقلة بالتزامن:

class ParallelToolExecutor:
    async def execute_tool_batch(self, tool_calls: list[dict]) -> list[dict]:
        """تنفيذ استدعاءات الأدوات المستقلة بالتوازي."""

        # تحليل التبعيات
        independent, dependent = self.analyze_dependencies(tool_calls)

        # تشغيل الاستدعاءات المستقلة بالتوازي
        results = {}
        if independent:
            parallel_results = await asyncio.gather(*[
                self.execute_tool(tc["tool"], tc["args"])
                for tc in independent
            ])
            for tc, result in zip(independent, parallel_results):
                results[tc["id"]] = result

        # تشغيل الاستدعاءات التابعة بالتسلسل
        for tc in dependent:
            # استبدال النتائج من التبعيات
            resolved_args = self.resolve_args(tc["args"], results)
            results[tc["id"]] = await self.execute_tool(tc["tool"], resolved_args)

        return [results[tc["id"]] for tc in tool_calls]

ملاحظة نيردية: هذه الأنماط ظهرت من الألم. كل واحد يمثل آلاف التشغيلات الفاشلة قبل أن يكتشف أحدهم الحل. تعلم من أخطائنا.

التالي: موارد لمواصلة رحلتك. :::