LangChain LCEL: componer cadenas con pipes y fallbacks

Visión General

Pegar una llamada a un LLM en tu código suele empezar con una función que llama al modelo, otra que post-procesa la salida y una tercera que vuelve a llamar al modelo. Tras unas iteraciones terminas con un script imperativo donde reintentos, streaming y fallbacks implican reescribir el mismo cableado una y otra vez. LangChain Expression Language (LCEL) reemplaza ese cableado con el operador pipe (|): cada paso declara qué hace y el framework maneja ejecución, streaming, batching y recuperación de errores de forma uniforme.

Esta receta construye cadenas LCEL paso a paso, desde un prompt | model | parser mínimo hasta ramas paralelas, componentes Python personalizados, reintentos y fallbacks. Para la versión a nivel de diseño de esta idea, consulta el patrón prompt chaining.

Cuándo Usar

• Workflows LLM multi-paso (resumir y luego traducir, extraer y luego clasificar).
• Pipelines que necesitan streaming, batching o lógica de reintento sin plomería manual.
• Componer prompts, herramientas y parsers en cadenas reutilizables y testeables.
• No es la mejor opción para agentes con estado que necesitan loops y grafos de llamadas a herramientas — eso es territorio de LangGraph, cubierto en la receta de máquina de estado de LangGraph.

Requisitos Previos

• Python 3.10+
• Paquetes langchain y langchain-openai
• Una API key de OpenAI en OPENAI_API_KEY

Solución

1. Instalar dependencias

pip install langchain langchain-openai

2. Cadena básica — Prompt | Model | Parser

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a concise technical writer."),
    ("user", "Explain {topic} in under 100 words."),
])

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
parser = StrOutputParser()

# El operador pipe de LCEL compone la cadena
chain = prompt | model | parser

result = chain.invoke({"topic": "vector databases"})
print(result)

3. Cadena multi-paso — resumir y luego traducir

summarize_prompt = ChatPromptTemplate.from_template(
    "Summarize this text in 3 bullet points:\n\n{text}"
)

translate_prompt = ChatPromptTemplate.from_template(
    "Translate to Spanish:\n\n{summary}"
)

summarize_chain = summarize_prompt | model | StrOutputParser()
translate_chain = translate_prompt | model | StrOutputParser()

# Encadenar la salida de summarize hacia translate
full_chain = (
    {"summary": summarize_chain}
    | translate_chain
)

result = full_chain.invoke({
    "text": "LangChain is a framework for building applications powered by LLMs..."
})

4. Ramas paralelas con RunnableParallel

from langchain_core.runnables import RunnableParallel

# Ejecutar múltiples cadenas simultáneamente sobre la misma entrada
classification_prompt = ChatPromptTemplate.from_template(
    "Classify this text as 'technical', 'marketing', or 'other':\n\n{text}"
)

sentiment_prompt = ChatPromptTemplate.from_template(
    "What is the sentiment of this text? (positive, neutral, negative):\n\n{text}"
)

parallel_chain = RunnableParallel(
    classification=classification_prompt | model | StrOutputParser(),
    sentiment=sentiment_prompt | model | StrOutputParser(),
)

result = parallel_chain.invoke({
    "text": "Our new product is amazing and uses cutting-edge AI!"
})
# {'classification': 'marketing', 'sentiment': 'positive'}

5. Componentes Runnable personalizados

from langchain_core.runnables import RunnableLambda

def extract_keywords(text: str) -> list[str]:
    """Extract keywords from text (simplified)."""
    stop_words = {"the", "a", "an", "is", "are", "and", "or", "in", "on"}
    words = [w.lower().strip(".,!?") for w in text.split()]
    return [w for w in words if w not in stop_words and len(w) > 3][:10]

def format_output(data: dict) -> str:
    """Format the final output."""
    return f"Keywords: {', '.join(data['keywords'])}\nSummary: {data['summary']}"

custom_chain = (
    RunnableParallel(
        keywords=RunnableLambda(extract_keywords),
        summary=summarize_chain,
    )
    | RunnableLambda(format_output)
)

result = custom_chain.invoke({
    "text": "Vector databases enable semantic search over embeddings..."
})

6. Agregar reintento y fallback

fast_model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
strong_model = ChatOpenAI(model="gpt-4o", temperature=0)

