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

لماذا المخفضات مهمة في الإنتاج

سؤال المقابلة الحقيقي (LangChain L6):

"لديك نظام متعدد الوكلاء حيث 3 وكلاء يبحثون بالتوازي. كل منهم يجد 10 مستندات. كيف تتراكم جميع الـ 30 مستنداً في الحالة بدون استبدال؟"

إجابة خاطئة:

"قم بإرجاع {"documents": new_docs} من كل عقدة." ❌ آخر كاتب يفوز، تبقى 10 مستندات فقط.

إجابة صحيحة:

"استخدم مخفض Annotated[list[str], operator.add]. LangGraph يدمج التحديثات بالإلحاق، وليس الاستبدال." ✅

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

دلالات تحديث الحالة: الدمج مقابل الاستبدال

اعتباراً من يناير 2026، يستخدم LangGraph الدمج الضحل افتراضياً:

السلوك الافتراضي: الدمج الضحل

from typing import TypedDict
from langgraph.graph import StateGraph

class AgentState(TypedDict):
    query: str
    documents: list[str]
    analysis: str

# العقدة تعيد تحديثات جزئية
def research_node(state: AgentState) -> dict:
    """العقدة تعيد فقط الحقول التي تريد تحديثها."""
    return {
        "documents": ["doc1", "doc2", "doc3"]
    }
    # query و analysis تبقى بدون تغيير!

# الرسم البياني يدمج التحديثات في الحالة الموجودة
graph = StateGraph(AgentState)
graph.add_node("research", research_node)

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

# العقد تعيد تحديثات حالة جزئية (dict)
# الرسم البياني ينفذ: new_state = {**old_state, **updates}
# هذا هو الدمج الضحل

مثال:

# الحالة الأولية
state = {
    "query": "AI trends 2026",
    "documents": [],
    "analysis": ""
}

# العقدة 1 تعيد
update1 = {"documents": ["doc1", "doc2"]}

# بعد الدمج: state = {**state, **update1}
state = {
    "query": "AI trends 2026",      # بدون تغيير
    "documents": ["doc1", "doc2"],   # استُبدل (لم يُلحق!)
    "analysis": ""                   # بدون تغيير
}

# العقدة 2 تعيد
update2 = {"documents": ["doc3", "doc4"]}

# بعد الدمج: state = {**state, **update2}
state = {
    "query": "AI trends 2026",
    "documents": ["doc3", "doc4"],   # استبدل ["doc1", "doc2"] ❌
    "analysis": ""
}

المشكلة: الدمج الافتراضي يستبدل حقول القائمة. تفقد المستندات السابقة!

الحل: استخدم المخفضات للتراكم.

المخفضات المشروحة: حل الإنتاج

operator.add للقوائم (الأكثر شيوعاً)

from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    """حالة مع تراكم قائم على المخفض."""
    query: str
    # المخفض: operator.add → يُلحق بدلاً من الاستبدال
    documents: Annotated[list[str], operator.add]
    key_findings: Annotated[list[str], operator.add]
    analysis: str  # لا مخفض → دمج افتراضي (استبدال)

# الآن العقد المتوازية يمكنها تراكم المستندات
def research_node_1(state: AgentState) -> dict:
    return {"documents": ["doc1", "doc2"]}

def research_node_2(state: AgentState) -> dict:
    return {"documents": ["doc3", "doc4"]}

# بعد كلتا العقدتين: documents = ["doc1", "doc2", "doc3", "doc4"] ✅

كيف يعمل:

# مع Annotated[list[str], operator.add]:
# new_value = old_value + update_value
# ["doc1", "doc2"] + ["doc3", "doc4"] = ["doc1", "doc2", "doc3", "doc4"]

حالات استخدام الإنتاج:

✅ تراكم نتائج البحث عبر الوكلاء
✅ جمع رسائل الخطأ من عقد متعددة
✅ بناء تاريخ المحادثة
✅ تجميع المقاييس (مثل عدد الرموز)

دوال المخفض المخصصة

لمنطق الدمج المعقد، اكتب مخفضات مخصصة:

from typing import Annotated

def merge_dicts(old: dict, new: dict) -> dict:
    """
    مخفض مخصص: دمج عميق للقواميس.
    يُستخدم للتكوين المتداخل أو البيانات الوصفية.
    """
    result = old.copy()
    for key, value in new.items():
        if key in result and isinstance(result[key], dict) and isinstance(value, dict):
            result[key] = merge_dicts(result[key], value)  # دمج تكراري
        else:
            result[key] = value  # استبدال الأوليات
    return result

