En este tutorial, implementamos un patrón de IA agente usando LangGraph que proxenetismo el razonamiento y la entusiasmo como un flujo de trabajo transaccional en motivo de una valor ú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 cerca de flujos de trabajo de IA estructurados y conscientes de la gobernanza que se ejecutan de forma confiable en Google Colab utilizando modelos OpenAI. Mira el Códigos completos aquí.
!pip -q install -U langgraph langchain-openai
import os, json, uuid, copy, math, re, operator
from typing import Any, Dict, List, Optional
from typing_extensions import TypedDict, Annotated
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, AnyMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
def _set_env_openai():
if os.environ.get("OPENAI_API_KEY"):
return
try:
from google.colab import userdata
k = userdata.get("OPENAI_API_KEY")
if k:
os.environ("OPENAI_API_KEY") = k
return
except Exception:
pass
import getpass
os.environ("OPENAI_API_KEY") = getpass.getpass("Enter OPENAI_API_KEY: ")
_set_env_openai()
MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")
llm = ChatOpenAI(model=MODEL, temperature=0)Configuramos el entorno de ejecución instalando LangGraph e inicializando el maniquí OpenAI. Cargamos de forma segura la secreto API y configuramos un LLM determinista, asegurando que todo el comportamiento de los agentes posteriores siga siendo reproducible y controlado. Mira el Códigos completos aquí.
SAMPLE_LEDGER = (
{"txn_id": "T001", "name": "Asha", "email": "(email protected)", "amount": "1,250.50", "date": "12/01/2025", "note": "Membership renewal"},
{"txn_id": "T002", "name": "Ravi", "email": "(email protected)", "amount": "-500", "date": "2025-12-02", "note": "Chargeback?"},
{"txn_id": "T003", "name": "Sara", "email": "(email protected)", "amount": "700", "date": "02-12-2025", "note": "Late fee waived"},
{"txn_id": "T003", "name": "Sara", "email": "(email protected)", "amount": "700", "date": "02-12-2025", "note": "Duplicate row"},
{"txn_id": "T004", "name": "Lee", "email": "(email protected)", "amount": "NaN", "date": "2025/12/03", "note": "Bad amount"},
)
ALLOWED_OPS = {"replace", "remove", "add"}
def _parse_amount(x):
if isinstance(x, (int, float)):
return float(x)
if isinstance(x, str):
try:
return float(x.replace(",", ""))
except:
return None
return None
def _iso_date(d):
if not isinstance(d, str):
return None
d = d.replace("/", "-")
p = d.split("-")
if len(p) == 3 and len(p(0)) == 4:
return d
if len(p) == 3 and len(p(2)) == 4:
return f"{p(2)}-{p(1)}-{p(0)}"
return None
def profile_ledger(rows):
seen, anomalies = {}, ()
for i, r in enumerate(rows):
if _parse_amount(r.get("amount")) is None:
anomalies.append(i)
if r.get("txn_id") in seen:
anomalies.append(i)
seen(r.get("txn_id")) = i
return {"rows": len(rows), "anomalies": anomalies}
def apply_patch(rows, patch):
out = copy.deepcopy(rows)
for op in sorted((p for p in patch if p("op") == "remove"), key=lambda x: x("idx"), reverse=True):
out.pop(op("idx"))
for op in patch:
if op("op") in {"add", "replace"}:
out(op("idx"))(op("field")) = op("value")
return out
def validate(rows):
issues = ()
for i, r in enumerate(rows):
if _parse_amount(r.get("amount")) is None:
issues.append(i)
if _iso_date(r.get("date")) is None:
issues.append(i)
return {"ok": len(issues) == 0, "issues": issues}Definimos la conceptualización del volumen anciano central contiguo con la deducción de parcheo, normalización y firmeza. Tratamos las transformaciones de datos como operaciones reversibles, lo que permite al agente razonar sobre los cambios de forma segura ayer de realizarlos. Mira el Códigos completos aquí.
class TxnState(TypedDict):
messages: Annotated(List(AnyMessage), add_messages)
raw_rows: List(Dict(str, Any))
sandbox_rows: List(Dict(str, Any))
patch: List(Dict(str, Any))
validation: Dict(str, Any)
approved: Optional(bool)
def node_profile(state):
p = profile_ledger(state("raw_rows"))
return {"messages": (AIMessage(content=json.dumps(p)))}
def node_patch(state):
sys = SystemMessage(content="Return a JSON patch list fixing amounts, dates, emails, duplicates")
usr = HumanMessage(content=json.dumps(state("raw_rows")))
r = llm.invoke((sys, usr))
patch = json.loads(re.search(r"(.*)", r.content, re.S).group())
return {"patch": patch, "messages": (AIMessage(content=json.dumps(patch)))}
def node_apply(state):
return {"sandbox_rows": apply_patch(state("raw_rows"), state("patch"))}
def node_validate(state):
v = validate(state("sandbox_rows"))
return {"validation": v, "messages": (AIMessage(content=json.dumps(v)))}
def node_approve(state):
decision = interrupt({"validation": state("validation")})
return {"approved": decision == "approve"}
def node_commit(state):
return {"messages": (AIMessage(content="COMMITTED"))}
def node_rollback(state):
return {"messages": (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. Mira el Códigos completos aquí.
builder = StateGraph(TxnState)
builder.add_node("profile", node_profile)
builder.add_node("patch", node_patch)
builder.add_node("apply", node_apply)
builder.add_node("validate", node_validate)
builder.add_node("approve", node_approve)
builder.add_node("commit", node_commit)
builder.add_node("rollback", node_rollback)
builder.add_edge(START, "profile")
builder.add_edge("profile", "patch")
builder.add_edge("patch", "apply")
builder.add_edge("apply", "validate")
builder.add_conditional_edges(
"validate",
lambda s: "approve" if s("validation")("ok") else "rollback",
{"approve": "approve", "rollback": "rollback"}
)
builder.add_conditional_edges(
"approve",
lambda s: "commit" if 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, firmeza, aprobación y finalización. Utilizamos ventajas condicionales para hacer cumplir las reglas de gobernanza en motivo de subordinarse de decisiones de modelos implícitas. Mira el Códigos completos aquí.
def run():
state = {
"messages": (),
"raw_rows": SAMPLE_LEDGER,
"sandbox_rows": (),
"patch": (),
"validation": {},
"approved": None,
}
cfg = {"configurable": {"thread_id": "txn-demo"}}
out = app.invoke(state, config=cfg)
if "__interrupt__" in out:
print(json.dumps(out("__interrupt__"), indent=2))
decision = input("approve / reject: ").strip()
out = app.invoke(Command(resume=decision), config=cfg)
print(out("messages")(-1).content)
run()Ejecutamos el agente transaccional y manejamos la aprobación humana mediante interrupciones gráficas. Reanudamos la ejecución de forma 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 firmeza 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 actual que requieren confianza, cumplimiento y capacidad de recuperación, y proporciona una pulvínulo praxis para crear flujos de trabajo autónomos de nivel de producción que sigan siendo seguros, transparentes y supervisados por humanos.
Mira el Códigos completos aquí. Adicionalmente, no dudes en seguirnos en Gorjeo y no olvides unirte a nuestro SubReddit de más de 100.000 ml y suscríbete a nuestro boletín. ¡Esperar! estas en telegrama? Ahora asimismo puedes unirte a nosotros en Telegram.
Asif Razzaq es el director ejecutante de Marktechpost Media Inc.. Como emprendedor e ingeniero iluminado, Asif está comprometido a beneficiarse el potencial de la inteligencia químico para el correctamente social. Su esfuerzo más flamante es el extensión de una plataforma de medios de inteligencia químico, Marktechpost, que se destaca por su cobertura en profundidad del enseñanza instintivo y las informativo sobre enseñanza 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 notorio.