Runnable

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

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)