LangGraph para Agents: Construye IA Agéntica con State Machines (Guía Práctica 2025)

¿Quieres construir AI agents que ejecuten workflows complejos en producción? LangGraph es el framework líder para crear sistemas agénticos con state machines, usado por Klarna, Uber y LinkedIn. En esta guía aprenderás a construir agents con grafos stateful, control de flujo avanzado, y arquitecturas que escalan en producción.

¿Qué es LangGraph y Por Qué Lo Necesitas?

LangGraph es un framework de orquestación low-level diseñado para construir, gestionar y deployar sistemas de AI agents sofisticados. Desarrollado por el equipo de LangChain, LangGraph lleva los agents más allá de simples cadenas lineales hacia grafos stateful con control de flujo completo.

El Problema Que Resuelve

Cuando construyes AI agents en producción, los frameworks tradicionales te limitan:

• Workflows lineales: Las cadenas (chains) solo van en una dirección, sin loops ni branching
• Sin persistencia de estado: Cada invocación empieza desde cero, sin memoria entre pasos
• Control de flujo limitado: No puedes implementar lógica condicional compleja o ciclos
• Falta de observability: Difícil debuggear qué hizo el agent en cada paso

LangGraph resuelve todo esto con arquitectura de grafos stateful:

┌─────────────────────────────────────────────────────┐
│          Problema              Solución LangGraph   │
├─────────────────────────────────────────────────────┤
│ Workflows lineales    →   Grafos cíclicos          │
│ Sin estado           →   State machines stateful   │
│ Flujo fijo           →   Conditional routing       │
│ No debuggeable       →   Checkpoints + LangSmith   │
│ Falla sin recuperar  →   Durable execution         │
└─────────────────────────────────────────────────────┘

Casos de Uso Reales

Empresas usando LangGraph en producción:

Paper oficial: LangGraph Documentation

State Machines: El Corazón de LangGraph

LangGraph modela agents como máquinas de estado (state machines), donde cada nodo procesa y actualiza un estado compartido.

Anatomía de un State Machine

┌──────────────────────────────────────────────┐
│          State Machine Workflow              │
├──────────────────────────────────────────────┤
│                                              │
│  START                                       │
│    │                                         │
│    ▼                                         │
│  ┌─────────────┐                            │
│  │  Node A     │  ──→ update state          │
│  │  (analyze)  │                            │
│  └─────────────┘                            │
│    │                                         │
│    ▼                                         │
│  ┌─────────────┐                            │
│  │  Node B     │  ──→ conditional           │
│  │  (decide)   │                            │
│  └─────────────┘                            │
│    │                                         │
│    ├──→ condition_true ──→ Node C           │
│    └──→ condition_false ──→ Node D          │
│                              │               │
│                              ▼               │
│                            END               │
└──────────────────────────────────────────────┘

Componentes Clave

1. State (Estado)
El estado es un objeto compartido que persiste a través del grafo. Se define como TypedDict:

from typing import TypedDict, List

class AgentState(TypedDict):
    messages: List[str]      # Historial de mensajes
    current_step: str        # Paso actual
    data: dict              # Datos procesados
    iterations: int         # Contador de iteraciones

2. Nodes (Nodos)
Cada nodo es una función que recibe el estado, lo procesa, y devuelve actualizaciones:

def analyze_node(state: AgentState) -> AgentState:
    """Analiza los datos del estado"""
    messages = state["messages"]
    # Procesar datos
    result = analyze_data(messages)

    return {
        "data": result,
        "current_step": "analysis_complete"
    }

3. Edges (Aristas)
Las aristas definen el flujo entre nodos:

• Normal edges: Transiciones directas e incondicionales
• Conditional edges: Branching basado en el estado

def route_decision(state: AgentState) -> str:
    """Decide el siguiente nodo basándose en el estado"""
    if state["data"]["confidence"] > 0.8:
        return "high_confidence_path"
    else:
        return "low_confidence_path"

Tu Primer Agent con LangGraph: Paso a Paso

Vamos a construir un agent de investigación que analiza preguntas, busca información, y genera respuestas.

Paso 1: Instalación

# Instalar LangGraph y dependencias
pip install langgraph langchain langchain-openai

# Variables de entorno
export OPENAI_API_KEY="tu-api-key"

Paso 2: Definir el Estado

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import HumanMessage, AIMessage

class ResearchState(TypedDict):
    """Estado del agent de investigación"""
    question: str                    # Pregunta original
    search_queries: List[str]        # Queries generadas
    search_results: List[dict]       # Resultados de búsqueda
    analysis: str                    # Análisis de resultados
    final_answer: str               # Respuesta final
    iterations: int                 # Contador de iteraciones

Paso 3: Crear Nodos de Procesamiento

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4", temperature=0)

