Despliega Agentes de IA con memoria a largo plazo a Producción en 15 Minutos (Sin Docker, Sin Kubernetes, Sin Dolor de Cabeza)

pip install -r requirements.txt

bedrock-agentcore
strands-agents
strands-agents-tools

"""
Agente de IA Listo para Producción con Memoria
Recuerda conversaciones y preferencias del usuario entre sesiones
"""
import os
from strands import Agent
from strands_tools import calculator
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from bedrock_agentcore.memory.integrations.strands.config import AgentCoreMemoryConfig, RetrievalConfig
from bedrock_agentcore.memory.integrations.strands.session_manager import AgentCoreMemorySessionManager

app = BedrockAgentCoreApp()

MEMORY_ID = os.getenv("BEDROCK_AGENTCORE_MEMORY_ID")
REGION = os.getenv("AWS_REGION", "us-west-2")
MODEL_ID = "us.anthropic.claude-3-7-sonnet-20250219-v1:0"

# Instancia global del agente
_agent = None

def get_or_create_agent(actor_id: str, session_id: str) -> Agent:
    """
    Obtener agente existente o crear uno nuevo con configuración de memoria.
    Como el contenedor está vinculado al ID de sesión, solo necesitamos un agente por contenedor.
    """
    global _agent

    if _agent is None:
        # Configurar memoria con recuperación para hechos y preferencias del usuario
        memory_config = AgentCoreMemoryConfig(
            memory_id=MEMORY_ID,
            session_id=session_id,
            actor_id=actor_id,
            retrieval_config={
                f"/users/{actor_id}/facts": RetrievalConfig(top_k=3, relevance_score=0.5),
                f"/users/{actor_id}/preferences": RetrievalConfig(top_k=3, relevance_score=0.5)
            }
        )

        # Crear agente con administrador de sesión de memoria
        _agent = Agent(
            model=MODEL_ID,
            session_manager=AgentCoreMemorySessionManager(memory_config, REGION),
            system_prompt="Eres un asistente útil con memoria. Recuerda las preferencias del usuario y hechos entre conversaciones. Usa la herramienta de cálculo para problemas matemáticos.",
            tools=[calculator]
        )

    return _agent

@app.entrypoint
def invoke(payload, context):
    """Punto de entrada de AgentCore Runtime con agente de carga perezosa"""
    if not MEMORY_ID:
        return {"error": "Memoria no configurada. Establece la variable de entorno BEDROCK_AGENTCORE_MEMORY_ID."}

    # Extraer información de sesión y actor
    actor_id = context.request_headers.get('X-Amzn-Bedrock-AgentCore-Runtime-Custom-Actor-Id', 'user') if context.request_headers else 'user'
    session_id = context.session_id or 'default_session'

    # Obtener o crear agente (carga perezosa)
    agent = get_or_create_agent(actor_id, session_id)

    prompt = payload.get("prompt", "¡Hola!")
    result = agent(prompt)

    return {
        "response": result.message.get('content', [{}])[0].get('text', str(result))
    }

if __name__ == "__main__":
    app.run()

python my_agent_memory.py

curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Mi nombre es Juan y me encanta la pizza"}'

curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{"prompt": "¿Qué recuerdas sobre mí?"}'

agentcore configure -e my_agent_memory.py

agentcore launch

import json
import uuid
import boto3

