مخططات TypedDict وأنماط تصميم الحالة — الغوص العميق في LangGraph: سير عمل الذكاء الاصطناعي الإنتاجية ذات الحالة — Nerd Level Tech

لماذا تصميم الحالة مهم في LangGraph الإنتاجي

سيناريو إنتاجي حقيقي (يناير 2026):

انتقل فريق في شركة Fortune 500 من CrewAI إلى LangGraph لأن نظام البحث متعدد الوكلاء الخاص بهم تعطل بعد معالجة 10K+ مستند. المشكلة؟ حالة غير منظمة نمت بلا حدود، استهلكت 64GB من الذاكرة.

بعد إعادة التصميم باستخدام مخططات TypedDict وحدود حجم الحالة، عالجوا 100K+ مستند مع <2GB بصمة ذاكرة.

هذا الدرس يعلمك: كيفية تصميم مخططات حالة جاهزة للإنتاج تتوسع، وتتجنب تضخم الذاكرة، وتجتاز مقابلات LangGraph في شركات مثل Anthropic و LangChain.

الحالة كمصدر واحد للحقيقة

في LangGraph، الحالة هي كل شيء. على عكس CrewAI (الذي يخفي الحالة في كائنات الطاقم) أو AutoGen (الذي يستخدم قوائم الرسائل)، يجعل LangGraph الحالة صريحة وقابلة للتحكم.

المبدأ الأساسي:

# الحالة تتدفق عبر جميع العقد
# كل عقدة تقرأ الحالة → تحولها → تعيد التحديثات
# الرسم البياني يدمج التحديثات مرة أخرى في الحالة

لماذا هذا مهم:

التصحيح: يمكنك فحص الحالة الدقيقة في أي نقطة
قابلية الاستئناف: يمكن إنشاء نقاط تفتيش للحالة واستئنافها
الاختبار: يمكنك اختبار العقد بوحدة مع حالة وهمية
قابلية المراقبة: تتبعات LangSmith تظهر تطور الحالة

سؤال المقابلة (Anthropic L5):

"لماذا يستخدم LangGraph حالة صريحة بدلاً من تمرير الرسائل الضمني مثل AutoGen؟"

إجابة قوية:

"تمكّن الحالة الصريحة من نقاط التفتيش، وتصحيح السفر عبر الزمن، والتحكم الدقيق في استخدام الذاكرة. مع الأنظمة القائمة على الرسائل، تفقد تاريخ الحالة عندما تتعطل العملية. تصميم LangGraph القائم على الحالة أولاً يجعل سير العمل قابلاً للاستئناف وجاهزاً للإنتاج، وهو أمر بالغ الأهمية للمهام متعددة الوكلاء طويلة المدى التي يمكن أن تفشل في منتصف الطريق."

TypedDict مقابل Pydantic: متى تستخدم كل منهما

TypedDict (موصى به لمعظم حالات استخدام LangGraph)

اعتباراً من يناير 2026، تستخدم حالة LangGraph الأصلية typing.TypedDict:

from typing import TypedDict, Annotated, Optional
from langgraph.graph import StateGraph
import operator

class AgentState(TypedDict):
    """
    حالة لسير عمل وكيل البحث.
    TypedDict يوفر تلميحات النوع في وقت الترجمة بدون تكلفة زمن التشغيل.
    """
    # الحقول المطلوبة
    query: str

    # الحقول الاختيارية مع القيم الافتراضية
    documents: Annotated[list[str], operator.add]  # تتراكم مع المخفض
    analysis: Optional[str]
    final_report: Optional[str]

    # حقول تدفق التحكم
    iteration_count: int
    max_iterations: int
    error_message: Optional[str]

# الاستخدام في الرسم البياني
graph = StateGraph(AgentState)

مزايا TypedDict:

تكلفة زمن تشغيل صفرية: فقط تلميحات النوع للإكمال التلقائي في IDE
دعم LangGraph الأصلي: يعمل بسلاسة مع StateGraph
بناء جملة بسيط: مألوف لمطوري Python
حقول مُشروحة: يدعم المخفضات (المزيد في الدرس 2)

