On this tutorial, we construct a glass-box agentic workflow that makes each resolution traceable, auditable, and explicitly ruled by human approval. We design the system to log every thought, motion, and remark right into a tamper-evident audit ledger whereas imposing dynamic permissioning for high-risk operations. By combining LangGraph’s interrupt-driven human-in-the-loop management with a hash-chained database, we exhibit how agentic programs can transfer past opaque automation and align with fashionable governance expectations. All through the tutorial, we deal with sensible, runnable patterns that flip governance from an afterthought right into a first-class system characteristic.
!pip -q set up -U langgraph langchain-core openai "pydantic<=2.12.3"
import os
import json
import time
import hmac
import hashlib
import secrets and techniques
import sqlite3
import getpass
from typing import Any, Dict, Record, Non-obligatory, Literal, TypedDict
from openai import OpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END
from langgraph.sorts import interrupt, Command
if not os.getenv("OPENAI_API_KEY"):
os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter OpenAI API Key: ")
consumer = OpenAI()
MODEL = "gpt-5"We set up all required libraries and import the core modules wanted for agentic workflows and governance. We securely accumulate the OpenAI API key via a terminal immediate to keep away from hard-coding secrets and techniques within the pocket book. We additionally initialize the OpenAI consumer and outline the mannequin 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,
function 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.join(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)
materials = f"{ts}|{actor}|{event_type}|{payload_json}|{prev_hash}".encode("utf-8")
row_hash = _sha256_hex(materials)
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, restrict: int = 50) -> Record[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 ?",
(restrict,),
).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.masses(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 {"okay": 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 {"okay": False, "at_id": id_, "purpose": "prev_hash mismatch"}
materials = f"{ts}|{actor}|{event_type}|{payload_json}|{prev_hash}".encode("utf-8")
expected_hash = _sha256_hex(materials)
if not hmac.compare_digest(expected_hash, row_hash):
return {"okay": False, "at_id": id_, "purpose": "row_hash mismatch"}
expected_prev = row_hash
return {"okay": True, "rows": len(rows), "message": "Hash chain legitimate."}
ledger = AuditLedger()We design a hash-chained SQLite ledger that information each agent and system occasion in an append-only method. We guarantee every log entry cryptographically hyperlinks to the earlier one, making post-hoc tampering detectable. We additionally present utilities to examine current occasions and confirm the integrity of the whole audit chain.
def mint_one_time_token(function: str, ttl_seconds: int = 600) -> Dict[str, str]:
token_id = secrets and techniques.token_hex(12)
token_plain = secrets and techniques.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, function, expires_unix, used) VALUES (?, ?, ?, ?, 0)",
(token_id, token_hash, function, expires),
)
ledger.conn.commit()
return {"token_id": token_id, "token_plain": token_plain, "function": function, "expires_unix": str(expires)}
def consume_one_time_token(token_id: str, token_plain: str, function: str) -> bool:
row = ledger.conn.execute(
"SELECT token_hash, function, 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 != function:
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 {"standing": "success", "transfer_id": "tx_" + secrets and techniques.token_hex(6), "amount_usd": amount_usd, "to_account": to_account}
def tool_rig_move(rig_id: str, path: Literal["UP", "DOWN"], meters: float) -> Dict[str, Any]:
return {"standing": "success", "rig_event_id": "rig_" + secrets and techniques.token_hex(6), "rig_id": rig_id, "path": path, "meters": meters}We implement a safe, single-use token mechanism that allows human approval for high-risk actions. We generate time-limited tokens, retailer solely their hashes, and invalidate them instantly after use. We additionally outline simulated restricted instruments that symbolize delicate operations equivalent to monetary transfers or bodily rig actions.
RestrictedTool = Literal["financial_transfer", "rig_move", "none"]
class GlassBoxState(TypedDict):
messages: Record[Any]
proposed_tool: RestrictedTool
tool_args: Dict[str, Any]
last_observation: Non-obligatory[Dict[str, Any]]
SYSTEM_POLICY = """You're a governance-first agent.
You MUST suggest actions in a structured JSON format with these keys:
- thought
- motion
- args
Return ONLY JSON."""
def llm_propose_action(messages: Record[Any]) -> Dict[str, Any]:
input_msgs = [{"role": "system", "content": SYSTEM_POLICY}]
for m in messages:
if isinstance(m, SystemMessage):
input_msgs.append({"position": "system", "content material": m.content material})
elif isinstance(m, HumanMessage):
input_msgs.append({"position": "person", "content material": m.content material})
elif isinstance(m, AIMessage):
input_msgs.append({"position": "assistant", "content material": m.content material})
resp = consumer.responses.create(mannequin=MODEL, enter=input_msgs)
txt = resp.output_text.strip()
attempt:
return json.masses(txt)
besides Exception:
return {"thought": "fallback", "motion": "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)
motion = proposal.get("motion", "no_op")
args = proposal.get("args", {})
if motion in ["financial_transfer", "rig_move"]:
state["proposed_tool"] = motion
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:
instrument = state["proposed_tool"]
if instrument == "none":
state["last_observation"] = {"standing": "no_op"}
return state
okay = consume_one_time_token(
state["tool_args"]["_token_id"],
state["tool_args"]["_human_token_plain"],
instrument,
)
if not okay:
state["last_observation"] = {"standing": "rejected"}
return state
if instrument == "financial_transfer":
state["last_observation"] = tool_financial_transfer(**state["tool_args"])
if instrument == "rig_move":
state["last_observation"] = tool_rig_move(**state["tool_args"])
return stateWe outline a governance-first system coverage that forces the agent to specific its intent in structured JSON. We use the language mannequin to suggest actions whereas explicitly separating thought, motion, and arguments. We then wire these choices into LangGraph nodes that put together, gate, and validate execution below strict management.
def node_finalize(state: GlassBoxState) -> GlassBoxState:
state["messages"].append(AIMessage(content material=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("assume", 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("assume")
g.add_conditional_edges("assume", 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 = enter("Enter approval token: ")
out = graph.invoke(Command(resume=token))
print(out["messages"][-1].content material)
run_case("Ship $2500 to vendor account ACCT-99213")We assemble the complete LangGraph workflow and join all nodes right into a managed resolution loop. We allow human-in-the-loop interruption, pausing execution till approval is granted or denied. We lastly run an end-to-end instance that demonstrates clear reasoning, enforced governance, and auditable execution in apply.
In conclusion, we applied an agent that now not operates as a black field however as a clear, inspectable resolution engine. We confirmed how real-time audit trails, one-time human approval tokens, and strict execution gates work collectively to forestall silent failures and uncontrolled autonomy. This strategy permits us to retain the ability of agentic workflows whereas embedding accountability instantly into the execution loop. Finally, we demonstrated that robust governance doesn’t sluggish brokers down; as a substitute, it makes them safer, extra reliable, and higher ready for real-world deployment in regulated, high-risk environments.
Try the FULL CODES right here. Additionally, be at liberty to observe us on Twitter and don’t overlook to affix our 100k+ ML SubReddit and Subscribe to our E-newsletter. Wait! are you on telegram? now you’ll be able to be a part of us on telegram as properly.
The publish Find out how to Construct Clear AI Brokers: Traceable Determination-Making with Audit Trails and Human Gates appeared first on MarkTechPost.
Elevate your perspective with NextTech Information, the place innovation meets perception.
Uncover the most recent breakthroughs, get unique updates, and join with a world community of future-focused thinkers.
Unlock tomorrow’s traits right now: learn extra, subscribe to our publication, and turn out to be a part of the NextTech group at NextTech-news.com