def generate_queries_node(state: ResearchState) -> dict:
    """Genera queries de búsqueda optimizadas"""
    question = state["question"]

    prompt = f"""
    Genera 3 queries de búsqueda optimizadas para responder:
    {question}

    Devuelve solo las queries, una por línea.
    """

    response = llm.invoke(prompt)
    queries = response.content.strip().split("\n")

    return {
        "search_queries": queries,
        "iterations": state.get("iterations", 0) + 1
    }

def search_node(state: ResearchState) -> dict:
    """Simula búsqueda (en producción: usar Tavily, SerpAPI, etc.)"""
    queries = state["search_queries"]

    # Aquí irían llamadas reales a APIs de búsqueda
    results = []
    for query in queries:
        results.append({
            "query": query,
            "snippet": f"Resultado simulado para: {query}",
            "url": f"https://example.com/{query.replace(' ', '-')}"
        })

    return {"search_results": results}

def analyze_node(state: ResearchState) -> dict:
    """Analiza resultados y genera respuesta"""
    question = state["question"]
    results = state["search_results"]

    results_text = "\n".join([
        f"- {r['snippet']}" for r in results
    ])

    prompt = f"""
    Pregunta: {question}

    Resultados de búsqueda:
    {results_text}

    Genera una respuesta completa y precisa basada en estos resultados.
    """

    response = llm.invoke(prompt)

    return {
        "analysis": results_text,
        "final_answer": response.content
    }

Paso 4: Construir el Grafo

# Crear el grafo
workflow = StateGraph(ResearchState)

# Añadir nodos
workflow.add_node("generate_queries", generate_queries_node)
workflow.add_node("search", search_node)
workflow.add_node("analyze", analyze_node)

# Definir flujo con aristas
workflow.add_edge(START, "generate_queries")
workflow.add_edge("generate_queries", "search")
workflow.add_edge("search", "analyze")
workflow.add_edge("analyze", END)

# Compilar el grafo
app = workflow.compile()

Paso 5: Ejecutar el Agent

# Invocar el agent
result = app.invoke({
    "question": "¿Cuáles son las mejores prácticas para deployar LLMs en producción?",
    "iterations": 0
})

print("=== RESPUESTA FINAL ===")
print(result["final_answer"])
print(f"\nIteraciones: {result['iterations']}")
print(f"Queries generadas: {len(result['search_queries'])}")

Output esperado:

=== RESPUESTA FINAL ===
Para deployar LLMs en producción se recomienda:
1. Usar frameworks de inferencia optimizados como vLLM o TensorRT-LLM
2. Implementar caching de embeddings y prompts frecuentes
3. Configurar rate limiting y monitoring de costos
4. Usar quantization (GPTQ, AWQ) para reducir VRAM
5. Implementar fallbacks y retry logic con timeouts

Iteraciones: 1
Queries generadas: 3

Conditional Routing: Workflows Dinámicos

Lo potente de LangGraph es el routing condicional: el agent decide dinámicamente qué nodo ejecutar basándose en el estado.

Ejemplo: Agent con Auto-Corrección

from typing import Literal

class CodeReviewState(TypedDict):
    code: str
    issues: List[str]
    fixed_code: str
    attempts: int
    max_attempts: int

def review_code_node(state: CodeReviewState) -> dict:
    """Revisa el código y detecta issues"""
    code = state["code"]

    # En producción: usar LLM para análisis real
    issues = []
    if "import os" not in code and "subprocess" in code:
        issues.append("Falta import os")
    if "try:" not in code:
        issues.append("Falta manejo de errores")

    return {
        "issues": issues,
        "attempts": state["attempts"] + 1
    }

def fix_code_node(state: CodeReviewState) -> dict:
    """Intenta corregir el código"""
    code = state["code"]
    issues = state["issues"]

    # Simular corrección (en producción: LLM)
    fixed = code
    for issue in issues:
        if "import" in issue:
            fixed = "import os\n" + fixed
        if "try:" in issue:
            fixed = f"try:\n    {fixed}\nexcept Exception as e:\n    print(e)"

    return {"fixed_code": fixed, "code": fixed}

def should_continue(state: CodeReviewState) -> Literal["fix_code", "end"]:
    """Decide si continuar iterando o terminar"""
    if len(state["issues"]) == 0:
        return "end"
    elif state["attempts"] >= state["max_attempts"]:
        return "end"
    else:
        return "fix_code"

# Construir grafo con routing condicional
workflow = StateGraph(CodeReviewState)

workflow.add_node("review", review_code_node)
workflow.add_node("fix_code", fix_code_node)

workflow.add_edge(START, "review")
workflow.add_conditional_edges(
    "review",
    should_continue,
    {
        "fix_code": "fix_code",
        "end": END
    }
)
workflow.add_edge("fix_code", "review")  # Loop back

app = workflow.compile()

# Ejecutar
result = app.invoke({
    "code": "subprocess.run(['ls'])",
    "issues": [],
    "attempts": 0,
    "max_attempts": 3
})