fast_chain = prompt | fast_model | StrOutputParser()
strong_chain = prompt | strong_model | StrOutputParser()

# Intentar el modelo rápido primero, fallback al fuerte si hay error
chain_with_fallback = fast_chain.with_fallbacks([strong_chain])

# Agregar lógica de reintento
chain_with_retry = fast_chain.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
)

7. Streaming de salida

# Streamear tokens a medida que llegan
streaming_chain = prompt | model | StrOutputParser()

for chunk in streaming_chain.stream({"topic": "GraphQL federation"}):
    print(chunk, end="", flush=True)

Para construir un endpoint de streaming completo alrededor de este patrón, consulta streaming de respuestas LLM.

8. Procesamiento batch

# Procesar múltiples entradas en paralelo
results = chain.batch([
    {"topic": "Redis caching"},
    {"topic": "PostgreSQL indexing"},
    {"topic": "Docker multi-stage builds"},
])

for r in results:
    print(r)
    print("---")

Cómo Funciona

Cada componente de una cadena — el prompt, el modelo, el parser, tus funciones Python — implementa la misma interfaz Runnable con invoke, stream y batch. El operador | los enlaza para que la salida de cada paso sea la entrada del siguiente, igual que los pipes de Unix conectan comandos. Como todo comparte la interfaz, capacidades como streaming y batching vienen gratis en cualquier cadena que armes.

El siguiente diagrama muestra cómo fluye una cadena de producción típica, incluidas las ramas paralelas del paso 4:

RunnableParallel vale la pena por la latencia cuando las ramas son independientes: dos llamadas de 800ms corren en ~800ms en vez de ~1600ms. RunnableLambda envuelve cualquier función Python para meter lógica personalizada — validación, extracción de keywords, llamadas a APIs — en medio de la cadena sin salir de la interfaz Runnable.

with_fallbacks y with_retry resuelven problemas distintos. El reintento maneja fallos transitorios (rate limits, timeouts) corriendo la misma cadena de nuevo; el fallback maneja problemas de capacidad cambiando a otra cadena — típicamente un modelo más fuerte o más barato — cuando la principal sigue fallando. Combinarlos (fast_chain.with_retry(...).with_fallbacks(...)) da primero reintentos transitorios y cambio de modelo solo cuando los reintentos se agotan.

batch empuja las entradas por la infraestructura paralela del proveedor, dramáticamente más rápido que iterar llamadas invoke, mientras que stream va entregando chunks a medida que el modelo los produce para que la UI pueda renderizar los primeros tokens temprano. Ambos son gratis porque cada runnable los implementa.

Variantes

Enrutamiento condicional

from langchain_core.runnables import RunnableBranch

summary_prompt = ChatPromptTemplate.from_template("Summarize briefly:\n\n{text}")
default_chain = summary_prompt | model | StrOutputParser()

technical_prompt = ChatPromptTemplate.from_template(
    "Write a technical summary preserving jargon:\n\n{text}"
)
marketing_prompt = ChatPromptTemplate.from_template(
    "Write a marketing-friendly summary:\n\n{text}"
)
technical_chain = technical_prompt | model | StrOutputParser()
marketing_chain = marketing_prompt | model | StrOutputParser()

# La primera condición que coincide gana; el último argumento es el fallback
branch_chain = RunnableBranch(
    (lambda x: "technical" in x["text"].lower(), technical_chain),
    (lambda x: "marketing" in x["text"].lower(), marketing_chain),
    default_chain,
)

result = branch_chain.invoke({"text": "This technical guide covers..."})

Salida JSON con parser estructurado

from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel

class ProductReview(BaseModel):
    rating: int
    summary: str
    pros: list[str]
    cons: list[str]

json_parser = JsonOutputParser(pydantic_object=ProductReview)

review_prompt = ChatPromptTemplate.from_template(
    "Analyze this review and extract structured data.\n{format_instructions}\n\nReview: {review}"
).partial(format_instructions=json_parser.get_format_instructions())

review_chain = review_prompt | model | json_parser

result = review_chain.invoke({"review": "Great product, fast delivery..."})
# {'rating': 5, 'summary': 'Positive review', 'pros': [...], 'cons': [...]}

Para más sobre obtener salida tipada de los modelos, consulta salida estructurada con function calling.

Ejecución async

import asyncio

