{"id":440,"date":"2026-02-20T14:28:08","date_gmt":"2026-02-20T06:28:08","guid":{"rendered":"https:\/\/connectword.dpdns.org\/?p=440"},"modified":"2026-02-20T14:28:08","modified_gmt":"2026-02-20T06:28:08","slug":"how-to-build-transparent-ai-agents-traceable-decision-making-with-audit-trails-and-human-gates","status":"publish","type":"post","link":"https:\/\/connectword.dpdns.org\/?p=440","title":{"rendered":"How to Build Transparent AI Agents: Traceable Decision-Making with Audit Trails and Human Gates"},"content":{"rendered":"

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\u2019s 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.<\/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-core openai \"pydantic<=2.12.3\"\n\n\nimport os\nimport json\nimport time\nimport hmac\nimport hashlib\nimport secrets\nimport sqlite3\nimport getpass\nfrom typing import Any, Dict, List, Optional, Literal, TypedDict\n\n\nfrom openai import OpenAI\nfrom langchain_core.messages import SystemMessage, HumanMessage, AIMessage\nfrom langgraph.graph import StateGraph, END\nfrom langgraph.types import interrupt, Command\n\n\nif not os.getenv(\"OPENAI_API_KEY\"):\n   os.environ[\"OPENAI_API_KEY\"] = getpass.getpass(\"Enter OpenAI API Key: \")\n\n\nclient = OpenAI()\nMODEL = \"gpt-5\"<\/code><\/pre>\n<\/div>\n<\/div>\n
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\u2019s reasoning loop.<\/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
CREATE_SQL = \"\"\"\nCREATE TABLE IF NOT EXISTS audit_log (\n   id INTEGER PRIMARY KEY AUTOINCREMENT,\n   ts_unix INTEGER NOT NULL,\n   actor TEXT NOT NULL,\n   event_type TEXT NOT NULL,\n   payload_json TEXT NOT NULL,\n   prev_hash TEXT NOT NULL,\n   row_hash TEXT NOT NULL\n);\n\n\nCREATE TABLE IF NOT EXISTS ot_tokens (\n   token_id TEXT PRIMARY KEY,\n   token_hash TEXT NOT NULL,\n   purpose TEXT NOT NULL,\n   expires_unix INTEGER NOT NULL,\n   used INTEGER NOT NULL DEFAULT 0\n);\n\"\"\"\n\n\ndef _sha256_hex(s: bytes) -> str:\n   return hashlib.sha256(s).hexdigest()\n\n\ndef _canonical_json(obj: Any) -> str:\n   return json.dumps(obj, sort_keys=True, separators=(\",\", \":\"), ensure_ascii=False)\n\n\nclass AuditLedger:\n   def __init__(self, path: str = \"glassbox_audit.db\"):\n       self.conn = sqlite3.connect(path, check_same_thread=False)\n       self.conn.executescript(CREATE_SQL)\n       self.conn.commit()\n\n\n   def _last_hash(self) -> str:\n       row = self.conn.execute(\"SELECT row_hash FROM audit_log ORDER BY id DESC LIMIT 1\").fetchone()\n       return row[0] if row else \"GENESIS\"\n\n\n   def append(self, actor: str, event_type: str, payload: Any) -> int:\n       ts = int(time.time())\n       prev_hash = self._last_hash()\n       payload_json = _canonical_json(payload)\n       material = f\"{ts}|{actor}|{event_type}|{payload_json}|{prev_hash}\".encode(\"utf-8\")\n       row_hash = _sha256_hex(material)\n       cur = self.conn.execute(\n           \"INSERT INTO audit_log (ts_unix, actor, event_type, payload_json, prev_hash, row_hash) VALUES (?, ?, ?, ?, ?, ?)\",\n           (ts, actor, event_type, payload_json, prev_hash, row_hash),\n       )\n       self.conn.commit()\n       return cur.lastrowid\n\n\n   def fetch_recent(self, limit: int = 50) -> List[Dict[str, Any]]:\n       rows = self.conn.execute(\n           \"SELECT id, ts_unix, actor, event_type, payload_json, prev_hash, row_hash FROM audit_log ORDER BY id DESC LIMIT ?\",\n           (limit,),\n       ).fetchall()\n       out = []\n       for r in rows[::-1]:\n           out.append({\n               \"id\": r[0],\n               \"ts_unix\": r[1],\n               \"actor\": r[2],\n               \"event_type\": r[3],\n               \"payload\": json.loads(r[4]),\n               \"prev_hash\": r[5],\n               \"row_hash\": r[6],\n           })\n       return out\n\n\n   def verify_integrity(self) -> Dict[str, Any]:\n       rows = self.conn.execute(\n           \"SELECT id, ts_unix, actor, event_type, payload_json, prev_hash, row_hash FROM audit_log ORDER BY id ASC\"\n       ).fetchall()\n       if not rows:\n           return {\"ok\": True, \"rows\": 0, \"message\": \"Empty ledger.\"}\n\n\n       expected_prev = \"GENESIS\"\n       for (id_, ts, actor, event_type, payload_json, prev_hash, row_hash) in rows:\n           if prev_hash != expected_prev:\n               return {\"ok\": False, \"at_id\": id_, \"reason\": \"prev_hash mismatch\"}\n           material = f\"{ts}|{actor}|{event_type}|{payload_json}|{prev_hash}\".encode(\"utf-8\")\n           expected_hash = _sha256_hex(material)\n           if not hmac.compare_digest(expected_hash, row_hash):\n               return {\"ok\": False, \"at_id\": id_, \"reason\": \"row_hash mismatch\"}\n           expected_prev = row_hash\n       return {\"ok\": True, \"rows\": len(rows), \"message\": \"Hash chain valid.\"}\n\n\nledger = AuditLedger()<\/code><\/pre>\n<\/div>\n<\/div>\n
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.<\/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 mint_one_time_token(purpose: str, ttl_seconds: int = 600) -> Dict[str, str]:\n   token_id = secrets.token_hex(12)\n   token_plain = secrets.token_urlsafe(20)\n   token_hash = _sha256_hex(token_plain.encode(\"utf-8\"))\n   expires = int(time.time()) + ttl_seconds\n   ledger.conn.execute(\n       \"INSERT INTO ot_tokens (token_id, token_hash, purpose, expires_unix, used) VALUES (?, ?, ?, ?, 0)\",\n       (token_id, token_hash, purpose, expires),\n   )\n   ledger.conn.commit()\n   return {\"token_id\": token_id, \"token_plain\": token_plain, \"purpose\": purpose, \"expires_unix\": str(expires)}\n\n\ndef consume_one_time_token(token_id: str, token_plain: str, purpose: str) -> bool:\n   row = ledger.conn.execute(\n       \"SELECT token_hash, purpose, expires_unix, used FROM ot_tokens WHERE token_id = ?\",\n       (token_id,),\n   ).fetchone()\n   if not row:\n       return False\n   token_hash_db, purpose_db, expires_unix, used = row\n   if used == 1:\n       return False\n   if purpose_db != purpose:\n       return False\n   if int(time.time()) > int(expires_unix):\n       return False\n   token_hash_in = _sha256_hex(token_plain.encode(\"utf-8\"))\n   if not hmac.compare_digest(token_hash_in, token_hash_db):\n       return False\n   ledger.conn.execute(\"UPDATE ot_tokens SET used = 1 WHERE token_id = ?\", (token_id,))\n   ledger.conn.commit()\n   return True\n\n\ndef tool_financial_transfer(amount_usd: float, to_account: str) -> Dict[str, Any]:\n   return {\"status\": \"success\", \"transfer_id\": \"tx_\" + secrets.token_hex(6), \"amount_usd\": amount_usd, \"to_account\": to_account}\n\n\ndef tool_rig_move(rig_id: str, direction: Literal[\"UP\", \"DOWN\"], meters: float) -> Dict[str, Any]:\n   return {\"status\": \"success\", \"rig_event_id\": \"rig_\" + secrets.token_hex(6), \"rig_id\": rig_id, \"direction\": direction, \"meters\": meters}<\/code><\/pre>\n<\/div>\n<\/div>\n
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.<\/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
RestrictedTool = Literal[\"financial_transfer\", \"rig_move\", \"none\"]\n\n\nclass GlassBoxState(TypedDict):\n   messages: List[Any]\n   proposed_tool: RestrictedTool\n   tool_args: Dict[str, Any]\n   last_observation: Optional[Dict[str, Any]]\n\n\nSYSTEM_POLICY = \"\"\"You are a governance-first agent.\nYou MUST propose actions in a structured JSON format with these keys:\n- thought\n- action\n- args\nReturn ONLY JSON.\"\"\"\n\n\ndef llm_propose_action(messages: List[Any]) -> Dict[str, Any]:\n   input_msgs = [{\"role\": \"system\", \"content\": SYSTEM_POLICY}]\n   for m in messages:\n       if isinstance(m, SystemMessage):\n           input_msgs.append({\"role\": \"system\", \"content\": m.content})\n       elif isinstance(m, HumanMessage):\n           input_msgs.append({\"role\": \"user\", \"content\": m.content})\n       elif isinstance(m, AIMessage):\n           input_msgs.append({\"role\": \"assistant\", \"content\": m.content})\n\n\n   resp = client.responses.create(model=MODEL, input=input_msgs)\n   txt = resp.output_text.strip()\n   try:\n       return json.loads(txt)\n   except Exception:\n       return {\"thought\": \"fallback\", \"action\": \"ask_human\", \"args\": {}}\n\n\ndef node_think(state: GlassBoxState) -> GlassBoxState:\n   proposal = llm_propose_action(state[\"messages\"])\n   ledger.append(\"agent\", \"THOUGHT\", {\"thought\": proposal.get(\"thought\")})\n   ledger.append(\"agent\", \"ACTION\", proposal)\n\n\n   action = proposal.get(\"action\", \"no_op\")\n   args = proposal.get(\"args\", {})\n\n\n   if action in [\"financial_transfer\", \"rig_move\"]:\n       state[\"proposed_tool\"] = action\n       state[\"tool_args\"] = args\n   else:\n       state[\"proposed_tool\"] = \"none\"\n       state[\"tool_args\"] = {}\n\n\n   return state\n\n\ndef node_permission_gate(state: GlassBoxState) -> GlassBoxState:\n   if state[\"proposed_tool\"] == \"none\":\n       return state\n\n\n   token = mint_one_time_token(state[\"proposed_tool\"])\n   payload = {\"token_id\": token[\"token_id\"], \"token_plain\": token[\"token_plain\"]}\n   human_input = interrupt(payload)\n\n\n   state[\"tool_args\"][\"_token_id\"] = token[\"token_id\"]\n   state[\"tool_args\"][\"_human_token_plain\"] = str(human_input)\n   return state\n\n\ndef node_execute_tool(state: GlassBoxState) -> GlassBoxState:\n   tool = state[\"proposed_tool\"]\n   if tool == \"none\":\n       state[\"last_observation\"] = {\"status\": \"no_op\"}\n       return state\n\n\n   ok = consume_one_time_token(\n       state[\"tool_args\"][\"_token_id\"],\n       state[\"tool_args\"][\"_human_token_plain\"],\n       tool,\n   )\n\n\n   if not ok:\n       state[\"last_observation\"] = {\"status\": \"rejected\"}\n       return state\n\n\n   if tool == \"financial_transfer\":\n       state[\"last_observation\"] = tool_financial_transfer(**state[\"tool_args\"])\n   if tool == \"rig_move\":\n       state[\"last_observation\"] = tool_rig_move(**state[\"tool_args\"])\n\n\n   return state<\/code><\/pre>\n<\/div>\n<\/div>\n
We 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.<\/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 node_finalize(state: GlassBoxState) -> GlassBoxState:\n   state[\"messages\"].append(AIMessage(content=json.dumps(state[\"last_observation\"])))\n   return state\n\n\ndef route_after_think(state: GlassBoxState) -> str:\n   return \"permission_gate\" if state[\"proposed_tool\"] != \"none\" else \"execute_tool\"\n\n\ng = StateGraph(GlassBoxState)\ng.add_node(\"think\", node_think)\ng.add_node(\"permission_gate\", node_permission_gate)\ng.add_node(\"execute_tool\", node_execute_tool)\ng.add_node(\"finalize\", node_finalize)\n\n\ng.set_entry_point(\"think\")\ng.add_conditional_edges(\"think\", route_after_think)\ng.add_edge(\"permission_gate\", \"execute_tool\")\ng.add_edge(\"execute_tool\", \"finalize\")\ng.add_edge(\"finalize\", END)\n\n\ngraph = g.compile()\n\n\ndef run_case(user_request: str):\n   state = {\n       \"messages\": [HumanMessage(content=user_request)],\n       \"proposed_tool\": \"none\",\n       \"tool_args\": {},\n       \"last_observation\": None,\n   }\n   out = graph.invoke(state)\n   if \"__interrupt__\" in out:\n       token = input(\"Enter approval token: \")\n       out = graph.invoke(Command(resume=token))\n   print(out[\"messages\"][-1].content)\n\n\nrun_case(\"Send $2500 to vendor account ACCT-99213\")<\/code><\/pre>\n<\/div>\n<\/div>\n
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.<\/p>\n
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.<\/p>\n
\n
Check out the\u00a0FULL CODES here<\/a>.\u00a0<\/strong>Also,\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 Build Transparent AI Agents: Traceable Decision-Making with Audit Trails and Human Gates<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"
In this tutorial, we build a g…<\/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-440","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\/440","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=440"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/440\/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=440"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=440"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=440"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}