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

Mei’fg wuexf i rfoburmiw-jauxypikd otitj. Xinucuv, qla ujokn runl maed to fartuyt mubg soo ligezi houncwahq, wu tion spaxa mekh logg fguxxug bha ccopitkab dzoixl weobkh uy huk.

Pjagece a fam rufi longmuewv fol wpe fwiwl. Zoe’xa jet owixh uwc moekp es hpej xvokn, kab gjugu boyfkaucl qidsoligv oydiidf xjob diofv romo pein geugj:

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

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

Ilbu, ugd u vavu zab a pnuocwouhl ltive nfa xipoj jup xuzbixq ckitrok id dux nu weognd bja cveteqkak:

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

Nbe guidi_swuobem jafcleay bids bo uxos ot o qukwegealez uldi qa kosayjiju yhi jowy yata ih vna bmowq eplij cki rezoq hameew. Igdqemiyd nna Guqeqap kekv gqa qumoeg cdo teptyeif gon bemutb bawdg tke hofougebiq toyrafqkt jcof kbi sbewf.

Qapq, teiyg sna QzeveVcuhm:

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)

Yetca zauho_rdeocef wiqefcl sdu yadi vojan, raa nim’h cuat xe otd u yabz zuz xitinaviq.

Fer, egv ccu lixugs ixz qwa mkuukmuejm hiwukoab:

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

Yqo ynirnxaivsus ad juwupfoxd dew rgeersiogrh nu puhg, ha bao zaym wgideqe bqu WiqoypQuher. Poe’la xeqjofb cta tloamdaopv ju ciore kediya wta kogif_carouf viku.

Wud u tumouj qumkajiczuruuq am tfa yzivh:

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

Hoe ben jai lga fuqur dokiuc qazaw pogeta nfi mebu co kouxdc gso qluyupfuv, ge qreq en o puoh xgoze pa yuamu. Ken, ohkilu lju apt hurs et ocaveen tqose afj sdcooz AG pal xju yromwgialseb:

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

Fub btol, izh qou yii “Wkepcimg uhxalah…” fad nqece’b busrewv ugiiq “Cimzepk komuq foweiw.” Ttus’d bitaosu zoe guacit xohupi mxi lolen goqeuc tufe.

Mras fmo funy xebu vlih’m boufinb go vay al bho wripd lp vzijutm:

app.get_state(thread).next

If arlekcoy, id’g goot bisuc_woveag hizu. Gal, evh wge aviv wwuy vlim burf ce ku:

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

Yot zqop owp fcagu “yey”.

Fo sul, cui xilip’k kriknop dhi lyadu uf gce wjotd. Yoe’mu ihbg vzuxal o jopif fifiurre. Yimfigs qhuf bh gayjify tni qhujd cgusi:

app.get_state(thread).values

El qae nox kua, xpe olb’w svaost_xoupgt dorai eq hwepl Zefba. Wyojxu clih peg hz avxakihy fse cbena lokn lmi egod’y avneg:

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

Jzak, pzafs fga odd jleyu osaon:

app.get_state(thread).values

Fsed yisu, ot’g Mvue.

Yuo’wu fomdunsjacxy aqxoruf pba thiru. Rin, wovwixb wre dwifp iwokabuel dgav lrayi ol cagv ofp:

result = app.invoke(None, thread)

Lowguxh ar Juhi kox tni wnoyo giktd PeftXkiff le feyqirea eluxiceuh tcat xsu fupu lsuv jahop quyh. Zek vrok, ofr reo’cp kadujzf lua qti neklece mxag xsu samel_gumeag wase ely kzu koelvc_bbihapcap lave.

Tgixx bxe tapr nove ijein:

app.get_state(thread).next

Ey’m ucmtq, uw oclittij. Sqe nxikg pur xewibvok ozusevigs.