print(f"Código corregido después de {result['attempts']} intentos:")
print(result['fixed_code'])

Human-in-the-Loop: Control Manual en Workflows

LangGraph permite pausar la ejecución para que un humano revise o modifique el estado.

Implementación con Checkpoints

from langgraph.checkpoint.sqlite import SqliteSaver

# Configurar persistencia
memory = SqliteSaver.from_conn_string(":memory:")

class ApprovalState(TypedDict):
    content: str
    approved: bool
    feedback: str

def generate_content_node(state: ApprovalState) -> dict:
    """Genera contenido que requiere aprobación"""
    content = "Este es el contenido generado automáticamente..."
    return {"content": content, "approved": False}

def approval_node(state: ApprovalState) -> dict:
    """Nodo de aprobación humana"""
    # En producción: mostrar UI para aprobación
    print(f"Contenido generado:\n{state['content']}\n")
    print("Esperando aprobación humana...")

    # Aquí el workflow se pausaría esperando input
    return {"approved": True}  # Simulado

def publish_node(state: ApprovalState) -> dict:
    """Publica solo si está aprobado"""
    if state["approved"]:
        print("✅ Contenido publicado!")
    return state

# Grafo con checkpointing
workflow = StateGraph(ApprovalState)
workflow.add_node("generate", generate_content_node)
workflow.add_node("approval", approval_node)
workflow.add_node("publish", publish_node)

workflow.add_edge(START, "generate")
workflow.add_edge("generate", "approval")
workflow.add_edge("approval", "publish")
workflow.add_edge("publish", END)

# Compilar con checkpointer
app = workflow.compile(checkpointer=memory)

# Ejecutar con thread_id para pausar/resumir
config = {"configurable": {"thread_id": "approval_flow_1"}}
result = app.invoke({"content": "", "approved": False}, config)

Multi-Agent Systems: Coordinación de Agents

LangGraph permite orquestar múltiples agents que colaboran en tareas complejas.

Ejemplo: Sistema de Investigación Multi-Agent

class MultiAgentState(TypedDict):
    task: str
    researcher_output: str
    writer_output: str
    reviewer_output: str
    final_report: str

def researcher_agent(state: MultiAgentState) -> dict:
    """Agent especializado en investigación"""
    task = state["task"]
    # Simular investigación profunda
    research = f"Investigación completa sobre: {task}"
    return {"researcher_output": research}

def writer_agent(state: MultiAgentState) -> dict:
    """Agent especializado en escritura"""
    research = state["researcher_output"]
    # Generar draft basado en investigación
    draft = f"Draft basado en:\n{research}"
    return {"writer_output": draft}

def reviewer_agent(state: MultiAgentState) -> dict:
    """Agent especializado en revisión"""
    draft = state["writer_output"]
    # Revisar y mejorar
    review = f"Versión revisada de:\n{draft}"
    return {"reviewer_output": review, "final_report": review}

# Grafo multi-agent
workflow = StateGraph(MultiAgentState)

workflow.add_node("researcher", researcher_agent)
workflow.add_node("writer", writer_agent)
workflow.add_node("reviewer", reviewer_agent)

# Pipeline secuencial
workflow.add_edge(START, "researcher")
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "reviewer")
workflow.add_edge("reviewer", END)

app = workflow.compile()

result = app.invoke({
    "task": "Analizar tendencias de IA en 2025"
})

print(result["final_report"])

LangGraph vs Otros Frameworks: Comparativa 2025

Tabla Comparativa Detallada

* Según benchmarks de CrewAI

Cuándo Usar LangGraph

✅ Usa LangGraph cuando:

• Necesitas control total sobre el flujo de ejecución
• Workflows complejos con branches, loops, y conditional logic
• Requieres persistencia de estado entre ejecuciones
• Deployando en producción con alta escala (ej: Klarna con 85M usuarios)
• Necesitas human-in-the-loop workflows
• Debugging y observability son críticos

❌ NO uses LangGraph si:

• Quieres prototipado ultra-rápido (usa CrewAI)
• Workflows simples lineales (usa LangChain Chains)
• No tienes experiencia con grafos/state machines
• Prefieres abstracciones de alto nivel

Patrones de Arquitectura Avanzados

1. Reflexión y Auto-Mejora

Agent que itera sobre su propia salida hasta alcanzar calidad:

class ReflectionState(TypedDict):
    task: str
    draft: str
    critique: str
    iterations: int
    max_iterations: int
    quality_score: float

def generate_draft(state: ReflectionState) -> dict:
    """Genera draft inicial"""
    return {
        "draft": f"Draft para: {state['task']}",
        "iterations": state["iterations"] + 1
    }

def critique_draft(state: ReflectionState) -> dict:
    """Critica el draft"""
    # En producción: LLM evalúa calidad
    score = 0.7 + (state["iterations"] * 0.1)  # Simular mejora
    critique = f"Score: {score}, necesita mejorar X, Y, Z"
    return {"critique": critique, "quality_score": score}

