6. Pipelines Declarativos con LCEL

En capítulos anteriores, hemos estado escribiendo código imperativo para orquestar interacciones con LLM: crear un prompt, invocar el modelo, analizar la respuesta. Esto funciona, pero a medida que las aplicaciones de IA se vuelven más complejas, este enfoque se vuelve verboso y más difícil de mantener. Terminas con llamadas a funciones profundamente anidadas, manejo manual de errores en cada paso y dificultad para comprender el flujo general de datos.

LangChain Expression Language (LCEL) resuelve esto permitiéndote declarar lo que quieres que suceda, no cómo hacerlo suceder. En lugar de escribir código procedimental que llama funciones en secuencia, compones componentes usando un simple operador pipe (|) que se lee como un pipeline de Unix. El resultado es código más limpio y mantenible que expresa claramente el flujo de datos a través de tu sistema de IA.

Este capítulo introduce LCEL para construir flujos de trabajo lineales - secuencias de operaciones donde los datos fluyen de principio a fin sin ramificaciones ni bucles. Cubriremos cuándo usar LCEL, cómo componer pipelines y cómo ejecutarlos tanto de forma síncrona como con salida en streaming.

6.1) ¿Por qué LCEL?

El Problema que LCEL Resuelve

Comencemos con un ejemplo concreto. Supongamos que estás construyendo un asistente de soporte al cliente que necesita:

1. Tomar una pregunta del usuario
2. Formatearla en un prompt con instrucciones del sistema y el mensaje del usuario
3. Enviarla a un LLM
4. Analizar la respuesta para extraer solo el contenido de texto