متى تستخدم TypedDict:

✅ سير عمل LangGraph القياسي (95% من الحالات)
✅ التطبيقات الحرجة للأداء
✅ عندما تكون الحالة بيانات مفتاح-قيمة بسيطة

Pydantic (للتحقق المتقدم)

from pydantic import BaseModel, Field, field_validator
from typing import Annotated
import operator

class AgentState(BaseModel):
    """
    حالة Pydantic مع التحقق من زمن التشغيل.
    استخدم عندما تحتاج عقود بيانات صارمة.
    """
    query: str = Field(..., min_length=10, max_length=500)
    documents: Annotated[list[str], operator.add] = Field(default_factory=list)
    analysis: Optional[str] = None

    iteration_count: int = Field(default=0, ge=0)  # >= 0
    max_iterations: int = Field(default=10, le=100)  # <= 100

    @field_validator('query')
    @classmethod
    def validate_query(cls, v: str) -> str:
        if any(word in v.lower() for word in ['hack', 'exploit']):
            raise ValueError("Query contains forbidden terms")
        return v

# تحويل إلى dict لـ LangGraph
graph = StateGraph(AgentState.model_json_schema())

متى تستخدم Pydantic:

✅ تحتاج للتحقق من زمن التشغيل (مثل إدخال المستخدم من API)
✅ قواعد العمل المعقدة (تبعيات الحقول، المحققون المخصصون)
✅ عقود بيانات صارمة عبر الفرق
❌ حلقات داخلية حرجة للأداء (التحقق يضيف تكلفة)

نمط الإنتاج (يناير 2026):

استخدم TypedDict للحالة، Pydantic للتحقق من إدخال/إخراج API. قم بالتحويل عند الحدود.

هيكل الحالة: مسطح مقابل متداخل

الحالة المسطحة (موصى بها)

class FlatAgentState(TypedDict):
    """
    هيكل مسطح: جميع الحقول في المستوى الأعلى.
    أسهل في القراءة والتحديث ونقاط التفتيش.
    """
    # الإدخال
    user_query: str

    # مرحلة البحث
    search_results: Annotated[list[str], operator.add]
    research_summary: Optional[str]

    # مرحلة التحليل
    key_findings: Annotated[list[str], operator.add]
    analysis_report: Optional[str]

    # الإخراج
    final_answer: Optional[str]

    # البيانات الوصفية
    iteration: int
    total_tokens_used: int

المزايا:

✅ بسيط للوصول: state["user_query"] (وليس state["input"]["query"])
✅ نقاط تفتيش أسهل: الهيكل المسطح يتسلسل بشكل نظيف
✅ وظائف المخفض تعمل بشكل طبيعي على القوائم ذات المستوى الأعلى

الحالة المتداخلة (استخدم باعتدال)

class NestedAgentState(TypedDict):
    """
    هيكل متداخل: مجمّع حسب الاهتمام.
    أكثر تنظيماً ولكن أصعب في العمل.
    """
    input: dict  # {"query": str, "max_results": int}
    research: dict  # {"results": list, "summary": str}
    analysis: dict  # {"findings": list, "report": str}
    output: dict  # {"answer": str, "confidence": float}
    metadata: dict  # {"iteration": int, "tokens": int}

متى يكون التداخل منطقياً:

✅ الحالة بها 15+ حقلاً (التجميع يحسن القابلية للقراءة)
✅ حدود نظام فرعي واضحة (مثل فرق وكلاء منفصلة)
❌ معظم الحالات - فضّل المسطح للبساطة

نمط الإنتاج:

ابدأ مسطحاً. فقط تداخل إذا كان لديك >15 حقلاً أو أنظمة فرعية مميزة.

اصطلاحات التسمية للإنتاج

استراتيجية تسمية الحقول:

class ProductionAgentState(TypedDict):
    """
    اصطلاحات التسمية للوضوح.
    """
    # استخدم أسماء واضحة ووصفية (وليست اختصارات)
    user_query: str  # ✅ واضح
    # usr_q: str     # ❌ غامض

    # جمع القوائم
    documents: list[str]  # ✅ الجمع للقائمة
    # document: list[str]  # ❌ المفرد مُربك

    # استخدم Optional[] للحقول التي قد لا توجد
    analysis: Optional[str]  # ✅ اختيارية صريحة
    # analysis: str         # ❌ يفترض الوجود دائماً

    # بادئة حقول تدفق التحكم
    iteration_count: int      # ✅ عداد واضح
    max_iterations: int       # ✅ حد واضح
    is_finished: bool         # ✅ قيمة منطقية مع بادئة is_

    # لاحقة البيانات الوصفية
    created_at: str           # ✅ لاحقة الطابع الزمني
    updated_at: str           # ✅ لاحقة الطابع الزمني
    error_message: Optional[str]  # ✅ تتبع الخطأ

الأخطاء الشائعة التي يجب تجنبها:

اسم سيء	اسم جيد	لماذا
`res`	`search_results`	الوضوح على الإيجاز
`doc`	`documents`	الجمع للقوائم
`err`	`error_message`	وصفي
`iter`	`iteration_count`	غرض صريح
`done`	`is_finished`	اصطلاح قيمة منطقية

نمط الإنتاج: وثّق حالتك

أمر حاسم للتعاون الجماعي:

from typing import TypedDict, Annotated, Optional, Literal
import operator

class ResearchAgentState(TypedDict):
    """
    حالة لسير عمل البحث متعدد الوكلاء.

    سير العمل:
    1. الباحث: query → documents (متراكمة)
    2. المحلل: documents → analysis
    3. الكاتب: analysis → final_report
    4. المشرف: يوجه بين الوكلاء، يتحقق من حد التكرار

    المؤلف: فريق AI
    آخر تحديث: 2026-01-15
    """

    # === الإدخال ===
    query: str
    """سؤال البحث للمستخدم. مطلوب. حد أقصى 500 حرف."""

    # === مرحلة البحث ===
    documents: Annotated[list[str], operator.add]
    """
    مستندات البحث المتراكمة.
    تستخدم مخفض operator.add للإلحاق (وليس الاستبدال).
    حد أقصى 100 مستند لمنع تضخم الذاكرة.
    """

    # === مرحلة التحليل ===
    analysis: Optional[str]
    """تحليل منظم للمستندات. يولده وكيل المحلل."""

    key_findings: Annotated[list[str], operator.add]
    """أهم 3-5 نتائج. متراكمة عبر التكرارات."""

    # === الإخراج ===
    final_report: Optional[str]
    """تقرير markdown النهائي. يولده وكيل الكاتب."""

    # === تدفق التحكم ===
    next_agent: Literal["researcher", "analyzer", "writer", "end"]
    """قرار التوجيه بواسطة المشرف. يجب أن يكون أحد القيم الحرفية."""

    iteration_count: int
    """رقم التكرار الحالي. يبدأ من 0."""

    max_iterations: int
    """حد صارم لمنع الحلقات اللانهائية. الافتراضي: 10."""

    # === معالجة الأخطاء ===
    error_message: Optional[str]
    """إذا فشلت أي عقدة، يُخزن الخطأ هنا للتصحيح."""

لماذا هذا مهم:

أعضاء الفريق الجدد يفهمون الحالة فوراً
يقلل وقت الإعداد من أيام إلى ساعات
المستندات تصبح ذاتية التفسير لتتبعات LangSmith

نمط الإنتاج: إدارة حجم الحالة

المشكلة: الحالة تنمو بلا حدود في سير العمل طويل المدى.

الحل: تتبع الحجم والتقليم دورياً.

import sys
from typing import TypedDict, Optional

class ManagedAgentState(TypedDict):
    """حالة مع تتبع الحجم."""
    documents: list[str]
    analysis: Optional[str]
    final_report: Optional[str]

    # تتبع الحجم
    state_size_bytes: int
    max_state_size_bytes: int  # مثل، 10MB