def should_continue(state: ReflectionState) -> Literal["generate", "end"]:
    if state["quality_score"] >= 0.9:
        return "end"
    elif state["iterations"] >= state["max_iterations"]:
        return "end"
    else:
        return "generate"

workflow = StateGraph(ReflectionState)
workflow.add_node("generate", generate_draft)
workflow.add_node("critique", critique_draft)

workflow.add_edge(START, "generate")
workflow.add_edge("generate", "critique")
workflow.add_conditional_edges(
    "critique",
    should_continue,
    {"generate": "generate", "end": END}
)

app = workflow.compile()

2. Subgrafos y Composición

Puedes componer grafos dentro de otros grafos:

def create_analysis_subgraph():
    """Subgrafo para análisis de datos"""
    subgraph = StateGraph(AnalysisState)
    subgraph.add_node("load_data", load_data_node)
    subgraph.add_node("clean_data", clean_data_node)
    subgraph.add_node("analyze", analyze_node)
    # ... configurar edges
    return subgraph.compile()

# Grafo principal que usa el subgrafo
main_workflow = StateGraph(MainState)
analysis_graph = create_analysis_subgraph()

main_workflow.add_node("preprocessing", preprocessing_node)
main_workflow.add_node("analysis", analysis_graph)  # ← Subgrafo como nodo
main_workflow.add_node("reporting", reporting_node)

3. Parallel Execution

Ejecutar múltiples nodos en paralelo:

from langgraph.graph import MessageGraph

def parallel_tasks_pattern(state: State) -> dict:
    """Pattern para ejecutar tareas en paralelo"""
    # LangGraph ejecuta nodos sin dependencias en paralelo
    return state

# Configurar nodos paralelos
workflow = StateGraph(State)
workflow.add_node("task_a", task_a_node)
workflow.add_node("task_b", task_b_node)
workflow.add_node("task_c", task_c_node)
workflow.add_node("merge", merge_results_node)

# Todos desde START en paralelo
workflow.add_edge(START, "task_a")
workflow.add_edge(START, "task_b")
workflow.add_edge(START, "task_c")

# Merge results
workflow.add_edge("task_a", "merge")
workflow.add_edge("task_b", "merge")
workflow.add_edge("task_c", "merge")
workflow.add_edge("merge", END)

Deployment en Producción

Opción 1: LangGraph Cloud

# Configurar para LangGraph Cloud
from langgraph.graph import StateGraph
from langgraph.checkpoint.postgres import PostgresSaver

# Usar PostgreSQL para checkpoints en producción
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@host:5432/db"
)

app = workflow.compile(checkpointer=checkpointer)

# Deploy a LangGraph Cloud
# Ver: https://langchain-ai.github.io/langgraph/cloud/

Opción 2: Docker Self-Hosted

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000"]

# api.py - FastAPI wrapper
from fastapi import FastAPI, BackgroundTasks
from langgraph.graph import StateGraph

app = FastAPI()
workflow = ...  # Tu grafo compilado

@app.post("/invoke")
async def invoke_agent(request: dict):
    result = workflow.invoke(request)
    return result

@app.post("/stream")
async def stream_agent(request: dict):
    for chunk in workflow.stream(request):
        yield chunk

Opción 3: Kubernetes

# k8s-deployment.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: your-registry/langgraph-agent:latest
        ports:
        - containerPort: 8000
        env:
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: openai-secret
              key: api-key
        resources:
          requests:
            memory: "2Gi"
            cpu: "1"
          limits:
            memory: "4Gi"
            cpu: "2"

Monitoring y Observability con LangSmith

LangGraph se integra nativamente con LangSmith para debugging y observability.

Configuración Básica

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "tu-langsmith-key"
os.environ["LANGCHAIN_PROJECT"] = "mi-agent-produccion"

# Ejecutar agent - todas las trazas van a LangSmith
result = app.invoke({"question": "..."})

Métricas Clave a Monitorear

# Custom metrics con callbacks
from langchain.callbacks import LangChainTracer

class ProductionMetrics(LangChainTracer):
    def on_chain_start(self, serialized, inputs, **kwargs):
        # Log inicio de chain
        print(f"Chain started: {serialized['name']}")

    def on_chain_end(self, outputs, **kwargs):
        # Log duración, tokens, costos
        duration = kwargs.get("duration_ms", 0)
        print(f"Chain completed in {duration}ms")

app = workflow.compile()
result = app.invoke(
    {"question": "..."},
    config={"callbacks": [ProductionMetrics()]}
)

Troubleshooting y Best Practices

Errores Comunes

1. Infinite Loops

# ❌ MAL: Sin límite de iteraciones
def should_continue(state):
    if state["done"]:
        return "end"
    return "continue"  # Puede loop infinito

# ✅ BIEN: Límite de seguridad
def should_continue(state):
    if state["done"] or state["iterations"] > 10:
        return "end"
    return "continue"