Así es como podrías escribir esto de forma imperativa (sin 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:
    # Paso 1: Formatear el prompt con la pregunta
    messages = prompt.invoke({"question": question})
    
    # Paso 2: Invocar el LLM
    response = llm.invoke(messages)
    
    # Paso 3: Analizar la salida para extraer el contenido de texto
    result = parser.invoke(response)
    
    return result
 
# Usarlo
result = answer_question("How do I reset my password?")
print(result)

Salida:

Para restablecer tu contraseña, sigue estos pasos:
1. Ve a la página de inicio de sesión
2. Haz clic en "Olvidé mi contraseña"
3. Ingresa tu dirección de correo electrónico
4. Revisa tu correo para obtener un enlace de restablecimiento
5. Sigue el enlace y crea una nueva contraseña

Esto funciona, pero observa los problemas:

Verbosidad: Cada paso requiere asignación explícita de variables y llamadas a funciones. La lógica real (formatear → invocar → analizar) está enterrada en código repetitivo.

Estructura Rígida: Si quieres agregar un paso (como validar la pregunta o registrar la respuesta), necesitas insertar código en medio de la función, aumentando la complejidad.

Sin Streaming Integrado: Para transmitir tokens a medida que llegan, necesitarías reescribir toda la función para usar llm.stream() y manejar la iteración asíncrona manualmente.

Flujo de Datos Poco Claro: Al leer el código, no es inmediatamente obvio que esto es un pipeline simple. Tienes que rastrear las asignaciones de variables para entender el flujo.

Ahora veamos la versión con LCEL:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
# Definir los componentes (igual que antes)
llm = ChatOpenAI(model="gpt-4o-mini")
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful customer support assistant."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
# Componerlos en una cadena usando el operador pipe
chain = prompt | llm | parser
 
# Usarlo
result = chain.invoke({"question": "How do I reset my password?"})
print(result)

Salida:

Para restablecer tu contraseña, sigue estos pasos:
1. Ve a la página de inicio de sesión
2. Haz clic en "Olvidé mi contraseña"
3. Ingresa tu dirección de correo electrónico
4. Revisa tu correo para obtener un enlace de restablecimiento
5. Sigue el enlace y crea una nueva contraseña

La salida es idéntica, pero el código es dramáticamente diferente:

Declarativo: chain = prompt | llm | parser expresa todo el flujo en una línea. Léelo de izquierda a derecha: prompt → LLM → parser.

Componible: Cada componente (prompt, llm, parser) es independiente y reutilizable. Puedes intercambiar componentes sin reescribir el pipeline.

Streaming Integrado: Cambia entre chain.invoke() y chain.stream() sin cambiar la definición del pipeline.

Intención Clara: El operador | hace que el flujo de datos sea obvio de un vistazo.

Dónde encaja LCEL: Flujos de trabajo lineales

LCEL está diseñado para flujos de trabajo lineales - secuencias donde los datos fluyen en una dirección desde el inicio hasta el final sin ramificaciones ni bucles.

Este patrón lineal cubre muchas aplicaciones de IA. Considera un sistema de preguntas y respuestas sobre documentos: recibes una pregunta → recuperas documentos relevantes → formateas un prompt → envías al LLM → analizas la respuesta. Cada paso es una secuencia clara donde la salida de un paso se convierte en la entrada del siguiente.

Pero ¿qué pasa si tu flujo de trabajo necesita:

• Que el LLM decida qué herramienta llamar basándose en la pregunta
• Llamar a una herramienta, observar el resultado, luego decidir qué hacer a continuación
• Reintentar operaciones fallidas con diferentes enfoques

Estos escenarios requieren bucles y ramificación condicional - cosas que LCEL no puede manejar. Por eso existe LangGraph (lo introduciremos en el Capítulo 15).

Aquí hay una comparación visual de los dos enfoques:

LCEL es perfecto para pipelines directos donde cada paso procesa la salida del paso anterior. Los datos fluyen en una sola dirección.

LangGraph es para cuando necesitas toma de decisiones y bucles. El agente puede actuar, observar resultados y pensar de nuevo.

Para este capítulo, nos estamos enfocando en LCEL. ¿Por qué dominar primero los pipelines lineales?

• Fundamento: El operador pipe (|) es la sintaxis central de LangChain. Entender esto hace que todo lo demás sea más fácil.
• Prerrequisito para LangGraph: Los agentes de LangGraph usan cadenas LCEL extensivamente dentro de sus nodos
• Patrón del mundo real: Los agentes complejos se construyen combinando cadenas LCEL

El resto de este capítulo te mostrará cómo construir estos pipelines lineales con el operador pipe (|).

6.2) Componiendo Pipelines con el Operador |

Cómo Funcionan los Pipes de LCEL

El operador pipe de LCEL (|) funciona gracias a la interfaz Runnable.

¿Qué es un Runnable?

Runnable es la interfaz estándar de LangChain. Cuando un componente implementa la interfaz Runnable, puede encadenarse con otros componentes usando el operador |.

Cada Runnable proporciona estos métodos:

• .invoke(input) - Ejecutar una vez y obtener el resultado completo
• .stream(input) - Ejecutar y recibir cada palabra a medida que el LLM la genera
• .batch(inputs) - Ejecutar múltiples veces con diferentes entradas y obtener todos los resultados

Para este capítulo, nos enfocaremos en .invoke() y .stream() (cubriremos .batch() más adelante cuando sea necesario).

¿Por qué funciona |?

Porque la clase Runnable implementa el operador | usando la sobrecarga de operadores de Python. Cuando escribes prompt | llm, crea un nuevo Runnable que ejecuta los dos componentes secuencialmente.

La mayoría de los componentes de LangChain son Runnables

Por eso puedes encadenar tantos componentes diferentes:

• ChatPromptTemplate es un Runnable
• ChatOpenAI es un Runnable
• StrOutputParser es un Runnable
• ¡Incluso las cadenas personalizadas que creas con | son Runnables!

Esto significa que puedes construir pipelines complejos combinando otros más simples.

Conectando Componentes: Tipos de Entrada/Salida

Al conectar componentes con |, necesitas asegurarte de que el tipo de salida de un componente coincida con el tipo de entrada del siguiente.

