Estado Tipado con TypedDict y Annotated

Descripción de la cápsula

El diseño del estado es LA decisión más importante que tomas al construir un grafo. Un estado bien diseñado hace que todo lo demás sea fácil — los nodos son simples, los edges son claros, y el debugging es directo. Un estado mal diseñado produce bugs silenciosos, datos que se pierden sin que te des cuenta, y nodos que no pueden comunicarse correctamente.

En la cápsula 02 viste TypedDict por primera vez cuando creaste tu primer StateGraph. Definiste un estado básico con messages y lo usaste para pasar información entre nodos. Ahora vas a ir mucho más profundo: vas a entender exactamente cómo funciona Annotated con reducers, por qué operator.add es crítico para listas de mensajes, qué pasa cuando NO lo usas (spoiler: pierdes datos), y cómo diseñar estados para diferentes tipos de aplicaciones.

Piensa en el estado como el plano arquitectónico de tu grafo. Así como un arquitecto no empieza a construir sin planos, tú no deberías escribir nodos ni edges sin antes diseñar tu estado. Esta cápsula te enseña a pensar como un arquitecto de workflows de AI.

TypedDict: la base del estado

TypedDict es la forma estándar de definir el estado de un grafo en LangGraph. Define un diccionario con claves y tipos específicos:

from typing import TypedDict

class MyState(TypedDict):
    messages: list[str]
    current_step: str
    is_done: bool

¿Por qué TypedDict y no una dataclass? Porque LangGraph opera internamente con diccionarios — cada nodo recibe un dict y retorna un dict. TypedDict te da type hints y autocompletado en el IDE sin cambiar esa naturaleza:

state: MyState = {"messages": ["hola"], "step_count": 0}
print(type(state))       # <class 'dict'>
print(state["messages"])  # ['hola']

El problema: sobrescritura silenciosa

Antes de entender reducers, necesitas ver el bug que resuelven. Este es el error más común en LangGraph y el más difícil de diagnosticar si no sabes qué buscar.

Sin reducers — datos que se pierden

Imagina un grafo con dos nodos. Ambos agregan un mensaje al estado:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class BuggyState(TypedDict):
    messages: list[str]
    final_answer: str

def node_a(state: BuggyState) -> dict:
    return {"messages": ["Hola desde nodo A"]}

def node_b(state: BuggyState) -> dict:
    return {"messages": ["Hola desde nodo B"]}

