Invoke, Stream y Batch

Descripción de la cápsula

Ya sabes inicializar modelos con init_chat_model y configurar sus parámetros. Pero hasta ahora solo has usado un modo de ejecución: invoke(). En la práctica, un modelo de lenguaje puede ejecutarse de tres formas fundamentales — cada una diseñada para un escenario diferente.

invoke() envía un mensaje y espera la respuesta completa. stream() devuelve la respuesta token por token, como cuando ves a ChatGPT "escribir" en tiempo real. batch() procesa múltiples inputs en paralelo, ideal para cuando necesitas hacer muchas llamadas al mismo modelo. Además, cada modo tiene su versión asíncrona (ainvoke, astream, abatch) para aplicaciones que usan async/await.

Entender cuándo usar cada modo es lo que separa un prototipo de una aplicación real. En el proyecto del módulo — el Chat Multi-Proveedor con Fallback — el streaming será esencial para que el usuario vea las respuestas progresivamente en vez de esperar varios segundos a que llegue el texto completo.

invoke(): respuesta completa

invoke() es el modo más simple: envías un mensaje, esperas, y recibes la respuesta completa de una vez.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

response = model.invoke("¿Qué es LangChain?")
print(response.content)
# Output esperado: LangChain es un framework open-source para construir
# aplicaciones con modelos de lenguaje...

El flujo es secuencial: tu código se bloquea hasta que el modelo termina de generar toda la respuesta. Para preguntas cortas esto está bien — la latencia es apenas 1-2 segundos. Para respuestas largas (500+ tokens), el usuario puede esperar 5-10 segundos sin ver nada.

invoke() acepta un string simple o una lista de mensajes (SystemMessage, HumanMessage) — el formato de lista es más común en aplicaciones reales porque te permite incluir un system prompt:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage

model = init_chat_model("openai:gpt-4.1-mini")

response = model.invoke([
    SystemMessage(content="Eres un experto en Python. Responde en máximo 2 frases."),
    HumanMessage(content="¿Qué es un decorador?")
])
print(response.content)
# Output esperado: Un decorador es una función que envuelve otra función
# para extender su comportamiento sin modificarla directamente...

AIMessage: la estructura de respuesta

Cada vez que llamas a invoke(), stream() o batch(), el modelo retorna un objeto AIMessage. No es solo texto — contiene metadata valiosa.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")
response = model.invoke("¿Qué es Docker en una frase?")

# 1. content — el texto de la respuesta
print(response.content)
# Output esperado: Docker es una plataforma de contenedores que permite
# empaquetar aplicaciones con todas sus dependencias.

# 2. response_metadata — información del proveedor
print(response.response_metadata)
# Output esperado:
# {
#     'token_usage': {'completion_tokens': 25, 'prompt_tokens': 15, 'total_tokens': 40},
#     'model_name': 'gpt-4.1-mini',
#     'finish_reason': 'stop',
#     ...
# }

# 3. usage_metadata — uso de tokens (formato estandarizado por LangChain)
print(response.usage_metadata)
# Output esperado:
# {
#     'input_tokens': 15,
#     'output_tokens': 25,
#     'total_tokens': 40
# }

Los tres campos clave de AIMessage

Puedes acceder a los tokens con response.usage_metadata['input_tokens'], ['output_tokens'] y ['total_tokens']. También puedes revisar response.response_metadata.get("finish_reason") para saber por qué el modelo paró.

¿Por qué importa finish_reason?

Si ves finish_reason: "length", significa que tu respuesta fue truncada y necesitas subir max_tokens.

stream(): tokens progresivos

stream() devuelve la respuesta token por token (o en chunks pequeños), permitiéndote mostrar texto progresivamente al usuario.

¿Por qué streaming?

Imagina una respuesta que tarda 5 segundos en generarse:

• Sin streaming (invoke): El usuario ve una pantalla en blanco durante 5 segundos, y luego aparece todo el texto de golpe.
• Con streaming (stream): El usuario empieza a ver texto en ~200ms. La experiencia se siente instantánea aunque el tiempo total sea el mismo.

Ejemplo básico

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

# stream() retorna un iterador de chunks
for chunk in model.stream("Explica qué es un API REST en 3 puntos"):
    print(chunk.content, end="", flush=True)

print()  # Salto de línea al final
# Output esperado (aparece progresivamente):
# Un API REST es:
# 1. Una interfaz de comunicación...
# 2. Usa métodos HTTP estándar...
# 3. Retorna datos en formato JSON...

