LangGraph: بناء Agent يحفظ حالته ويطلب الموافقة

في مقال LangChain: متى يفيدك إطار العمل بنينا وكيلاً Agent بسطر واحد مع create_agent، وذكرنا أن هذا الوكيل يعمل في الحقيقة كرسم بياني Graph في LangGraph. والآن لنفرض أن مساعد الموظفين يفتح تذكرة في نظام حقيقي، وأن المدير يجب أن يوافق قبل أي تذكرة، وأن المدير لن يرد قبل ساعتين. فماذا يحدث للوكيل في هذه الساعتين؟ إذا كان حلقة for في ذاكرة البرنامج كما في تجربة الوكيل السابقة، فسوف ينتظر البرنامج مفتوحاً، وإذا أعيد تشغيل السيرفر أو تحديث Container ضاعت المحادثة والخطوة التي وصل إليها، وضاع معها طلب الموافقة نفسه.

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

وسوف نناقش في هذا المقال ما يلي:

• لماذا لا تكفي الحلقة البسيطة عندما يحتاج الوكيل أن ينتظر أو يكمل بعد التوقف.
• أجزاء الرسم البياني: الحالة State والعقد Nodes والحواف Edges والحواف الشرطية Conditional Edges والحلقات Loops، ثم حلقة الوكيل نفسها كرسم بياني.
• الحفظ Persistence عبر Checkpointer على SQLite أو PostgreSQL، وموافقة الإنسان Human-in-the-loop بالدالة interrupt.
• تجربة حقيقية: Agent بأداتين، يتوقف قبل فتح التذكرة، ونعيد التشغيل، ثم نوافق أو نرفض من Container جديد.
• البث Streaming والذاكرة Memory وأنماط تعدد الوكلاء Multi-agent.
• علاقة LangGraph بـ LangChain، والفرق بين المكتبة المفتوحة والمنصة المدفوعة، ومتى تكفيك حلقة بسيطة.

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

لماذا لا تكفي الحلقة البسيطة؟

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

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

أجزاء الرسم البياني: State وNodes وEdges

الرسم البياني في LangGraph له ثلاثة أجزاء، والصورة التالية تبينها على مثال صغير دون أي نموذج لغة:

• الحالة State: قاموس له مخطط Schema تحدده بـ TypedDict أو Pydantic، وهو الذاكرة المشتركة بين كل الخطوات. ولكل مفتاح فيه دالة دمج اسمها Reducer تحدد كيف يدمج التحديث الجديد مع القيمة القديمة، فإذا لم تحددها استبدلت القيمة الجديدة القديمة، ومع الرسائل تستخدم الدالة add_messages التي تضيف الرسائل الجديدة إلى القائمة بدلاً من استبدالها، كما في توثيق Graph API.
• العقد Nodes: دوال Python عادية تأخذ الحالة وترجع تحديثاً لجزء منها، فقد تستدعي نموذجاً، أو تنفذ أداة، أو تحسب شيئاً دون أي نموذج.
• الحواف Edges: حافة ثابتة تقول «بعد A اذهب دائماً إلى B»، وحافة شرطية Conditional Edge تستدعي دالة تقرأ الحالة وترجع اسم العقدة التالية، وإذا رجعت الحافة إلى عقدة سابقة صار عندك حلقة Loop.

والتنفيذ يتم بخطوات اسمها Super-steps، وقد استلهمت LangGraph هذا التصميم من نظام Pregel من Google، ففي كل خطوة تعمل العقد النشطة، وقد تعمل أكثر من عقدة في نفس الخطوة بالتوازي، ثم ترسل مخرجاتها عبر الحواف إلى العقد التالية. ولاحظ أن الحد الأقصى لعدد الخطوات Recursion Limit أصبح افتراضياً 1000 خطوة منذ الإصدار 1.0.6 بحسب نفس التوثيق، وهذا رقم كبير لوكيل يدخل في حلقة مع نموذج يدفع عنه بالتوكن، لذلك مرر حداً أصغر في config مثل {"recursion_limit": 10} لكل وكيل.

والبرنامج graph.py التالي يبني الرسم البياني الذي في الصورة، والعقدة draft تحاكي نموذجاً يكتب عنوان تذكرة طويلاً ثم يختصره، والحافة الشرطية ترجعه إليها حتى يصبح العنوان 30 حرفاً أو أقل أو تنتهي المحاولات:

# رسم بياني دون أي نموذج لغة: State وNodes وEdges وConditional Edge وحلقة
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    title: str
    attempts: int

def draft(state: State) -> dict:
    # نحاكي نموذجاً يكتب عنواناً طويلاً، ثم يختصره في كل محاولة
    words = state["title"].split()
    return {"title": " ".join(words[: max(3, len(words) - 4)]), "attempts": state["attempts"] + 1}

def route(state: State) -> str:
    if len(state["title"]) <= 30 or state["attempts"] >= 3:
        return "done"
    return "again"

builder = StateGraph(State)
builder.add_node("draft", draft)
builder.add_edge(START, "draft")
builder.add_conditional_edges("draft", route, {"again": "draft", "done": END})
graph = builder.compile()

