Cadenas Secuenciales

Aprende patrones de ejecución secuencial en LangGraph, pasando estado entre nodos y usando reductores de estado para gestión avanzada de estado.

Aunque LangGraph soporta topologías de grafo complejas, muchas aplicaciones se construyen sobre cadenas secuenciales simples: una secuencia lineal de nodos donde cada paso construye sobre el anterior.

Ejecución Secuencial en LangGraph

Una cadena secuencial es un pipeline lineal: cada nodo se ejecuta después de que el anterior se completa, pasando estado a lo largo del camino.

python

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

class PipelineState(TypedDict):
    input_text: str
    cleaned: str
    analyzed: str
    result: str

def clean(state: PipelineState) -> dict:
    return {"cleaned": state["input_text"].strip().lower()}

def analyze(state: PipelineState) -> dict:
    analysis = f"Análisis de '{state['cleaned']}': {len(state['cleaned'])} caracteres"
    return {"analyzed": analysis}

def format(state: PipelineState) -> dict:
    return {"result": f"=== RESULTADO ===\n{state['analyzed']}\n=== FIN ==="}

builder = StateGraph(PipelineState)
builder.add_node("clean", clean)
builder.add_node("analyze", analyze)
builder.add_node("format", format)
builder.add_edge(START, "clean")
builder.add_edge("clean", "analyze")
builder.add_edge("analyze", "format")
builder.add_edge("format", END)
app = builder.compile()

ℹ️Nota

Cada nodo solo lee los campos que necesita del estado y escribe los campos que produce. La topología del pipeline garantiza que se ejecuten en el orden correcto.

Pasando Estado Entre Nodos

El estado fluye automáticamente. Cuando el nodo A escribe {"cleaned": valor}, el nodo B puede leer state["cleaned"].

python

def node_a(state: State) -> dict:
    return {"intermediate": "Procesado por A"}

def node_b(state: State) -> dict:
    intermediate = state["intermediate"]
    return {"result": f"B recibió: {intermediate}"}

Reglas de Flujo de Estado

Escenario	Comportamiento
A escribe clave X, B lee X	B ve el valor de A
B escribe clave X después de A	El valor de A se sobrescribe
C escribe clave Y, D no toca Y	D ve el valor de C para Y
Ningún nodo escribe clave Y	Y mantiene su valor inicial

Reductores de Estado

Por defecto, cuando un nodo escribe en una clave de estado, reemplaza el valor. Los reductores cambian este comportamiento.

python

from typing import Annotated
from typing_extensions import TypedDict
from operator import add

class AccumState(TypedDict):
    messages: Annotated[list, add]
    total: int

def step_1(state: AccumState) -> dict:
    return {"messages": ["Paso 1 completo"], "total": 10}

def step_2(state: AccumState) -> dict:
    return {"messages": ["Paso 2 completo"], "total": state["total"] + 20}

# Resultado: messages = ["Paso 1 completo", "Paso 2 completo"]
# Resultado: total = 30

⚠️Advertencia

El reductor add solo funciona con listas. Usa el operador + de Python, así que ambos lados deben ser listas.

Fan-Out y Fan-In

python

def validate(state: State) -> dict:
    return {"validated": True}

def search_web(state: State) -> dict:
    return {"web_results": "..."}

def search_db(state: State) -> dict:
    return {"db_results": "..."}

def merge(state: State) -> dict:
    combined = f"Web: {state['web_results']}\nDB: {state['db_results']}"
    return {"merged": combined}

builder.add_edge(START, "validate")
builder.add_edge("validate", "search_web")
builder.add_edge("validate", "search_db")
builder.add_edge("search_web", "merge")
builder.add_edge("search_db", "merge")
builder.add_edge("merge", END)

✅Éxito

Fan-out con ramas paralelas, luego fan-in para fusionar resultados. Este patrón combina el poder de cadenas secuenciales con procesamiento paralelo.

Comparación: LangChain vs LangGraph

Aspecto	LangChain Chain	LangGraph Secuencial
Definición	`chain = A \| B \| C`	`add_edge(A, B); add_edge(B, C)`
Paso de estado	Cada salida es la siguiente entrada	Objeto de estado compartido
Acceso intermedio	Perdido tras ejecutar la chain	Disponible en el estado
Depuración	Difícil	Fácil (streaming por nodo)
Añadir pasos	Requiere reconstruir la chain	Solo añadir edge
Recuperación de errores	La chain falla completamente	Manejo por nodo