def deduplicate_list(old: list, new: list) -> list:
    """
    مخفض مخصص: إلحاق + إزالة التكرار.
    يُستخدم لتراكم مستندات فريدة.
    """
    return list(dict.fromkeys(old + new))  # يحفظ الترتيب، يزيل التكرارات

class AdvancedState(TypedDict):
    documents: Annotated[list[str], deduplicate_list]
    metadata: Annotated[dict, merge_dicts]
    query: str

# الاستخدام
def node1(state: AdvancedState) -> dict:
    return {
        "documents": ["doc1", "doc2"],
        "metadata": {"source": "web", "count": 2}
    }

def node2(state: AdvancedState) -> dict:
    return {
        "documents": ["doc2", "doc3"],  # doc2 مكرر
        "metadata": {"source": "academic", "quality": "high"}
    }

# النتيجة:
# documents = ["doc1", "doc2", "doc3"] (بدون تكرار)
# metadata = {"source": "academic", "count": 2, "quality": "high"} (مدمج عميقاً)

متى تستخدم المخفضات المخصصة:

✅ منطق إزالة التكرار (مستندات فريدة، معرفات)
✅ الدمج العميق للهياكل المتداخلة
✅ التجميع (sum، max، min للمقاييس)
✅ الدمج القائم على الأولوية (الأحدث يستبدل الأقدم)

نمط الإنتاج: العداد مع operator.add

تتبع التكرارات، استخدام الرموز، أو التكاليف:

from typing import Annotated
import operator

class AgentState(TypedDict):
    query: str
    documents: Annotated[list[str], operator.add]

    # العدادات مع operator.add
    iteration_count: Annotated[int, operator.add]
    total_tokens: Annotated[int, operator.add]
    total_cost_usd: Annotated[float, operator.add]

# كل عقدة تزيد العدادات
def research_node(state: AgentState) -> dict:
    # ... قم بالبحث ...
    tokens_used = 1500
    cost = tokens_used * 0.00002  # $0.02 لكل 1K رمز

    return {
        "documents": ["doc1", "doc2"],
        "iteration_count": 1,        # يضيف للعداد
        "total_tokens": tokens_used, # يتراكم
        "total_cost_usd": cost       # يتراكم
    }

def analysis_node(state: AgentState) -> dict:
    tokens_used = 2000
    cost = tokens_used * 0.00002

    return {
        "iteration_count": 1,
        "total_tokens": tokens_used,
        "total_cost_usd": cost
    }

# بعد كلتا العقدتين:
# iteration_count = 2
# total_tokens = 3500
# total_cost_usd = 0.07

لماذا هذا يعمل:

operator.add على int/float ينفذ الجمع: 1 + 1 = 2
لا حاجة لـ state["iteration_count"] + 1 داخل العقد
تحديثات نظيفة وتعريفية

التحديثات الثابتة للتصحيح

نمط الإنتاج: أعد كائنات جديدة، لا تحوّل في المكان.

❌ سيء: التحويل في المكان (صعب التصحيح)

def bad_node(state: AgentState) -> dict:
    """يحوّل الحالة مباشرة - يكسر تصحيح السفر عبر الزمن."""
    state["documents"].append("new_doc")  # ❌ يعدّل القائمة الأصلية
    state["analysis"] = "updated"          # ❌ يعدّل السلسلة الأصلية

    return {}  # يعيد dict فارغ، لكن الحالة معدّلة بالفعل

المشاكل:

نقاط التفتيش تنكسر: حافظ نقاط التفتيش يخزن الحالة المعدّلة، وليس لقطات نظيفة
تصحيح السفر عبر الزمن يفشل: لا يمكن إعادة التشغيل من نقطة التفتيش
شروط السباق: العقد المتزامنة تعدّل الحالة المشتركة

✅ جيد: تحديثات ثابتة (جاهزة للإنتاج)

def good_node(state: AgentState) -> dict:
    """يعيد حالة جديدة بدون تعديل الأصلية."""
    # لا تعدّل: state["documents"].append(...)
    # بدلاً من ذلك: أعد قائمة جديدة
    new_documents = state["documents"] + ["new_doc"]

    return {
        "documents": new_documents,
        "analysis": "updated"
    }
    # الحالة الأصلية بدون تغيير، الرسم البياني يدمج التحديثات