print(graph.get_graph().draw_mermaid())
start = {"title": "شبكة VPN في مكتب جدة متوقفة منذ الصباح ولا يستطيع أحد الدخول إلى الأنظمة", "attempts": 0}
for update in graph.stream(start, stream_mode="updates"):
    print(update)

والمخرج سوف يكون كما يلي:

---
config:
  flowchart:
    curve: linear
---
graph TD;
	__start__([<p>__start__</p>]):::first
	draft(draft)
	__end__([<p>__end__</p>]):::last
	__start__ --> draft;
	draft -. &nbsp;done&nbsp; .-> __end__;
	draft -. &nbsp;again&nbsp; .-> draft;
	classDef default fill:#f2f0ff,line-height:1.2
	classDef first fill-opacity:0
	classDef last fill:#bfb6fc

{'draft': {'title': 'شبكة VPN في مكتب جدة متوقفة منذ الصباح ولا يستطيع', 'attempts': 1}}
{'draft': {'title': 'شبكة VPN في مكتب جدة متوقفة', 'attempts': 2}}

في المخرج أعلاه لاحظ التالي:

• الدالة draw_mermaid رسمت الرسم البياني بلغة Mermaid، وهو نص تضعه في أي محرر يدعمها فترى الرسم، وهذا مفيد للتوثيق ومراجعة الكود، فالخطوات مكتوبة كبيانات وليست مخفية داخل حلقة.
• الحافة الشرطية ظهرت بخط متقطع ومعها القيمتان again وdone، والسهم من draft إلى draft هو الحلقة.
• مع stream_mode="updates" جاء بعد كل خطوة التحديث الذي رجعته العقدة فقط، فعملت draft مرتين، وفي الثانية أصبح العنوان 27 حرفاً فذهب التنفيذ إلى END.

حلقة الوكيل كرسم بياني

الآن نضع نموذج اللغة في الرسم البياني، فالوكيل في أبسط أشكاله عقدتان وحافة شرطية: عقدة llm تستدعي النموذج، وعقدة tools تنفذ الأدوات التي طلبها، والحافة الشرطية بعد llm تسأل: هل طلب النموذج أداة؟ فإذا طلب ذهبت إلى tools ثم رجعت إلى llm، وإذا لم يطلب انتهى التنفيذ. وهذا بالضبط ما تبنيه create_agent داخلياً. أما في وكيلنا فأضفنا عقدة ثالثة اسمها approve تقف بين النموذج وأي أداة تكتب، والصورة التالية تبين الرسم البياني الذي سوف نبنيه في التجربة:

وقد يتساءل البعض: لماذا لا نستخدم create_agent مع الطبقة HumanInTheLoopMiddleware التي ذكرناها في مقال LangChain؟ والإجابة أنها الخيار الأقصر إذا كان كل ما تحتاجه موافقة قبل أدوات معينة، ولكننا نكتب الرسم البياني هنا بأيدينا حتى ترى ما يحدث تحته، ولأنك سوف تحتاج هذا المستوى عندما تضع بين خطوات النموذج خطوات ثابتة يقررها الكود، مثل التحقق من صلاحية الموظف في نظام الصلاحيات قبل عرض الطلب على المدير.

الحفظ Persistence عبر Checkpointer

الـ Checkpointer هو المكون الذي يحفظ الحالة، فبحسب توثيق Checkpointers يحفظ لقطة Snapshot من حالة الرسم البياني عند نهاية كل Super-step، ويجمع اللقطات في خيوط Threads، فكل محادثة أو مهمة لها معرف thread_id تمرره مع كل استدعاء، وبدونه لا يستطيع أن يحفظ ولا أن يكمل. والصورة التالية تبين ما يحفظ في تجربتنا:

وتأتي الـ Checkpointers في حزم منفصلة تثبتها بحسب قاعدة بياناتك، وكلها برخصة MIT:

• InMemorySaver في الحزمة langgraph-checkpoint: يحفظ في الذاكرة، ويصفه التوثيق بأنه للتجربة، فإذا توقف البرنامج ضاع كل شيء.
• SqliteSaver في الحزمة langgraph-checkpoint-sqlite: ملف واحد، ويصفه التوثيق بأنه مثالي للتجربة والعمل المحلي، وهو ما نستخدمه في التجربة.
• PostgresSaver في الحزمة langgraph-checkpoint-postgres: للإنتاج Production، وتستدعي الدالة setup() أول مرة لإنشاء الجداول، وإذا كان عندك PostgreSQL فقد شرحنا لماذا يكفيك لأغلب الحاجات في مقال PostgreSQL لكل شيء.

ولاحظ أن الحالة تحفظ كاملة، أي أن نص كل رسالة ونتيجة كل أداة موجودان في قاعدة البيانات، فإذا كانت المحادثات فيها بيانات حساسة فاستخدم الخيار EncryptedSerializer الذي يشفر الحالة بمفتاح AES من المتغير LANGGRAPH_AES_KEY كما في نفس التوثيق، وطبق على هذه القاعدة نفس سياسة النسخ الاحتياطي والحذف التي تطبقها على أي بيانات للموظفين.

موافقة الإنسان Human-in-the-loop بالدالة interrupt

