LangChain LCEL na prática: do chain tradicional à resposta em streaming

No ano passado, peguei um projeto antigo e travei no momento em que abri o código-fonte: um único chain de conversa tinha mais de 200 linhas. Inicializava PromptTemplate, configurava LLMChain, tratava manualmente o mapeamento de entrada e saída e ainda tinha callback próprio para streaming. O pior era que ninguém no time queria mexer naquele trecho. “Se está rodando, deixa” tinha virado consenso.

Esse é um legado das primeiras versões do LangChain. APIs antigas como LLMChain e SequentialChain ainda eram comuns em 2023, mas hoje já foram marcadas como deprecated pela documentação oficial. O problema é que boa parte dos tutoriais na internet continua usando essa escrita antiga.

Este é o artigo 13 da série AI desenvolvimento na prática. Vou comparar código real para mostrar por que o LCEL (LangChain Expression Language) consegue reduzir em 70% o código para a mesma funcionalidade, e como ele resolve automaticamente streaming, execução assíncrona e outras coisas que antes exigiam muito boilerplate.

Aliás, se você está construindo um sistema RAG ou uma aplicação com Agent, este texto conversa bem com RAG na prática: otimização de sistemas e Gerenciamento de estado com LangGraph, porque todos fazem parte do núcleo inevitável do ecossistema LangChain.

Capítulo 1: o que é LCEL e por que usar?

Se você começou a aprender LangChain por tutoriais de 2023, provavelmente escreveu algo parecido com isto:

# Escrita tradicional com LLMChain (deprecated)
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import OpenAI

# 1. Inicializa o modelo
llm = OpenAI(temperature=0.7)

# 2. Define o template do Prompt
template = """Você é um {role} experiente.
Pergunta do usuário: {question}
Responda de forma profissional:"""
prompt = PromptTemplate(
    template=template,
    input_variables=["role", "question"]
)

# 3. Cria o chain
chain = LLMChain(llm=llm, prompt=prompt)

# 4. Chama o chain (repare na forma de passar parâmetros)
result = chain.run(role="engenheiro frontend", question="Como escolher entre React e Vue?")
print(result)

Parece aceitável, certo? Mas, se você precisa adicionar saída em streaming, processamento em lote ou combinação de vários chains, o volume de código cresce rápido. O chain tradicional tem três problemas bem sérios:

Primeiro: suporte fraco a streaming. LLMChain não oferece saída em streaming por padrão. Você precisa escrever callbacks manualmente e escutar eventos de geração de token. O código fica pesado, e o tratamento assíncrono se torna mais propenso a erro.

Segundo: composição pesada. Quer encadear dois chains? Use SequentialChain. Quer executar em paralelo? Use outra API. A cada forma de composição, uma nova interface para aprender.

Terceiro: mapeamento explícito e verboso de entrada e saída. Cada chain precisa declarar input_variables e output_variables, e quando os dados passam de um chain para outro você ainda precisa alinhar os nomes dos campos manualmente.

O LCEL foi criado justamente para atacar esses problemas. Veja como a mesma funcionalidade fica em LCEL:

# Escrita com LCEL (recomendada em LangChain v0.3+)
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

# 1. Define o modelo
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# 2. Define o Prompt
prompt = ChatPromptTemplate.from_template(
    "Você é um {role} experiente.\nPergunta do usuário: {question}\nResponda de forma profissional:"
)

# 3. Conecta os componentes com o operador de pipe
chain = prompt | model

# 4. Chama o chain (o mapeamento de entrada e saída é automático)
result = chain.invoke({"role": "engenheiro frontend", "question": "Como escolher entre React e Vue?"})
print(result.content)

O código cai de 15 para 9 linhas, mas a parte realmente interessante é esta:

• Suporte automático a streaming: troque invoke por stream; o restante do código fica igual
• Suporte automático a assíncrono: use ainvoke ou astream, e a execução assíncrona fica em uma linha
• Suporte automático a lote: use o método batch, passe uma lista e ele executa em paralelo

A inspiração do operador Pipe | vem dos pipelines do Linux. No Linux, cat log.txt | grep error | wc -l encadeia três comandos: a saída de um vira a entrada do próximo. O LCEL leva a mesma ideia para o LangChain: prompt | model | output_parser. Os dados fluem da esquerda para a direita, e o código parece uma frase.

