LangGraph في بيئة الإنتاج: خمسة أنماط تحافظ على أمان الوكلاء

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

إطار العمل LangGraph هو الأساس الذي نبني عليه لهذا النوع من المهام. فهو يجبرك على نمذجة الوكيل كآلة حالة صريحة (explicit state machine): خطوات مسماة، انتقالات محددة، وحالة محفوظة. لكن إطار العمل وحده لا يجعل الوكيل آمناً. يجب عليك بناء الأمان بنفسك، وهذه الأنماط الخمسة هي طريقتنا في تحقيق ذلك.

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

النمط 1: نقاط الحفظ المستمرة، مع Postgres كخيار افتراضي

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

يقوم نظام نقاط الحفظ (checkpointer) في LangGraph بحفظ حالة الرسم البياني بالكامل في كل خطوة. عندما تعود العملية، يستأنف الرسم البياني من آخر خطوة محفوظة. وبدون ذلك، تؤدي إعادة التشغيل إلى ضياع عمل العميل واضطرار شخص ما لإعادته يدوياً.

يُعد Postgres هو الخيار الافتراضي الصحيح لنقاط الحفظ في بيئة الإنتاج. فهو متين، ويتعامل مع العديد من المحادثات المتزامنة، ومعظم الفرق تقوم بتشغيله وأخذ نسخ احتياطية منه بالفعل. توفر LangChain هذا كحزمة langgraph-checkpoint-postgres، والتي تحتاج إلى استدعاء .setup() لإنشاء جداولها.

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

DB_URI = os.environ["CHECKPOINT_DB_URI"]

async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
    await checkpointer.setup()  # creates checkpoint tables on first run
    graph = builder.compile(checkpointer=checkpointer)

    # Each conversation or case gets its own thread_id
    config = {"configurable": {"thread_id": "case-48213"}}
    result = await graph.ainvoke({"user_input": message}, config=config)

استخدم SQLite أو الحفظ في الذاكرة للتطوير المحلي والاختبارات فقط. فهي لا تصمد أمام انتهاء مهلة السيرفرلس أو عند استخدام خادم ثانٍ.

نوعان من الذاكرة، وقاعدة بيانات واحدة

تحتفظ نقاط الحفظ بـ حالة التنفيذ: أين وصل سير العمل، ما هي الأدوات التي تم تشغيلها، ما الذي عاد منها، وماذا سيحدث تالياً. عادةً ما يحتاج الوكلاء أيضاً إلى ذاكرة طويلة المدى: تفضيلات العميل، الحالات السابقة، والحقائق التي تعيش أطول من سير عمل واحد.

غالباً ما تقوم الفرق بتقسيم هذه البيانات عبر قاعدة بيانات المتجهات، وذاكرة تخزين مؤقت Redis، وقاعدة بيانات للتطبيق. هذا يعني ثلاث فواتير وثلاث طرق لفقدان التزامن. توفر LangGraph PostgresStore للذاكرة طويلة المدى، مع بحث دلالي اختياري. تشغيل نقاط الحفظ والذاكرة في مثيل Postgres واحد يحافظ على سياسة نسخ احتياطي واحدة، ونموذج تحكم في الوصول واحد، وشيء واحد يجب مراقبته.

لماذا يفشل نهج "إرسال السجل بالكامل"

الاختصار الذي تتخذه العديد من المشاريع التجريبية هو إعادة إرسال المحادثة بأكملها إلى النموذج في كل خطوة. التكلفة تنمو بشكل أسرع مما تفترضه معظم الميزانيات. إذا كان السجل ينمو بمقدار g من الرموز لكل خطوة على مدار n من الخطوات، فإن إجمالي الرموز المدخلة لكل جلسة يكون تقريباً:

g × n × (n + 1) / 2

عند 750 رمزاً لكل خطوة على مدار 20 خطوة، تنتهي الجلسة بـ 15,000 رمز من السجل. يكون النموذج قد قرأ 157,500 رمز إدخال للوصول إلى هناك. اضرب ذلك في سعر الرمز الخاص بك وعدد الجلسات الشهرية لترى الرقم الخاص بك. كما أن الموجّهات الطويلة تقلل من دقة اتباع التعليمات، لذا من المرجح أن يتجاهل الوكيل قيداً مبكراً. تتيح الحالة المحفوظة لكل خطوة إرسال ما تحتاجه فقط، كما أن التخزين المؤقت للموجّهات يجعل الجزء الذي يجب عليك إعادة إرساله رخيصاً.

