Despliega y prueba tu agente en un entorno simulado

Introducción: El Momento de la Verdad para tu Agente

Has diseñado, construido y configurado un agente de LangGraph con memoria y herramientas para la gestión de proyectos. Ahora llega la fase crítica: sacarlo del entorno de desarrollo aislado y exponerlo a un entorno simulado que imite las condiciones del mundo real. Esta lección no trata solo de ejecutar un script; se trata de validar que tu creación es robusta, útil y capaz de manejar la complejidad y la imprevisibilidad de las interacciones reales. El despliegue en un entorno simulado es el puente obligatorio entre el prototipo y la aplicación productiva, donde se descubren problemas de concurrencia, fallos en la lógica de herramientas y limitaciones en la memoria a largo plazo.

En este contexto avanzado, "desplegar" significa empaquetar tu grafo de LangGraph dentro de una aplicación servidora (como FastAPI) que pueda recibir peticiones HTTP, mantener el estado de múltiples conversaciones o proyectos de forma aislada, y gestionar recursos como conexiones a bases de datos o APIs externas. "Probar" en un entorno simulado implica crear flujos de trabajo automatizados o interfaces sencillas que permitan a usuarios ficticios (o a ti mismo actuando en diferentes roles) interactuar con el agente, sometiéndolo a cargas de trabajo variables y escenarios de borde. El objetivo final es ganar una confianza operacional sólida antes de un lanzamiento real.

Concepto Clave: El Entorno Simulado como Campo de Pruebas

Imagina que has construido un nuevo modelo de automóvil de carreras. El taller de desarrollo (tu script de Python) es donde ajustas el motor y la aerodinámica. Pero antes de llevarlo a una carrera real, lo llevas a un circuito de pruebas. Este circuito tiene curvas cerradas, superficies resbaladizas, y zonas para probar frenadas de emergencia. No es una carretera pública, pero reproduce fielmente sus desafíos. Un entorno simulado para tu agente de IA es exactamente eso: un circuito de pruebas controlado. Es un ecosistema de software que replica la interfaz, la carga y las condiciones de error de un entorno de producción, pero dentro de un contenedor seguro donde los fallos no tienen coste real.

Los componentes clave de este entorno son: un servidor de aplicaciones que aloja la lógica del agente y maneja múltiples solicitudes concurrentes; un mecanismo de estado persistente (como Redis o una base de datos) para la memoria del grafo entre sesiones; un simulador de entradas de usuario que puede ser una suite de pruebas automatizada, un script que envía prompts predefinidos, o incluso una interfaz web básica; y un sistema de monitoreo y logging detallado para capturar cada paso, decisión del agente y llamada a herramienta. La analogía se completa con los ingenieros de pruebas (tus scripts de simulación) que "conducen" al agente de formas impredecibles para encontrar sus puntos débiles.

Tip Clave: La simulación no debe probar solo el "camino feliz". Diseña escenarios que incluyan solicitudes ambiguas, herramientas que fallen, recuperación de conversaciones antiguas y picos de carga repentinos. La resiliencia es tan importante como la funcionalidad.

Cómo Funciona en la Práctica: Arquitectura de Despliegue y Flujo de Pruebas

El proceso práctico sigue una secuencia lógica. Primero, debes encapsular tu grafo. En lugar de interactuar directamente con el objeto `Graph` en una consola, lo integrarás en una clase o fábrica que pueda instanciar un grafo por sesión o proyecto. Este grafo se configurará con un `Checkpointer` persistente (por ejemplo, usando `SqliteSaver`) para que el estado de la conversación y la memoria sobrevivan a reinicios del servidor. Segundo, debes exponer una API. Utilizarás un framework como FastAPI para crear endpoints, típicamente uno para iniciar una nueva sesión/conversación de gestión de proyectos y otro para enviar un mensaje (una instrucción del usuario) a una sesión existente. El servidor se desplegará localmente usando Uvicorn.

