Por qué el cumplimiento es un problema de grafos
La mayoría de los frameworks de aplicaciones de IA optimizan para la flexibilidad — le das al modelo un prompt, él decide qué hacer a continuación, llama a herramientas, razona. Eso funciona bien para asistentes de propósito abierto. Es un problema en la gestión de cumplimiento.
En Zertia y en traze.ai, procesamos evidencias de auditoría para ISO 27001, ISO 9001 y GDPR en más de cincuenta clientes empresariales. La característica definitoria de un flujo de auditoría es que tiene una secuencia fija y auditable de pasos. Un auditor no quiere que una IA decida saltarse el paso de recopilación de evidencias porque se siente suficientemente segura. Un auditor quiere un sistema que demuestre — de forma comprobable, con un registro — que cada paso se ejecutó, en orden, con entradas y salidas específicas vinculadas.
Esta es la idea central que hace de LangGraph la herramienta correcta para la automatización del cumplimiento. LangGraph modela la ejecución como un grafo dirigido de transiciones de estado tipadas. Defines nodos (pasos de procesamiento discretos), defines aristas (la lógica de enrutamiento entre ellos) y defines un esquema de estado compartido que cada nodo lee y escribe. El grafo se compila una vez y se ejecuta de forma determinista. No puede saltarse nodos. No puede inventar aristas. La traza de ejecución es un objeto de primera clase.
Este último punto importa enormemente en contextos regulados. Bajo ISO 19011, las actividades de auditoría deben documentarse. Bajo el Artículo 5(2) del GDPR, los responsables del tratamiento deben demostrar el cumplimiento mediante el principio de responsabilidad proactiva. Una ejecución compilada de LangGraph te da ambas cosas: cada invocación de nodo tiene marca de tiempo, cada transición de estado se registra y el estado final lleva una cadena de procedencia completa.
Mapeando flujos de auditoría ISO a nodos de grafo
Una auditoría ISO 27001, simplificada, se ve así: planificar (definir alcance, objetivos, criterios) → recopilar evidencias (documentos, entrevistas, exportaciones del sistema) → evaluar controles (análisis de brechas contra el conjunto de controles del Anexo A) → generar hallazgos (no conformidades, observaciones) → producir informe (informe de auditoría con plan de acción).
Cada uno de esos es un nodo del grafo. El esquema de estado lleva todo lo que el grafo sabe en cualquier momento. Esta es la definición de estado que usamos en producción:
from typing import Annotated, TypedDict, Literal
from datetime import datetime
import operator
class ControlFinding(TypedDict):
control_id: str # e.g. "A.9.2.3"
status: Literal["conformant", "minor", "major", "na"]
evidence_refs: list[str]
justification: str
generated_at: datetime
class AuditState(TypedDict):
# Immutable inputs — set once, never mutated
audit_id: str
standard: Literal["ISO27001", "ISO9001", "GDPR"]
scope: str
criteria: list[str]
# Accumulated across nodes — operator.add reducer merges parallel writes
evidence: Annotated[list[dict], operator.add]
findings: Annotated[list[ControlFinding], operator.add]
trace: Annotated[list[str], operator.add]
# Mutable workflow state
current_stage: str
report_url: str | None
error: str | None
El patrón Annotated[list, operator.add] es importante. Las actualizaciones de estado en LangGraph son dicts parciales — un nodo devuelve solo las claves que modificó. El reducer operator.add significa que cuando dos nodos paralelos añaden a findings, la lista se fusiona correctamente en lugar de que un resultado sobrescriba al otro. Este es el mecanismo que hace segura la recopilación paralela de evidencias.
Definiciones de nodos y enrutamiento de aristas
Cada nodo es una función Python simple que recibe el estado actual y devuelve una actualización parcial. Mantenemos los nodos delgados: llaman a una capa de servicio, añaden a la traza y devuelven. Sin lógica de negocio dentro del nodo en sí.
from langgraph.graph import StateGraph, END
from datetime import datetime, timezone
from .services import evidence_service, evaluation_service, report_service
from .state import AuditState
def collect_evidence(state: AuditState) -> dict:
ts = datetime.now(timezone.utc).isoformat()
docs = evidence_service.fetch_for_scope(
audit_id=state["audit_id"],
criteria=state["criteria"],
)
return {
"evidence": docs,
"current_stage": "evidence_collected",
"trace": [f"collect_evidence completed at {ts}, {len(docs)} documents"],
}
def evaluate_controls(state: AuditState) -> dict:
ts = datetime.now(timezone.utc).isoformat()
findings = evaluation_service.evaluate(
evidence=state["evidence"],
standard=state["standard"],
)
return {
"findings": findings,
"current_stage": "controls_evaluated",
"trace": [f"evaluate_controls completed at {ts}, {len(findings)} findings"],
}
def route_after_evaluation(state: AuditState) -> str:
# Deterministic routing: if any major nonconformity, escalate.
has_major = any(f["status"] == "major" for f in state["findings"])
return "escalate" if has_major else "generate_report"
def generate_report(state: AuditState) -> dict:
ts = datetime.now(timezone.utc).isoformat()
url = report_service.generate(state)
return {
"report_url": url,
"current_stage": "report_generated",
"trace": [f"generate_report completed at {ts}"],
}
# Graph compilation
builder = StateGraph(AuditState)
builder.add_node("collect_evidence", collect_evidence)
builder.add_node("evaluate_controls", evaluate_controls)
builder.add_node("generate_report", generate_report)
builder.add_node("escalate", escalate_to_reviewer)
builder.set_entry_point("collect_evidence")
builder.add_edge("collect_evidence", "evaluate_controls")
builder.add_conditional_edges(
"evaluate_controls",
route_after_evaluation,
{"generate_report": "generate_report", "escalate": "escalate"},
)
builder.add_edge("generate_report", END)
builder.add_edge("escalate", END)
audit_graph = builder.compile()
La función de enrutamiento route_after_evaluation es un predicado Python simple — sin LLM involucrado. Esta es la disciplina clave: las llamadas al LLM viven dentro de los nodos; las decisiones de enrutamiento son Python determinista. El LLM te dijo status: "major" en el hallazgo, pero es Python quien decide a dónde ir a continuación basándose en ese valor. Las dos responsabilidades se mantienen separadas y testables de forma independiente.
Trazabilidad y patrones de pista de auditoría
En entornos regulados necesitas más que el checkpointing integrado de LangGraph. Necesitas una pista inmutable y de solo adición que se mapee a tus registros de evidencia de cumplimiento. El campo trace en AuditState es una capa — cada nodo registra lo que hizo y cuándo. Pero añadimos dos capas más encima.
LangSmith para observabilidad operativa
Las trazas de LangSmith te dan la vista completa a nivel de token de cada llamada al LLM: prompt de entrada, salida, latencia, versión del modelo y coste. Para una auditoría de cumplimiento quieres poder mostrar a un auditor exactamente qué se le preguntó al modelo y qué respondió cuando clasificó el control A.9.2.3. Configura LANGCHAIN_TRACING_V2=true y tus trazas se almacenan automáticamente. En producción en traze.ai etiquetamos cada ejecución con {"audit_id": "...", "client_id": "..."} mediante el parámetro metadata para poder recuperar la traza completa de cualquier auditoría específica desde la interfaz de LangSmith sin buscar en logs.
Log de eventos estructurado con PostgreSQL
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any
@dataclass
class AuditEvent:
audit_id: str
node: str
input_snapshot: dict[str, Any]
output_snapshot: dict[str, Any]
created_at: datetime = field(
default_factory=lambda: datetime.now(timezone.utc)
)
model_version: str | None = None
def make_audit_callback(db, audit_id: str):
"""
Returns a node wrapper that logs input/output to audit_events
before and after execution. The table has an append-only trigger
enforced at the database level — rows cannot be updated or deleted.
Usage: collect_evidence = make_audit_callback(db, id)(collect_evidence)
"""
def wrapper(node_fn):
def traced(state: AuditState) -> dict:
input_snap = {
k: state[k]
for k in ("current_stage", "standard", "audit_id")
}
result = node_fn(state)
event = AuditEvent(
audit_id=audit_id,
node=node_fn.__name__,
input_snapshot=input_snap,
output_snapshot=result,
)
db.audit_events.insert(event)
return result
return traced
return wrapper
Almacenamos snapshots de entrada y salida, no diffs. El almacenamiento es barato; reconstruir lo que pasó cuando un cliente importante disputa un hallazgo no lo es. La tabla audit_events tiene un trigger de solo adición aplicado a nivel de base de datos — el código de aplicación no puede actualizar ni eliminar filas.
Lecciones de producción en traze.ai
Después de ejecutar esta arquitectura en más de 50 empresas durante un año, las cosas que más nos sorprendieron no fueron técnicas.
La ubicación del human-in-the-loop importa más de lo que esperas. Inicialmente pusimos un paso de revisión humana solo en la etapa final del informe. Los clientes con procesos estrictos de aprobación interna necesitaban revisión después del paso de evaluación de controles, antes de que los hallazgos se bloquearan. Añadir un nodo de interrupción human_review con el mecanismo interrupt_before de LangGraph fue directo, pero nos obligó a repensar cómo serializábamos y reanudábamos el estado a través de solicitudes HTTP. Usa el checkpointer de Postgres de LangGraph desde el día uno — implementarlo después es doloroso.
El coste de tokens se acumula a escala. Una auditoría ISO 27001 de tamaño medio tiene aproximadamente 130 controles del Anexo A, cada uno requiriendo recuperación de evidencias y una llamada de evaluación al LLM. Con 50 clientes al mes eso son más de 6.500 llamadas al LLM por ciclo de auditoría. Agrupamos controles en lotes de 10 por llamada al LLM (salida estructurada con tipo de retorno lista), lo que redujo costes en ~73% sin pérdida de calidad medible. La lógica de agrupación vive en el nodo de evaluación, no en la estructura del grafo — la topología del grafo se mantuvo limpia.
Versionar el grafo es un requisito de cumplimiento. Si actualizas tu prompt de evaluación de controles, cada ejecución de auditoría después de ese cambio puede producir clasificaciones diferentes a las ejecuciones anteriores. Debes poder reproducir resultados históricos. Versionamos el grafo compilado junto con la versión del modelo y almacenamos ambos en el AuditEvent. Los despliegues solo afectan a nuevas auditorías; las auditorías en curso se ejecutan hasta completarse con la versión del grafo con la que empezaron, usando el estado con checkpoint de LangGraph.
Errores comunes: no determinismo, alucinaciones y estrategias de testing
La estructura del grafo es determinista. Las llamadas al LLM dentro de los nodos no lo son. Esta es la tensión central que debes gestionar.
La temperatura no es suficiente
Configurar temperature=0 reduce la varianza pero no la elimina. Tanto OpenAI como Anthropic documentan que incluso a temperatura cero, las salidas pueden variar entre versiones de API, actualizaciones de modelo y ejecución paralela. Para fines de cumplimiento, "reducir la varianza" no es suficiente — necesitas validar cada salida del LLM contra un esquema explícito antes de permitirle mutar el estado.
Usamos modelos Pydantic como contrato para cada llamada al LLM. El evaluation_service llama al modelo con with_structured_output(ControlFindingList) y re-lanza un ValidationError si el modelo devuelve un valor de estado fuera del conjunto Literal["conformant", "minor", "major", "na"]. El nodo entonces marca el hallazgo como requiriendo revisión manual en lugar de propagar silenciosamente un valor incorrecto.
Alucinación en citas de evidencias
El modo de fallo más dañino es un modelo citando evidencia que no existe — fabricando una referencia de documento que respalda un hallazgo "conformant". Mitigamos esto con generación fundamentada: la evidencia pasada al nodo de evaluación se fragmenta y se vectoriza en el momento de la ingesta; el LLM solo recibe los top-k fragmentos recuperados como contexto. Cada cita en la salida se valida contra el conjunto de IDs de fragmentos pasados al prompt. Las citas a IDs que no están en ese conjunto se rechazan y se marcan para revisión manual.
Estrategia de testing
Mantenemos tres capas de test:
- Tests unitarios de nodos — mock de la capa de servicio, aserciones sobre el dict de estado devuelto. Sin llamadas al LLM. Se ejecutan en CI en cada push, completan en menos de 10 segundos.
- Tests de integración del grafo — ejecutan el grafo compilado completo contra un fixture de auditoría sintético con respuestas LLM mockeadas (fixtures JSON deterministas). Aserciones sobre el estado final y contenidos de la traza. Validan la lógica de enrutamiento y el comportamiento de los reducers.
- Harness de evaluación — semanalmente, contra un conjunto de 30 auditorías históricas reales con hallazgos ground-truth conocidos. Mide la precisión de hallazgos, validez de citas y tasa de alucinación. Los resultados se almacenan en evaluaciones de LangSmith y se comparan con la ejecución de la semana anterior antes de aprobar cualquier actualización de modelo para producción.
El harness de evaluación es la inversión más valiosa. Llevó dos semanas construirlo y ha detectado tres regresiones en doce meses — cada vez que un proveedor de modelos actualizó silenciosamente sus pesos y nuestra precisión de clasificación cayó entre 4 y 8 puntos en controles de casos límite.
El grafo controla el flujo de trabajo. El modelo hace el razonamiento. Mantén esas responsabilidades separadas y podrás testear, auditar y explicar cada una de forma independiente.
Esa separación es lo que hace de LangGraph el sustrato correcto para la automatización del cumplimiento. El modelo no decide qué hacer a continuación. Informa una decisión que Python determinista toma. Los auditores entienden condicionales de Python. No entienden los pesos de atención de un transformer. Construye bajo esa restricción y la IA de cumplimiento se convierte en algo que realmente puedes certificar.