الفوائد:

✅ نقاط التفتيش تعمل بشكل صحيح
✅ تصحيح السفر عبر الزمن ممكّن
✅ آمن للخيوط للتنفيذ المتزامن
✅ أسهل للاختبار (دوال نقية)

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

حدّث الحقول فقط عند الحاجة:

from typing import Optional

def conditional_node(state: AgentState) -> dict:
    """حدّث التحليل فقط إذا وجدت المستندات."""
    updates = {}

    # تحديث حقل شرطي
    if len(state["documents"]) > 0:
        updates["analysis"] = analyze_documents(state["documents"])
    else:
        updates["error_message"] = "No documents to analyze"

    # زِد التكرار دائماً
    updates["iteration_count"] = 1

    return updates  # يعيد فقط الحقول التي تغيرت

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

يتجنب مكالمات LLM غير الضرورية
يبقي الحالة نظيفة (لا تلوث فارغ/null)
تدفق تحكم واضح في تتبعات LangSmith

مخفض لتمرير الرسائل بين الوكلاء

حالة الاستخدام: أنظمة متعددة الوكلاء حيث يتواصل الوكلاء.

from typing import TypedDict, Annotated
import operator

class Message(TypedDict):
    """رسالة من وكيل إلى وكيل."""
    sender: str
    recipient: str
    content: str
    timestamp: str

class MultiAgentState(TypedDict):
    query: str
    # تراكم الرسائل بين الوكلاء
    messages: Annotated[list[Message], operator.add]
    final_report: str

def researcher_node(state: MultiAgentState) -> dict:
    """وكيل الباحث يرسل رسالة للكاتب."""
    message = Message(
        sender="researcher",
        recipient="writer",
        content="Found 5 papers on AI safety",
        timestamp="2026-01-15T10:30:00Z"
    )

    return {"messages": [message]}

def writer_node(state: MultiAgentState) -> dict:
    """الكاتب يقرأ الرسائل من الباحث."""
    researcher_messages = [
        msg for msg in state["messages"]
        if msg["sender"] == "researcher" and msg["recipient"] == "writer"
    ]

    # ... معالجة الرسائل ...

    response = Message(
        sender="writer",
        recipient="researcher",
        content="Draft complete",
        timestamp="2026-01-15T10:35:00Z"
    )

    return {"messages": [response]}

# messages تتراكم تاريخ الاتصالات

نمط الإنتاج:

استخدم Annotated[list, operator.add] لمسارات التدقيق، وسجلات الرسائل، والاتصال بين الوكلاء.

الأخطاء الشائعة والإصلاحات

الخطأ 1: نسيان المخفض، القوائم تُستبدل

# ❌ سيء
class State(TypedDict):
    documents: list[str]  # لا مخفض!

# آخر كاتب يفوز
node1_output = {"documents": ["doc1", "doc2"]}
node2_output = {"documents": ["doc3", "doc4"]}
# النهائي: ["doc3", "doc4"] - فُقد ["doc1", "doc2"]

# ✅ جيد
class State(TypedDict):
    documents: Annotated[list[str], operator.add]  # مخفض!

# كلاهما يتراكم
# النهائي: ["doc1", "doc2", "doc3", "doc4"]

الخطأ 2: استخدام المخفض على حقول غير القائمة

# ❌ سيء
class State(TypedDict):
    analysis: Annotated[str, operator.add]  # مخفض على سلسلة؟

node1 = {"analysis": "First insight"}
node2 = {"analysis": " Second insight"}
# النتيجة: "First insight Second insight" (دمج السلاسل!)
# ربما ليس ما تريده

# ✅ جيد
class State(TypedDict):
    analysis: str  # لا مخفض، دمج عادي (استبدال)

# آخر كاتب يفوز (متوقع لحقول القيمة الواحدة)

الخطأ 3: نمو لا نهائي بدون تقليم

# ❌ سيء: تراكم غير محدود
class State(TypedDict):
    documents: Annotated[list[str], operator.add]
    # بعد 1000 تكرار: 10,000+ مستند في الذاكرة!

# ✅ جيد: تقليم في المشرف
def supervisor_node(state: State) -> dict:
    """احتفظ بآخر 100 مستند فقط."""
    if len(state["documents"]) > 100:
        # استبدل بقائمة مقلّمة
        return {"documents": state["documents"][-100:]}
    return {}

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

س1: "ما الفرق بين دلالات الدمج والاستبدال؟"

