Building Agentic Workflows in Python with LangGraph

The landscape of artificial intelligence is rapidly evolving, with a particular focus on creating sophisticated AI agents capable of complex reasoning and task execution. A significant advancement in this domain comes from LangGraph, a Python library that provides a structured and intuitive framework for building these agentic workflows. This article delves into the capabilities of LangGraph, illustrating how to construct a comprehensive agent from a simple model call to a fully functional tool-using agent with persistent conversational memory.

The challenges in developing advanced AI agents extend beyond single-turn interactions. While many existing AI setups can effectively handle a user’s query by invoking a model and returning a response, real-world applications often demand more. Agents may need to access external databases, retain the context of ongoing conversations, or provide transparency into their decision-making processes. Overcoming these hurdles without resorting to bespoke, complex plumbing for each new use case is a critical factor in the scalability and effectiveness of AI agent development. LangGraph addresses these complexities by representing an agent as a graph, where distinct units of work are nodes, the flow of execution is dictated by edges, and a shared state object serves as the central repository for all information, including the complete message history. Crucially, model inferences, tool invocations, and agent responses are all integrated into this graph’s state, rendering the entire execution flow transparent, auditable, and accessible to subsequent steps.

This exploration will cover the foundational elements of LangGraph, including the fundamental primitives of state, nodes, and edges. We will examine how to automatically manage conversation history using the MessagesState class, integrate language model calls within graph nodes, and connect these nodes to the broader graph structure. Furthermore, the article will detail the process of registering tools, routing tool calls back to the model for interpretation, tracing the message sequence to understand model behavior at each stage, and implementing persistent conversations across separate invocations using a checkpointer. The development process will commence with the necessary installation steps, building the graph incrementally from the ground up.

Setting Up Your Development Environment

To begin building with LangGraph, a straightforward setup process is required. The essential packages can be installed using pip:

pip install langgraph langchain-openai python-dotenv

Following the installation, it is necessary to configure your OpenAI API key. Create a file named .env in the root directory of your project and add your key as follows:

OPENAI_API_KEY="your_key_here"

This environment variable will be loaded into your script using the python-dotenv library. Ensure this loading step precedes any LangChain or LangGraph imports:

from dotenv import load_dotenv
load_dotenv()

The python-dotenv library facilitates the reading of the .env file, making your OpenAI API key available as an environment variable, which is then accessible by the LangChain and LangGraph libraries.

Understanding the Core Components: State, Nodes, and Edges

At the heart of every LangGraph implementation lie three fundamental components: state, nodes, and edges. A clear understanding of these elements from the outset is crucial for managing the complexity of evolving graphs.

State serves as the shared memory for the entire graph. It is typically defined as a TypedDict, allowing for structured data. Every node within the graph reads from this state and writes its updates back to it. Information does not pass between nodes through any other mechanism. Fields within the state that are not explicitly modified by a node remain unchanged, ensuring that only intended updates are applied.

Nodes are essentially plain Python functions. Each node accepts the current state as input and returns a dictionary containing the fields it intends to modify. By registering these functions with add_node in the StateGraph builder, they become integral parts of the graph without requiring special decorators or base classes. If a function is passed without an explicit name, LangGraph automatically uses the function’s name as its identifier within the graph.

Edges dictate the sequence of execution. A simple add_edge(A, B) command establishes a direct flow: upon completion of node A, node B will execute next. For more dynamic routing, add_conditional_edges allows for conditional execution based on the output of a routing function, enabling the graph to navigate to different paths. Every LangGraph must define a START node as its entry point and include at least one path leading to an END node to signify completion.

By default, when a node returns a value for a state field, this value replaces the existing content. However, for fields that are intended to accumulate data across multiple nodes – such as logs or message histories – a reducer function can be specified. Annotating a state field with a reducer, like operator.add for a list, ensures that new data is appended rather than overwriting existing content.

Consider the following example demonstrating this concept with a TicketState:

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 snippet produces the following output:

'customer_message': 'My invoice looks wrong', 'log': ['Received: My invoice looks wrong', 'Assigned to support queue']

Both the log_received and log_assigned nodes contributed to the log field, and both entries are present in the final state. The customer_message field remained unchanged as neither node returned an update for it. This accumulation mechanism is fundamental to how MessagesState manages its message history, employing a specialized reducer called add_messages that also handles deduplication and ordering.

