Design Transactional Agentic AI with LangGraph
Learn to implement a transactional AI system using LangGraph and two-phase commit.
Overview
In this tutorial, we implement an agentic AI pattern using LangGraph that treats reasoning and action as a transactional workflow rather than a single-shot decision. We model a two-phase commit system where an agent stages reversible changes, validates strict invariants, pauses for human approval via graph interrupts, and commits or rolls back only thereafter.
Safety and Auditability
With this, we demonstrate how agentic systems can be designed with safety, auditability, and controllability at their core, moving beyond reactive chat agents toward structured, governance-aware AI workflows that run reliably in Google Colab using OpenAI models.
Environment Setup
We set up the execution environment as follows:
!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)Ledger Abstraction
Next, we define the core ledger abstraction:
SAMPLE_LEDGER = [
{"txn_id": "T001", "name": "Asha", "email": "ASHA@Example.com", "amount": "1,250.50", "date": "12/01/2025", "note": "Membership renewal"},
{"txn_id": "T002", "name": "Ravi", "email": "ravi@example.com", "amount": "-500", "date": "2025-12-02", "note": "Chargeback?"},
{"txn_id": "T003", "name": "Sara", "email": "sara@example.com", "amount": "700", "date": "02-12-2025", "note": "Late fee waived"},
]
ALLOWED_OPS = {"replace", "remove", "add"}Workflow Modeling
We model the agent’s internal state and define each node in the LangGraph workflow:
class TxnState(TypedDict):
messages: Annotated[List[AnyMessage], add_messages]
raw_rows: List[Dict[str, Any]]
sandbox_rows: List[Dict[str, Any]]
patch: List[Dict[str, Any]]
validation: Dict[str, Any]
approved: Optional[bool]
# Define your nodes
...Running the Agent
Finally, we execute the transactional agent:
def run():
state = {
"messages": [],
"raw_rows": SAMPLE_LEDGER,
"sandbox_rows": [],
"patch": [],
"validation": {},
"approved": None,
}
out = app.invoke(state)
print(out["messages"][-1].content)
run()In conclusion, this tutorial demonstrates how LangGraph enables us to build agents that reason over states, enforce validation gates, and collaborate with humans at precisely defined control points. We treated the agent not as an oracle, but as a transaction coordinator capable of staging, inspecting, and reversing its actions while maintaining a full audit trail.
Сменить язык
Читать эту статью на русском