Site Logo
כניסה
  • בלוג
  • חמש דוגמאות LangGraph לקראת הוובינר השבוע

חמש דוגמאות LangGraph לקראת הוובינר השבוע

09/02/2026
יומי

הי חברים מה קורה? השבוע במפגש של מדברים AI אציג את LangGraph ואת הרעיונות המרכזיים בו. פוסט זה מסכם את עיקרי הדברים ואני מקווה שיעזור לכם להתחיל עם LangGraph גם אם לא תגיעו לוובינר (או להיזכר בדברים אם כן תגיעו).

תוכן עניינים

  1. דוגמה 1 - החלק של ה Lang
  2. דוגמה 2 - הגרף
  3. דוגמה 3 - ביצוע פעולות במקביל
  4. דוגמה 4 - קשתות תנאי
  5. דוגמה 5 - שיחות עם AI
  6. מה הלאה?

1. דוגמה 1 - החלק של ה Lang

לנגגרף היא ספריית פייתון ליצירת אוטומציות בעזרת AI. היא מגיעה עם סט כלים עצום לשיחות עם AI ואבסטרקציה של גרף איתה נמדל את תהליך העבודה. החוזקה המרכזית של LangGraph ומה שמבדיל אותה מאינסוף ספריות מתחרות הוא אותה אבסטרקציית גרף והיכולת לטפל בתהליכי אוטומציה מורכבים. בהקשר הזה כדאי להסתכל על לנגגרף בתור גרסת המתכנתים ל n8n. יש לכם תהליך עבודה מורכב שגם משלב פניות למודל שפה? לנגגרף יכולה לתת לכם את הכלים לקודד אותו בצורה טובה.

לנגגרף הוא רק מוצר אחד מבין 3 מוצרים שבנויים על אותה תשתית: LangChain, LangGraph ו Deep Agents SDK. לנגצ'יין אחראי על המבנים הבסיסיים לשיחות עם מודל שפה, לנגגרף מוסיף את האבסטרקציה של הגרף ומידול של תהליכי עבודה ו Deep Agents מוסיף את המבנה של סוכן, חלוקה לתתי סוכנים, יכולות, תכנון מתקדם ושיתוף פעולה בין סוכנים.

אפשר לראות אל כל הדוגמאות לוובינר ולפוסט זה במקום אחד במחברת קולאב בקישור:

https://colab.research.google.com/drive/1471oo_HWx-h6PPRFgY7sMsPTjP1jdWdH?usp=sharing

נתחיל בדוגמה הראשונה שבסך הכל שולחת הודעה ל Gemini:

from google.colab import userdata
from langchain_google_genai import ChatGoogleGenerativeAI

api_key = userdata.get("GEMINI_API_KEY")

llm = ChatGoogleGenerativeAI(
    model= "gemini-2.5-flash",
    temperature=1.0,
    google_api_key=api_key,
)

res=llm.invoke(f"Translate 'Hello' to Spanish")
print(res)

פקודת userdata.get מגיעה מקולאב ולכן הדבר היחיד שקיבלתי מ LangGraph או למעשה מ LangChain הוא ChatGoogleGenerativeAI. אוביקט זה מייצג קשר למודל שפה והפונקציה invoke שלו שולחת הודעה למודל.

בעולם של LangChain יש תמיכה מובנית ב Streaming ולכן הרבה פעמים נראה את הקוד הבא:

from google.colab import userdata
from langchain_google_genai import ChatGoogleGenerativeAI

api_key = userdata.get("GEMINI_API_KEY")

llm = ChatGoogleGenerativeAI(
    model= "gemini-2.5-flash",
    temperature=1.0,
    google_api_key=api_key,
)

query = "Translate 'Hello' to Spanish. Explain different translations in different spanish speaking countries"

for chunk in llm.stream(query):
    # chunk.content contains the text part of the stream
    print(chunk.content, end="", flush=True)

