Source: MarkTechPost
In this tutorial, we build a glass-box agentic workflow that makes every decision traceable, auditable, and explicitly governed by human approval. We design the system to log each thought, action, and observation into a tamper-evident audit ledger while enforcing dynamic permissioning for high-risk operations. By combining LangGraph’s interrupt-driven human-in-the-loop control with a hash-chained database, we demonstrate how agentic systems can move beyond opaque automation and align with modern governance expectations. Throughout the tutorial, we focus on practical, runnable patterns that turn governance from an afterthought into a first-class system feature.
!pip -q install -U langgraph langchain-core openai "pydantic<=2.12.3" import os import json import time import hmac import hashlib import secrets import sqlite3 import getpass from typing import Any, Dict, List, Optional, Literal, TypedDict from openai import OpenAI from langchain_core.messages import SystemMessage, HumanMessage, AIMessage from langgraph.graph import StateGraph, END from langgraph.types import interrupt, Command if not os.getenv("OPENAI_API_KEY"): os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter OpenAI API Key: ") client = OpenAI() MODEL = "gpt-5"We install all required libraries and import the core modules needed for agentic workflows and governance. We securely collect the OpenAI API key through a terminal prompt to avoid hard-coding secrets in the notebook. We also initialize the OpenAI client and define the model that drives the agent’s reasoning loop.
CREATE_SQL = """ CREATE TABLE IF NOT EXISTS audit_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts_unix INTEGER NOT NULL, actor TEXT NOT NULL, event_type TEXT NOT NULL, payload_json TEXT NOT NULL, prev_hash TEXT NOT NULL, row_hash TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS ot_tokens ( token_id TEXT PRIMARY KEY, token_hash TEXT NOT NULL, purpose TEXT NOT NULL, expires_unix INTEGER NOT NULL, used INTEGER NOT NULL DEFAULT 0 ); """ def _sha256_hex(s: bytes) -> str: return hashlib.sha256(s).hexdigest() def _canonical_json(obj: Any) -> str: return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) class AuditLedger: def __init__(self, path: str = "glassbox_audit.db"): self.conn = sqlite3.connect(path, check_same_thread=False) self.conn.executescript(CREATE_SQL) self.conn.commit() def _last_hash(self) -> str: row = self.conn.execute("SELECT row_hash FROM audit_log ORDER BY id DESC LIMIT 1").fetchone() return row[0] if row else "GENESIS" def append(self, actor: str, event_type: str, payload: Any) -> int: ts = int(time.time()) prev_hash = self._last_hash() payload_json = _canonical_json(payload) material = f"{ts}|{actor}|{event_type}|{payload_json}|{prev_hash}".encode("utf-8") row_hash = _sha256_hex(material) cur = self.conn.execute( "INSERT INTO audit_log (ts_unix, actor, event_type, payload_json, prev_hash, row_hash) VALUES (?, ?, ?, ?, ?, ?)", (ts, actor, event_type, payload_json, prev_hash, row_hash), ) self.conn.commit() return cur.lastrowid def fetch_recent(self, limit: int = 50) -> List[Dict[str, Any]]: rows = self.conn.execute( "SELECT id, ts_unix, actor, event_type, payload_json, prev_hash, row_hash FROM audit_log ORDER BY id DESC LIMIT ?", (limit,), ).fetchall() out = [] for r in rows[::-1]: out.append({ "id": r[0], "ts_unix": r[1], "actor": r[2], "event_type": r[3], "payload": json.loads(r[4]), "prev_hash": r[5], "row_hash": r[6], }) return out def verify_integrity(self) -> Dict[str, Any]: rows = self.conn.execute( "SELECT id, ts_unix, actor, event_type, payload_json, prev_hash, row_hash FROM audit_log ORDER BY id ASC" ).fetchall() if not rows: return {"ok": True, "rows": 0, "message": "Empty ledger."} expected_prev = "GENESIS" for (id_, ts, actor, event_type, payload_json, prev_hash, row_hash) in rows: if prev_hash != expected_prev: return {"ok": False, "at_id": id_, "reason": "prev_hash mismatch"} material = f"{ts}|{actor}|{event_type}|{payload_json}|{prev_hash}".encode("utf-8") expected_hash = _sha256_hex(material) if not hmac.compare_digest(expected_hash, row_hash): return {"ok": False, "at_id": id_, "reason": "row_hash mismatch"} expected_prev = row_hash return {"ok": True, "rows": len(rows), "message": "Hash chain valid."} ledger = AuditLedger()We design a hash-chained SQLite ledger that records every agent and system event in an append-only manner. We ensure each log entry cryptographically links to the previous one, making post-hoc tampering detectable. We also provide utilities to inspect recent events and verify the integrity of the entire audit chain.
def mint_one_time_token(purpose: str, ttl_seconds: int = 600) -> Dict[str, str]: token_id = secrets.token_hex(12) token_plain = secrets.token_urlsafe(20) token_hash = _sha256_hex(token_plain.encode("utf-8")) expires = int(time.time()) + ttl_seconds ledger.conn.execute( "INSERT INTO ot_tokens (token_id, token_hash, purpose, expires_unix, used) VALUES (?, ?, ?, ?, 0)", (token_id, token_hash, purpose, expires), ) ledger.conn.commit() return {"token_id": token_id, "token_plain": token_plain, "purpose": purpose, "expires_unix": str(expires)} def consume_one_time_token(token_id: str, token_plain: str, purpose: str) -> bool: row = ledger.conn.execute( "SELECT token_hash, purpose, expires_unix, used FROM ot_tokens WHERE token_id = ?", (token_id,), ).fetchone() if not row: return False token_hash_db, purpose_db, expires_unix, used = row if used == 1: return False if purpose_db != purpose: return False if int(time.time()) > int(expires_unix): return False token_hash_in = _sha256_hex(token_plain.encode("utf-8")) if not hmac.compare_digest(token_hash_in, token_hash_db): return False ledger.conn.execute("UPDATE ot_tokens SET used = 1 WHERE token_id = ?", (token_id,)) ledger.conn.commit() return True def tool_financial_transfer(amount_usd: float, to_account: str) -> Dict[str, Any]: return {"status": "success", "transfer_id": "tx_" + secrets.token_hex(6), "amount_usd": amount_usd, "to_account": to_account} def tool_rig_move(rig_id: str, direction: Literal["UP", "DOWN"], meters: float) -> Dict[str, Any]: return {"status": "success", "rig_event_id": "rig_" + secrets.token_hex(6), "rig_id": rig_id, "direction": direction, "meters": meters}We implement a secure, single-use token mechanism that enables human approval for high-risk actions. We generate time-limited tokens, store only their hashes, and invalidate them immediately after use. We also define simulated restricted tools that represent sensitive operations such as financial transfers or physical rig movements.
RestrictedTool = Literal["financial_transfer", "rig_move", "none"] class GlassBoxState(TypedDict): messages: List[Any] proposed_tool: RestrictedTool tool_args: Dict[str, Any] last_observation: Optional[Dict[str, Any]] SYSTEM_POLICY = """You are a governance-first agent. You MUST propose actions in a structured JSON format with these keys: - thought - action - args Return ONLY JSON.""" def llm_propose_action(messages: List[Any]) -> Dict[str, Any]: input_msgs = [{"role": "system", "content": SYSTEM_POLICY}] for m in messages: if isinstance(m, SystemMessage): input_msgs.append({"role": "system", "content": m.content}) elif isinstance(m, HumanMessage): input_msgs.append({"role": "user", "content": m.content}) elif isinstance(m, AIMessage): input_msgs.append({"role": "assistant", "content": m.content}) resp = client.responses.create(model=MODEL, input=input_msgs) txt = resp.output_text.strip() try: return json.loads(txt) except Exception: return {"thought": "fallback", "action": "ask_human", "args": {}} def node_think(state: GlassBoxState) -> GlassBoxState: proposal = llm_propose_action(state["messages"]) ledger.append("agent", "THOUGHT", {"thought": proposal.get("thought")}) ledger.append("agent", "ACTION", proposal) action = proposal.get("action", "no_op") args = proposal.get("args", {}) if action in ["financial_transfer", "rig_move"]: state["proposed_tool"] = action state["tool_args"] = args else: state["proposed_tool"] = "none" state["tool_args"] = {} return state def node_permission_gate(state: GlassBoxState) -> GlassBoxState: if state["proposed_tool"] == "none": return state token = mint_one_time_token(state["proposed_tool"]) payload = {"token_id": token["token_id"], "token_plain": token["token_plain"]} human_input = interrupt(payload) state["tool_args"]["_token_id"] = token["token_id"] state["tool_args"]["_human_token_plain"] = str(human_input) return state def node_execute_tool(state: GlassBoxState) -> GlassBoxState: tool = state["proposed_tool"] if tool == "none": state["last_observation"] = {"status": "no_op"} return state ok = consume_one_time_token( state["tool_args"]["_token_id"], state["tool_args"]["_human_token_plain"], tool, ) if not ok: state["last_observation"] = {"status": "rejected"} return state if tool == "financial_transfer": state["last_observation"] = tool_financial_transfer(**state["tool_args"]) if tool == "rig_move": state["last_observation"] = tool_rig_move(**state["tool_args"]) return stateWe define a governance-first system policy that forces the agent to express its intent in structured JSON. We use the language model to propose actions while explicitly separating thought, action, and arguments. We then wire these decisions into LangGraph nodes that prepare, gate, and validate execution under strict control.
def node_finalize(state: GlassBoxState) -> GlassBoxState: state["messages"].append(AIMessage(content=json.dumps(state["last_observation"]))) return state def route_after_think(state: GlassBoxState) -> str: return "permission_gate" if state["proposed_tool"] != "none" else "execute_tool" g = StateGraph(GlassBoxState) g.add_node("think", node_think) g.add_node("permission_gate", node_permission_gate) g.add_node("execute_tool", node_execute_tool) g.add_node("finalize", node_finalize) g.set_entry_point("think") g.add_conditional_edges("think", route_after_think) g.add_edge("permission_gate", "execute_tool") g.add_edge("execute_tool", "finalize") g.add_edge("finalize", END) graph = g.compile() def run_case(user_request: str): state = { "messages": [HumanMessage(content=user_request)], "proposed_tool": "none", "tool_args": {}, "last_observation": None, } out = graph.invoke(state) if "__interrupt__" in out: token = input("Enter approval token: ") out = graph.invoke(Command(resume=token)) print(out["messages"][-1].content) run_case("Send $2500 to vendor account ACCT-99213")We assemble the full LangGraph workflow and connect all nodes into a controlled decision loop. We enable human-in-the-loop interruption, pausing execution until approval is granted or denied. We finally run an end-to-end example that demonstrates transparent reasoning, enforced governance, and auditable execution in practice.
In conclusion, we implemented an agent that no longer operates as a black box but as a transparent, inspectable decision engine. We showed how real-time audit trails, one-time human approval tokens, and strict execution gates work together to prevent silent failures and uncontrolled autonomy. This approach allows us to retain the power of agentic workflows while embedding accountability directly into the execution loop. Ultimately, we demonstrated that strong governance does not slow agents down; instead, it makes them safer, more trustworthy, and better prepared for real-world deployment in regulated, high-risk environments.
Check out the FULL CODES here. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
Michal Sutter
Michal Sutter is a data science professional with a Master of Science in Data Science from the University of Padova. With a solid foundation in statistical analysis, machine learning, and data engineering, Michal excels at transforming complex datasets into actionable insights.
