Inicio / Artículos / El historial de chat, los hechos, el estado del flujo de trabajo y los puntos de control son cuatro almacenamientos diferentes.

langchainlanggraphmemorycheckpointsagentspersistence

El historial de chat, los hechos, el estado del flujo de trabajo y los puntos de control son cuatro almacenamientos diferentes.

Deje de llamar “memoria” a todo. Separe las transcripciones de sesiones, los hechos duraderos, el estado del flujo de trabajo de los tickets y los puntos de control de LangGraph, estableciendo reglas de retención y autenticación para cada uno.

2467 palabras

Parte 9 de 14: Historial de chat separado, datos guardados y flujo de trabajo reanudable

Novena entrega de una serie de catorce artículos sobre la creación de un servicio de soporte técnico que guía a LangChain desde su primera llamada al modelo hasta las prácticas de producción. Los artículos posteriores convierten el sistema final en ejercicios para entrevistas.

La entrega anterior abordó la búsqueda en el manual de operaciones: consultar un corpus seleccionado, conservar metadatos de origen y bloquear consejos que hagan referencia a documentos que la búsqueda nunca devolvió.

Alguien de turno pregunta si el sistema puede “recordar” el incidente al día siguiente. La solicitud es poco específica: ¿guardar las conversaciones? Las preferencias del equipo? Las etiquetas y los fragmentos recuperados? Una escritura en espera de aprobación? ¿Cada campo intermedio del grafo? La gente agrupa todo esto bajo una etiqueta vaga. Cada uno necesita su propia clave, tiempo de vida, políticas de acceso y política ante fallos.

Esta parte separa cuatro conceptos:

chat history
  ordered messages for one conversation

saved facts
  selected application data about a user or accountworkflow state
  the current named values for one runcheckpoint
  a saved snapshot of workflow state that can be loaded later

Pasar mensajes anteriores a un ejecutable estático de LangChain no activa mágicamente la posibilidad de pausar y reanudar. Los hilos duraderos y las instantáneas provienen del modelo de estado y checkpointer de LangGraph.

El caso en estudio

Reutilicemos el incidente conocido como ejemplo:

Después del lanzamiento de las 14:05, las llamadas a checkout desde la región de la UE fallan. Los registros de checkout-api indican que la base de datos rechazó nuevas conexiones.

Asigne a cada tipo de registro su propio espacio de claves:

chat session:    chat:INC-2048
user facts:      user-17
workflow thread: ticket:INC-2048

Tratar el identificador del incidente como si fuera un identificador de persona fusiona espacios de nombres no relacionados. De igual manera, una única lista de transcripciones compartidas mezcla casos separados.

Primero, dejen de decir “la memoria”

Nombren el registro al que se refieren.

Historial de chat

Una lista ordenada como esta:

human: The failure began after 14:05.
assistant: I recorded the start time.
human: The failed requests are only in the EU region.
assistant: I added the affected region to the investigation context.

La secuencia es determinante cuando más adelante se ensamblan las instrucciones a partir de esos turnos.

Hechos guardados

Campos seleccionados como:

{
  "team": "commerce-platform",
  "timezone": "America/Los_Angeles"
}

Los hechos pueden sobrevivir más allá de un único chat. Únicamente se deben conservar mediante reglas explícitas de la aplicación, y no rastreando cada afirmación que invente el modelo.

Estatus del flujo de trabajo

Datos actuales para un proceso de ticket:

{
  "ticket_id": "INC-2048",
  "details": "checkout-api reports database connection refused",
  "classification": "database",
  "recommendation": "Compare database settings with the last good release.",
  "audit": [
    "ticket_received",
    "classified:database",
    "recommendation_created"
  ]
}

El estado cambia a medida que se ejecutan los pasos.

Punto de control

Considere un punto de control como una imagen congelada del flujo de trabajo, además de los registros contables que necesita el sistema en ejecución para continuar. Se carga la imagen más reciente de un hilo, se reanuda después de una pausa, se inspecciona lo que vio un paso y se recupera tras una falla. Los mecanismos de guardado locales al proceso desaparecen al salir; la recuperación en entornos de producción requiere un mecanismo de guardado respaldado por una base de datos.

