How to Design Transactional Agentic AI Systems with LangGraph Using Two-Phase Commit, Human Interrupts, and Safe Rollbacks

How to Design Transactional Agentic AI Systems with LangGraph Using Two-Phase Commit, Human Interrupts, and Safe Rollbacks

This tutorial implements an agentic AI pattern using LangGraph, treating reasoning and action as a transactional workflow rather than a single-shot decision. The system models a two-phase commit process with reversible changes, invariant validation, human approval, and safe rollbacks, running reliably in Google Colab using OpenAI models.

Why This Matters

Traditional AI agents often operate reactively, making decisions without a clear understanding of potential consequences or a mechanism for recovery. This can lead to unpredictable behavior and costly errors, especially in critical applications like financial transactions or data management. LangGraph enables a more structured approach, mirroring database transaction principles to ensure data integrity and allow for human oversight, reducing the risk of irreversible mistakes and increasing trust in AI-driven workflows.

Key Insights

• Two-Phase Commit: This database concept ensures all parts of a transaction agree before changes are finalized, preventing partial updates and data corruption.
• Agentic AI with State: LangGraph facilitates building agents that maintain and transform internal state, allowing for complex, multi-step reasoning processes.
• Human-in-the-Loop Interrupts: The system pauses for human approval at critical stages, enabling intervention and preventing potentially harmful actions.

Working Example

!pip -q install -U langgraph langchain-openai
import os, json, uuid, copy, math, re, operator
from typing import Any, Dict, List, Optional
from typing_extensions import TypedDict, Annotated
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, AnyMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
def _set_env_openai():
if os.environ.get("OPENAI_API_KEY"):
return
try:
from google.colab import userdata
k = userdata.get("OPENAI_API_KEY")
if k:
os.environ["OPENAI_API_KEY"] = k
return
except Exception:
pass
import getpass
os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter OPENAI_API_KEY: ")
_set_env_openai()
MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")
llm = ChatOpenAI(model=MODEL, temperature=0)

SAMPLE_LEDGER = [
{"txn_id": "T001", "name": "Asha", "email": "[email protected]", "amount": "1,250.50", "date": "12/01/2025", "note": "Membership renewal"},
{"txn_id": "T002", "name": "Ravi", "email": "[email protected]", "amount": "-500", "date": "2025-12-02", "note": "Chargeback?"},
{"txn_id": "T003", "name": "Sara", "email": "[email protected]", "amount": "700", "date": "02-12-2025", "note": "Late fee waived"},
{"txn_id": "T003", "name": "Sara", "email": "[email protected]", "amount": "700", "date": "02-12-2025", "note": "Duplicate row"},
{"txn_id": "T004", "name": "Lee", "email": "[email protected]", "amount": "NaN", "date": "2025/12/03", "note": "Bad amount"},
]
ALLOWED_OPS = {"replace", "remove", "add"}
def _parse_amount(x):
if isinstance(x, (int, float)):
return float(x)
if isinstance(x, str):
try:
return float(x.replace(",", ""))
except:
return None
return None
def _iso_date(d):
if not isinstance(d, str):
return None
d = d.replace("/", "-")
p = d.split("-")
if len(p) == 3 and len(p[0]) == 4:
return d
if len(p) == 3 and len(p[2]) == 4:
return f"{p[2]}-{p[1]}-{p[0]}"
return None
def profile_ledger(rows):
seen, anomalies = {}, []
for i, r in enumerate(rows):
if _parse_amount(r.get("amount")) is None:
anomalies.append(i)
if r.get("txn_id") in seen:
anomalies.append(i)
seen[r.get("txn_id")] = i
return {"rows": len(rows), "anomalies": anomalies}
def apply_patch(rows, patch):
out = copy.deepcopy(rows)
for op in sorted([p for p in patch if p["op"] == "remove"], key=lambda x: x["idx"], reverse=True):
out.pop(op["idx"])
for op in patch:
if op["op"] in {"add", "replace"}:
out[op["idx"]][op["field"]] = op["value"]
return out
def validate(rows):
issues = []
for i, r in enumerate(rows):
if _parse_amount(r.get("amount")) is None:
issues.append(i)
if _iso_date(r.get("date")) is None:
issues.append(i)
return {"ok": len(issues) == 0, "issues": issues}

Practical Applications

• Financial Institutions: Automating transaction monitoring and fraud detection with human oversight for complex cases.
• Data Pipelines: Ensuring data quality and consistency by validating transformations before applying them to production datasets, preventing data corruption.
• Pitfall: Over-reliance on the LLM for critical decisions without proper validation and human review can lead to errors and unintended consequences. Always implement robust validation steps and human-in-the-loop mechanisms.

References:

• https://www.marktechpost.com/2025/12/31/how-to-design-transactional-agentic-ai-systems-with-langgraph-using-two-phase-commit-human-interrupts-and-safe-rollbacks/