__start__
]):::first draft(draft) __end__([__end__
]):::last __start__ --> draft; draft -. done .-> __end__; draft -. again .-> 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](https://docs.langchain.com/oss/python/langgraph/checkpointers?ref=arabroot.io) يحفظ لقطة Snapshot من حالة الرسم البياني عند نهاية كل Super-step، ويجمع اللقطات في خيوط Threads، فكل محادثة أو مهمة لها معرف thread\_id تمرره مع كل استدعاء، وبدونه لا يستطيع أن يحفظ ولا أن يكمل. والصورة التالية تبين ما يحفظ في تجربتنا:  كل خطوة تحفظ لقطة في SQLite، فيكمل Container جديد من آخر لقطة وتأتي الـ Checkpointers في حزم منفصلة تثبتها بحسب قاعدة بياناتك، وكلها برخصة MIT: - InMemorySaver في الحزمة langgraph-checkpoint: يحفظ في الذاكرة، ويصفه التوثيق بأنه للتجربة، فإذا توقف البرنامج ضاع كل شيء. - SqliteSaver في الحزمة [langgraph-checkpoint-sqlite](https://pypi.org/project/langgraph-checkpoint-sqlite/?ref=arabroot.io): ملف واحد، ويصفه التوثيق بأنه مثالي للتجربة والعمل المحلي، وهو ما نستخدمه في التجربة. - PostgresSaver في الحزمة [langgraph-checkpoint-postgres](https://pypi.org/project/langgraph-checkpoint-postgres/?ref=arabroot.io): للإنتاج Production، وتستدعي الدالة setup() أول مرة لإنشاء الجداول، وإذا كان عندك PostgreSQL فقد شرحنا لماذا يكفيك لأغلب الحاجات في [مقال PostgreSQL لكل شيء](https://arabroot.io/articles/postgresql-%D9%84%D9%83%D9%84-%D8%B4%D9%8A%D8%A1-%D9%85%D8%AA%D9%89-%D8%AA%D9%83%D9%81%D9%8A%D9%83-%D9%82%D8%A7%D8%B9%D8%AF%D8%A9-%D8%A8%D9%8A%D8%A7%D9%86%D8%A7%D8%AA-%D9%88%D8%A7%D8%AD%D8%AF%D8%A9/). ولاحظ أن الحالة تحفظ كاملة، أي أن نص كل رسالة ونتيجة كل أداة موجودان في قاعدة البيانات، فإذا كانت المحادثات فيها بيانات حساسة فاستخدم الخيار EncryptedSerializer الذي يشفر الحالة بمفتاح AES من المتغير LANGGRAPH\_AES\_KEY كما في نفس التوثيق، وطبق على هذه القاعدة نفس سياسة النسخ الاحتياطي والحذف التي تطبقها على أي بيانات للموظفين. ## موافقة الإنسان Human-in-the-loop بالدالة interrupt الدالة interrupt توقف التنفيذ في أي مكان داخل عقدة، وترجع إلى من استدعى الرسم البياني قيمة JSON تصف ما ينتظر الموافقة، ثم يحفظ الـ Checkpointer الحالة، ويبقى الرسم البياني متوقفاً إلى أن تستدعيه مرة أخرى بنفس thread\_id ومعه Command(resume=...)، فتصبح هذه القيمة هي ما ترجعه interrupt داخل العقدة. والصورة التالية تبين هذه الدورة:  عند الاستكمال تعاد العقدة من أولها، لذلك ضع أي فعل له أثر بعد interrupt أو في عقدة منفصلة وأهم ما في [توثيق Interrupts](https://docs.langchain.com/oss/python/langgraph/interrupts?ref=arabroot.io) قواعده، لأن مخالفتها تعطيك أخطاء لا تظهر في التجربة الأولى: - عند الاستكمال لا يكمل التنفيذ من السطر الذي فيه interrupt، وإنما تعاد العقدة كلها من أولها، لذلك أي كود قبل interrupt يعمل مرة أخرى، فإذا كان هذا الكود يكتب في نظام خارجي فسوف يكتب مرتين. والحل أن يكون الكود قبلها Idempotent، أي أن تكراره لا يغير النتيجة، أو أن تضع الفعل بعدها أو في عقدة منفصلة، ولهذا فعقدة approve في تجربتنا تقرأ الحالة فقط قبل interrupt، والكتابة تحدث في عقدة tools بعدها. - لا تقم بوضع interrupt داخل try/except عام، والسبب أنها تعمل برمي استثناء Exception خاص، فإذا التقطته توقفت الموافقة عن العمل دون أي رسالة. - لا تقم بوضع interrupt داخل حلقة while في نفس العقدة، لأن كل استكمال يعيد كل الدورات السابقة. ## تجربة حقيقية: Agent يحفظ حالته ويطلب الموافقة نستخدم نفس بيئة [تجربة LangChain](https://arabroot.io/articles/%D9%85%D8%A7-%D9%87%D9%8A-langchain-%D9%88%D9%85%D8%AA%D9%89-%D8%AA%D8%AD%D8%AA%D8%A7%D8%AC%D9%87%D8%A7/#lab-setup): الشبكة 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: ```bash 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، ثم نعرف اختصاراً للأمر الطويل الذي نشغل به البرامج: ```bash 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 لاستكمال خيط متوقف بقرار المدير: ```python # 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: طلب فتح تذكرة يتوقف عند الموافقة ```bash lg python agent.py ask t1 "Open a ticket for the network team: the VPN in the Jeddah office is down." ``` والمخرج سوف يكون كما يلي: ```text [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 ثالث: ```bash docker restart lab-lc-ollama lg python agent.py state t1 lg python agent.py resume t1 approve ``` والمخرج سوف يكون كما يلي: ```text 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 داخل الخيط وخارجه ```bash 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?" ``` والمخرج سوف يكون كما يلي: ```text [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](https://docs.langchain.com/oss/python/langgraph/stores?ref=arabroot.io). وتذكر أن المحادثة الطويلة ترسل كلها إلى النموذج في كل خطوة، فتزيد التوكنز وزمن الاستجابة، لذلك تحتاج مع الوقت تلخيصاً أو قصاً للرسائل القديمة. ### الخطوة 4: الرفض، ثم موافقة على طلب ناقص في الخيط t3 نطلب تذكرة لفريق الأجهزة ثم نرفضها: ```bash 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 ``` والمخرج سوف يكون كما يلي: ```text [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 ونوافق عليه رغم أنه ناقص: ```bash 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 ``` طلب النموذج نفس الاستدعاء الناقص، ثم كان مخرج الاستكمال كما يلي: ```text [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: نفس الطلب بالعربية ```bash lg python agent.py ask t4 "افتح تذكرة لفريق الشبكات: شبكة VPN في مكتب جدة متوقفة" ``` والمخرج سوف يكون كما يلي: ```text [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 وعدد اللقطات لكل خيط، ثم ملف التذاكر: ```python 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()) ``` ```bash lg python inspect_db.py ``` والمخرج سوف يكون كما يلي: ```text ['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 ما زال ينتظر الموافقة، وهذا ما تريده بالضبط من وكيل يكتب في أنظمة حقيقية. ### إزالة التجربة ```bash 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](https://docs.langchain.com/oss/python/langgraph/streaming?ref=arabroot.io): - 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](https://docs.langchain.com/oss/python/langchain/multi-agent?ref=arabroot.io) بتنبيه مهم: ليست كل مهمة معقدة تحتاج هذا، فوكيل واحد بالأدوات والتعليمات المناسبة قد يحقق نتيجة مشابهة. ثم يذكر الأنماط التالية، والصورة بعدها تبين أشهرها: - الوكلاء الفرعيون Subagents: وكيل رئيسي يستدعي وكلاء متخصصين كأدوات، وكل التوجيه يمر عليه. - التسليم Handoffs: استدعاء أداة يغير متغيراً في الحالة فيتغير الوكيل أو أدواته وتعليماته. - المهارات Skills: وكيل واحد يحمل تعليمات ومعرفة متخصصة عند الحاجة. - الموجه Router: خطوة تصنف الطلب وترسله إلى وكيل أو أكثر، ثم تجمع النتائج. - سير عمل مخصص Custom Workflow: رسم بياني في LangGraph يجمع خطوات ثابتة مع خطوات وكيل، وهو ما بنيناه في التجربة.  الوكيل الرئيسي يقرر في Subagents، والكود يقرر في Router، وأنت ترسم الخطوات في Custom Workflow وإذا قرأت عن نمط المشرف Supervisor فقد كان له حزمة اسمها langgraph-supervisor، وهي اليوم بحسب [دليل الانتقال الرسمي](https://docs.langchain.com/oss/python/migrate/langgraph-supervisor?ref=arabroot.io) غير مطورة بنشاط، والبديل نمط Subagents: وكيل رئيسي بـ create\_agent كل أداة فيه تستدعي وكيلاً فرعياً. ويذكر الدليل قاعدة مهمة إذا كان في الوكلاء الفرعيين interrupt: الـ Checkpointer يوضع على الرسم البياني الخارجي فقط، وتمرر thread\_id في الاستدعاء الخارجي. ولاحظ أن كل وكيل إضافي يعني استدعاءات أكثر للنموذج وكلفة أكبر ومكاناً جديداً للخطأ، لذلك ابدأ بوكيل واحد، ولا تقسمه إلا عندما تقيس أنه يخلط بين أدواته. ## علاقة LangGraph بـ LangChain تطور الشركة نفسها LangChain Inc المكتبتين، ويقول [توثيق LangGraph](https://docs.langchain.com/oss/python/langgraph/overview?ref=arabroot.io) إنها مبنية من صانعي 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 دون أن ترسم شيئاً، والطريقتان تعملان على نفس بيئة التشغيل بحسب صفحة [الاختيار بين الطريقتين](https://docs.langchain.com/oss/python/langgraph/choosing-apis?ref=arabroot.io). وللمكتبة نسخة بلغة TypeScript في المستودع [langgraphjs](https://github.com/langchain-ai/langgraphjs?ref=arabroot.io) برخصة MIT. ## المكتبة المفتوحة والمنصة المدفوعة هذا القسم مهم لمن يبني على سيرفره، لأن الأسماء تغيرت وتداخلت. والصورة التالية تفصل بين ما تستطيع تشغيله بحرية وما يحتاج ترخيصاً:  المكتبة وحفظ الحالة مفتوحان بالكامل، وسيرفر التشغيل الجاهز هو الجزء التجاري - المكتبة langgraph وحزم الـ Checkpointers: [برخصة MIT](https://github.com/langchain-ai/langgraph?ref=arabroot.io)، فتستطيع تشغيلها داخل أي خدمة تكتبها، مثل واجهة FastAPI تستقبل الطلب وتستدعي الرسم البياني مع PostgresSaver، وهذا كل ما تحتاجه لأغلب الحالات. - الـ Agent Server: سيرفر جاهز يقدم API للخيوط والتشغيلات Runs والطوابير والبث، وحزمته [langgraph-api برخصة Elastic-2.0](https://pypi.org/project/langgraph-api/?ref=arabroot.io)، وهي رخصة مصدر متاح Source-available وليست رخصة مفتوحة المصدر بالمعنى الكامل، ومن شروطها كما في [نص الرخصة](https://www.elastic.co/licensing/elastic-license?ref=arabroot.io) ألا تقدم البرنامج لطرف ثالث كخدمة مدارة، وألا تتجاوز آلية مفتاح الترخيص. - التشغيل المستقل على سيرفرك Standalone: يحتاج بحسب [توثيق Standalone Servers](https://docs.langchain.com/langsmith/deploy-standalone-server?ref=arabroot.io) PostgreSQL وRedis، ومفتاح ترخيص في المتغير LANGGRAPH\_CLOUD\_LICENSE\_KEY، واتصالاً خارجياً إلى beacon.langchain.com للتحقق من الترخيص وتقارير الاستخدام إلا في وضع الشبكة المعزولة Air-gapped. - المنصة نفسها: كان اسمها LangGraph Platform ثم أصبحت LangSmith Deployment، وأصبح LangGraph Studio اسمه LangSmith Studio، بحسب [سجل تغييرات LangSmith](https://docs.langchain.com/langsmith/changelog?ref=arabroot.io) في أسبوع 13 إلى 17 أكتوبر 2025، وتشغيل LangSmith كاملاً على سيرفرك إضافة على خطة Enterprise كما في [توثيق Self-hosted](https://docs.langchain.com/langsmith/self-hosted?ref=arabroot.io). - للتطوير المحلي يوجد الأمر langgraph dev من الحزمة langgraph-cli، ويعمل دون Docker ويحفظ الحالة في الذاكرة وفي مجلد محلي، ويضع [توثيقه](https://docs.langchain.com/langsmith/local-dev-testing?ref=arabroot.io) مفتاح LangSmith API في المتطلبات، ولا تقم باستخدامه كسيرفر إنتاج. والخلاصة العملية لمن يريد كل شيء على سيرفره دون اشتراك: استخدم المكتبة مع PostgresSaver داخل خدمتك، واكتب بنفسك الواجهة التي تعرض طلبات الموافقة وترسل Command(resume=...)، وهي أسطر قليلة كما رأيت في agent.py. ## متى تكفيك حلقة بسيطة؟ LangGraph ليست الخيار الصحيح لكل وكيل، فالحلقة التي كتبناها في [مقال من RAG إلى Agents](https://arabroot.io/articles/rag-%D8%A7%D9%85-fine-tuning-%D8%A7%D9%85-mcp-%D8%A7%D9%85-agents/#lab-agent) بعشرين سطراً تكفيك في الحالات التالية: - الوكيل ينهي عمله في طلب 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 جزء تجاري بترخيص. - إذا كان وكيلك ينهي عمله في طلب واحد ولا ينتظر أحداً، فالحلقة البسيطة تكفيك. وفي المقالات القادمة من [مرحلة الذكاء الاصطناعي على سيرفرك](https://arabroot.io/start/#stage-ai) بإذن الله سوف نضع هذه الأجزاء معاً في مساعد واحد للموظفين، مع RAG وأدوات MCP وموافقة إنسان. [دليلLangChain: متى يفيدك إطار العمل ومتى يكفيك الكودما الذي تحله LangChain فعلاً في تطبيقات نماذج اللغة، وكيف تجربها على سيرفرك مع Ollama، ومتى يكفيك استدعاء مباشر دونها.](https://arabroot.io/articles/%D9%85%D8%A7-%D9%87%D9%8A-langchain-%D9%88%D9%85%D8%AA%D9%89-%D8%AA%D8%AD%D8%AA%D8%A7%D8%AC%D9%87%D8%A7/)[دليلمن RAG إلى Agents: أي طريقة يحتاجها مساعد شركتك؟ما الذي تحله كل طريقة لبناء مساعد ذكي للموظفين وما الذي لا تحله، ولماذا تبدأ بالأرخص ولا تنتقل إلى الأغلى إلا عند الحاجة.](https://arabroot.io/articles/rag-%D8%A7%D9%85-fine-tuning-%D8%A7%D9%85-mcp-%D8%A7%D9%85-agents/)[دليلRAG للمبتدئين: نموذج لغة يجيب من مستنداتك على سيرفرككيف تجعل نموذج اللغة يجيب من مستندات مؤسستك ويذكر مصدره، دون أن يخرج أي مستند من سيرفرك.](https://arabroot.io/articles/%D8%A7%D9%84%D8%A8%D8%AD%D8%AB-%D9%81%D9%8A-%D9%85%D8%B3%D8%AA%D9%86%D8%AF%D8%A7%D8%AA%D9%83-%D9%85%D8%B9-rag-%D9%84%D9%84%D9%85%D8%A8%D8%AA%D8%AF%D8%A6%D9%8A%D9%86/)[دليلبروتوكول MCP للمبتدئين: كيف تربط النماذج بأدواتك بأمانكيف يربط بروتوكول MCP نماذج اللغة بأنظمتك وبياناتك، ولماذا الأمان أهم ما فيه، مع سيرفر MCP صغير تبنيه بنفسك.](https://arabroot.io/articles/%D8%A8%D8%B1%D9%88%D8%AA%D9%88%D9%83%D9%88%D9%84-mcp-%D9%84%D9%84%D9%85%D8%A8%D8%AA%D8%AF%D8%A6%D9%8A%D9%86/) ## سجل التحديثات - أكتوبر 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.