Execução Paralela

Avançado⏱35 min📖3 min de leituraLição 8 de 10

Execução Paralela

Domine a execução paralela de nós em LangGraph, padrões de fan-out/fan-in, coordenação de ramos, segurança de thread e sincronização de resultados paralelos.

LangGraph suporta execução paralela de nós por meio de arestas de fan-out. Esta lição cobre quando e como paralelizar, coordenar ramos paralelos, mesclar resultados e tratar a segurança de threads.

Como Funciona a Execução Paralela

Quando um nó tem múltiplas arestas de saída, todos os nós de destino são executados simultaneamente em threads 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 ramo recebe uma cópia do estado no ponto de fan-out. Quando todos os ramos terminam, as atualizações deles são mescladas antes da execução do nó 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

Os três nós de processamento (A, B, C) rodam em paralelo. O nó de mesclagem espera os três terminarem antes de executar.

Paralelismo Dinâmico

Faça fan-out dinamicamente com base no número de itens a processar:

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 um fan-out verdadeiramente dinâmico, use um padrão 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}

Segurança de Thread

Ramos paralelos compartilham o estado, mas são executados em threads separados. Cada ramo recebe um snapshot do estado no ponto 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

Ramos paralelos não devem depender dos resultados uns dos outros. Cada ramo trabalha a partir do mesmo snapshot de estado. Use nós de fan-in para coordenação após a conclusão do trabalho paralelo.

Padrões de Redução (Fan-In)

Agregação Simples

python

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

Fusão 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}

Fusão por Votação

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]}