الدالة interrupt توقف التنفيذ في أي مكان داخل عقدة، وترجع إلى من استدعى الرسم البياني قيمة JSON تصف ما ينتظر الموافقة، ثم يحفظ الـ Checkpointer الحالة، ويبقى الرسم البياني متوقفاً إلى أن تستدعيه مرة أخرى بنفس thread_id ومعه Command(resume=...)، فتصبح هذه القيمة هي ما ترجعه interrupt داخل العقدة. والصورة التالية تبين هذه الدورة:

وأهم ما في توثيق Interrupts قواعده، لأن مخالفتها تعطيك أخطاء لا تظهر في التجربة الأولى:

• عند الاستكمال لا يكمل التنفيذ من السطر الذي فيه interrupt، وإنما تعاد العقدة كلها من أولها، لذلك أي كود قبل interrupt يعمل مرة أخرى، فإذا كان هذا الكود يكتب في نظام خارجي فسوف يكتب مرتين. والحل أن يكون الكود قبلها Idempotent، أي أن تكراره لا يغير النتيجة، أو أن تضع الفعل بعدها أو في عقدة منفصلة، ولهذا فعقدة approve في تجربتنا تقرأ الحالة فقط قبل interrupt، والكتابة تحدث في عقدة tools بعدها.
• لا تقم بوضع interrupt داخل try/except عام، والسبب أنها تعمل برمي استثناء Exception خاص، فإذا التقطته توقفت الموافقة عن العمل دون أي رسالة.
• لا تقم بوضع interrupt داخل حلقة while في نفس العقدة، لأن كل استكمال يعيد كل الدورات السابقة.

تجربة حقيقية: Agent يحفظ حالته ويطلب الموافقة

نستخدم نفس بيئة تجربة LangChain: الشبكة lab-lc-net، وContainer لـ Ollama باسم lab-lc-ollama مع النموذج qwen2.5:1.5b، وDocker Image باسم lab-lc-py:1 فيها langgraph 1.2.14 وlanggraph-checkpoint-sqlite 3.1.1. وإذا لم تكن قد جهزتها بعد فهذه أوامرها مختصرة، والملف Dockerfile كاملاً في مقال LangChain:

docker network create lab-lc-net
docker volume create lab-lc-ollama
docker run -d --name lab-lc-ollama --network lab-lc-net \
  -v lab-lc-ollama:/root/.ollama ollama/ollama:0.40.2
docker exec lab-lc-ollama ollama pull qwen2.5:1.5b
docker build -t lab-lc-py:1 .

ونضيف Volume باسم lab-lc-data لملف SQLite وملف التذاكر، حتى يبقيا بعد حذف Container، ثم نعرف اختصاراً للأمر الطويل الذي نشغل به البرامج:

docker volume create lab-lc-data
alias lg='docker run --rm --name lab-lc-run --network lab-lc-net -v "$PWD:/lab" -v lab-lc-data:/data lab-lc-py:1'

لاحظ الخيار --rm، فكل أمر يشغل Container جديداً ويحذفه عند الانتهاء، أي أن كل خطوة في التجربة عملية جديدة لا تعرف شيئاً عن التي قبلها إلا ما حفظ في lab-lc-data، وهذا أقسى من إعادة تشغيل سيرفر.

البرنامج agent.py

البرنامج التالي يبني الرسم البياني الذي في الصورة الثانية، بأداة قراءة get_ticket_status وأداة كتابة create_ticket تكتب التذاكر في ملف JSON حتى نرى هل نفذت فعلاً، ويستقبل من سطر الأوامر الأمر ask لسؤال جديد، أو resume لاستكمال خيط متوقف بقرار المدير:

# Agent بأداتين، وCheckpointer على SQLite، وموافقة إنسان قبل أي أداة تكتب
import json, sqlite3, sys
from typing import Literal
from langchain_ollama import ChatOllama
from langchain.tools import tool
from langchain.messages import SystemMessage, ToolMessage
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import Command, interrupt

TICKETS = "/data/tickets.json"

@tool
def get_ticket_status(ticket_id: str) -> str:
    """Return the current status of a helpdesk ticket, for example IT-2041."""
    data = {"IT-2041": {"status": "in_progress", "team": "network-team"}}
    return json.dumps(data.get(ticket_id, {"error": "ticket not found"}))

@tool
def create_ticket(title: str, team: str) -> str:
    """Create a new helpdesk ticket for a team (network-team, hardware-team or access-team)."""
    try:
        tickets = json.load(open(TICKETS))
    except FileNotFoundError:
        tickets = []
    ticket = {"id": f"IT-{2042 + len(tickets)}", "title": title, "team": team}
    tickets.append(ticket)
    json.dump(tickets, open(TICKETS, "w"), ensure_ascii=False)
    return json.dumps(ticket, ensure_ascii=False)

TOOLS = [get_ticket_status, create_ticket]
NEEDS_APPROVAL = {"create_ticket"}
SYSTEM = SystemMessage("You are the IT helpdesk assistant. Use the tools when needed. Answer in Arabic in one sentence.")
model = ChatOllama(model="qwen2.5:1.5b", base_url="http://lab-lc-ollama:11434", temperature=0, seed=1).bind_tools(TOOLS)

