6. Pipelines Declarativos com LCEL

Nos capítulos anteriores, escrevemos código imperativo para orquestrar interações com LLM: criar um prompt, invocar o modelo, analisar a resposta. Isso funciona, mas à medida que as aplicações de IA se tornam mais complexas, essa abordagem se torna verbosa e mais difícil de manter. Você acaba com chamadas de função profundamente aninhadas, tratamento manual de erros em cada etapa e dificuldade para entender o fluxo geral de dados.

A LangChain Expression Language (LCEL) resolve isso permitindo que você declare o que deseja que aconteça, não como fazer acontecer. Em vez de escrever código procedural que chama funções em sequência, você compõe componentes usando um operador de pipe simples (|) que se lê como um pipeline Unix. O resultado é um código mais limpo e fácil de manter que expressa claramente o fluxo de dados através do seu sistema de IA.

Este capítulo apresenta a LCEL para construir fluxos de trabalho lineares - sequências de operações onde os dados fluem do início ao fim sem ramificações ou loops. Vamos cobrir quando usar LCEL, como compor pipelines e como executá-los tanto de forma síncrona quanto com saída em streaming.

6.1) Por Que LCEL?

O Problema Que a LCEL Resolve

Vamos começar com um exemplo concreto. Suponha que você esteja construindo um assistente de suporte ao cliente que precisa:

1. Receber uma pergunta do usuário
2. Formatá-la em um prompt com instruções do sistema e a mensagem do usuário
3. Enviá-la para um LLM
4. Analisar a resposta para extrair apenas o conteúdo de texto

Aqui está como você poderia escrever isso de forma imperativa (sem LCEL):

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
llm = ChatOpenAI(model="gpt-4o-mini")
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful customer support assistant."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
def answer_question(question: str) -> str:
    # Etapa 1: Formatar o prompt com a pergunta
    messages = prompt.invoke({"question": question})
    
    # Etapa 2: Invocar o LLM
    response = llm.invoke(messages)
    
    # Etapa 3: Analisar a saída para extrair o conteúdo de texto
    result = parser.invoke(response)
    
    return result
 
# Usar
result = answer_question("How do I reset my password?")
print(result)

Saída:

Para redefinir sua senha, siga estas etapas:
1. Vá para a página de login
2. Clique em "Esqueci a Senha"
3. Digite seu endereço de e-mail
4. Verifique seu e-mail para obter um link de redefinição
5. Siga o link e crie uma nova senha

Isso funciona, mas observe os problemas:

Verbosidade: Cada etapa requer atribuição explícita de variável e chamadas de função. A lógica real (formatar → invocar → analisar) está enterrada em código boilerplate.

Estrutura Rígida: Se você quiser adicionar uma etapa (como validar a pergunta ou registrar a resposta), precisa inserir código no meio da função, aumentando a complexidade.

Sem Streaming Integrado: Para transmitir tokens conforme chegam, você precisaria reescrever toda a função para usar llm.stream() e lidar com a iteração assíncrona manualmente.

Fluxo de Dados Pouco Claro: Ao ler o código, não é imediatamente óbvio que isso é um pipeline simples. Você precisa rastrear as atribuições de variáveis para entender o fluxo.

Agora vamos ver a versão LCEL:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
# Definir os componentes (igual ao anterior)
llm = ChatOpenAI(model="gpt-4o-mini")
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful customer support assistant."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
# Compô-los em uma cadeia usando o operador pipe
chain = prompt | llm | parser
 
# Usar
result = chain.invoke({"question": "How do I reset my password?"})
print(result)

Saída:

Para redefinir sua senha, siga estas etapas:
1. Vá para a página de login
2. Clique em "Esqueci a Senha"
3. Digite seu endereço de e-mail
4. Verifique seu e-mail para obter um link de redefinição
5. Siga o link e crie uma nova senha

A saída é idêntica, mas o código é dramaticamente diferente:

Declarativo: chain = prompt | llm | parser expressa todo o fluxo em uma linha. Leia da esquerda para a direita: prompt → LLM → parser.

Componível: Cada componente (prompt, llm, parser) é independente e reutilizável. Você pode trocar componentes sem reescrever o pipeline.

Streaming Integrado: Alterne entre chain.invoke() e chain.stream() sem alterar a definição do pipeline.

Intenção Clara: O operador | torna o fluxo de dados óbvio à primeira vista.

Onde a LCEL se Encaixa: Fluxos de Trabalho Lineares

A LCEL é projetada para fluxos de trabalho lineares - sequências onde os dados fluem em uma direção do início ao fim sem ramificações ou loops.

Esse padrão linear cobre muitas aplicações de IA. Considere um sistema de perguntas e respostas sobre documentos: você recebe uma pergunta → recupera documentos relevantes → formata um prompt → envia para o LLM → analisa a resposta. Cada etapa é uma sequência clara onde a saída de uma etapa se torna a entrada da próxima.

Mas e se seu fluxo de trabalho precisar:

• Fazer o LLM decidir qual ferramenta chamar com base na pergunta
• Chamar uma ferramenta, observar o resultado e então decidir o que fazer a seguir
• Tentar novamente operações que falharam com abordagens diferentes

Esses cenários requerem loops e ramificação condicional - coisas que a LCEL não pode lidar. É por isso que o LangGraph existe (vamos apresentá-lo no Capítulo 15).

Aqui está uma comparação visual das duas abordagens:

LCEL é perfeita para pipelines diretos onde cada etapa processa a saída da etapa anterior. Os dados fluem em apenas uma direção.

LangGraph é para quando você precisa de tomada de decisão e loops. O agente pode agir, observar resultados e pensar novamente.

Para este capítulo, estamos focando na LCEL. Por que dominar pipelines lineares primeiro?

• Fundação: O operador pipe (|) é a sintaxe central do LangChain. Entender isso torna todo o resto mais fácil.
• Pré-requisito para LangGraph: Agentes LangGraph usam cadeias LCEL extensivamente dentro de seus nós
• Padrão do mundo real: Agentes complexos são construídos combinando cadeias LCEL

O restante deste capítulo mostrará como construir esses pipelines lineares com o operador pipe (|).

6.2) Compondo Pipelines com o Operador |

Como os Pipes LCEL Funcionam

O operador pipe (|) da LCEL funciona por causa da interface Runnable.

O Que é um Runnable?

Runnable é a interface padrão do LangChain. Quando um componente implementa a interface Runnable, ele pode ser encadeado com outros componentes usando o operador |.

Todo Runnable fornece estes métodos:

• .invoke(input) - Executar uma vez e obter o resultado completo
• .stream(input) - Executar e receber cada palavra conforme o LLM gera
• .batch(inputs) - Executar várias vezes com entradas diferentes e obter todos os resultados

Para este capítulo, vamos focar em .invoke() e .stream() (vamos cobrir .batch() mais tarde quando necessário).

Por Que o | Funciona?

Porque a classe Runnable implementa o operador | usando sobrecarga de operador do Python. Quando você escreve prompt | llm, isso cria um novo Runnable que executa os dois componentes sequencialmente.

A Maioria dos Componentes LangChain São Runnables

É por isso que você pode encadear tantos componentes diferentes:

• ChatPromptTemplate é um Runnable
• ChatOpenAI é um Runnable
• StrOutputParser é um Runnable
• Até cadeias personalizadas que você cria com | são elas mesmas Runnables!

Isso significa que você pode construir pipelines complexos combinando os mais simples.

Conectando Componentes: Tipos de Entrada/Saída

Ao conectar componentes com |, você precisa garantir que o tipo de saída de um componente corresponda ao tipo de entrada do próximo.

Assinaturas de componentes principais:

Exemplo de fluxo:

chain = prompt | llm | parser

Aqui está como o tipo de dados se transforma conforme passa por cada componente:

• prompt: Recebe dict como entrada e converte para list[BaseMessage]
• llm: Recebe list[BaseMessage] como entrada e converte para AIMessage
• parser: Recebe AIMessage como entrada e converte para str

Vamos ver isso em ação:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()
 
chain = prompt | llm | parser
 
result = chain.invoke({"question": "What is 2+2?"})
print(result)  # "2+2 equals 4."

O que acontece em cada etapa:

O que acontece se os tipos não corresponderem?

Se você tentar conectar componentes incompatíveis, receberá um erro:

# ERRO: Isso não funcionará
chain = llm | prompt  # LLM produz AIMessage, mas prompt requer dict como entrada

A mensagem de erro informará qual tipo de entrada o próximo componente espera versus qual tipo de entrada ele realmente recebeu.

6.3) Executando uma Cadeia: .invoke() e .stream()

Executando um Pipeline

Depois de compor uma cadeia, você a executa usando o método .invoke(). Esta é a maneira síncrona de executar um pipeline - ele aguarda a resposta completa antes de retornar o resultado.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# Invocação síncrona
result = chain.invoke({"question": "What is 2+2?"})
print(result)

Saída:

2+2 é igual a 4.

O método .invoke() é direto: passe o parâmetro de entrada que o primeiro componente espera e obtenha de volta a saída produzida pelo último componente.

Transmitindo Tokens da Mesma Cadeia

Embora .invoke() seja simples, ele tem uma limitação para aplicações voltadas ao usuário: os usuários não veem nada até que a resposta completa esteja pronta. Para respostas longas (10-20 segundos), isso cria uma experiência ruim para o usuário.

Streaming mostra tokens imediatamente conforme são gerados, em vez de esperar pela resposta completa. É o efeito de digitação que você vê no ChatGPT.

Vamos ver o streaming em ação:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# Transmitir chunks conforme chegam
for chunk in chain.stream({"question": "Explain what Python is in one sentence"}):
    print(chunk, end="", flush=True)

Saída (exibida em tempo real, token por token):

Python é uma linguagem de programação versátil e de alto nível, conhecida por sua simplicidade e legibilidade, amplamente usada em desenvolvimento web, ciência de dados e automação.

Observe que a definição da cadeia é idêntica ao exemplo .invoke() acima. Não precisamos reconstruí-la - apenas chamamos .stream() em vez de .invoke().

O método .stream() retorna resultados como chunks enquanto o LLM gera tokens. O print(chunk, end="", flush=True) exibe cada chunk imediatamente na tela, criando o efeito de digitação ao vivo.

Quando Usar .invoke() vs .stream()

Use .invoke() quando:

• Você só precisa do resultado final (análise, tradução, classificação)
• A resposta é curta e o tempo de espera não é um problema
• Você precisa da saída completa antes de prosseguir para a próxima etapa

Use .stream() quando:

• Os usuários precisam ver o progresso (chat, geração de conteúdo)
• A resposta é longa e o tempo de espera seria perceptível
• Você está construindo uma UI onde feedback em tempo real importa

Ambos os métodos funcionam na mesma cadeia. Defina a cadeia uma vez e escolha o modo de execução com base em suas necessidades.

Neste capítulo, você aprendeu LCEL - a sintaxe de pipeline declarativo do LangChain:

• Construir pipelines com o operador pipe: prompt | llm | parser
• Executar com flexibilidade: Use .invoke() para resultados completos ou .stream() para saída em tempo real
• Segurança de tipo: Corresponda tipos de saída aos tipos de entrada ao encadear componentes

A mesma cadeia funciona para ambos os modos de execução - defina uma vez, use em qualquer lugar.

Próximo: O Capítulo 7 cobre saída estruturada com Pydantic, permitindo que você extraia dados JSON validados de respostas LLM.