LangGraph

In this section, we will learn about LangGraph, a tool that helps us use large language models (LLMs) together with structured data in an organized way. LangGraph helps us build apps that can think in steps and make better decisions, instead of just giving one simple answer.

What is LangGraph?

LangGraph is a framework that helps developers build AI apps that need to do many tasks, one after another, before giving a final answer.

Meaning: In an AI app, the LLM may need to do different things - like calling a tool, searching the web, checking a database, using memory, or asking another model for help. If we try to write all this using lots of if-else conditions, the code becomes messy, confusing, and hard to manage. LangGraph solves this by letting us organize all these steps as a graph. This makes the app cleaner, easier to understand, and easier to grow.

Docs: LangGraph Documentation

How LangGraph Works

LangGraph lets developers build workflows as a graph (like a flowchart), where nodes are the tasks, and edges show the order in which these tasks should run. This graph style makes it easier to handle multi-step thinking and decision-making, so the LLM can complete tricky, multi-step jobs properly.

Parts of LangGraph:

• Nodes: Each node is one task or step that the LLM needs to do. It could be checking a database, calling an outside API, or working with what the user typed.
• Edges: Edges join the nodes together and decide the order in which the steps should run.
• State: State is like a shared storage box that holds information which can be used by all the nodes in the graph. Every node takes the current state as input, does its job, and then gives back a new state, which the next node can use.

Create State

As we said above, state is like a shared storage box that holds information used across different nodes in the graph. Every node takes the current state as input, does its job, and returns a new state that the next node can use.

State is a way to store information that can be used across different nodes in the graph. Each node takes the current state as input and performs its operation and outputs a new state that can be used by subsequent nodes.

First, install LangGraph:

uv add langgraph

Now let’s create a state object:

from typing_extensions import TypedDict, Annotated
from langgraph.graph.messages import add_message
from langgraph.graph import StateGraph

class MyState(TypedDict):
    message: Annotated[list, add_message]


graphBuilder = StateGraph(MyState)  # builder used to add nodes and edges

Create Nodes

As we said above, each node is one task or step that the LLM needs to do. It could be checking a database, calling an outside API, or working with what the user typed.

Each node takes the current state as input and does its job, then gives back a new state that the next node can use.

A node is just a function that does one specific task. It takes the current state as input and returns a new state that the next node can use.

To create a node, we write a normal function that takes state as its parameter. This function takes the current state, does its job, and returns a new state that later nodes can use. Then we register this node in the graph using the add_node method of the StateGraph class.

# create nodes
def my_node1(state: MyState):
    return {"message": ["Hello from my_node1!"]}

def my_node2(state: MyState):
    return {"message": ["Hello from my_node2!"]}

# register nodes in the graph
graphBuilder.add_node("my_node1", my_node1)
graphBuilder.add_node("my_node2", my_node2)

Note: the second value passed to add_node must be the actual function (like my_node1), not a text string. The first value is just the name we give the node.

It looks like the function is just returning a dict, but this returned value actually gets added to the state object. So this updated state can be used in the next node.

Full Code

from typing_extensions import TypedDict, Annotated
from langgraph.graph.messages import add_message
from langgraph.graph import StateGraph

class MyState(TypedDict):
    message: Annotated[list, add_message]

def my_node1(state: MyState):
    return {"message": ["Hello from my_node1!"]}

def my_node2(state: MyState):
    return {"message": ["Hello from my_node2!"]}

graphBuilder = StateGraph(MyState) # builder used to add nodes and edges

graphBuilder.add_node("my_node1", my_node1)
graphBuilder.add_node("my_node2", my_node2)

Create Edges

As we said above, edges join the nodes together and decide the order in which the steps should run. To create an edge, we use the add_edge method of the StateGraph class.

Edges connect the nodes and decide the order of execution between the different steps in the workflow. To create an edge, we use the add_edge method of the StateGraph class.

from langgraph.graph import START, END

graphBuilder.add_edge(START, "my_node1")
graphBuilder.add_edge("my_node1", "my_node2")
graphBuilder.add_edge("my_node2", END)

graph = graphBuilder.compile()  # builds the runnable graph

Full Code

from typing_extensions import TypedDict, Annotated
from langgraph.graph.messages import add_message
from langgraph.graph import StateGraph, START, END

class MyState(TypedDict):
    message: Annotated[list, add_message]

def my_node1(state: MyState):
    return {"message": ["Hello from my_node1!"]}

def my_node2(state: MyState):
    return {"message": ["Hello from my_node2!"]}

graphBuilder = StateGraph(MyState) # builder used to add nodes and edges

graphBuilder.add_node("my_node1", my_node1)
graphBuilder.add_node("my_node2", my_node2)