def llm(state: MessagesState) -> dict:
    return {"messages": [model.invoke([SYSTEM] + state["messages"])]}

def route(state: MessagesState) -> str:
    calls = state["messages"][-1].tool_calls
    if not calls:
        return END
    return "approve" if any(c["name"] in NEEDS_APPROVAL for c in calls) else "tools"

def approve(state: MessagesState) -> Command[Literal["tools", "llm"]]:
    calls = state["messages"][-1].tool_calls
    decision = interrupt({"question": "approve these tool calls?", "tool_calls": [{"name": c["name"], "args": c["args"]} for c in calls]})
    if decision == "approve":
        return Command(goto="tools")
    refused = [ToolMessage("The manager rejected this action. Do not retry it.", tool_call_id=c["id"]) for c in calls]
    return Command(goto="llm", update={"messages": refused})

builder = StateGraph(MessagesState)
builder.add_node("llm", llm)
builder.add_node("approve", approve)
builder.add_node("tools", ToolNode(TOOLS))
builder.add_edge(START, "llm")
builder.add_conditional_edges("llm", route, ["tools", "approve", END])
builder.add_edge("tools", "llm")

checkpointer = SqliteSaver(sqlite3.connect("/data/checkpoints.db", check_same_thread=False))
graph = builder.compile(checkpointer=checkpointer)

def show(update):
    for node, value in update.items():
        if node == "__interrupt__":
            print("  [interrupt]", json.dumps(value[0].value, ensure_ascii=False))
            continue
        messages = (value or {}).get("messages", [])
        if not messages:
            print(f"  [{node}] (no new messages)")
        for m in messages:
            print(f"  [{node}] {m.type}:", m.tool_calls if getattr(m, "tool_calls", None) else m.content)

cmd, thread = sys.argv[1], sys.argv[2]
config = {"configurable": {"thread_id": thread}}
if cmd == "ask":
    inp = {"messages": [{"role": "user", "content": sys.argv[3]}]}
elif cmd == "resume":
    inp = Command(resume=sys.argv[3])
if cmd in ("ask", "resume"):
    for update in graph.stream(inp, config, stream_mode="updates"):
        show(update)
state = graph.get_state(config)
print("next:", state.next, "| checkpoints:", len(list(graph.get_state_history(config))))

في الكود أعلاه لاحظ التالي:

• الحالة MessagesState فيها مفتاح واحد messages مع الدالة add_messages كـ Reducer، لذلك كل عقدة ترجع الرسائل الجديدة فقط.
• الدالة route هي الحافة الشرطية، فالقرار هل الأداة تحتاج موافقة يتخذه الكود من قائمة NEEDS_APPROVAL، ولا يتخذه النموذج.
• العقدة approve ترجع Command فيه اسم العقدة التالية، وعند الرفض تضيف رسالة tool لكل استدعاء، والسبب أن النموذج ينتظر نتيجة لكل أداة طلبها، وقد ترفض واجهات بعض المزودين المحادثة إذا نقصت.
• العقدة ToolNode من الحزمة langgraph.prebuilt تنفذ الأدوات التي في آخر رسالة، وترجع الخطأ للنموذج كرسالة بدلاً من أن توقف البرنامج كما سوف ترى.
• كل الحفظ سطران: SqliteSaver على ملف في /data، ثم compile(checkpointer=...).

الخطوة 1: طلب فتح تذكرة يتوقف عند الموافقة

lg python agent.py ask t1 "Open a ticket for the network team: the VPN in the Jeddah office is down."

والمخرج سوف يكون كما يلي:

  [llm] ai: [{'name': 'create_ticket', 'args': {'title': 'VPN in Jeddah Office', 'team': 'network-team'}, 'id': '80d6e158-92a1-47b0-a583-895896ece51c', 'type': 'tool_call'}]
  [interrupt] {"question": "approve these tool calls?", "tool_calls": [{"name": "create_ticket", "args": {"title": "VPN in Jeddah Office", "team": "network-team"}}]}
next: ('approve',) | checkpoints: 3

طلب النموذج الأداة create_ticket بعنوان واسم فريق صحيح، فأرسلته الحافة الشرطية إلى approve، فتوقف التنفيذ عند interrupt وظهر طلب الموافقة في البث، ثم انتهى البرنامج وحذف Container، والحالة تقول إن الخطوة التالية approve بعد ثلاث لقطات محفوظة.

الخطوة 2: إعادة التشغيل ثم الموافقة من Container جديد

الآن نعيد تشغيل Container الخاص بـ Ollama أيضاً، ثم نقرأ حالة الخيط t1 من Container جديد، ثم نوافق من Container ثالث:

docker restart lab-lc-ollama
lg python agent.py state t1
lg python agent.py resume t1 approve

والمخرج سوف يكون كما يلي:

next: ('approve',) | checkpoints: 3
  [approve] (no new messages)
  [tools] tool: {"id": "IT-2042", "title": "VPN in Jeddah Office", "team": "network-team"}
  [llm] ai: The ticket for the network team regarding the downed VPN in the Jeddah office has been created successfully. The ticket ID is IT-2042.
next: () | checkpoints: 6