Firmas de componentes clave:

Flujo de ejemplo:

chain = prompt | llm | parser

Así es como el tipo de datos se transforma a medida que pasa por cada componente:

• prompt: Toma dict como entrada y convierte a list[BaseMessage]
• llm: Toma list[BaseMessage] como entrada y convierte a AIMessage
• parser: Toma AIMessage como entrada y convierte a str

Veamos esto en acción:

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."

Lo que sucede en cada paso:

¿Qué sucede si los tipos no coinciden?

Si intentas conectar componentes incompatibles, obtendrás un error:

# ERROR: Esto no funcionará
chain = llm | prompt  # LLM produce AIMessage, pero prompt requiere dict como entrada

El mensaje de error te dirá qué tipo de entrada espera el siguiente componente versus qué tipo de entrada realmente recibió.

6.3) Ejecutando una Cadena: .invoke() y .stream()

Ejecutando un Pipeline

Una vez que has compuesto una cadena, la ejecutas usando el método .invoke(). Esta es la forma síncrona de ejecutar un pipeline - espera la respuesta completa antes de devolver el 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()
 
# Invocación síncrona
result = chain.invoke({"question": "What is 2+2?"})
print(result)

Salida:

2+2 es igual a 4.

El método .invoke() es directo: pasa el parámetro de entrada que el primer componente espera y obtén la salida producida por el último componente.

Transmitiendo Tokens desde la Misma Cadena

Aunque .invoke() es simple, tiene una limitación para aplicaciones orientadas al usuario: los usuarios no ven nada hasta que la respuesta completa está lista. Para respuestas largas (10-20 segundos), esto crea una mala experiencia de usuario.

Streaming muestra tokens inmediatamente a medida que se generan, en lugar de esperar la respuesta completa. Es el efecto de escritura que ves en ChatGPT.

Veamos streaming en acción:

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 fragmentos a medida que llegan
for chunk in chain.stream({"question": "Explain what Python is in one sentence"}):
    print(chunk, end="", flush=True)

Salida (mostrada en tiempo real, token por token):

Python es un lenguaje de programación versátil de alto nivel conocido por su simplicidad y legibilidad, ampliamente utilizado en desarrollo web, ciencia de datos y automatización.

Observa que la definición de la cadena es idéntica al ejemplo de .invoke() anterior. No necesitamos reconstruirla - solo llamamos .stream() en lugar de .invoke().

El método .stream() devuelve resultados como fragmentos mientras el LLM genera tokens. El print(chunk, end="", flush=True) muestra cada fragmento inmediatamente en pantalla, creando el efecto de escritura en vivo.

Cuándo Usar .invoke() vs .stream()

Usa .invoke() cuando:

• Solo necesitas el resultado final (análisis, traducción, clasificación)
• La respuesta es corta y el tiempo de espera no es un problema
• Necesitas la salida completa antes de proceder al siguiente paso

Usa .stream() cuando:

• Los usuarios necesitan ver el progreso (chat, generación de contenido)
• La respuesta es larga y el tiempo de espera sería notable
• Estás construyendo una UI donde la retroalimentación en tiempo real importa

Ambos métodos funcionan en la misma cadena. Define la cadena una vez, luego elige el modo de ejecución según tus necesidades.

En este capítulo, aprendiste LCEL - la sintaxis declarativa de pipelines de LangChain:

• Construir pipelines con el operador pipe: prompt | llm | parser
• Ejecutar flexiblemente: Usa .invoke() para resultados completos o .stream() para salida en tiempo real
• Seguridad de tipos: Coincide los tipos de salida con los tipos de entrada al encadenar componentes

La misma cadena funciona para ambos modos de ejecución - define una vez, usa en cualquier lugar.

Siguiente: El Capítulo 7 cubre salida estructurada con Pydantic, permitiéndote extraer datos JSON validados de respuestas LLM.