Para la fase de prueba, crearás un script de simulación de cliente. Este script actuará como un Product Owner, un Desarrollador y un Stakeholder no técnico, enviando secuencias de mensajes preprogramadas pero realistas a tu API. Por ejemplo, el Product Owner podría crear una tarea, el Desarrollador preguntar por sus dependencias, y luego el Stakeholder pedir un resumen del progreso, todo en la misma sesión de proyecto. El script registrará cada respuesta y verificará que el agente mantenga el contexto, utilice las herramientas correctas (como `query_task_database` o `assign_resource`) y proporcione respuestas coherentes. Paralelamente, puedes usar herramientas como `locust` o `pytest` con plugins asíncronos para simular carga de múltiples usuarios simultáneos.

Código en Acción: Servidor FastAPI y Cliente de Simulación

A continuación, un ejemplo completo y funcional de un servidor FastAPI básico para nuestro agente de gestión de proyectos, asumiendo que tenemos un grafo ya construido llamado `project_agent_graph`. Luego, un script de simulación que prueba tres flujos de usuario.

Servidor FastAPI para el Agente (app.py)

from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing import Dict, Optional
import uuid
from langgraph.checkpoint.sqlite import SqliteSaver
from my_agent_builder import create_project_agent  # Tu función de construcción
import asyncio

# Configuración de la aplicación y el estado global
app = FastAPI(title="LangGraph Project Agent API")
sqlite_checkpointer = SqliteSaver.from_conn_string(":memory:")  # Usa un archivo para producción
agent_graph = create_project_agent(sqlite_checkpointer)  # Grafo configurado con persistencia

# Almacén en memoria para sesiones (en producción, usa una base de datos)
sessions: Dict[str, dict] = {}

# Modelos Pydantic
class MessageRequest(BaseModel):
    message: str
    user_id: Optional[str] = "default_user"  # Para personalización

class SessionResponse(BaseModel):
    session_id: str
    response: str

# Endpoints
@app.post("/create_session", response_model=SessionResponse)
async def create_session():
    """Crea una nueva sesión de gestión de proyectos."""
    session_id = str(uuid.uuid4())
    # Configura el estado inicial del grafo para esta sesión
    initial_config = {"configurable": {"thread_id": session_id}}
    sessions[session_id] = {"config": initial_config}
    return SessionResponse(session_id=session_id, response="Sesión de proyecto creada. ¡Puedes empezar!")

@app.post("/send_message/{session_id}", response_model=SessionResponse)
async def send_message(session_id: str, request: MessageRequest):
    """Envía un mensaje a una sesión de agente específica."""
    if session_id not in sessions:
        raise HTTPException(status_code=404, detail="Sesión no encontrada")

    config = sessions[session_id]["config"]
    # Invoca el grafo de LangGraph de forma asíncrona
    try:
        inputs = {"messages": [("user", request.message)], "user_id": request.user_id}
        # Usamos `astream_events` para capturar el output final de forma más clara
        final_output = None
        async for event in agent_graph.astream_events(inputs, config=config, version="v1"):
            if event["event"] == "on_chain_end" and event["name"] == "agent":
                final_output = event["data"].get("output")
                break

        if final_output and hasattr(final_output, 'messages') and final_output.messages:
            agent_response = final_output.messages[-1].content
        else:
            agent_response = "No se pudo generar una respuesta."

        return SessionResponse(session_id=session_id, response=agent_response)

    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error en el agente: {str(e)}")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Script de Simulación de Cliente (simulate.py)

import asyncio
import aiohttp
import json