async def run_chain():
    # Invoke no bloqueante — usar dentro de FastAPI o workers async
    result = await chain.ainvoke({"topic": "vector databases"})
    print(result)

    # Streaming async
    async for chunk in chain.astream({"topic": "Redis caching"}):
        print(chunk, end="", flush=True)

    # Batch async
    results = await chain.abatch([
        {"topic": "Redis caching"},
        {"topic": "PostgreSQL indexing"},
    ])
    print(results)

asyncio.run(run_chain())

Cadena conversacional con memoria

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

conversational_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("placeholder", "{history}"),
    ("human", "{input}"),
])

base_chain = conversational_prompt | model | StrOutputParser()

chain_with_history = RunnableWithMessageHistory(
    base_chain,
    lambda session_id: InMemoryChatMessageHistory(),
    input_messages_key="input",
    history_messages_key="history",
)

result = chain_with_history.invoke(
    {"input": "What is Redis?"},
    config={"configurable": {"session_id": "user123"}},
)

Mejores Prácticas

• Usa StrOutputParser para salida de texto simple: extrae el contenido string del mensaje de la IA en vez de devolver el objeto mensaje completo.
• Divide cadenas complejas en componentes más pequeños, cada uno con una sola responsabilidad; una cadena que puedes diagramar en una servilleta es una cadena que puedes depurar.
• Envuelve las cadenas de producción con with_fallbacks: las APIs de LLM hacen rate-limit y timeout en el peor momento posible.
• Prefiere stream para todo lo que un usuario mira en tiempo real; la latencia del primer token importa más que la latencia total.
• Mantén las claves de entrada consistentes entre pasos. La mayoría de los bugs de LCEL son un runnable que recibe {"text": ...} cuando esperaba {"summary": ...}.

Errores Comunes

Pasar claves de entrada incorrectas entre pasos es el fallo número uno en runtime: cada runnable espera claves específicas y el desajuste solo aparece cuando la cadena ejecuta, no cuando se define.

Correr ramas independientes en secuencia desperdicia latencia: si dos cadenas no dependen de la salida de la otra, envuélvelas en RunnableParallel y obtén ambos resultados en el tiempo de la más lenta.

Encadenar sin manejo de errores significa que un solo rate-limit mata todo el pipeline. Un wrapper with_fallbacks cuesta dos líneas y te ahorra el aviso a las 3 AM.

Por último, llamar invoke dentro de código async (FastAPI, workers asíncronos) bloquea el event loop. Usa ainvoke, astream y abatch, que existen en cada runnable.

Solución de Problemas

KeyError o errores de entrada faltante entre pasos. El runnable anterior produce una clave de dict que el siguiente prompt no declara, o el prompt espera una clave que nadie produce. Imprime la salida intermedia con RunnablePassthrough() o chain.get_graph().print_ascii() para ver los datos fluyendo entre pasos.

Una rama paralela falla y mata toda la cadena. RunnableParallel lanza la excepción en cuanto una rama falla, y los demás resultados se pierden. Envuelve cada rama frágil en su propio with_fallbacks para que un fallo degrade con gracia en vez de abortar el agregado.

Los reintentos se disparan en cada llamada y el costo de tokens sube. with_retry reintenta ante cualquier excepción, incluidos errores que causa tu propio prompt (parsers de salida malformados, entrada inválida). Registra los fallos antes de añadir reintentos: una cadena que siempre falla reintentará 3 veces y triplicará la factura sin arreglar nada.

El streaming funciona pero la salida final es incorrecta. Cuando consumes .stream() chunk a chunk, recuerda acumularlos si el código posterior necesita el string completo; los parsers de la cadena ven la salida completa, tu loop ve fragmentos.

invoke se cuelga dentro de FastAPI o un worker async. Las llamadas síncronas bloquean el event loop. Cambia a await chain.ainvoke(...) y toda la cadena corre sin bloqueo, incluida la llamada al modelo.

Ver También

• Guía completa de LangChain en producción para alternativas y el ecosistema más amplio.
• Streaming de respuestas LLM para endpoints de streaming en producción.
• Salida estructurada con function calling para salida tipada del modelo sin LCEL.
• Documentación de LCEL de LangChain para la referencia completa de la interfaz Runnable.

Código companion: proyecto de cadenas LCEL: las cadenas de esta página organizadas como módulos importables, más un main.py ejecutable.