التحميل الديناميكي للقدرات

قوائم الأدوات الثابتة لا تتوسع. وكيلك يحتاج قدرات مختلفة لـ "حلل هذا CSV" مقابل "انشر للإنتاج." التحميل الديناميكي يعطي الوكلاء الأدوات الصحيحة في الوقت الصحيح.

استراتيجية التحميل

class DynamicSkillLoader:
    def __init__(self, skills_dir: str = "skills"):
        self.skills_dir = Path(skills_dir)
        self.skill_registry = {}
        self.loaded_tools = {}
        self.scan_skills()

    def scan_skills(self):
        """بناء سجل المهارات المتاحة."""
        for skill_path in self.skills_dir.glob("*/SKILL.md"):
            skill_name = skill_path.parent.name
            self.skill_registry[skill_name] = {
                "path": skill_path.parent,
                "manifest": self.parse_manifest(skill_path),
                "loaded": False
            }

    def load_skill(self, skill_name: str) -> list[dict]:
        """استيراد وتهيئة مهارة ديناميكياً."""
        if skill_name not in self.skill_registry:
            raise ValueError(f"مهارة غير معروفة: {skill_name}")

        skill = self.skill_registry[skill_name]
        if skill["loaded"]:
            return self.loaded_tools[skill_name]

        # استيراد وحدة المهارة
        module_path = skill["path"] / "tools.py"
        spec = importlib.util.spec_from_file_location(skill_name, module_path)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)

        # استخراج تعريفات الأدوات
        tools = module.get_tools()
        self.loaded_tools[skill_name] = tools
        skill["loaded"] = True

        return tools

اكتشاف الأدوات المبني على النية

دع النموذج يقرر ما يحتاجه:

async def discover_tools_for_intent(user_message: str, loader: DynamicSkillLoader) -> list[dict]:
    """استخدم نموذجاً صغيراً لتحديد المهارات المطلوبة."""

    # بناء ملخص المهارات للمصنف
    skill_summaries = "\n".join([
        f"- {name}: {info['manifest']['description']}"
        for name, info in loader.skill_registry.items()
    ])

    response = await llm.chat(
        model="claude-3-haiku-20240307",  # مصنف سريع ورخيص
        messages=[{
            "role": "user",
            "content": f"""بالنظر لطلب المستخدم هذا، أي مهارات مطلوبة؟

طلب المستخدم: {user_message}

المهارات المتاحة:
{skill_summaries}

أعد مصفوفة JSON لأسماء المهارات. أدرج فقط المهارات ذات الصلة المباشرة بالمهمة.
مثال: ["filesystem", "web"]"""
        }]
    )

    needed_skills = json.loads(response.content)

    # تحميل المهارات المطلوبة فقط
    tools = []
    for skill_name in needed_skills:
        tools.extend(loader.load_skill(skill_name))

    return tools

التحميل الكسول مع التخزين المؤقت

class CachedSkillLoader:
    def __init__(self):
        self.loader = DynamicSkillLoader()
        self.tool_cache = {}
        self.cache_ttl = 300  # 5 دقائق

    async def get_tools(self, user_message: str) -> list[dict]:
        """الحصول على الأدوات مع التخزين المؤقت للأنماط المتكررة."""
        cache_key = self.compute_intent_hash(user_message)

        if cache_key in self.tool_cache:
            entry = self.tool_cache[cache_key]
            if time.time() - entry["timestamp"] < self.cache_ttl:
                return entry["tools"]

        tools = await discover_tools_for_intent(user_message, self.loader)
        self.tool_cache[cache_key] = {
            "tools": tools,
            "timestamp": time.time()
        }

        return tools

    def compute_intent_hash(self, message: str) -> str:
        """تجزئة مبنية على كلمات النية الرئيسية، ليس الرسالة الدقيقة."""
        # استخراج كلمات النية المفتاحية لمطابقة الكاش
        keywords = set(re.findall(r'\b(file|web|code|email|database)\b', message.lower()))
        return hashlib.md5(",".join(sorted(keywords)).encode()).hexdigest()

التحميل الساخن لخوادم MCP

الاتصال بخوادم MCP عند الطلب:

class MCPDynamicConnector:
    def __init__(self, config_path: str):
        self.config = json.load(open(config_path))
        self.active_connections = {}

    async def connect_server(self, server_name: str):
        """بدء اتصال خادم MCP عند الطلب."""
        if server_name in self.active_connections:
            return self.active_connections[server_name]

        server_config = self.config["mcpServers"][server_name]

        # بدء عملية الخادم
        process = await asyncio.create_subprocess_exec(
            server_config["command"],
            *server_config.get("args", []),
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE
        )

        # تهيئة اتصال MCP
        connection = MCPConnection(process)
        await connection.initialize()

        self.active_connections[server_name] = connection
        return connection

    async def get_tools_from_server(self, server_name: str) -> list[dict]:
        """جلب الأدوات من خادم MCP محدد."""
        connection = await self.connect_server(server_name)
        return await connection.list_tools()

    async def disconnect_idle_servers(self, idle_threshold: int = 300):
        """تنظيف الخوادم غير المستخدمة مؤخراً."""
        now = time.time()
        for name, conn in list(self.active_connections.items()):
            if now - conn.last_used > idle_threshold:
                await conn.close()
                del self.active_connections[name]

خط الأنابيب الكامل

async def process_with_dynamic_tools(user_message: str):
    """التدفق الكامل: اكتشف → حمّل → نفّذ."""

    # 1. تصنيف النية واكتشاف الأدوات المطلوبة
    tools = await cached_loader.get_tools(user_message)

    # 2. تحقق إذا كانت خوادم MCP مطلوبة
    if needs_external_capability(user_message):
        mcp_tools = await mcp_connector.get_tools_from_server("relevant-server")
        tools.extend(mcp_tools)

    # 3. استدعاء LLM بمجموعة أدوات مجمعة ديناميكياً
    response = await llm.chat(
        messages=[{"role": "user", "content": user_message}],
        tools=tools
    )

    # 4. تنفيذ استدعاءات الأدوات
    while response.tool_calls:
        results = await execute_tools(response.tool_calls)
        response = await llm.chat(
            messages=[...],  # تضمين نتائج الأدوات
            tools=tools
        )

    # 5. تنظيف الاتصالات الخاملة
    await mcp_connector.disconnect_idle_servers()

    return response.content

ملاحظة نيردية: هكذا يعمل Claude Code. لا يحمل كل الـ 50+ أداة دفعة واحدة—ينشط ما هو مطلوب بناءً على طلبك.

الوحدة التالية: أخذ الوكلاء للإنتاج. :::