Cada chunk es un AIMessageChunk — una versión parcial de AIMessage. La clave es end="" y flush=True: end="" evita saltos de línea entre chunks, y flush=True fuerza la impresión inmediata.

¿Qué contiene cada chunk?

Cada chunk es un AIMessageChunk. El primer chunk suele tener content vacío (contiene metadata inicial). Los siguientes traen fragmentos pequeños de texto — a veces una palabra, a veces un par de caracteres. Puedes inspeccionar los chunks con un enumerate:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

for i, chunk in enumerate(model.stream("Di hola")):
    print(f"Chunk {i}: '{chunk.content}'")
    if i > 4:
        break
# Output esperado:
# Chunk 0: ''
# Chunk 1: '¡'
# Chunk 2: 'Hola'
# Chunk 3: '!'
# Chunk 4: ' ¿'

Acumular chunks para obtener la respuesta completa

Si necesitas tanto streaming como la respuesta final (con metadata de tokens), acumula los chunks con el operador +. LangChain combina los AIMessageChunk en un mensaje completo:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

accumulated = None
for chunk in model.stream("Di hola"):
    if accumulated is None:
        accumulated = chunk
    else:
        accumulated = accumulated + chunk

print(f"Content: {accumulated.content}")
print(f"Usage: {accumulated.usage_metadata}")
# Output esperado:
# Content: ¡Hola! ¿En qué puedo ayudarte?
# Usage: {'input_tokens': 9, 'output_tokens': 12, 'total_tokens': 21}

Este patrón es útil cuando quieres streaming para el usuario, pero también necesitas usage_metadata para tracking de costos.

batch(): procesamiento paralelo

batch() envía múltiples inputs al modelo y los procesa en paralelo. En vez de hacer N llamadas secuenciales, hace N llamadas simultáneas.

Úsalo cuando tienes una lista de preguntas que procesar, necesitas generar respuestas para múltiples usuarios, o quieres clasificar/resumir/traducir múltiples textos en una sola operación.

Ejemplo básico

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

questions = [
    "¿Qué es Docker?",
    "¿Qué es Kubernetes?",
    "¿Qué es Terraform?",
]

responses = model.batch(questions)

for question, response in zip(questions, responses):
    print(f"Q: {question}")
    print(f"A: {response.content[:80]}...")
    print()
# Output esperado:
# Q: ¿Qué es Docker?
# A: Docker es una plataforma de contenedores que permite empaquetar aplicaciones...
#
# Q: ¿Qué es Kubernetes?
# A: Kubernetes es un sistema de orquestación de contenedores que automatiza el de...
#
# Q: ¿Qué es Terraform?
# A: Terraform es una herramienta de infraestructura como código (IaC) que permite...

Con 5 inputs, batch() suele ser 3-5x más rápido que un loop de invoke(). Con 20+ inputs, la diferencia es dramática. En el Ejercicio 3 medirás esta diferencia exacta.

Controlar concurrencia con max_concurrency

Por defecto, batch() lanza todas las peticiones en paralelo. Si tienes muchos inputs, puedes limitar la concurrencia para no saturar el rate limit del proveedor:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

questions = [f"¿Cuántos habitantes tiene el país #{i}?" for i in range(20)]

# Máximo 5 llamadas simultáneas
responses = model.batch(
    questions,
    config={"max_concurrency": 5}
)

print(f"Procesados: {len(responses)} inputs")
# Output esperado: Procesados: 20 inputs

Valores recomendados: 5-10 para desarrollo, 3-5 para producción con rate limit bajo, 10-20 para rate limit alto, 1-2 para APIs locales (Ollama).

batch() también acepta listas de mensajes con SystemMessage y HumanMessage, no solo strings — lo verás en el Ejercicio 5.

Versiones async: ainvoke, astream, abatch

Cada modo síncrono tiene su equivalente asíncrono. Usa las versiones async cuando tu aplicación corre en un event loop (FastAPI, notebooks, aplicaciones web).

En una aplicación web (FastAPI, por ejemplo), mientras esperas la respuesta de un LLM, el servidor debería poder atender otras peticiones. Las versiones síncronas bloquean el hilo — las async no.

ainvoke()

import asyncio
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

async def main():
    response = await model.ainvoke("¿Qué es FastAPI?")
    print(response.content)

asyncio.run(main())
# Output esperado: FastAPI es un framework web moderno y de alto rendimiento
# para construir APIs con Python...

astream()

import asyncio
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

async def main():
    async for chunk in model.astream("Explica qué es WebSocket en 2 frases"):
        print(chunk.content, end="", flush=True)
    print()

asyncio.run(main())
# Output esperado (progresivo):
# WebSocket es un protocolo de comunicación bidireccional...