def estimate_state_size(state: dict) -> int:
    """
    تقدير حجم الحالة بالبايتات.
    يُستخدم في الإنتاج لمنع تضخم الذاكرة.
    """
    import json
    return len(json.dumps(state, default=str).encode('utf-8'))

def prune_state_if_needed(state: ManagedAgentState) -> ManagedAgentState:
    """
    تقليم الحالة إذا تجاوزت الحجم الأقصى.
    يُستدعى دورياً في عقدة المشرف.
    """
    current_size = estimate_state_size(state)
    state["state_size_bytes"] = current_size

    if current_size > state["max_state_size_bytes"]:
        # احتفظ بآخر 10 مستندات فقط
        state["documents"] = state["documents"][-10:]
        print(f"⚠️ تم تقليم الحالة من {current_size} إلى {estimate_state_size(state)} بايت")

    return state

# الاستخدام في العقدة
def supervisor_node(state: ManagedAgentState) -> ManagedAgentState:
    """مشرف مع إدارة الحالة."""
    state = prune_state_if_needed(state)

    # ... منطق التوجيه ...

    return {"next_agent": "researcher"}

حدود الإنتاج (يناير 2026):

PostgresSaver: 1GB لكل نقطة تفتيش (قابلة للتكوين)
Redis: 512MB افتراضي (زيادة لسير العمل الكبير)
في الذاكرة: راقب بـ sys.getsizeof()، قلّم عند 100MB+

أسئلة المقابلة الشائعة

س1: "هل يجب أن تتضمن الحالة نتائج وسيطة أو فقط المخرجات النهائية؟"

إجابة قوية:

"قم بتضمين النتائج الوسيطة للتصحيح وقابلية الاستئناف. على سبيل المثال، قم بتخزين كلٍ من documents (نتائج البحث الخام) وanalysis (الرؤى المعالجة). هذا يسمح لك بفحص الفشل في منتصف سير العمل والاستئناف من نقاط التفتيش دون إعادة تشغيل مكالمات LLM المكلفة. في الإنتاج، أقلّم النتائج الوسيطة القديمة عندما تتجاوز الحالة حدود الحجم لمنع تضخم الذاكرة."

س2: "كيف تتعامل مع تعارضات الحالة في سير العمل المتزامن؟"

الإجابة:

"يستخدم LangGraph معرّفات الخيوط لعزل الحالة بين التشغيلات المتزامنة. كل سير عمل يحصل على مساحة اسم حالة خاصة به في حافظ نقاط التفتيش. للموارد المشتركة (مثل اتصالات قاعدة البيانات)، أمررها عبر التكوين، وليس الحالة. للتنسيق الجماعي، أستخدم حقول تمرير الرسائل في الحالة مثل team_messages: Annotated[list, operator.add] حيث يلحق الوكلاء بدون تعارضات."

س3: "متى ستستخدم Pydantic بدلاً من TypedDict للحالة؟"

الإجابة:

"استخدم Pydantic عندما تأتي الحالة من مصادر خارجية (APIs، إدخال المستخدم) التي تحتاج للتحقق من زمن التشغيل. على سبيل المثال، إذا قدم المستخدمون استعلامات عبر REST API، يتحقق Pydantic من الطول والشكل وقواعد العمل قبل الدخول إلى الرسم البياني. للحالة الداخلية ضمن عقد LangGraph، يُفضل TypedDict لأنه ليس له تكلفة زمن تشغيل ويعمل أصلياً مع StateGraph."

✅ استخدم TypedDict لمعظم حالة LangGraph (تكلفة صفرية، دعم أصلي) ✅ ابدأ بحالة مسطحة (فقط تداخل إذا >15 حقلاً) ✅ وثّق جميع الحقول بسلاسل التوثيق (وضوح للفرق) ✅ استخدم Optional[] للحقول التي قد لا توجد ✅ تتبع حجم الحالة في سير العمل طويل المدى (قلّم عند الحاجة) ✅ سمّ الحقول بوضوح (لا اختصارات، استخدم الاصطلاحات)

التالي: تعلم كيفية تحويل الحالة بشكل صحيح مع المخفضات ودلالات التحديث في الدرس 2.

:::