النمط 2: بوابات الموافقة باستخدام interrupt()

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

استدعاء interrupt() داخل عقدة يوقف الرسم البياني مؤقتاً ويحفظ حالته من خلال نظام نقاط الحفظ. لا يستهلك الوكيل أي موارد حوسبة أثناء انتظاره. ترسل واجهة برمجة التطبيقات الخاصة بك البيانات إلى المراجع، ويستأنف Command(resume=...) التشغيل بقراره. توثق LangChain أربعة استجابات مدمجة للمراجعين للمقاطعة: الموافقة، التعديل، الرفض، أو الرد المباشر.

from langgraph.types import interrupt, Command

def human_approval(state: AgentState) -> dict:
    action = state["pending_action"]
    decision = interrupt({
        "type": "approval_required",
        "action": action,
        "risk": state.get("risk_level", "medium"),
    })
    if decision["approved"]:
        return {"approved": True}
    return {"approved": False, "rejection_reason": decision.get("reason", "")}

# API layer, called when the reviewer clicks approve or reject
async def handle_decision(thread_id: str, approved: bool, reason: str = ""):
    config = {"configurable": {"thread_id": thread_id}}
    return await graph.ainvoke(
        Command(resume={"approved": approved, "reason": reason}),
        config=config,
    )

استخدم آلية إيقاف مؤقت واحدة، وليس اثنتين. إذا قمت أيضاً بالترجمة البرمجية باستخدام interrupt_before=["human_approval"]، سيتوقف الرسم البياني مرة قبل العقدة ومرة أخرى بداخلها. سيرى المراجعون حينها إيقافاً وهمياً يجب عليهم تجاوزه.

ضع بوابة على أي إجراء لا رجعة فيه، أو يكلف مالاً لكل استدعاء، أو يكون مرئياً خارج الشركة، أو يعتمد على نية مستنتجة. تحديد مكان البوابة وما يراه المراجع هو مهمة تصميم بحد ذاتها. إذا كان الإجراء المقيد يصل إلى العملاء وتختلف القواعد حسب المنتج أو المنطقة، فإن البوابة تُعد أداة تحكم في الامتثال، وليست ميزة في تجربة المستخدم. هذه هي المشكلة التي يعالجها Verel Coordinator للعمل الوارد: التوجيه القائم على القواعد وفحص الامتثال قبل أن يصل أي شيء إلى العميل.

النمط 3: إعادة المحاولة المحدودة، والأخطاء كحالة

كل استدعاء خارجي يفشل في بعض الأحيان: واجهات برمجة تطبيقات النماذج تُرجع حدود الاستخدام، والخدمات الداخلية تنتهي مهلتها. بدون إعادة المحاولة، يمكن لخلل بسيط أن يوقف سير العمل. لكن إعادة المحاولة غير المحدودة أسوأ. فهي تزيد الحمل على خدمة تعاني بالفعل وتستنزف حد الاستخدام الخاص بك.

الحل هو ميزانية إعادة المحاولة: عدد ثابت من المحاولات، مع تأخيرات متزايدة وتغيير عشوائي. الإصدار 1.x من LangGraph يحتوي على هذا بشكل مدمج. لا تحتاج إلى كتابة حلقة يدوياً.

from langgraph.types import RetryPolicy

TRANSIENT = (httpx.TimeoutException, RateLimitError)

builder.add_node(
    "execute_tool",
    execute_tool,
    retry_policy=RetryPolicy(
        max_attempts=3,
        initial_interval=1.0,
        backoff_factor=2.0,
        max_interval=30.0,
        jitter=True,
        retry_on=TRANSIENT,
    ),
)

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

async def execute_tool(state: AgentState) -> dict:
    try:
        result = await crm.update(state["pending_action"])  # transient errors propagate
        return {"tool_result": result, "error": None}
    except CRMValidationError as e:
        return {"tool_result": None, "error": str(e)}