graphBuilder.add_edge(START, "my_node1")
graphBuilder.add_edge("my_node1", "my_node2")
graphBuilder.add_edge("my_node2", END)

graph = graphBuilder.compile() # builds the runnable graph

Create Conditional Edges

Normal edges always go from one fixed node to another fixed node. But sometimes we don’t want a fixed path - we want the graph to look at the current state and decide, on its own, which node to go to next. This is what conditional edges are for.

Think of it like a fork in the road. Instead of always turning left, we check something first (like a value in the state) and then decide - go left or go right. In LangGraph, we do this using the add_conditional_edges method of the StateGraph class.

To use a conditional edge, we need:

• A condition function - a normal function that looks at the current state and returns the name of the next node to go to.
• A mapping (optional) - a dictionary that connects what the condition function returns to the actual node names in the graph.

Here is a simple example. Suppose after my_node1, we want to check if the message is happy or sad, and go to a different node depending on that:

def check_mood(state: MyState):
    last_message = state["message"][-1]
    if "sad" in str(last_message).lower():
        return "sad_node"
    return "happy_node"

def happy_node(state: MyState):
    return {"message": ["Glad to hear that!"]}

def sad_node(state: MyState):
    return {"message": ["I'm sorry to hear that."]}

# register the new nodes
graphBuilder.add_node("happy_node", happy_node)
graphBuilder.add_node("sad_node", sad_node)

# add a conditional edge from my_node1
graphBuilder.add_conditional_edges(
    "my_node1",       # from this node
    check_mood,       # this function decides where to go next
    {
        "happy_node": "happy_node",  # if check_mood returns "happy_node", go here
        "sad_node": "sad_node",      # if check_mood returns "sad_node", go here
    }
)

Note: the check_mood function does not update the state - it only looks at the state and returns a plain string telling the graph which node to go to next. The actual state updates still happen inside the nodes (like happy_node and sad_node).

In the full example below, my_node1 picks a random mood (“happy” or “sad”) using Python’s random.choice. This is just to show the branching in action - check_mood always looks at the latest message (the one from my_node1), so whichever mood my_node1 picks is the one that decides which branch runs next.

Full Code

from typing_extensions import TypedDict, Annotated
from langgraph.graph.messages import add_message
from langgraph.graph import StateGraph, START, END
import random

class MyState(TypedDict):
    message: Annotated[list, add_message]

def my_node1(state: MyState):
    random_choice = random.choice(["happy", "sad"])
    return {"message": ["Hello from my_node1! I am feeling " + random_choice]}

def check_mood(state: MyState):
    last_message = state["message"][-1]
    if "sad" in str(last_message).lower():
        return "sad_node"
    return "happy_node"

def happy_node(state: MyState):
    return {"message": ["Glad to hear that!"]}

def sad_node(state: MyState):
    return {"message": ["I'm sorry to hear that."]}

graphBuilder = StateGraph(MyState) # builder used to add nodes and edges

graphBuilder.add_node("my_node1", my_node1)
graphBuilder.add_node("happy_node", happy_node)
graphBuilder.add_node("sad_node", sad_node)

graphBuilder.add_edge(START, "my_node1")

graphBuilder.add_conditional_edges(
    "my_node1",       # from this node
    check_mood,       # this function decides where to go next
    {
        "happy_node": "happy_node",  # if check_mood returns "happy_node", go here
        "sad_node": "sad_node",      # if check_mood returns "sad_node", go here
    }
)

graphBuilder.add_edge("happy_node", END)
graphBuilder.add_edge("sad_node", END)

graph = graphBuilder.compile() # builds the runnable graph

output = graph.invoke({"message": "Hello, this is the starting message"}) # runs the graph, returns final state

print(output)

Run the Graph

To run the graph, we call the graph.invoke() method. This method needs an initial state passed to it as a parameter.

output = graph.invoke({"message": "Some message"})  # runs the graph, returns final state

Note: we just pass a normal dictionary here, like {"message": "Some message"}. There’s no need to wrap it as MyState(...) - MyState is only a type definition, not something we need to call like a function.

Full Code

from typing_extensions import TypedDict, Annotated
from langgraph.graph.messages import add_message
from langgraph.graph import StateGraph, START, END

class MyState(TypedDict):
    message: Annotated[list, add_message]

def my_node1(state: MyState):
    return {"message": ["Hello from my_node1!"]}

def my_node2(state: MyState):
    return {"message": ["Hello from my_node2!"]}

graphBuilder = StateGraph(MyState) # builder used to add nodes and edges

graphBuilder.add_node("my_node1", my_node1)
graphBuilder.add_node("my_node2", my_node2)

graphBuilder.add_edge(START, "my_node1")
graphBuilder.add_edge("my_node1", "my_node2")
graphBuilder.add_edge("my_node2", END)

