تصميم سجل الأدوات

سجل الأدوات هو نظام مركزي لإدارة واكتشاف وتوفير الأدوات للوكلاء. تصميم السجل الجيد يمكّن الوكلاء من اختيار الأدوات المناسبة ديناميكياً لأي مهمة.

لماذا سجل الأدوات؟

المشكلة	كيف يحلها السجل
أدوات كثيرة للسياق	البحث الدلالي عن الأدوات ذات الصلة
الأدوات تتغير بشكل متكرر	تسجيل ديناميكي، بدون إعادة تشغيل الوكيل
إصدار الأدوات	إصدارات متعددة، ترحيل تدريجي
التحكم بالوصول	أذونات الأدوات لكل وكيل

هندسة السجل الأساسية

from dataclasses import dataclass
from typing import Callable, Optional
import json

@dataclass
class Tool:
    name: str
    description: str
    parameters: dict  # مخطط JSON
    handler: Callable
    version: str = "1.0.0"
    category: str = "general"
    embedding: Optional[list] = None  # للبحث الدلالي

class ToolRegistry:
    def __init__(self, embedding_model):
        self.tools = {}
        self.embedder = embedding_model
        self.embeddings = {}  # الاسم -> التضمين

    async def register(self, tool: Tool):
        """تسجيل أداة مع تضمين دلالي."""
        # توليد التضمين للاكتشاف الدلالي
        description_text = f"{tool.name}: {tool.description}"
        tool.embedding = await self.embedder.embed(description_text)

        self.tools[tool.name] = tool
        self.embeddings[tool.name] = tool.embedding

    def get(self, name: str) -> Optional[Tool]:
        """الحصول على أداة بالاسم الدقيق."""
        return self.tools.get(name)

    async def search(self, query: str, top_k: int = 5) -> list:
        """البحث الدلالي عن الأدوات ذات الصلة."""
        query_embedding = await self.embedder.embed(query)

        scores = []
        for name, embedding in self.embeddings.items():
            score = self._cosine_similarity(query_embedding, embedding)
            scores.append((name, score))

        # الترتيب حسب التشابه
        scores.sort(key=lambda x: x[1], reverse=True)

        return [self.tools[name] for name, _ in scores[:top_k]]

    def list_by_category(self, category: str) -> list:
        """عرض الأدوات حسب الفئة."""
        return [t for t in self.tools.values() if t.category == category]

    def to_openai_format(self, tools: list) -> list:
        """تحويل الأدوات إلى صيغة استدعاء دوال OpenAI."""
        return [
            {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.parameters
                }
            }
            for tool in tools
        ]

اختيار الأدوات الديناميكي

اختيار الأدوات ذات الصلة بناءً على استعلام المستخدم:

class SmartAgent:
    def __init__(self, llm, registry: ToolRegistry, max_tools: int = 10):
        self.llm = llm
        self.registry = registry
        self.max_tools = max_tools

    async def run(self, task: str) -> str:
        # الخطوة 1: إيجاد الأدوات ذات الصلة لهذه المهمة
        relevant_tools = await self.registry.search(task, top_k=self.max_tools)

        # الخطوة 2: تشغيل الوكيل مع الأدوات المختارة
        tools_formatted = self.registry.to_openai_format(relevant_tools)

        messages = [{"role": "user", "content": task}]

        while True:
            response = await self.llm.complete(
                messages=messages,
                tools=tools_formatted
            )

            if response.tool_calls:
                for call in response.tool_calls:
                    tool = self.registry.get(call.function.name)
                    result = await tool.handler(**json.loads(call.function.arguments))
                    messages.append({
                        "role": "tool",
                        "content": str(result),
                        "tool_call_id": call.id
                    })
            else:
                return response.content

إصدار الأدوات

دعم إصدارات متعددة للترحيل التدريجي:

class VersionedToolRegistry(ToolRegistry):
    def __init__(self, embedding_model):
        super().__init__(embedding_model)
        self.versions = {}  # الاسم -> {الإصدار -> الأداة}

    async def register(self, tool: Tool):
        """التسجيل مع تتبع الإصدارات."""
        if tool.name not in self.versions:
            self.versions[tool.name] = {}

        self.versions[tool.name][tool.version] = tool

        # الإصدار الأخير هو الافتراضي
        await super().register(tool)

    def get(self, name: str, version: str = "latest") -> Optional[Tool]:
        """الحصول على إصدار محدد من أداة."""
        if name not in self.versions:
            return None

        if version == "latest":
            return self.tools.get(name)

        return self.versions[name].get(version)

    def deprecate(self, name: str, version: str, replacement_version: str):
        """وسم إصدار كمهمل."""
        if name in self.versions and version in self.versions[name]:
            tool = self.versions[name][version]
            tool.deprecated = True
            tool.replacement = replacement_version

فئات الأدوات والأذونات

@dataclass
class ToolPermission:
    tool_name: str
    allowed_agents: list  # معرفات الوكلاء الذين يمكنهم استخدام هذه الأداة
    rate_limit: Optional[int] = None  # الاستدعاءات في الدقيقة
    requires_approval: bool = False

class SecureToolRegistry(ToolRegistry):
    def __init__(self, embedding_model):
        super().__init__(embedding_model)
        self.permissions = {}

    def set_permission(self, permission: ToolPermission):
        self.permissions[permission.tool_name] = permission

    async def get_for_agent(self, agent_id: str, query: str) -> list:
        """الحصول على الأدوات المسموح للوكيل باستخدامها."""
        # الحصول على الأدوات ذات الصلة
        relevant = await self.search(query)

        # التصفية حسب الأذونات
        allowed = []
        for tool in relevant:
            perm = self.permissions.get(tool.name)
            if perm is None or agent_id in perm.allowed_agents:
                allowed.append(tool)

        return allowed

تركيب الأدوات

بناء أدوات معقدة من أدوات أبسط:

class ComposedTool:
    """أداة تجمع أدوات متعددة."""

    def __init__(self, name: str, steps: list):
        self.name = name
        self.steps = steps  # قائمة من (اسم_الأداة، تحويل_المدخل)

    async def execute(self, registry: ToolRegistry, initial_input: dict) -> str:
        current_result = initial_input

        for tool_name, transform in self.steps:
            tool = registry.get(tool_name)
            # تحويل النتيجة السابقة إلى مدخل الأداة
            tool_input = transform(current_result)
            current_result = await tool.handler(**tool_input)

        return current_result

# مثال: أداة مركبة "بحث_ولخص"
research_summarize = ComposedTool(
    name="research_and_summarize",
    steps=[
        ("web_search", lambda x: {"query": x["topic"]}),
        ("summarize", lambda x: {"text": x["results"]})
    ]
)

نصيحة للمقابلة

عند مناقشة سجلات الأدوات، ركّز على:

1. الاكتشاف الدلالي - الوكلاء لا يحتاجون معرفة جميع الأدوات
2. قابلية التوسع - التعامل مع مئات الأدوات بكفاءة
3. الأمان - ليس كل الوكلاء يجب أن يصلوا لكل الأدوات
4. التطور - الأدوات تتغير، الوكلاء يجب ألا ينكسروا

بعد ذلك، سنستكشف إدارة الحالة والذاكرة للوكلاء. :::