Sequential Chain Pattern na prática
Como construir um pipeline em etapas, em que a saída de uma chain alimenta a próxima, ideal para fluxos com tradução, resumo e transformação.
Quando um prompt só não basta
Imagine que você recebe textos em português e precisa entregar, em inglês, um resumo e a análise de sentimento. Dá para pedir tudo num prompt só, mas o resultado costuma piorar: o modelo mistura as tarefas, pula etapas e fica difícil saber onde algo deu errado.
O Sequential Chain Pattern divide o trabalho em etapas pequenas, cada uma com um prompt especializado, e liga as etapas em sequência. Cada etapa faz uma coisa só, e faz bem.
Pré-requisito: este tutorial usa o ChatPromptTemplate e o operador | do LangChain. Se ainda não conhece, comece pelo Prompt Template Pattern com LangChain.
Como o padrão funciona
A saída de cada etapa vira a entrada da próxima, e o resultado final reúne o que cada etapa produziu:
- A saída da primeira operação alimenta a segunda.
- Cada etapa tem o seu próprio prompt e a sua própria lógica.
- O resultado final agrega todas as transformações.
Um prompt por etapa
O primeiro passo é escrever um template para cada etapa. No repositório pattern_chain, eles ficam juntos em prompts.py:
from langchain_core.prompts import ChatPromptTemplate
def create_translation_prompt() -> ChatPromptTemplate:
"""Traduz o texto para o idioma de destino."""
system = ("system", "Translate the following text to {target_language}")
user = ("user", "{text}")
return ChatPromptTemplate.from_messages([system, user])
def create_summary_prompt() -> ChatPromptTemplate:
"""Resume o texto de forma concisa."""
system = ("system", "Provide a concise summary of the following text")
user = ("user", "{text}")
return ChatPromptTemplate.from_messages([system, user])
def create_sentiment_prompt() -> ChatPromptTemplate:
"""Analisa o sentimento do texto e explica brevemente."""
system = ("system", "Analyze the sentiment of the following text and provide a brief explanation")
user = ("user", "{text}")
return ChatPromptTemplate.from_messages([system, user])
Repare que todos recebem {text}. Essa interface em comum é o que permite ligar uma etapa na outra sem adaptação.
Encadeando as etapas
Cada prompt vira uma chain pequena (prompt | llm | parser). Uma função Python chama as chains em ordem, passando a saída de uma para a próxima:
from langchain_core.output_parsers import StrOutputParser
from langchain.chat_models import init_chat_model
from sequential_chain.prompts import (
create_translation_prompt,
create_summary_prompt,
create_sentiment_prompt,
)
def create_analysis_chain():
"""Pipeline: traduzir -> resumir -> analisar sentimento."""
llm = init_chat_model(model="gemini-2.5-flash", model_provider="google_genai")
parser = StrOutputParser()
translate_chain = create_translation_prompt() | llm | parser
summary_chain = create_summary_prompt() | llm | parser
sentiment_chain = create_sentiment_prompt() | llm | parser
def combined_chain(input_dict):
translated = translate_chain.invoke({
"target_language": input_dict.get("language", "English"),
"text": input_dict["text"],
})
summary = summary_chain.invoke({"text": translated})
sentiment = sentiment_chain.invoke({"text": summary})
return {
"original_text": input_dict["text"],
"translated": translated,
"summary": summary,
"sentiment": sentiment,
}
return combined_chain
O ponto de entrada só monta a chain e mostra o resultado de cada etapa:
from dotenv import load_dotenv
load_dotenv()
from sequential_chain import create_analysis_chain
def main():
chain = create_analysis_chain()
result = chain({
"text": "A inteligência artificial está transformando o mundo moderno "
"de maneiras sem precedentes.",
"language": "English",
})
print("=== Translated ===")
print(result["translated"])
print("\n=== Summary ===")
print(result["summary"])
print("\n=== Sentiment ===")
print(result["sentiment"])
if __name__ == "__main__":
main()
poetry install
python src/main_sequential.py
Essa versão é simples de ler e de depurar: é Python comum, com um passo por linha. O preço é que a função não é uma chain LCEL, então não ganha batch, stream nem o rastreamento automático de ferramentas como o LangSmith.
Versão com LCEL
O mesmo pipeline pode ser escrito só com LCEL. O RunnablePassthrough.assign mantém o dicionário de entrada e acrescenta uma nova chave com a saída da etapa, então cada etapa enxerga tudo o que veio antes:
from operator import itemgetter
from langchain_core.runnables import RunnablePassthrough
analysis_chain = (
RunnablePassthrough.assign(
translated={"target_language": itemgetter("language"), "text": itemgetter("text")}
| create_translation_prompt() | llm | parser
)
| RunnablePassthrough.assign(
summary={"text": itemgetter("translated")} | create_summary_prompt() | llm | parser
)
| RunnablePassthrough.assign(
sentiment={"text": itemgetter("summary")} | create_sentiment_prompt() | llm | parser
)
)
result = analysis_chain.invoke({"text": "Texto em português...", "language": "English"})
# {'text': ..., 'language': ..., 'translated': ..., 'summary': ..., 'sentiment': ...}
O dicionário entre chaves ({"text": itemgetter("translated")}) monta a entrada que o próximo prompt espera, pegando o valor certo do resultado acumulado.
| Função Python | LCEL | |
|---|---|---|
| Leitura | Mais simples para quem está começando | Exige conhecer assign e itemgetter |
| Depuração | Breakpoint e print em qualquer linha | Rastreamento por etapa em ferramentas como o LangSmith |
| batch / stream / async | Precisa implementar | Vem pronto |
| Composição | Não entra em outra chain com | | Vira uma etapa de chains maiores |
Inspecionando cada etapa
Uma das maiores vantagens do padrão é poder olhar a saída intermediária. Se o sentimento saiu estranho, você descobre na hora se o problema veio da tradução, do resumo ou da análise, porque as três saídas estão no resultado.
- Devolva as saídas intermediárias, como as duas versões acima fazem, em vez de só o resultado final.
- Teste cada etapa isolada:
summary_chain.invoke({"text": "..."})roda só o resumo, sem gastar chamadas nas outras etapas. - Registre entradas e saídas de cada etapa nos logs quando o pipeline estiver em produção.
Boas práticas e cuidados
| Benefício | O que você ganha |
|---|---|
| Pipeline complexo | Várias operações numa única execução |
| Reutilização | Cada etapa funciona sozinha em outros fluxos |
| Depuração | Saída de cada etapa visível |
| Flexibilidade | Adicionar ou remover etapas sem reescrever o resto |
Cuidados
- Cada etapa é uma chamada ao modelo. Três etapas custam mais tempo e mais tokens do que uma. Junte etapas que não ganham nada separadas.
- Erros se propagam. Uma tradução ruim contamina o resumo e o sentimento. Valide as saídas que alimentam etapas críticas.
- Mantenha interfaces simples. Prompts que recebem as mesmas variáveis (como
{text}) se ligam sem adaptação. - Escolha o modelo por etapa. Etapas simples podem usar um modelo mais rápido e barato; só a etapa difícil precisa do mais capaz.
- Etapas independentes não precisam ser sequenciais. Se duas etapas usam a mesma entrada e não dependem uma da outra, rode-as em paralelo.
Resumo e referências
- O Sequential Chain divide uma tarefa grande em etapas pequenas, cada uma com seu prompt.
- A saída de cada etapa vira a entrada da próxima, e o resultado guarda todas as saídas.
- Uma função Python é o jeito mais simples de começar; LCEL com
RunnablePassthrough.assigndábatch,streame composição. - Mais etapas significam mais custo e mais pontos de falha: separe só o que ganha com a separação.