graph = graphBuilder.compile() # builds the runnable graph

output = graph.invoke({"message": "Some message"}) # runs the graph, returns final state

print(output)

Integration with LLMs

We can connect LangGraph with large language models (LLMs) to build smarter apps. By linking LLMs with structured data and setting up workflows as graphs, we let the LLM do deep, multi-step thinking and decision-making. A simple example is shown below:

from typing_extensions import TypedDict, Annotated
from langgraph.graph.messages import add_message
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model

from dotenv import load_dotenv

load_dotenv()

llm = init_chat_model(
  model="gpt-4",
  model_provider="openai",
)

class State(TypedDict):
    message: Annotated[list, add_message]


def chat_node(state: State):
    response = llm.invoke(state.get("message"))
    return {"message": [response]}

def other_node(state: State):
    return {"message": ["Hello from other_node!"]}

graphBuilder = StateGraph(State)

graphBuilder.add_node("chat_node", chat_node)
graphBuilder.add_node("other_node", other_node)

graphBuilder.add_edge(START, "chat_node")
graphBuilder.add_edge("chat_node", "other_node")
graphBuilder.add_edge("other_node", END)

graph = graphBuilder.compile()

graph_output = graph.invoke({"message": "Hello, how are you?"})

print(graph_output)

Checkpointing

It is important to know that LangGraph does not automatically save the state of the graph after each node runs. If you want to save the state at certain points, you can add checkpointing - this means saving the state to a database or file after specific nodes. This way, you can resume the graph from where it left off, instead of starting over.

Here we will use MongoDB to save the state of the graph after each node runs. This way, we can resume the graph from a saved state whenever we need to.

Pypi: https://pypi.org/project/langgraph-checkpoint-mongodb/

Create a docker-compose.yml file to run a MongoDB container:

# docker-compose.yml
services:
  mongodb:
    image: mongo
    container_name: mongodb
    ports:
      - "27017:27017"
    environment:
      MONGO_INITDB_ROOT_USERNAME: admin
      MONGO_INITDB_ROOT_PASSWORD: password
    volumes:
      - ./.mongoData/db:/data/db

Run the following command to start the MongoDB container:

docker-compose up -d

Install the required packages:

uv add pymongo langgraph langgraph-checkpoint-mongodb

# add this to your existing LangGraph code to turn on checkpointing with MongoDB
from langgraph.checkpoint.mongodb import MongoDBSaver

connection_string = "mongodb://admin:password@localhost:27017"
db_name = "dbname"

with MongoDBSaver.from_conn_string(connection_string, db_name) as checkpointer:
    graph = graphBuilder.compile(checkpointer=checkpointer)
    output = graph.invoke({"message": "Hello, how are you?"})
    print(output)

Note: graph.invoke() must be called inside the with block, not after it. The with block opens the MongoDB connection, and as soon as the block ends, that connection closes. If we call graph.invoke() after the with block ends, LangGraph will try to use a checkpointer whose connection is already closed, which will cause an error. So anything that actually runs the graph needs to stay inside the with block.

conflict

Using just one checkpointer like this can cause a problem, because every user session ends up sharing the same saved state. To fix this, add a thread_id so each session gets its own separate saved state:

config = {"configurable": {"thread_id": "user_session_123"}}

Now, whenever the thread_id changes, LangGraph will save and load the state separately for that session, instead of mixing everyone’s state together.

# updated code with a thread_id, so each session keeps its own separate saved state
from langgraph.checkpoint.mongodb import MongoDBSaver

connection_string = "mongodb://admin:password@localhost:27017"
db_name = "dbname"
config = {"configurable": {"thread_id": "user_session_123"}}

with MongoDBSaver.from_conn_string(connection_string, db_name) as checkpointer:
    graph = graphBuilder.compile(checkpointer=checkpointer)
    # pass config to invoke so this session's state is saved and loaded separately
    graph_output = graph.invoke({"message": "Hello, how are you?"}, config=config)
    print(graph_output)

Note: config also needs to be passed into graph.invoke() - just creating the config dictionary isn’t enough on its own. Passing it to invoke() is what tells LangGraph which thread_id to save and load the state for.

Full Code

from typing_extensions import TypedDict, Annotated
from langgraph.graph.messages import add_message
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.mongodb import MongoDBSaver

from dotenv import load_dotenv

load_dotenv()

llm = init_chat_model(
  model="gpt-4",
  model_provider="openai",
)

class State(TypedDict):
    message: Annotated[list, add_message]

def chat_node(state: State):
    response = llm.invoke(state.get("message"))
    return {"message": [response]}

graphBuilder = StateGraph(State)

