Despliegue

Despliega aplicaciones LangGraph con LangGraph Cloud/Platform, configura APIs del grafo, gestiona opciones de despliegue y escala para producción.

Desplegar aplicaciones LangGraph en producción requiere comprender LangGraph Cloud (Platform), la configuración de la API, la autenticación, el escalado y las consideraciones operativas.

Visión General de LangGraph Cloud/Platform

LangGraph Platform ofrece una infraestructura de despliegue gestionada para aplicaciones LangGraph:

Característica	Descripción
API gestionada	Generación automática de API REST a partir de tu grafo
Escalado	Autoescalado según la demanda
Persistencia	PostgreSQL gestionado para checkpointing
Autenticación	Compatibilidad con clave de API y OAuth
Monitoreo	Integración con LangSmith integrada
Streaming	Server-Sent Events (SSE) para salida en tiempo real

Arquitectura de Despliegue

┌─────────────┐     ┌──────────────┐     ┌─────────────┐
│   Client    │────▶│ LangGraph    │────▶│   LLM       │
│  (App/Web)  │     │  Platform    │     │   APIs      │
└─────────────┘     └──────────────┘     └─────────────┘
                           │
                           ▼
                    ┌──────────────┐     ┌─────────────┐
                    │  PostgreSQL  │     │  LangSmith   │
                    │ (State/Pers.)│     │ (Observ.)    │
                    └──────────────┘     └─────────────┘

Configuración de la API del Grafo

Crea un archivo de configuración langgraph.json en la raíz de tu proyecto:

json

{
  "name": "my-agent",
  "version": "1.0.0",
  "graphs": {
    "agent": "./src/agent.py:graph"
  },
  "dependencies": {
    "pip": ["langchain-openai", "tavily-python"],
    "python": "3.11"
  },
  "env": {
    "OPENAI_API_KEY": "sk-...",
    "TAVILY_API_KEY": "tvly-..."
  },
  "checkpointer": {
    "type": "postgres",
    "url": "postgresql://user:pass@host:5432/langgraph"
  }
}

📌Importante

El campo graphs asigna nombres de endpoints a rutas de importación de Python. El valor "./src/agent.py:graph" significa "importar graph desde src/agent.py".

Despliegue con Dockerfile

Para despliegues personalizados, usa la imagen oficial de LangGraph para Docker:

docker

FROM langchain/langgraph:latest

WORKDIR /app
COPY pyproject.toml .
RUN pip install -e .

COPY . .

# The graph is auto-detected from langgraph.json
EXPOSE 8080
CMD ["langgraph", "serve"]

Opciones de Despliegue

Opción 1: LangGraph Cloud (gestionado)

bash

# Install CLI
pip install langgraph-cli

# Deploy to LangGraph Cloud
langgraph deploy --project my-agent

# Set environment variables
langgraph secrets set OPENAI_API_KEY sk-...

Opción 2: Docker autoalojado

bash

# Build the Docker image
docker build -t my-agent .

# Run with environment variables
docker run -p 8080:8080 \
  -e OPENAI_API_KEY=sk-... \
  -e DATABASE_URL=postgresql://... \
  my-agent

Opción 3: Kubernetes

yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  name: langgraph-agent
spec:
  replicas: 3
  selector:
    matchLabels:
      app: langgraph-agent
  template:
    metadata:
      labels:
        app: langgraph-agent
    spec:
      containers:
      - name: agent
        image: my-agent:latest
        ports:
        - containerPort: 8080
        env:
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: api-keys
              key: openai-key
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: db-credentials
              key: url

ℹ️Nota

En producción, usa secretos de Kubernetes o un servicio de vault para las claves de API. Nunca dejes secretos fijos en el código ni en los archivos de configuración.

Endpoints de la API

LangGraph Platform genera estos endpoints REST:

Endpoint	Method	Descripción
`/threads`	POST	Crea un nuevo hilo de conversación
`/threads/{id}/runs`	POST	Invoca el grafo en un hilo
`/threads/{id}/state`	GET	Obtiene el estado del hilo
`/threads/{id}/state`	PATCH	Actualiza el estado del hilo
`/threads/{id}/runs/{run_id}/stream`	GET	Transmite los eventos de ejecución

Ejemplo de Cliente

python

import requests

API_URL = "https://my-agent.langgraph.app"

# Create a thread
resp = requests.post(f"{API_URL}/threads", json={})
thread_id = resp.json()["thread_id"]

# Invoke the graph
resp = requests.post(
    f"{API_URL}/threads/{thread_id}/runs",
    json={"input": {"query": "What is LangGraph?"}}
)
print(resp.json())

# Get state
state = requests.get(f"{API_URL}/threads/{thread_id}/state")
print(state.json())

Streaming desde la API Desplegada

python

# Stream events via SSE
import json
import requests

with requests.post(
    f"{API_URL}/threads/{thread_id}/runs/stream",
    json={"input": {"query": "Write a poem"}},
    stream=True
) as resp:
    for line in resp.iter_lines():
        if line:
            event = json.loads(line.decode("utf-8").removeprefix("data: "))
            print(event)

Autenticación

python

# API Key authentication
headers = {
    "X-Api-Key": "lgv2_...",
    "Content-Type": "application/json"
}

resp = requests.post(
    f"{API_URL}/threads",
    headers=headers,
    json={}
)

⚠️Advertencia

Rota las claves de API regularmente. Usa claves de corta duración en producción y revoca de inmediato las claves comprometidas.

Consideraciones de Escalado

Dimension	Estrategia
Usuarios concurrentes	Escalado horizontal (más réplicas)
Grafos de larga duración	Ejecución asíncrona con callbacks de webhook
Base de datos	Agrupación de conexiones PostgreSQL (PgBouncer)
Límites de tasa del LLM	Encola solicitudes e implementa reintentos con backoff
Memoria	Monitorea el tamaño de los checkpoints y poda los antiguos
Cold starts	Mantén las réplicas mínimas activas

Conexión con Base de Datos Pooling

python

from langgraph.checkpoint.postgres import PostgresSaver

# Use connection pooling for production
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@host:5432/langgraph",
    pool_size=10,
    max_overflow=20
)

Verificaciones de Salud

python

from fastapi import FastAPI
from langgraph.graph import StateGraph

app = FastAPI()
graph_app = builder.compile(checkpointer=checkpointer)

@app.get("/health")
def health():
    return {"status": "ok", "graph": "ready"}

@app.post("/invoke")
def invoke(input: dict):
    result = graph_app.invoke(input)
    return result