graph_builder = StateGraph(BuggyState)
graph_builder.add_node("a", node_a)
graph_builder.add_node("b", node_b)
graph_builder.add_edge(START, "a")
graph_builder.add_edge("a", "b")
graph_builder.add_edge("b", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": [], "final_answer": ""})
print(result["messages"])
# Output: ['Hola desde nodo B']
# ¡¡ "Hola desde nodo A" se PERDIÓ !!

El nodo A escribió ["Hola desde nodo A"] en messages. Luego el nodo B escribió ["Hola desde nodo B"]. Sin un reducer, LangGraph simplemente reemplaza el valor — el último nodo gana, y el mensaje del nodo A desaparece.

En una aplicación de chat, esto significaría que cada vez que un nodo procesa un mensaje, todos los mensajes anteriores se borran. Tu chatbot tendría amnesia total.

Annotated y reducers: la solución

Qué es un reducer

Un reducer es una función que le dice a LangGraph cómo combinar el valor que retorna un nodo con el valor que ya existe en el estado. En lugar de reemplazar, el reducer define la lógica de merge.

La sintaxis usa Annotated de Python:

from typing import TypedDict, Annotated
import operator

class MyState(TypedDict):
    messages: Annotated[list[str], operator.add]  # Reducer: acumula
    current_step: str                              # Sin reducer: reemplaza

Annotated[list[str], operator.add] dice: "este campo es una lista de strings, y cuando un nodo retorna un nuevo valor, concatena la lista nueva con la existente en vez de reemplazarla."

operator.add para listas: ACUMULA

Ahora corrijamos el bug del ejemplo anterior:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class FixedState(TypedDict):
    messages: Annotated[list[str], operator.add]  # ← Reducer
    final_answer: str

def node_a(state: FixedState) -> dict:
    return {"messages": ["Hola desde nodo A"]}

def node_b(state: FixedState) -> dict:
    return {"messages": ["Hola desde nodo B"]}

graph_builder = StateGraph(FixedState)
graph_builder.add_node("a", node_a)
graph_builder.add_node("b", node_b)
graph_builder.add_edge(START, "a")
graph_builder.add_edge("a", "b")
graph_builder.add_edge("b", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": [], "final_answer": ""})
print(result["messages"])
# Output: ['Hola desde nodo A', 'Hola desde nodo B']
# ¡Ambos mensajes preservados!

La única diferencia es Annotated[list[str], operator.add]. Ahora:

1. El estado inicial tiene messages: []
2. El nodo A retorna {"messages": ["Hola desde nodo A"]} → LangGraph ejecuta [] + ["Hola desde nodo A"] → ["Hola desde nodo A"]
3. El nodo B retorna {"messages": ["Hola desde nodo B"]} → LangGraph ejecuta ["Hola desde nodo A"] + ["Hola desde nodo B"] → ["Hola desde nodo A", "Hola desde nodo B"]

Sin Annotated para escalares: REEMPLAZA

Los campos sin Annotated usan el comportamiento por defecto: reemplazo. Esto es exactamente lo que quieres para valores que representan "el estado actual" de algo:

from typing import TypedDict, Annotated
import operator

class ProcessState(TypedDict):
    messages: Annotated[list[str], operator.add]  # Acumula
    current_topic: str                             # Reemplaza
    confidence: float                              # Reemplaza
    is_complete: bool                              # Reemplaza

Aquí current_topic, confidence e is_complete se reemplazan cada vez que un nodo los actualiza — que es el comportamiento correcto. Si el nodo de análisis determina que el topic actual es "machine learning" con confianza 0.85, quieres ese valor, no concatenarlo con valores anteriores.

La regla de oro

• ✅ Listas que acumulan datos (mensajes, fuentes, resultados) → Annotated[list[...], operator.add]
• ✅ Escalares que representan estado actual (topic, confidence, step) → Sin Annotated (reemplaza)
• ❌ Listas sin reducer → Bug silencioso, datos se pierden
• ❌ Escalares con operator.add → Error de tipos (no puedes sumar strings con + de esta forma)

Custom reducers

operator.add cubre la mayoría de casos, pero a veces necesitas lógica de merge personalizada. Un custom reducer es cualquier función que recibe dos argumentos (el valor actual y el nuevo) y retorna el valor combinado.

Ejemplo: deduplicar fuentes

from typing import TypedDict, Annotated
import operator

def deduplicate_sources(current: list[str], new: list[str]) -> list[str]:
    seen = set(current)
    result = list(current)
    for source in new:
        if source not in seen:
            result.append(source)
            seen.add(source)
    return result

class ResearchState(TypedDict):
    messages: Annotated[list, operator.add]
    sources: Annotated[list[str], deduplicate_sources]  # Sin duplicados
    current_topic: str

Si el nodo A retorna {"sources": ["arxiv.org/123", "wiki/AI"]} y el nodo B retorna {"sources": ["wiki/AI", "docs.python.org"]}, el resultado será ["arxiv.org/123", "wiki/AI", "docs.python.org"].

Ejemplo: mantener el valor más alto

def keep_highest(current: float, new: float) -> float:
    return max(current, new)

class AnalysisState(TypedDict):
    best_confidence: Annotated[float, keep_highest]
    current_step: str

Cuándo escribir un custom reducer

• ✅ Deduplicar items en una lista
• ✅ Aplicar un límite máximo (ej: mantener solo los últimos N mensajes)
• ✅ Merge con lógica de negocio (ej: priorizar ciertos valores sobre otros)
• ❌ Acumular listas simples → usa operator.add
• ❌ Reemplazar valores → no uses reducer

MessagesState: el atajo para chat

LangGraph incluye un estado prebuilt para aplicaciones de chat: MessagesState. Ya tiene messages configurado con operator.add y usa el tipo de mensajes de LangChain:

from langgraph.graph import MessagesState

# MessagesState es equivalente a:
# class MessagesState(TypedDict):
#     messages: Annotated[list[AnyMessage], operator.add]

Usando MessagesState directamente

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, MessagesState, START, END

model = init_chat_model("openai:gpt-4.1-mini")

def chatbot(state: MessagesState) -> dict:
    response = model.invoke(state["messages"])
    return {"messages": [response]}

graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": [HumanMessage(content="¿Qué es Python?")]})
print(result["messages"][-1].content)
# Output: Python es un lenguaje de programación de alto nivel, interpretado...

Extender MessagesState con campos adicionales

Para la mayoría de aplicaciones reales, necesitas más campos además de mensajes. Puedes extender MessagesState:

from typing import Annotated
import operator
from langgraph.graph import MessagesState

class ChatbotState(MessagesState):
    user_name: str
    conversation_topic: str
    sources: Annotated[list[str], operator.add]

Esto hereda messages con su reducer ya configurado y agrega tus campos custom.

¿Cuándo usar MessagesState vs estado custom?

• ✅ Usa MessagesState cuando tu grafo es primariamente un chatbot o un sistema conversacional
• ✅ Extiende MessagesState cuando necesitas campos adicionales pero el core sigue siendo conversación
• ✅ Usa estado custom cuando tu grafo procesa datos que no son conversacionales (ej: pipeline de datos, ETL)
• ❌ No fuerces MessagesState en grafos que no son de chat — diseña tu propio estado

Diseñar estado para diferentes casos de uso

El estado varía según el tipo de aplicación. Estos tres patrones cubren la mayoría de escenarios:

from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage

# Chat: mensajes + contexto del usuario
class ChatState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]  # Acumula
    user_id: str                                          # Reemplaza
    conversation_summary: str                             # Reemplaza

# Investigación: múltiples fuentes que se acumulan
class ResearchState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]   # Acumula
    sources: Annotated[list[str], operator.add]           # Acumula
    findings: Annotated[list[str], operator.add]          # Acumula
    current_topic: str                                     # Reemplaza
    confidence: float                                      # Reemplaza
    is_complete: bool                                      # Reemplaza