الإجابة:

"يستخدم LangGraph الدمج الضحل افتراضياً: new_state = {**old_state, **updates}. للحقول بدون مخفضات، هذا يستبدل القيمة. للحقول مع Annotated[list, operator.add]، يتراكم بالإلحاق. دلالات الدمج تمنع العقد من استبدال تحديثات بعضها البعض عن طريق الخطأ، وهو أمر بالغ الأهمية في التنفيذ المتوازي."

س2: "متى تستخدم مخفض مخصص بدلاً من operator.add؟"

الإجابة:

"استخدم مخفضات مخصصة لمنطق معقد لا يستطيع operator.add التعامل معه. أمثلة: إزالة التكرار (إلحاق + إزالة التكرارات)، دمج عميق للقواميس المتداخلة، تحديثات قائمة على الأولوية (الأحدث يستبدل الأقدم)، أو دوال التجميع (max، min، متوسط). على سبيل المثال، في نظام متعدد الوكلاء يتتبع درجات الثقة، سأستخدم مخفضاً مخصصاً للاحتفاظ بأقصى ثقة لكل نتيجة."

س3: "كيف تؤثر المخفضات على أداء نقاط التفتيش؟"

الإجابة:

"المخفضات لا تؤثر مباشرة على نقاط التفتيش—تحدد كيفية دمج الحالة أثناء تنفيذ الرسم البياني. ومع ذلك، يمكن أن يؤدي التراكم غير المحدود عبر المخفضات إلى تضخم حجم نقطة التفتيش. على سبيل المثال، تراكم 10K مستند مع operator.add ينشئ نقطة تفتيش 50MB. أخفف من هذا بتقليم الحالة دورياً في عقدة المشرف، والاحتفاظ بحجم نقطة التفتيش تحت 10MB لأوقات استئناف سريعة."

مثال كود الإنتاج: نمط كامل

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

class ResearchState(TypedDict):
    """حالة الإنتاج مع المخفضات."""
    # الإدخال
    query: str

    # متراكم مع المخفض
    documents: Annotated[list[str], operator.add]
    key_findings: Annotated[list[str], operator.add]

    # قيمة واحدة (دمج/استبدال)
    analysis: Optional[str]
    final_report: Optional[str]

    # عدادات مع المخفض
    iteration_count: Annotated[int, operator.add]
    total_tokens: Annotated[int, operator.add]

    # تدفق التحكم
    max_iterations: int
    is_finished: bool

def research_node(state: ResearchState) -> dict:
    """بحث المستندات (تحديث ثابت)."""
    new_docs = search_documents(state["query"])

    return {
        "documents": new_docs,          # يُلحق عبر المخفض
        "iteration_count": 1,           # يزيد عبر المخفض
        "total_tokens": len(new_docs) * 100
    }

def analysis_node(state: ResearchState) -> dict:
    """تحليل المستندات (دلالات الاستبدال)."""
    analysis = analyze_docs(state["documents"])

    return {
        "analysis": analysis,           # يستبدل (لا مخفض)
        "key_findings": extract_findings(analysis),  # يُلحق
        "total_tokens": 2000
    }

def should_continue(state: ResearchState) -> str:
    """حافة شرطية مع فحص الأمان."""
    if state["iteration_count"] >= state["max_iterations"]:
        return "end"
    if state["is_finished"]:
        return "end"
    return "research"

# بناء الرسم البياني
graph = StateGraph(ResearchState)
graph.add_node("research", research_node)
graph.add_node("analysis", analysis_node)
graph.add_conditional_edges("analysis", should_continue, {
    "research": "research",
    "end": END
})
graph.set_entry_point("research")

app = graph.compile()

✅ استخدم Annotated[list, operator.add] للتراكم (مستندات، رسائل، نتائج) ✅ استخدم مخفضات مخصصة لإزالة التكرار، الدمج العميق، أو المنطق المعقد ✅ أعد تحديثات ثابتة (لا تحوّل الحالة في المكان) ✅ قلّم الحالة المتراكمة لمنع تضخم الذاكرة ✅ استخدم مخفضات للعدادات (التكرار، الرموز، تتبع التكلفة) ✅ الدمج الافتراضي يستبدل حقول القيمة الواحدة (سلوك متوقع)

التالي: تعلم أنماط الحالة المتقدمة للإنتاج في الدرس 3: التحقق، حالات الخطأ، وبيانات التعافي الوصفية.

:::