Para ser sincero, na primeira vez que vi essa sintaxe estranhei um pouco. Em Python, | não era operador bitwise OR? Depois entendi que isso usa o açúcar sintático introduzido no Python 3.10, junto com o método mágico __or__, para implementar semântica de pipeline. É um desenho bem esperto.

Capítulo 2: como o operador Pipe funciona

O operador Pipe parece simples, mas há um desenho completo por trás. Comece por este experimento:

from langchain_core.runnables import RunnableLambda

# Cria dois Runnables simples
def add_one(x: int) -> int:
    return x + 1

def multiply_two(x: int) -> int:
    return x * 2

# Envolve funções comuns com RunnableLambda
add_one_runnable = RunnableLambda(add_one)
multiply_two_runnable = RunnableLambda(multiply_two)

# Conecta com o operador de pipe
chain = add_one_runnable | multiply_two_runnable

# Executa
result = chain.invoke(3)  # 3 -> 4 -> 8
print(result)  # Saída: 8

Quando a linha chain = add_one_runnable | multiply_two_runnable é executada, o Python na prática chama add_one_runnable.__or__(multiply_two_runnable).

Railway - Plataforma de deploy moderna, sem ops complexas, no ar em minutos

Começar agora →Anúncio

A classe Runnable do LangChain implementa o método __or__ e retorna um novo objeto RunnableSequence. Esse objeto guarda internamente todos os Runnables encadeados. Quando você chama invoke, ele executa cada componente em ordem e passa a saída do anterior para o próximo.

Runnable é a abstração central do LCEL. Ele define uma interface unificada; qualquer componente que implemente estes quatro métodos pode participar de uma composição por pipeline:

Cada método também tem versões assíncronas correspondentes: astream, abatch e abatch_as_completed.

Essa interface unifica a forma de chamar todos os componentes do LangChain. Seja PromptTemplate, ChatModel, OutputParser ou um RunnableLambda escrito por você, todos podem ser chamados do mesmo jeito.

# Forma unificada de chamada
chain.invoke({"input": "hello"})        # chamada única
chain.stream({"input": "hello"})        # chamada em streaming
chain.batch([{"input": "a"}, {"input": "b"}])  # chamada em lote

# Versões assíncronas
await chain.ainvoke({"input": "hello"})
async for chunk in chain.astream({"input": "hello"}):
    print(chunk, end="", flush=True)

Como os dados fluem dentro do pipeline? Veja um exemplo um pouco mais complexo:

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Três componentes
prompt = ChatPromptTemplate.from_template("Traduza para inglês: {text}")
model = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

# Combinação
chain = prompt | model | parser

# Chamada
result = chain.invoke({"text": "Olá, mundo"})
print(result)  # Saída: Hello World

O fluxo de dados fica assim:

{"text": "Olá, mundo"}
       ↓
   [prompt] → ChatPromptValue(messages=[HumanMessage("Traduza para inglês: Olá, mundo")])
       ↓
   [model]  → AIMessage(content="Hello World")
       ↓
   [parser] → "Hello World" (str)

Cada componente tem um contrato para o tipo de entrada que recebe e o tipo de saída que devolve. Prompt recebe dict e produz ChatPromptValue; Model recebe PromptValue e produz AIMessage; Parser recebe Message e produz str.

Esse contrato de tipos deixa a composição por pipeline mais segura. Se você errar a ordem, por exemplo model | prompt, o código falha em tempo de execução indicando incompatibilidade de tipos. A IDE também consegue encontrar parte desses problemas antes, com type hints.

Capítulo 3: respostas em streaming na prática

Streaming é a funcionalidade do LCEL que mais me surpreendeu.

Imagine que você está construindo um chatbot de atendimento, e o usuário faz uma pergunta complexa: “me ajude a analisar os pontos fortes e fracos deste produto e compare com os concorrentes”. O GPT-4o-mini pode levar algo entre 5 e 8 segundos para gerar a resposta completa.

Sem streaming, o usuário fica olhando para uma tela em branco por 8 segundos. Nesse intervalo, ele pensa: o sistema travou? A rede caiu? Devo atualizar a página? A ansiedade sobe na hora.