# Pipeline multi-paso: datos que se transforman + errores que se acumulan
class PipelineState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]   # Acumula
    raw_input: str                                         # Reemplaza
    analysis_results: Annotated[list[dict], operator.add] # Acumula
    errors: Annotated[list[str], operator.add]            # Acumula
    current_step: str                                      # Reemplaza

El patrón es consistente: listas que crecen usan operator.add, escalares que representan "lo actual" se reemplazan.

Buenas prácticas para diseñar estado

Qué incluir en el estado

• ✅ Datos que necesitan fluir entre nodos — si un nodo produce algo que otro nodo consume, va en el estado
• ✅ Historial de la conversación — casi siempre messages con operator.add
• ✅ Flags de control de flujo — como is_complete, should_continue, current_step
• ✅ Resultados intermedios — si necesitas acumular datos de múltiples nodos

Qué NO incluir en el estado

• ❌ Configuración estática — API keys, model names, constantes. Usa variables de módulo o config
• ❌ Objetos no serializables — conexiones a bases de datos, file handles, clients HTTP
• ❌ Datos enormes — si un campo va a tener megabytes de datos, reconsidera tu diseño
• ❌ Duplicados — si puedes derivar un valor de otros campos del estado, no lo dupliques

Estado como fuente única de verdad

Un nodo nunca debería depender de variables globales, archivos temporales, o side effects. Todo lo que un nodo necesita para hacer su trabajo debe estar en el estado:

# MAL — depende de variable global
results_cache = []

def search_node(state):
    results = do_search(state["current_topic"])
    results_cache.extend(results)  # Side effect!
    return {"current_step": "analyze"}

# BIEN — todo en el estado
def search_node(state):
    results = do_search(state["current_topic"])
    return {
        "findings": results,
        "current_step": "analyze"
    }

Actualizaciones parciales

