ابنِ وكيلك الأول من الصفر إلى النهاية: مشروع كامل بـ LangChain وLangGraph
دليل عملي تبني فيه وكيل حجوزات كاملاً سطراً بسطر: ثلاث أدوات، رسم حالة بعقدتين، حفظ دائم على القرص، وبوابة موافقة بشرية قبل أي حجز. التحذيران اللذان يُكلّفان مالاً حقيقياً مكتوبان داخل البناء لا في نهايته: أن InMemorySaver يفقد كل شيء عند إعادة تشغيل البرنامج، وأن استئناف المقاطعة يعيد العقدة من بدايتها فيجب تصميم الحجز كي لا يتكرر. وينتهي الدليل بقائمة تحقق من ثمانية أسئلة تشخيصية، بعد عرض الرسم عبر draw_mermaid_png والبثّ التشخيصي عبر stream_events بنسخته v3.

في هذا الدليل ستكتب ملفاً واحداً من أعلى إلى أسفل، ويصبح عندك وكيل حجوزات يعمل: يبحث عن الرحلات، ويحجز — بعد موافقة بشرية — ولا يحجز مرتين لو استؤنف. كل خطوة تُضاف إلى الملف نفسه، فلا مقاطع منفصلة تتعطّل وحدها.
قبل الشيفرة، ثمانية مصطلحات ستتكرر ألافاً: الوكيل هو النموذج اللغوي داخل حلقة يبادر هو في اختيار أدواته، وليس روبوت محادثة. الأداة دالة يستطيع النموذج استدعاؤها. الحالة هي لقطة مشتركة بين العقد. دالة الدمج تقرّر كيف يُدمج ما تعيده العقدة مع الموجود. نقطة الحفظ هي لقطة للحالة تُكتب عند حدود الخطوات الكبرى. مُحافِظ النقاط هو الكائن الذي يكتب هذه اللقطات. thread_id معرّف المحادثة التي ستستأنف منها. ومقاطعة تفاعلية (interrupt) هي تعليق صريح تنتظر فيه إجابة إنسان.
ما الذي سيبنيه هذا الدليل، وما الذي لن نبنيه فيه؟
سنبني وكيل حجوزات طيران كاملاً: ثلاث أدوات، رسم حالة بعقدتين، حفظ دائم في ملف SQLite، وبوابة موافقة بشرية قبل أي حجز فعلي.
أما ما لن نضعه — بتقصد، لا إهمالاً:
- لا نشر. لا حاويات، ولا خوادم، ولا أسماء نطاقات. ما سيخرج من هذا الدليل ملف Python وملف قاعدة بيانات محلياً.
- لا لانغ سميث (LangSmith). تتبّع وأرشفة وتقييم على منصة تجارية ليست مطلوبة لتشغيل وكيلك. كل تشخيص في هذا الدليل يتم عبر أدوات مفتوحة: رسم المخطط وحالة الرسم والبثّ.
- لا مصادقة ولا تعدد مستأجرين. لا حسابات، ولا صلاحيات، ولا عزل بيانات بين مستخدمين.
- لا ذاكرة طويلة المدى. لا
storeولا نطاقات مُسمّاة. هذا مشروع محادثة واحدة تحفظ نفسها عبر إغلاق البرنامج. - لا ذكاء اصطناعي متعدد الوكلاء. وكيل واحد، عقدتان. التوسّع لاحقاً — وله مقاله.
وإن كان كل ما تريده وكيلاً بسيطاً يجيب عن أسئلة، فربما لا تحتاج هذه السلسلة كلها: توثيق لانغ تشين نفسه يقول (You do not need to know LangGraph for basic LangChain agent usage.).
وأخيراً، إن كنت تتابع درساً قديماً على يوتيوب أو مدوّنة، احكم عليه بالتاريخ لا بالمظهر: النسخة التي على PyPI اليوم هي langchain 1.4.3 بتاريخ Released: Sep 28, 2026، وأغلب ما قبل ذلك يعلّم أسماء نقلت إلى مكان آخر.
ما الذي تحتاج تثبيته قبل كتابة أول سطر؟
تحتاج بايثون Requires Python <4.0.0, >=3.10.0 أو أحدث، وثلاث حزم للبدء:
pip install -U langchain langgraph langgraph-checkpoint-sqlite
استخدم -U دائماً: سطر تثبيت بلا -U قد يبقيك على إصدار أقدم تحمل أسماء وواجهات مختلفة عمّا تشرحه هذه المقالة.
الحزمة الثالثة ليست اختيارية في هذا الدليل: langgraph-checkpoint-sqlite: An implementation of LangGraph checkpointer that uses SQLite database (SqliteSaver / AsyncSqliteSaver). Ideal for experimentation and local workflows. Needs to be installed separately.
بقيت خطوة واحدة لا أستطيع أن أخمّنها عنك: حزمة التكامل الخاصة بمزوّدك. التوثيق يطلب منك ضبط ANTHROPIC_API_KEY ثم يحيلك إلى صفحة بعينها لكل المزوّدين: See chat model integrations for all available providers. لا تخمّن اسم الحزمة من شكل المعرّف — هذا بالضبط ما يتغيّر بين الإصدارات.
والآن: كل ما تحت يُضاف إلى ملف واحد اسمه agent.py. والخطوات تراكمية — كل خطوة تفترض أن ما قبلها مكتوب، فلا تشغّل مقطعاً وحده.
كيف تكتب ثلاث أدوات يفهمها النموذج؟
الأداة في لانغ تشين ليست كياناً سحرياً: Under the hood, tools are callable functions with well-defined inputs and outputs that get passed to a chat model. The model decides when to invoke a tool based on the conversation context, and what input arguments to provide.
يقع جوهر العقد بينك وبين النموذج في سطرين: Type hints are **required** as they define the tool's input schema. و By default, the function's docstring becomes the tool's description. أي أن التسمية النوعية تبني المُخطَّط، ونص التوثيق يصير تعليمات يقرأها النموذج ليقرّر متى يستدعيك.
هذه أول إضافة للملف. ثلاث أدوات: بحث، حجز، إلغاء.
from langchain.tools import tool
from langchain.chat_models import init_chat_model
model = init_chat_model("claude-sonnet-4-6", temperature=0)
# سجل تجريبي يحاكي جدول الحجوزات؛ في مشروع حقيقي استبدله بقاعدة بياناتك
BOOKINGS = {}
@tool
def search_flights(origin: str, destination: str) -> str:
"""Search available flights between two cities.
Args:
origin: Departure city code, for example RUH
destination: Arrival city code, for example DXB
"""
return (
f"{origin} to {destination}: one morning departure, "
"one evening departure, one overnight return."
)
@tool
def book_flight(flight_code: str, passenger: str) -> str:
"""Create a booking for one passenger on one flight.
Args:
flight_code: The flight code returned by search_flights
passenger: Full name exactly as it must appear on the ticket
"""
key = f"{flight_code}:{passenger}"
existing = BOOKINGS.get(key)
if existing:
return f"Booking {existing} already exists for {passenger}. Nothing new created."
booking_code = f"BK-{len(BOOKINGS) + 1}"
BOOKINGS[key] = booking_code
return f"Booking {booking_code} created for {passenger} on {flight_code}."
@tool
def cancel_booking(booking_code: str) -> str:
"""Cancel an existing booking.
Args:
booking_code: The booking code returned by book_flight
"""
key = next((k for k, v in BOOKINGS.items() if v == booking_code), None)
if key is None:
return f"No booking found with code {booking_code}."
del BOOKINGS[key]
return f"Booking {booking_code} cancelled."
# النموذج لا يعرف الأدوات إلا بعد bind_tools
tools = [search_flights, book_flight, cancel_booking]
tools_by_name = {t.name: t for t in tools}
model_with_tools = model.bind_tools(tools)
ثلاث قواعد توفّر عليك ساعات:
- الاسم
snake_case.Prefer snake_case for tool names (e.g., web_search instead of Web Search). Some model providers have issues with or reject names containing spaces or special characters with errors.الخطأ هنا لا يظهر إلا عند النموذج، فيبدو كأنه «أخطأ في التفسير» بينما الاسم مكسور. configوruntimeمحجوزان.The following parameter names are reserved and cannot be used as tool arguments. Using these names will cause runtime errors.لا تسمّ وسيطكruntime، استخدمToolRuntimeإن احتجت حالة من داخل الأداة.- لا تترجم الوعد إلى الكود. النص العربي أو الإنجليزي داخل الـdocstring هو ما يقرأه النموذج لتقرير الوكيل؛ الشرح الموجّه للزميل يوضع في تعليق
#لا في التوثيق.
كيف تعرّف حالة الوكيل بحيث تنجو الرسائل؟
الحالة هي قاموس مُنمَّط (TypedDict) يشترك فيه كل العقد؛ وتُلحق الرسائل بدالة دمج بدل استبدال القائمة كاملة في كل مرة.
أضف إلى agent.py:
from langchain.messages import AnyMessage
from langgraph.graph import add_messages
from typing_extensions import TypedDict, Annotated
class BookingState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
llm_calls: int
عندنا حقلان فقط. messages هو ذاكرة المحادثة، وllm_calls عدّاد تشخيصي: كل مرة تستدعي فيها عقدة النموذج يزيد واحد، فتنظر لاحقاً وتعرف كم أنفقت على الاستدعاءات فعلاً بدل التخمين.
انتبه إلى فرق واحد عن صفحة البداية السريعة: هناك تُستخدم Annotated[list[AnyMessage], operator.add] — دمج أعمى يُلحق كل ما يُعاد. أنا أستخدم add_messages لأنها تضمّ الرسائل حسب هويتها، وهذا بالضبط ما تحتاجه في مشروع تتكرّر فيه العقد. لو كنت تكتب وكيلاً بلا بوابة موافقة، فـoperator.add تكفيك.
كيف تكتب عقدتَي النموذج والأدوات؟
عقدة النموذج تسأل النموذج ماذا يفعل، وعقدة الأدوات نفّذ ما طلبه. لا أكثر.
from langchain.messages import SystemMessage, ToolMessage
SYSTEM_PROMPT = (
"You are a flight-booking agent. Search before you book. "
"Never invent a flight code: it must come from search_flights. "
"Ask for the passenger's full name if it is missing."
)
def llm_call(state: BookingState):
"""LLM decides whether to call a tool or not"""
return {
"messages": [
model_with_tools.invoke(
[SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]
)
],
"llm_calls": 1 + state.get("llm_calls", 0),
}
def tool_node(state: BookingState):
"""Performs the tool call"""
result = []
for tool_call in state["messages"][-1].tool_calls:
# الاسم `tool` هنا يحجب المُزيّن — لا بأس، فقد استُعمل قبل هذه الحلقة
tool = tools_by_name[tool_call["name"]]
observation = tool.invoke(tool_call["args"])
result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
return {"messages": result}
قارن بين العقدتين: الأولى تُعيد messages وllm_calls، والثانية تُعيد messages فقط. لا تكتب عقدة تُعيد شيئاً لم تعرّفه في BookingState — الخطأ يظهر وقت التشغيل لا وقت التجميع.
messages[-1] في عقدة الأدوات تفترض أن آخر رسالة جاءت من النموذج. إن كتبت يوماً عقدة أدوات تعمل بلا نموذج قبلها، فهذه الافتراض سيوقعك في خطأ عند أول استدعاء.
كيف يقرّر الوكيل أن يتوقف؟
يتوقّف الوكيل حين لا يعود النموذج يطلب أداة، والحكم يقع في دالة الحافة الشرطية should_continue.
from typing import Literal
from langgraph.graph import StateGraph, START, END
def should_continue(state: BookingState) -> Literal["tool_node", END]:
"""Route to the tool node only while the model keeps asking for tools"""
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tool_node"
return END
تلميح النوع Literal["tool_node", END] ليس زينة: هو ما يجعل الرسم يفحص أن كل مسار تعلنه الدالة مسجّل في add_conditional_edges أدناه، فيمسك خطأ بشرياً بدل أن يعلق الرسم بلا سبب.
لماذا لا يُشغَّل الرسم قبل التجميع؟
StateGraph وصفٌ للرسم، وcompile() هو ما يحوّله إلى كائن قابل للاستدعاء. الوكيل الذي ستستدعيه ليس builder بل ناتج compile().
builder = StateGraph(BookingState)
builder.add_node("llm_call", llm_call)
builder.add_node("tool_node", tool_node)
builder.add_edge(START, "llm_call")
builder.add_conditional_edges(
"llm_call",
should_continue,
["tool_node", END]
)
builder.add_edge("tool_node", "llm_call")
agent = builder.compile()
result = agent.invoke({"messages": [{"role": "user", "content": "Find me a flight from RUH to DXB."}]})
print(result["messages"][-1].content)
تلاحظ استدعاء stream_events ليس هنا: سنصل إليه في الخطوة الأخيرة، لأنه يحتاج thread_id ومُحافِظ نقاط، وكلاهما لا وجود لهما الآن.
كيف يحفظ الوكيل حالته بعد إغلاق البرنامج؟
بـSqliteSaver تكتب الحالة في ملف على القرص؛ بالمحافظ في الذاكرة تفقد كل شيء عند إعادة التشغيل.
هذه هي المشكلة التي لا يشرحها أحد: أكثر أمثلة الوكلاء تبدأ بسطر InMemorySaver()، ثم يكتشف المطوّر بعد أسبوع أن محادثته تختفي كلما أوقف البرنامج. الوثائق تسميه بلا تجميل: MemorySaver and InMemorySaver store checkpoints in RAM. When the process restarts, all checkpoints are lost.
استبدل سطر التجميع بما يلي، وأضف thread_id:
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
checkpointer = SqliteSaver(sqlite3.connect("bookings.db", check_same_thread=False))
agent = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "trip-ruh-dxb"}}
استبدل استدعاء agent.invoke في الخطوة السابقة بما يلي، وشغّله مرتين:
agent.invoke({"messages": [{"role": "user", "content": "My name is Sara."}]}, config)
agent.invoke({"messages": [{"role": "user", "content": "What is my name?"}]}, config)
في المرة الثانية ستعرف اسمك. الآن أوقف البرنامج، افتحه من جديد، واطرح السؤال نفسه بنفس thread_id: ما زال يعرف. هذه هي الفارق كله بين InMemorySaver وSqliteSaver — لا تحسّن في الأداء، بل في البقاء.
ثلاث قواعد مرافقة:
thread_idهو مؤشرك الدائم. استخدم نفس القيمة لاستئناف نفس المحادثة، وقيمة جديدة لبدء محادثة فارغة.The thread_id you choose is effectively your persistent cursor.- لا تجعله طويلاً بلا سبب.
Fix: Keep thread_id values under 255 characters. Use a UUID or hash if you need deterministic IDs— خطأ طويل جداً لا يظهر إلا عند الترقية إلىPostgresSaver. - بانتظارك
PostgresSaver، لاSqliteSaver. صفحة مُحافِظات النقاط تصفه بـIdeal for using in production. التخزين المحلي يبقى للتجريب وسير العمل المحلي.
كيف توقف الوكيل قبل أي حجز وتطلب موافقة إنسان؟
بـinterrupt() تُعلّق الحالة قبل تنفيذ الحجز، ثم تستأنف بـCommand(resume=...) عندما يجيب الإنسان.
الآن — وعند هذه النقطة تحديداً — اقرأ القاعدة التالية بتأنٍّ، لأنها تغيّر تصميمك: When execution resumes (after you provide the requested input), the runtime restarts the entire node from the beginning—it does not resume from the exact line where interrupt() was called.
بعبارة أخرى: عند الاستئناف، تُعاد العقدة كلها من أولها. وكل ما سبق interrupt() داخلها يُنفَّذ من جديد.
الآن انظر إلى book_flight التي كتبتها في الخطوة الثانية: هي تكتب في BOOKINGS دون أي بوابة. لو أضفت interrupt() في نهايتها، فسيحجز مرتين: مرة قبل الموافقة، ومرة بعد كل استئناف. هذا بالضبط هو الخطأ الذي يكلّف مالاً حقيقياً.
الحل ليس الحذر، بل التصميم. استبدل الدالة بهذه، وأضف الاستيراد في أعلى الملف:
from langgraph.types import interrupt
@tool
def book_flight(flight_code: str, passenger: str) -> str:
"""Create a booking for one passenger on one flight.
Args:
flight_code: The flight code returned by search_flights
passenger: Full name exactly as it must appear on the ticket
"""
key = f"{flight_code}:{passenger}"
# البوابة أولًا: قبل أي أثر جانبي
decision = interrupt(
{
"question": "Approve this booking?",
"details": {"flight_code": flight_code, "passenger": passenger},
}
)
if decision.get("action") != "approve":
return f"Booking cancelled by the reviewer for {passenger}."
# الأثر الجانبي بعد البوابة، ومفتاحه مشتق من الوسائط فيكون قابلاً للتكرار
existing = BOOKINGS.get(key)
if existing:
return f"Booking {existing} already exists for {passenger}. Nothing new created."
booking_code = f"BK-{len(BOOKINGS) + 1}"
BOOKINGS[key] = booking_code
return f"Booking {booking_code} created for {passenger} on {flight_code}."
لاحظ ثلاث أشياء في الترتيب نفسه:
- البوابة قبل الأثر الجانبي. التوثيق يوصي بذلك صراحةً:
Place side effects after interrupt() calls. - الأثر قابل للتكرار. المفتاح
keyمشتق منflight_codeوpassenger، فإعادة التنفيذ لا تُنتج حجزاً ثانياً لنفس المسافر على نفس الرحلة. التوثيق يصطلح على ذلك:Side effects called before interrupt must be idempotent، ويضرب المثال:This will create duplicate records on each resume. interrupt()ليس داخلtry/except. لأن التعليق ينفَّذ برمي استثناء، وtry/exceptعريض يبتلعه فيختفي التوقّف كأنه لم يحدث.
ونمط خاطئ آخر، للعلم فقط — لا تنسخه:
# ❌ أثر جانبي قبل البوابة: يُنفَّذ من جديد عند كل استئناف
booking_code = f"BK-{len(BOOKINGS) + 1}"
BOOKINGS[key] = booking_code
approved = interrupt({"question": "Approve this booking?"})
ولا تُلِف نداء interrupt() في حلقة while True. التوثيق يحذّر بأن النتيجة هي the first resume replays 1 iteration, the second replays 2, and so on — أي إعادة تنفيذ أُسّية. الطريقة الصحيحة: استدعِ interrupt() مرة واحدة في العقدة، ثم عُد بحافة شرطية.
الآن شغّل البوابة. هذه أول مرة تستعمل فيها stream_events:
from langchain.messages import HumanMessage
from langgraph.types import Command
stream = agent.stream_events(
{"messages": [HumanMessage(content="Book the morning flight from RUH to DXB for Sara.")]},
config=config,
version="v3",
)
for message in stream.messages:
for token in message.text:
print(token, end="", flush=True)
if stream.interrupted:
print("\n---", stream.interrupts)
stream = agent.stream_events(
Command(resume={"action": "approve"}),
config=config,
version="v3",
)
print("\n", stream.output["messages"][-1].content)
stream.interrupted تخبرك إن كان الرسم قد توقّف، وstream.interrupts تحمل ما مرّرته إلى interrupt() — وهو ما تحتاجه لتبني استمارة موافقة فوقها.
وstream.output هي حالة الرسم النهائية. انتبه لقاعدة صارمة: Command(resume=...) is the only Command pattern intended as input. لا تمرّر Command(update=...) كمدخل؛ هذا يربك الرسم ويبدو معلّقاً بلا سبب.
وأخيراً تحذير صدق: BOOKINGS سجل تجريبي في ذاكرة العملية، بينما نقاط حفظ الحالة تعيش في ملف SQLite. أي أن «عدم تكرار الحجز» يصمد أمام إعادة تشغيل العقدة، لكنه لا يصمد أمام إعادة تشغيل البرنامج. في نظام حقيقي اجعل الحجز نفسه في قاعدة بيانات بــupsert على مفتاح ثابت، وعندها تتحول قاعدة «قابلية التكرار» من نصيحة إلى ضمان.
كيف ترسم وكيلك وتشخّصه أثناء التشغيل؟
draw_mermaid_png() تعطيك شكل الرسم، وstream_events تعطيك حركته — بما فيها لحظات التوقّف نفسها.
from IPython.display import Image, display
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))
ستحصل على دائرة بعقدتين وسهمين، بالضبط كما بنيتها. إن كانت الدالة تفشل عندك (بيئة مغلقة أو بلا شبكة)، استخدم draw_mermaid() التي تعيد نص Mermaid الخام وافتحه في أي عارض.
والأدق من الصورة للحالة نفسها:
snapshot = agent.get_state(config)
print(snapshot.values) # كل ما في الحالة الآن
print(snapshot.next) # العقد التالية — () يعني أن الرسم انتهى
You can view the latest state of the graph by calling graph.get_state(config) — وهذه هي الصورة التي ستسألك عنها في أي عطل: أين توقّف، وما الذي ينتظر.
وبثٌّ واحد يخدم الاثنين، فكل إسقاط يُقرأ من المجرى نفسه دون أن يستهلك ما يحتاجه الآخر:
stream = agent.stream_events(
{"messages": [HumanMessage(content="Which flights did you find?")]},
config=config,
version="v3",
)
for snapshot in stream.values:
print(snapshot)
final_state = stream.output
stream.values يمرّ عليك بلقطة الحالة بعد كل خطوة، وstream.output ينتظرها حتى النهاية. وعند المقاطعة نفسها، وجّه البثّ إلى stream.interrupts وstream.interrupted واقرأ الحمولة، ثم استأنف — لا مزاوجة بين أسلوبين.
ما الأسئلة الثمانية التي يجب أن تجيب عنها عن وكيلك؟
إذا لم تستطع الإجابة عن هذه الثمانية من ذهنك وبجملة واحدة، فالوكيل ليس جاهزاً بعد. هذه أسئلة تشخيص لا امتحان.
أسئلة عن الوكيل:
- أين يتوقّف وكيلك الآن، وما الذي ينتظره تحديداً؟ أجب من
agent.get_state(config)وحدها. - كم استدعاءً للنموذج جرت في هذه المحادثة؟ اقرأ
llm_callsمن الحالة، ولا تخمّن. - أي أداة عندك تغيّر المال أو ترسل شيئاً إلى الخارج، وهل تمرّ كل واحدة منها ببوابة موافقة؟
أسئلة عن نفسك:
- هل ينجو وكيلك من إعادة التشغيل؟ جرّب عملياً: أوقف البرنامج، افتحه، وتابع نفس المحادثة بنفس
thread_id. - هل يوجد في كودك أي أثر جانبي قبل
interrupt()؟ إن وُجد، فكم مرة ينفَّذ بعد ثلاثة استئنافات؟ الجواب: مرة لكل استئناف، إن لم يكن قابلاً للتكرار. - هل تستأنف بـ
Command(resume=...)وحدها، أم تمرّرCommand(update=...)كمدخل؟ الثانية تربك الرسم ويبدو معلّقاً بلا سبب. - ما قيمة
thread_idالتي ستختارها لمستخدم حقيقي، وهل تعرف لماذا تختارها قبل أن تختارها فعلاً؟ - لو أخطأ النموذج وطلب حجزاً مكرراً، أي طبقة عندك تمنعه — منطق الوكيل أم منطق عملك داخل الأدوات؟ الإجابة الصحيحة الثانية.
إن كانت إجاباتك عن الأربعة الأخيرة غائبة، فراجع بوابة الموافقة قبل أن تكتب سطراً آخر. هذا أرخص مكان لتكتشف الخطأ.
أين تكمل السلسلة؟
هذا المقال هو نهاية مسار البناء، لا بدايته. إن أردت أن تفهم لماذا وُضع كل سطر أعلاه، ابدأ من الدليل الرئيسي ثم امشِ بالترتيب:
- لماذا تختار لانغ تشين وحدها أم تحتاج لانغ غراف تحتها — القرار الذي يسبق أي كود.
- الواجهة الموحّدة للنماذج — لماذا غيّرت سطراً واحداً فأصبح الوكيل يعمل مع أي مزوّد.
- الأدوات واستدعاؤها — عقد الـdocstring والتسمية، بالتفصيل الذي استعنا إليه هنا.
- الوسيطة و
create_agent— الطريق الأقصر إن كان وكيلك لا يحتاج تحكماً دقيقاً. - الحالة ونقاط الحفظ — دوال الدمج والمخازن، إن أردت ذاكرة أطول من محادثة.
- أنماط سير العمل والمقاطعات — إن أردت بديل
while TrueبنمطSendأوCommand.
وإن أردت أن تأخذ ما بنيته إلى ما بعده، فالخطوة التالية الطبيعية هي: استبدال SqliteSaver بـPostgresSaver، ثم الانتباه إلى أن القنوات الخاصة في الحالة لا تُخفى أثناء البثّ — فحدّد ما يُعرض صراحةً.
تحتاج وكيلاً يعمل لفريقك؟
نصمّم ونبني أنظمة وكلاء للشركات والأفراد فوق لانغ تشين (LangChain) ولانغ غراف (LangGraph) — من إثبات المفهوم في أسبوع إلى نظام إنتاجي يُشغَّل يومياً. ابدأ بطلبك من صفحة التواصل.
المصادر
كل ادعاء في المقال مرتبط بمصدره. الروابط تفتح في نافذة جديدة.
- 01Tools — LangChain ↗
docs.langchain.com
الأداة تقنياً دالة استدعاء لها مدخلات ومخرجات محدّدة تُمرَّر إلى نموذج محادثة، والنموذج هو من يقرّر متى يستدعيها وبأي وسائط.
- 02Tools — LangChain ↗
docs.langchain.com
التلميحات النوعية إلزامية لأنها تعرّف مُخطَّط مدخلات الأداة، ونص التوثيق (docstring) يصير افتراضاً هو وصف الأداة الذي يقرأه النموذج.
- 03Interrupts — LangGraph docs ↗
docs.langchain.com
عند الاستئناف يعيد وقت التشغيل العقدة بأكملها من بدايتها ولا يكمل من سطر interrupt()، ولذلك يجب أن تكون الآثار الجانبية التي تسبقها قابلة للتكرار، وإلا ظهرت سجلات مكرّرة عند كل استئناف.
- 04Checkpointers — LangGraph ↗
docs.langchain.com
مُحافِظ نقاط الحفظ على SQLite يأتي في حزمة منفصلة هي langgraph-checkpoint-sqlite، بينما PostgresSaver هو الخيار الموصوف به بأنه مناسب للإنتاج.
- 05
- 06Tools — LangChain ↗
docs.langchain.com
يجب تفضيل أسماء الأدوات بنمط snake_case لأن بعض مزوّدي النماذج يرفضون الأسماء التي فيها مسافات أو رموز خاصة.
- 07LangGraph — Persistence ↗
docs.langchain.com
يحفظ MemorySaver وInMemorySaver نقاط الحفظ في الذاكرة العشوائية، وتفقد كل النقاط عند إعادة تشغيل العملية؛ لذا صفحة الاستمرارية توجّه إلى PostgresSaver أو SqliteSaver.
- 08LangGraph Quickstart — Use the Graph API ↗
docs.langchain.com
يوثيق لانغ غراف يعرض بناء الوكيل برسم StateGraph ثم تجميعه بـ compile قبل الاستدعاء، مع مثال فوري على عرض المخطط عبر draw_mermaid_png.