Streaming muda essa experiência. Assim que o usuário envia a pergunta, o primeiro caractere aparece na tela, depois as palavras vão surgindo uma a uma, como se alguém estivesse digitando em tempo real. Psicologicamente, a sensação de espera desaparece.

No LangChain tradicional, implementar streaming exigia bastante código:

# Implementação tradicional de streaming (era do LLMChain)
from langchain.chains import LLMChain
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

llm = OpenAI(
    temperature=0.7,
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)
chain = LLMChain(llm=llm, prompt=prompt)
chain.run(role="atendimento", question="...")

Essa solução tem alguns problemas:

1. Callbacks dão trabalho. Se você quiser lógica customizada, como enviar tokens para o frontend, precisa herdar de BaseCallbackHandler e escrever sua própria classe de callback.
2. Alternar entre streaming e não streaming exige mudar código. streaming=True/False é parâmetro de inicialização; não dá para trocar em runtime.
3. Streaming assíncrono é ainda mais complexo. Você precisa combinar com AsyncCallbackHandler, e o volume de código dobra.

O LCEL transforma streaming em capacidade embutida:

# Implementação de streaming com LCEL
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

model = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template(
    "Você é um {role} profissional. Responda à pergunta do usuário: {question}"
)
chain = prompt | model

# Chamada sem streaming
result = chain.invoke({"role": "atendente", "question": "Analise os pontos fortes e fracos deste produto"})
print(result.content)

# Chamada em streaming (só muda o nome do método)
for chunk in chain.stream({"role": "atendente", "question": "Analise os pontos fortes e fracos deste produto"}):
    print(chunk.content, end="", flush=True)

Simples assim. invoke vira stream, e o restante do código não muda.

Veja um exemplo completo de aplicação de chat em tempo real:

import asyncio
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

# 1. Define modelo e Prompt
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
prompt = ChatPromptTemplate.from_messages([
    ("system", "Você é um assistente de IA amigável e bom em responder dúvidas técnicas."),
    ("human", "{input}")
])
parser = StrOutputParser()

# 2. Constrói o chain base
chain = prompt | model | parser

# 3. Adiciona histórico de conversa (memória independente por usuário)
memory = ChatMessageHistory()

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history=lambda session_id: memory,
    input_messages_key="input",
    history_messages_key="chat_history"
)

# 4. Função de conversa em streaming
async def chat_stream(user_input: str):
    """Emite a resposta da conversa em streaming"""
    print("IA: ", end="", flush=True)

    async for chunk in chain_with_history.astream(
        {"input": user_input},
        config={"configurable": {"session_id": "demo"}}
    ):
        print(chunk, end="", flush=True)

    print("\n")  # quebra de linha

# 5. Executa a conversa
async def main():
    print("=== Assistente de IA (demonstração de streaming) ===")

    await chat_stream("O que é LangChain?")
    await chat_stream("Para que ele pode ser usado?")
    await chat_stream("Qual é a relação com LCEL?")

if __name__ == "__main__":
    asyncio.run(main())

Resultado em execução:

=== Assistente de IA (demonstração de streaming) ===
IA: LangChain é um framework open source para criar aplicações baseadas em grandes modelos de linguagem...
IA: Ele pode ser usado para criar chatbots, sistemas RAG, aplicações com Agent e outros fluxos...
IA: LCEL é LangChain Expression Language, um componente central do LangChain...

Cada caractere aparece em tempo real. O usuário não precisa esperar.

Comparei a diferença entre streaming e não streaming na experiência do usuário com dados reais:

No cenário sem streaming, a latência do primeiro token é igual ao tempo total de geração. O usuário espera 8 segundos para ver qualquer feedback. Com streaming, a latência do primeiro token é apenas o tempo de gerar o primeiro token, geralmente abaixo de 1 segundo.

Como o streaming do LCEL funciona? O ponto-chave é que o método stream de Runnable chama recursivamente o stream de cada componente do pipeline. Para componentes de modelo, ele chama diretamente a API em streaming da OpenAI; para Prompt e Parser, que normalmente não precisam de streaming, ele retorna o resultado completo. O comportamento de streaming do pipeline inteiro é coordenado automaticamente pelos componentes.