Los nodos no necesitan retornar el estado completo. Retornan solo los campos que quieren actualizar:

class MyState(TypedDict):
    messages: Annotated[list, operator.add]
    current_step: str
    confidence: float
    is_complete: bool

def analyze_node(state: MyState) -> dict:
    # Solo actualiza 2 de 4 campos
    return {
        "confidence": 0.92,
        "current_step": "summarize"
    }
    # messages e is_complete no se tocan

LangGraph aplica el update solo a los campos retornados. Los campos no mencionados mantienen su valor actual.

Ejemplo completo: pipeline de análisis

Juntemos todo en un grafo con dos nodos que acumulan hallazgos:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END

class AnalysisState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    findings: Annotated[list[str], operator.add]
    current_step: str
    confidence: float

model = init_chat_model("openai:gpt-4.1-mini")

def extract_topics(state: AnalysisState) -> dict:
    last_message = state["messages"][-1].content
    response = model.invoke(
        f"Extrae los 3 temas principales, separados por comas:\n{last_message}"
    )
    topics = [t.strip() for t in response.content.split(",")]
    return {"findings": [f"Temas: {', '.join(topics)}"], "current_step": "sentiment"}

def analyze_sentiment(state: AnalysisState) -> dict:
    last_message = state["messages"][-1].content
    response = model.invoke(
        f"Sentimiento general (positivo/negativo/neutro):\n{last_message}"
    )
    return {
        "findings": [f"Sentimiento: {response.content.strip()}"],
        "confidence": 0.85, "current_step": "done"
    }

graph_builder = StateGraph(AnalysisState)
graph_builder.add_node("extract", extract_topics)
graph_builder.add_node("sentiment", analyze_sentiment)
graph_builder.add_edge(START, "extract")
graph_builder.add_edge("extract", "sentiment")
graph_builder.add_edge("sentiment", END)
graph = graph_builder.compile()

result = graph.invoke({
    "messages": [HumanMessage(content=(
        "Python sigue dominando en ciencia de datos y AI. "
        "Su ecosistema de librerías lo hace indispensable."
    ))],
    "findings": [], "current_step": "extract", "confidence": 0.0
})

print("Hallazgos acumulados:")
for finding in result["findings"]:
    print(f"  - {finding}")
print(f"Confianza: {result['confidence']}")
# Output:
# Hallazgos acumulados:
#   - Temas: Python, ciencia de datos, AI
#   - Sentimiento: positivo
# Confianza: 0.85

findings acumula de ambos nodos (gracias a operator.add). current_step y confidence se reemplazan con el último valor. Cada nodo solo retorna los campos que actualiza — el resto se mantiene intacto.

Troubleshooting

Problema 1: Lista de mensajes solo muestra el último valor

Síntoma: Tu campo messages solo contiene lo que retornó el último nodo, perdiendo todo lo anterior. Causa: Olvidaste agregar Annotated[..., operator.add] al campo. Solución:

# MAL
class MyState(TypedDict):
    messages: list[str]  # Sin reducer → reemplaza

# BIEN
class MyState(TypedDict):
    messages: Annotated[list[str], operator.add]  # Con reducer → acumula

Problema 2: TypeError al usar operator.add con strings

Síntoma: TypeError: can only concatenate str (not "str") to str o resultados como "hola" + "mundo" = "holamundo". Causa: Usaste operator.add en un campo str en lugar de list[str]. El operador + en strings concatena los caracteres. Solución: operator.add es para listas. Para escalares que quieres reemplazar, no uses reducer:

# MAL
class MyState(TypedDict):
    current_topic: Annotated[str, operator.add]  # "tema1" + "tema2" = "tema1tema2"

# BIEN
class MyState(TypedDict):
    current_topic: str  # Reemplaza: "tema1" → "tema2"

Problema 3: El nodo no puede leer campos del estado

Síntoma: KeyError al acceder a un campo del estado dentro de un nodo. Causa: El campo no fue incluido en el estado inicial al invocar el grafo. Solución: Incluye todos los campos en el input inicial con valores por defecto:

# MAL — falta 'findings' en el input
result = graph.invoke({"messages": [HumanMessage(content="hola")]})

# BIEN — todos los campos presentes
result = graph.invoke({
    "messages": [HumanMessage(content="hola")],
    "findings": [],
    "current_step": "start",
    "confidence": 0.0
})

Problema 4: Custom reducer no se ejecuta

Síntoma: Tu función reducer personalizada nunca se llama; el campo se comporta como si no tuviera reducer. Causa: La función no está correctamente referenciada en Annotated, o el nodo no retorna ese campo. Solución: Verifica que la función es invocable y que el nodo incluye el campo en su retorno:

def my_reducer(current: list, new: list) -> list:
    print(f"Reducer called: {current} + {new}")  # Debug
    return current + new

class MyState(TypedDict):
    items: Annotated[list[str], my_reducer]

def my_node(state: MyState) -> dict:
    return {"items": ["new_item"]}  # Debe incluir 'items' para activar el reducer

Ejercicios

Ejercicio 1: Diagnosticar el bug de sobrescritura (Fácil)

El siguiente grafo tiene un bug: los mensajes del primer nodo se pierden. Identifica el problema y corrígelo.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class ChatState(TypedDict):
    messages: list[str]
    user_name: str

def greet(state):
    return {"messages": [f"Hola, {state['user_name']}!"]}

def ask_question(state):
    return {"messages": ["¿En qué puedo ayudarte?"]}

graph_builder = StateGraph(ChatState)
graph_builder.add_node("greet", greet)
graph_builder.add_node("ask", ask_question)
graph_builder.add_edge(START, "greet")
graph_builder.add_edge("greet", "ask")
graph_builder.add_edge("ask", END)
graph = graph_builder.compile()

result = graph.invoke({"messages": [], "user_name": "Carlos"})
print(result["messages"])
# Actual: ['¿En qué puedo ayudarte?'] — ¡falta el saludo!
# Esperado: ['Hola, Carlos!', '¿En qué puedo ayudarte?']

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class ChatState(TypedDict):
    messages: Annotated[list[str], operator.add]  # ← Agregar reducer
    user_name: str

def greet(state):
    return {"messages": [f"Hola, {state['user_name']}!"]}

def ask_question(state):
    return {"messages": ["¿En qué puedo ayudarte?"]}

graph_builder = StateGraph(ChatState)
graph_builder.add_node("greet", greet)
graph_builder.add_node("ask", ask_question)
graph_builder.add_edge(START, "greet")
graph_builder.add_edge("greet", "ask")
graph_builder.add_edge("ask", END)
graph = graph_builder.compile()

result = graph.invoke({"messages": [], "user_name": "Carlos"})
print(result["messages"])
# Output: ['Hola, Carlos!', '¿En qué puedo ayudarte?']

Ejercicio 2: Estado para un sistema de reseñas (Fácil)

Diseña un TypedDict llamado ReviewState para un grafo que analiza reseñas de productos. El grafo tiene tres nodos: uno que extrae keywords, otro que determina el sentimiento, y otro que genera un resumen. Define qué campos deben tener reducer y cuáles no. No necesitas implementar el grafo — solo el estado.

from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage

class ReviewState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    keywords: Annotated[list[str], operator.add]
    sentiment: str
    sentiment_score: float
    summary: str
    review_text: str

Ejercicio 3: Custom reducer para limitar mensajes (Medio)

Escribe un custom reducer que mantenga solo los últimos 5 mensajes en la lista. Úsalo en un estado y demuestra que funciona creando un grafo donde 7 nodos agregan un mensaje cada uno.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END

def keep_last_five(current: list[str], new: list[str]) -> list[str]:
    """Acumula mensajes pero mantiene solo los últimos 5."""
    combined = current + new
    return combined[-5:]

class LimitedState(TypedDict):
    messages: Annotated[list[str], keep_last_five]
    step: int

def make_node(node_id: int):
    def node_fn(state: LimitedState) -> dict:
        return {
            "messages": [f"Mensaje del nodo {node_id}"],
            "step": node_id
        }
    return node_fn

