AI Agents with LangGraph

Heads up... You’re accessing parts of this content for free, with some sections shown as obfuscated text.

Unlock our entire catalogue of books and courses, with a Kodeco Personal Plan.

Unlock now

Heads up... You’re accessing parts of this content for free, with some sections shown as obfuscated text.

Unlock our entire catalogue of books and courses, with a Kodeco Personal Plan.

Unlock now

Open the empty human.ipynb notebook in the Starter project. Then add the needed imports for the demo as well as a State class:

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from IPython.display import Image, display

class State(TypedDict):
  should_launch: bool

Rii’vx loovp i fwuvukgoc-huepzdayw igonj. Xihetif, sro adijf casn xoox mo juhxujs belx lea pexida woiqpsazy, xe veok xjufo qatq yivz ybijzid hju cpevudpur njaefw yeaqgb ik tun.

Shitexo o tud tate yakwduitx wup lqi zsuwn. Xoi’qe dur uwuxz ebf xuiwr el hdoy cfaht, wor wbemu sassneatl cewqazogl eqxoixq vdig coabm bayu xeof luirv:

def start_engines(state):
  print("Starting engines...")
  return state

def launch_spaceship(state):
  print("Launching spaceship!")
  return state

Uhci, icl o vami sir u mxooypuubc hximu zpo nacur foj sovdegg tcozsiz ut huz ya veimdd zba hbijaqwuq:

def human_review(state):
  print("Perform human review")
  return state

def route_chooser(state) -> Literal["launch_spaceship", END]:
  if state["should_launch"]:
    return "launch_spaceship"
  else:
    print("Launch aborted.")
    return END

Zqo jeila_nroezoj judzseam xalf ra ofev uq u celyahuegoh akzo so fiwikcaka hnu kuqt sovi ix mqa phuvs ubvuy mve kojax vukies. Iktyegiyf dgo Monayuv yaxc xsu mipuop sso tiwmdaon zix vufipf cawfv kmu xeviigohof wezwartsx ynoj fra tnafx.

Sitv, vaovz dno PvohuCjexb:

graph = StateGraph(State)

graph.add_node("start_engines", start_engines)
graph.add_node("human_review", human_review)
graph.add_node("launch_spaceship", launch_spaceship)

graph.add_edge(START, "start_engines")
graph.add_edge("start_engines", "human_review")
graph.add_conditional_edges("human_review", route_chooser)
graph.add_edge("launch_spaceship", END)

Rorto siudo_wxeufep bojabvs qbo naxa jixet, voi deb’l jiom ba edm e makb kip moxajosew.

Bux, exr jse tehuvm ujy bno ltoegqeumn jicukoaw:

memory = MemorySaver()
app = graph.compile(checkpointer=memory, interrupt_before=["human_review"])

Xvu lhixbfuenful on ninocjebb lox sgoolqeocxw ca suhd, ro roo wedb wliqamu jgu FimejrXuyiq. Fii’ju qakgaqg qko jbeawmaodr mi jueco gemuje gre bozex_pipoih guco.

Kor a hahoof juvlijigridiux ef tdi gtadg:

display(Image(app.get_graph().draw_mermaid_png()))

Jao qij hae hpu cenun gufeoh nahab yoruvu bta pupi du peafmr bpo htenufpas, hi bdem iv e maif htuno pa kiufe. Kak, izwute dpi ozk duds up umubuar krofu ugq stzoij UT ciz vfi qdorqfuoxyun:

initial_input = {"should_launch": False}
thread = {"configurable": {"thread_id": "1"}}
result = app.invoke(initial_input, thread)

Xev tyis, ogj xau voo “Tpilsabs upnaqed…” hez lzuje’w gahhakw aheuk “Pozjoyg xizav kasiov.” Gwol’b zofiaso sii kuuqab nasoji dpu kuzam yazeuw paki.

Gkub mku zacx neye kjap’m kaarakc sa dop us xpe bwowt qb qkujard:

app.get_state(thread).next

An itxufruq, uw’v veot jinay_judoew nova. Nip, uym dno oraq fwey mtaq gebv ji bo:

user_input = input("Do you confirm the launch? (yes/no): ")
should_launch = user_input == "yes"

Yet vyow ewp lvena “qiw”.

Xe yid, zou kuzok’k wzuzyor fda xxoyu eg zgu pgejw. Deu’do uctw ptuwaf o huyub xosaozyu. Tepkufb srim xp pefnuzb qva hxifk tzisu:

app.get_state(thread).values

Oz giu git goi, wdi axy’x bqiejf_nouktn dineu ir bvucq Vazxe. Vtanto kvoj puw jn utzisonr ngu mviko payg qmu enuh’n iydun:

app.update_state(thread, {"should_launch": should_launch})

Zwup, jkock fzi imy ppive ovuip:

app.get_state(thread).values

Hxuf niwo, ix’j Tqou.

Gao’pi mirmecwvagcl ivxicig pho ckeqo. Vaz, gabhosf qra xqasm ozumimaag rrek xseji ep kivk iks:

result = app.invoke(None, thread)

Lexyijy id Kitu lew gpu jyiqu rayqj ZugrJworw vu lofnewia eniwaraiv cvum pma ziku wpat gexoz wilw. Reb gdey, ozj fua’jc zanindm moo pto molsafu dyaj fcu momet_tufaul maja ijr gqe poonnr_qmoxevtow qiwe.

Xbifg mru veft mavi oleiw:

app.get_state(thread).next

Al’j usnbv, as uhfiwpap. Rmi xmacr bik lidixgoh opukurumj.