Vultr - Servidor cloud NVMe de alto desempenho, 32 locais globais, deploy Docker com um clique

Ver preços →Anúncio

Isso significa que você não precisa se preocupar com qual componente suporta streaming e qual não suporta. O LCEL cuida disso. Se um componente não oferece streaming, ele é tratado dentro do pipeline como um “retorno único”, sem prejudicar a saída em streaming do conjunto.

Capítulo 4: componentes Runnable em detalhes

O operador Pipe resolve o encadeamento de componentes, mas projetos reais têm vários cenários mais complexos: executar múltiplos ramos em paralelo, passar resultados intermediários, transformar dados de forma customizada. O LangChain oferece um conjunto de componentes Runnable para lidar com essas necessidades.

RunnableParallel: execução paralela

Em sistemas RAG, uma necessidade comum é consultar várias fontes de dados ao mesmo tempo: banco de dados vetorial, busca por palavras-chave e grafo de conhecimento. Com RunnableParallel, essas buscas podem rodar em paralelo:

from langchain_core.runnables import RunnableParallel

# Define três recuperadores (exemplo simulado com RunnableLambda)
def vector_search(query: str) -> str:
    return f"Resultado da busca vetorial: 3 documentos relacionados a {query}"

def keyword_search(query: str) -> str:
    return f"Resultado da busca por palavra-chave: 5 registros correspondem a {query}"

def graph_search(query: str) -> str:
    return f"Resultado da busca no grafo: 2 entidades relacionadas a {query}"

# Cria o chain de busca paralela
retrievers = RunnableParallel(
    vector=RunnableLambda(vector_search),
    keyword=RunnableLambda(keyword_search),
    graph=RunnableLambda(graph_search)
)

# Executa (as três buscas rodam ao mesmo tempo)
results = retrievers.invoke("LangChain LCEL")
print(results)
# Saída: {
#   'vector': 'Resultado da busca vetorial: 3 documentos relacionados a LangChain LCEL',
#   'keyword': 'Resultado da busca por palavra-chave: 5 registros correspondem a LangChain LCEL',
#   'graph': 'Resultado da busca no grafo: 2 entidades relacionadas a LangChain LCEL'
# }

RunnableParallel retorna um dicionário: as chaves são os nomes definidos na construção, e os valores são os resultados de cada ramo. Esses resultados podem ser passados para componentes posteriores para combinação e processamento.

RunnablePassthrough: passar a entrada adiante

Às vezes você precisa preservar a entrada original dentro do pipeline para usá-la depois. Em um sistema RAG, por exemplo, o recuperador precisa da consulta original, enquanto o gerador precisa dos documentos recuperados + a consulta original:

from langchain_core.runnables import RunnablePassthrough

# Recuperador simulado
def retrieve(query: dict) -> str:
    return "Conteúdo dos documentos recuperados..."

# Constrói o chain: preserva a query original e faz a recuperação
chain = RunnableParallel(
    retrieved_docs=RunnableLambda(retrieve),
    original_query=RunnablePassthrough()
)

result = chain.invoke({"query": "O que é LCEL?"})
print(result)
# Saída: {
#   'retrieved_docs': 'Conteúdo dos documentos recuperados...',
#   'original_query': {'query': 'O que é LCEL?'}
# }

RunnablePassthrough não faz nada: só repassa a entrada como recebeu. Parece pouco útil, mas em pipelines complexos é essencial.

RunnableLambda: transformação com função customizada

O LangChain oferece muitos componentes prontos, mas sempre aparece um caso que precisa de lógica própria. RunnableLambda envolve uma função Python comum como Runnable, permitindo que ela participe do pipeline:

from langchain_core.runnables import RunnableLambda

# Define uma função de formatação
def format_output(result: dict) -> str:
    """Formata os resultados recuperados como entrada do Prompt"""
    docs = result["retrieved_docs"]
    query = result["original_query"]["query"]
    return f"Material de referência: {docs}\nPergunta do usuário: {query}\nResponda com base no material:"

# Uso
chain = RunnableParallel(
    retrieved_docs=RunnableLambda(retrieve),
    original_query=RunnablePassthrough()
) | RunnableLambda(format_output)