2. State No Actualizado

# ❌ MAL: Mutar estado directamente
def bad_node(state):
    state["value"] = 123  # No funciona
    return state

# ✅ BIEN: Retornar dict con updates
def good_node(state):
    return {"value": 123}  # LangGraph mergea

3. Memory Leaks con Checkpoints

# ✅ Limpiar checkpoints antiguos
from langgraph.checkpoint import CheckpointManager

manager = CheckpointManager(checkpointer)
manager.cleanup_old_checkpoints(days=7)

Best Practices

1. Diseño de Estado Mínimo

# ✅ Solo lo esencial
class GoodState(TypedDict):
    query: str
    result: str

# ❌ Demasiado complejo
class BadState(TypedDict):
    query: str
    intermediate_1: dict
    intermediate_2: list
    temp_var_a: str
    # ... 10 campos más

2. Timeouts en Nodos

import asyncio
from functools import wraps

def with_timeout(seconds):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            return await asyncio.wait_for(
                func(*args, **kwargs),
                timeout=seconds
            )
        return wrapper
    return decorator

@with_timeout(30)
async def llm_node(state):
    # Este nodo tiene timeout de 30s
    result = await llm.ainvoke(prompt)
    return {"result": result}

3. Testing de Grafos

import pytest

def test_research_agent():
    app = create_research_agent()

    result = app.invoke({
        "question": "Test question",
        "iterations": 0
    })

    assert "final_answer" in result
    assert len(result["search_queries"]) > 0
    assert result["iterations"] == 1

def test_conditional_routing():
    app = create_conditional_agent()

    # Test path A
    result_a = app.invoke({"confidence": 0.9})
    assert result_a["path_taken"] == "high_confidence"

    # Test path B
    result_b = app.invoke({"confidence": 0.3})
    assert result_b["path_taken"] == "low_confidence"

Performance y Optimización

Benchmarks Reales (2025)

Según datos de producción de LangChain:

Optimizaciones Clave

1. Batch Processing

# Procesar múltiples requests en batch
results = []
for request in batch_requests:
    results.append(app.invoke(request))

# Mejor: Async batch
import asyncio

async def process_batch(requests):
    tasks = [app.ainvoke(req) for req in requests]
    return await asyncio.gather(*tasks)

results = asyncio.run(process_batch(batch_requests))

2. Caching de Resultados

from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_llm_call(prompt: str) -> str:
    """Cache LLM calls con mismo prompt"""
    return llm.invoke(prompt).content

def smart_node(state):
    prompt = state["query"]
    result = cached_llm_call(prompt)  # Usa cache
    return {"result": result}

3. Streaming para UX

# Streaming de outputs intermedios
for chunk in app.stream({"question": "..."}):
    node_name = chunk.get("__node__")
    data = chunk.get("data", {})
    print(f"[{node_name}] {data}")
    # Enviar a frontend en real-time

Casos de Uso Avanzados

1. Code Generation Pipeline

class CodeGenState(TypedDict):
    requirements: str
    pseudocode: str
    code: str
    tests: str
    review: str
    final_code: str

workflow = StateGraph(CodeGenState)

workflow.add_node("plan", generate_pseudocode_node)
workflow.add_node("implement", generate_code_node)
workflow.add_node("test", generate_tests_node)
workflow.add_node("review", review_code_node)

# Pipeline con auto-corrección
workflow.add_edge(START, "plan")
workflow.add_edge("plan", "implement")
workflow.add_edge("implement", "test")
workflow.add_conditional_edges(
    "test",
    lambda s: "review" if s["tests_pass"] else "implement",
    {"review": "review", "implement": "implement"}
)
workflow.add_edge("review", END)

2. Data Analysis Agent

class AnalysisState(TypedDict):
    dataset_path: str
    data: pd.DataFrame
    insights: List[str]
    visualizations: List[str]
    report: str

def load_data(state):
    df = pd.read_csv(state["dataset_path"])
    return {"data": df}

def analyze_patterns(state):
    df = state["data"]
    insights = []
    # Análisis estadístico
    insights.append(f"Mean: {df.mean()}")
    insights.append(f"Std: {df.std()}")
    return {"insights": insights}

def generate_viz(state):
    # Generar visualizaciones
    return {"visualizations": ["chart1.png", "chart2.png"]}

def create_report(state):
    insights = state["insights"]
    viz = state["visualizations"]
    report = f"Report with {len(insights)} insights and {len(viz)} charts"
    return {"report": report}

3. Customer Support Agent

class SupportState(TypedDict):
    customer_message: str
    intent: str
    user_data: dict
    solution: str
    escalate: bool

def classify_intent(state):
    message = state["customer_message"]
    # Clasificar: technical, billing, general
    intent = "technical"  # Simulated
    return {"intent": intent}

