LangChain LCEL en práctica: del encadenamiento tradicional a la respuesta en streaming — un paradigma moderno

El año pasado heredé un proyecto antiguo. Al abrir el código, me quedé helado: una sola cadena de diálogo tenía más de 200 líneas. Inicializar PromptTemplate, configurar LLMChain, mapear manualmente entradas y salidas, escribir callbacks para el streaming. Peor aún: nadie en el equipo se atrevía a tocarlo — «mientras funcione» era el consenso.

Es el legado de las primeras versiones de LangChain. LLMChain, SequentialChain y compañía eran la norma en 2023; hoy el equipo oficial los ha marcado como obsoletos. El problema: la mayoría de tutoriales en línea siguen usando la sintaxis antigua.

Este artículo es el número 13 de la serie Desarrollo de IA en la práctica. Con comparaciones de código reales, te muestro por qué LCEL (LangChain Expression Language) reduce el código un 70 % para la misma funcionalidad y cómo gestiona automáticamente streaming, ejecución asíncrona y procesamiento por lotes — antes montañas de código repetitivo.

Si estás construyendo un sistema RAG o una aplicación Agent, este artículo encaja con Optimización de sistemas RAG en la práctica y Gestión de estado con LangGraph — pilares ineludibles del ecosistema LangChain.

Capítulo 1: ¿Qué es LCEL? ¿Por qué usarlo?

Si aprendiste LangChain con tutoriales de 2023, probablemente escribiste algo así:

# Enfoque LLMChain tradicional (obsoleto)
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import OpenAI

# 1. Inicializar el modelo
llm = OpenAI(temperature=0.7)

# 2. Definir la plantilla de prompt
template = """Eres un(a) {role} con experiencia.
Pregunta del usuario: {question}
Respuesta profesional:"""
prompt = PromptTemplate(
    template=template,
    input_variables=["role", "question"]
)

# 3. Crear la cadena
chain = LLMChain(llm=llm, prompt=prompt)

# 4. Invocar la cadena (fíjate en cómo se pasan los parámetros)
result = chain.run(role="ingeniero frontend", question="¿React o Vue?")
print(result)

¿Parece razonable? Añade streaming, procesamiento por lotes o varias cadenas en serie y el código explota. Las cadenas tradicionales tienen tres problemas graves:

Primero: streaming deficiente. LLMChain no hace streaming por defecto. Hay que escribir callbacks para escuchar cada token — código pesado y async frágil.

Segundo: composición torpe. ¿Dos cadenas en serie? SequentialChain. ¿En paralelo? Otra API. Cada modo de composición implica una interfaz nueva que aprender.

Tercero: mapeo entrada/salida explícito y tedioso. Cada cadena declara input_variables y output_variables; hay que alinear manualmente los nombres entre pasos.

LCEL apunta exactamente a esos problemas. La misma funcionalidad con LCEL:

# Enfoque LCEL (recomendado LangChain v0.3+)
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

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

# 2. Definir el prompt
prompt = ChatPromptTemplate.from_template(
    "Eres un(a) {role} con experiencia.\nPregunta del usuario: {question}\nRespuesta profesional:"
)

# 3. Conectar con el pipe
chain = prompt | model

# 4. Invocar (mapeo automático)
result = chain.invoke({"role": "ingeniero frontend", "question": "¿React o Vue?"})
print(result.content)

De 15 a 9 líneas — pero lo importante es:

• Streaming automático: cambia invoke por stream sin tocar nada más
• Async automático: ainvoke o astream en una línea
• Batch automático: método batch con una lista de entradas

El operador pipe | se inspira en los pipes de Linux. cat log.txt | grep error | wc -l encadena tres comandos; LCEL hace lo mismo: prompt | model | output_parser, los datos fluyen de izquierda a derecha.

Al principio la sintaxis me desconcertó — en Python, ¿no es | un OR a nivel de bits? En realidad es azúcar sintáctico de Python 3.10 vía __or__. Elegante.

Capítulo 2: Cómo funciona el operador pipe

El pipe parece simple; detrás hay un diseño completo. Un experimento:

Railway - Plataforma de despliegue moderna, sin ops complejas, en producción en minutos

Empezar ahora →Anuncio

from langchain_core.runnables import RunnableLambda

def add_one(x: int) -> int:
    return x + 1

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

add_one_runnable = RunnableLambda(add_one)
multiply_two_runnable = RunnableLambda(multiply_two)

chain = add_one_runnable | multiply_two_runnable

result = chain.invoke(3)  # 3 -> 4 -> 8
print(result)  # Salida: 8

chain = add_one_runnable | multiply_two_runnable llama a add_one_runnable.__or__(multiply_two_runnable).

