Introduction to LangGraph and Stateful Workflows

LangGraph is a low-level orchestration framework developed by the LangChain team for building stateful, multi-agent applications. Unlike traditional linear chains, LangGraph allows you to define cyclical loops, conditional routing, and persistent states. This makes it ideal for complex, autonomous agent behaviors modeled as Finite State Machines (FSM).

At the core of any LangGraph workflow are three foundational concepts:

1. State: The shared memory read from and written to by all steps in the graph.
2. Nodes: Independent Python functions that perform specific operations (such as processing input, querying an LLM, or calling a tool) and return updates to the State.
3. Edges: Connections that define the transition logic from one node to the next.

Before diving into LangGraph, make sure you understand the basics of LangChain. Refer to the LangChain Getting Started guide as a prerequisite.

Defining Graph State

The State is the single source of truth for the entire workflow. It is represented as a Python dictionary schema, typically using TypedDict, which outlines the keys and data types available for nodes to read and modify.

Import the necessary utilities:

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

Now, define a class SimpleState that contains input_text and output_text keys:

class SimpleState(TypedDict):
    input_text: str
    output_text: str

You can initialize and inspect this state schema like any standard dictionary:

SimpleState(input_text="hello", output_text="HELLO")

{'input_text': 'hello', 'output_text': 'HELLO'}

Creating Custom Nodes

Nodes are standard Python functions that execute the business logic of your application. Each node takes the current state as its input and returns a dictionary containing only the keys it wishes to update or overwrite in the shared state.

Let's define the first node, process_input, which converts the input_text string to uppercase and writes it to output_text:

def process_input(state: SimpleState) -> SimpleState:
    output_text = state['input_text'].upper()
    return {'output_text': output_text}

We can test this node function in isolation by passing it a sample state dictionary:

state = {'input_text': 'hello', 'output_text': ''}
process_input(state)

{'output_text': 'HELLO'}

Next, define a second node, add_prefix, which adds an introductory phrase to the output_text:

def add_prefix(state: SimpleState):
    print("Current State [Prefix]", state)
    output = "Hey, i have added something here. " + state['output_text']
    return {'output_text': output}

We can chain these function calls manually to inspect the cumulative changes:

add_prefix(process_input(state))

Current State [Prefix] {'output_text': 'HELLO'}
{'output_text': 'Hey, i have added something here. HELLO'}

Finally, define the third node, add_suffix, which appends an ending phrase to output_text:

def add_suffix(state: SimpleState):
    print("Current State [Suffix]", state)
    output = state["output_text"] + ". i have added suffix!"
    return {'output_text': output}

Test all three nodes sequentially by nesting their calls:

add_suffix(add_prefix(process_input(state)))

Current State [Prefix] {'output_text': 'HELLO'}
Current State [Suffix] {'output_text': 'Hey, i have added something here. HELLO'}
{'output_text': 'Hey, i have added something here. HELLO. i have added suffix!'}

Composing the StateGraph

Instead of nesting function calls manually, we can define a graph structure using StateGraph. This class takes our state schema (SimpleState) as an argument and compiles it into an executable pipeline.

We define the compilation inside a helper function create_simple_graph():

1. Initialize the StateGraph builder with our state schema.
2. Add the custom node functions to the graph using builder.add_node(name, function).
3. Create edges using builder.add_edge(source, destination). The flow starts at the START sentinel, moves through the nodes sequentially, and ends at the END sentinel.
4. Compile the graph builder to create a LangChain-runnable object.

def create_simple_graph():
    builder = StateGraph(SimpleState)

    # Add the nodes
    builder.add_node("process_input", process_input)
    builder.add_node("add_prefix", add_prefix)
    builder.add_node("add_suffix", add_suffix)

    # Connect the edges
    builder.add_edge(START, "process_input")
    builder.add_edge("process_input", "add_prefix")
    builder.add_edge("add_prefix", "add_suffix")
    builder.add_edge("add_suffix", END)

    # Compile the graph builder
    graph = builder.compile()

    return graph

Instantiate the graph object:

graph = create_simple_graph()
graph

<langgraph.graph.state.CompiledStateGraph object at 0x0000018D8F4EC680>

Invoking the Graph

To run the workflow, pass an initial state dictionary containing the required input_text key to the .invoke() method. LangGraph will automatically manage the shared state and pass it along the edges to each node.

initial_state = {'input_text': "hello"}
graph.invoke(initial_state)

Current State [Prefix] {'input_text': 'hello', 'output_text': 'HELLO'}
Current State [Suffix] {'input_text': 'hello', 'output_text': 'Hey, i have added something here. HELLO'}
{'input_text': 'hello',
 'output_text': 'Hey, i have added something here. HELLO. i have added suffix!'}

State Isolation and Persistence

LangGraph isolates keys that are not defined in the state schema. If you pass extra metadata (e.g., "some_other_state"), it will be ignored by the graph runtime and won't pollute the final output:

initial_state = {'input_text': "hello", "some_other_state": 'hi'}
graph.invoke(initial_state)

Current State [Prefix] {'input_text': 'hello', 'output_text': 'HELLO'}
Current State [Suffix] {'input_text': 'hello', 'output_text': 'Hey, i have added something here. HELLO'}
{'input_text': 'hello',
 'output_text': 'Hey, i have added something here. HELLO. i have added suffix!'}

Furthermore, even if you pre-populate a key that is managed by a node (like output_text), the node will overwrite it when it runs:

initial_state = {'input_text': "hello", "output_text": 'hi'}
graph.invoke(initial_state)

Current State [Prefix] {'input_text': 'hello', 'output_text': 'HELLO'}
Current State [Suffix] {'input_text': 'hello', 'output_text': 'Hey, i have added something here. HELLO'}
{'input_text': 'hello',
 'output_text': 'Hey, i have added something here. HELLO. i have added suffix!'}