Most LLM demos collapse the moment you close the session. A stateful agent LangGraph implementation keeps conversation history, tool results, and intermediate reasoning in a persisted graph that survives process restarts, making it viable for real automation.
Prerequisites
- Python 3.10 or newer (LangGraph relies on modern typing features)
langgraph,langchain-openai, andlangchain-coreinstalled (pip install langgraph langchain-openai)- An OpenAI API key, or any OpenAI-compatible endpoint. Export
OPENAI_API_KEYin your environment. - Familiarity with Python functions and basic async is helpful but not required.
Define the agent state
LangGraph treats state as a typed container passed between nodes. The simplest robust pattern uses a list of messages with a reducer that appends. The reducer runs whenever a node returns a partial state, merging it with existing values.
from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
iteration: intThe add_messages reducer concatenates new messages from a node onto the existing list, which is the core of a stateful agent LangGraph design. The iteration field is a plain integer; nodes must return its new value or it falls back to the schema default.
Build the graph nodes
We need an agent node that calls the model and a tool node that executes functions the model requests.
The agent node
Use ChatOpenAI bound to a tool. The node returns message updates only.
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
@tool
def calculator(expression: str) -> str:
"""Evaluate a basic arithmetic expression."""
try:
return str(eval(expression))
except Exception as e:
return f"Error: {e}"
tools = [calculator]
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
model_with_tools = model.bind_tools(tools)
def agent_node(state: AgentState):
response = model_with_tools.invoke(state["messages"])
return {"messages": [response], "iteration": state["iteration"] + 1}The model returns an AIMessage that may contain tool_calls. We do not execute them here; the graph routes to the tool node.
The tool node
LangGraph provides ToolNode to run tools from the last AI message. It parses tool_calls, invokes the matching function, and appends ToolMessage objects.
from langgraph.prebuilt import ToolNode
tool_node = ToolNode(tools)
def should_continue(state: AgentState):
last = state["messages"][-1]
if hasattr(last, "tool_calls") and last.tool_calls:
return "tools"
return "end"Wire up the stateful agent LangGraph
Create a StateGraph, add nodes, set edges, and compile with an in-memory checkpointer.
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
"agent",
should_continue,
{"tools": "tools", "end": END}
)
workflow.add_edge("tools", "agent")
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)This compiles a stateful agent LangGraph that stores each thread’s state in MemorySaver. The conditional edge sends control back to the agent after tools run, creating a reasoning loop.
Run with checkpointing
Invoke with a thread_id to persist across calls. The checkpointer keys state by that id.
from langchain_core.messages import HumanMessage
config = {"configurable": {"thread_id": "demo-1"}}
result = app.invoke(
{"messages": [HumanMessage(content="What is 12 * 8 + 3?")], "iteration": 0},
config
)
for m in result["messages"]:
print(type(m).__name__, ":", m.content if hasattr(m, "content") else m)Expected output
The first run prints an AI message with a tool call, a ToolMessage with “99”, and a final AI message confirming the result.
AIMessage : I'll calculate that for you.
ToolMessage : 99
AIMessage : 12 * 8 + 3 equals 99.Because the checkpointer is active, a second invocation with the same thread_id retains the prior messages. Send a follow-up:
app.invoke(
{"messages": [HumanMessage(content="Now subtract 10 from that.")]},
config
)The model sees the full history and answers 89 without re-explaining the prior step. The iteration field is now 2.
Persist across processes with SQLite
MemorySaver dies with the process. Swap it for SqliteSaver for durable state.
from langgraph.checkpoint.sqlite import SqliteSaver
with SqliteSaver.from_conn_string("state.db") as saver:
app = workflow.compile(checkpointer=saver)
app.invoke(
{"messages": [HumanMessage(content="What is 5 ** 3?")], "iteration": 0},
{"configurable": {"thread_id": "demo-2"}}
)The state.db file now holds the thread. Restart the script, recompile with the same saver, and resume via the same thread_id. No messages are lost.
Inspect and manipulate state
LangGraph exposes get_state for debugging.
state_snapshot = app.get_state(config)
print("Iteration:", state_snapshot.values["iteration"])
print("Last message:", state_snapshot.values["messages"][-1].content)You can also branch by cloning state with app.get_state(config).next and custom configs, enabling parallel exploration of agent trajectories. This is useful for what-if analysis without mutating the main thread.
Extend state with domain fields
You are not limited to messages. Add a task_status to track progress machine-readably.
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
iteration: int
task_status: str
def agent_node(state: AgentState):
response = model_with_tools.invoke(state["messages"])
status = "completed" if not getattr(response, "tool_calls", None) else "tool_pending"
return {"messages": [response], "iteration": state["iteration"] + 1, "task_status": status}Now the state carries progress that downstream systems can read without parsing LLM output.
Async execution
LangGraph supports ainvoke for non-blocking runs.
import asyncio
async def main():
cfg = {"configurable": {"thread_id": "async-1"}}
await app.ainvoke(
{"messages": [HumanMessage(content="Calculate 2**10")], "iteration": 0},
cfg
)
asyncio.run(main())The checkpointer works identically under async; SqliteSaver uses a sync connection, so prefer the async Postgres saver in high-concurrency servers.
Using an OpenAI-compatible gateway
If you run multiple models or need resilience, point ChatOpenAI at an OpenAI-compatible endpoint. For example, n4n.ai provides one OpenAI-compatible endpoint that addresses 240+ models and automatically falls back when a provider is rate-limited, which keeps a stateful agent LangGraph online during provider outages. Set base_url and api_key accordingly:
model = ChatOpenAI(
model="anthropic/claude-3.5-sonnet",
base_url="https://api.n4n.ai/v1",
api_key="your-key",
temperature=0
)The graph code stays identical; only the transport changes. Provider cache-control hints are forwarded, so you keep observability.
Production considerations
- Keep state schemas explicit. Add fields like
user_idorpending_approvalsas needed. - Trim message history before invoking if context windows are tight; LangGraph’s reducer makes this easy by replacing the list.
- Use
SqliteSaverfor single-node deployments, or a Postgres-backed checkpointer for distributed workers. - Tools should be side-effect free where possible, since the graph may replay them on branch restores.
- Stream tokens with
app.streamto show progress in UIs; the state still updates only on node completion.
A stateful agent LangGraph gives you deterministic control over multi-turn LLM workflows without hand-rolling conversation stores. The checkpointer abstraction is the difference between a toy script and a recoverable service.