abatch()

abatch() sigue la misma lógica — acepta una lista de inputs y los procesa en paralelo de forma asíncrona:

responses = await model.abatch(["¿Qué es Redis?", "¿Qué es Kafka?"])

astream_events(): eventos semánticos

astream_events() va más allá de streaming de tokens — emite eventos semánticos que describen qué está pasando en la ejecución. Es especialmente útil en pipelines complejos (chains, agentes) donde quieres saber cuándo empieza y termina cada paso.

import asyncio
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

async def main():
    async for event in model.astream_events(
        "Di hola en 3 idiomas",
        version="v2"
    ):
        kind = event["event"]
        if kind == "on_chat_model_start":
            print(">>> Modelo iniciado")
        elif kind == "on_chat_model_stream":
            print(event["data"]["chunk"].content, end="", flush=True)
        elif kind == "on_chat_model_end":
            print("\n>>> Modelo terminado")
            usage = event["data"]["output"].usage_metadata
            if usage:
                print(f">>> Tokens: {usage}")

asyncio.run(main())
# Output esperado:
# >>> Modelo iniciado
# ¡Hola! (español), Hello! (inglés), Bonjour! (francés)
# >>> Modelo terminado
# >>> Tokens: {'input_tokens': 12, 'output_tokens': 18, 'total_tokens': 30}

Los eventos principales son on_chat_model_start, on_chat_model_stream y on_chat_model_end. En pipelines más complejos también verás on_chain_start/end y on_tool_start/end (Módulo 2). Para un modelo solo, astream_events se comporta similar a astream — su poder real aparece cuando necesitas distinguir qué componente genera output.

Comparación: ¿Cuándo usar cada modo?

Reglas rápidas de decisión

• ✅ ¿Es un script o pipeline de datos? → invoke() o batch()
• ✅ ¿El usuario ve la respuesta en una UI? → stream()
• ✅ ¿Tienes muchos inputs que procesar? → batch()
• ✅ ¿Estás en un framework async (FastAPI)? → ainvoke() o astream()
• ✅ ¿Necesitas saber qué pasa dentro de un pipeline? → astream_events()

Conexión con el proyecto

En el Chat Multi-Proveedor con Fallback (Cápsula 08), el streaming es una feature central. El usuario verá tokens aparecer progresivamente, y cuando un proveedor falle, el sistema cambiará al siguiente y empezará a hacer streaming inmediatamente. Usarás usage_metadata para mostrar qué proveedor respondió y cuántos tokens consumió.

El patrón base será:

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

providers = ["openai:gpt-4.1-mini", "anthropic:claude-haiku-4-20250514"]

def chat_with_fallback(question: str):
    for provider_id in providers:
        try:
            model = init_chat_model(provider_id)
            for chunk in model.stream(question):
                print(chunk.content, end="", flush=True)
            print()
            return
        except Exception:
            print(f"\n[Fallback] {provider_id} falló, intentando siguiente...")
    print("[Error] Todos los proveedores fallaron")

chat_with_fallback("¿Qué es Docker?")

Troubleshooting

Problema 1: stream() no muestra nada hasta el final

Causa: Falta flush=True en print(), o el output está siendo buffered. Solución:

# Asegúrate de incluir end="" y flush=True
for chunk in model.stream("pregunta"):
    print(chunk.content, end="", flush=True)

Problema 2: batch() lanza RateLimitError

Causa: Demasiadas peticiones simultáneas exceden el rate limit del proveedor. Solución:

responses = model.batch(
    questions,
    config={"max_concurrency": 3}
)

Problema 3: asyncio.run() lanza "Event loop already running"

Causa: Estás en un entorno que ya tiene un event loop (Jupyter notebook, por ejemplo). Solución: En notebooks, usa await directamente sin asyncio.run():

# En Jupyter notebook
response = await model.ainvoke("Hola")
print(response.content)

# En scripts normales, sí usa asyncio.run()

Problema 4: usage_metadata es None

Causa: Algunos proveedores o modos no incluyen metadata de uso por defecto. En streaming, la metadata suele llegar en el último chunk. Solución: Acumula chunks con + para obtener la metadata completa:

accumulated = None
for chunk in model.stream("pregunta"):
    accumulated = chunk if accumulated is None else accumulated + chunk
print(accumulated.usage_metadata)

Problema 5: stream() retorna chunks vacíos al inicio

Causa: Es comportamiento normal. El primer chunk suele contener metadata inicial sin texto. Solución: Simplemente ignóralo — print(chunk.content, end="") ya lo maneja porque imprime un string vacío sin efecto visual.