def test_short_memory(agent_arn, region=None):
    """Probar memoria a corto plazo dentro de una sola sesión"""

    # Extraer región del ARN si no se proporciona
    if not region:
        region = agent_arn.split(':')[3]

    # Inicializar cliente
    client = boto3.client('bedrock-agentcore', region_name=region)

    # Generar ID de sesión
    session_id = str(uuid.uuid4())

    print(f"Probando memoria a corto plazo en sesión: {session_id}")
    print(f"Región: {region}")
    print("-" * 50)

    try:
        # Primer mensaje - establecer contexto
        print("Mensaje 1: Estableciendo contexto...")
        payload1 = json.dumps({"prompt": "Mi nombre es Alicia y me gusta el helado de chocolate"}).encode()

        response1 = client.invoke_agent_runtime(
            agentRuntimeArn=agent_arn,
            runtimeSessionId=session_id,
            payload=payload1,
            qualifier="DEFAULT"
        )

        content1 = []
        for chunk in response1.get("response", []):
            content1.append(chunk.decode('utf-8'))

        result1 = json.loads(''.join(content1))
        print(f"Agente: {result1.get('response', 'Sin respuesta')}")
        print()

        # Segundo mensaje - probar recuperación de memoria
        print("Mensaje 2: Probando recuperación de memoria...")
        payload2 = json.dumps({"prompt": "¿Cuál es mi nombre y qué me gusta?"}).encode()

        response2 = client.invoke_agent_runtime(
            agentRuntimeArn=agent_arn,
            runtimeSessionId=session_id,  # Misma sesión
            payload=payload2,
            qualifier="DEFAULT"
        )

        content2 = []
        for chunk in response2.get("response", []):
            content2.append(chunk.decode('utf-8'))

        result2 = json.loads(''.join(content2))
        print(f"Agente: {result2.get('response', 'Sin respuesta')}")

        print("\n✓ Prueba de memoria a corto plazo completada")

    except Exception as e:
        print(f"Error: {e}")

python test_short_memory.py "AGENT_ARN"

import json
import uuid
import boto3
import time

def test_long_memory(agent_arn, region=None):
    """Probar memoria a largo plazo entre diferentes sesiones"""

    # Extraer región del ARN si no se proporciona
    if not region:
        region = agent_arn.split(':')[3]

    # Inicializar cliente
    client = boto3.client('bedrock-agentcore', region_name=region)

    # Generar IDs de sesión
    session_1 = str(uuid.uuid4())
    session_2 = str(uuid.uuid4())

    print(f"Probando memoria a largo plazo entre sesiones")
    print(f"Región: {region}")
    print("-" * 50)

    try:
        # Sesión 1: Almacenar información del usuario
        print(f"Sesión 1: {session_1}")
        print("Almacenando preferencias del usuario...")

        payload1 = json.dumps({"prompt": "Mi nombre es Sara y soy ingeniera de software. Prefiero Python sobre JavaScript."}).encode()

        response1 = client.invoke_agent_runtime(
            agentRuntimeArn=agent_arn,
            runtimeSessionId=session_1,
            payload=payload1,
            qualifier="DEFAULT"
        )

        content1 = []
        for chunk in response1.get("response", []):
            content1.append(chunk.decode('utf-8'))

        result1 = json.loads(''.join(content1))
        print(f"Agente: {result1.get('response', 'Sin respuesta')}")
        print()

        # Esperar extracción de memoria a largo plazo
        print("Esperando 25 segundos para extracción de memoria a largo plazo...")
        time.sleep(25)

        # Sesión 2: Probar recuperación de memoria
        print(f"Sesión 2: {session_2}")
        print("Probando recuperación de memoria entre sesiones...")

        payload2 = json.dumps({"prompt": "¿Qué recuerdas sobre mí? ¿Cuál es mi nombre y qué prefiero?"}).encode()

        response2 = client.invoke_agent_runtime(
            agentRuntimeArn=agent_arn,
            runtimeSessionId=session_2,  # Sesión diferente
            payload=payload2,
            qualifier="DEFAULT"
        )

        content2 = []
        for chunk in response2.get("response", []):
            content2.append(chunk.decode('utf-8'))

        result2 = json.loads(''.join(content2))
        print(f"Agente: {result2.get('response', 'Sin respuesta')}")

        print("\n✓ Prueba de memoria a largo plazo completada")

    except Exception as e:
        print(f"Error: {e}")

python test_long_memory.py "AGENT_ARN"

# O usando variables de entorno
export AGENT_ARN="tu-agent-arn"
python test_long_memory.py

agentcore destroy