graphBuilder.add_node("chat_node", chat_node)

graphBuilder.add_edge(START, "chat_node")
graphBuilder.add_edge("chat_node", END)

connection_string = "mongodb://admin:password@localhost:27017"
db_name = "dbname"
config = {"configurable": {"thread_id": "user_session_123"}}

with MongoDBSaver.from_conn_string(connection_string, db_name) as checkpointer:
    graph = graphBuilder.compile(checkpointer=checkpointer)
    graph_output = graph.invoke({"message": "Hello, how are you?"}, config=config)
    print(graph_output)

Streaming

So far, we have used graph.invoke() to run the graph. But invoke() waits for the whole graph to finish, and only then gives us the final result. This means if the LLM is generating a long answer, we won’t see anything until it is 100% done.

Streaming solves this problem. Instead of waiting for everything to finish, LangGraph can send us updates piece by piece, as soon as they are ready. This is useful for chat apps, where we want to show the reply word-by-word, like typing, instead of showing it all at once after a long wait.

To stream, we use graph.stream() instead of graph.invoke(). It works almost the same way, but instead of returning one final result, it gives us a stream of updates that we can loop over.

for chunk in graph.stream({"message": "Tell me a joke"}):
    print(chunk)

Note: graph.stream() returns a Python generator. This means the code inside the for loop runs again and again, once for every new update that comes in, instead of running just once like invoke() does.

Stream Modes

LangGraph lets us choose what kind of updates we want to see, using the stream_mode parameter. The two most common and beginner-friendly modes are:

• "values" - after every node runs, we get the full state as it stands right now. Good when we always want to see the complete picture.
• "updates" - after every node runs, we only get what that node changed, not the whole state. Good when we only care about what’s new.

# see the full state after every node runs
for chunk in graph.stream({"message": "Tell me a joke"}, stream_mode="values"):
    print(chunk)

# see only what changed after every node runs
for chunk in graph.stream({"message": "Tell me a joke"}, stream_mode="updates"):
    print(chunk)

Streaming Directly from the LLM

The examples above stream node-by-node, meaning we still wait for the whole LLM reply before seeing that node’s update. If we want to see the LLM’s answer being typed out word-by-word (also called token-by-token), we need stream_mode="messages". This mode gives us small pieces of the LLM’s reply as they are generated, instead of waiting for the full reply.

# see the LLM's reply piece by piece, as it is being generated
for chunk, metadata in graph.stream({"message": "Tell me a joke"}, stream_mode="messages"):
    print(chunk.content, end="", flush=True)

Note: with stream_mode="messages", each item in the loop is a pair: the small piece of text (chunk) and some extra information about it (metadata), like which node it came from. That’s why we write for chunk, metadata in ... instead of just for chunk in ....

Full Code

from typing_extensions import TypedDict, Annotated
from langgraph.graph.messages import add_message
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model

from dotenv import load_dotenv

load_dotenv()

llm = init_chat_model(
  model="gpt-4",
  model_provider="openai",
)

class State(TypedDict):
    message: Annotated[list, add_message]

def chat_node(state: State):
    response = llm.invoke(state.get("message"))
    return {"message": [response]}

graphBuilder = StateGraph(State)

graphBuilder.add_node("chat_node", chat_node)

graphBuilder.add_edge(START, "chat_node")
graphBuilder.add_edge("chat_node", END)

graph = graphBuilder.compile()

# stream the LLM's reply piece by piece, as it is being generated
for chunk, metadata in graph.stream({"message": "Tell me a joke"}, stream_mode="messages"):
    print(chunk.content, end="", flush=True)

Full Code (Streaming + MongoDB Checkpointing)

from typing_extensions import TypedDict, Annotated
from langgraph.graph.messages import add_message
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.mongodb import MongoDBSaver

from dotenv import load_dotenv

load_dotenv()

llm = init_chat_model(
  model="gpt-4",
  model_provider="openai",
)

class State(TypedDict):
    message: Annotated[list, add_message]

def chat_node(state: State):
    response = llm.invoke(state.get("message"))
    return {"message": [response]}

graphBuilder = StateGraph(State)

graphBuilder.add_node("chat_node", chat_node)

graphBuilder.add_edge(START, "chat_node")
graphBuilder.add_edge("chat_node", END)

connection_string = "mongodb://admin:password@localhost:27017"
db_name = "dbname"
config = {"configurable": {"thread_id": "user_session_123"}}

with MongoDBSaver.from_conn_string(connection_string, db_name) as checkpointer:
    graph = graphBuilder.compile(checkpointer=checkpointer)

    for chunk, metadata in graph.stream({"message": "Tell me a joke"}, config=config, stream_mode="messages"):
        print(chunk.content, end="", flush=True)