Managing Conversation History with MessagesState

For conversational agents, maintaining a coherent and complete message history is paramount. The MessagesState class, provided by LangGraph, is specifically designed for this purpose. It is a TypedDict containing a single messages field that automatically utilizes the add_messages reducer. This ensures that every new message, whether it’s a user input, a model response, or a tool output, is appended to the existing history rather than replacing it. This built-in functionality eliminates the need for manual stitching of conversational context.

The MessagesState definition is as follows:

from langgraph.graph import MessagesState

This state definition is sufficient for most single-agent graphs. It can be extended with additional fields as needed, such as customer_id or a priority flag, to accommodate specific application requirements. The messages field, however, is pre-configured to handle the accumulation of conversational data.

Integrating Language Model Calls within Nodes

The core of any LangGraph agent involves a node responsible for interacting with a language model. This node takes the current message list from the state, passes it to the model, and appends the model’s response. The model typically returns an AIMessage, which, when returned within a dictionary keyed to "messages", is automatically added to the graph’s state.

Here’s an example of a node that invokes a language model:

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage

llm = ChatOpenAI(model="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]

The ChatOpenAI class from langchain-openai provides an interface to the OpenAI API. This node can be easily adapted to different LLM providers (e.g., Anthropic, Google, or local models via Ollama) by modifying the import statement and the model string. The SystemMessage is included in each model call without being stored in the persistent history, ensuring a clean and consistent context for the model.

To integrate this node into a graph and execute it, the following structure can be used:

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["messages"] contains the complete sequence of messages, starting with the initial HumanMessage and including the AIMessage generated by the model. Accessing [-1] retrieves the most recent message.

Registering Tools and Routing Tool Calls

While language models can answer general knowledge questions, accessing specific data requires the use of tools. Tools allow agents to interact with external systems, databases, or APIs. The model itself determines when a tool is necessary and what arguments to pass. The developer’s role is to define the available tools and how they should be executed.

Tools are defined using the @tool decorator. The docstring of the tool function is critical, as it provides the model with the necessary information to understand the tool’s purpose and usage.

from langchain_core.tools import tool

@tool
def get_customer_tier(customer_id: str) -> str:
    """Look up the subscription tier for a customer by their ID.
    Returns 'free', 'pro', or 'enterprise'.
    """
    tiers = 
        "cust_1001": "enterprise",
        "cust_2002": "pro",
        "cust_3003": "free",
    
    return tiers.get(customer_id, "not found")

The model uses the tool’s docstring to decide whether to invoke it and with which parameters. Precise and clear docstrings are essential to prevent misinterpretations or incorrect argument assignments by the model.

To enable the model to use this tool, it must be bound to the language model. This is achieved using the bind_tools method. The run_model node is then updated to use the model configured with tools.

tools = [get_customer_tier]
llm_with_tools = llm.bind_tools(tools)

def run_model(state: MessagesState) -> dict:
    system = SystemMessage("You are a support agent for a SaaS product. Use available tools when you need account-specific information.")
    response = llm_with_tools.invoke([system] + state["messages"])
    return "messages": [response]

When the model decides to use a tool, its response will be an AIMessage with a populated tool_calls field, rather than plain text content.

To handle the execution of these tool calls, LangGraph provides a ToolNode. This node is responsible for invoking the correct tool with the arguments provided by the model and appending the result as a ToolMessage back into the state. The tools_condition function is used to route the graph’s execution based on whether the last AIMessage contains tool_calls.

from langgraph.prebuilt import ToolNode, tools_condition

tool_node = ToolNode(tools)

builder = StateGraph(MessagesState)

builder.add_node("run_model", run_model)
builder.add_node("tools", tool_node)

builder.add_edge(START, "run_model")
builder.add_conditional_edges("run_model", tools_condition)
builder.add_edge("tools", "run_model")

graph = builder.compile()

The execution flow now includes a loop: the model calls a tool, the ToolNode executes it, and the result is fed back to the model. This is a representation of the ReAct (Reasoning and Acting) pattern. Each tool use typically incurs two model calls: one to determine which tool to use and with what arguments, and a second one to interpret the tool’s output and formulate a final response. Understanding this two-step process is crucial for managing latency and cost, especially as the number of available tools increases.

Tracing the Reasoning Loop

To gain a deeper insight into the agent’s decision-making process, especially when tools are involved, it’s beneficial to trace the entire message sequence. This reveals the intricate steps of the reasoning loop.

Consider the following invocation and how the messages are processed:

result = graph.invoke(
    "messages": [HumanMessage("Can you check what plan customer cust_1001 is on?")]
)
for msg in result["messages"]:
    print(type(msg).__name__, ":", msg.content or msg.tool_calls)

The sample output demonstrates the flow:

HumanMessage : Can you check what plan customer cust_1001 is on?
AIMessage : ['name': 'get_customer_tier', 'args': 'customer_id': 'cust_1001', 'id': 'call_Rx7kLmNpQ2wJtA3s', 'type': 'tool_call']
ToolMessage : enterprise
AIMessage : Customer cust_1001 is on the enterprise plan.

In this sequence, the initial HumanMessage prompts the first model call. The resulting AIMessage contains tool_calls, indicating the model’s intent to use the get_customer_tier tool. The tools_condition routes execution to the ToolNode, which executes get_customer_tier("cust_1001"). The output, "enterprise," is captured in a ToolMessage. This ToolMessage is then added to the state, and the graph’s execution returns to the run_model node. The model now has the original query, the tool call details, and the tool’s result as context. It uses this information to generate the final AIMessage containing the answer. The tools_condition is checked again, finds no further tool calls, and the graph terminates.

Persisting Conversations Across Invocations

By default, each call to graph.invoke() starts with a fresh graph state, meaning the agent has no memory of previous interactions. To enable persistent conversations, LangGraph offers persistence mechanisms. Attaching a checkpointer when compiling the graph allows it to save and restore the state between invocations.

This is achieved by initializing a checkpointer and passing it during graph compilation:

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

To maintain a conversation thread, the same thread_id must be provided in the config argument for each invocation:

config = "configurable": "thread_id": "ticket-7741"

# First invocation
graph.invoke(
    "messages": [HumanMessage("Hi, I can't access my account.")],
    config,
)

# Second invocation
result = graph.invoke(
    "messages": [HumanMessage("My ID is cust_2002, can you check my plan?")],
    config,
)
print(result["messages"][-1].content)

The sample output from the second invocation demonstrates the continuity:

You're on the pro plan, cust_2002. Since you're having trouble accessing your account, I'd recommend resetting your password first. Pro accounts also have priority support available if the issue continues.

The second invocation successfully leverages the context from the first because the checkpointer restored the thread’s state before execution and saved the updated state afterward. Using a different thread_id would initiate a new, independent conversation.

The InMemorySaver stores checkpoints in process memory, suitable for development and testing. For production environments, it’s recommended to use persistent checkpointers backed by databases or other durable storage solutions. The core graph logic remains unchanged regardless of the underlying persistence mechanism.

Complementary Storage Solutions

Checkpointers are specifically designed to persist graph state for a given thread. However, applications may also require independent storage for data not directly tied to a specific conversation, such as user profiles, preferences, or long-term memories shared across multiple threads. Stores complement checkpointers by providing durable, application-level storage that graphs can access during execution. This distinction is crucial for building comprehensive and scalable agentic applications.

Conclusion

This article has guided you through the construction of a complete LangGraph agent, starting from basic principles and progressing to advanced features like tool integration and persistent memory. We’ve explored how state flows through a graph, how nodes execute tasks, how tools are incorporated into the execution loop, and how checkpointers enable conversations to persist across separate invocations. The foundational building blocks of LangGraph are versatile and scalable, supporting the development of everything from simple chatbots to highly sophisticated agentic workflows.

A significant advantage of LangGraph lies in the modularity of its components. Language models can be swapped, new tools can be registered, and persistence strategies can be altered without necessitating a complete redesign of the graph. Communication occurs exclusively through shared state, ensuring predictable behavior and ease of extension.

These principles also extend seamlessly to the realm of multi-agent systems. A coordinating agent that directs requests to specialized agents can still be represented as a graph with state, nodes, and conditional edges. While the architecture scales in complexity, the underlying primitives remain consistent.

For those seeking to delve deeper into this technology, further exploration of LangGraph’s documentation and related resources is highly recommended. The ability to build dynamic, stateful, and tool-aware AI agents efficiently positions LangGraph as a pivotal tool in the advancement of artificial intelligence applications.