{"id":214,"date":"2025-12-31T23:16:09","date_gmt":"2025-12-31T15:16:09","guid":{"rendered":"https:\/\/connectword.dpdns.org\/?p=214"},"modified":"2025-12-31T23:16:09","modified_gmt":"2025-12-31T15:16:09","slug":"how-to-design-transactional-agentic-ai-systems-with-langgraph-using-two-phase-commit-human-interrupts-and-safe-rollbacks","status":"publish","type":"post","link":"https:\/\/connectword.dpdns.org\/?p=214","title":{"rendered":"How to Design Transactional Agentic AI Systems with LangGraph Using Two-Phase Commit, Human Interrupts, and Safe Rollbacks"},"content":{"rendered":"

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 in which an agent stages reversible changes, validates strict invariants, pauses for human approval via graph interrupts, and commits or rolls back only then. 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. Check out the\u00a0Full Codes here<\/a><\/strong>.<\/p>\n

\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
!pip -q install -U langgraph langchain-openai\n\n\nimport os, json, uuid, copy, math, re, operator\nfrom typing import Any, Dict, List, Optional\nfrom typing_extensions import TypedDict, Annotated\n\n\nfrom langchain_openai import ChatOpenAI\nfrom langchain_core.messages import SystemMessage, HumanMessage, AIMessage, AnyMessage\nfrom langgraph.graph import StateGraph, START, END\nfrom langgraph.graph.message import add_messages\nfrom langgraph.checkpoint.memory import InMemorySaver\nfrom langgraph.types import interrupt, Command\n\n\ndef _set_env_openai():\n   if os.environ.get(\"OPENAI_API_KEY\"):\n       return\n   try:\n       from google.colab import userdata\n       k = userdata.get(\"OPENAI_API_KEY\")\n       if k:\n           os.environ[\"OPENAI_API_KEY\"] = k\n           return\n   except Exception:\n       pass\n   import getpass\n   os.environ[\"OPENAI_API_KEY\"] = getpass.getpass(\"Enter OPENAI_API_KEY: \")\n\n\n_set_env_openai()\n\n\nMODEL = os.environ.get(\"OPENAI_MODEL\", \"gpt-4o-mini\")\nllm = ChatOpenAI(model=MODEL, temperature=0)<\/code><\/pre>\n<\/div>\n<\/div>\n
We set up the execution environment by installing LangGraph and initializing the OpenAI model. We securely load the API key and configure a deterministic LLM, ensuring that all downstream agent behavior remains reproducible and controlled. Check out the\u00a0Full Codes here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
SAMPLE_LEDGER = [\n   {\"txn_id\": \"T001\", \"name\": \"Asha\", \"email\": \"ASHA@Example.com\", \"amount\": \"1,250.50\", \"date\": \"12\/01\/2025\", \"note\": \"Membership renewal\"},\n   {\"txn_id\": \"T002\", \"name\": \"Ravi\", \"email\": \"ravi@example.com\", \"amount\": \"-500\", \"date\": \"2025-12-02\", \"note\": \"Chargeback?\"},\n   {\"txn_id\": \"T003\", \"name\": \"Sara\", \"email\": \"sara@example.com\", \"amount\": \"700\", \"date\": \"02-12-2025\", \"note\": \"Late fee waived\"},\n   {\"txn_id\": \"T003\", \"name\": \"Sara\", \"email\": \"sara@example.com\", \"amount\": \"700\", \"date\": \"02-12-2025\", \"note\": \"Duplicate row\"},\n   {\"txn_id\": \"T004\", \"name\": \"Lee\", \"email\": \"lee@example.com\", \"amount\": \"NaN\", \"date\": \"2025\/12\/03\", \"note\": \"Bad amount\"},\n]\n\n\nALLOWED_OPS = {\"replace\", \"remove\", \"add\"}\n\n\ndef _parse_amount(x):\n   if isinstance(x, (int, float)):\n       return float(x)\n   if isinstance(x, str):\n       try:\n           return float(x.replace(\",\", \"\"))\n       except:\n           return None\n   return None\n\n\ndef _iso_date(d):\n   if not isinstance(d, str):\n       return None\n   d = d.replace(\"\/\", \"-\")\n   p = d.split(\"-\")\n   if len(p) == 3 and len(p[0]) == 4:\n       return d\n   if len(p) == 3 and len(p[2]) == 4:\n       return f\"{p[2]}-{p[1]}-{p[0]}\"\n   return None\n\n\ndef profile_ledger(rows):\n   seen, anomalies = {}, []\n   for i, r in enumerate(rows):\n       if _parse_amount(r.get(\"amount\")) is None:\n           anomalies.append(i)\n       if r.get(\"txn_id\") in seen:\n           anomalies.append(i)\n       seen[r.get(\"txn_id\")] = i\n   return {\"rows\": len(rows), \"anomalies\": anomalies}\n\n\ndef apply_patch(rows, patch):\n   out = copy.deepcopy(rows)\n   for op in sorted([p for p in patch if p[\"op\"] == \"remove\"], key=lambda x: x[\"idx\"], reverse=True):\n       out.pop(op[\"idx\"])\n   for op in patch:\n       if op[\"op\"] in {\"add\", \"replace\"}:\n           out[op[\"idx\"]][op[\"field\"]] = op[\"value\"]\n   return out\n\n\ndef validate(rows):\n   issues = []\n   for i, r in enumerate(rows):\n       if _parse_amount(r.get(\"amount\")) is None:\n           issues.append(i)\n       if _iso_date(r.get(\"date\")) is None:\n           issues.append(i)\n   return {\"ok\": len(issues) == 0, \"issues\": issues}<\/code><\/pre>\n<\/div>\n<\/div>\n
We define the core ledger abstraction along with the patching, normalization, and validation logic. We treat data transformations as reversible operations, allowing the agent to reason about changes safely before committing them. Check out the\u00a0Full Codes here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
class TxnState(TypedDict):\n   messages: Annotated[List[AnyMessage], add_messages]\n   raw_rows: List[Dict[str, Any]]\n   sandbox_rows: List[Dict[str, Any]]\n   patch: List[Dict[str, Any]]\n   validation: Dict[str, Any]\n   approved: Optional[bool]\n\n\ndef node_profile(state):\n   p = profile_ledger(state[\"raw_rows\"])\n   return {\"messages\": [AIMessage(content=json.dumps(p))]}\n\n\ndef node_patch(state):\n   sys = SystemMessage(content=\"Return a JSON patch list fixing amounts, dates, emails, duplicates\")\n   usr = HumanMessage(content=json.dumps(state[\"raw_rows\"]))\n   r = llm.invoke([sys, usr])\n   patch = json.loads(re.search(r\"[.*]\", r.content, re.S).group())\n   return {\"patch\": patch, \"messages\": [AIMessage(content=json.dumps(patch))]}\n\n\ndef node_apply(state):\n   return {\"sandbox_rows\": apply_patch(state[\"raw_rows\"], state[\"patch\"])}\n\n\ndef node_validate(state):\n   v = validate(state[\"sandbox_rows\"])\n   return {\"validation\": v, \"messages\": [AIMessage(content=json.dumps(v))]}\n\n\ndef node_approve(state):\n   decision = interrupt({\"validation\": state[\"validation\"]})\n   return {\"approved\": decision == \"approve\"}\n\n\ndef node_commit(state):\n   return {\"messages\": [AIMessage(content=\"COMMITTED\")]}\n\n\ndef node_rollback(state):\n   return {\"messages\": [AIMessage(content=\"ROLLED BACK\")]}<\/code><\/pre>\n<\/div>\n<\/div>\n
We model the agent\u2019s internal state and define each node in the LangGraph workflow. We express agent behavior as discrete, inspectable steps that transform state while preserving message history. Check out the\u00a0Full Codes here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
builder = StateGraph(TxnState)\n\n\nbuilder.add_node(\"profile\", node_profile)\nbuilder.add_node(\"patch\", node_patch)\nbuilder.add_node(\"apply\", node_apply)\nbuilder.add_node(\"validate\", node_validate)\nbuilder.add_node(\"approve\", node_approve)\nbuilder.add_node(\"commit\", node_commit)\nbuilder.add_node(\"rollback\", node_rollback)\n\n\nbuilder.add_edge(START, \"profile\")\nbuilder.add_edge(\"profile\", \"patch\")\nbuilder.add_edge(\"patch\", \"apply\")\nbuilder.add_edge(\"apply\", \"validate\")\n\n\nbuilder.add_conditional_edges(\n   \"validate\",\n   lambda s: \"approve\" if s[\"validation\"][\"ok\"] else \"rollback\",\n   {\"approve\": \"approve\", \"rollback\": \"rollback\"}\n)\n\n\nbuilder.add_conditional_edges(\n   \"approve\",\n   lambda s: \"commit\" if s[\"approved\"] else \"rollback\",\n   {\"commit\": \"commit\", \"rollback\": \"rollback\"}\n)\n\n\nbuilder.add_edge(\"commit\", END)\nbuilder.add_edge(\"rollback\", END)\n\n\napp = builder.compile(checkpointer=InMemorySaver())<\/code><\/pre>\n<\/div>\n<\/div>\n
We construct the LangGraph state machine and explicitly encode the control flow between profiling, patching, validation, approval, and finalization. We use conditional edges to enforce governance rules rather than rely on implicit model decisions. Check out the\u00a0Full Codes here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
def run():\n   state = {\n       \"messages\": [],\n       \"raw_rows\": SAMPLE_LEDGER,\n       \"sandbox_rows\": [],\n       \"patch\": [],\n       \"validation\": {},\n       \"approved\": None,\n   }\n\n\n   cfg = {\"configurable\": {\"thread_id\": \"txn-demo\"}}\n   out = app.invoke(state, config=cfg)\n\n\n   if \"__interrupt__\" in out:\n       print(json.dumps(out[\"__interrupt__\"], indent=2))\n       decision = input(\"approve \/ reject: \").strip()\n       out = app.invoke(Command(resume=decision), config=cfg)\n\n\n   print(out[\"messages\"][-1].content)\n\n\nrun()<\/code><\/pre>\n<\/div>\n<\/div>\n
We run the transactional agent and handle human-in-the-loop approval through graph interrupts. We resume execution deterministically, demonstrating how agentic workflows can pause, accept external input, and safely conclude with either a commit or rollback.<\/p>\n
In conclusion, we showed 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 that can stage, inspect, and reverse its own actions while maintaining a full audit trail. This approach highlights how agentic AI can be applied to real-world systems that require trust, compliance, and recoverability, and it provides a practical foundation for building production-grade autonomous workflows that remain safe, transparent, and human-supervised.<\/p>\n
\n
Check out the\u00a0Full Codes here<\/a><\/strong>.\u00a0Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0100k+ ML SubReddit<\/a><\/strong>\u00a0and Subscribe to\u00a0our Newsletter<\/a><\/strong>. Wait! are you on telegram?\u00a0now you can join us on telegram as well.<\/a><\/strong><\/p>\n
The post How to Design Transactional Agentic AI Systems with LangGraph Using Two-Phase Commit, Human Interrupts, and Safe Rollbacks<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"
In this tutorial, we implement…<\/p>\n","protected":false},"author":1,"featured_media":29,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-214","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/214","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=214"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/214\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/media\/29"}],"wp:attachment":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=214"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=214"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=214"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}