في المخرج أعلاه لاحظ التالي:

• الـ Container الجديد لم ير المحادثة الأولى، ومع ذلك عرف أن الخيط t1 متوقف عند approve، لأنه قرأ الحالة من ملف checkpoints.db.
• عند الاستكمال أعيدت العقدة approve من أولها، ورجعت interrupt هذه المرة بالقيمة approve، فذهب التنفيذ إلى tools فنفذ create_ticket مرة واحدة، ثم رجع إلى llm فكتب الإجابة برقم التذكرة IT-2042.
• لم يتغير شيء في الكود بين التوقف والاستكمال، والانتظار يمكن أن يكون ثانية أو أسبوعاً، فالحالة في قاعدة البيانات وليست في ذاكرة برنامج.
• الإجابة جاءت بالإنجليزية رغم أن التعليمات تقول «Answer in Arabic»، وهذا من النموذج الصغير، كما رأينا في مقال LangChain.

الخطوة 3: الذاكرة Memory داخل الخيط وخارجه

lg python agent.py ask t1 "What is the ID of the ticket you opened for me?"
lg python agent.py ask t2 "What is the ID of the ticket you opened for me?"

والمخرج سوف يكون كما يلي:

  [llm] ai: The ticket ID is IT-2042.
next: () | checkpoints: 9
  [llm] ai: I'm sorry, but I don't have the ticket ID. Could you please provide it?
next: () | checkpoints: 3

في الخيط t1 عرف النموذج رقم التذكرة لأن المحادثة كلها في الحالة المحفوظة، وزادت اللقطات إلى تسع، أما الخيط t2 فهو محادثة جديدة لا تعرف شيئاً عن t1. وهذه هي الذاكرة قصيرة المدى Short-term Memory في LangGraph: الـ Checkpointer لكل خيط. أما الذاكرة طويلة المدى Long-term Memory التي تتشارك بين الخيوط، مثل تفضيلات موظف تريد أن يتذكرها المساعد في كل محادثة، فلها مكون آخر اسمه Store، منه InMemoryStore للتجربة وPostgresStore للإنتاج كما في توثيق Stores. وتذكر أن المحادثة الطويلة ترسل كلها إلى النموذج في كل خطوة، فتزيد التوكنز وزمن الاستجابة، لذلك تحتاج مع الوقت تلخيصاً أو قصاً للرسائل القديمة.

الخطوة 4: الرفض، ثم موافقة على طلب ناقص

في الخيط t3 نطلب تذكرة لفريق الأجهزة ثم نرفضها:

lg python agent.py ask t3 "Open a ticket for the hardware team: my laptop screen is broken."
lg python agent.py resume t3 reject

والمخرج سوف يكون كما يلي:

  [llm] ai: [{'name': 'create_ticket', 'args': {'title': 'broken laptop screen'}, 'id': 'ca124b3b-9010-4233-8ef8-c7ffba8de8fd', 'type': 'tool_call'}]
  [interrupt] {"question": "approve these tool calls?", "tool_calls": [{"name": "create_ticket", "args": {"title": "broken laptop screen"}}]}
next: ('approve',) | checkpoints: 3
  [approve] tool: The manager rejected this action. Do not retry it.
  [llm] ai: I'm sorry, but the manager has rejected this action. Please try again or contact the IT support team directly.
next: () | checkpoints: 5

لاحظ أولاً أن النموذج طلب create_ticket دون المعامل team رغم أنه مطلوب في الأداة، وأن طلب الموافقة أظهر ذلك للمدير قبل أي تنفيذ، فعقدة الموافقة ليست حماية من الأفعال الخطرة فقط، وإنما فرصة لرؤية ما سوف ينفذ بالضبط. ولاحظ ثانياً أن النموذج بعد الرفض قال «Please try again» رغم أن رسالة الرفض تقول «Do not retry it»، لذلك لا تعتمد على أن النموذج يحترم الرفض، فالكود هو الذي يمنع الأداة، وهو ما فعلته عقدة approve. والآن لنكسر التجربة لنفهمها، فنكرر نفس الطلب في الخيط t5 ونوافق عليه رغم أنه ناقص:

lg python agent.py ask t5 "Open a ticket for the hardware team: my laptop screen is broken."
lg python agent.py resume t5 approve

طلب النموذج نفس الاستدعاء الناقص، ثم كان مخرج الاستكمال كما يلي:

  [approve] (no new messages)
  [tools] tool: Error invoking tool 'create_ticket' with kwargs {'title': 'broken laptop screen'} with error:
 team: Field required
 Please fix the error and try again.
  [llm] ai: The team field is required. Please provide the team name.
next: () | checkpoints: 6

رفضت ToolNode الاستدعاء لأن Pydantic تحققت من المعاملات قبل تشغيل الدالة، ورجع الخطأ إلى النموذج كرسالة tool بدلاً من أن يتوقف البرنامج، فطلب النموذج اسم الفريق من الموظف. وهذا سلوك جيد هنا، ولكن تذكر أن الخطأ الذي يرجع إلى النموذج قد يدفعه إلى محاولة أخرى، وكل محاولة استدعاء جديد، وهذا سبب آخر لوضع حد للخطوات.