Ejercicios

Ejercicio 1: Explorar AIMessage (Fácil)

Llama a un modelo con invoke() e imprime los tres campos clave de la respuesta: content, response_metadata y usage_metadata. Calcula el costo aproximado sabiendo que GPT-4.1 Mini cuesta $0.40 por millón de input tokens y $1.60 por millón de output tokens.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")
response = model.invoke("¿Qué es un microservicio? Responde en 2 frases.")

print(f"Content: {response.content}")
print(f"Response metadata: {response.response_metadata}")
print(f"Usage metadata: {response.usage_metadata}")

if response.usage_metadata:
    input_cost = response.usage_metadata["input_tokens"] * 0.40 / 1_000_000
    output_cost = response.usage_metadata["output_tokens"] * 1.60 / 1_000_000
    total_cost = input_cost + output_cost
    print(f"\nCosto estimado: ${total_cost:.6f}")
# Output esperado:
# Content: Un microservicio es un patrón arquitectónico...
# Response metadata: {'token_usage': {...}, 'model_name': 'gpt-4.1-mini', ...}
# Usage metadata: {'input_tokens': 18, 'output_tokens': 45, 'total_tokens': 63}
#
# Costo estimado: $0.000079

Ejercicio 2: Stream con contador (Fácil)

Usa stream() para mostrar la respuesta progresivamente y al final imprime cuántos chunks recibiste y el texto completo acumulado.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

chunk_count = 0
full_text = ""

for chunk in model.stream("Nombra 5 lenguajes de programación populares"):
    full_text += chunk.content
    chunk_count += 1
    print(chunk.content, end="", flush=True)

print(f"\n\n--- Estadísticas ---")
print(f"Chunks recibidos: {chunk_count}")
print(f"Caracteres totales: {len(full_text)}")
print(f"Texto completo: {full_text}")
# Output esperado:
# 1. Python
# 2. JavaScript
# 3. Java
# 4. TypeScript
# 5. Go
#
# --- Estadísticas ---
# Chunks recibidos: 35
# Caracteres totales: 85
# Texto completo: 1. Python...

Ejercicio 3: batch vs secuencial (Medio)

Compara el tiempo de ejecución entre procesar 5 preguntas con invoke() en un loop y con batch(). Imprime el speedup.

import time
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

questions = [
    "¿Qué es Python?",
    "¿Qué es JavaScript?",
    "¿Qué es Rust?",
    "¿Qué es Go?",
    "¿Qué es TypeScript?",
]

# Secuencial
start = time.time()
sequential_results = []
for q in questions:
    r = model.invoke(q)
    sequential_results.append(r)
time_seq = time.time() - start

# Batch
start = time.time()
batch_results = model.batch(questions)
time_batch = time.time() - start

print(f"Secuencial: {time_seq:.2f}s")
print(f"Batch:      {time_batch:.2f}s")
print(f"Speedup:    {time_seq / time_batch:.1f}x")

# Verificar que ambos dieron respuestas
for q, r in zip(questions, batch_results):
    print(f"  {q} → {r.content[:50]}...")
# Output esperado:
# Secuencial: 6.10s
# Batch:      1.80s
# Speedup:    3.4x
#   ¿Qué es Python? → Python es un lenguaje de programación de alto nivel...
#   ...

Ejercicio 4: Streaming con formato (Medio)

Crea una función stream_with_border(question) que muestre la respuesta con streaming dentro de un recuadro visual. Imprime una línea superior antes de empezar, stream del contenido, y una línea inferior al terminar. Incluye el conteo de tokens al final.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

def stream_with_border(question: str):
    print(f"\n{'=' * 60}")
    print(f"Pregunta: {question}")
    print(f"{'-' * 60}")

    accumulated = None
    for chunk in model.stream(question):
        print(chunk.content, end="", flush=True)
        if accumulated is None:
            accumulated = chunk
        else:
            accumulated = accumulated + chunk

    print(f"\n{'-' * 60}")
    if accumulated and accumulated.usage_metadata:
        tokens = accumulated.usage_metadata
        print(f"Tokens: {tokens['input_tokens']} in / {tokens['output_tokens']} out")
    print(f"{'=' * 60}")

stream_with_border("¿Qué es un container en 2 frases?")
# Output esperado:
# ============================================================
# Pregunta: ¿Qué es un container en 2 frases?
# ------------------------------------------------------------
# Un container es un paquete ligero y portable que contiene
# una aplicación y todas sus dependencias...
# ------------------------------------------------------------
# Tokens: 15 in / 35 out
# ============================================================