הפונקציה stream במקום invoke מחזירה Generator בו כל chunk הוא הקטע הבא שמודל השפה מחזיר ב Streaming.

2. דוגמה 2 - הגרף

אמרנו ש LangGraph יעיל מאוד כשאנחנו בונים תהליכי עבודה. הסיבה היא שהגיוני לייצג תהליך עבודה בצורת גרף - הצמתים מייצגים פעולות והקשתות מייצגות את התלות בין הפעולות, כלומר קשת בין פעולה A לפעולה B אומרת שפעולה B תוכל להתבצע רק אחרי ש A תסתיים.

נקרא את הקוד הבא שמתאר גרף הפעם עדיין בלי לדבר עם מודלי שפה:

from langgraph.graph import StateGraph, MessagesState, START, END
from pydantic import BaseModel
from random import randint

class GraphState(BaseModel):
    a: int
    b: int
    c: int


def randomize_2_numbers(state: GraphState):
    return {
        "a": randint(0, 100),
        "b": randint(0, 100)
    }

def sum_2_numbers(state: GraphState):
    return {
        "c": state.a + state.b
    }

graph = StateGraph(GraphState)
graph.add_node("A", randomize_2_numbers)
graph.add_node("B", sum_2_numbers)

graph.add_edge(START, "A")
graph.add_edge("A", "B")
graph.add_edge("B", END)

app = graph.compile()
print(app.get_graph().draw_ascii())

הקוד יוצר גרף ובו שני צמתים ו-3 קשתות. הצומת A מבצע את הפעולה שמקודדת בפונקציה randomize_2_numbers. הצומת B מבצע את הפעולה שמקודדת בפונקציה sum_2_numbers. הקשתות מתארות את התהליך, תחילה יופעל A, אחריו יופעל B ואחריו התהליך יסתיים. הערכים START ו END הם ערכים מיוחדים של לנגגרף המייצגים את ההתחלה והסיום של התהליך.

ומה לגבי GraphState? נשים לב שכל פונקציית צומת מקבלת פרמטר מסוג GraphState. זו מחלקה שאני הגדרתי שמייצגת את המידע שהתהליך אוסף או מייצר. לאורך ריצת התהליך האוביקט יתמלא ובסוף נקבל אוביקט שלם עם שני מספרים אקראיים ב a ו b וסכומם ב c. שווה גם לשים לב שכל פונקציית פעולה בצומת לא צריכה למזג בעצמה את השדות לתוך אובייקט ה state שהיא קיבלה אלא רק להחזיר את ההבדל, לכן הפונקציה B יכולה להחזיר רק את הסכום - לנגגרף בעצמו כבר ימזג את הערך שחוזר מ B לאוביקט ה state שרץ לאורך הגרף.

הקוד שהוצג עדיין לא עושה כלום אלא רק מדפיס את הגרף, כן אותו גרף שמייצג את תהליך העבודה. כשמריצים אותו מקבלים את הפלט:

+-----------+  
| __start__ |  
+-----------+  
      *        
      *        
      *        
    +---+      
    | A |      
    +---+      
      *        
      *        
      *        
    +---+      
    | B |      
    +---+      
      *        
      *        
      *        
 +---------+   
 | __end__ |   
 +---------+

בשביל להריץ את תהליך העבודה עלינו להעביר אוביקט סטייט ראשוני לגרף ולתת לו לרוץ בעזרת פונקציה שנקראת invoke:

state = GraphState(a=0, b=0, c=0)
final_state = app.invoke(state)
print(final_state)

התוצאה היא הסטייט אחרי כל התהליך כלומר:

{'a': 37, 'b': 93, 'c': 130}

אגב לנגגרף מאפשר לנו גם להריץ את הגרף בשלבים ולקבל אירוע אחרי כל פעולה בעזרת קוד שנראה כך:

for event in app.stream(state):
    print(event)

ואז התוצאה תהיה:

{'A': {'a': 74, 'b': 78}}
{'B': {'c': 152}}

3. דוגמה 3 - ביצוע פעולות במקביל

ראינו את אבסטרקציית הגרף וכאן כבר המקום לשאול - למה זה טוב? למה אנחנו צריכים גרף פעולות ואת כל הבלאגן הזה? התשובה בגדול היא שמבנה של גרף עם אוביקט סטייט מאוד מקל על קידוד של דברים מסוימים למשל:

  1. קל לבנות תהליך בו חלק מהפעולות קורות אחת אחרי השניה ופעולות אחרות במקביל.
  2. קל להפסיק באמצע את ריצת התהליך, לשמור את הסטייט ואז להמשיך מאותה נקודה יותר מאוחר (למשל אם לא הצלחת לקבל תשובה ממודל שפה כי בדיוק עברת את ה rate limit).
  3. קל לשמור cache של צמתים מסוימים וכך "לדלג" על שלבים כשאנחנו יודעים שהפלט יהיה זהה.
  4. קל לבדוק כל פעולה בנפרד במנותק משאר התהליך.

בדוגמה הבאה נראה רעיון שמאוד קל לממש באמצעות גרף והוא הרצה במקביל של חלק מהפעולות. התהליך שנבנה יהיה:

  1. מקבלים שורת טקסט מהמשתמש.
  2. מתרגמים את הטקסט במקביל לעברית וספרדית בשתי קריאות לשני מודלי שפה - כרגע בשביל הדוגמה הקוד רק ממתין 3 שניות. בדוגמה הבאה נהפוך את זה כבר לפנייה ל AI.
  3. מאחדים את התוצאות לאוביקט סטייט ומסיימים.

הנה הקוד:

from typing import Optional
from langgraph.graph import StateGraph, MessagesState, START, END
from pydantic import BaseModel
import asyncio
from google.colab import userdata
from langchain_google_genai import ChatGoogleGenerativeAI

api_key = userdata.get("GEMINI_API_KEY")

llm = ChatGoogleGenerativeAI(
    model= "gemini-2.5-flash",
    temperature=1.0,
    google_api_key=api_key,
)


class GraphState(BaseModel):
    en: Optional[str] = None
    he: Optional[str] = None
    es: Optional[str] = None

def read_english_text(state: GraphState):
    return {
        "en": input("Type something in English: ")
    }

async def translate_to_hebrew(state: GraphState):
    await asyncio.sleep(3)
    translation = "תרגום לא נמצא"
    return {
        "he": translation
    }

async def translate_to_spanish(state: GraphState):
    await asyncio.sleep(3)
    translation = "traduccion no encontrada"
    return {
        "es": translation
    }

graph = StateGraph(GraphState)
graph.add_node("read_english_text", read_english_text)
graph.add_node("translate_to_hebrew", translate_to_hebrew)
graph.add_node("translate_to_spanish", translate_to_spanish)

graph.add_edge(START, "read_english_text")
graph.add_edge("read_english_text", "translate_to_hebrew")
graph.add_edge("read_english_text", "translate_to_spanish")
graph.add_edge("translate_to_hebrew", END)
graph.add_edge("translate_to_spanish", END)

app = graph.compile()
print(app.get_graph().draw_ascii())

הפעלה שלו כרגע רק מדפיסה את הגרף:

                      +-----------+                         
                      | __start__ |                         
                      +-----------+                         
                             *                              
                             *                              
                             *                              
                  +-------------------+                     
                  | read_english_text |                     
                  +-------------------+                     
                   ***               **                     
                ***                    ***                  
              **                          **                
+---------------------+           +----------------------+  
| translate_to_hebrew |           | translate_to_spanish |  
+---------------------+           +----------------------+  
                   ***               **                     
                      ***         ***                       
                         **     **                          
                       +---------+                          
                       | __end__ |                          
                       +---------+