الخطوة 5: نفس الطلب بالعربية

lg python agent.py ask t4 "افتح تذكرة لفريق الشبكات: شبكة VPN في مكتب جدة متوقفة"

والمخرج سوف يكون كما يلي:

  [llm] ai: [{'name': 'create_ticket', 'args': {'title': 'شبكة VPN في مكتب جدة', 'team': 'network-team'}, 'id': '0f4ea00d-f4d0-4263-9ce6-d367bfaacffc', 'type': 'tool_call'}]
  [interrupt] {"question": "approve these tool calls?", "tool_calls": [{"name": "create_ticket", "args": {"title": "شبكة VPN في مكتب جدة", "team": "network-team"}}]}
next: ('approve',) | checkpoints: 3

هذه المرة طلب النموذج الأداة من سؤال عربي، وترجم «فريق الشبكات» إلى network-team الصحيح، بينما لم يطلب أداة القراءة من سؤال عربي في مقال LangChain. وهذا التفاوت بين سؤال وآخر هو سلوك النموذج الصغير، والدرس أن تختبر وكيلك بأسئلة بنفس لغة موظفيك قبل أن تحكم عليه.

الخطوة 6: ماذا يوجد في قاعدة البيانات؟

البرنامج inspect_db.py يعرض جداول SQLite وعدد اللقطات لكل خيط، ثم ملف التذاكر:

import sqlite3
db = sqlite3.connect("/data/checkpoints.db")
print([r[0] for r in db.execute("select name from sqlite_master where type='table'")])
for row in db.execute("select thread_id, count(*) from checkpoints group by thread_id"):
    print(row)
print(open("/data/tickets.json").read())

lg python inspect_db.py

والمخرج سوف يكون كما يلي:

['checkpoints', 'writes']
('t1', 9)
('t2', 3)
('t3', 5)
('t4', 3)
('t5', 6)
[{"id": "IT-2042", "title": "VPN in Jeddah Office", "team": "network-team"}]

جدولان فقط: checkpoints للقطات، وwrites للكتابات المعلقة لكل عقدة داخل الخطوة، وهي التي تجعل العقد الناجحة لا تعاد إذا تعطلت عقدة أخرى في نفس الخطوة كما يشرح توثيق Checkpointers. ولاحظ أن ملف التذاكر فيه تذكرة واحدة فقط رغم خمسة خيوط: t3 رفض، وt5 فشل في التحقق، وt4 ما زال ينتظر الموافقة، وهذا ما تريده بالضبط من وكيل يكتب في أنظمة حقيقية.

إزالة التجربة

unalias lg
docker volume rm lab-lc-data
docker rm -f lab-lc-ollama
docker volume rm lab-lc-ollama
docker network rm lab-lc-net
docker rmi lab-lc-py:1

البث Streaming

استخدمنا في التجربة graph.stream مع stream_mode="updates" فوصلنا تحديث كل عقدة بعد انتهائها، وهذه أوضاع البث الأخرى بحسب توثيق Streaming:

• values: الحالة كاملة بعد كل خطوة، وهو مفيد للتتبع ولكنه أثقل.
• messages: توكنز النموذج أولاً بأول مع اسم العقدة التي جاءت منها، وهو ما تحتاجه لواجهة محادثة يظهر فيها النص وهو يكتب.
• custom: بيانات ترسلها أنت من داخل العقدة، مثل «جاري البحث في 3 مصادر».
• checkpoints وtasks وdebug: لمراقبة التنفيذ نفسه.

ويقول نفس التوثيق إن التطبيقات الجديدة ينصح لها بواجهة Event Streaming التي ظهرت في LangGraph v1.2 عبر graph.stream_events(..., version="v3")، وفيها تصل طلبات الموافقة في stream.interrupts، أما graph.stream وgraph.invoke فما زالتا تعملان كما رأيت.

تعدد الوكلاء Multi-agent

عندما يكثر عدد الأدوات على وكيل واحد يبدأ النموذج بالخلط بينها، وهنا يظهر اقتراح «وكلاء متعددون». ويبدأ توثيق Multi-agent بتنبيه مهم: ليست كل مهمة معقدة تحتاج هذا، فوكيل واحد بالأدوات والتعليمات المناسبة قد يحقق نتيجة مشابهة. ثم يذكر الأنماط التالية، والصورة بعدها تبين أشهرها:

• الوكلاء الفرعيون Subagents: وكيل رئيسي يستدعي وكلاء متخصصين كأدوات، وكل التوجيه يمر عليه.
• التسليم Handoffs: استدعاء أداة يغير متغيراً في الحالة فيتغير الوكيل أو أدواته وتعليماته.
• المهارات Skills: وكيل واحد يحمل تعليمات ومعرفة متخصصة عند الحاجة.
• الموجه Router: خطوة تصنف الطلب وترسله إلى وكيل أو أكثر، ثم تجمع النتائج.
• سير عمل مخصص Custom Workflow: رسم بياني في LangGraph يجمع خطوات ثابتة مع خطوات وكيل، وهو ما بنيناه في التجربة.

