Mini Tutorial: Agentes de IA con MongoDB y LangGraph

Curso para Crear Agentes de AI con MongoDB

COMPARTE ESTE ARTÍCULO Y MUESTRA LO QUE APRENDISTE

<h1>Mini Tutorial: Agentes de IA con MongoDB y LangGraph</h1>

🎯 Introducción

Los agentes de IA son sistemas autónomos que pueden percibir su entorno, razonar sobre situaciones complejas, tomar decisiones y ejecutar acciones para alcanzar objetivos específicos. A diferencia de simples prompts o sistemas RAG, los agentes pueden usar herramientas, planificar múltiples pasos y aprender de sus interacciones.

📚 Conceptos Fundamentales

¿Qué son los Agentes de IA?

Los agentes de IA representan un salto evolutivo en sistemas inteligentes:

Simple Prompting: Una pregunta → Una respuesta directa
RAG (Retrieval-Augmented Generation): Pregunta → Búsqueda en documentos → Respuesta contextual
Agentes de IA: Análisis → Planificación → Uso de herramientas → Ejecución iterativa → Resultado

Cuándo usar Agentes vs RAG

Usa RAG cuando:

Necesites respuestas basadas en documentación estática
La tarea sea relativamente simple y directa
Solo requieras recuperar y resumir información

Usa Agentes cuando:

Necesites tomar decisiones complejas
Debas ejecutar múltiples acciones secuenciales
Requieras interactuar con APIs o servicios externos
El contexto cambie dinámicamente

🧩 Los 4 Componentes Esenciales

1. Percepción

La capacidad del agente para entender el lenguaje natural y extraer intención.

# Ejemplo conceptual
user_input = "Busca información sobre optimización de índices en MongoDB"
# El LLM procesa y entiende la intención
intent = "search_documentation"
query = "mongodb index optimization"

2. Herramientas (Tools)

Funciones que el agente puede ejecutar para realizar acciones específicas.

from langchain.tools import tool

@tool
def buscar_documentacion(query: str) -> str:
    """Busca en la documentación de MongoDB usando búsqueda vectorial"""
    # Implementación de búsqueda semántica
    resultados = vector_search(query)
    return resultados

3. Planificación

El cerebro que decide qué herramientas usar y en qué orden.

# El agente decide el plan de acción:
# 1. Buscar en documentación
# 2. Si no hay respuesta, buscar en web
# 3. Formatear respuesta final

4. Memoria

Almacenamiento del contexto y conversaciones previas.

# Memoria persistente con MongoDB
{
  "conversation_id": "user_123",
  "messages": [
    {"role": "user", "content": "¿Cómo optimizar queries?"},
    {"role": "assistant", "content": "Para optimizar..."}
  ],
  "timestamp": "2025-01-23T10:30:00Z"
}

🚀 Implementación Práctica

Paso 1: Configuración del Entorno

# Instalar dependencias
pip install langchain langgraph pymongo openai sentence-transformers

# Variables de entorno
export OPENAI_API_KEY="tu-api-key"
export MONGODB_URI="mongodb+srv://usuario:password@cluster.mongodb.net"

Paso 2: Configurar MongoDB con Búsqueda Vectorial

from pymongo import MongoClient
from sentence_transformers import SentenceTransformer

# Conectar a MongoDB
client = MongoClient("mongodb+srv://...")
db = client["agentes_ia"]
collection = db["documentacion"]

# Modelo para embeddings
model = SentenceTransformer('all-MiniLM-L6-v2')

# Crear índice vectorial en MongoDB Atlas
# (Se hace desde la interfaz de Atlas)
# Index name: vector_index
# Field: embedding
# Dimensions: 384
# Similarity: cosine

Paso 3: Crear Herramienta de Búsqueda Vectorial

from langchain.tools import tool
from typing import List

@tool
def busqueda_vectorial_mongodb(consulta: str) -> str:
    """
    Realiza búsqueda semántica en documentación de MongoDB
    usando embeddings y el índice vectorial.
    """
    # Generar embedding de la consulta
    query_embedding = model.encode(consulta).tolist()
    
    # Búsqueda vectorial en MongoDB
    resultados = collection.aggregate([
        {
            "$vectorSearch": {
                "index": "vector_index",
                "path": "embedding",
                "queryVector": query_embedding,
                "numCandidates": 100,
                "limit": 5
            }
        },
        {
            "$project": {
                "titulo": 1,
                "contenido": 1,
                "score": {"$meta": "vectorSearchScore"}
            }
        }
    ])
    
    # Formatear resultados
    docs_encontrados = []
    for doc in resultados:
        docs_encontrados.append({
            "titulo": doc["titulo"],
            "contenido": doc["contenido"][:500],
            "relevancia": doc["score"]
        })
    
    return str(docs_encontrados)

Paso 4: Orquestar el Agente con LangGraph

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated
import operator

# Definir el estado del agente
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    herramienta_usada: str
    respuesta_final: str

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

# Crear el grafo del agente
workflow = StateGraph(AgentState)

# Nodo 1: Analizar consulta
def analizar_consulta(state):
    ultimo_mensaje = state["messages"][-1]
    # El LLM decide qué hacer
    prompt = f"""
    Analiza esta consulta y decide si necesitas usar la herramienta de búsqueda:
    {ultimo_mensaje}
    
    Responde SOLO: "BUSCAR" o "RESPONDER_DIRECTO"
    """
    decision = llm.invoke(prompt).content
    return {"herramienta_usada": decision}