def route_after_tool(state: AgentState) -> str:
    return "error" if state.get("error") else "success"

إعادة المحاولة تعني أيضاً أن العقدة يمكن أن تعمل أكثر من مرة. أي أداة تكتب في نظام آخر تحتاج إلى مفتاح تساوي الأثر، يُشتق عادةً من معرّف المسار والخطوة. وإلا فإن إعادة المحاولة قد ترسل نفس البريد الإلكتروني مرتين.

النمط 4: الحالة المكتوبة، والتحقق حيث يهم الأمر

العديد من أخطاء الإنتاج الدقيقة هي أخطاء في الحالة. تكتب إحدى العقد customer_id كعدد صحيح وتتوقعه العقدة التالية كنص. الحقل الذي تم تعيينه في الدور السابق لا يتم مسحه أبداً. العروض التجريبية لا تكتشف هذه الأخطاء لأن بيانات العرض التجريبي نظيفة.

قم بتعريف الحالة كـ TypedDict مع أنواع صريحة. ثم تحقق من صحة البيانات التي تدخل فعلياً من الخارج: مخرجات النموذج واستجابات الأدوات.

from typing import Literal, NotRequired, TypedDict
from pydantic import BaseModel

class PlannedAction(BaseModel):
    tool: Literal["crm_update", "send_email", "refund"]
    args: dict
    description: str

class AgentState(TypedDict):
    session_id: str
    user_input: str
    pending_action: NotRequired[dict | None]
    risk_level: NotRequired[Literal["low", "medium", "high"]]
    approved: NotRequired[bool | None]
    error: NotRequired[str | None]

async def plan_action(state: AgentState) -> dict:
    raw = await planner_llm.ainvoke(state["user_input"])
    action = PlannedAction.model_validate_json(raw.content)  # fails loudly here
    return {"pending_action": action.model_dump(), "approved": None}

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

النمط 5: التتبع والسفر عبر الزمن

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

تتكامل LangSmith مباشرة مع LangGraph وتُظهر المسار المتخذ، ومدخلات ومخرجات كل عقدة، والرموز، وزمن الاستجابة. تُعد Langfuse بديلاً جيداً يمكن استضافته ذاتياً وتتصل من خلال معالج استدعاء. في كلتا الحالتين، قم بتكوينها من خلال البيئة وأضف علامات لكل تشغيل. تسرد صفحة تسعير LangChain خطة مطوري LangSmith كمجانية لمقعد واحد مع 5,000 تتبع أساسي شهرياً، وخطة Plus بسعر 39 دولاراً لكل مقعد.

# Environment, never hardcoded:
# LANGSMITH_TRACING=true
# LANGSMITH_API_KEY=...
# LANGSMITH_PROJECT=support-agent-prod

config = {
    "configurable": {"thread_id": thread_id},
    "metadata": {"tenant": tenant_id, "workflow": "refund_request"},
    "tags": ["production", f"release-{RELEASE}"],
}

تتبع ثلاثة أرقام من اليوم الأول:

1. ▸معدل الخطأ لكل عقدة. هذا يخبرك أين يجب أن تعزز الأمان أولاً.
2. ▸معدل الموافقة. إذا كان مرتفعاً جداً، سيبدأ المراجعون في الموافقة التلقائية. وإذا كان منخفضاً جداً، فقد يكون الوكيل يتصرف دون رقابة.
3. ▸زمن الاستجابة للمسار البطيء (p95) حسب مسار الرسم البياني. هذا يوضح أي الفروع تجعل العملاء ينتظرون.

التتبعات هي أيضاً المادة الخام لـ التقييم على مستوى الخطوة لهذه التتبعات. يقوم القاضي بتقييم استدعاء أداة واحدة مقابل سؤال واحد، بدلاً من التشغيل بأكمله.