وإذا قرأت عن نمط المشرف Supervisor فقد كان له حزمة اسمها langgraph-supervisor، وهي اليوم بحسب دليل الانتقال الرسمي غير مطورة بنشاط، والبديل نمط Subagents: وكيل رئيسي بـ create_agent كل أداة فيه تستدعي وكيلاً فرعياً. ويذكر الدليل قاعدة مهمة إذا كان في الوكلاء الفرعيين interrupt: الـ Checkpointer يوضع على الرسم البياني الخارجي فقط، وتمرر thread_id في الاستدعاء الخارجي. ولاحظ أن كل وكيل إضافي يعني استدعاءات أكثر للنموذج وكلفة أكبر ومكاناً جديداً للخطأ، لذلك ابدأ بوكيل واحد، ولا تقسمه إلا عندما تقيس أنه يخلط بين أدواته.

علاقة LangGraph بـ LangChain

تطور الشركة نفسها LangChain Inc المكتبتين، ويقول توثيق LangGraph إنها مبنية من صانعي LangChain ولكن يمكن استخدامها دون LangChain. وهذا صحيح بمعنى أنك لا تحتاج الحزمة langchain ولا create_agent، ولكن الحزمة langgraph نفسها تعتمد على langchain-core، وقد رأينا ذلك في متطلباتها عند بناء Docker Image، فالرسائل والأدوات وRunnableConfig تأتي من langchain-core دائماً. أما في الاتجاه الآخر فـ create_agent في LangChain ترجع رسماً بيانياً في LangGraph، فتستطيع أن تضعه عقدة داخل رسمك البياني الأكبر.

ولـ LangGraph طريقتان للكتابة: Graph API التي استخدمناها، حيث تعرف العقد والحواف صراحة، وFunctional API حيث تكتب دالة Python عادية بالمزخرف entrypoint وتحصل على نفس الحفظ وinterrupt دون أن ترسم شيئاً، والطريقتان تعملان على نفس بيئة التشغيل بحسب صفحة الاختيار بين الطريقتين. وللمكتبة نسخة بلغة TypeScript في المستودع langgraphjs برخصة MIT.

المكتبة المفتوحة والمنصة المدفوعة

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

• المكتبة langgraph وحزم الـ Checkpointers: برخصة MIT، فتستطيع تشغيلها داخل أي خدمة تكتبها، مثل واجهة FastAPI تستقبل الطلب وتستدعي الرسم البياني مع PostgresSaver، وهذا كل ما تحتاجه لأغلب الحالات.
• الـ Agent Server: سيرفر جاهز يقدم API للخيوط والتشغيلات Runs والطوابير والبث، وحزمته langgraph-api برخصة Elastic-2.0، وهي رخصة مصدر متاح Source-available وليست رخصة مفتوحة المصدر بالمعنى الكامل، ومن شروطها كما في نص الرخصة ألا تقدم البرنامج لطرف ثالث كخدمة مدارة، وألا تتجاوز آلية مفتاح الترخيص.
• التشغيل المستقل على سيرفرك Standalone: يحتاج بحسب توثيق Standalone Servers PostgreSQL وRedis، ومفتاح ترخيص في المتغير LANGGRAPH_CLOUD_LICENSE_KEY، واتصالاً خارجياً إلى beacon.langchain.com للتحقق من الترخيص وتقارير الاستخدام إلا في وضع الشبكة المعزولة Air-gapped.
• المنصة نفسها: كان اسمها LangGraph Platform ثم أصبحت LangSmith Deployment، وأصبح LangGraph Studio اسمه LangSmith Studio، بحسب سجل تغييرات LangSmith في أسبوع 13 إلى 17 أكتوبر 2025، وتشغيل LangSmith كاملاً على سيرفرك إضافة على خطة Enterprise كما في توثيق Self-hosted.
• للتطوير المحلي يوجد الأمر langgraph dev من الحزمة langgraph-cli، ويعمل دون Docker ويحفظ الحالة في الذاكرة وفي مجلد محلي، ويضع توثيقه مفتاح LangSmith API في المتطلبات، ولا تقم باستخدامه كسيرفر إنتاج.

والخلاصة العملية لمن يريد كل شيء على سيرفره دون اشتراك: استخدم المكتبة مع PostgresSaver داخل خدمتك، واكتب بنفسك الواجهة التي تعرض طلبات الموافقة وترسل Command(resume=...)، وهي أسطر قليلة كما رأيت في agent.py.

متى تكفيك حلقة بسيطة؟

LangGraph ليست الخيار الصحيح لكل وكيل، فالحلقة التي كتبناها في مقال من RAG إلى Agents بعشرين سطراً تكفيك في الحالات التالية:

• الوكيل ينهي عمله في طلب HTTP واحد، ولا ينتظر إنساناً، ولا يحتاج أن يكمل بعد إعادة التشغيل.
• الخطوات قليلة وثابتة، فسير عمل Workflow بكود عادي أوضح من رسم بياني، والسبب أن الرسم البياني يضيف مفاهيم مثل Reducer وSuper-step وCheckpoint يجب أن يفهمها كل من يقرأ الكود.
• المحادثة قصيرة وتحفظها أصلاً في جدول رسائل عندك، وتعيد إرسالها في كل طلب.