graph_builder = StateGraph(LimitedState)

for i in range(1, 8):
    graph_builder.add_node(f"node_{i}", make_node(i))

graph_builder.add_edge(START, "node_1")
for i in range(1, 7):
    graph_builder.add_edge(f"node_{i}", f"node_{i+1}")
graph_builder.add_edge("node_7", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": [], "step": 0})

print(f"Total mensajes: {len(result['messages'])}")
for msg in result["messages"]:
    print(f"  - {msg}")
# Output:
# Total mensajes: 5
#   - Mensaje del nodo 3
#   - Mensaje del nodo 4
#   - Mensaje del nodo 5
#   - Mensaje del nodo 6
#   - Mensaje del nodo 7

Ejercicio 4: Migrar a MessagesState (Medio)

Tienes el siguiente estado custom. Refactorízalo para que extienda MessagesState en lugar de definir messages manualmente. Verifica que el comportamiento es idéntico.

from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage

class MyState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    user_name: str
    topic: str
    sources: Annotated[list[str], operator.add]

from dotenv import load_dotenv
load_dotenv()

from typing import Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, MessagesState, START, END

class MyState(MessagesState):
    user_name: str
    topic: str
    sources: Annotated[list[str], operator.add]

model = init_chat_model("openai:gpt-4.1-mini")

def greet(state: MyState) -> dict:
    return {
        "messages": [HumanMessage(content=f"Hola, soy {state['user_name']}")],
        "sources": ["user_input"]
    }

def respond(state: MyState) -> dict:
    response = model.invoke(state["messages"])
    return {
        "messages": [response],
        "topic": "greeting",
        "sources": ["model_response"]
    }

graph_builder = StateGraph(MyState)
graph_builder.add_node("greet", greet)
graph_builder.add_node("respond", respond)
graph_builder.add_edge(START, "greet")
graph_builder.add_edge("greet", "respond")
graph_builder.add_edge("respond", END)

graph = graph_builder.compile()
result = graph.invoke({
    "messages": [],
    "user_name": "Ana",
    "topic": "",
    "sources": []
})

print(f"Mensajes: {len(result['messages'])}")
print(f"Sources: {result['sources']}")
print(f"Topic: {result['topic']}")
# Output:
# Mensajes: 2
# Sources: ['user_input', 'model_response']
# Topic: greeting

Ejercicio 5: Estado complejo con múltiples reducers (Avanzado)

Diseña e implementa un grafo de 3 nodos (search, analyze, summarize) para un pipeline de investigación. El estado debe usar operator.add para sources y findings, un custom reducer para confidence (que mantenga el mayor), y reemplazo normal para summary.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END

def keep_highest(current: float, new: float) -> float:
    return max(current, new)

class ResearchState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    sources: Annotated[list[str], operator.add]
    findings: Annotated[list[str], operator.add]
    confidence: Annotated[float, keep_highest]
    summary: str
    current_step: str

model = init_chat_model("openai:gpt-4.1-mini")

def search_node(state: ResearchState) -> dict:
    topic = state["messages"][-1].content
    response = model.invoke(f"Encuentra 2 datos sobre: {topic}. Sepáralos con '|'.")
    findings = [f.strip() for f in response.content.split("|")]
    return {
        "sources": ["web_search", "knowledge_base"],
        "findings": findings,
        "confidence": 0.6,
        "current_step": "analyze"
    }

def analyze_node(state: ResearchState) -> dict:
    findings_text = "\n".join(state["findings"])
    response = model.invoke(f"Analiza y agrega una observación concisa:\n{findings_text}")
    return {
        "findings": [f"Análisis: {response.content.strip()}"],
        "confidence": 0.82,
        "current_step": "summarize"
    }

def summarize_node(state: ResearchState) -> dict:
    findings_text = "\n".join(state["findings"])
    response = model.invoke(f"Resume en una frase:\n{findings_text}")
    return {"summary": response.content.strip(), "current_step": "done"}

graph_builder = StateGraph(ResearchState)
graph_builder.add_node("search", search_node)
graph_builder.add_node("analyze", analyze_node)
graph_builder.add_node("summarize", summarize_node)
graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "analyze")
graph_builder.add_edge("analyze", "summarize")
graph_builder.add_edge("summarize", END)
graph = graph_builder.compile()

