Estado tipado

2.1El plan que se borró a sí mismo

En L1 dejaste el estado de Elcano como una variable local mensajes. Vamos a darle forma —y a tropezar con el primer bug que el tipado previene.

Elcano ya planifica. Investiga la cuenta, escribe un plan de tres pasos y lo guarda en el estado para ejecutarlo. Defines el estado como un diccionario con un campo plan y un campo pasos_hechos. Dos nodos lo tocan: el planificador escribe el plan, el ejecutor añade a pasos_hechos cada paso completado.

Lanzas un caso real. El ejecutor termina el paso 1, devuelve {"pasos_hechos": ["consultó la cuenta"]} para registrarlo. Y al mirar el estado, pasos_hechos tiene un solo elemento. El paso 2 lo vuelve a dejar con uno. El historial de lo hecho se borra en cada actualización.

El reflejo es buscar el bug en la lógica del ejecutor. No está ahí. Está en cómo definiste el estado: por defecto, cada campo se sobrescribe. Devolver {"pasos_hechos": [...]} no añade a la lista; la reemplaza entera. Para que un campo acumule en vez de pisar, hay que decírselo con un reducer. Esta lección es sobre esa decisión —sobrescribir o acumular— porque elegirla mal corrompe el estado en silencio.

2.2Qué vas a poder hacer

Al terminar serás capaz de:

• Definir el estado de un agente con TypedDict y type hints.
• Distinguir overwrite (sobrescribir) de acumulación, y elegir el reducer correcto por campo.
• Usar add_messages y MessagesState para el historial de mensajes, y operator.add para listas que crecen.
• Decidir cuándo Pydantic como estado aporta y cuándo sus límites no compensan.

Necesitas saber antes:

• De N0·L1: que el estado del agente debe ser explícito y legible, no una variable local.
• Qué es un TypedDict: un diccionario con un tipo declarado por clave (de typing_extensions).
• Qué es la anotación de tipo Annotated[tipo, metadato]: un tipo con metadatos adjuntos.

Esta lección define el estado. Los nodos que lo leen y actualizan llegan en L3.

2.3Recupera

Antes de seguir, responde de memoria.

1. En el agente-bucle de L1, ¿dónde vivía el estado y por qué no podías auditarlo?
2. Si un campo del estado es una lista y dos nodos devuelven una lista para ese campo, ¿qué crees que pasa por defecto: se juntan o se pisa una a otra?
3. ¿Qué tipo de dato lleva el historial de conversación de un agente?

2.4El concepto: estado, reducers y la decisión por campo

Tres ideas, en orden: cómo se declara el estado, qué hace un reducer, y cuál elegir para cada campo de Elcano.

El estado es un esquema tipado

El estado de un grafo en LangGraph se define con un TypedDict: un esquema que declara qué claves tiene y de qué tipo. Cada nodo recibe el estado, hace su trabajo y devuelve un diccionario parcial con los campos que quiere actualizar. LangGraph combina ese parcial con el estado actual.

La pregunta clave es cómo combina. Y la respuesta tiene una regla por defecto y una excepción.

1from typing_extensions import Annotated, TypedDict
2import operator
3
4class State(TypedDict):
5    messages: Annotated[list, operator.add]  # reducer: append
6    counter: int                             # sin reducer: overwrite

• Campo sin reducer (counter) → overwrite. La actualización reemplaza el valor anterior. Es el comportamiento por defecto.
• Campo Annotated[tipo, reducer] (messages) → el reducer combina el valor viejo y el nuevo. Aquí operator.add aplica viejo + nuevo, que en listas es append.

Ese es el mecanismo entero. Un reducer es la función que decide cómo se fusiona la actualización de un nodo con el estado existente. Sin reducer, la fusión es "tira lo viejo".

La analogía: el estado es una pizarra compartida entre nodos. Por defecto, escribir en una casilla borra lo que había. Un reducer es una regla de escritura distinta para esa casilla: "no borres, añade debajo". Límite de la analogía: en una pizarra física tú controlas el orden de escritura. Aquí varios nodos pueden actualizar la misma casilla en el mismo superstep, y el reducer es lo único que evita que se pisen.

El reducer de mensajes: add_messages

El historial de conversación no quiere overwrite —quieres que cada turno se añada—. Y quiere algo más que un append ciego. LangGraph trae el reducer hecho.

add_messages (de langgraph.graph.message, también exportado en langgraph.graph) es el reducer de mensajes: hace append, deduplica por id y admite RemoveMessage para borrar mensajes. Esa deduplicación por id importa: si un nodo reescribe un mensaje con el mismo id, lo actualiza en su sitio en vez de duplicarlo.

