Why I Stopped Letting aI Agents Write Directly to my Database (and Built MemState)

Recently, I decided to build a classic AI agent for ordering pizza. The goal was simple: ask for the pizza type, ask for toppings, confirm the order, and save it.

I  used the standard stack: LangChain, LangGraph, and SQLite. \n Here is what my first version looked like:

import sqlite3 from langchain_core.tools import tool from langchain.agents import create_agent from langchain.chat_models import init_chat_model from langgraph.checkpoint.sqlite import SqliteSaver @tool def create_order(pizza_type: str, size: str): """Create a new pizza order.""" # Simulation: Just printing, no real state management here! print(f"Creating order: {size} {pizza_type}") return "Order created." @tool def update_order(new_details: str): """Update existing order.""" return "Order updated." @tool def confirm_order(): """Call this to finalize the order.""" return f"Order sent to kitchen!" llm = init_chat_model(model="gpt-4o", model_provider="openai") agent = create_agent( llm, tools=[create_order, update_order, confirm_order], checkpointer=SqliteSaver(sqlite3.connect("agent.db", check_same_thread=False)), ) config = {"configurable": {"thread_id": "session_1"}} agent.invoke( {"messages": [("user", "I want a large Pepperoni.")]}, config=config )

\ The logic seems fine, right?

1. User says "I want Pepperoni" → Agent calls create_order.
2. Database executes INSERT INTO orders ....
3. User says "No onions please" → Agent calls update_order.
4. Database executes UPDATE orders ....

Then I realized the architecture was broken.

Imagine if the user says on step 3: "Actually, I changed my mind. I don't want pizza, I want sushi."

Now, my production database has a "dirty" record of a Pepperoni order that was never finished. I have to write logic to delete it, handle cancellations, and clean up the garbage.

I was letting the Agent's "thought process", which is chaotic and prone to mistakes, write directly to my production database. This creates Dirty Writes.

Attempt #1: Vector Memory?

Many developers suggest using tools like Mem0 or Zep. But those are for semantic memory. They help the agent remember that "Alice likes spicy food."

They do not solve the transactional state problem. Vectors cannot guarantee that my order ID is unique or that the price is a valid number.

The Solution: MemState (A "Buffer" for Agents)

I needed a middle layer. A "Draft" area where the agent can mold the data like clay, make mistakes, fix them, and only commit when everything is perfect.

I couldn't find a simple tool for this, so I built MemState.

Think of it as Git, but for Agent data:

1. Strict Types (Pydantic): The agent cannot save garbage data.
2. Transactions: Every change is logged. I can rollback if the agent hallucinates.
3. Constraints: I can prevent duplicates automatically.

Here is the new Agent code:

from langchain_core.tools import tool from langchain.agents import create_agent from langchain.chat_models import init_chat_model from pydantic import BaseModel from memstate import MemoryStore, Fact, Constraint, SQLiteStorage from memstate.integrations.langgraph import MemStateCheckpointer class PizzaOrder(BaseModel): status: str = "new" pizza_type: str size: str toppings: list[str] = [] storage = SQLiteStorage("pizza_shop.db") memory = MemoryStore(storage) # 🔥 KILLER FEATURE: Singleton Constraint # A single user can have ONLY ONE active order in a single session. # If the agent attempts to create a second one, MemState will automatically update the first. memory.register_schema("order", PizzaOrder, Constraint(singleton_key="session_id")) checkpointer = MemStateCheckpointer(memory=memory) @tool def update_order(pizza_type: str, size: str, toppings: list[str]): """Call this tool to create or update the pizza order.""" # We use thread_id as a unique key (singleton_key) # In a real application, thread_id is passed through the context (config) session_id = "session_1" # The agent simply "throws" the fact. It doesn't need to check whether the order exists. # MemState will decide for itself: INSERT or UPDATE. fid = memory.commit( Fact( type="order", payload={ "session_id": session_id, "pizza_type": pizza_type, "size": size, "toppings": toppings } ) ) return f"Order state saved. Fact ID: {fid}" @tool def confirm_order(): """Call this to finalize the order.""" orders = memory.query(typename="order", json_filters={"session_id": "session_1"}) return f"Order {orders[0]['payload']} sent to kitchen!" llm = init_chat_model(model="gpt-4o", model_provider="openai") agent = create_agent( llm, tools=[update_order, confirm_order], checkpointer=checkpointer, ) config = {"configurable": {"thread_id": "session_1"}} agent.invoke( {"messages": [("user", "I want a large Pepperoni.")]}, config=config )

Why is this better?

1. The "Draft" Concept: While the user is changing their mind ("add mushrooms", "remove cheese"), we are only updating the local MemState. My main production database stays clean.
2. Validation: If the Agent hallucinates and tries to set the pizza price to "one million", the Pydantic schema in MemState will reject it before it corrupts the state.
3. One Clean Write: When the user finally says "Confirm", I can simply query the final, validated JSON from MemState and do one clean INSERT into my main database.

Summary

MemState turns the chaos of a conversation into a structured transaction. It supports rollback() (Time Travel), LangGraph checkpoints, and runs on SQLite or Redis.

It’s Open Source. I would love your feedback:

https://github.com/scream4ik/MemState