Ejercicio 5: Traductor batch (Medio)

Crea un traductor que reciba un texto y una lista de idiomas destino, y use batch() para traducir el texto a todos los idiomas en paralelo. Usa mensajes con system prompt para cada traducción.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage

model = init_chat_model("openai:gpt-4.1-mini")

def translate_batch(text: str, target_languages: list[str]) -> dict[str, str]:
    """Traduce un texto a múltiples idiomas en paralelo."""
    inputs = [
        [
            SystemMessage(content=f"Traduce el siguiente texto a {lang}. "
                                  f"Responde SOLO con la traducción, sin explicaciones."),
            HumanMessage(content=text)
        ]
        for lang in target_languages
    ]

    responses = model.batch(inputs)

    return {
        lang: response.content
        for lang, response in zip(target_languages, responses)
    }

text = "La inteligencia artificial está transformando el mundo"
languages = ["inglés", "francés", "japonés", "portugués"]

translations = translate_batch(text, languages)
for lang, translation in translations.items():
    print(f"{lang:>12}: {translation}")
# Output esperado:
#       inglés: Artificial intelligence is transforming the world
#      francés: L'intelligence artificielle transforme le monde
#     japonés: 人工知能が世界を変革している
#   portugués: A inteligência artificial está transformando o mundo

Ejercicio 6: async stream con gather (Difícil)

Crea una función async astream_and_collect(model, question, label) que haga streaming imprimiendo cada chunk con un prefijo [label], acumule los chunks, y retorne un dict con text, chunks (cantidad), y usage. Ejecuta 2 preguntas en paralelo con asyncio.gather.

import asyncio
from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4.1-mini")

async def astream_and_collect(model, question: str, label: str) -> dict:
    accumulated = None
    chunk_count = 0
    async for chunk in model.astream(question):
        if chunk.content:
            print(f"[{label}] {chunk.content}", end="", flush=True)
        chunk_count += 1
        accumulated = chunk if accumulated is None else accumulated + chunk
    print()
    return {
        "text": accumulated.content if accumulated else "",
        "chunks": chunk_count,
        "usage": accumulated.usage_metadata if accumulated else None,
    }

async def main():
    results = await asyncio.gather(
        astream_and_collect(model, "¿Qué es Redis?", "Q1"),
        astream_and_collect(model, "¿Qué es Kafka?", "Q2"),
    )
    print("\n--- Resumen ---")
    for i, r in enumerate(results, 1):
        print(f"Q{i}: {r['chunks']} chunks, {len(r['text'])} chars, usage={r['usage']}")

asyncio.run(main())
# Output esperado (intercalado porque corren en paralelo):
# [Q1] Redis es una base...[Q2] Kafka es una plataforma...
#
# --- Resumen ---
# Q1: 28 chunks, 120 chars, usage={'input_tokens': 10, 'output_tokens': 35, ...}
# Q2: 32 chunks, 140 chars, usage={'input_tokens': 10, 'output_tokens': 40, ...}

Resumen

En esta cápsula aprendiste:

• invoke() envía un mensaje y espera la respuesta completa — ideal para scripts y procesamiento backend
• AIMessage contiene tres campos clave: content (texto), response_metadata (info del proveedor), usage_metadata (tokens usados)
• stream() retorna la respuesta token por token — esencial para UIs de chat donde la experiencia del usuario importa
• batch() procesa múltiples inputs en paralelo — significativamente más rápido que un loop de invoke()
• Las versiones async (ainvoke, astream, abatch) son necesarias en frameworks como FastAPI
• astream_events() emite eventos semánticos — su poder real aparece en pipelines y agentes
• Usa end="" y flush=True para streaming, y max_concurrency para controlar batch
• Acumular chunks con + te da la metadata completa después del streaming

Próxima cápsula: Structured Output — aprenderás a obtener respuestas en formatos tipados (Pydantic, TypedDict, JSON Schema) en vez de texto libre, para que tu código nunca tenga que parsear strings.

Recursos adicionales

1. Chat Models: invoke, stream, batch - Documentación oficial de los modos de ejecución
2. How to stream chat model responses - Guía de streaming
3. How to stream events from a Runnable - astream_events en detalle
4. AIMessage API Reference - Referencia completa de AIMessage
5. Async Programming with LangChain - Guía de async/await
6. How to batch calls to a Runnable - Batch processing y paralelismo
7. LangChain Runnable Interface - Interface unificada que habilita invoke/stream/batch
8. OpenAI Streaming Guide - Cómo funciona streaming a nivel de API

Módulo 1 — LangChain & LangGraph: From Chains to Agents