Runnable implementa __or__ y devuelve un RunnableSequence que ejecuta cada componente en secuencia, pasando la salida al siguiente.

Runnable es la abstracción central de LCEL. Cuatro métodos unifican todos los componentes:

Versiones async: astream, abatch, abatch_as_completed.

PromptTemplate, ChatModel, OutputParser o RunnableLambda — misma interfaz:

chain.invoke({"input": "hello"})
chain.stream({"input": "hello"})
chain.batch([{"input": "a"}, {"input": "b"}])

await chain.ainvoke({"input": "hello"})
async for chunk in chain.astream({"input": "hello"}):
    print(chunk, end="", flush=True)

Flujo de datos en un pipe algo más complejo:

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

prompt = ChatPromptTemplate.from_template("Traduce al inglés: {text}")
model = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

chain = prompt | model | parser

result = chain.invoke({"text": "你好世界"})
print(result)  # Salida: Hello World

Flujo:

{"text": "你好世界"}
       ↓
   [prompt] → ChatPromptValue(messages=[HumanMessage("Traduce al inglés: 你好世界")])
       ↓
   [model]  → AIMessage(content="Hello World")
       ↓
   [parser] → "Hello World" (str)

Cada componente tiene tipos de entrada/salida acordados. Prompt: dict → ChatPromptValue; Model: PromptValue → AIMessage; Parser: Message → str.

Si el orden es incorrecto (model | prompt), error en ejecución — el IDE también puede avisar con type hints.

Capítulo 3: Streaming en la práctica

El streaming es lo que más me impresionó de LCEL.

Imagina un bot de atención al cliente. Pregunta compleja: «Analiza las ventajas y desventajas de este producto y compáralo con la competencia». GPT-4o-mini tarda 5-8 segundos.

Sin streaming, pantalla en blanco 8 segundos. El usuario se pregunta: ¿se colgó? ¿la red? ¿refrescar?

Con streaming, el primer carácter aparece al instante, palabra tras palabra — como una respuesta en directo. La espera psicológica desaparece.

Antes, el streaming en LangChain exigía mucho código:

# Streaming tradicional (era 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="atención al cliente", question="...")

Problemas:

1. Callbacks engorrosos — para enviar tokens al frontend, heredar BaseCallbackHandler.
2. Cambio stream/no-stream — streaming=True/False al inicializar, no en ejecución.
3. Async stream — AsyncCallbackHandler, código duplicado.

LCEL integra el streaming:

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

model = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template(
    "Eres un(a) {role} profesional. Responde: {question}"
)
chain = prompt | model

result = chain.invoke({"role": "atención al cliente", "question": "Analiza las ventajas y desventajas de este producto"})
print(result.content)

for chunk in chain.stream({"role": "atención al cliente", "question": "Analiza las ventajas y desventajas de este producto"}):
    print(chunk.content, end="", flush=True)

invoke → stream, y listo.

Ejemplo completo de chat en tiempo 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

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
prompt = ChatPromptTemplate.from_messages([
    ("system", "Eres un asistente de IA amigable, experto en preguntas técnicas."),
    ("human", "{input}")
])
parser = StrOutputParser()

chain = prompt | model | parser
memory = ChatMessageHistory()

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

async def chat_stream(user_input: str):
    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")

async def main():
    print("=== Asistente de IA (demo streaming) ===")
    await chat_stream("¿Qué es LangChain?")
    await chat_stream("¿Para qué sirve?")
    await chat_stream("¿Qué relación tiene con LCEL?")

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

Resultado:

=== Asistente de IA (demo streaming) ===
IA: LangChain es un framework de código abierto para construir aplicaciones basadas en grandes modelos de lenguaje...
IA: Sirve para construir chatbots, sistemas RAG, aplicaciones Agent, etc...
IA: LCEL es LangChain Expression Language, componente central de LangChain...

Cada carácter en tiempo real.

Comparación medida stream vs no-stream:

Vultr - Servidor cloud NVMe de alto rendimiento, 32 ubicaciones globales, despliegue Docker con un clic

Ver precios →Anuncio

Sin streaming, la latencia del primer token equivale al tiempo total. Con streaming, es el tiempo del primer token — a menudo < 1 s.

Mecanismo: stream llama recursivamente al stream de cada componente. El modelo usa la API streaming de OpenAI; prompt y parser suelen devolver un bloque completo. LCEL coordina todo — un componente no streamable se trata como retorno único sin romper el flujo.

Capítulo 4: Componentes Runnable en detalle

El pipe encadena; los proyectos reales exigen paralelismo, resultados intermedios y transformaciones personalizadas. LangChain ofrece Runnables dedicados.

