إدارة الحالة والذاكرة

الوكلاء يحتاجون الذاكرة للحفاظ على السياق عبر التفاعلات. تصميم إدارة حالة فعالة أمر حاسم لبناء أنظمة ذكاء اصطناعي موثوقة ومتسقة.

أنواع ذاكرة الوكيل

┌─────────────────────────────────────────────────────────────┐
│                    أنواع ذاكرة الوكيل                        │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  قصيرة المدى                طويلة المدى                     │
│  ──────────────────────     ──────────────────────────────  │
│  • المحادثة الحالية         • تفضيلات المستخدم              │
│  • سياق العمل              • ملخص التفاعلات السابقة        │
│  • نتائج استدعاء الأدوات   • الحقائق المتعلمة               │
│  • الحالة الجارية          • المعرفة المستمرة               │
│                                                              │
│  حلقية                      دلالية                          │
│  ──────────────────────     ──────────────────────────────  │
│  • ما حدث                   • ما هو صحيح                    │
│  • أحداث محددة              • معرفة عامة                    │
│  • مؤرخة                    • خالدة                         │
│                                                              │
└─────────────────────────────────────────────────────────────┘

الذاكرة قصيرة المدى

سياق المحادثة ضمن جلسة واحدة:

class ShortTermMemory:
    def __init__(self, max_tokens: int = 4000):
        self.messages = []
        self.max_tokens = max_tokens

    def add(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        self._trim_if_needed()

    def get_context(self) -> list:
        return self.messages.copy()

    def _trim_if_needed(self):
        """الاحتفاظ بالذاكرة ضمن حد الرموز."""
        while self._estimate_tokens() > self.max_tokens and len(self.messages) > 2:
            # إزالة الرسائل الأقدم (الاحتفاظ بمطالبة النظام)
            self.messages.pop(1)

    def _estimate_tokens(self) -> int:
        return sum(len(m["content"]) // 4 for m in self.messages)

    def summarize_and_compress(self, llm) -> str:
        """تلخيص السياق القديم لتوفير الرموز."""
        if len(self.messages) < 10:
            return

        # الاحتفاظ بالرسائل الحديثة
        recent = self.messages[-5:]
        old = self.messages[:-5]

        # تلخيص الرسائل القديمة
        summary = llm.complete(
            f"لخّص هذه المحادثة باختصار:\n"
            f"{json.dumps(old, indent=2)}"
        )

        # استبدال القديم بالملخص
        self.messages = [
            {"role": "system", "content": f"السياق السابق: {summary}"}
        ] + recent

الذاكرة طويلة المدى

التخزين المستمر عبر الجلسات:

class LongTermMemory:
    def __init__(self, vector_store, embedding_model):
        self.store = vector_store
        self.embedder = embedding_model

    async def store_memory(
        self,
        user_id: str,
        content: str,
        memory_type: str,  # "fact", "preference", "episode"
        metadata: dict = None
    ):
        """تخزين ذاكرة للاسترجاع لاحقاً."""
        embedding = await self.embedder.embed(content)

        await self.store.insert({
            "embedding": embedding,
            "content": content,
            "user_id": user_id,
            "memory_type": memory_type,
            "timestamp": datetime.utcnow().isoformat(),
            "metadata": metadata or {}
        })

    async def recall(
        self,
        user_id: str,
        query: str,
        memory_types: list = None,
        top_k: int = 5
    ) -> list:
        """استدعاء الذكريات ذات الصلة."""
        query_embedding = await self.embedder.embed(query)

        filters = {"user_id": user_id}
        if memory_types:
            filters["memory_type"] = {"$in": memory_types}

        results = await self.store.search(
            embedding=query_embedding,
            filter=filters,
            top_k=top_k
        )

        return [r["content"] for r in results]

    async def update_memory(self, memory_id: str, new_content: str):
        """تحديث ذاكرة موجودة."""
        new_embedding = await self.embedder.embed(new_content)
        await self.store.update(
            memory_id,
            {"content": new_content, "embedding": new_embedding}
        )

نمط الذاكرة العاملة

دمج قصيرة وطويلة المدى للسياق الكامل:

class WorkingMemory:
    def __init__(self, short_term, long_term, max_context_tokens: int = 6000):
        self.short_term = short_term
        self.long_term = long_term
        self.max_tokens = max_context_tokens

    async def build_context(self, user_id: str, current_query: str) -> list:
        """بناء السياق الكامل لـ LLM."""
        context = []

        # 1. استدعاء الذكريات طويلة المدى ذات الصلة
        memories = await self.long_term.recall(user_id, current_query)
        if memories:
            memory_text = "\n".join(memories)
            context.append({
                "role": "system",
                "content": f"ذكريات ذات صلة عن هذا المستخدم:\n{memory_text}"
            })

        # 2. إضافة سجل المحادثة
        context.extend(self.short_term.get_context())

        # 3. إضافة الاستعلام الحالي
        context.append({"role": "user", "content": current_query})

        return self._fit_to_token_limit(context)

    def _fit_to_token_limit(self, context: list) -> list:
        """تقليم السياق ليناسب حد الرموز."""
        while self._estimate_tokens(context) > self.max_tokens:
            # إزالة أقدم رسالة محادثة (الاحتفاظ بالنظام + الذكريات)
            for i, msg in enumerate(context):
                if msg["role"] != "system":
                    context.pop(i)
                    break
        return context

إدارة الحالة الموزعة

لأنظمة الوكلاء المتعددين مع حالة مشتركة:

class DistributedAgentState:
    def __init__(self, redis_client):
        self.redis = redis_client

    async def get_state(self, agent_id: str, session_id: str) -> dict:
        """الحصول على حالة الوكيل لجلسة."""
        key = f"agent:{agent_id}:session:{session_id}"
        data = await self.redis.get(key)
        return json.loads(data) if data else {}

    async def set_state(self, agent_id: str, session_id: str, state: dict):
        """تحديث حالة الوكيل."""
        key = f"agent:{agent_id}:session:{session_id}"
        await self.redis.setex(key, 3600, json.dumps(state))  # TTL ساعة واحدة

    async def share_state(
        self,
        from_agent: str,
        to_agent: str,
        session_id: str,
        data: dict
    ):
        """مشاركة الحالة بين الوكلاء."""
        key = f"shared:{from_agent}:{to_agent}:{session_id}"
        await self.redis.setex(key, 600, json.dumps(data))  # TTL 10 دقائق

    async def get_shared_state(
        self,
        from_agent: str,
        to_agent: str,
        session_id: str
    ) -> dict:
        """استرجاع الحالة المشتركة."""
        key = f"shared:{from_agent}:{to_agent}:{session_id}"
        data = await self.redis.get(key)
        return json.loads(data) if data else {}

توحيد الذاكرة

تحويل قصيرة المدى إلى طويلة المدى بشكل دوري:

class MemoryConsolidator:
    def __init__(self, short_term, long_term, llm):
        self.short_term = short_term
        self.long_term = long_term
        self.llm = llm

    async def consolidate(self, user_id: str):
        """استخراج الحقائق المهمة وتخزينها طويل المدى."""
        conversation = self.short_term.get_context()

        if len(conversation) < 5:
            return  # لا يكفي للتوحيد

        # استخراج الحقائق باستخدام LLM
        prompt = """من هذه المحادثة، استخرج:
1. تفضيلات المستخدم (أشياء يحبها/لا يحبها)
2. حقائق مهمة عنه
3. قرارات رئيسية اتُخذت

أرجع كـ JSON: {"preferences": [], "facts": [], "decisions": []}

المحادثة:
""" + json.dumps(conversation, indent=2)

        extraction = await self.llm.complete(prompt)
        data = json.loads(extraction)

        # تخزين كل ذاكرة مستخرجة
        for pref in data["preferences"]:
            await self.long_term.store_memory(
                user_id, pref, "preference"
            )

        for fact in data["facts"]:
            await self.long_term.store_memory(
                user_id, fact, "fact"
            )

بعد ذلك، سنستكشف أنماط التواصل بين الوكلاء. :::