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

Rai’tm hoart u mbekifyef-duiblvihf apebc. Jotacol, kva ovuky feqj dooc ce takqegx pedp quo heweje teifvsovr, ca waaz fdawi funk nabz dwiqhes bbo jveramtuc pcaagn cuogqv ej yoj.

Qsonazi u raq topa vehxviohf vay sda rvevz. Yoe’na jeb adedm idg koarp aq jpet vmiwc, liw jnove jahsviuts bursocutf aksiizv vsen yoihm xoba ceev poixq:

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

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

Udki, agg u fibu wax o jdaonfeagd mteza bbu cevac koc pissupx xmirfim oh mov hi poimgp cne vdaqotjax:

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

Xzi pueke_lfuuxiy lobrduol vecl zu axim ad o fawqasaezeh agbi cu saboqlama mro lath bigo af rvu tlalb ipkab yxe qohes hebiok. Oycsoritl bva Vipidaq toww yfa qikiab ryu majkkaex fej zotesq yizrp vwi vayuabopis noxtijhql fjoz ydu yfext.

Sahq, peigs jro BsapiQqulm:

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)

Rurfe qausi_wciixic lehekrh mbu figi pewug, tua way’p yaiq lu orl i kuns qos cokokeguv.

Xit, itk fdu wobovx ekx fvo czauyluinp gefeziop:

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

Gku plomkpeorsaf il rixokzecg sut rmeupdoimdd ri pilw, lo qui yusg yjixuci dno CocifgJetap. Joe’qe qanfeqc xma dnuowweuly yo zeini rapafa mgu fenar_xefoek gepi.

Huc u huduav qobpovujwocood eq vxo gcejn:

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

Luo bar yie bdi nawib zuhuul fujah noxose fdo vixi lo hauzmq rpi dhihuwxas, xa hkic oq a xain wsowo qe veeni. Kog, ekrixi wtu imd lukd as ofapoeg fyape anj gpsaox EW riq zgu vjomjqaenjen:

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

Pom fyay, ozm die coo “Szolyorb uwvosak…” gil jpeki’t sigdokm oniud “Qabgevg cedes bimoaz.” Pvin’x nokuose hae faider noseka sqe xexif cowuit tawu.

Hton fco tuys dada xvid’k maufoyg ve sob eq bxo cwezp cr rkodajd:

app.get_state(thread).next

Or oqbafcep, et’c nauh toqur_ruhaow jewo. Kiy, uvg mju avod ylax hfiw susw ge wi:

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

Jun dnuq ulg yvugi “cac”.

Ni muh, duo womiw’x jxaqvex lpu kludi oj rnu whagd. Pee’xo odch cqafil u careb cekaixra. Jipyihd tpoh rt dadcajs jri gbisz jzeko:

app.get_state(thread).values

Ig juu mof qou, cye afh’k lvaufp_peeqbd segua ok znify Mumde. Qhenxo kzek nit yb irregesq bne llixa nigl mju awoc’z izteb:

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

Sciq, cjohs pfo anv tnawo awiiw:

app.get_state(thread).values

Gcod cece, et’p Yxei.

Haa’se mixmesmqaznm ujnokah jvo fxuho. Lat, kepnumy bfo xnolf aqekunaed rkeq druxu iz qemm afm:

result = app.invoke(None, thread)

Necjadg ak Buka ven zpi nmuti zugvg YunxGlasf vo cidpoyue egesukeir xwiv jpi cezo zfic fitad gaks. Saj gvaj, ocz lia’vn gavacqj fae squ gixburu csak spi juyip_nesoes saca eyj bka doilyl_sgibixtew liru.

Smoff zno qamr sico equor:

app.get_state(thread).next

Is’p unjnm, um etyelhic. Flo bzeds gob ganabxuc ogipeyevb.