הרצה מקבילית דורשת הרצה עם async/await באמצעות הקוד הבא:

result = await app.ainvoke({})
print(result)

עכשיו נעבור על הקוד ונראה את הדגשים המרכזיים:

  1. מאוד קל להגדיר פעולה אסינכרונית! במקום להגדיר פעולה בתור פונקציה אני מגדיר אותה בתור פונקציה אסינכרונית. זה הכל.

  2. בבניית הגרף אני מגדיר ביצוע פעולות במקביל באמצעות הוצאת מספר קשתות מאותו צומת.

רוצים לראות איך זה נראה כשפונים למודל שפה בשביל התרגום? אין בעיה, לכל llm בלנגגרף יש פונקציית ainvoke שמחזירה תשובה בצורה אסינכרונית. זה הקוד:

from typing import Optional
from langgraph.graph import StateGraph, MessagesState, START, END
from pydantic import BaseModel
import asyncio
from langchain_google_genai import ChatGoogleGenerativeAI

api_key = userdata.get("GEMINI_API_KEY")

llm = ChatGoogleGenerativeAI(
    model= "gemini-2.5-flash",
    temperature=1.0,
    google_api_key=api_key,
)

class GraphState(BaseModel):
    en: Optional[str] = None
    he: Optional[str] = None
    es: Optional[str] = None

def read_english_text(state: GraphState):
    return {
        "en": input("Type something in English: ")
    }

async def translate_to_hebrew(state: GraphState):
    text_to_translate = state.en

    messages = [
        ("system", "You are a professional translator. Translate the user text to Hebrew only."),
        ("human", text_to_translate)
    ]
    response = await llm.ainvoke(messages)

    return {
        "he": response.content
    }

async def translate_to_spanish(state: GraphState):
    text_to_translate = state.en

    messages = [
        ("system", "You are a professional translator. Translate the user text to Spanish only."),
        ("human", text_to_translate)
    ]
    response = await llm.ainvoke(messages)

    return {
        "es": response.content
    }

graph = StateGraph(GraphState)
graph.add_node("read_english_text", read_english_text)
graph.add_node("translate_to_hebrew", translate_to_hebrew)
graph.add_node("translate_to_spanish", translate_to_spanish)

graph.add_edge(START, "read_english_text")
graph.add_edge("read_english_text", "translate_to_hebrew")
graph.add_edge("read_english_text", "translate_to_spanish")
graph.add_edge("translate_to_hebrew", END)
graph.add_edge("translate_to_spanish", END)

app = graph.compile()

result = await app.ainvoke({})
print(result)

נ.ב. במקרה הזה בעולם האמיתי לא הייתי פונה לשני מודלי שפה כדי לתרגם לשתי שפות אלא למודל שפה אחד עם פרומפט שמבקש לקבל את התרגום לשתי השפות.

4. דוגמה 4 - קשתות תנאי

בואו נקח את דוגמת התרגום עוד צעד קדימה. נניח שאנחנו מזהים שמשתמשים מקלידים לתרגומון מילים מסוימות שאנחנו לא רוצים שיתרגמו ונרצה לסנן את הבקשות האלה לפני שנשלח אותן ל llm. במקרה כזה נוכל להוסיף קשת "תנאי". קשת תנאי היא קשת שמחוברת לפונקציה ואחרי מעבר בה הצומת הבא הוא הצומת שחוזר מהפונקציה. הנה דוגמה לכזו פונקציית ניתוב:

def input_validation(state: GraphState):
  if state.en in blacklisted_words:
    return "blacklist"
  else:
    return "translate"

הפונקציה מנתבת את השליטה לאחד משני צמתי יציאה. אם הטקסט הוא אחד המשפטים במערך blacklisted_words אז נעביר את השליטה לצומת בשם blacklist ואם לא נמשיך את התהליך בצומת בשם translate. צומת blacklist יוכל לנקות את הסטייט וצומת translate יוכל לשלוח את הטקסט לתרגום בשני סוכנים במקביל.

