بناء وكلاء الذكاء الاصطناعي باستخدام LangGraph: دليل عملي

مقدمة

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

على عكس السلاسل الخطية التقليدية، يتعامل LangGraph مع سير عمل الوكلاء كرسوم بيانية موجّهة، حيث تمثّل العُقد خطوات الحساب وتُعرّف الحواف تدفق التحكم. هذا النهج يجعل من الممكن بناء وكلاء قادرين على التكرار والتفرّع والحفاظ على حالة معقدة عبر التفاعلات.

ما الذي يميّز LangGraph

بُني LangGraph لمعالجة القيود في النهج التقليدية القائمة على السلاسل. تشمل المفاهيم الأساسية ما يلي:

رسوم بيانية ذوو حالة: تُمرَّر الحالة بين العُقد ويمكن تعديلها في كل خطوة
سير عمل دوري: يمكن للوكلاء العودة إلى الخطوات السابقة، مما يتيح التفكير التكراري
حواف شرطية: يمكن أن يتفرّع تدفق التحكم بناءً على الحالة الحالية
الاستمرارية: يمكن حفظ الحالة كنقاط تفتيش، مما يتيح تفاعل الإنسان في الحلقة والتعافي

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

بناء وكيل بحثي

لنبنِ وكيلًا بحثيًا عمليًا قادرًا على البحث في الويب، وتلخيص النتائج، وتقرير ما إذا كانت لديه معلومات كافية للإجابة على سؤال.

إعداد الرسم البياني

أولًا، ثبّت الحزم المطلوبة:

pip install langgraph langchain-openai langchain-community

الآن لنُعرّف الحالة ونبنى الرسم البياني:

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchRun
import operator

class AgentState(TypedDict):
    question: str
    search_results: Annotated[List[str], operator.add]
    summary: str
    iterations: int
    final_answer: str

llm = ChatOpenAI(model="gpt-4o", temperature=0)
search = DuckDuckGoSearchRun()

تعريف العُقد

كل عقدة هي دالة تأخذ الحالة الحالية وتُرجع تحديثًا جزئيًا للحالة:

def search_node(state: AgentState) -> dict:
    """إجراء بحث على الويب بناءً على السؤال."""
    query = state["question"]
    results = search.run(query)
    return {
        "search_results": [results],
        "iterations": state["iterations"] + 1
    }

def summarize_node(state: AgentState) -> dict:
    """تلخيص نتائج البحث."""
    combined = "\n".join(state["search_results"])
    prompt = f"لخّص نتائج البحث هذه للسؤال: {state['question']}\n\nالنتائج:\n{combined}"
    summary = llm.invoke(prompt).content
    return {"summary": summary}

def evaluate_node(state: AgentState) -> dict:
    """تقرير ما إذا كانت لدينا معلومات كافية للإجابة."""
    prompt = f"""بناءً على هذا الملخص، هل يمكنك الإجابة بشكل كامل على السؤال: {state['question']}؟

الملخص: {state['summary']}

إذا نعم، قدّم الإجابة النهائية. إذا لا، قل 'NEED_MORE_INFO'."""
    response = llm.invoke(prompt).content
    if "NEED_MORE_INFO" in response:
        return {"final_answer": ""}
    return {"final_answer": response}

إضافة حواف شرطية

مفتاح السلوك التكراري للوكيل هو التوجيه الشرطي:

def should_continue(state: AgentState) -> str:
    """تقرير ما إذا كان سيتم البحث مرة أخرى أو الانتهاء."""
    if state["final_answer"] and "NEED_MORE_INFO" not in state["final_answer"]:
        return "end"
    if state["iterations"] >= 3:
        return "end"
    return "search"

# بناء الرسم البياني
workflow = StateGraph(AgentState)
workflow.add_node("search", search_node)
workflow.add_node("summarize", summarize_node)
workflow.add_node("evaluate", evaluate_node)

workflow.set_entry_point("search")
workflow.add_edge("search", "summarize")
workflow.add_edge("summarize", "evaluate")
workflow.add_conditional_edges("evaluate", should_continue, {
    "search": "search",
    "end": END
})

app = workflow.compile()

تشغيل الوكيل

result = app.invoke({
    "question": "ما أحدث التطورات في الحوسبة الكمومية؟",
    "search_results": [],
    "summary": "",
    "iterations": 0,
    "final_answer": ""
})

print(result["final_answer"])

استمرارية الحالة وتدخل الإنسان

من أقوى ميزات LangGraph استمرارية الحالة. يمكنك حفظ الحالة كنقاط تفتيش في أي عقدة، مما يتيح:

تدخل الإنسان: إيقاف التنفيذ وطلب موافقة بشرية قبل المتابعة
السفر عبر الزمن: إعادة التشغيل من أي نقطة تفتيش مع حالة معدّلة
التعافي من الأخطاء: الاستئناف من آخر نقطة تفتيش ناجحة بعد الفشل

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

# التشغيل مع معرّف خيط لتتبع الحالة
config = {"configurable": {"thread_id": "thread-1"}}
result = app.invoke(initial_state, config)

متى تستخدم LangGraph مقابل LangChain

يعتمد الاختيار بين LangGraph و LangChain على حالة الاستخدام:

استخدم LangChain عندما:

تحتاج إلى سلاسل خطية بسيطة من العمليات
تبني خطوط أنابيب RAG مباشرة
النماذج الأولية السريعة بالأنماط القياسية
سير العمل لا يتطلب حلقات أو تفرّعًا معقدًا

استخدم LangGraph عندما:

تبني وكلاء يحتاجون إلى تفكير تكراري
سير العمل يتطلب تفرّعًا شرطيًا وحلقات
تحتاج إلى استمرارية الحالة ونقاط التفتيش
تفاعل الإنسان في الحلقة مطلوب
وكلاء متعددون يحتاجون إلى التنسيق ومشاركة الحالة

أفضل الممارسات

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

خاتمة

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