formatted = chain.invoke({"query": "O que é LCEL?"})
print(formatted)
# Saída: Material de referência: Conteúdo dos documentos recuperados...
#        Pergunta do usuário: O que é LCEL?
#        Responda com base no material:

A flexibilidade do RunnableLambda faz dele uma peça coringa dentro do pipeline. Qualquer função Python pode ser empacotada e entrar no fluxo.

Implementação completa de um pipeline RAG

Juntando esses componentes, um pipeline RAG completo fica assim:

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableParallel, RunnableLambda
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# 1. Inicializa o banco vetorial (exemplo)
embeddings = OpenAIEmbeddings()
# Em um projeto real, aqui você carregaria vetores de documentos reais
vectorstore = FAISS.from_texts(
    ["LCEL é a linguagem de expressão do LangChain",
     "O operador Pipe serve para encadear componentes",
     "Runnable é a abstração central do LCEL"],
    embeddings
)
retriever = vectorstore.as_retriever()

# 2. Define o Prompt
rag_prompt = ChatPromptTemplate.from_template(
    """Responda à pergunta do usuário com base no material de referência abaixo.

Material de referência:
{context}

Pergunta do usuário: {question}

Dê uma resposta precisa e detalhada:"""
)

# 3. Define a função de formatação (converte resultados recuperados em string)
def format_docs(docs) -> str:
    return "\n".join(doc.page_content for doc in docs)

# 4. Constrói o chain RAG completo
rag_chain = (
    # Executa em paralelo: recuperação + passagem da pergunta original
    RunnableParallel(
        context=retriever | RunnableLambda(format_docs),
        question=RunnablePassthrough()
    )
    # Monta o Prompt
    | rag_prompt
    # Chama o modelo
    | ChatOpenAI(model="gpt-4o-mini")
    # Analisa a saída
    | StrOutputParser()
)

# 5. Uso
answer = rag_chain.invoke("O que é LCEL?")
print(answer)
# Saída: LCEL é LangChain Expression Language, a linguagem de expressão do LangChain...

A estrutura desse chain RAG pode ser desenhada como um fluxograma:

{"question": "O que é LCEL?"}
              ↓
    ┌─────────┴─────────┐
    ↓                   ↓
[retriever]        [Passthrough]
    ↓                   ↓
format_docs        question
    ↓                   ↓
    └─────────┬─────────┘
              ↓
        {"context": "...", "question": "..."}
              ↓
         [rag_prompt]
              ↓
           [model]
              ↓
          [parser]
              ↓
        "Conteúdo da resposta..."

Essa implementação RAG conversa diretamente com RAG na prática: otimização de sistemas, outro texto da série. Se você estiver lendo aquele artigo, vai perceber que muitas técnicas, como reordenação de resultados e recuperação por múltiplas rotas, encaixam diretamente nesta estrutura LCEL.

Capítulo 5: migração prática a partir de chains antigos

Se o seu projeto ainda usa LLMChain, migrar para LCEL não é tão difícil. No ano passado, ajudei um projeto de e-commerce a fazer essa migração, e o módulo inteiro do chatbot de atendimento levou dois dias. Aqui estão alguns padrões comuns de migração.

LLMChain → sintaxe Pipe

Esta é a migração mais básica. O núcleo do LLMChain é Prompt + Model; depois da migração, basta conectar os dois com o pipe:

# Código antigo (LLMChain)
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import OpenAI

llm = OpenAI(temperature=0.7)
prompt = PromptTemplate(
    template="Pergunta do usuário: {question}\nResponda:",
    input_variables=["question"]
)
chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run(question="...")

# Código novo (LCEL)
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

model = ChatOpenAI(temperature=0.7)
prompt = ChatPromptTemplate.from_template("Pergunta do usuário: {question}\nResponda:")
chain = prompt | model
result = chain.invoke({"question": "..."})

Alguns pontos de atenção:

1. A classe do modelo mudou. O código antigo usa OpenAI (Completion API); no novo código, prefira ChatOpenAI (Chat API). A Chat API é a direção principal da OpenAI, enquanto a Completion API vem ficando cada vez mais periférica.
2. A classe do Prompt mudou. PromptTemplate ainda funciona, mas ChatPromptTemplate oferece formatos mais ricos, como system message e diálogo com múltiplos papéis.
3. O método de chamada mudou. chain.run() vira chain.invoke(), e o retorno deixa de ser string para ser um objeto Message. Você precisa acessar .content para obter o texto.