La recuperación no es ninguna de estas opciones

El índice del manual de procedimientos curado funciona como un corpus de búsqueda. Abrirlo durante un incidente no convierte los resultados en historial de chat. Los fragmentos no deben promocionarse automáticamente a datos permanentes del perfil. Los índices de embeddings no son bases de datos de puntos de control. Aíslen las bases de datos incluso cuando una sola solicitud HTTP afecte a varias.

Qué recuerda por defecto una cadena fija

A través de invocaciones independientes, la cadena no recuerda nada a menos que su aplicación inyecte o guarde contexto.

Esta llamada:

result = chain.invoke(current_input)

no reenvía automáticamente las entradas o salidas anteriores. Puede añadir usted mismo las interacciones previas o utilizar envoltorios de historial más antiguos. En la versión de LangChain analizada aquí, RunnableWithMessageHistory emite advertencias y orienta el nuevo trabajo hacia la persistencia en LangGraph.

Con una cadena estática, gestionar el historial en el código de la aplicación suele ser más sencillo de comprender:

read permitted messages
  -> select the messages needed for this request
  -> call the chain
  -> store the new turn under the correct session ID

Eso es lo que hace el código complementario.

Estructura del proyecto

La captura de estado de la Parte 9 contiene:

langchain-helpdesk/
├── app.py
├── checkpoint_graph.py
├── facts.py
├── history.py
└── tests/
    └── test_state.py

Instalar paquetes:

python -m pip install -U langchain-core langgraph pydantic pytest

El ejemplo no realiza llamadas a ningún proveedor.

Paso 1: Almacenar mensajes de chat por sesión

Crear history.py:

from dataclasses import dataclass, field

from langchain_core.messages import (
    AIMessage,
    BaseMessage,
    HumanMessage,
)
@dataclass
class ChatHistoryStore:
    histories: dict[str, list[BaseMessage]] = field(
        default_factory=dict
    )    def read(self, session_id: str) -> list[BaseMessage]:
        return list(self.histories.get(session_id, []))    def add_turn(
        self,
        session_id: str,
        user_text: str,
        reply_text: str,
    ) -> None:
        history = self.histories.setdefault(session_id, [])
        history.extend(
            [
                HumanMessage(content=user_text),
                AIMessage(content=reply_text),
            ]
        )    def prior_turn_count(self, session_id: str) -> int:
        return len(self.histories.get(session_id, [])) // 2

Las claves histories relacionan una sesión con una lista ordenada. read crea copias para que quienes lo llamen no puedan agregar elementos al almacenamiento por efecto secundario. add_turn registra una pareja de usuario/asistente. prior_turn_count reduce la longitud a la mitad porque el ejemplo solo almacena pares completos. Las transcripciones en tiempo real también incluyen mensajes de herramientas, turnos incompletos y errores; no asuma que habrá una asociación ordenada en entornos de producción.

La historia necesita una regla de retención

Mantener cada turno para siempre no es una característica del producto. La política debe especificar qué se puede conservar, por cuánto tiempo, quién puede leerlo, qué campos se deben ocultar, cómo funciona la eliminación y cuántos turnos se incluyen en la siguiente llamada al modelo. Los historiales demasiado grandes desperdician tokens y dinero. Los resúmenes pueden ser útiles, pero también generan errores; trátelos como artefactos derivados con reglas claras de procedencia.

Paso 2: Almacenar los hechos seleccionados por separado

Cree facts.py:

from dataclasses import dataclass, field

@dataclass
class UserFactsStore:
    records: dict[str, dict[str, str]] = field(
        default_factory=dict
    )    def put(self, user_id: str, key: str, value: str) -> None:
        self.records.setdefault(user_id, {})[key] = value    def get(self, user_id: str) -> dict[str, str]:
        return dict(self.records.get(user_id, {}))

Indexe los hechos por user_id, nunca por ID de sesión o incidente. Acepte solo campos con nombre definido; nunca almacene toda la transcripción bajo una única clave. Las rutas reales de put necesitan listas de permisos, validación, autorización y eventos de auditoría. Una indicación del modelo no constituye autorización para persistirlo.

