LangGraph مع Ollama

LangGraph يمكّن من بناء سير عمل ذكاء اصطناعي متعدد الخطوات والحالات مع دورات ومنطق شرطي. مع Ollama، يمكنك إنشاء وكلاء ذكاء اصطناعي محليين متطورين.

التثبيت

pip install langgraph langchain-ollama

مفاهيم LangGraph

┌─────────────────────────────────────────────────────────────────┐
│                    المفاهيم الأساسية لـ LangGraph                 │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  الحالة: بيانات مشتركة تستمر عبر العقد                          │
│  العقد: دوال تعالج وتحدث الحالة                                 │
│  الحواف: اتصالات بين العقد (يمكن أن تكون شرطية)                 │
│  الرسم البياني: تعريف سير العمل الكامل                          │
│                                                                 │
│  الميزة الرئيسية: الدورات! العقد يمكنها الرجوع للتكرار          │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

رسم بياني خطي بسيط

from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_ollama import ChatOllama

# حدد الحالة
class State(TypedDict):
    question: str
    answer: str

# هيئ Ollama
llm = ChatOllama(model="llama3.2")

# حدد العقد
def think(state: State) -> State:
    """ولّد إجابة."""
    response = llm.invoke(f"Answer concisely: {state['question']}")
    return {"answer": response.content}

# ابنِ الرسم البياني
graph = StateGraph(State)
graph.add_node("think", think)
graph.set_entry_point("think")
graph.add_edge("think", END)

# جمّع وشغّل
app = graph.compile()
result = app.invoke({"question": "What is Python?"})
print(result["answer"])

التوجيه الشرطي

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_ollama import ChatOllama

class State(TypedDict):
    query: str
    query_type: str
    response: str

llm = ChatOllama(model="llama3.2")

def classify_query(state: State) -> State:
    """صنف نوع الاستعلام."""
    response = llm.invoke(
        f"Classify this query as 'code' or 'general'. "
        f"Reply with just the word.\nQuery: {state['query']}"
    )
    query_type = response.content.strip().lower()
    return {"query_type": query_type}

def handle_code_query(state: State) -> State:
    """تعامل مع أسئلة البرمجة."""
    response = llm.invoke(
        f"As a coding expert, answer: {state['query']}"
    )
    return {"response": response.content}

def handle_general_query(state: State) -> State:
    """تعامل مع الأسئلة العامة."""
    response = llm.invoke(
        f"Answer this general question: {state['query']}"
    )
    return {"response": response.content}

def route_query(state: State) -> Literal["code", "general"]:
    """وجّه بناءً على نوع الاستعلام."""
    if "code" in state["query_type"]:
        return "code"
    return "general"

# ابنِ رسم بياني مع توجيه شرطي
graph = StateGraph(State)

graph.add_node("classify", classify_query)
graph.add_node("code", handle_code_query)
graph.add_node("general", handle_general_query)

graph.set_entry_point("classify")
graph.add_conditional_edges(
    "classify",
    route_query,
    {"code": "code", "general": "general"}
)
graph.add_edge("code", END)
graph.add_edge("general", END)

app = graph.compile()

# اختبر
result = app.invoke({"query": "How do I write a for loop in Python?"})
print(result["response"])

التحسين التكراري (الدورات)

from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_ollama import ChatOllama

class State(TypedDict):
    task: str
    draft: str
    feedback: str
    iteration: int

llm = ChatOllama(model="llama3.2")

def write_draft(state: State) -> State:
    """اكتب أو حسّن مسودة."""
    if state.get("draft"):
        prompt = f"""Improve this draft based on feedback:
Draft: {state['draft']}
Feedback: {state['feedback']}
Write an improved version:"""
    else:
        prompt = f"Write a short paragraph about: {state['task']}"

    response = llm.invoke(prompt)
    return {
        "draft": response.content,
        "iteration": state.get("iteration", 0) + 1
    }