result = graph.invoke({
    "messages": [HumanMessage(content="¿Qué es RAG?")],
    "sources": [], "findings": [], "confidence": 0.0,
    "summary": "", "current_step": "search"
})

print(f"Sources: {result['sources']}")
print(f"Findings: {len(result['findings'])} items")
print(f"Confianza (mayor): {result['confidence']}")
print(f"Resumen: {result['summary'][:100]}...")
# Output (ejemplo):
# Sources: ['web_search', 'knowledge_base']
# Findings: 3 items
# Confianza (mayor): 0.82
# Resumen: RAG combina búsqueda de documentos con generación de texto para...

Ejercicio 6: Reducer con validación (Avanzado)

Crea un reducer que acumule errores pero lance una excepción si supera 3 errores (circuit breaker). Demuéstralo con un grafo que reporta errores progresivamente.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

MAX_ERRORS = 3

def accumulate_with_limit(current: list[str], new: list[str]) -> list[str]:
    combined = current + new
    if len(combined) > MAX_ERRORS:
        raise ValueError(f"Demasiados errores ({len(combined)}). Máximo: {MAX_ERRORS}")
    return combined

class RobustState(TypedDict):
    results: Annotated[list[str], operator.add]
    errors: Annotated[list[str], accumulate_with_limit]

def step_with_error(state: RobustState) -> dict:
    return {"errors": ["Error: timeout en API"]}

def step_with_two_errors(state: RobustState) -> dict:
    return {"errors": ["Error: datos inválidos", "Error: formato incorrecto"]}

graph_builder = StateGraph(RobustState)
graph_builder.add_node("error1", step_with_error)
graph_builder.add_node("error2", step_with_two_errors)
graph_builder.add_node("error3", step_with_error)
graph_builder.add_edge(START, "error1")
graph_builder.add_edge("error1", "error2")
graph_builder.add_edge("error2", "error3")
graph_builder.add_edge("error3", END)
graph = graph_builder.compile()

try:
    result = graph.invoke({"results": [], "errors": []})
except ValueError as e:
    print(f"Grafo detenido: {e}")
# Output: Grafo detenido: Demasiados errores (4). Máximo: 3

Resumen

En esta cápsula aprendiste:

• TypedDict es la base del estado en LangGraph — define qué campos tiene tu grafo con tipos explícitos
• Sin Annotated, los campos se reemplazan (el último nodo que escribe gana)
• Con Annotated[list, operator.add], los campos se acumulan (cada nodo agrega items)
• Custom reducers te dan control total sobre cómo se combinan valores (deduplicar, limitar, validar)
• MessagesState es el atajo prebuilt para aplicaciones de chat (messages con operator.add ya configurado)
• El estado es la fuente única de verdad — los nodos no deben usar variables globales ni side effects
• Los nodos retornan actualizaciones parciales — solo los campos que quieren modificar
• Diseñar bien el estado es diseñar bien el grafo — invierte tiempo aquí antes de escribir nodos

Próxima cápsula: Compilación y Ejecución — cómo compilar tu grafo, ejecutarlo con invoke y stream, y visualizarlo con draw_mermaid_png().

Recursos adicionales

1. State Management — LangGraph Docs — Referencia oficial de cómo funciona el estado en LangGraph
2. Reducers — LangGraph Docs — Documentación detallada de reducers y Annotated
3. MessagesState — LangGraph API — Referencia del estado prebuilt para chat
4. TypedDict — Python Docs — Documentación oficial de TypedDict
5. Annotated — Python Docs — Documentación oficial de Annotated
6. operator — Python Docs — Referencia del módulo operator, incluyendo operator.add
7. How to define graph state — Tutorial paso a paso para definir estado
8. LangGraph Quick Start — Tutorial introductorio oficial con ejemplos de estado

Módulo 5 — LangChain & LangGraph: From Chains to Agents