Cómo diseñar sistemas de IA agentes transaccionales con LangGraph utilizando compromisos de dos fases, interrupciones humanas y reversiones seguras

En este tutorial, implementamos un patrón de IA agente usando LangGraph que trata el razonamiento y la acción como un flujo de trabajo transaccional en lugar de una decisión única. Modelamos un sistema de confirmación de dos fases en el que un agente realiza cambios reversibles, valida invariantes estrictos, hace una pausa para la aprobación humana mediante interrupciones gráficas y solo entonces confirma o revierte. Con esto, demostramos cómo se pueden diseñar sistemas agentic teniendo en cuenta la seguridad, la auditabilidad y la controlabilidad, yendo más allá de los agentes de chat reactivos hacia flujos de trabajo de IA estructurados y conscientes de la gobernanza que se ejecutan de manera confiable en Google Colab utilizando modelos OpenAI. Consulte los códigos completos aquí.

!pip -q install -U langgraph langchain-openai importar os, json, uuid, copiar, matemáticas, re, operador de escribir importar Cualquiera, Dict, Lista, Opcional de escribir_extensiones importar TypedDict, Anotado de langchain_openai importar ChatOpenAI de langchain_core.messages importar SystemMessage, HumanMessage, AIMessage, AnyMessage de langgraph.graph importar StateGraph, START, END desde langgraph.graph.message importar add_messages desde langgraph.checkpoint.memory importar InMemorySaver desde langgraph.types importar interrupción, Comando def _set_env_openai(): if os.environ.get(“OPENAI_API_KEY”): volver intentar: desde google.colab importar datos de usuario k = userdata.get(“OPENAI_API_KEY”) si k: os.environ[“OPENAI_API_KEY”] = k retorno excepto Excepción: pasar importar getpass os.environ[“OPENAI_API_KEY”] = getpass.getpass(“Ingrese OPENAI_API_KEY: “) _set_env_openai() MODELO = os.environ.get(“OPENAI_MODEL”, “gpt-4o-mini”) llm = ChatOpenAI(modelo=MODELO, temperatura=0)

Configuramos el entorno de ejecución instalando LangGraph e inicializando el modelo OpenAI. Cargamos de forma segura la clave API y configuramos un LLM determinista, asegurando que todo el comportamiento de los agentes posteriores siga siendo reproducible y controlado. Consulte los códigos completos aquí.

MUESTRA_LEDGER = [
{“txn_id”: “T001”, “name”: “Asha”, “email”: “[email protected]”, “monto”: “1,250.50”, “fecha”: “01/12/2025”, “nota”: “Renovación de membresía”}, {“txn_id”: “T002”, “nombre”: “Ravi”, “email”: “[email protected]”, “monto”: “-500”, “fecha”: “2025-12-02”, “nota”: “¿Devolución de cargo?”}, {“txn_id”: “T003”, “nombre”: “Sara”, “correo electrónico”: “[email protected]”, “monto”: “700”, “fecha”: “12-02-2025”, “nota”: “Exención del cargo por pago atrasado”}, {“txn_id”: “T003”, “nombre”: “Sara”, “correo electrónico”: “[email protected]”, “monto”: “700”, “fecha”: “12-02-2025”, “nota”: “Fila duplicada”}, {“txn_id”: “T004”, “nombre”: “Lee”, “email”: “[email protected]”, “cantidad”: “NaN”, “fecha”: “03/12/2025”, “nota”: “Cantidad incorrecta”}, ]ALLOWED_OPS = {“reemplazar”, “eliminar”, “agregar”} def _parse_amount(x): if isinstance(x, (int, float)): devolver float(x) if isinstance(x, str): intentar: return float(x.replace(“,”, “”)) excepto: return Ninguno return Ninguno def _iso_date(d): si no es isinstance(d, str): return Ninguno d = d.replace(“/”, “-“) p = d.split(“-“) if len(p) == 3 y len(p[0]) == 4: devuelve d si len(p) == 3 y len(p[2]) == 4: devolver f”{p[2]}-{pag[1]}-{pag[0]}” return Ninguno def perfil_ledger(filas): visto, anomalías = {}, []
para i, r en enumerar (filas): si _parse_amount(r.get(“cantidad”)) es Ninguno: anomalías.append(i) si r.get(“txn_id”) visto: anomalías.append(i) visto[r.get(“txn_id”)] = devuelvo {“filas”: len(filas), “anomalías”: anomalías} def apply_patch(filas, parche): out = copy.deepcopy(filas) para op in sorted([p for p in patch if p[“op”] == “eliminar”]clave=lambda x: x[“idx”]reverso=Verdadero): out.pop(op[“idx”]) para op en parche: si op[“op”] en {“agregar”, “reemplazar”}: fuera[op[“idx”]][op[“field”]]= operación[“value”]
devolver def validar (filas): problemas = []
para i, r en enumerar (filas): si _parse_amount(r.get(“cantidad”)) es Ninguno: issues.append(i) si _iso_date(r.get(“date”)) es Ninguno: issues.append(i) return {“ok”: len(issues) == 0, “issues”: issues}

Definimos la abstracción del libro mayor central junto con la lógica de parcheo, normalización y validación. Tratamos las transformaciones de datos como operaciones reversibles, lo que permite al agente razonar sobre los cambios de forma segura antes de realizarlos. Consulte los códigos completos aquí.

