Ejecución Paralela

Avanzado⏱35 min📖3 min de lecturaLección 8 de 10

Ejecución Paralela

Domina la ejecución paralela de nodos en LangGraph, patrones de fan-out/fan-in, coordinación de ramas, seguridad de hilos y sincronización de resultados paralelos.

LangGraph admite la ejecución paralela de nodos mediante aristas de fan-out. Esta lección cubre cuándo y cómo paralelizar, coordinar ramas paralelas, fusionar resultados y tratar la seguridad de los hilos.

Cómo Funciona la Ejecución Paralela

Cuando un nodo tiene múltiples aristas de salida, todos los nodos de destino se ejecutan simultáneamente en hilos separados:

python

# Fan-out: B and C run in parallel after A
builder.add_edge("A", "B")
builder.add_edge("A", "C")

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

Cada rama recibe una copia del estado en el punto de fan-out. Cuando todas las ramas terminan, sus actualizaciones se fusionan antes de que se ejecute el nodo de fan-in.

Fan-Out/Fan-In Básico

python

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

class ParallelState(TypedDict):
    input_text: str
    results: Annotated[List[str], add]
    merged: str

def splitter(state: ParallelState) -> dict:
    """Fan-out: this node passes state to multiple parallel nodes."""
    return {}  # Passes state through unchanged

def process_a(state: ParallelState) -> dict:
    return {"results": [f"Process A: {state['input_text'].upper()}"]}

def process_b(state: ParallelState) -> dict:
    return {"results": [f"Process B: {state['input_text'].lower()}"]}

def process_c(state: ParallelState) -> dict:
    return {"results": [f"Process C: {state['input_text'][::-1]}"]}

def merger(state: ParallelState) -> dict:
    combined = "\n".join(state["results"])
    return {"merged": combined}

builder = StateGraph(ParallelState)
builder.add_node("splitter", splitter)
builder.add_node("process_a", process_a)
builder.add_node("process_b", process_b)
builder.add_node("process_c", process_c)
builder.add_node("merger", merger)

builder.add_edge(START, "splitter")
builder.add_edge("splitter", "process_a")
builder.add_edge("splitter", "process_b")
builder.add_edge("splitter", "process_c")
builder.add_edge("process_a", "merger")
builder.add_edge("process_b", "merger")
builder.add_edge("process_c", "merger")
builder.add_edge("merger", END)

app = builder.compile()

ℹ️Nota

Los tres nodos de procesamiento (A, B, C) se ejecutan en paralelo. El nodo de fusión espera a que los tres terminen antes de ejecutarse.

Paralelismo Dinámico

Haz fan-out dinámicamente según el número de elementos a procesar:

python

def dispatch(state: State) -> dict:
    items = state["items"]
    # Dynamic dispatch creates parallel work
    return {"item_count": len(items)}

# The splitter node and multiple parallel workers
# process different chunks simultaneously

Para un fan-out verdaderamente dinámico, usa un patrón map:

python

def map_node(state: State) -> dict:
    items = state["items_to_process"]
    results = []

    # Process each item (can parallelize by fan-out)
    for item in items:
        result = process_item(item)
        results.append(result)

    return {"all_results": results}

Seguridad de Hilos

Las ramas paralelas comparten el estado, pero se ejecutan en hilos separados. Cada rama recibe una instantánea del estado en el punto de fan-out:

python

import threading
import time

def parallel_worker(state: State) -> dict:
    thread_id = threading.get_ident()
    print(f"Thread {thread_id} processing: {state.get('shared_value', 'N/A')}")

    # Simulate work
    time.sleep(1)

    return {"worker_result": f"Thread {thread_id} done"}

# Each worker sees the same state snapshot
# Their updates are merged when all complete

📌Importante

Las ramas paralelas no deben depender de los resultados de las demás. Cada rama trabaja a partir de la misma instantánea de estado. Usa nodos de fan-in para coordinar una vez terminado el trabajo paralelo.

Patrones de Reducción (Fan-In)

Agregación Simple

python

def reducer(state: State) -> dict:
    all_results = "\n".join(state.get("results", []))
    return {"final": all_results}

Fusión Ponderada

python

def weighted_merger(state: State) -> dict:
    results = state.get("results", [])
    weights = state.get("weights", {})
    final = ""

    for result in results:
        source = result.split(":")[0]
        weight = weights.get(source, 1.0)
        if weight > 0.5:
            final += f"[High confidence] {result}\n"

    return {"merged": final}

Fusión por Votación

python

def voting_merger(state: State) -> dict:
    results = state.get("results", [])
    from collections import Counter
    answers = Counter(results)
    most_common = answers.most_common(1)[0]
    return {"consensus": most_common[0], "votes": most_common[1]}