LangGraph: Revolutionizing AI Workflows in Python

Introduction to LangGraph: A New Approach to AI Workflows

In the field of artificial intelligence, managing conversational agents often boils down to simple interactions: asking a question, getting an answer. However, as tasks become more complex, the challenges multiply. An agent may need to access a database, remember past conversations, or provide explanations for its decisions. These complex requirements often necessitate custom solutions, which can be a barrier for many implementations.

LangGraph offers an elegant solution to these problems. It structures AI agents as graphs, where each node represents a unit of work, and the edges determine the order of execution. A shared state object retains the history of messages, allowing each step of the process to be visible and accessible for subsequent nodes. This approach makes the execution flow transparent and inspectable, thereby facilitating the management of AI agents.

In this article, we will explore how LangGraph handles state, node, and edge primitives. We will see how it automatically manages conversation history with MessagesState, calls a language model within a node, registers tools, routes tool calls, and persists conversations with a checkpointer. We will build a graph step by step, starting with the installation of the necessary tools.

Initial Setup

To start using LangGraph, it is essential to install the required packages. Here are the commands to execute:

pip install langgraph langchain-[openai](/dossier/openai) python-dotenv

Once the packages are installed, create a .env file at the root of your project to store your OpenAI API key:

OPENAI_API_KEY="your_key_here"

Load this file at the beginning of your script to set the key as an environment variable, before any imports of LangChain or LangGraph:

from dotenv import load_dotenv
load_dotenv()

The python-dotenv module reads the .env file and sets the key as an environment variable, simplifying the management of sensitive configurations.

Understanding the Components of LangGraph

Each LangGraph graph relies on three fundamental elements: state, nodes, and edges. Understanding these components is crucial to avoid confusion as the graph becomes more complex.

• State: This is a TypedDict that serves as shared memory for the entire graph. Each node can read and write updates to this state. Unmodified fields remain unchanged, and only the fields you wish to modify are returned.

• Nodes: These are standard Python functions that take the current state as an argument and return a dictionary of fields to update. By registering a function with add_node, it becomes an integral part of the graph without requiring a special decorator or base class.

• Edges: They define the order of execution. For example, add_edge(A, B) means that node B executes after node A. Conditional edges allow directing the flow based on the results of nodes.

By default, when a node updates a state field, this update replaces the previous value. For fields requiring accumulation, such as a log or message history, a reduction function is used. For example, operator.add on a list field allows adding elements rather than replacing them.

from typing import Annotated
import operator
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class TicketState(TypedDict):
    customer_message: str
    log: Annotated[list, operator.add]

def log_received(state: TicketState) -> dict:
    return {"log": [f"Received: {state['customer_message']}"]}

def log_assigned(state: TicketState) -> dict:
    return {"log": ["Assigned to support queue"]}

builder = StateGraph(TicketState)
builder.add_node("log_received", log_received)
builder.add_node("log_assigned", log_assigned)
builder.add_edge(START, "log_received")
builder.add_edge("log_received", "log_assigned")
builder.add_edge("log_assigned", END)

graph = builder.compile()
result = graph.invoke({"customer_message": "My invoice looks wrong", "log": []})
print(result)

This code produces a log where each node adds an entry, and the customer's message remains unchanged as no node has modified it. This is how MessagesState manages its message field, using a specialized reducer called add_messages that also ensures deduplication and order of messages.

Managing Conversation History with MessagesState

In a LangGraph graph, each node reads the current state and writes updates. For a conversational agent, the state must include the complete history of messages, encompassing user inputs, model responses, and tool outputs, so that the model always has the necessary context to decide on the next action.

LangGraph offers MessagesState, an integrated state type that uses the add_messages reducer. This reducer adds new messages to the existing list instead of replacing it, thereby simplifying the management of conversation history.

from langgraph.graph import MessagesState

This state definition is sufficient for most single-agent graphs. You can extend it with additional fields, such as a customer_id or a priority indicator, depending on your nodes' needs. However, the messages field is already configured to accumulate automatically.

Integrating the Language Model into a Node

With the state in place, the central node of a LangGraph agent is a function that passes the current list of messages to a language model and adds its response. The model returns an AIMessage, which is then added to the state.

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage

llm = ChatOpenAI(model="[gpt](/glossaire/gpt)-4o-mini")

def run_model(state: MessagesState) -> dict:
    system = SystemMessage("You are a support agent for a SaaS product. Be concise and helpful.")
    response = llm.invoke([system] + state["messages"])
    return {"messages": [response]}

ChatOpenAI encapsulates the OpenAI API with a standard LangChain chat model interface. Switching to another provider, such as Anthropic or Google, simply requires changing the import and model string, without affecting the rest of the node. SystemMessage defines the model's role at each call, without being stored in the state, keeping the history clean.

To integrate this model into a graph and execute it:

from langgraph.graph import StateGraph, START, END
from langchain_core.messages import HumanMessage

builder = StateGraph(MessagesState)
builder.add_node("run_model", run_model)
builder.add_edge(START, "run_model")
builder.add_edge("run_model", END)

graph = builder.compile()
result = graph.invoke({"messages": [HumanMessage("My dashboard isn't loading. What should I try?")]})
print(result["messages"][-1].content)

The result contains the complete list of messages, including the original HumanMessage and the AIMessage generated by the model. The last message is retrieved with [-1].

Recording and Routing Tool Calls

While the model can answer general questions based on its training data, specific information related to your data, such as account details or ticket history, requires tool calls. The model decides when a tool should be called, allowing advanced functionalities to be integrated into the agent's workflow.