async def test_agent_flow():
    base_url = "http://localhost:8000"
    async with aiohttp.ClientSession() as session:
        # 1. Crear una nueva sesión de proyecto
        async with session.post(f"{base_url}/create_session") as resp:
            session_data = await resp.json()
            session_id = session_data["session_id"]
            print(f"[+] Sesión creada: {session_id}")
            print(f"Respuesta: {session_data['response']}\n")

        # Secuencia de pruebas simulando diferentes roles
        test_messages = [
            ("product_owner", "Crea una nueva tarea llamada 'Diseñar interfaz de usuario' con alta prioridad."),
            ("developer", "¿Cuáles son las tareas de alta prioridad asignadas a mí?"),
            ("product_owner", "Asigna la tarea 'Diseñar interfaz de usuario' al desarrollador 'Ana López'."),
            ("stakeholder", "Dame un resumen del progreso del proyecto 'Dashboard Admin'."),
            ("developer", "Marca la tarea 'Diseñar interfaz de usuario' como completada en un 50%."),
        ]

        for role, message in test_messages:
            print(f"[{role.upper()}] > {message}")
            payload = {"message": message, "user_id": role}
            async with session.post(f"{base_url}/send_message/{session_id}", json=payload) as resp:
                response_data = await resp.json()
                print(f"[AGENTE] < {response_data['response'][:200]}...")  # Truncamos por brevedad
                print("-" * 50)
            await asyncio.sleep(0.5)  # Pequeña pausa entre mensajes

        # 2. Prueba de recuperación de memoria (nuevo "cliente" reconectándose)
        print("\n[+] Prueba de Memoria: Reconexión a la misma sesión.")
        memory_test_msg = "¿Qué tareas hemos discutido hasta ahora?"
        print(f"[USER] > {memory_test_msg}")
        payload = {"message": memory_test_msg, "user_id": "product_owner"}
        async with session.post(f"{base_url}/send_message/{session_id}", json=payload) as resp:
            response_data = await resp.json()
            print(f"[AGENTE] < {response_data['response'][:300]}...")

if __name__ == "__main__":
    asyncio.run(test_agent_flow())

Ejemplo de Herramienta con Persistencia Simulada (my_tools.py)

from langchain.tools import tool
import sqlite3
from datetime import datetime
from typing import List, Dict, Any