def critique(state: State) -> State:
    """انتقد المسودة الحالية."""
    response = llm.invoke(
        f"Critique this draft briefly. What could be better?\n\n{state['draft']}"
    )
    return {"feedback": response.content}

def should_continue(state: State) -> str:
    """قرر إذا يجب التحسين أكثر."""
    if state["iteration"] >= 3:  # حد أقصى 3 تكرارات
        return "end"
    return "continue"

# ابنِ رسم بياني مع دورة
graph = StateGraph(State)

graph.add_node("write", write_draft)
graph.add_node("critique", critique)

graph.set_entry_point("write")
graph.add_edge("write", "critique")
graph.add_conditional_edges(
    "critique",
    should_continue,
    {"continue": "write", "end": END}
)

app = graph.compile()

result = app.invoke({"task": "Explain machine learning to a child"})
print(f"المسودة النهائية (بعد {result['iteration']} تكرارات):")
print(result["draft"])

نمط وكيل ReAct

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langchain_ollama import ChatOllama
import operator

class AgentState(TypedDict):
    question: str
    thoughts: Annotated[list[str], operator.add]
    answer: str
    step: int

llm = ChatOllama(model="llama3.2")

def think(state: AgentState) -> AgentState:
    """فكر في المشكلة."""
    response = llm.invoke(
        f"Question: {state['question']}\n"
        f"Previous thoughts: {state['thoughts']}\n"
        f"Think step by step. What's your next thought?"
    )
    return {
        "thoughts": [response.content],
        "step": state.get("step", 0) + 1
    }

def decide(state: AgentState) -> AgentState:
    """قرر إذا لدينا ما يكفي للإجابة."""
    response = llm.invoke(
        f"Question: {state['question']}\n"
        f"Thoughts so far: {state['thoughts']}\n"
        f"Can you answer now? Reply YES or NO."
    )
    return {"answer": response.content}

def answer(state: AgentState) -> AgentState:
    """ولّد الإجابة النهائية."""
    response = llm.invoke(
        f"Question: {state['question']}\n"
        f"Based on these thoughts: {state['thoughts']}\n"
        f"Give a clear, final answer:"
    )
    return {"answer": response.content}

def should_answer(state: AgentState) -> str:
    """تحقق إذا جاهز للإجابة."""
    if state["step"] >= 3:  # فرض الإجابة بعد 3 خطوات
        return "answer"
    if "YES" in state.get("answer", "").upper():
        return "answer"
    return "think"

graph = StateGraph(AgentState)

graph.add_node("think", think)
graph.add_node("decide", decide)
graph.add_node("answer", answer)

graph.set_entry_point("think")
graph.add_edge("think", "decide")
graph.add_conditional_edges(
    "decide",
    should_answer,
    {"think": "think", "answer": "answer"}
)
graph.add_edge("answer", END)

app = graph.compile()

result = app.invoke({
    "question": "What's the best way to learn programming?",
    "thoughts": [],
    "step": 0
})
print(f"عملية التفكير: {len(result['thoughts'])} خطوات")
print(f"الإجابة: {result['answer']}")

تصور الرسم البياني

# ولّد تصور الرسم البياني (يتطلب graphviz)
from IPython.display import Image, display

# إذا تستخدم Jupyter
display(Image(app.get_graph().draw_mermaid_png()))

# أو احفظ لملف
with open("graph.png", "wb") as f:
    f.write(app.get_graph().draw_mermaid_png())

ملخص الأنماط الرئيسية

النمط	حالة الاستخدام
خطي	معالجة تسلسلية بسيطة
شرطي	توجيه بناءً على المحتوى/التصنيف
دوري	تحسين تكراري، تصحيح ذاتي
ReAct	حلقات استدلال + عمل

LangGraph يمكّن سير عمل ذكاء اصطناعي معقد ومتعدد الحالات بالكامل على عتاد محلي. في الوحدة التالية، سنستكشف أنماط متقدمة بما في ذلك RAG المحلي واستدعاء الوظائف. :::