بناء وكيل بحث محادثة مدعوم بالذكاء الاصطناعي باستخدام LangGraph: إعادة تشغيل الخطوات ونقاط التفتيش للسفر عبر الزمن
يُقدم هذا البرنامج التعليمي شرحًا عمليًا لكيفية استخدام LangGraph لإدارة تدفق المحادثات بطريقة منظمة، مع إمكانية “السفر عبر الزمن” من خلال نقاط التفتيش. من خلال بناء روبوت محادثة يدمج نموذج Gemini المجاني وأداة ويكيبيديا، يمكننا إضافة خطوات متعددة إلى الحوار، وتسجيل كل نقطة تفتيش، وإعادة تشغيل سجل الحالة الكامل، بل وحتى استئناف المحادثة من حالة سابقة. يُتيح لنا هذا النهج العملي رؤية كيفية تسهيل تصميم LangGraph لتتبع وتلاعب تقدم المحادثة بوضوح وتحكم في الوقت الحقيقي.
الخطوة الأولى: الإعداد والتركيب
نبدأ بتثبيت المكتبات المطلوبة، وإعداد مفتاح API الخاص بـ Gemini، واستيراد جميع الوحدات اللازمة. نقوم بعد ذلك بتهيئة نموذج Gemini باستخدام LangChain لاستخدامه كنموذج لغة أساسي في سير عمل LangGraph.
pip -q install -U langgraph langchain langchain-google-genai google-generativeai typing_extensions
!pip -q install "requests==2.32.4"بعد ذلك، نقوم باستيراد الوحدات الضرورية:
import os
import json
import textwrap
import getpass
import time
from typing import Annotated, List, Dict, Any, Optional
from typing_extensions import TypedDict
from langchain.chat_models import init_chat_model
from langchain_core.messages import BaseMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import ToolNode, tools_condition
import requests
from requests.adapters import HTTPAdapter, Retry
if not os.environ.get("GOOGLE_API_KEY"):
os.environ["GOOGLE_API_KEY"] = getpass.getpass("Enter your Google API Key (Gemini): ")
llm = init_chat_model("google_genai:gemini-2.0-flash")الخطوة الثانية: إنشاء أداة بحث ويكيبيديا
نقوم بإنشاء أداة بحث في ويكيبيديا باستخدام جلسة مخصصة، وإعادة المحاولات، ووكيل مستخدم مهذب. نعرف _wiki_search_raw لاستعلام واجهة برمجة تطبيقات MediaWiki، ثم نلفها كأداة LangChain، مما يسمح لنا باستدعائها بسلاسة داخل سير عمل LangGraph.
WIKI_SEARCH_URL = "https://en.wikipedia.org/w/api.php"
_session = requests.Session()
_session.headers.update({
"User-Agent": "LangGraph-Colab-Demo/1.0 (contact: example@example.com)",
"Accept": "application/json",
})
retry = Retry(
total=5,
connect=5,
read=5,
backoff_factor=0.5,
status_forcelist=(429, 500, 502, 503, 504),
allowed_methods=("GET", "POST")
)
_session.mount("https://", HTTPAdapter(max_retries=retry))
_session.mount("http://", HTTPAdapter(max_retries=retry))
def _wiki_search_raw(query: str, limit: int = 3) -> List[Dict[str, str]]:
# ... (rest of the code remains the same) ...
@tool
def wiki_search(query: str) -> List[Dict[str, str]]:
# ... (rest of the code remains the same) ...
TOOLS = [wiki_search]الخطوة الثالثة: بناء الرسم البياني وتسجيل نقاط التفتيش
نعرف حالة الرسم البياني لتخزين سلسلة الرسائل الجارية، وربط نموذج Gemini بأداة wiki_search، مما يسمح له باستدعائها عند الحاجة. نضيف عقدة روبوت محادثة وعقدة أدوات، ونربطها بحواف شرطية، ونُمَكِّن نقاط التفتيش باستخدام مُخزن مؤقت في الذاكرة. نقوم الآن بتجميع الرسم البياني حتى نتمكن من إضافة خطوات، وإعادة تشغيل السجل، واستئناف المحادثة من أي نقطة تفتيش.
class State(TypedDict):
messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)
llm_with_tools = llm.bind_tools(TOOLS)
SYSTEM_INSTRUCTIONS = textwrap.dedent("""
You are ResearchBuddy, a careful research assistant.
- If the user asks you to "research", "find info", "latest", "web", or references a library/framework/product, you SHOULD call the `wiki_search` tool at least once before finalizing your answer.
- When you call tools, be concise in the text you produce around the call.
- After receiving tool results, cite at least the page titles you used in your summary.
""").strip()
def chatbot(state: State) -> Dict[str, Any]:
# ... (rest of the code remains the same) ...
graph_builder.add_node("chatbot", chatbot)
memory = InMemorySaver()
graph = graph_builder.compile(checkpointer=memory)الخطوة الرابعة: إضافة وظائف مساعدة
نقوم بإضافة وظائف مساعدة لتسهيل فحص سير عمل LangGraph والتحكم فيه. نستخدم print_last_message لعرض الاستجابة الأخيرة بشكل مرتب، و show_state_history لسرد جميع نقاط التفتيش المحفوظة، و pick_checkpoint_by_next لتحديد نقطة تفتيش حيث يكون الرسم البياني على وشك تشغيل عقدة محددة، مثل خطوة الأدوات.
# ... (code for print_last_message, show_state_history, pick_checkpoint_by_next functions remains the same) ...الخطوة الخامسة: التفاعل مع الروبوت وإعادة تشغيل المحادثة
نقوم بمحاكاة تفاعلين للمستخدم في نفس السلسلة من خلال بث الأحداث عبر الرسم البياني. نقدم أولاً تعليمات النظام ونطلب من المساعد البحث عن LangGraph، ثم نتبع ذلك برسالة مستخدم ثانية حول إنشاء وكيل مستقل. يتم وضع نقطة تفتيش لكل خطوة، مما يسمح لنا بإعادة تشغيل المحادثة أو استئنافها من هذه الحالات لاحقًا.
# ... (code for simulating user interactions and streaming events remains the same) ...الخطوة السادسة: إعادة تشغيل المحادثة من نقطة تفتيش سابقة
نقوم بإعادة تشغيل سجل نقاط التفتيش الكامل لنرى كيف تتطور محادثتنا عبر الخطوات، ونحدد نقطة مفيدة للاستئناف. ثم “نسافر عبر الزمن” بإعادة التشغيل من نقطة تفتيش محددة، ومن أي فهرس يدوي اختياري، حتى نستمر في الحوار بالضبط من تلك الحالة المحفوظة.
# ... (code for replaying the conversation from a checkpoint remains the same) ...الخلاصة
لقد حصلنا على صورة أوضح لكيفية توفير إمكانيات نقاط التفتيش والسفر عبر الزمن في LangGraph المرونة والشفافية لإدارة المحادثات. من خلال المرور عبر العديد من أدوار المستخدم، وإعادة تشغيل سجل الحالة، واستئناف المحادثة من نقاط سابقة، يمكننا تجربة قوة هذا الإطار في بناء وكلاء بحث موثوقين أو مساعدين مستقلين. نحن ندرك أن هذا السير العمل ليس مجرد عرض توضيحي، بل هو أساس يمكننا توسيعه إلى تطبيقات أكثر تعقيدًا، حيث تكون الإمكانية للتكرار والتعقب بنفس أهمية الإجابات نفسها.
اترك تعليقاً