זה תיאור הגרף:

graph = StateGraph(GraphState)
graph.add_node("read_english_text", read_english_text)
graph.add_node("translate_to_hebrew", translate_to_hebrew)
graph.add_node("translate_to_spanish", translate_to_spanish)
graph.add_node("blacklist", blacklist)
graph.add_node("translate", translate)

graph.add_edge(START, "read_english_text")
graph.add_conditional_edges("read_english_text", input_validation)

graph.add_edge("translate", "translate_to_hebrew")
graph.add_edge("translate", "translate_to_spanish")
graph.add_edge("translate_to_hebrew", END)
graph.add_edge("translate_to_spanish", END)
graph.add_edge("blacklist", END)

וקוד התוכנית המלא הוא:

from typing import Optional
from langgraph.graph import StateGraph, MessagesState, START, END
from google.colab import userdata

from pydantic import BaseModel
import asyncio
from langchain_google_genai import ChatGoogleGenerativeAI

api_key = userdata.get("GEMINI_API_KEY")

blacklisted_words = ['stop']

llm = ChatGoogleGenerativeAI(
    model= "gemini-2.5-flash",
    temperature=1.0,
    google_api_key=api_key,
)

class GraphState(BaseModel):
    en: Optional[str] = None
    he: Optional[str] = None
    es: Optional[str] = None

def input_validation(state: GraphState):
  if state.en in blacklisted_words:
    return "blacklist"
  else:
    return "translate"

def translate(state: GraphState):
  return state

def blacklist(state: GraphState):
  return GraphState(en="", he="", es="")

def read_english_text(state: GraphState):
    return {
        "en": input("Type something in English: ")
    }

async def translate_to_hebrew(state: GraphState):
    text_to_translate = state.en

    messages = [
        ("system", "You are a professional translator. Translate the user text to Hebrew only."),
        ("human", text_to_translate)
    ]
    response = await llm.ainvoke(messages)

    return {
        "he": response.content
    }

async def translate_to_spanish(state: GraphState):
    text_to_translate = state.en

    messages = [
        ("system", "You are a professional translator. Translate the user text to Spanish only."),
        ("human", text_to_translate)
    ]
    response = await llm.ainvoke(messages)

    return {
        "es": response.content
    }

graph = StateGraph(GraphState)
graph.add_node("read_english_text", read_english_text)
graph.add_node("translate_to_hebrew", translate_to_hebrew)
graph.add_node("translate_to_spanish", translate_to_spanish)
graph.add_node("blacklist", blacklist)
graph.add_node("translate", translate)

graph.add_edge(START, "read_english_text")
graph.add_conditional_edges("read_english_text", input_validation)

graph.add_edge("translate", "translate_to_hebrew")
graph.add_edge("translate", "translate_to_spanish")
graph.add_edge("translate_to_hebrew", END)
graph.add_edge("translate_to_spanish", END)
graph.add_edge("blacklist", END)

app = graph.compile()

# print(app.get_graph().draw_ascii())
result = await app.ainvoke({})
print(result)

5. דוגמה 5 - שיחות עם AI

לדוגמה האחרונה בואו נבנה סוכן שיחה בסיסי באמצעות LangGraph. סוכן זה יקבל הודעה מהמשתמש, יוסיף אותה למערך הודעות ואז יפנה למודל השפה עם כל היסטוריית ההודעות כדי לקבל השלמה.

מבנה כזה יהיה קצת שונה מהדוגמאות הקודמות שראינו - בכל הדוגמאות הקודמות היה לנו תהליך, קלט נכנס בתחילת התהליך, התקדם לאורך התהליך (למשל עבר תרגום) וסיימנו עם אוביקט מצב מלא עם כל השדות. גם סוכן שיחה משתמש בתהליך אבל זה רק חלק מהתמונה:

  1. התוכנית מגדירה תהליך פשוט בו הודעה מתווספת למערך הודעות, המערך כולו נשלח ל AI ומקבלים חזרה אוביקט מצב עם התשובה של המודל.

  2. התהליך עצמו רק בלולאה, בכל איטרציה אנחנו לוקחים שורת קלט מהמשתמש ומריצים מחדש את התהליך.

