En este tutorial, creamos un flujo de trabajo agente de caja de cristal que hace que cada decisión sea rastreable, auditable y explícitamente gobernada por la aprobación humana. Diseñamos el sistema para registrar cada pensamiento, acción y observación en un libro de auditoría a prueba de manipulaciones mientras aplicamos permisos dinámicos para operaciones de alto riesgo. Al combinar el control humano en el circuito impulsado por interrupciones de LangGraph con una base de datos en cadena hash, demostramos cómo los sistemas agentes pueden ir más allá de la automatización opaca y alinearse con las expectativas de gobernanza modernas. A lo largo del tutorial, nos centramos en patrones prácticos y ejecutables que convierten la gobernanza de una idea de último momento en una característica del sistema de primera clase.
!pip -q install -U langgraph langchain-core openai “pydantic<=2.12.3" importar os importar json importar tiempo importar hmac importar hashlib importar secretos importar sqlite3 importar getpass de escribir importar Cualquiera, Dict, Lista, Opcional, Literal, TypedDict de openai importar OpenAI de langchain_core.messages importar SystemMessage, HumanMessage, AIMessage de langgraph.graph importa StateGraph, FIN de langgraph.types importa interrupción, Comando si no os.getenv("OPENAI_API_KEY"): os.environ["OPENAI_API_KEY"] = getpass.getpass("Ingrese la clave API de OpenAI: ") cliente = OpenAI() MODELO = "gpt-5"
Instalamos todas las bibliotecas necesarias e importamos los módulos principales necesarios para los flujos de trabajo y la gobernanza agentes. Recopilamos de forma segura la clave API de OpenAI a través de un mensaje de terminal para evitar secretos de codificación en el portátil. También inicializamos el cliente OpenAI y definimos el modelo que impulsa el ciclo de razonamiento del agente.
CREATE_SQL = “”” CREAR TABLA SI NO EXISTE audit_log (id INTEGER PRIMARY KEY AUTOINCREMENT, ts_unix INTEGER NOT NULL, actor TEXTO NOT NULL, event_type TEXTO NOT NULL, payload_json TEXTO NOT NULL, prev_hash TEXTO NOT NULL, row_hash TEXTO NOT NULL); CREAR TABLA SI NO EXISTE ot_tokens ( token_id TEXTO CLAVE PRIMARIA, token_hash TEXTO NO NULO, propósito TEXTO NO NULO, expires_unix INTEGER NO NULO, usado INTEGER NOT NULL DEFAULT 0 ); json.dumps(obj, sort_keys=True, separators=(“,”, “:”), sure_ascii=False) clase AuditLedger: def __init__(self, ruta: str = “glassbox_audit.db”): self.conn = sqlite3.connect(ruta, check_same_thread=False) self.conn.executescript(CREATE_SQL) self.conn.commit() def _last_hash(self) -> str: fila = self.conn.execute(“SELECT row_hash FROM audit_log ORDER BY id DESC LIMIT 1”).fetchone() devolver fila[0] if fila 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( “INSERTAR EN audit_log (ts_unix, actor, event_type, payload_json, prev_hash, row_hash) VALORES (?, ?, ?, ?, ?, ?)”, (ts, actor, event_type, payload_json, prev_hash, row_hash), ) self.conn.commit() return cur.lastrowid def fetch_recent(self, limit: int = 50) -> Lista[Dict[str, Any]]: filas = self.conn.execute( “SELECT id, ts_unix, actor, event_type, payload_json, prev_hash, row_hash FROM audit_log ORDER BY id DESC LIMIT?”, (límite,), ).fetchall() out = []
para r en filas[::-1]: out.append({ “id”: r[0]”ts_unix”: r[1]”actor”: r[2]”tipo_evento”: r[3]”carga útil”: json.loads(r[4]), “prev_hash”: r[5]”row_hash”: r[6]}) devuelve def verificar_integridad(self) -> Dict[str, Any]: filas = self.conn.execute( “SELECT id, ts_unix, actor, event_type, payload_json, prev_hash, row_hash FROM audit_log ORDER BY id ASC” ).fetchall() si no filas: return {“ok”: True, “rows”: 0, “message”: “Empty ledger.”} esperado_prev = “GENESIS” para (id_, ts, actor, event_type, payload_json, prev_hash, row_hash) en filas: 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) si no 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”: “Cadena hash válida.”} libro mayor = AuditLedger()
Diseñamos un libro de contabilidad SQLite con cadena hash que registra cada agente y evento del sistema de forma que solo se pueda agregar. Nos aseguramos de que cada entrada del registro se vincule criptográficamente con la anterior, lo que hace que la manipulación post hoc sea detectable. También proporcionamos servicios públicos para inspeccionar eventos recientes y verificar la integridad de toda la cadena de auditoría.
def mint_one_time_token(propósito: str, ttl_segundos: 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”)) expira = int(time.time()) + ttl_segundos ledger.conn.execute( “INSERT INTO ot_tokens (token_id, token_hash, propósito, expira_unix, usado) VALORES (?, ?, ?, ?, 0)”, (token_id, token_hash, propósito, expira), ) ledger.conn.commit() return {“token_id”: token_id, “token_plain”: token_plain, “propósito”: propósito, “expires_unix”: str(expires)} def consume_one_time_token(token_id: str, token_plain: str, propósito: str) -> bool: fila = ledger.conn.execute( “SELECCIONE token_hash, propósito, expires_unix, usado DESDE ot_tokens DONDE token_id =?”, (token_id,), ).fetchone() si no es fila: devuelve Falso token_hash_db, propósito_db, expira_unix, usado = fila si se usa == 1: devuelve Falso si propósito_db != propósito: devolver False si int(time.time()) > int(expires_unix): devolver False token_hash_in = _sha256_hex(token_plain.encode(“utf-8”)) si no hmac.compare_digest(token_hash_in, token_hash_db): devolver False ledger.conn.execute(“ACTUALIZAR ot_tokens SET usado = 1 DONDE token_id =?”, (token_id,)) ledger.conn.commit() return True def tool_financial_transfer(amount_usd: float, to_account: str) -> Dict[str, Any]: return {“status”: “éxito”, “transfer_id”: “tx_” + secrets.token_hex(6), “amount_usd”: monto_usd, “to_account”: to_account} def tool_rig_move(rig_id: str, dirección: Literal[“UP”, “DOWN”]metros: flotador) -> Dict[str, Any]: return {“status”: “success”, “rig_event_id”: “rig_” + secrets.token_hex(6), “rig_id”: rig_id, “direction”: dirección, “meters”: metros}
Implementamos un mecanismo de token seguro y de un solo uso que permite la aprobación humana de acciones de alto riesgo. Generamos tokens por tiempo limitado, almacenamos solo sus hashes y los invalidamos inmediatamente después de su uso. También definimos herramientas restringidas simuladas que representan operaciones sensibles, como transferencias financieras o movimientos físicos de plataformas.
Herramienta restringida = Literal[“financial_transfer”, “rig_move”, “none”]
clase GlassBoxState (TypedDict): mensajes: Lista[Any]
propuesta_herramienta: herramienta restringida tool_args: Dict[str, Any]
última_observación: opcional[Dict[str, Any]]SYSTEM_POLICY = “””Usted es un agente que prioriza la gobernanza. DEBE proponer acciones en un formato JSON estructurado con estas claves: – pensamiento – acción – args Devolver SÓLO JSON.””” def llm_propose_action(messages: List[Any]) -> Dictar[str, Any]: mensajes_entrada = [{“role”: “system”, “content”: SYSTEM_POLICY}]
para m en mensajes: 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) excepto Excepción: return {“thinked”: “fallback”, “action”: “ask_human”, “args”: {}} def node_think(state: GlassBoxState) -> GlassBoxState: propuesta = llm_propose_action(estado[“messages”]) ledger.append(“agente”, “PENSAMIENTO”, {“pensamiento”: propuesta.get(“pensamiento”)}) libro mayor.append(“agente”, “ACCIÓN”, propuesta) acción = propuesta.get(“acción”, “no_op”) args = propuesta.get(“args”, {}) si la acción está en [“financial_transfer”, “rig_move”]: estado[“proposed_tool”] = estado de acción[“tool_args”] = argumentos más: estado[“proposed_tool”] = estado “ninguno”[“tool_args”] = {} devolver estado def node_permission_gate(estado: GlassBoxState) -> GlassBoxState: si estado[“proposed_tool”] == “ninguno”: token de estado de retorno = mint_one_time_token(estado[“proposed_tool”]) carga útil = {“token_id”: token[“token_id”]”token_plain”: token[“token_plain”]} human_input = estado de interrupción (carga útil)[“tool_args”][“_token_id”] = ficha[“token_id”]
estado[“tool_args”][“_human_token_plain”] = str(human_input) devolver estado def node_execute_tool(estado: GlassBoxState) -> GlassBoxState: herramienta = estado[“proposed_tool”]
si herramienta == “ninguna”: estado[“last_observation”] = {“status”: “no_op”} devolver estado ok = consumir_one_time_token( estado[“tool_args”][“_token_id”]estado[“tool_args”][“_human_token_plain”]herramienta, ) si no está bien: estado[“last_observation”] = {“status”: “rejected”} devolver estado si herramienta == “financial_transfer”: estado[“last_observation”] = herramienta_transferencia_financiera(**estado[“tool_args”]) si herramienta == “rig_move”: estado[“last_observation”] = herramienta_rig_move(**estado[“tool_args”]) estado de retorno
Definimos una política de sistema de gobernanza primero que obliga al agente a expresar su intención en JSON estructurado. Usamos el modelo de lenguaje para proponer acciones mientras separamos explícitamente pensamiento, acción y argumentos. Luego conectamos estas decisiones a nodos LangGraph que preparan, controlan y validan la ejecución bajo un control estricto.
def node_finalize(estado: GlassBoxState) -> GlassBoxState: estado[“messages”].append(AIMessage(content=json.dumps(estado[“last_observation”]))) devolver estado def route_after_think(estado: GlassBoxState) -> str: devolver “permission_gate” si estado[“proposed_tool”] != “ninguno” 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”: Ninguno, } out = graph.invoke(state) if “__interrupt__” in out: token = input(“Ingrese el token de aprobación: “) out = graph.invoke(Command(resume=token)) print(out[“messages”][-1].content) run_case(“Enviar $2500 a la cuenta del proveedor ACCT-99213”)
Montamos el flujo de trabajo completo de LangGraph y conectamos todos los nodos en un ciclo de decisión controlado. Permitimos la interrupción humana en el circuito, pausando la ejecución hasta que se conceda o deniegue la aprobación. Finalmente ejecutamos un ejemplo de extremo a extremo que demuestra un razonamiento transparente, una gobernanza impuesta y una ejecución auditable en la práctica.
En conclusión, implementamos un agente que ya no opera como una caja negra sino como un motor de decisiones transparente e inspeccionable. Mostramos cómo los seguimientos de auditoría en tiempo real, los tokens de aprobación humana únicos y las puertas de ejecución estrictas funcionan en conjunto para evitar fallas silenciosas y una autonomía incontrolada. Este enfoque nos permite conservar el poder de los flujos de trabajo agentes y al mismo tiempo incorporar la responsabilidad directamente en el ciclo de ejecución. En última instancia, demostramos que una gobernanza sólida no frena a los agentes; en cambio, los hace más seguros, más confiables y mejor preparados para la implementación en el mundo real en entornos regulados y de alto riesgo.
Consulta los CÓDIGOS COMPLETOS aquí. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 100.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.
La publicación Cómo crear agentes de IA transparentes: toma de decisiones rastreable con seguimientos de auditoría y puertas humanas apareció por primera vez en MarkTechPost.