def fetch_user_data(state):
    # Buscar datos del usuario en DB
    data = {"tier": "premium", "history": [...]}
    return {"user_data": data}

def route_to_solution(state):
    intent = state["intent"]
    if intent == "technical":
        return "technical_support"
    elif intent == "billing":
        return "billing_support"
    else:
        return "general_support"

workflow = StateGraph(SupportState)
workflow.add_node("classify", classify_intent)
workflow.add_node("fetch_data", fetch_user_data)
workflow.add_node("technical_support", technical_node)
workflow.add_node("billing_support", billing_node)
workflow.add_node("general_support", general_node)

workflow.add_edge(START, "classify")
workflow.add_edge("classify", "fetch_data")
workflow.add_conditional_edges(
    "fetch_data",
    route_to_solution,
    {
        "technical_support": "technical_support",
        "billing_support": "billing_support",
        "general_support": "general_support"
    }
)

Migración desde Otros Frameworks

Desde LangChain Chains

# Antes: LangChain Chain
from langchain.chains import LLMChain

chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run(input="question")

# Después: LangGraph
def chain_as_node(state):
    result = llm.invoke(state["input"])
    return {"output": result.content}

workflow = StateGraph(State)
workflow.add_node("llm_step", chain_as_node)
workflow.add_edge(START, "llm_step")
workflow.add_edge("llm_step", END)

app = workflow.compile()
result = app.invoke({"input": "question"})

Desde CrewAI

# Antes: CrewAI
from crewai import Agent, Task, Crew

researcher = Agent(role="researcher", goal="research")
task = Task(description="research X", agent=researcher)
crew = Crew(agents=[researcher], tasks=[task])
result = crew.kickoff()

# Después: LangGraph (más control)
def researcher_node(state):
    # Mismo concepto pero más control de flujo
    research = do_research(state["task"])
    return {"research": research}

workflow = StateGraph(State)
workflow.add_node("researcher", researcher_node)
# ... más nodos y control fino

Recursos y Comunidad

Documentación Oficial

• LangGraph Docs: https://langchain-ai.github.io/langgraph/
• LangChain Docs: https://python.langchain.com/
• LangSmith: https://docs.smith.langchain.com/

Repositorios de Ejemplo

• LangGraph Examples: https://github.com/langchain-ai/langgraph-examples
• Templates: https://github.com/langchain-ai/langgraph-template

Comunidad

• Discord LangChain: https://discord.gg/langchain
• GitHub Discussions: https://github.com/langchain-ai/langgraph/discussions
• Twitter/X: @LangChainAI

Cursos y Tutoriales

• DeepLearning.AI: AI Agents in LangGraph (curso gratuito)
• LangChain Academy: Cursos oficiales
• YouTube: Canal oficial de LangChain

Blogs Recomendados

• Blog oficial de LangChain
• Real Python tutorials
• Towards Data Science (Medium)

Comparativa con Artículos Relacionados

Si te interesó LangGraph, estos artículos complementan tu conocimiento:

• vLLM: Deploy LLMs a Escala en Producción – Infrastructure para servir los LLMs que usan tus agents
• Quantization de LLMs: GGUF vs GPTQ vs AWQ – Optimiza modelos para reducir costos en agents
• Fine-Tuning de LLMs: Guía Completa – Entrena modelos especializados para tus agents
• Homelab con GPU 24GB+: Stack Completo – Hardware para ejecutar agents localmente
• IA Agéntica: Revolución de Agents Autónomos – Conceptos fundamentales de IA agéntica

Preguntas Frecuentes (FAQs)

1. ¿LangGraph funciona solo con OpenAI o puedo usar otros LLMs?

LangGraph es agnóstico al modelo. Puedes usar:

• Modelos OpenAI (GPT-4, GPT-3.5)
• Anthropic Claude
• Google Gemini
• Modelos open-source (Llama, Mistral, etc.) vía Ollama o vLLM
• Cualquier LLM compatible con LangChain

# Ejemplo con diferentes LLMs
from langchain_anthropic import ChatAnthropic
from langchain_community.llms import Ollama

# Claude
llm = ChatAnthropic(model="claude-3-sonnet")

# Ollama local
llm = Ollama(model="llama2")

# Usar en tus nodos
def node_with_claude(state):
    result = llm.invoke(state["prompt"])
    return {"output": result.content}

2. ¿Cuánto cuesta ejecutar agents con LangGraph en producción?

El costo depende principalmente de las llamadas a LLM, no de LangGraph (que es open-source):

Ejemplo de cálculo (GPT-4):

• Input: $0.03 / 1K tokens
• Output: $0.06 / 1K tokens
• Agent típico: 3-5 llamadas LLM por request
• Promedio: 2K tokens input + 500 tokens output por llamada
• Costo por request: ~$0.20-$0.35

Optimizaciones para reducir costos:

• Usar GPT-3.5-turbo para tareas simples ($0.001/1K tokens)
• Implementar caching de prompts frecuentes
• Usar modelos open-source locales para desarrollo
• Batch processing cuando sea posible