גרף של לנגגרף ממדל בצורה טובה תהליך, כלומר משהו שיש לו התחלה, אמצע וסוף. סוכן הוא מנגנון יותר מורכב ואפשר להגיד שהסוכן מפעיל תהליכים. ככל שהסוכן יצטרך להפעיל יותר תהליכים ולפי לוגיקה יותר מתוחכמת נרצה לחשוב על מעבר לספריית סוכנים.

קוד? בטח זאת התוכנית:

from langgraph.graph import StateGraph, MessagesState, START, END
from google.colab import userdata

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import HumanMessage


# --- setup ---
api_key = userdata.get("GEMINI_API_KEY")

llm = ChatGoogleGenerativeAI(
    model="gemini-2.5-flash",
    temperature=1.0,
    google_api_key=api_key,
)

class GraphState(MessagesState):
    pass


# --- graph node ---
async def chat_node(state: GraphState):
    response = await llm.ainvoke(state["messages"])
    return {
        "messages": [response]   # reducer appends
    }


# --- build graph ---
graph = StateGraph(GraphState)
graph.add_node("chat", chat_node)

graph.add_edge(START, "chat")
graph.add_edge("chat", END)

app = graph.compile()


# --- interactive loop (Colab-safe) ---
state: GraphState = {"messages": []}

try:
    while True:
        user_input = input("You: ").strip()

        state = await app.ainvoke(
            {"messages": [HumanMessage(content=user_input)]},
            state,
        )

        print("AI:", state["messages"][-1].content)

except KeyboardInterrupt:
    print("\nExiting chat. Goodbye!")

6. מה הלאה?

רוצים לראות את כל הדוגמאות האלה בוובינר לייב עם כוס קפה? מוזמנים בשמחה, הוובינרים שלנו בחינם לגמרי נפגשים כל יום חמישי בעשר בבוקר וצריך רק להירשם כאן ומקבלים את הלינק לזום אליכם למייל:

https://www.tocode.co.il/talking_ai

וכמובן הדרך הכי טובה ללמוד ספריית תוכנה היא לכתוב באמצעותה קוד. הנה כמה רעיונות לדברים שתוכלו לכתוב עם LangGraph בהתבסס על הדוגמאות כאן ועם קצת עזרה מ AI ומהתיעוד:

  1. תהליך לכתיבה ועריכה של פוסט - פונים למודל שפה כדי לכתוב פוסט, אחרי זה מעבירים את התוצאה למודל נוסף שיערוך ומודל שלישי שיבדוק אם התוצאה מספיק טובה. אם כן אפשר לסיים ואם לא חוזרים לעריכה וממשיכים לשפר.

  2. משחק איקס עיגול נגד מודל שפה - כתבו משחק איקס עיגול שאת מהלכי המחשב מבצע מודל שפה. בכל תור קחו מהלך מהשחקן, סמנו אותו באוביקט ה State ואז עברו למודל השפה כדי שימלא את מהלך המחשב. בשביל המימוש יהיה מעניין להשתמש במנגנון Human In The Loop המובנה ב LangGraph.

  3. סוכן קידוד שבונה דפי נחיתה - כתבו תהליך לבניית דף נחיתה (עמוד html יחיד). התהליך לוקח קלט ממשתמש שמתאר את הדף, פונה למודל שפה כדי להפוך את זה לאפיון מסודר של דף נחיתה ואז את האפיון שולח למודל שפה נוסף כדי לקבל קובץ HTML אותו רושמים בסוף לקובץ.