No tienes que cablearlo a mano cada vez. LangGraph trae un estado prebuilt:

1# MessagesState es un TypedDict prebuilt:
2#   messages: Annotated[list[AnyMessage], add_messages]
3from langgraph.graph import MessagesState
4
5class State(MessagesState):
6    extra_field: int

MessagesState es un TypedDict prebuilt cuyo único campo es messages: Annotated[list[AnyMessage], add_messages]. Heredas de él y añades tus campos. Así el historial ya viene con el reducer correcto y tú solo declaras lo tuyo.

La decisión, sobre Elcano

Ahora el bug del §2.1 tiene cura. El estado de Elcano necesita decisiones distintas por campo:

1from typing_extensions import Annotated
2from langgraph.graph import MessagesState
3import operator
4
5class EstadoOps(MessagesState):              # hereda messages con add_messages
6    cuenta_id: str                           # overwrite: identifica el thread, no cambia
7    plan: list[str]                          # overwrite: replanificar REEMPLAZA el plan
8    pasos_hechos: Annotated[list, operator.add]   # append: el historial de pasos ACUMULA

Lee la decisión campo a campo, porque es exactamente lo que la rúbrica C0 evalúa:

• messages → add_messages (heredado). Append con dedup por id. Es la conversación; nunca se pisa.
• cuenta_id → overwrite. Es un identificador fijo del run. No acumula.
• plan → overwrite, a propósito. Cuando Elcano replanifica, el plan nuevo sustituye al viejo; no quieres concatenar dos planes contradictorios.
• pasos_hechos → operator.add. Aquí está el arreglo del §2.1: este campo acumula. Cada ejecutor añade su paso sin borrar los anteriores.

El bug del plan que se borraba era pasos_hechos sin reducer. La cura es una anotación: Annotated[list, operator.add].

Normaliza el error: el reducer equivocado corrompe en silencio

Es común confundir los dos fallos simétricos, y ninguno lanza excepción.

• Acumular donde querías sobrescribir. Si pones plan con operator.add, replanificar concatena el plan nuevo al viejo. Elcano arrastra pasos obsoletos y ejecuta de más.
• Sobrescribir donde querías acumular. El bug del §2.1: pasos_hechos sin reducer pierde el historial.

La diferencia clave: overwrite para lo que representa un valor actual (el plan vigente, el id de cuenta); reducer para lo que representa una historia (los mensajes, los pasos hechos). Pregúntate por cada campo: ¿esto es un estado actual o un registro que crece?

Pydantic como estado: qué aporta y qué no

Quizá prefieras Pydantic BaseModel por la validación. Se puede, con límites que conviene conocer antes.

Un BaseModel puede usarse como esquema de estado: valida los inputs que llegan al primer nodo. Pero hay matices documentados a jun 2026:

• La salida de invoke() es un dict, no una instancia Pydantic. No esperes recuperar tu modelo tipado al final.
• Hay limitaciones abiertas (issues #6401, #4060, #6675): populate_by_name, genéricos, y model_dump() perdiendo tool_calls.

Por eso el curso usa TypedDict como opción por defecto, y reserva Pydantic para cuando necesitas validación estricta de entrada y aceptas esos límites. Para Elcano, TypedDict heredando MessagesState cubre el caso sin fricción.

2.5Míralo: el estado de Elcano, comentado

Aquí está el esquema de estado de Elcano que arrastrarás durante todo el curso. Léelo entero una vez; luego la anotación. La dificultad no está en ninguna línea suelta, sino en ver por qué cada campo eligió su regla de fusión.

1# estado.py — el estado tipado de Elcano (baseline N0)
2from typing_extensions import Annotated
3from langgraph.graph import MessagesState
4import operator
5
6class EstadoOps(MessagesState):
7    # messages: heredado de MessagesState → Annotated[list[AnyMessage], add_messages]
8    #   append + dedup por id. La conversación: investigación, decisiones, resultados de tools.
9
10    cuenta_id: str
11    #   overwrite. Identifica la cuenta del cliente en este run. Fijo, no acumula.
12
13    plan: list[str]
14    #   overwrite a propósito. Replanificar REEMPLAZA el plan, no lo concatena.
15
16    pasos_hechos: Annotated[list, operator.add]
17    #   append. Registro de pasos completados. Acumula sin pisar (arregla el bug del §2.1).
18
19    accion_propuesta: dict | None
20    #   overwrite. La acción consecuente que el agente quiere ejecutar
21    #   (p. ej. {"tipo": "reembolso", "cargo_id": ...}). En N1 será lo que el
22    #   humano apruebe antes de tocar el dinero. Aquí solo la declaramos.

Tres observaciones:

• El historial viene gratis y correcto. Al heredar MessagesState, messages ya trae add_messages. No reimplementas el reducer.
• plan y pasos_hechos son el par que enseña la lección. Mismo tipo (list), reglas opuestas. Uno es estado actual, el otro es historia.
• accion_propuesta mira a N1. La declaramos ahora para que la acción consecuente viva en el estado, separada de su ejecución. Es la frontera donde N1 colgará el gate humano.

Self-explanation —respóndela antes de seguir: ¿por qué plan usa overwrite y pasos_hechos usa operator.add, si ambos son listas?

2.6Hazlo tú

Ejercicio 1 — andamiaje parcial

Añades a Elcano un campo errores: cada vez que una tool falla, el nodo correspondiente quiere registrar el error sin borrar los anteriores. Te doy la clase a medias. Completa la anotación de errores:

1class EstadoOps(MessagesState):
2    cuenta_id: str
3    plan: list[str]
4    pasos_hechos: Annotated[list, operator.add]
5    errores: ______________________   # quieres ACUMULAR cada error, sin pisar

1    errores: Annotated[list, operator.add]

Ejercicio 2 — autónomo

Diseñas el estado de un agente distinto: un asistente que mantiene un contador de reintentos del paso actual y un conjunto de documentos ya leídos (sin duplicados). Sin mirar la solución, escribe las dos anotaciones y justifica cada una en una frase.

1reintentos_paso: int                    # overwrite: es el valor actual, se reinicia por paso
2docs_leidos: Annotated[set, operator.or_]   # acumula sin duplicados: unión de conjuntos

Elaborative interrogation —antes de leer: ¿por qué add_messages deduplica por id en vez de hacer un append ciego como operator.add?

2.7Comprueba

Sin pistas. Para cada campo, di si debe ser overwrite o llevar un reducer de acumulación, y por qué.

1. usuario_id: el id del cliente del run.
2. historial_mensajes: la conversación completa.
3. plan_actual: el plan que el agente sigue ahora mismo.
4. tools_invocadas: registro de todas las herramientas llamadas durante el run, para auditoría.

2.8Conecta

El bug del §2.1 —el plan que se borraba— no era de lógica. Era una decisión de fusión mal tomada: un campo histórico sin reducer. Ahora sabes pedirle al estado que acumule o que sobrescriba, campo a campo. Eso es la dimensión 1 de la rúbrica C0: estado tipado correcto, con reducers apropiados y sin campos muertos.

Tienes el estado de Elcano. Lo que falta es lo que lo mueve. En L3 construyes el StateGraph: los nodos que leen este estado y devuelven los parciales que acabas de aprender a fusionar, las aristas entre ellos, la transición condicional y el ciclo que termina por condición. El estado tipado es el contrato; los nodos son quienes lo firman.

→ Monta el grafo: nodos, edges y ciclos

2.9Reflexiona

Tómate dos minutos.

• Con tus palabras: ¿qué pregunta te haces sobre un campo para decidir si lleva reducer o no?
• ¿Qué bug introdujiste, o casi, alguna vez por pisar estado que querías acumular? Si nunca te pasó, anota cómo lo cazarías ahora.
• ¿Qué sigue sin estar claro? Si es "¿quién devuelve estos parciales y cuándo se fusionan?", es la pregunta de L3.

Referencia rápida

• Estado: TypedDict (de typing_extensions). Cada nodo devuelve un dict parcial; LangGraph lo fusiona con el estado.
• Overwrite (por defecto): campo sin reducer → la actualización reemplaza. Para valores actuales (id, plan vigente).
• Reducer: campo Annotated[tipo, reducer] → fusiona viejo + nuevo. Para historia que crece.
• operator.add: append en listas. Para registros que acumulan (pasos_hechos, errores).
• add_messages (langgraph.graph.message): reducer de mensajes — append, dedup por id, admite RemoveMessage. MessagesState lo trae prebuilt: messages: Annotated[list[AnyMessage], add_messages].
• Pydantic como estado: valida la entrada al primer nodo, pero invoke() devuelve dict (no instancia); issues abiertos (#6401, #4060, #6675). Default: TypedDict; Pydantic solo si necesitas validación estricta.
• No usar nunca: operator.add para el historial de mensajes (pierde dedup/borrado); un reducer de acumulación en el plan vigente (concatena planes contradictorios).