clase TxnState (TypedDict): mensajes: anotados[List[AnyMessage]add_messages]raw_rows: Lista[Dict[str, Any]]sandbox_rows: Lista[Dict[str, Any]]parche: Lista[Dict[str, Any]]validación: Dict[str, Any]
aprobado: Opcional[bool]

def perfil_nodo(estado): p = libro mayor_perfil(estado)[“raw_rows”]) devolver {“mensajes”: [AIMessage(content=json.dumps(p))]} def node_patch(estado): sys = SystemMessage(content=”Devolver una lista de parches JSON arreglando cantidades, fechas, correos electrónicos, duplicados”) usr = HumanMessage(content=json.dumps(estado)[“raw_rows”])) r = llm.invocar([sys, usr]) parche = json.loads(re.search(r”\[.*\]”, r.content, re.S).group()) return {“patch”: parche, “mensajes”: [AIMessage(content=json.dumps(patch))]} def node_apply(estado): return {“sandbox_rows”: apply_patch(estado[“raw_rows”]estado[“patch”])} def node_validate(estado): v = validar(estado[“sandbox_rows”]) devolver {“validación”: v, “mensajes”: [AIMessage(content=json.dumps(v))]} def node_approve(estado): decisión = interrupción({“validación”: estado[“validation”]}) devolver {“aprobado”: decisión == “aprobar”} def node_commit(estado): devolver {“mensajes”: [AIMessage(content=”COMMITTED”)]} def node_rollback(estado): devolver {“mensajes”: [AIMessage(content=”ROLLED BACK”)]}

Modelamos el estado interno del agente y definimos cada nodo en el flujo de trabajo de LangGraph. Expresamos el comportamiento del agente como pasos discretos e inspeccionables que transforman el estado preservando el historial del mensaje. Consulte los códigos completos aquí.

constructor = StateGraph(TxnState) builder.add_node(“perfil”, node_profile) builder.add_node(“patch”, node_patch) builder.add_node(“apply”, node_apply) builder.add_node(“validate”, node_validate) builder.add_node(“aprobar”, node_approve) builder.add_node(“commit”, node_commit) builder.add_node(“rollback”, node_rollback) builder.add_edge(START, “perfil”) builder.add_edge(“perfil”, “parche”) builder.add_edge(“parche”, “aplicar”) builder.add_edge(“aplicar”, “validar”) builder.add_conditional_edges( “validar”, lambda s: “aprobar” si s[“validation”][“ok”] else “revertir”, {“aprobar”: “aprobar”, “revertir”: “revertir”}) builder.add_conditional_edges( “aprobar”, lambda s: “confirmar” si s[“approved”] else “rollback”, {“commit”: “commit”, “rollback”: “rollback”} ) builder.add_edge(“commit”, END) builder.add_edge(“rollback”, END) app = builder.compile(checkpointer=InMemorySaver())

Construimos la máquina de estado LangGraph y codificamos explícitamente el flujo de control entre creación de perfiles, parches, validación, aprobación y finalización. Utilizamos ventajas condicionales para hacer cumplir las reglas de gobernanza en lugar de depender de decisiones de modelos implícitas. Consulte los códigos completos aquí.

def ejecutar(): estado = { “mensajes”: []”raw_rows”: MUESTRA_LEDGER, “sandbox_rows”: []”parche”: []”validación”: {}, “aprobado”: Ninguno, } cfg = {“configurable”: {“thread_id”: “txn-demo”}} out = app.invoke(state, config=cfg) if “__interrupt__” in out: print(json.dumps(out)[“__interrupt__”]sangría=2)) decisión = input(“aprobar/rechazar: “).strip() out = app.invoke(Command(resume=decisión), config=cfg) print(out[“messages”][-1].contenido) ejecutar()

Ejecutamos el agente transaccional y manejamos la aprobación humana mediante interrupciones gráficas. Reanudamos la ejecución de manera determinista, demostrando cómo los flujos de trabajo agentes pueden pausar, aceptar entradas externas y concluir de forma segura con una confirmación o una reversión.

En conclusión, mostramos cómo LangGraph nos permite crear agentes que razonan sobre estados, imponen puertas de validación y colaboran con humanos en puntos de control definidos con precisión. Tratamos al agente no como un oráculo, sino como un coordinador de transacciones que puede organizar, inspeccionar y revertir sus propias acciones mientras mantiene un seguimiento de auditoría completo. Este enfoque destaca cómo se puede aplicar la IA agente a sistemas del mundo real que requieren confianza, cumplimiento y capacidad de recuperación, y proporciona una base práctica para crear flujos de trabajo autónomos de nivel de producción que sigan siendo seguros, transparentes y supervisados ​​por humanos.

Consulte 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.

Asif Razzaq es el director ejecutivo de Marktechpost Media Inc.. Como emprendedor e ingeniero visionario, Asif está comprometido a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, Marktechpost, que se destaca por su cobertura en profundidad del aprendizaje automático y las noticias sobre aprendizaje profundo que es técnicamente sólida y fácilmente comprensible para una amplia audiencia. La plataforma cuenta con más de 2 millones de visitas mensuales, lo que ilustra su popularidad entre el público.