RunnableParallel: ejecución en paralelo

En RAG, a menudo se consultan vector, palabras clave y grafo a la vez:

from langchain_core.runnables import RunnableParallel

def vector_search(query: str) -> str:
    return f"Resultado vectorial: 3 documentos para {query}"

def keyword_search(query: str) -> str:
    return f"Resultado por palabras clave: 5 registros para {query}"

def graph_search(query: str) -> str:
    return f"Resultado de grafo: 2 entidades para {query}"

retrievers = RunnableParallel(
    vector=RunnableLambda(vector_search),
    keyword=RunnableLambda(keyword_search),
    graph=RunnableLambda(graph_search)
)

results = retrievers.invoke("LangChain LCEL")
print(results)

RunnableParallel devuelve un dict {nombre_de_rama: resultado}.

RunnablePassthrough: conservar la entrada

En RAG, el retriever quiere la consulta original; el generador quiere documentos + consulta:

from langchain_core.runnables import RunnablePassthrough

def retrieve(query: dict) -> str:
    return "Contenido del documento recuperado..."

chain = RunnableParallel(
    retrieved_docs=RunnableLambda(retrieve),
    original_query=RunnablePassthrough()
)

result = chain.invoke({"query": "¿Qué es LCEL?"})
print(result)

RunnablePassthrough pasa la entrada tal cual — crucial en pipes complejos.

RunnableLambda: transformación personalizada

from langchain_core.runnables import RunnableLambda

def format_output(result: dict) -> str:
    docs = result["retrieved_docs"]
    query = result["original_query"]["query"]
    return f"Referencias: {docs}\nPregunta: {query}\nResponde a partir de las referencias:"

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

formatted = chain.invoke({"query": "¿Qué es LCEL?"})
print(formatted)

RunnableLambda es la «cola universal» del pipe.

Pipeline RAG completo

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

embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts(
    ["LCEL es el lenguaje de expresión de LangChain",
     "El operador pipe sirve para encadenar componentes",
     "Runnable es la abstracción central de LCEL"],
    embeddings
)
retriever = vectorstore.as_retriever()

rag_prompt = ChatPromptTemplate.from_template(
    """Responde a partir de las siguientes referencias.

Referencias:
{context}

Pregunta: {question}

Respuesta precisa y detallada:"""
)

def format_docs(docs) -> str:
    return "\n".join(doc.page_content for doc in docs)

rag_chain = (
    RunnableParallel(
        context=retriever | RunnableLambda(format_docs),
        question=RunnablePassthrough()
    )
    | rag_prompt
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

answer = rag_chain.invoke("¿Qué es LCEL?")
print(answer)

Esquema:

{"question": "¿Qué es LCEL?"}
              ↓
    ┌─────────┴─────────┐
    ↓                   ↓
[retriever]        [Passthrough]
    ↓                   ↓
format_docs        question
    ↓                   ↓
    └─────────┬─────────┘
              ↓
        {"context": "...", "question": "..."}
              ↓
         [rag_prompt] → [model] → [parser]
              ↓
        "Contenido de la respuesta..."

Esta implementación conecta con Optimización de sistemas RAG en la práctica — reranking, multi-recall, etc. se aplican directamente a esta estructura LCEL.

Capítulo 5: Migración de cadenas antiguas

Si tu proyecto aún usa LLMChain, la migración es accesible. El año pasado migré un bot de atención al cliente de e-commerce en dos días. Algunos patrones habituales.

LLMChain → sintaxis pipe

# Antiguo (LLMChain)
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import OpenAI

llm = OpenAI(temperature=0.7)
prompt = PromptTemplate(
    template="Pregunta: {question}\nRespuesta:",
    input_variables=["question"]
)
chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run(question="...")

# Nuevo (LCEL)
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

model = ChatOpenAI(temperature=0.7)
prompt = ChatPromptTemplate.from_template("Pregunta: {question}\nRespuesta:")
chain = prompt | model
result = chain.invoke({"question": "..."})

Puntos clave:

1. Modelo: OpenAI (Completion) → ChatOpenAI (Chat API).
2. Prompt: ChatPromptTemplate para system message y multirrol.
3. Invocación: run() → invoke(), retorno Message — usa .content.

SequentialChain → RunnableParallel

El código antiguo encadenaba varios pasos con SequentialChain; tras la migración, conecta directamente con el pipe:

# Antiguo (SequentialChain)
from langchain.chains import SequentialChain, LLMChain

title_chain = LLMChain(
    llm=llm, prompt=title_prompt,
    output_key="title"
)

content_chain = LLMChain(
    llm=llm, prompt=content_prompt,
    output_key="content"
)

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

# Nuevo (LCEL)
from langchain_core.runnables import RunnableParallel

title_chain = title_prompt | model
content_chain = content_prompt | model

full_chain = RunnableParallel(
    title=title_chain,
    content=content_chain
)
result = full_chain.invoke({"topic": "Desarrollo de IA"})
print(result["title"].content, result["content"].content)

SequentialChain ejecuta en serie — cada cadena espera la anterior. RunnableParallel paraleliza y acelera. Para serie (el segundo paso depende del primero), encadena con el pipe:

chain = (
    title_prompt | model | StrOutputParser()
    | (lambda title: {"topic": topic, "title": title})
    | content_prompt | model
)

TransformChain → RunnableLambda

TransformChain insertaba lógica personalizada; migra a RunnableLambda:

# Antiguo (TransformChain)
from langchain.chains import TransformChain

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

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

# Nuevo (RunnableLambda)
from langchain_core.runnables import RunnableLambda

transform_chain = RunnableLambda(transform_func)

RunnableLambda es más flexible — sin declaración explícita de variables, participación directa en el pipe.

Trampas de migración

Algunas trampas que encontré:

Trampa 1: el tipo de retorno cambió

LLMChain.run() devuelve string; LCEL.invoke() devuelve un objeto Message.

KiloClaw - OpenClaw gestionado para empresas: orquestación de agentes IA y enrutamiento inteligente

Empezar →Anuncio

result = chain.run(...)  # str
result = chain.invoke(...)  # AIMessage
text = result.content  # str

Solución: añade StrOutputParser al final del pipe.

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

Trampa 2: migración de componentes de memoria

Código antiguo con ConversationChain:

from langchain.chains import ConversationChain

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

Código nuevo con 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"
)