Paso 3: Definir el estado del flujo de trabajo

Pase a un flujo de trabajo con estado. Use TypedDict en checkpoint_graph.py:

from operator import add
from typing import Annotated, TypedDict

class TicketWorkflowState(TypedDict, total=False):
    ticket_id: str
    details: str
    classification: str
    recommendation: str
    audit: Annotated[list[str], add]

total=False permite que los campos permanezcan sin valor hasta que un nodo los escriba. Los eventos de auditoría utilizan un reductor:

Annotated[list[str], add]

Cuando un nodo devuelve más filas de auditoría, el reductor las concatena en lugar de sobrescribirlas. Elija los reductores intencionadamente: use “append” para flujos de eventos y “replace” para campos escalares.

Paso 4: Crear nodos pequeños y deterministas

Teaching Graph utiliza Python puro, por lo que el comportamiento de los puntos de control es visible:

def classify_node(state: TicketWorkflowState) -> TicketWorkflowState:
    details = state["details"].lower()

    if "database" in details or "connection refused" in details:
        category = "database"
    elif "access" in details or "role" in details:
        category = "access"
    else:
        category = "unknown"    return {
        "classification": category,
        "audit": [f"classified:{category}"],
    }

Cada nodo lee el estado y devuelve una actualización; nunca modifica el diccionario de entrada. El nodo de recomendaciones consume los resultados de la clasificación:

def recommend_node(state: TicketWorkflowState) -> TicketWorkflowState:
    category = state["classification"]

    if category == "database":
        recommendation = (
            "Compare database settings with the last good release."
        )
    elif category == "access":
        recommendation = (
            "Confirm the requested role and current access policy."
        )
    else:
        recommendation = "Ask a person to classify the ticket."    return {
        "recommendation": recommendation,
        "audit": ["recommendation_created"],
    }

Se trata de funciones normales de Python; esta sección se centra en el estado y la persistencia, no en la precisión del clasificador.

Paso 5: Construir el grafo

from langgraph.graph import END, START, StateGraph

def build_checkpointed_graph(checkpointer=None):
    builder = StateGraph(TicketWorkflowState)
    builder.add_node("classify", classify_node)
    builder.add_node("recommend", recommend_node)
    builder.add_edge(START, "classify")
    builder.add_edge("classify", "recommend")
    builder.add_edge("recommend", END)    return builder.compile(
        checkpointer=checkpointer or InMemorySaver()
    )

StateGraph(TicketWorkflowState) relaciona el estado compartido con el diccionario tipado. Los nodos y aristas determinan el orden; compile valida la estructura y conecta los puntos de verificación. La ruta se mantiene lineal: el grafo funciona bien porque el estado y los puntos de verificación son de primera categoría, no porque el diagrama sea sofisticado.

Paso 6: Asignar un ID de hilo a cada flujo de trabajo

def thread_config(thread_id: str) -> dict[str, dict[str, str]]:
    return {"configurable": {"thread_id": thread_id}}

Ejecutar la tarea:

config = thread_config("ticket:INC-2048")

result = graph.invoke(
    {
        "ticket_id": "INC-2048",
        "details": (
            "checkout-api reports database connection refused"
        ),
        "audit": ["ticket_received"],
    },
    config,
)

thread_id divide la historia de los puntos de verificación. Reutilizar un mismo hilo en incidentes no relacionados provoca fugas de estado entre ellos.

Paso 7: Leer el estado guardado

snapshot = graph.get_state(config)
print(snapshot.values)

Los valores contienen:

{
  "ticket_id": "INC-2048",
  "details": "checkout-api reports database connection refused",
  "classification": "database",
  "recommendation": "Compare database settings with the last good release.",
  "audit": [
    "ticket_received",
    "classified:database",
    "recommendation_created"
  ]
}

Esa instantánea es únicamente datos del flujo de trabajo; no constituye ni un almacén de perfiles ni el corpus del manual de operaciones.

Qué puede y qué no puede hacer InMemorySaver