3. ¿Puedo ejecutar LangGraph completamente offline/local?

Sí, configurando todo el stack localmente:

# LLM local con Ollama
from langchain_community.llms import Ollama

llm = Ollama(model="llama2")

# Embeddings locales
from langchain_community.embeddings import OllamaEmbeddings

embeddings = OllamaEmbeddings(model="llama2")

# Vector store local
from langchain_community.vectorstores import Chroma

vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

# Checkpoints locales (SQLite)
from langgraph.checkpoint.sqlite import SqliteSaver

checkpointer = SqliteSaver.from_conn_string("./checkpoints.db")

# Todo local!
app = workflow.compile(checkpointer=checkpointer)

4. ¿Cómo debuggeo un agent que no funciona correctamente?

Estrategias de debugging:

• Logging detallado en cada nodo:

def debug_node(state):
    print(f"Estado actual: {state}")
    result = process(state)
    print(f"Estado después: {result}")
    return result

• LangSmith tracing:

os.environ["LANGCHAIN_TRACING_V2"] = "true"
# Ver trazas en https://smith.langchain.com/

• Streaming de estados intermedios:

for step in app.stream({"input": "..."}):
    print(f"Paso: {step}")

• Tests unitarios por nodo:

def test_specific_node():
    result = my_node({"input": "test"})
    assert result["output"] == "expected"

5. ¿LangGraph escala para millones de usuarios como Klarna?

Sí, con arquitectura adecuada:

Klarna case study (85M usuarios):

• Horizontal scaling con instancias stateless
• PostgreSQL para checkpoints (no SQLite en prod)
• Load balancing con NGINX/AWS ALB
• Caching agresivo de prompts comunes
• Rate limiting por usuario
• Monitoring con Datadog/Grafana

Arquitectura tipo:

┌─────────────┐
│ Load        │
│ Balancer    │
└──────┬──────┘
       │
   ┌───┴───┬───────┬───────┐
   │       │       │       │
┌──▼──┐ ┌──▼──┐ ┌──▼──┐ ┌──▼──┐
│ Pod │ │ Pod │ │ Pod │ │ Pod │ (LangGraph instances)
└──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘
   │       │       │       │
   └───┬───┴───────┴───────┘
       │
┌──────▼──────────┐
│ PostgreSQL      │ (Shared checkpoints)
│ (Checkpoints)   │
└─────────────────┘

6. ¿Cuál es la diferencia entre LangGraph y LangChain?

Cuándo usar cada uno:

• LangChain: RAG simple, Q&A básico, pipelines lineales
• LangGraph: Multi-step agents, workflows con decisiones, human-in-loop

7. ¿Necesito experiencia con grafos o state machines para usar LangGraph?

No es requisito, pero ayuda entender conceptos básicos:

Conceptos clave (fáciles de aprender):

• Nodo: Una función que procesa datos
• Arista: Conexión entre funciones
• Estado: Datos que pasan entre funciones
• Grafo: Conjunto de nodos + aristas

Aprende haciendo:

# Esto ya es un grafo válido:
workflow = StateGraph(State)
workflow.add_node("step1", my_function)
workflow.add_node("step2", another_function)
workflow.add_edge("step1", "step2")  # step1 → step2

# Si entiendes esto, puedes usar LangGraph!

8. ¿Puedo combinar LangGraph con LangChain components?

Absolutamente. LangGraph es compatible con todo el ecosistema LangChain:

from langchain.chains import RetrievalQA
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

# Chain de LangChain como nodo de LangGraph
def retrieval_node(state):
    vectorstore = Chroma(...)
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        retriever=vectorstore.as_retriever()
    )
    result = qa_chain.run(state["question"])
    return {"answer": result}

# Usar en grafo
workflow.add_node("rag", retrieval_node)

9. ¿Qué pasa si un nodo falla en medio de la ejecución?

Con checkpointing, el agent puede resumir desde el último checkpoint:

from langgraph.checkpoint.sqlite import SqliteSaver

checkpointer = SqliteSaver.from_conn_string("./checkpoints.db")
app = workflow.compile(checkpointer=checkpointer)

try:
    result = app.invoke(
        {"input": "..."},
        config={"configurable": {"thread_id": "session_1"}}
    )
except Exception as e:
    print(f"Error: {e}")
    # Reintentar desde último checkpoint
    result = app.invoke(
        {"input": "..."},
        config={"configurable": {"thread_id": "session_1"}}
    )
    # Automáticamente resume desde donde falló

Sin checkpointing: El agent falla completamente y debe reiniciar.

10. ¿Cuánto tiempo toma aprender LangGraph?

Timeline realista:

• 1-2 días: Conceptos básicos, primer agent simple funcionando
• 1 semana: Dominar conditional routing, multi-node workflows
• 2-3 semanas: Checkpointing, human-in-loop, patterns avanzados
• 1-2 meses: Deployment en producción, optimización, scaling

