AI Agents with LangGraph

Transcript

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

You’ll build a spaceship-launching agent. However, the agent will need to confirm with you before launching, so your state will mark whether the spaceship should launch or not.

Prepare a few node functions for the graph. You’re not using any tools in this graph, but these functions represent actions that could have been tools:

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

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

Also, add a node for a breakpoint where the human can confirm whether or not to launch the spaceship:

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

The route_chooser function will be used in a conditional edge to determine the next node in the graph after the human review. Including the Literal with the values the function can return helps the visualizer correctly draw the graph.

Next, build the StateGraph:

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)

Since route_chooser returns the node names, you don’t need to add a path map parameter.

Now, add the memory and the breakpoint location:

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

The checkpointer is necessary for breakpoints to work, so you must provide the MemorySaver. You’re setting the breakpoint to pause before the human_review node.

Get a visual representation of the graph:

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

You can see the human review comes before the node to launch the spaceship, so this is a good place to pause. Now, invoke the app with an initial state and thread ID for the checkpointer:

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

Run that, and you see “Starting engines…” but there’s nothing about “Perform human review.” That’s because you paused before the human review node.

Show the next node that’s waiting to run in the graph by writing:

app.get_state(thread).next

As expected, it’s your human_review node. Now, ask the user what they want to do:

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

Run that and write “yes”.

So far, you haven’t changed the state of the graph. You’ve only stored a local variable. Confirm that by getting the graph state:

app.get_state(thread).values

As you can see, the app’s should_launch value is still False. Change that now by updating the state with the user’s input:

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

Then, check the app state again:

app.get_state(thread).values

This time, it’s True.

You’ve successfully updated the state. Now, restart the graph execution from where it left off:

result = app.invoke(None, thread)

Passing in None for the state tells LangGraph to continue execution from the node that comes next. Run that, and you’ll finally see the message from the human_review node and the launch_spaceship node.

Check the next node again:

app.get_state(thread).next

It’s empty, as expected. The graph has finished executing.