Mantiene los puntos de control solo durante la vida útil del proceso de Python, lo cual es adecuado para pruebas unitarias y cuadernos de notas. No sobrevivirá a los reinicios, no abarcará réplicas de servicio ni satisfará las necesidades de retención, cifrado o respaldo. La documentación actual orienta la memoria del agente de producción y los hilos reanudables hacia un mecanismo de almacenamiento basado en base de datos, como Postgres.

Estructura para producción:

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    graph = build_checkpointed_graph(checkpointer)

Los secretos de conexión, las migraciones, la gestión de conexiones compartidas y la limpieza siguen siendo responsabilidades de la aplicación. Mantenga las URI de la base de datos fuera del código fuente comprometido.

Dónde termina LangChain y comienza LangGraph

LangChain fijo es suficiente cuando

la secuencia es fija; una sola solicitud puede completarse sin necesidad de pausas humanas; está bien reiniciar toda la solicitud; el estado intermedio no necesita ser duradero; el código de aplicación ordinario puede almacenar el pequeño historial que se necesita.

LangGraph es más adecuado cuando

Los bucles o ramas de flujo de control operan sobre estados nombrados; es necesario que una persona apruebe el proceso durante su ejecución; el trabajo continúa más tarde en el mismo hilo de ejecución; al reiniciar un proceso se debe mantener el estado pendiente; los operadores necesitan instantáneas inspeccionables; la recuperación debe reanudarse desde un punto guardado en lugar de comenzar de nuevo.

La función auxiliar create_agent actual ya devuelve un agente establecido en LangGraph. Se proporciona un punto de control y los flujos de persistencia provienen de ese entorno de ejecución; se trata de la arquitectura esperada, no de una fuga accidental.

Un punto de control no es un registro de auditoría

Los puntos de control existen para que el entorno de ejecución pueda continuar. Los registros de auditoría existen para que los responsables de seguridad y de gestión empresarial puedan reconstruir las acciones realizadas. A veces comparten campos, pero sus propósitos son diferentes. Una fila de auditoría debe indicar el solicitante, la herramienta propuesta, quien aprobó, los argumentos utilizados, el resultado y la fecha y hora. No se debe tratar una instantánea serializada interna como un registro de auditoría de nivel de cumplimiento.

Puntos de control y efectos secundarios

El estado persistente no hace que una escritura externa sea idempotente. Si el proceso actualiza un ticket y luego se detiene antes del siguiente punto de control, la reanudación podría repetir la escritura. Las herramientas necesitan claves de idempotencia o verificaciones de “ya aplicado”. Coloque los efectos secundarios después de la aprobación; firme los IDs de operación estables; documente las semánticas de reintentos. La siguiente etapa se ejecuta antes de la escritura en el ticket y determina si se aprueba o rechaza.

Pruebe la separación

Tres pruebas sin conexión en el snapshot complementario.

Los historiales de mensajes permanecen separados

history.add_turn("chat:first", "First note", "First reply")
history.add_turn("chat:first", "Second note", "Second reply")
history.add_turn("chat:second", "Other ticket", "Other reply")

assert history.prior_turn_count("chat:first") == 2
assert history.prior_turn_count("chat:second") == 1

Los hechos guardados no son mensajes de chat

facts.put("user-17", "team", "commerce-platform")

assert facts.get("user-17") == {
    "team": "commerce-platform"
}
assert history.read("user-17") == []

Los puntos de control permanecen separados por hilo de ticket

first, first_config = run_ticket(
    graph,
    "INC-2048",
    "checkout-api reports database connection refused",
)
second, second_config = run_ticket(
    graph,
    "INC-2050",
    "identity-api denied an access role request",
)

assert graph.get_state(first_config).values["ticket_id"] == "INC-2048"
assert graph.get_state(second_config).values["ticket_id"] == "INC-2050"

Ejecute:

pytest -q

Se espera:

3 passed

Prueban los límites de los espacios de nombres y la aislación de hilos, pero no demuestran la durabilidad de la base de datos al utilizar InMemorySaver.

Errores comunes

Una sola lista de historial global

Esto provoca colisiones entre usuarios o incidentes no relacionados. Siempre identifique el historial con un identificador autenticado y de alcance restringido.

Guardar cada declaración del modelo como un hecho

Los modelos se crean con confianza excesiva. Solo persista los campos autorizados a través de un camino de escritura validado.