Ruta de aprendizaje recomendada:

• Tutorial oficial de LangChain (2h)
• Documentación LangGraph (4h)
• Construir 3-5 agents de ejemplo (1 semana)
• Proyecto real pequeño (1 semana)
• Deployment y producción (2 semanas)

11. ¿LangGraph es mejor que construir agents desde cero?

Depende del caso:

✅ Usa LangGraph si:

• Necesitas checkpointing/persistencia nativo
• Quieres integración con ecosistema LangChain
• Valoras debugging con LangSmith
• Deployando en producción rápido

❌ Construye desde cero si:

• Tienes requisitos muy específicos que LangGraph no cubre
• Quieres zero dependencies
• Performance crítico (nanosegundos importan)
• Ya tienes sistema custom funcionando

En general: LangGraph ahorra semanas de desarrollo vs. custom.

12. ¿Hay límites en el número de nodos o complejidad del grafo?

No hay límites técnicos duros, pero considera:

Best practices:

• Nodos: < 20 nodos por grafo (más allá: usar subgrafos)
• Profundidad: < 10 niveles de anidamiento
• Estado: < 10 MB por state object
• Iteraciones: Siempre poner límite máximo (ej: max_iterations=100)

Si necesitas más:

• Dividir en subgrafos
• Usar colas para procesamiento asíncrono
• Implementar patterns de microservicios

13. ¿Puedo monetizar aplicaciones construidas con LangGraph?

Sí, LangGraph tiene licencia MIT (completamente libre para uso comercial).

Modelos de monetización comunes:

• SaaS con agents como servicio
• APIs de agents especializados
• Consultoría de implementación
• Plugins/extensiones para plataformas

No hay royalties ni restricciones de uso comercial.

14. ¿Cómo manejo secrets y API keys de forma segura?

Best practices de seguridad:

# ✅ Variables de entorno
import os
api_key = os.getenv("OPENAI_API_KEY")

# ✅ Vaults (producción)
from azure.keyvault.secrets import SecretClient

client = SecretClient(vault_url, credential)
api_key = client.get_secret("openai-key").value

# ❌ NUNCA hardcodear
api_key = "sk-xxx..."  # MAL!

# ❌ NUNCA commitear a Git
# Usar .gitignore para .env files

Docker secrets:

# docker-compose.yml
services:
  app:
    image: langgraph-agent
    secrets:
      - openai_key
    environment:
      OPENAI_API_KEY_FILE: /run/secrets/openai_key

secrets:
  openai_key:
    file: ./secrets/openai.txt

15. ¿Dónde encuentro más ejemplos de código y templates?

Recursos oficiales:

• GitHub LangGraph Examples: https://github.com/langchain-ai/langgraph-examples

• 50+ ejemplos completos
• Casos de uso reales
• Templates production-ready

• LangChain Templates: https://github.com/langchain-ai/langchain/tree/master/templates

• Templates específicos por caso de uso
• docker-compose incluido

• LangSmith Hub: https://smith.langchain.com/hub

• Prompts compartidos por la comunidad
• Workflows testeados

• Este blog:

• Tutorial de IA Agéntica
• Homelab IA Stack

Conclusión: El Futuro de los AI Agents

LangGraph representa el estado del arte en frameworks de IA agéntica para 2025. Su arquitectura basada en grafos stateful, combinada con checkpointing robusto y control de flujo completo, lo convierte en la elección ideal para construir agents que van a producción.

Puntos Clave

✅ Flexibilidad: Control total sobre workflows complejos con conditional routing y loops
✅ Production-Ready: Usado por Klarna (85M usuarios), Uber, LinkedIn en producción real
✅ State Management: Persistencia nativa de estado entre ejecuciones
✅ Observability: Integración con LangSmith para debugging profundo
✅ Escalabilidad: Arquitectura probada para alto throughput y baja latencia

Próximos Pasos

• Instala LangGraph y construye tu primer agent simple
• Experimenta con conditional routing y human-in-the-loop
• Deploy a producción con checkpointing y monitoring
• Únete a la comunidad LangChain en Discord
• Itera y mejora basándote en métricas reales

Recursos Finales

• 📚 Documentación Oficial
• 💬 Discord LangChain
• 🎓 Curso DeepLearning.AI
• 🐙 GitHub Examples

¿Listo para construir tu primer AI agent con LangGraph? Empieza con los ejemplos de código de este artículo y escala desde ahí. El futuro de la IA es agéntico, y LangGraph es tu mejor herramienta para construirlo.

¿Te gustó esta guía? Comparte con otros desarrolladores que estén construyendo AI agents. ¿Tienes dudas? Déjalas en los comentarios.

Tags: #LangGraph #AIAgents #LangChain #MachineLearning #Python #StateMachines #IA #AgenticAI #LLMs #Producción