Annotated[list[str], operator.add] no es decoración de tipos: es la
instrucción de fusión que LangGraph ejecuta.
Por eso el ejemplo de la guía concatena a mano
(state["pasos"] + ["reverso"]): está emulando un reducer que nunca
declaró. Funciona mientras el grafo sea lineal. Con dos ramas paralelas
escribiendo la misma clave, la concatenación manual pierde una de las dos
— o LangGraph aborta con un error de actualización concurrente.
import operator
from typing import Annotated, TypedDict
class Estado(TypedDict):
pasos: Annotated[list[str], operator.add]
texto: str # sin anotar: se sobrescribe
Ahora cada nodo devuelve solo su aportación:
def nodo_mayusculas(state):
return {"texto": state["texto"].upper(), "pasos": ["mayusculas"]}
LangGraph aplica anterior + nuevo por ti. El nodo deja de necesitar saber
qué había antes, que era justo el acoplamiento que el estado compartido
venía a quitar. Un reducer es cualquier callable
(actual, nuevo) -> fusionado.
from typing import Annotated
from langgraph.graph.message import add_messages
class Estado(TypedDict):
messages: Annotated[list, add_messages]
Hace tres cosas que operator.add no hace:
| Qué hace | Para qué sirve |
|---|---|
Normaliza a objetos Message | Puedes pasar diccionarios sueltos y no rompe |
Asigna id a los que no lo traen | Cada mensaje queda identificado |
Reemplaza si el id ya existe | Permite editar el historial, no solo hacerlo crecer |
MessagesState es exactamente ese estado ya escrito. Si tu grafo es un chat
y no necesitas más claves, úsalo. Si necesitas más, hereda:
class MiEstado(MessagesState): intentos: int.
| Clave | Reducer | Por qué |
|---|---|---|
messages | add_messages | El historial crece y a veces se corrige |
pasos, logs | operator.add | Traza acumulativa |
intentos | ninguno | Es un contador: quieres el último valor, no la suma de todos |
respuesta_final | ninguno | El último nodo manda |
La guía menciona Annotated[list, operator.add] dos veces — en
el troubleshooting de la primera lección y en las notas de la tercera — y
las dos veces lo aplaza a «lecciones avanzadas» que nunca llegan. Mientras
tanto, la cuarta lección usa MessagesState, que es
un reducer, sin decir que lo es.
Antes de dar por buena la explicación de acumulación de mensajes: ¿estás confiando en el reducer, o en que el ejemplo concreto tenía la lista vacía al empezar? Es el mismo agujero de la primera lección, un nivel más arriba.
Coge el grafo de la lección 01, declara
pasos: Annotated[list[str], operator.add] y quita la concatenación
manual de nodo_reverso. Ejecútalo y compara el resultado con el de
antes.
Después, sin cambiar nada más, invócalo dos veces reutilizando el
mismo diccionario de entrada como variable de Python. ¿pasos empieza
vacío la segunda vez? Anota qué te dice eso sobre dónde vive el
estado entre invocaciones: lo vas a necesitar en la lección de memoria.