Almacenar secretos en el estado

Las instantáneas se copian, inspeccionan y conservan. Deje los secretos en un lugar seguro y proporcione referencias en su lugar.

Usar thread_id como autorización

Un ID de hilo permite acceder al estado, pero nunca demuestra que quien lo solicita tenga permiso para leerlo. Autorice por separado.

Calificar a un almacén de vectores como “memoria a largo plazo”

El eslogan oculta la propiedad y la eliminación de datos. Indique los registros, los autores, la ruta de consulta y la política de eliminación.

Esperar un punto de control para reparar un error

Las instantáneas conservan todo lo que se escribió, incluidos los errores. La validación y las pruebas siguen siendo obligatorias.

Resultado de la Parte 9

Cuatro límites de almacenamiento con nombre:

session ID -> ordered chat messages
user ID    -> selected saved facts
thread ID  -> current workflow state
checkpoint -> persisted workflow snapshot

Una cadena estática sigue siendo adecuada para tareas de un solo paso. LangGraph resulta más claro cuando se necesita un estado duradero, posponer, reanudar o recuperar datos. A continuación viene la primera decisión real del agente: las herramientas de solo lectura pueden ejecutarse automáticamente; las mutaciones de tickets esperan a que una persona tome una decisión.

Revisión de documentación: verificada en comparación con la documentación sobre memoria a corto plazo de LangChain y la persistencia en LangGraph el 26 de agosto de 2026. Las APIs del paquete han cambiado.

Leer más: memoria a corto plazo de LangChain, agentes de LangChain, persistencia en LangGraph.

Los sistemas de servicio de soporte para la producción suelen necesitar las cuatro almacenaciones al mismo tiempo: un búfer de chat con alcance por sesión para el ingeniero actual, un almacén de datos con alcance por usuario para las preferencias duraderas, un estado del flujo de trabajo con alcance por hilo para el grafo de tickets, y un índice de manual operativo buscable que nunca sirva como historial ni como punto de control. Nombrar esos límites en las revisiones de código evita el atajo común de meter todo en una sola lista de Redis etiquetada como “memory”. Al incorporar a un nuevo compañero, pídale que dibuje las cuatro cajas y etiquete las claves; si no puede hacerlo, el diseño aún no está listo para interrumpir y reanudar.

Cuando más adelante se introduce la aprobación humana (Parte 10), el punto de control se convierte en el lugar donde el flujo de trabajo se detiene a la espera. El historial de chat continúa de forma independiente, de modo que el ingeniero puede hacer preguntas para aclarar cosas sin modificar la escritura pendiente. Los datos permanecen fuera de la ruta de interrupción a menos que una regla explícita copie un campo. Esa separación es lo que impide que “reanudar después del almuerzo” se convierta en “reproducir toda la conversación en una actualización de ticket”.

Mezclar políticas de retención entre almacenamientos

Las transcripciones de chat, los datos duraderos, los puntos de control del flujo de trabajo y las incrustaciones de manuales de operaciones casi nunca comparten el mismo reloj de retención. Alinearlos “por simplicidad” suele violar ya sea las solicitudes de eliminación por privacidad o las necesidades de reproducir incidentes. Documente cuatro relojes, cuatro responsables y cuatro puntos de entrada para la eliminación, incluso si dos de ellos apuntan actualmente a la misma instancia de Redis.

Tratar las advertencias de obsolescencia como opcionales

Cuando la biblioteca advierte que los envoltorios de historial se están trasladando a la persistencia de LangGraph, considérelo como una señal de diseño. Lanzar una nueva función de soporte técnico por la ruta obsoleta implica tener que reescribirla más adelante dentro del plazo establecido. Prefiera el modelo de checkpointer para cualquier flujo que pueda detenerse.

Olvidar que los reducers forman parte del esquema

Los equipos discuten los nombres de los campos durante horas y luego añaden casualmente un reducer de apéndice a un campo que debería reemplazarlo. El error aparece semanas después en forma de clasificaciones duplicadas o filas de auditoría eliminadas. Revise los reducers en la misma solicitud de cambios que el TypedDict.

Lecturas relacionadas

Todos los artículosEnviar el brief