Más parámetros — nombres de campo explícitos. Consulta Llamadas a herramientas Agent en la práctica para un sistema de diálogo completo.

Trampa 3: rutas de importación LangChain v0.3

Muchos componentes pasan de langchain a langchain_core o langchain_community:

# Imports antiguos
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

# Imports nuevos
from langchain_core.runnables import RunnableLambda, RunnableParallel
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_message_histories import ChatMessageHistory

El IDE señala los errores — sigue las sugerencias.

Caso de migración en producción

Bot de atención al cliente de e-commerce migrado el año pasado: ~300 líneas (LLMChain + SequentialChain + TransformChain), lógica central tras la migración:

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

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)

intent_prompt = ChatPromptTemplate.from_template(
    """Analiza la intención del usuario, devuelve una categoría:
    - product_query (consulta de producto)
    - order_status (consulta de pedido)
    - complaint (queja o sugerencia)
    - other (otro)

    Mensaje: {message}
    Categoría:"""
)

product_prompt = ChatPromptTemplate.from_template(
    "Consulta de producto: {message}\nBusca en el catálogo y responde:"
)
order_prompt = ChatPromptTemplate.from_template(
    "Consulta de pedido: {message}\nConsulta el estado y responde:"
)

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"

intent_chain = intent_prompt | model | StrOutputParser() | RunnableLambda(route_by_intent)

full_chain = (
    {"message": RunnablePassthrough()}
    | RunnableParallel(
        intent=intent_chain,
        original=RunnablePassthrough()
    )
    # Enrutamiento por rama — RunnableBranch en producción
)

async for chunk in full_chain.astream({"message": "Quiero consultar el pedido 12345"}):
    print(chunk, end="", flush=True)

~150 líneas tras la migración — la mitad. Streaming y async, antes desarrollos aparte, caben en una línea.

Resumen

LCEL es la arquitectura recomendada de LangChain v0.3+. El pipe simplifica el código, Runnable unifica la invocación y el streaming integrado mejora la UX.

El reto no es la sintaxis sino el cambio de mentalidad: declaración explícita frente a flujo implícito en el pipe.

Migra por etapas: primero cadenas simples, luego composición compleja, con LangSmith para depurar tipos.

A continuación: Gestión de estado con LangGraph en la práctica. LangGraph + LCEL: cadenas simples en LCEL, estado complejo en LangGraph — combinación madura hoy.

Navegación — serie Desarrollo de IA en la práctica

• Parte 1: Claude API: de la autenticación al diálogo multiturno
• Parte 2: Prompt Engineering avanzado en la práctica
• Parte 8: Optimización de sistemas RAG: equilibrio entre precisión de recuperación y calidad de generación
• Parte 11: Gestión de estado con LangGraph — arquitectura Agent 2026
• Parte 13: LangChain LCEL en práctica: del encadenamiento tradicional a la respuesta en streaming (este artículo)
• Parte 13: Llamadas a herramientas Agent: hacer que la IA invoque APIs y servicios externos