Parallel Chain Pattern com LangChain

Quando a ordem não importa

Na Parte 2, o pipeline traduzia o texto, resumia a tradução e analisava o sentimento do resumo. Cada etapa esperava a anterior, e fazia sentido: o sentimento precisava do resumo pronto.

Mas muitas análises não dependem umas das outras. Depois de traduzir, o resumo, o sentimento e as palavras-chave só precisam do texto traduzido. Rodá-las em fila é esperar à toa: cada chamada ao modelo leva segundos, e esses segundos se somam.

O Parallel Chain Pattern agrupa essas etapas independentes e executa todas ao mesmo tempo. O tempo total deixa de ser a soma das etapas e passa a ser o da etapa mais lenta.

Como o padrão funciona

Com quatro chamadas de duração parecida, a diferença fica assim:

1. Identifique as etapas que dependem só da mesma entrada.
2. Agrupe essas etapas num RunnableParallel: todas recebem a mesma entrada.
3. O resultado é um dicionário com a saída de cada etapa, uma chave por etapa.

RunnableParallel na prática

Cada etapa é uma chain comum. O RunnableParallel recebe as chains como argumentos nomeados, e o nome vira a chave do resultado:

from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel

from parallel_chain.prompts import create_keywords_prompt
from sequential_chain.prompts import create_sentiment_prompt, create_summary_prompt


def create_analyses(llm):
    """Three independent analyses that receive the same {"text": ...} input."""
    parser = StrOutputParser()
    return RunnableParallel(
        summary=create_summary_prompt() | llm | parser,
        sentiment=create_sentiment_prompt() | llm | parser,
        keywords=create_keywords_prompt() | llm | parser,
    )


analyses = create_analyses(llm)
analyses.invoke({"text": "Artificial intelligence is transforming..."})
# {"summary": "...", "sentiment": "...", "keywords": "AI, assistants, ..."}

Os prompts de resumo e sentimento são os mesmos da Parte 2; só o de palavras-chave é novo. Como todos recebem {text}, o mesmo dicionário de entrada serve para os três.

Sequência + paralelo

Pipelines reais misturam os dois. A tradução continua sequencial, porque as análises dependem dela; depois dela, as três análises rodam juntas:

from operator import itemgetter
from langchain_core.runnables import RunnablePassthrough


def create_parallel_analysis_chain(llm=None):
    if llm is None:
        llm = init_chat_model(model="gemini-2.5-flash", model_provider="google_genai")

    parser = StrOutputParser()
    translate = (
        {"target_language": itemgetter("language"), "text": itemgetter("text")}
        | create_translation_prompt()
        | llm
        | parser
    )

    return (
        RunnablePassthrough.assign(translated=translate)          # 1 etapa
        | RunnablePassthrough.assign(
            analysis={"text": itemgetter("translated")} | create_analyses(llm)
        )                                                          # 3 etapas juntas
    )

O resultado guarda tudo:

result = chain.invoke({"text": "A inteligência artificial está...", "language": "English"})

result["translated"]               # texto traduzido
result["analysis"]["summary"]      # resumo
result["analysis"]["sentiment"]    # sentimento
result["analysis"]["keywords"]     # palavras-chave

Um detalhe que pouca gente nota: RunnablePassthrough.assign com várias chaves também roda em paralelo, porque por baixo ele cria um RunnableParallel. Esta versão tem o mesmo efeito e deixa o resultado "plano", sem a chave analysis:

to_text = {"text": itemgetter("translated")}

chain = (
    RunnablePassthrough.assign(translated=translate)
    | RunnablePassthrough.assign(
        summary=to_text | create_summary_prompt() | llm | parser,
        sentiment=to_text | create_sentiment_prompt() | llm | parser,
        keywords=to_text | create_keywords_prompt() | llm | parser,
    )
)

Medindo o ganho

Para medir sem gastar chamadas de API, rodei o pipeline com um modelo falso que espera 0,5 s a cada chamada:

O ganho cresce com o número de etapas independentes: com dez análises em vez de três, a fila sequencial levaria 5,5 s e a paralela continuaria em 1,0 s.

O main_parallel.py do repositório imprime o tempo total, então você pode medir com o seu modelo real:

python src/main_parallel.py

Controlando a concorrência

Paralelo demais também é um problema: os provedores limitam quantas requisições você pode fazer por minuto. O parâmetro max_concurrency define quantas chamadas rodam ao mesmo tempo:

result = chain.invoke(
    {"text": texto, "language": "English"},
    config={"max_concurrency": 3},
)

Ele vale também para o batch, onde faz ainda mais diferença: um batch de 50 textos com três análises cada soma 200 chamadas, e sem um limite definido por você muitas delas saem ao mesmo tempo.

results = chain.batch(
    [{"text": t, "language": "English"} for t in textos],
    config={"max_concurrency": 5},
)

Em aplicações assíncronas, como uma API FastAPI, use ainvoke: os ramos rodam em paralelo do mesmo jeito, sem bloquear o servidor.

Cuidados

• Só paralelize o que é independente. Se uma etapa precisa da saída de outra, ela tem que vir depois. Paralelizar o sentimento e o resumo só funciona porque, aqui, o sentimento é calculado sobre o texto traduzido, não sobre o resumo.
• Uma falha derruba o grupo. Se uma das análises lança erro, a chain inteira falha e você perde também as que deram certo. Para etapas não essenciais, trate o erro dentro do ramo (com with_fallbacks, assunto da Parte 11).
• Paralelo não é mais barato. O número de chamadas e de tokens é o mesmo; o que muda é o tempo de espera. O custo continua igual, e o pico de requisições fica maior.
• Respeite o rate limit. Use max_concurrency, principalmente com batch.
• Não conte com a ordem de execução. Os ramos podem terminar em qualquer ordem. O resultado sempre vem com as chaves certas, mas efeitos colaterais (como gravar em banco dentro de um ramo) não têm ordem garantida.

Resumo e referências

• O Parallel Chain executa ao mesmo tempo etapas que dependem só da mesma entrada.
• RunnableParallel, um dicionário dentro da chain ou um assign com várias chaves fazem a mesma coisa.
• Sequência e paralelo se combinam: primeiro o que tem dependência, depois o que é independente.
• max_concurrency controla quantas chamadas rodam juntas; ainvoke mantém o paralelismo em código assíncrono.

Na Parte 6 (Few-Shot Prompting), voltamos ao prompt em si: como usar exemplos para guiar o formato e o estilo das respostas.

Referências

• Repositório pattern_chain (código completo)
• Referência do LangChain: runnables (RunnableParallel, RunnablePassthrough)
• Documentação oficial do LangChain