أنظمة وكيل-إلى-وكيل البيئية

المستقبل ليس وكيلاً خارقاً واحداً. إنها شبكات من الوكلاء المتخصصين الذين يكتشفون، يفاوضون، ويتعاونون ذاتياً.

رؤية النظام البيئي

┌─────────────────────────────────────────────────────────┐
│                    سوق الوكلاء                          │
├─────────────────────────────────────────────────────────┤
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐│
│  │  وكيل   │  │  وكيل   │  │  وكيل   │  │  وكيل   ││
│  │  الكود   │  │  البحث   │  │ البيانات│  │  الأمان  ││
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘│
│       │             │             │             │       │
│       └─────────────┴─────────────┴─────────────┘       │
│                         │                               │
│              ┌──────────┴──────────┐                   │
│              │    طبقة الاكتشاف   │                   │
│              │   (سجل الوكلاء)   │                   │
│              └─────────────────────┘                   │
└─────────────────────────────────────────────────────────┘

بروتوكول اكتشاف الوكلاء

الوكلاء يعلنون عن قدراتهم ويجدون متعاونين:

from dataclasses import dataclass

@dataclass
class AgentCapability:
    name: str
    description: str
    input_schema: dict
    output_schema: dict
    cost_per_call: float
    avg_latency_ms: int
    success_rate: float

class AgentRegistry:
    def __init__(self):
        self.agents = {}  # agent_id -> AgentInfo
        self.capabilities = {}  # capability_name -> [agent_ids]

    def register_agent(self, agent_id: str, capabilities: list[AgentCapability]):
        """تسجيل وكيل وقدراته."""
        self.agents[agent_id] = {
            "capabilities": capabilities,
            "registered_at": datetime.now(),
            "status": "active"
        }

        for cap in capabilities:
            self.capabilities.setdefault(cap.name, []).append(agent_id)

    def discover_agents(self, capability_needed: str, constraints: dict = None) -> list[str]:
        """إيجاد وكلاء يمكنهم تلبية قدرة."""
        candidates = self.capabilities.get(capability_needed, [])

        if constraints:
            # تصفية حسب القيود
            filtered = []
            for agent_id in candidates:
                agent = self.agents[agent_id]
                cap = next(c for c in agent["capabilities"] if c.name == capability_needed)

                if constraints.get("max_cost") and cap.cost_per_call > constraints["max_cost"]:
                    continue
                if constraints.get("max_latency") and cap.avg_latency_ms > constraints["max_latency"]:
                    continue
                if constraints.get("min_success") and cap.success_rate < constraints["min_success"]:
                    continue

                filtered.append(agent_id)
            return filtered

        return candidates

بروتوكول تواصل الوكلاء

رسائل موحدة لتعاون الوكلاء:

@dataclass
class AgentMessage:
    sender: str
    recipient: str
    message_type: str  # "request"، "response"، "negotiate"، "delegate"
    payload: dict
    correlation_id: str
    timestamp: datetime

class AgentCommunicator:
    def __init__(self, agent_id: str, registry: AgentRegistry):
        self.agent_id = agent_id
        self.registry = registry
        self.pending_requests = {}

    async def request_collaboration(
        self,
        capability: str,
        task: dict,
        budget: float = None
    ) -> dict:
        """طلب مساعدة من وكيل آخر."""

        # إيجاد وكلاء قادرين
        candidates = self.registry.discover_agents(
            capability,
            constraints={"max_cost": budget} if budget else None
        )

        if not candidates:
            return {"error": f"لم يُعثر على وكلاء للقدرة: {capability}"}

        # اختيار أفضل مرشح (يمكن أن يكون أكثر تعقيداً)
        target_agent = candidates[0]

        # إرسال الطلب
        message = AgentMessage(
            sender=self.agent_id,
            recipient=target_agent,
            message_type="request",
            payload={"capability": capability, "task": task, "budget": budget},
            correlation_id=str(uuid.uuid4()),
            timestamp=datetime.now()
        )

        response = await self.send_and_wait(message)
        return response

    async def handle_request(self, message: AgentMessage) -> AgentMessage:
        """التعامل مع طلب تعاون وارد."""
        task = message.payload["task"]
        budget = message.payload.get("budget")

        # تقدير التكلفة
        estimated_cost = self.estimate_task_cost(task)

        if budget and estimated_cost > budget:
            # تفاوض
            return AgentMessage(
                sender=self.agent_id,
                recipient=message.sender,
                message_type="negotiate",
                payload={
                    "required_budget": estimated_cost,
                    "reason": "تعقيد المهمة يتجاوز الميزانية المعروضة"
                },
                correlation_id=message.correlation_id,
                timestamp=datetime.now()
            )

        # تنفيذ والرد
        result = await self.execute_task(task)
        return AgentMessage(
            sender=self.agent_id,
            recipient=message.sender,
            message_type="response",
            payload={"result": result, "cost": estimated_cost},
            correlation_id=message.correlation_id,
            timestamp=datetime.now()
        )