SequentialChain → RunnableParallel

No código antigo, SequentialChain encadeava várias etapas. Depois da migração, você pode usar pipes diretamente:

# Código antigo (SequentialChain)
from langchain.chains import SequentialChain, LLMChain

# Primeira etapa: gerar título
title_chain = LLMChain(
    llm=llm, prompt=title_prompt,
    output_key="title"
)

# Segunda etapa: gerar corpo do texto
content_chain = LLMChain(
    llm=llm, prompt=content_prompt,
    output_key="content"
)

# Encadeamento
full_chain = SequentialChain(
    chains=[title_chain, content_chain],
    input_variables=["topic"],
    output_variables=["title", "content"]
)
result = full_chain({"topic": "desenvolvimento de IA"})
print(result["title"], result["content"])

# Código novo (LCEL)
from langchain_core.runnables import RunnableParallel

# Define dois ramos
title_chain = title_prompt | model
content_chain = content_prompt | model

# Executa em paralelo (se quiser serial, conecte diretamente com |)
full_chain = RunnableParallel(
    title=title_chain,
    content=content_chain
)
result = full_chain.invoke({"topic": "desenvolvimento de IA"})
print(result["title"].content, result["content"].content)

SequentialChain executa em série por padrão: cada chain espera o anterior terminar. RunnableParallel, no LCEL, executa em paralelo e tende a ser mais rápido. Se você realmente precisa de execução serial, por exemplo quando a segunda etapa depende da saída da primeira, use o pipe:

# Serial: a saída da primeira etapa passa para a segunda
chain = (
    title_prompt | model | StrOutputParser()
    | (lambda title: {"topic": topic, "title": title})  # passa o resultado intermediário
    | content_prompt | model
)

TransformChain → RunnableLambda

TransformChain serve para inserir lógica customizada dentro do chain. A migração correspondente usa RunnableLambda:

# Código antigo (TransformChain)
from langchain.chains import TransformChain

def transform_func(inputs: dict) -> dict:
    text = inputs["text"]
    processed = text.upper()  # algum tipo de processamento
    return {"processed_text": processed}

transform_chain = TransformChain(
    input_variables=["text"],
    output_variables=["processed_text"],
    transform=transform_func
)

# Código novo (RunnableLambda)
from langchain_core.runnables import RunnableLambda

def transform_func(inputs: dict) -> dict:
    text = inputs["text"]
    processed = text.upper()
    return {"processed_text": processed}

transform_chain = RunnableLambda(transform_func)

RunnableLambda é mais flexível: não exige declarar input_variables e output_variables explicitamente, e entra direto no pipeline.

Armadilhas comuns na migração

Na migração, já caí em algumas armadilhas:

Armadilha 1: o tipo de retorno mudou

KiloClaw - OpenClaw gerenciado para empresas: orquestração de agentes IA e roteamento inteligente

Começar →Anúncio

run() no LLMChain retorna string; invoke() no LCEL retorna um objeto Message.

# Antigo: recebe a string diretamente
result = chain.run(...)  # str

# Novo: precisa acessar content
result = chain.invoke(...)  # AIMessage
text = result.content  # str

A solução é adicionar StrOutputParser no fim do pipeline, para converter automaticamente Message em string.

chain = prompt | model | StrOutputParser()
result = chain.invoke(...)  # retorna str diretamente

Armadilha 2: migração de componentes de memória

No código antigo, ConversationChain já trazia memória:

# Código antigo
from langchain.chains import ConversationChain

chain = ConversationChain(llm=llm, memory=memory)

No código novo, use RunnableWithMessageHistory:

from langchain_core.runnables import RunnableWithMessageHistory

chain = prompt | model
chain_with_memory = RunnableWithMessageHistory(
    chain,
    get_session_history=get_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)

São vários parâmetros, e você precisa declarar explicitamente o nome do campo de entrada e o nome do campo de histórico. Para um exemplo completo de sistema de conversa, veja Chamada de ferramentas com Agent na prática, outro artigo da série.

Armadilha 3: os caminhos de importação mudaram no LangChain v0.3