تمنحك نقاط الحفظ أيضاً القدرة على السفر عبر الزمن. يسرد graph.get_state_history(config) كل خطوة محفوظة لمسار معين. إذا مرر الوكيل وسيطة مشوهة في الخطوة السابعة، يمكن للمشغل تصحيح الحالة باستخدام graph.update_state(...) والاستئناف من نقطة الحفظ تلك. لا يتم إعادة تنفيذ الخطوات من واحد إلى ستة، ولا يبدأ العميل من جديد. للحصول على مجموعة أوسع من الأشياء التي تسوء بين العرض التجريبي والإطلاق، راجع 12 نمط فشل من إثبات المفهوم إلى الإنتاج.

الهيكل، مجمّعاً

from langgraph.graph import StateGraph, START, END

builder = StateGraph(AgentState)

builder.add_node("understand_intent", understand_intent)
builder.add_node("retrieve_context", retrieve_context)
builder.add_node("plan_action", plan_action)              # Pattern 4
builder.add_node("human_approval", human_approval)        # Pattern 2
builder.add_node("execute_tool", execute_tool,
                 retry_policy=TOOL_RETRY)                 # Pattern 3
builder.add_node("synthesize_response", synthesize_response)
builder.add_node("error_handler", error_handler)

builder.add_edge(START, "understand_intent")
builder.add_edge("understand_intent", "retrieve_context")
builder.add_edge("retrieve_context", "plan_action")

builder.add_conditional_edges("plan_action", route_by_risk, {
    "needs_approval": "human_approval",
    "safe_to_execute": "execute_tool",
})

builder.add_conditional_edges("human_approval",
    lambda s: "execute" if s["approved"] else "rejected", {
    "execute": "execute_tool",
    "rejected": "synthesize_response",
})

builder.add_conditional_edges("execute_tool", route_after_tool, {
    "success": "synthesize_response",
    "error": "error_handler",
})

builder.add_edge("synthesize_response", END)
builder.add_edge("error_handler", END)

# Pattern 1. No interrupt_before: the pause lives inside human_approval.
graph = builder.compile(checkpointer=checkpointer)

بوابة ما قبل الإطلاق

استخدم هذا الجدول كفحص للقرار. كل صف يمثل فشلاً سيلاحظه العميل.

الأسئلة الشائعة

كيف تقارن LangGraph بـ CrewAI و AutoGen في بيئة الإنتاج؟ تتيح لك CrewAI و AutoGen تشغيل نموذج أولي بكود أقل. بينما LangGraph أكثر تفصيلاً. في المقابل، تحصل على تحكم صريح في الحالة، والاستمرارية، والإيقافات المؤقتة، ومسارات الأخطاء. بيئة الإنتاج تحتاج بالضبط إلى هذه الأشياء. تختلف العائلات الثلاث حول ماهية تشغيل الوكيل أصلاً، وكيفية مقارنة LangGraph مع CrewAI و AutoGen تعود إلى هذا الاختلاف.

هل تؤدي نقاط الحفظ إلى إبطاء الوكيل؟ تضيف كل خطوة عملية كتابة واحدة في قاعدة البيانات. يعتمد الوقت الذي تستغرقه على حجم الحالة ومدى قرب Postgres من الوكيل. بالنسبة لسير عمل الأعمال حيث يستغرق كل استدعاء للنموذج ثوانٍ، نادراً ما يلاحظ المستخدمون عملية الكتابة. حافظ على الحالة خفيفة عن طريق تخزين معرّفات المستندات بدلاً من المستندات بأكملها. يتحكم إعداد durability في LangGraph أيضاً في متى يتم كتابة نقاط الحفظ، إذا كنت بحاجة إلى ضبطه.

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

هذا الأسبوع: قم بتشغيل الصف الأول من البوابة

خذ الصف الأول من جدول ما قبل الإطلاق وقم بتشغيله على وكيلك الحالي، في بيئة الاختبار إذا كان لديك واحدة. ابدأ تشغيلاً باستخدام thread_id حقيقي، وأوقف العملية في منتصفها، ثم أعد تشغيلها. إذا استأنف نفس المسار عند الخطوة الصحيحة، فلديك النمط 1. إذا بدأ من جديد أو فقد الحالة، فقم بإصلاح ذلك أولاً. إنه يكلف أقل من أي تغيير تفكر فيه على النموذج.