سلاسل التفويض

وكلاء يفوضون المهام الفرعية للمتخصصين:

class DelegatingAgent:
    def __init__(self, agent_id: str, communicator: AgentCommunicator):
        self.agent_id = agent_id
        self.communicator = communicator

    async def execute_complex_task(self, task: dict) -> dict:
        """تقسيم وتفويض المهام المعقدة."""

        # تحليل المهمة وتحديد المهام الفرعية
        subtasks = await self.decompose_task(task)

        results = {}
        for subtask in subtasks:
            if self.can_handle(subtask):
                # التعامل محلياً
                results[subtask["id"]] = await self.handle_locally(subtask)
            else:
                # التفويض لمتخصص
                capability_needed = subtask["required_capability"]
                result = await self.communicator.request_collaboration(
                    capability=capability_needed,
                    task=subtask,
                    budget=subtask.get("budget")
                )
                results[subtask["id"]] = result

        # دمج النتائج
        return await self.combine_results(task, results)

الثقة والسمعة

تتبع موثوقية الوكيل عبر الوقت:

class ReputationSystem:
    def __init__(self):
        self.interactions = {}  # agent_id -> [سجلات التفاعل]
        self.reputation_scores = {}

    def record_interaction(
        self,
        agent_id: str,
        success: bool,
        response_time_ms: int,
        cost_accuracy: float  # الفعلي مقابل المعروض
    ):
        """تسجيل تفاعل مع وكيل."""
        self.interactions.setdefault(agent_id, []).append({
            "timestamp": datetime.now(),
            "success": success,
            "response_time_ms": response_time_ms,
            "cost_accuracy": cost_accuracy
        })
        self.update_reputation(agent_id)

    def update_reputation(self, agent_id: str):
        """حساب درجة السمعة من التفاعلات الأخيرة."""
        recent = self.interactions[agent_id][-50:]  # آخر 50

        if len(recent) < 5:
            self.reputation_scores[agent_id] = 0.5  # محايد للوكلاء الجدد
            return

        success_rate = sum(1 for i in recent if i["success"]) / len(recent)
        avg_cost_accuracy = sum(i["cost_accuracy"] for i in recent) / len(recent)

        # درجة موزونة
        self.reputation_scores[agent_id] = (
            0.7 * success_rate +
            0.3 * avg_cost_accuracy
        )

    def get_trusted_agents(self, capability: str, min_reputation: float = 0.8) -> list[str]:
        """الحصول على وكلاء عاليي التصنيف لقدرة."""
        candidates = registry.discover_agents(capability)
        return [
            a for a in candidates
            if self.reputation_scores.get(a, 0.5) >= min_reputation
        ]

ملاحظة نيردية: أنظمة الوكلاء البيئية تحتاج حوافز اقتصادية لتعمل. لماذا سيساعد وكيل وكيلاً آخر؟ مكافآت رمزية، مكاسب سمعة، خدمات متبادلة. نظرية اللعبة مهمة.

التالي: أنماط ناشئة من عمليات النشر الإنتاجية. :::