# Simulación de una base de datos de tareas
def init_task_db():
    conn = sqlite3.connect(":memory:")  # En producción, usa un archivo
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS tasks (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            priority TEXT,
            assignee TEXT,
            progress INTEGER DEFAULT 0,
            project_id TEXT,
            created_at TIMESTAMP
        )
    ''')
    conn.commit()
    return conn

TASK_DB = init_task_db()

@tool
def create_new_task(title: str, priority: str = "medium", project_id: str = "default") -> str:
    """Crea una nueva tarea en la base de datos del proyecto."""
    try:
        cursor = TASK_DB.cursor()
        cursor.execute('''
            INSERT INTO tasks (title, priority, project_id, created_at)
            VALUES (?, ?, ?, ?)
        ''', (title, priority, project_id, datetime.now()))
        TASK_DB.commit()
        task_id = cursor.lastrowid
        return f"Tarea creada exitosamente: '{title}' (ID: {task_id}, Prioridad: {priority})."
    except Exception as e:
        return f"Error al crear la tarea: {str(e)}"

@tool
def query_tasks_by_priority(priority: str, project_id: str = "default") -> List[Dict[str, Any]]:
    """Consulta tareas filtradas por prioridad y proyecto."""
    cursor = TASK_DB.cursor()
    cursor.execute('''
        SELECT id, title, assignee, progress FROM tasks
        WHERE priority = ? AND project_id = ?
    ''', (priority, project_id))
    rows = cursor.fetchall()
    tasks = [{"id": r[0], "title": r[1], "assignee": r[2], "progress": r[3]} for r in rows]
    return tasks

@tool
def assign_task_to_member(task_id: int, assignee_name: str) -> str:
    """Asigna una tarea existente a un miembro del equipo."""
    try:
        cursor = TASK_DB.cursor()
        cursor.execute('''
            UPDATE tasks SET assignee = ? WHERE id = ?
        ''', (assignee_name, task_id))
        TASK_DB.commit()
        if cursor.rowcount > 0:
            return f"Tarea ID {task_id} asignada a '{assignee_name}'."
        else:
            return f"Tarea con ID {task_id} no encontrada."
    except Exception as e:
        return f"Error en la asignación: {str(e)}"

Errores Comunes y Cómo Evitarlos

1. Estado de Grafo No Aislado entre Sesiones: El error más grave es usar la misma instancia de grafo sin un `thread_id` configurable único por conversación, mezclando las memorias de todos los usuarios. Cómo evitarlo: Siempre utiliza un `Checkpointer` persistente (como `SqliteSaver`) y pasa un diccionario `config` con una clave `"thread_id"` única (por ejemplo, el `session_id`) en cada invocación. Asegúrate de que tu función de construcción del grafo acepte y use este checkpointer.

2. Bloqueo del Event Loop en Herramientas Síncronas: Si tus herramientas realizan operaciones de E/S bloqueantes (como peticiones HTTP sin `async/await`) dentro de un servidor asíncrono como FastAPI, bloquearás el event loop y matarás la concurrencia. Cómo evitarlo: Implementa todas las herramientas usando bibliotecas asíncronas (por ejemplo, `aiohttp` en lugar de `requests`) o ejecuta las operaciones bloqueantes en un `ThreadPoolExecutor` usando `asyncio.to_thread`.

3. Falta de Timeouts y Manejo de Errores en Herramientas: Una herramienta que consulta una API externa puede colgarse indefinidamente, dejando al agente y al usuario en espera. Cómo evitarlo: Envuelve todas las llamadas a herramientas en timeouts explícitos usando `asyncio.wait_for` y define respuestas de fallback claras que el agente pueda usar para informar al usuario y continuar el flujo.

4. Logging Insuficiente para Depuración: Cuando el agente da una respuesta inesperada en producción, sin logs detallados es imposible rastrear si el fallo estuvo en la herramienta, en el LLM, o en la lógica del grafo. Cómo evitarlo: Instrumenta tu grafo usando `astream_events` para registrar cada paso (nodo invocado, herramienta ejecutada, entrada/salida del LLM) en un sistema estructurado. En el entorno simulado, revisa estos logs meticulosamente.

5. Simulaciones que Solo Siguen el "Camino Feliz": Si tus pruebas solo envían prompts perfectos a los que el agente ya sabe responder, no estarás preparado para la realidad. Cómo evitarlo: Diseña escenarios de prueba que incluyan: solicitudes ambiguas o contradictorias, intentos de usar herramientas con parámetros inválidos, recuperación de conversaciones muy antiguas, y simulación de múltiples usuarios interactuando con el mismo recurso (condiciones de carrera).

Checklist de Dominio

• ¿Has encapsulado tu grafo de LangGraph dentro de un servidor de aplicaciones (por ejemplo, FastAPI) con endpoints claros para crear sesiones y enviar mensajes?
• ¿Has implementado un mecanismo de persistencia de estado (Checkpointer) como `SqliteSaver` o `RedisSaver` para aislar y mantener la memoria entre reinicios y por sesión?
• ¿Has creado y ejecutado un script de simulación que prueba al menos tres flujos de usuario diferentes (por ejemplo, creación, consulta y modificación) en una sola sesión?
• ¿Has validado que el agente mantiene el contexto correctamente a lo largo de una conversación larga y puede recuperar información de interacciones previas?
• ¿Has probado el manejo de errores de tu agente cuando una herramienta falla o devuelve una respuesta inesperada?
• ¿Has revisado los logs de ejecución del grafo (usando `astream_events`) para asegurarte de que cada nodo y herramienta se ejecuta en el orden y con los parámetros esperados?
• ¿Has simulado una carga básica de múltiples solicitudes concurrentes para identificar posibles cuellos de botella o problemas de concurrencia en tu código?
• ¿Has documentado el proceso de despliegue y los requisitos del entorno (variables de entorno, dependencias, puertos) para una futura puesta en producción real?