Muitos componentes saíram de langchain e foram para langchain_core ou langchain_community:

# Import antigo
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

# Import novo
from langchain_core.runnables import RunnableLambda, RunnableParallel
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_message_histories import ChatMessageHistory

A IDE normalmente aponta o erro de importação; basta seguir a sugestão e ajustar.

Caso de migração em produção

O chatbot de atendimento do e-commerce que migrei no ano passado tinha cerca de 300 linhas no código original e usava LLMChain + SequentialChain + TransformChain. A lógica principal depois da migração ficou assim:

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda

# Inicialização
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)

# Prompt de classificação de intenção
intent_prompt = ChatPromptTemplate.from_template(
    """Analise a intenção do usuário e retorne uma das categorias abaixo:
    - product_query (consulta sobre produto)
    - order_status (consulta de pedido)
    - complaint (reclamação ou sugestão)
    - other (outros)

    Mensagem do usuário: {message}
    Categoria de intenção:"""
)

# Prompts de tratamento para cada intenção
product_prompt = ChatPromptTemplate.from_template(
    "O usuário perguntou sobre um produto: {message}\nBusque no catálogo e responda:"
)
order_prompt = ChatPromptTemplate.from_template(
    "O usuário quer consultar um pedido: {message}\nVerifique o status do pedido e responda:"
)

# Constrói a lógica de roteamento
def route_by_intent(result):
    intent = result.content.strip().lower()
    if "product" in intent:
        return "product"
    elif "order" in intent:
        return "order"
    else:
        return "default"

# Chain completo
intent_chain = intent_prompt | model | StrOutputParser() | RunnableLambda(route_by_intent)

# Roteamento por ramo (pseudocódigo; na prática, use RunnableBranch)
full_chain = (
    {"message": RunnablePassthrough()}
    | RunnableParallel(
        intent=intent_chain,
        original=RunnablePassthrough()
    )
    # Escolhe diferentes ramos de processamento com base em intent
    # ... o código real usa RunnableBranch
)

# Saída em streaming
async for chunk in full_chain.astream({"message": "Quero consultar o pedido 12345"}):
    print(chunk, end="", flush=True)

Depois da migração, o código ficou com 150 linhas, uma redução pela metade. Mais importante: streaming e processamento assíncrono, que antes exigiam desenvolvimento extra, agora ficam resolvidos com uma linha.

Resumo

LCEL é a arquitetura recomendada oficialmente no LangChain v0.3+. Ele simplifica o código com o operador Pipe, unifica a forma de chamada com a interface Runnable e melhora a experiência do usuário com suporte nativo a streaming.

O maior desafio da migração não é converter sintaxe, e sim mudar o modo de pensar. Chains tradicionais enfatizam “declaração explícita”: cada chain precisa deixar claros os campos de entrada e saída. O LCEL enfatiza “fluxo implícito”: os dados passam automaticamente pelo pipeline, e os contratos de tipo ficam dentro dos componentes.

Se você tem um projeto antigo ainda usando LLMChain, vale migrar em lotes: primeiro os chains simples de conversa, depois as composições mais complexas. Durante a migração, depurar com LangSmith ajuda a encontrar rapidamente problemas como incompatibilidade de tipos.

O próximo passo pode ser ler Gerenciamento de estado com LangGraph na prática. LangGraph é a nova geração de framework para Agent lançada pela equipe do LangChain; usado com LCEL, ele permite construir aplicações de Agent mais complexas. Use LCEL para tarefas simples em chain e LangGraph para gerenciamento de estado complexo. Essa é uma combinação já bastante madura.

Navegação da série AI desenvolvimento na prática

• Artigo 1: Claude API para iniciantes: da autenticação ao diálogo multi-turno
• Artigo 2: Prompt Engineering avançado na prática
• Artigo 8: RAG na prática: equilíbrio entre precisão da recuperação e qualidade da geração
• Artigo 11: Gerenciamento de estado com LangGraph: melhores práticas para arquitetura de Agent em 2026
• Artigo 13: LangChain LCEL na prática: do chain tradicional à resposta em streaming (este artigo)
• Artigo 14: Chamada de ferramentas com Agent: fazendo a IA chamar APIs e serviços externos