How to Build Transparent AI Agents: Trackable Decision Making with Research Trails and Human Gateways
In this tutorial, we build a glass-box workflow that makes every decision traceable, auditable, and transparently governed by human approval. We design a system to enter each thought, action, and observation into a tamper-evident checklist while enforcing flexible consent for high-risk activities. By combining the human-in-the-loop interrupt-driven LangGraph controller with a hash-chain database, we show how agent systems can move beyond opaque automation and align with modern governance expectations. Throughout the course, we focus on practical, actionable patterns that transform governance from an afterthought into a first-rate 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 include all the required libraries and import the essential modules required for further agent operation and governance. We securely collect the OpenAI API key via a terminal command to avoid coding secrets in the manual. We also implement the OpenAI client and define the model that drives the agent's logic 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 SQLite ledger with a hash-chain that records every agent and system event in an append-only manner. We ensure that each log entry is correlated with the previous one, making post-hoc tampering visible. We also provide resources to check the latest events and ensure the integrity of the entire research 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 use a secure, single-use token approach that allows human authorization for high-risk actions. We create time-limited tokens, store only their hashes, and invalidate them immediately after use. We also describe limited simulation tools that represent sensitive operations such as financial transfers or firm 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 first define a system policy that forces the agent to express its intent in structured JSON. We use a language model to suggest actions while clearly distinguishing thought, action, and arguments. We then connect these decisions to LangGraph nodes that configure, gate, and verify performance under tight 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 integrate a full LangGraph workflow and connect all nodes in a controlled decision loop. We allow for human-in-the-loop interruption, pausing the execution until it is approved or rejected. Finally we run a bottom-up example that demonstrates transparent reasoning, enforced governance, and testable performance in practice.
In conclusion, we used the agent no longer as a black box but as a transparent, testable decision engine. We've shown that real-time test methods, one-time authentication tokens, and strong kill gates work together to prevent silent failures and unmanageable autonomy. This approach allows us to retain the functionality of the agent workflow while embedding the accountability directly into the work loop. Finally, we showed that strong governance does not deter agents; instead, it makes them safer, more reliable, and better suited for real-world deployments in controlled, high-risk environments.
Check it out FULL CODES here. Also, feel free to follow us Twitter and don't forget to join our 100k+ ML SubReddit and Subscribe to Our newspaper. Wait! are you on telegram? now you can join us on telegram too.
The post How to Build Transparent AI Agents: Trackable Decision Making with Research Trails and Human Gateways appeared first on MarkTechPost.
