Saltar al contenido principal
Jordi Izquierdo Montalbán
Proyectos

Arquitectura Empresarial con LangGraph

Diseño de grafos cíclicos con reductores de estado acumulativos, persistencia duradera en SQLite (SqliteSaver), pausas Human-in-the-Loop, rebobinado temporal (Time Travel), memoria cross-thread y streaming granular de eventos.

Código privado: te lo enseño en una entrevista

LangGraph, Python, SQLite (SqliteSaver), BaseStore, TypedDict, Streaming, pytest

El problema

Los agentes hechos con cadenas secuenciales simples o bucles while improvisados se rompen ante requisitos empresariales reales:

  • Si el servidor se reinicia, la memoria RAM se borra y la conversación se pierde.
  • Cuando una interacción se alarga, los mensajes saturan la ventana de contexto y los costes se disparan.
  • Si el modelo se equivoca en un paso intermedio, no hay forma de rebobinar y corregirlo sin empezar de cero.
  • Una ejecución monolítica impide modularizar y testear los componentes por separado.

La solución

Estructura del núcleo de LangGraph: estado con reductores → nodos y aristas condicionales → checkpoints duraderos en SQLite → Time Travel (inspección y rebobinado) → memoria cross-thread.

Estudié a fondo los seis módulos de la arquitectura de LangGraph y con ellos construí un motor de agentes con estado tipado, persistencia en base de datos, rebobinado temporal y streaming en tiempo real:

  1. Reductores de estado tipados (state-reducers). Modelo el estado compartido con TypedDict y operadores reductores acumulativos (Annotated[list, operator.add] y add_messages). Varios nodos escriben en el estado global sin pisar los datos de los demás.
  2. Persistencia y puntos de control (checkpoints). Guardo el estado en SQLite (SqliteSaver.from_conn_string("checkpoints.db")). Cada paso del grafo genera una instantánea inmutable indexada por thread_id y checkpoint_id, así que una conversación interrumpida se reanuda sin perder contexto.
  3. Control humano y Time Travel. Reviso el historial de estados con graph.get_state_history(config). Si el agente toma una rama que no quiero, vuelvo a un checkpoint anterior, edito los datos con graph.update_state() y bifurco (branching) la ejecución.
  4. Ventana de contexto y memoria cross-thread:
    • Borro los mensajes obsoletos con RemoveMessage(id=...) y los sustituyo por resúmenes dinámicos, para que el consumo de tokens no crezca.
    • Guardo la memoria a largo plazo fuera del hilo (BaseStore, InMemoryStore), en espacios de nombres (user_id, "memories"), de modo que el agente recuerda las preferencias del usuario entre sesiones.
  5. Subgrafos jerárquicos. Divido las arquitecturas complejas en subgrafos modulares, cada uno con su contrato de entrada y salida (InputState, OutputState), que se invocan como nodos de un grafo principal.
  6. Streaming de producción. Emito eventos en tiempo real con modos configurables (stream_mode="updates", "values" o "messages"), para que el usuario no espere a ver nada.

Stack desglosado

  • Orquestación: LangGraph (StateGraph, MessagesState, Command, interrupt).
  • Persistencia: SQLite (SqliteSaver) y BaseStore para memoria persistente multi-hilo.
  • Validación y tipado: Pydantic y Python TypedDict.
  • Streaming: Modos nativos de streaming de eventos y valores.
  • Empaquetado: Manifiesto langgraph.json para despliegue en producción.

Resultados y aprendizajes

Lo que funcionó fue tratar el estado como una base de datos de eventos inmutables. Con los checkpoints, un agente tiene la fiabilidad que se le pide a cualquier backend transaccional.

Lo que aprendí: construir agentes para producción no va de escribir mejores prompts. Va de diseñar grafos con estado resistente, control humano incorporado y límites claros de memoria.