أما عندما يحتاج الوكيل أن يتوقف لينتظر موافقة، أو أن يكمل بعد عطل في منتصف مهمة طويلة، أو أن تراجع تاريخ خطواته لتعرف لماذا اتخذ قراراً معيناً، فهنا يوفر عليك LangGraph كتابة نظام حفظ واستعادة دقيق بنفسك، وهذا هو سبب وجوده.

أخطاء شائعة

• InMemorySaver في الإنتاج: يعمل في التجربة ثم تضيع كل الموافقات المعلقة مع أول تحديث للخدمة، فاستخدم PostgresSaver أو SqliteSaver على Volume.
• فعل له أثر قبل interrupt: العقدة تعاد من أولها عند الاستكمال، فيتكرر الإرسال أو الإنشاء، لذلك ضع الأفعال بعد interrupt أو في عقدة منفصلة.
• نسيان thread_id أو تكراره بين موظفين: دون thread_id لا يحفظ شيء، ومع نفس thread_id لموظفين مختلفين يرى كل منهما محادثة الآخر، لذلك اربطه بمعرف المستخدم والمحادثة في تطبيقك، وتحقق من الصلاحية قبل تحميل أي خيط.
• ترك قرار الموافقة للنموذج: «اسأل المستخدم إذا كان الفعل خطيراً» في التعليمات ليست موافقة، والموافقة الحقيقية حافة في الكود كما فعلنا بالدالة route.
• الاعتماد على حد الخطوات الافتراضي: الحد 1000 خطوة، فمرر recursion_limit صغيراً لكل وكيل.
• وكلاء متعددون من البداية: قسم الوكيل فقط عندما تقيس أنه يخلط بين أدواته، وكل تقسيم يضيف استدعاءات وكلفة.

قائمة التحقق

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

تدريب عملي

• أضف إلى agent.py حداً للخطوات بالقيمة {"recursion_limit": 6} في config، ثم اجعل get_ticket_status ترجع خطأ دائماً، ولاحظ كم خطوة يحاول النموذج قبل أن يتوقف التنفيذ بالخطأ GraphRecursionError.
• عدل العقدة approve حتى تقبل القرار edit مع عنوان جديد، فتغير معاملات create_ticket قبل تنفيذها، ثم جربها على الخيط t4 المتوقف.

الخلاصة

وصلنا لنهاية الموضوع، وأهم ما فيه:

• LangGraph تكتب خطوات الوكيل كرسم بياني: حالة State مشتركة، وعقد Nodes تعمل، وحواف Edges ثابتة أو شرطية تقرر الخطوة التالية، والحلقة حافة ترجع إلى الخلف.
• الـ Checkpointer يحفظ لقطة بعد كل خطوة في خيط thread_id، فيكمل الوكيل بعد التوقف من Container جديد، كما رأيت عندما أعدنا التشغيل ثم وافقنا من عملية أخرى.
• الدالة interrupt توقف التنفيذ لينتظر إنساناً، وعند الاستكمال تعاد العقدة من أولها، لذلك ضع الأفعال بعدها.
• التجربة بينت أن عقدة الموافقة تكشف أيضاً أخطاء النموذج مثل معامل ناقص، وأن النموذج قد لا يحترم الرفض، فالمنع مكانه الكود.
• المكتبة وحفظ الحالة مفتوحان برخصة MIT ويعملان على سيرفرك دون أي مفتاح، والـ Agent Server وLangSmith Deployment جزء تجاري بترخيص.
• إذا كان وكيلك ينهي عمله في طلب واحد ولا ينتظر أحداً، فالحلقة البسيطة تكفيك.

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

دليلLangChain: متى يفيدك إطار العمل ومتى يكفيك الكودما الذي تحله LangChain فعلاً في تطبيقات نماذج اللغة، وكيف تجربها على سيرفرك مع Ollama، ومتى يكفيك استدعاء مباشر دونها.دليلمن RAG إلى Agents: أي طريقة يحتاجها مساعد شركتك؟ما الذي تحله كل طريقة لبناء مساعد ذكي للموظفين وما الذي لا تحله، ولماذا تبدأ بالأرخص ولا تنتقل إلى الأغلى إلا عند الحاجة.دليلRAG للمبتدئين: نموذج لغة يجيب من مستنداتك على سيرفرككيف تجعل نموذج اللغة يجيب من مستندات مؤسستك ويذكر مصدره، دون أن يخرج أي مستند من سيرفرك.دليلبروتوكول MCP للمبتدئين: كيف تربط النماذج بأدواتك بأمانكيف يربط بروتوكول MCP نماذج اللغة بأنظمتك وبياناتك، ولماذا الأمان أهم ما فيه، مع سيرفر MCP صغير تبنيه بنفسك.

سجل التحديثات

• أكتوبر 2026: كتابة المقال، واختبار التجربة على Docker Engine 29.8.0 (معالج فقط) مع Ollama 0.40.2 والنموذج qwen2.5:1.5b، وPython 3.13.16 مع langgraph 1.2.14 وlanggraph-checkpoint-sqlite 3.1.1 وlanggraph-prebuilt 1.1.0 وlangchain-core 1.6.9 وlangchain-ollama 1.1.0.