Runnable invoke
El método invoke es la forma más directa y fundamental de ejecutar cualquier componente en LangChain. Representa la ejecución síncrona de un Runnable, donde proporcionas una entrada y recibes una salida de manera inmediata. Este método constituye la base sobre la cual se construyen todas las demás formas de ejecución.
Cuando trabajas con LCEL (LangChain Expression Language), cada componente que creas implementa automáticamente la interfaz Runnable, lo que significa que todos heredan el método invoke junto con sus capacidades de ejecución. Esta uniformidad permite que cualquier cadena, modelo o parser se comporte de manera consistente.
Ejecución básica con invoke
La sintaxis del método invoke es extremadamente simple: pasas los datos de entrada como argumento y obtienes el resultado procesado. Veamos cómo funciona con diferentes tipos de componentes:
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
# Configuración del modelo
llm = ChatOpenAI(model="gpt-4")
# Ejecución directa del modelo
response = llm.invoke("¿Cuál es la capital de Francia?")
print(response.content) # París
El método invoke maneja automáticamente la conversión de tipos cuando es necesario. Si pasas una cadena de texto a un modelo de chat, LangChain la convierte internamente en el formato de mensaje apropiado, simplificando significativamente el código.
Ejemplo de invoke:
Invoke en cadenas LCEL
Cuando construyes cadenas usando LCEL, el método invoke ejecuta toda la secuencia de componentes de manera secuencial, pasando la salida de cada paso como entrada del siguiente:
from langchain_core.output_parsers import StrOutputParser
# Construcción de la cadena
prompt = ChatPromptTemplate.from_template(
"Explica el concepto de {tema} en {nivel} palabras máximo"
)
chain = prompt | llm | StrOutputParser()
# Ejecución de la cadena completa
result = chain.invoke({
"tema": "inteligencia artificial",
"nivel": "50"
})
print(result)
En este ejemplo, invoke ejecuta secuencialmente: el prompt formatea la entrada, el modelo genera la respuesta, y el parser extrae el contenido como string. Todo esto ocurre en una sola llamada al método invoke.
Manejo de configuración en invoke
El método invoke acepta un segundo parámetro opcional llamado config
que permite personalizar el comportamiento de la ejecución sin modificar la cadena original:
from langchain_core.runnables import RunnableConfig
# Configuración personalizada
config = RunnableConfig(
tags=["produccion", "usuario-premium"],
metadata={"session_id": "abc123", "user_id": "user456"}
)
# Ejecución con configuración
result = chain.invoke(
{"tema": "machine learning", "nivel": "100"},
config=config
)
Esta configuración es especialmente útil para trazabilidad y debugging, ya que permite etiquetar y rastrear ejecuciones específicas sin alterar la lógica de la cadena.
Invoke con componentes complejos
Los componentes más sofisticados también implementan invoke de manera consistente. Por ejemplo, cuando trabajas con retrievers o herramientas de búsqueda:
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
# Cadena con múltiples componentes
template = """Basándote en el siguiente contexto: {context}
Pregunta: {question}
Respuesta:"""
prompt = ChatPromptTemplate.from_template(template)
# Cadena que procesa contexto y pregunta
chain = (
{"context": RunnablePassthrough(), "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
# Ejecución con datos estructurados
response = chain.invoke({
"context": "LangChain es un framework para desarrollar aplicaciones con LLMs",
"question": "¿Qué es LangChain?"
})
Gestión de errores en invoke
El método invoke propaga automáticamente las excepciones que puedan ocurrir durante la ejecución, permitiendo un manejo de errores claro y predecible:
try:
result = chain.invoke({"tema": "", "nivel": "invalid"})
except ValueError as e:
print(f"Error de validación: {e}")
except Exception as e:
print(f"Error durante la ejecución: {e}")
Esta aproximación directa al manejo de errores hace que invoke sea ideal para casos donde necesitas control inmediato sobre el flujo de ejecución y quieres manejar problemas de manera síncrona.
El método invoke representa la forma más fundamental de interactuar con componentes LangChain, proporcionando una interfaz simple pero completa para la ejecución síncrona de cualquier Runnable en tu aplicación.
Runnable batch
Guarda tu progreso
Inicia sesión para no perder tu progreso y accede a miles de tutoriales, ejercicios prácticos y nuestro asistente de IA.
Más de 25.000 desarrolladores ya confían en CertiDevs
El método batch permite ejecutar múltiples entradas de manera simultánea, optimizando significativamente el rendimiento cuando necesitas procesar varios elementos a la vez. A diferencia de invoke que maneja una sola entrada, batch está diseñado para procesamiento en lotes, aprovechando el paralelismo para reducir el tiempo total de ejecución.
Esta funcionalidad resulta especialmente valiosa cuando trabajas con grandes volúmenes de datos o cuando necesitas procesar múltiples consultas similares de manera eficiente. LangChain maneja automáticamente la distribución de carga y la sincronización de resultados.
Sintaxis básica de batch
El método batch acepta una lista de entradas y devuelve una lista de resultados en el mismo orden. La estructura es consistente con invoke, pero escalada para múltiples elementos:
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
llm = ChatOpenAI(model="gpt-4")
# Lista de preguntas para procesar
preguntas = [
"¿Cuál es la capital de España?",
"¿Cuál es la capital de Italia?",
"¿Cuál es la capital de Alemania?"
]
# Procesamiento en lote
respuestas = llm.batch(preguntas)
for i, respuesta in enumerate(respuestas):
print(f"Pregunta {i+1}: {respuestas[i].content}")
El método batch mantiene el orden de las entradas, garantizando que la respuesta en la posición i
corresponde exactamente a la entrada en la misma posición.
Ejemplo de batch:
Batch con cadenas LCEL
Cuando aplicas batch a cadenas construidas con LCEL, cada entrada se procesa a través de toda la cadena de manera independiente, pero de forma paralela:
from langchain_core.output_parsers import StrOutputParser
prompt = ChatPromptTemplate.from_template(
"Resume en {palabras} palabras el tema: {tema}"
)
chain = prompt | llm | StrOutputParser()
# Datos de entrada estructurados
entradas = [
{"tema": "inteligencia artificial", "palabras": "30"},
{"tema": "blockchain", "palabras": "25"},
{"tema": "computación cuántica", "palabras": "35"}
]
# Procesamiento en lote de la cadena completa
resultados = chain.batch(entradas)
for resultado in resultados:
print(f"Resumen: {resultado}\n")
Configuración de concurrencia
El método batch permite controlar el nivel de paralelismo mediante el parámetro config
, especificando cuántas operaciones pueden ejecutarse simultáneamente:
from langchain_core.runnables import RunnableConfig
# Configuración con límite de concurrencia
config = RunnableConfig(
max_concurrency=3 # Máximo 3 operaciones simultáneas
)
# Procesamiento con concurrencia limitada
resultados = chain.batch(entradas, config=config)
Esta configuración es crucial cuando trabajas con APIs externas que tienen límites de velocidad o cuando quieres controlar el uso de recursos del sistema.
Manejo de errores en batch
Una característica importante de batch es su capacidad para manejar errores de manera granular. Si una entrada falla, las demás continúan procesándose:
# Entradas con una potencialmente problemática
entradas_mixtas = [
{"tema": "machine learning", "palabras": "20"},
{"tema": "", "palabras": "invalid"}, # Entrada problemática
{"tema": "deep learning", "palabras": "25"}
]
try:
resultados = chain.batch(entradas_mixtas)
for i, resultado in enumerate(resultados):
if resultado is not None:
print(f"Resultado {i}: {resultado}")
else:
print(f"Error en entrada {i}")
except Exception as e:
print(f"Error general en el lote: {e}")
Optimización de rendimiento
El método batch implementa optimizaciones automáticas que van más allá del simple paralelismo. Para modelos de chat, puede agrupar múltiples mensajes en una sola llamada a la API cuando es posible:
import time
# Comparación de rendimiento
start_time = time.time()
# Procesamiento secuencial con invoke
resultados_secuencial = []
for entrada in entradas:
resultado = chain.invoke(entrada)
resultados_secuencial.append(resultado)
tiempo_secuencial = time.time() - start_time
# Procesamiento en lote
start_time = time.time()
resultados_batch = chain.batch(entradas)
tiempo_batch = time.time() - start_time
print(f"Tiempo secuencial: {tiempo_secuencial:.2f}s")
print(f"Tiempo batch: {tiempo_batch:.2f}s")
print(f"Mejora: {tiempo_secuencial/tiempo_batch:.1f}x más rápido")
Batch con componentes heterogéneos
Puedes usar batch con diferentes tipos de componentes en la misma cadena, manteniendo la eficiencia del procesamiento paralelo:
from langchain_core.runnables import RunnablePassthrough
# Cadena con procesamiento condicional
def procesar_entrada(data):
if data["tipo"] == "pregunta":
return f"Responde: {data['contenido']}"
else:
return f"Analiza: {data['contenido']}"
chain = (
RunnablePassthrough.assign(prompt=procesar_entrada)
| (lambda x: ChatPromptTemplate.from_template(x["prompt"]).format_prompt(contenido=x["contenido"]))
| llm
| StrOutputParser()
)
entradas_heterogeneas = [
{"tipo": "pregunta", "contenido": "¿Qué es Python?"},
{"tipo": "analisis", "contenido": "El mercado de criptomonedas"},
{"tipo": "pregunta", "contenido": "¿Cómo funciona Git?"}
]
resultados = chain.batch(entradas_heterogeneas)
El método batch transforma la manera en que manejas el procesamiento masivo en LangChain, proporcionando una interfaz simple para operaciones complejas que requieren alta eficiencia y paralelismo controlado.
Runnable stream
El método stream permite procesar respuestas de manera incremental, recibiendo fragmentos de datos a medida que se generan en lugar de esperar a que se complete toda la operación. Esta funcionalidad es especialmente valiosa para aplicaciones interactivas donde quieres mostrar resultados progresivamente, mejorando significativamente la experiencia del usuario.
A diferencia de invoke que devuelve un resultado completo o batch que procesa múltiples entradas simultáneamente, stream está diseñado para manejar flujos de datos continuos, permitiendo que tu aplicación responda de manera más fluida y natural.
Funcionamiento básico del streaming
El método stream devuelve un generador que produce fragmentos de la respuesta conforme se van generando. Esto es particularmente útil con modelos de lenguaje que pueden generar texto de manera incremental:
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
llm = ChatOpenAI(model="gpt-4")
# Streaming básico con un modelo
for chunk in llm.stream("Explica qué es la programación funcional"):
print(chunk.content, end="", flush=True)
Cada chunk contiene una porción de la respuesta completa, permitiendo que tu aplicación muestre el texto conforme se genera, similar a como funciona ChatGPT en su interfaz web.
Ejemplo:
Stream con cadenas LCEL
Cuando aplicas streaming a cadenas construidas con LCEL, el comportamiento se adapta inteligentemente a cada componente de la cadena:
from langchain_core.output_parsers import StrOutputParser
prompt = ChatPromptTemplate.from_template(
"Escribe un artículo de {palabras} palabras sobre {tema}"
)
chain = prompt | llm | StrOutputParser()
# Streaming de una cadena completa
for chunk in chain.stream({
"tema": "inteligencia artificial",
"palabras": "200"
}):
print(chunk, end="", flush=True)
En este caso, el prompt se procesa instantáneamente, pero el modelo y el parser generan contenido de manera incremental, proporcionando una experiencia fluida al usuario final.
Manejo de diferentes tipos de chunks
Los chunks que recibes pueden contener diferentes tipos de información dependiendo del componente que los genera. Es importante entender cómo manejar cada tipo:
# Streaming con información detallada de chunks
for chunk in llm.stream("Describe el proceso de fotosíntesis"):
if hasattr(chunk, 'content') and chunk.content:
print(f"Contenido: {chunk.content}")
if hasattr(chunk, 'response_metadata'):
print(f"Metadata: {chunk.response_metadata}")
Stream con configuración personalizada
Puedes personalizar el comportamiento del streaming mediante configuraciones específicas, controlando aspectos como el tamaño de los chunks o metadatos adicionales:
from langchain_core.runnables import RunnableConfig
config = RunnableConfig(
tags=["streaming", "tiempo-real"],
metadata={"session_id": "stream_001"}
)
# Streaming con configuración personalizada
for chunk in chain.stream(
{"tema": "machine learning", "palabras": "150"},
config=config
):
print(chunk, end="")
Implementación de streaming en aplicaciones web
El streaming es especialmente útil para aplicaciones web donde quieres mostrar respuestas progresivamente. Aquí un ejemplo de cómo estructurar el código para una aplicación interactiva:
import asyncio
from typing import AsyncGenerator
async def stream_response(query: str) -> AsyncGenerator[str, None]:
"""Generador asíncrono para streaming de respuestas"""
chain = prompt | llm | StrOutputParser()
async for chunk in chain.astream({"tema": query, "palabras": "100"}):
yield chunk
# Pequeña pausa para simular procesamiento natural
await asyncio.sleep(0.01)
# Uso en una aplicación web
async def handle_user_query(user_input: str):
response_buffer = ""
async for chunk in stream_response(user_input):
response_buffer += chunk
# Aquí enviarías el chunk al frontend
print(f"Enviando chunk: {chunk}")
return response_buffer
Stream con manejo de errores
El streaming requiere un manejo de errores más sofisticado, ya que los errores pueden ocurrir en cualquier momento durante la generación:
def safe_stream(query: str):
"""Streaming con manejo robusto de errores"""
try:
for chunk in chain.stream({"tema": query, "palabras": "100"}):
yield chunk
except Exception as e:
yield f"\n[Error durante la generación: {e}]"
return
# Uso con manejo de errores
for chunk in safe_stream("inteligencia artificial"):
print(chunk, end="", flush=True)
Aprendizajes de esta lección
- Comprender el método invoke para la ejecución síncrona de componentes Runnable.
- Aprender a utilizar batch para procesar múltiples entradas en paralelo y optimizar el rendimiento.
- Conocer el método stream para recibir respuestas incrementales y mejorar la experiencia interactiva.
- Manejar configuraciones personalizadas y control de concurrencia en la ejecución de componentes.
- Implementar técnicas de manejo de errores y optimización en los diferentes métodos de ejecución.
Completa LangChain y certifícate
Únete a nuestra plataforma y accede a miles de tutoriales, ejercicios prácticos, proyectos reales y nuestro asistente de IA personalizado para acelerar tu aprendizaje.
Asistente IA
Resuelve dudas al instante
Ejercicios
Practica con proyectos reales
Certificados
Valida tus conocimientos
Más de 25.000 desarrolladores ya se han certificado con CertiDevs