Implantação

Implante aplicações LangGraph com LangGraph Cloud/Platform, configure APIs do grafo, gerencie opções de implantação e escale para produção.

Implantar aplicações LangGraph em produção exige entender o LangGraph Cloud (Platform), a configuração de API, autenticação, escalonamento e considerações operacionais.

Visão Geral do LangGraph Cloud/Platform

O LangGraph Platform oferece uma infraestrutura de implantação gerenciada para aplicações LangGraph:

Funcionalidade	Descrição
API gerenciada	Geração automática de API REST a partir do seu grafo
Escalonamento	Autoescalonamento com base na demanda
Persistência	PostgreSQL gerenciado para checkpointing
Autenticação	Suporte para chave de API e OAuth
Monitoramento	Integração com LangSmith nativa
Streaming	Server-Sent Events (SSE) para saída em tempo real

Arquitetura de Implantação

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

Configuração da API do Grafo

Crie um arquivo de configuração langgraph.json na raiz do seu projeto:

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

O campo graphs mapeia nomes de endpoints para caminhos de importação em Python. O valor "./src/agent.py:graph" significa "importar graph de src/agent.py".

Implantação com Dockerfile

Para implantações personalizadas, use a imagem oficial do 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"]

Opções de Implantação

Opção 1: LangGraph Cloud (gerenciado)

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

Opção 2: Docker auto-hospedado

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

Opção 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

Em produção, use secrets do Kubernetes ou um serviço de vault para as chaves de API. Nunca deixe secrets fixos no código ou nos arquivos de configuração.

Endpoints da API

O LangGraph Platform gera estes endpoints REST:

Endpoint	Method	Descrição
`/threads`	POST	Cria um novo thread de conversação
`/threads/{id}/runs`	POST	Invoca o grafo em um thread
`/threads/{id}/state`	GET	Obtém o estado do thread
`/threads/{id}/state`	PATCH	Atualiza o estado do thread
`/threads/{id}/runs/{run_id}/stream`	GET	Faz streaming dos eventos de execução

Exemplo 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 da API Implantada

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)

Autenticação

python

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

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

⚠️Aviso

Gire as chaves de API regularmente. Use chaves de curta duração em produção e revogue imediatamente as chaves comprometidas.

Considerações de Escala

Dimension	Estratégia
Usuários concorrentes	Escalonamento horizontal (mais réplicas)
Grafos de longa duração	Execução assíncrona com callbacks de webhook
Banco de dados	Pool de conexões PostgreSQL (PgBouncer)
Limites de taxa do LLM	Enfileire solicitações e implemente retry com backoff
Memória	Monitore o tamanho dos checkpoints e descarte os antigos
Cold starts	Mantenha as réplicas mínimas em quente

Conexão com Banco de Dados 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
)

Verificações de Saúde

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