# Nodo 2: Ejecutar búsqueda
def ejecutar_busqueda(state):
    consulta = state["messages"][-1]
    resultado = busqueda_vectorial_mongodb.invoke(consulta)
    return {"messages": [f"Resultados de búsqueda: {resultado}"]}

# Nodo 3: Generar respuesta
def generar_respuesta(state):
    contexto = "\n".join(state["messages"])
    prompt = f"""
    Basándote en este contexto:
    {contexto}
    
    Genera una respuesta clara y útil para el usuario.
    """
    respuesta = llm.invoke(prompt).content
    return {"respuesta_final": respuesta}

# Construir el grafo
workflow.add_node("analizar", analizar_consulta)
workflow.add_node("buscar", ejecutar_busqueda)
workflow.add_node("responder", generar_respuesta)

# Definir el flujo
workflow.set_entry_point("analizar")

def decidir_siguiente_paso(state):
    if state["herramienta_usada"] == "BUSCAR":
        return "buscar"
    return "responder"

workflow.add_conditional_edges(
    "analizar",
    decidir_siguiente_paso,
    {
        "buscar": "buscar",
        "responder": "responder"
    }
)

workflow.add_edge("buscar", "responder")
workflow.add_edge("responder", END)

# Compilar el agente
agente = workflow.compile()

Paso 5: Agregar Memoria Persistente

from datetime import datetime

def guardar_conversacion(conversation_id: str, mensaje: dict):
    """Guarda mensajes en MongoDB para memoria persistente"""
    db.conversaciones.update_one(
        {"conversation_id": conversation_id},
        {
            "$push": {"messages": mensaje},
            "$set": {"last_updated": datetime.utcnow()}
        },
        upsert=True
    )

def cargar_conversacion(conversation_id: str) -> list:
    """Carga el historial de conversación"""
    conv = db.conversaciones.find_one({"conversation_id": conversation_id})
    return conv.get("messages", []) if conv else []

# Usar con el agente
def ejecutar_agente_con_memoria(consulta: str, user_id: str):
    # Cargar memoria
    historial = cargar_conversacion(user_id)
    
    # Ejecutar agente
    resultado = agente.invoke({
        "messages": historial + [consulta]
    })
    
    # Guardar en memoria
    guardar_conversacion(user_id, {"role": "user", "content": consulta})
    guardar_conversacion(user_id, {
        "role": "assistant", 
        "content": resultado["respuesta_final"]
    })
    
    return resultado["respuesta_final"]

Paso 6: Ejemplo de Uso Completo

# Caso de uso: Agente de soporte técnico MongoDB
consulta_usuario = "¿Cómo puedo optimizar las consultas lentas en mi base de datos?"

respuesta = ejecutar_agente_con_memoria(
    consulta=consulta_usuario,
    user_id="usuario_123"
)

print(respuesta)
# Output: "Para optimizar consultas lentas en MongoDB, te recomiendo:
# 1. Crear índices apropiados usando createIndex()
# 2. Analizar consultas con explain()
# 3. Usar el MongoDB Profiler para identificar cuellos de botella..."

🎓 Conceptos Clave de LangGraph

Nodos, Aristas y Estados

Nodos: Funciones que procesan el estado (analizar, buscar, responder)
Aristas: Conexiones entre nodos que definen el flujo
Estado: Diccionario compartido que pasa entre nodos

# Visualización del flujo
#
#    [INICIO]
#       ↓
#   [Analizar]
#       ↓
#    ¿Necesita búsqueda?
#    ↙        ↘
# [Buscar]  [Responder]
#    ↓           ↓
# [Responder]  [FIN]
#    ↓
#  [FIN]

💡 Mejores Prácticas

1. Diseño de Herramientas

Mantén cada herramienta enfocada en una tarea específica
Usa docstrings descriptivos (el LLM los lee)
Valida entradas y maneja errores

2. Optimización de Búsqueda Vectorial

Experimenta con diferentes modelos de embeddings
Ajusta el número de candidatos y límites según tu caso
Usa filtros combinados (vectorial + metadata)

3. Gestión de Memoria

Limita el tamaño del historial para evitar tokens excesivos
Implementa estrategias de resumen para conversaciones largas
Usa índices en MongoDB para consultas rápidas

4. Manejo de Costos

Implementa caché de respuestas frecuentes
Usa modelos más pequeños para tareas simples
Monitorea el uso de tokens

🔍 Casos de Uso Reales

Asistente de Documentación: Responde preguntas técnicas buscando en docs
Agente de Soporte: Resuelve tickets con acceso a base de conocimientos
Analista de Datos: Ejecuta queries en MongoDB según lenguaje natural
Automatización de Tareas: Ejecuta workflows complejos paso a paso

📊 Próximos Pasos

Experimenta con diferentes arquitecturas de grafos
Agrega más herramientas (APIs externas, calculadoras, etc.)
Implementa evaluación y métricas de calidad
Explora agentes multi-agente y colaborativos

📚 Recursos Adicionales

¡Felicidades! 🎉 Ahora tienes las bases para construir agentes de IA inteligentes y escalables.