Asistente de WhatsApp con IA: Twilio, LangGraph y OpenAI

Introducción

Construir experiencias conversacionales inteligentes en WhatsApp pide decisiones de arquitectura cuidadas y una buena integración de tools. Esta guía muestra cómo armar un asistente de WhatsApp con IA sobre Twilio, listo para producción. Procesa mensajes de voz, maneja eventos de calendario, hace búsquedas web y mantiene el contexto de la conversación entre sesiones.

La implementación usa Python, FastAPI, LangGraph y los modelos GPT-4 de OpenAI. Esto no es una exploración teórica. La arquitectura que describo acá sale de una implementación que funciona, disponible en github.com/GonzaGomezDev/whatsapp-ai-assistant-starting-setup, y cada decisión técnica que menciono refleja trade-offs reales.

Qué vas a construir

La arquitectura del sistema incluye estas capacidades:

• Procesamiento de mensajes de texto y de voz por WhatsApp
• Transcripción automática de audio con OpenAI Whisper
• Integración con Google Calendar (crear, ver y borrar eventos)
• Búsqueda web en tiempo real con la API de Tavily
• Historial de conversación persistente en PostgreSQL
• Ruteo entre varias tools con la máquina de estados de LangGraph

Este es un sistema de IA agéntico. La distinción importa: los chatbots tradicionales siguen flujos de conversación predefinidos, mientras que los sistemas agénticos razonan qué tools usar según la intención del usuario. Las implicancias de esa elección tocan todo, desde el diseño de la base de datos hasta el manejo de errores.

Por qué Twilio WhatsApp para asistentes de IA

La WhatsApp Business API de Twilio tiene ventajas técnicas concretas para sistemas conversacionales con IA:

Escala y alcance: WhatsApp tiene una base de 2+ mil millones de usuarios. Tu implementación llega a todo el mundo sin pedirle a nadie que instale otra app ni aprenda otra interfaz.

Multimedia: a diferencia de los sistemas basados en SMS, WhatsApp soporta mensajes de voz, imágenes y documentos. Para un asistente de IA eso importa, porque habilita transcripción de audio, análisis de imágenes y procesamiento de documentos en un solo canal.

Integración con sistemas de negocio: la API de Twilio maneja los mensajes por webhooks, y eso integra limpio con los frameworks web modernos. Conectás CRMs, bases de datos y herramientas de automatización con interfaces HTTP estándar.

Control programable: con acceso completo a la API controlás el ruteo de mensajes, la transformación de contenido y la lógica de integración. No te limitan las restricciones típicas de los constructores de chatbots no-code.

Panorama de la arquitectura

El sistema usa estos componentes centrales:

• FastAPI: framework web async de Python para atender los webhooks de Twilio con baja latencia
• LangGraph: framework de máquina de estados para workflows agénticos, con persistencia en PostgreSQL
• OpenAI GPT-4o-mini: el modelo de lenguaje que entiende el pedido y elige las tools
• PostgreSQL: dos bases separadas, una para el checkpointing de LangGraph y otra para el historial de mensajes
• Twilio WhatsApp API: la plataforma de entrega de mensajes, con eventos por webhook
• OpenAI Whisper: modelo de speech-to-text para transcribir los mensajes de voz
• Google Calendar API: manejo del calendario con autenticación OAuth 2.0
• Tavily Search: API de búsqueda web para traer información en tiempo real

El sistema sigue un patrón conversacional con estado. Cada usuario, identificado por su número de teléfono, mantiene un thread de conversación independiente, con su propio checkpoint de estado en PostgreSQL. Esa elección tiene implicancias concretas en concurrencia, aislamiento de datos y manejo de memoria.

Requisitos

Antes de meternos en la implementación, asegurate de tener:

• Python 3.10 o superior instalado
• Una base PostgreSQL (local o en la nube)
• Una cuenta de Twilio con sandbox de WhatsApp o número de negocio aprobado
• Una API key de OpenAI con acceso a los modelos GPT-4 y a Whisper
• Un proyecto de Google Cloud con la Calendar API habilitada
• Una API key de Tavily para la búsqueda web
• Manejo básico de Python, APIs REST y conceptos de bases de datos

Paso 1: preparar el entorno de desarrollo

Clonar el repositorio

git clone https://github.com/GonzaGomezDev/whatsapp-ai-assistant-starting-setup.git
cd whatsapp-ai-assistant-starting-setup/backend

Instalar las dependencias de Python

pip install -r requirements.txt

Dependencias clave de requirements.txt:

• fastapi y uvicorn para el server web
• langchain, langgraph y langchain-openai para orquestar la IA
• langgraph-checkpoint-postgres para persistir la conversación
• openai para integrar GPT-4 y Whisper
• twilio para la API de WhatsApp
• psycopg2-binary para conectarse a PostgreSQL
• google-api-python-client para la integración con Calendar
• langchain_tavily para la búsqueda web
• sqlalchemy como ORM del historial de mensajes

Configurar las variables de entorno

Creá un archivo .env.development en el directorio backend:

# OpenAI Configuration
OPENAI_API_KEY=sk-proj-xxxxxxxxxxxxx

# Twilio WhatsApp Configuration
TWILIO_ACCOUNT_SID=ACxxxxxxxxxxxxx
TWILIO_AUTH_TOKEN=xxxxxxxxxxxxx

# PostgreSQL Database
DB_USER=postgres
DB_PASSWORD=your_secure_password
DB_HOST=localhost
DB_PORT=5432
DB_NAME=langgraph
DB_DRIVER=postgresql+psycopg2

# Tavily Search API
TAVILY_API_KEY=tvly-xxxxxxxxxxxxx

# Google Calendar OAuth
GOOGLE_CALENDAR_SCOPES=https://www.googleapis.com/auth/calendar.events
GOOGLE_CALENDAR_CREDENTIALS_FILE=./credentials.json
GOOGLE_CALENDAR_TOKEN_FILE=./token.json
GOOGLE_CALENDAR_DEFAULT_CALENDAR_ID=primary

Paso 2: base de datos para persistir la conversación

El asistente usa dos mecanismos de base de datos separados:

1. Checkpointing de LangGraph para el estado de la conversación (se maneja solo)

2. Una tabla propia de mensajes para registrar el historial

    

Crear la base en PostgreSQL

psql -U postgres
CREATE DATABASE langgraph; \q

Entender el esquema de la base

La implementación en models.py define una tabla Message:

class Message(Base):
    __tablename__ = "messages"

    id = Column(Integer, primary_key=True, index=True)
    _from = Column(String(50), nullable=False)
    _to = Column(String(50), nullable=False)
    content = Column(Text, nullable=False)
    created_at = Column(String(50), nullable=False)
    message_type = Column(String(20), nullable=False)  # "user" or "ai"

Esta tabla guarda el historial completo de mensajes, aparte del sistema de checkpointing de LangGraph. Esa decisión de diseño te da:

• Traza de auditoría de todas las conversaciones
• Poder consultar el historial sin depender del estado de la conversación
• Un respaldo por si se hace pruning de los checkpoints de LangGraph
• Capacidad de analítica y reportes

La tabla se crea sola cuando arranca la aplicación, con Base.metadata.create_all() de SQLAlchemy.

Checkpointing de LangGraph

El PostgresSaver de LangGraph crea sus propias tablas para el estado:

• checkpoints: guarda snapshots completos del estado de la conversación
• checkpoint_writes: registra cada mutación del estado
• checkpoint_metadata: guarda los identificadores de thread y la configuración

La implementación en assistant.py inicializa PostgresSaver con lógica de fallback:

# Try URL form first, then DSN
for candidate in (connection_string, dsn_fallback):
    try:
        cm = PostgresSaver.from_conn_string(candidate)
        self.memory = self._exit_stack.enter_context(cm)
        break
    except Exception as e:
        last_err = e

Este enfoque de doble conexión cubre distintos métodos de autenticación de PostgreSQL. El ExitStack se asegura de limpiar bien el context manager de la conexión.

Importante: PostgresSaver.from_conn_string() devuelve un context manager, no una instancia de saver. Si no entrás al contexto, vas a tener errores en runtime. La implementación usa ExitStack para manejar bien ese ciclo de vida.

Paso 3: integración con Google Calendar

La integración con el calendario muestra cómo una interfaz de lenguaje natural puede controlar lógica de negocio estructurada. El usuario expresa su intención conversando, y el sistema lo traduce a llamadas concretas a la API.

Configurar OAuth en Google Cloud

1. Entrá a la Google Cloud Console
2. Creá un proyecto nuevo o elegí uno existente
3. Habilitá la Google Calendar API en la API Library
4. Andá a APIs & Services > Credentials
5. Hacé clic en Create Credentials > OAuth 2.0 Client ID
6. Elegí Desktop application como tipo de aplicación
7. Descargá el archivo JSON de credenciales
8. Guardalo como credentials.json en tu directorio backend

Entender las tools de calendario

La implementación en tools/calendar.py expone tres funciones como tools de LangChain:

1. Crear un evento

def create_calendar_event(
    summary: str,
    start: str,
    end: str,
    description: Optional[str] = None,
    attendees: Optional[List[str]] = None,
    location: Optional[str] = None,
    calendar_id: Optional[str] = None,
) -> dict:

Esta función crea el evento y notifica automáticamente a los invitados. Detalles clave de la implementación:

• Valida que el datetime de end sea posterior al de start
• Convierte los strings ISO 8601 en objetos datetime con timezone
• Asume UTC si no le pasás timezone
• Manda las invitaciones por mail con el parámetro sendUpdates="all"
• Devuelve el recurso del evento creado, tal como lo da la Google Calendar API

2. Traer los eventos

def get_calendar_events(
    time_min: str, 
    time_max: str, 
    calendar_id: Optional[str] = None
) -> List[dict]:

Trae los eventos dentro de un rango de tiempo. La implementación:

• Usa singleEvents=True para expandir los eventos recurrentes
• Ordena por startTime para mostrarlos cronológicamente
• Devuelve los recursos crudos de la Google Calendar API

3. Borrar un evento

def delete_calendar_event(
    start_time: str, 
    calendar_id: Optional[str] = None
) -> None:

Borra eventos haciendo match por hora de inicio. El enfoque:

• Consulta los eventos que arrancan en la hora indicada
• Toma el primero que coincide (asume horas de inicio únicas)
• Lo borra vía Google Calendar API

Este diseño asume que el usuario no tiene varios eventos con exactamente la misma hora de inicio. En producción puede hacer falta lógica extra de desambiguación para los casos borde.

Flujo de OAuth y persistencia del token

La función _load_credentials() en calendar.py implementa OAuth con cache del token:

def _load_credentials() -> Credentials:
    creds: Optional[Credentials] = None

    # Load existing token if present
    if TOKEN_FILE and os.path.exists(TOKEN_FILE):
        try:
            creds = Credentials.from_authorized_user_file(TOKEN_FILE, SCOPES)
        except Exception:
            creds = None

    # Refresh if expired
    if creds and creds.expired and creds.refresh_token:
        try:
            creds.refresh(Request())
        except Exception as e:
            creds = None

    # If no valid creds available, start browser-based flow
    if not creds:
        flow = InstalledAppFlow.from_client_secrets_file(CREDENTIALS_FILE, SCOPES)
        try:
            creds = flow.run_local_server(port=0)
        except Exception:
            creds = flow.run_console()

La implementación prueba primero run_local_server() por una UX más cómoda, y cae a run_console() si el server local falla. El token queda guardado en token.json para las corridas siguientes.

Paso 4: entender la arquitectura del agente en LangGraph

El asistente usa LangGraph para implementar un workflow agéntico. Esta sección mira la implementación real para entender cómo maneja el razonamiento en varios pasos y la ejecución de tools.

Manejo del estado

El estado de la conversación en assistant/state.py usa un TypedDict mínimo:

class State(TypedDict):
    messages: Annotated[list, add_messages]

El diseño es deliberadamente mínimo. El estado contiene solo el historial de mensajes, y el reducer add_messages se encarga de acumularlos. Eso significa:

• El estado se persiste en los límites de ejecución del grafo
• Cada checkpoint es inmutable y versionado
• Ante una falla, la recuperación vuelve al último checkpoint exitoso
• El uso de memoria crece linealmente con el largo de la conversación

La estructura del grafo

La implementación en assistant.py arma un grafo con dos nodos principales:

self.graph_builder = StateGraph(State)
self.graph_builder.add_node("chat", self.chat)
self.graph_builder.add_node("tools", self.tool_node)

Nodo de chat: procesa los mensajes e invoca al modelo:

def chat(self, state: State):
    """Chat node that processes messages and generates responses."""
    return {
        "messages": [self.agent.invoke(state["messages"])]
    }

El nodo de chat es simple. Toma el historial de mensajes actual, se lo pasa al modelo vinculado y devuelve su respuesta.

Nodo de tools: ejecuta las tools pedidas con BasicToolNode:

class BasicToolNode:
    def __call__(self, inputs: dict):
        if messages := inputs.get("messages", []):
            message = messages[-1]
        else:
            raise ValueError("No message found in input")
        outputs = []
        for tool_call in message.tool_calls:
            tool_result = self.tools_by_name[tool_call["name"]].invoke(
                tool_call["args"]
            )
            outputs.append(
                ToolMessage(
                    content=json.dumps(tool_result),
                    name=tool_call["name"],
                    tool_call_id=tool_call["id"],
                )
            )
        return {"messages": outputs}

El nodo de tools extrae las llamadas del último mensaje de la IA, ejecuta cada una con los argumentos recibidos y devuelve objetos ToolMessage con los resultados.

Lógica de ruteo condicional

El ruteo entre nodos usa BasicToolNode.route_tools():

@staticmethod
def route_tools(state: State):
    if isinstance(state, list):
        ai_message = state[-1]
    elif messages := state.get("messages", []):
        ai_message = messages[-1]
    else:
        raise ValueError(f"No messages found in input state")
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "tools"
    return END

Esta función revisa si el último mensaje tiene llamadas a tools. Si las tiene, rutea al nodo de tools. Si no, termina la ejecución del grafo.

Las conexiones del grafo:

self.graph_builder.add_conditional_edges(
    "chat",
    BasicToolNode.route_tools,
    {"tools": "tools", END: END},
)
self.graph_builder.add_edge("tools", "chat")
self.graph_builder.add_edge(START, "chat")

Eso arma un loop: START → chatbot → tools (opcional) → summarize → END

El sistema no tiene un máximo de iteraciones hardcodeado. El modelo decide cuándo dejar de pedir tools. En la práctica, GPT-4o-mini rara vez necesita más de 2-3 invocaciones por request, pero en teoría un loop infinito es posible.

Binding de tools

Las tools se registran en el modelo con bind_tools de LangChain:

self.tools = [
    TavilySearch(max_results=5),
    create_calendar_event,
    get_calendar_events,
    delete_calendar_event,
]
self.agent = init_chat_model(
    "gpt-4o-mini", 
    temperature=0.5, 
    use_responses_api=True
).bind_tools(self.tools)

El modelo aprende a usar las tools desde las firmas de función, los docstrings y los type hints de los parámetros. La implementación usa temperature=0.5, que mete algo de aleatoriedad controlada sin perder consistencia al elegir tools.

El parámetro use_responses_api=True habilita el parseo de salida estructurada para las llamadas a tools, que es lo que necesita la ejecución para funcionar bien.

Paso 5: implementar el handler del webhook de Twilio

La implementación del webhook con FastAPI en main.py recibe los mensajes de WhatsApp y los rutea por el asistente.

El endpoint de mensajes

@app.post("/message")
async def receive_message(request: Request):
    form_data = await request.form()
    from_number = form_data.get("From")
    to_number = form_data.get("To")

    if not from_number or not to_number:
        raise HTTPException(
            status_code=400, 
            detail="Missing From or To fields in webhook payload"
        )

    body: str | None = None
    assistant = Assistant()

    # Detect audio media attachment and attempt transcription
    media_content_type = form_data.get("MediaContentType0")
    if media_content_type and media_content_type.startswith("audio/"):
        media_url = form_data.get("MediaUrl0")
        # ... audio handling code
    else:
        body = form_data.get("Body") or ""

    try:
        message = await assistant.generate_response(
            prompt=body, 
            from_number=from_number, 
            to_number=to_number
        )
    except Exception as e:
        print(f"[receive_message] Error generating response: {e}")
        raise HTTPException(status_code=500, detail="Failed to generate response")

    return {"status": "Message sent"}

El endpoint extrae los números de teléfono y el contenido del mensaje del payload de Twilio, detecta los mensajes de voz y los procesa según corresponda.

Transcripción de mensajes de voz

El código que maneja el audio descarga el archivo y lo transcribe:

# Twilio media URLs require basic auth
account_sid = os.getenv("TWILIO_ACCOUNT_SID")
auth_token = os.getenv("TWILIO_AUTH_TOKEN")

try:
    resp = requests.get(
        media_url, 
        auth=(account_sid, auth_token), 
        timeout=30
    )
    resp.raise_for_status()
    audio_bytes = resp.content
    
    # Derive filename extension from content-type
    ext = "ogg" if media_content_type == "audio/ogg" else media_content_type.split("/")[-1][:5]
    transcript = await assistant.transcribe_audio(audio_bytes, filename=f"voice.{ext}")
    
    body = transcript.strip() or "(Unintelligible audio or empty transcription)"
except Exception as e:
    print(f"[receive_message] Audio transcription failed: {e}")
    body = "(Error transcribing audio message)"

La implementación usa HTTP Basic Auth para bajar el archivo del storage temporal de Twilio. Los bytes del audio van al método de transcripción del asistente.

La implementación de la transcripción en assistant.py:

async def transcribe_audio(self, audio_bytes: bytes, filename: str = "audio.ogg") -> str:
    # Wrap bytes in a file-like object with a name attr
    audio_file_obj = io.BytesIO(audio_bytes)
    audio_file_obj.name = filename

    try:
        model = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))

        transcription = await model.audio.transcriptions.create(
            model="whisper-1",
            file=audio_file_obj,
            response_format="text",
        )
        return transcription or ""
    except Exception as e:
        print(f"[transcribe_audio] Failed to transcribe audio: {e}")
        return ""

Whisper tiene un límite de 25MB por archivo. La implementación envuelve los bytes en un objeto BytesIO con atributo name, porque la librería de OpenAI necesita objetos tipo archivo con nombre para inferir el formato.

El parámetro response_format="text" devuelve el texto transcripto plano en vez de metadata JSON. Para aplicaciones conversacionales es más simple de manejar.

Procesar los mensajes con el agente

El método generate_response() en assistant.py orquesta toda la interacción:

async def generate_response(self, prompt: str, from_number: str, to_number: str) -> str:
    # Store prompt in DB
    db = SessionLocal()
    try:
        msg_record = Message(
            _from=from_number,
            _to=to_number,
            content=prompt,
            created_at=date.today().isoformat(),
            message_type="user"
        )
        db.add(msg_record)
        db.commit()
    except Exception as e:
        print(f"Error saving incoming message to DB: {e}")
    finally:
        db.close()
    
    # Create messages with system instructions
    messages = [
        {
            "role": "system", 
            "content": self.assistant_instructions + 
                      f"\n\nCurrent date is {date.today().isoformat()} and default timezone is UTC -3 (ART)."
        }
    ]
    
    # Add current user message
    messages.append({"role": "user", "content": prompt})
    
    # Use phone number as thread ID for persistent memory
    config = {"configurable": {"thread_id": from_number}}

    final_response = ""
    
    for step in self.graph.stream({"messages": messages}, config, stream_mode="messages"):
        # Process streaming responses
        if isinstance(step, tuple) and len(step) == 2:
            message_chunk, metadata = step
            
            if (hasattr(message_chunk, 'content') and message_chunk.content):
                content = message_chunk.content
                
                # Handle both string and list content
                if isinstance(content, list):
                    text_content = ""
                    for item in content:
                        if isinstance(item, dict) and 'text' in item:
                            text_content += item['text']
                        elif isinstance(item, str):
                            text_content += item
                        elif hasattr(item, 'text'):
                            text_content += item.text
                    content = text_content
                
                # Filter out JSON responses and empty content
                if content and not str(content).startswith('{'):
                    final_response += content

Las decisiones de arquitectura clave:

1. Thread ID: usa from_number como identificador del thread, así cada usuario tiene su propio thread persistente en el sistema de checkpoints de LangGraph.
2. Instrucciones del sistema: carga el prompt desde prompts/_evo_001 y le suma la fecha y la timezone actuales. El system prompt se carga una sola vez al inicializar, no en cada request.
3. Modo streaming: usa stream_mode="messages", que va devolviendo los chunks del mensaje a medida que se generan. La Responses API puede devolver el contenido como lista de bloques, así que la implementación maneja strings y listas.
4. Filtrado de respuestas: descarta las respuestas JSON, que serían metadata de llamadas a tools, para no mandarle datos estructurados al usuario.

Una vez armada la respuesta completa, la implementación la manda por Twilio y la guarda en la base de historial.

Paso 6: configurar Twilio WhatsApp

Para recibir mensajes de usuarios de WhatsApp, configurá Twilio para que mande los webhooks a tu aplicación.

Usar el sandbox de WhatsApp (desarrollo)

1. Entrá a tu Twilio Console
2. Andá a Messaging > Try it out > Send a WhatsApp message
3. Seguí las instrucciones para unirte al sandbox mandando el código indicado
4. Configurá la URL del webhook para los mensajes entrantes

Configurar ngrok para desarrollo local

Como Twilio necesita una URL pública para mandar los webhooks, usá ngrok para exponer tu server local:

ngrok http 8000

ngrok te da una URL pública tipo https://abc123.ngrok.io

Configurar el webhook

En la Twilio Console:

1. Andá a la configuración del sandbox de WhatsApp
2. Poné el webhook de "When a message comes in" en: https://your-ngrok-url.ngrok.io/message
3. Poné el método en HTTP POST
4. Guardá la configuración

Ahora, cuando alguien le mande un mensaje a tu número de WhatsApp de Twilio, va a llegar a tu aplicación FastAPI.

Despliegue en producción

Para producción:

• Desplegá tu app FastAPI en un proveedor cloud con dominio público
• Pedí una cuenta de Twilio WhatsApp Business (es obligatoria en producción)
• Configurá la URL del webhook de producción en Twilio
• Implementá validación de firma del webhook para evitar spoofing
• Armá monitoreo y logging
• Implementá rate limiting

Paso 7: probar el asistente

Levantar la aplicación

cd backend
uvicorn main:app --reload --host 0.0.0.0 --port 8000

Tenés que ver:

INFO:     Uvicorn running on http://0.0.0.0:8000
INFO:     Application startup complete.

Probar distintos escenarios

1. Conversación simple

You: Hello!
Assistant: [Responds with greeting and capabilities]

2. Búsqueda web

You: What's the latest AI news?
Assistant: [Searches via Tavily and provides current information]

3. Manejo del calendario

You: Schedule a meeting tomorrow at 2 PM with john@example.com
Assistant: [Creates calendar event and confirms]

4. Mensaje de voz

You: [Sends voice message]
Assistant: [Transcribes and responds to content]

5. Retención de contexto

You: Schedule a meeting with Sarah tomorrow at 3 PM
Assistant: [Creates event]
You: Actually, make that 4 PM
Assistant: [Understands "that" refers to Sarah's meeting and updates]

La retención de contexto muestra el checkpointing de LangGraph manteniendo el estado entre turnos.

Rendimiento y escalabilidad

Fuentes de latencia

La implementación tiene varias fuentes de latencia:

1. Transcripción con Whisper: 2-5 segundos para un mensaje de voz típico
2. Streaming de LangGraph: suma un overhead mínimo (50-100ms)
3. Ejecución de tools: variable según la tool (Calendar API: 200-500ms, Tavily: 500-1000ms)
4. Escrituras en la base: impacto mínimo con connection pooling
5. Envío del mensaje por Twilio: 200-500ms

El tiempo total de respuesta suele ir de 2-8 segundos, según si se invocan tools o no.

Optimización de la base

La implementación usa el connection pooling de SQLAlchemy para la base de historial:

engine = create_engine(
    url, 
    connect_args={"options": "-c client_encoding=UTF8"}, 
    pool_pre_ping=True
)

El parámetro pool_pre_ping=True valida las conexiones antes de usarlas, y así evita errores por conexiones viejas.

Para el sistema de checkpointing de LangGraph, mirá:

• Sumar índices en thread_id para recuperar checkpoints más rápido
• Hacer pruning de checkpoints de conversaciones de más de N días
• Monitorear cuánto crece la tabla de checkpoints

Optimización de costo

Desglose de costo por turno de conversación:

• GPT-4o-mini: ~$0.0001-0.0005 por mensaje (según el largo del contexto)
• Whisper: ~$0.006 por minuto de audio
• Twilio WhatsApp: $0.005 por mensaje entrante + $0.005 saliente
• Tavily Search: variable según el plan

La implementación usa temperature=0.5, que da respuestas consistentes sin disparar el uso de tokens. Para bajar más el costo, mirá:

• Resumir la conversación para achicar la ventana de contexto
• Cachear las consultas frecuentes
• Usar system prompts más cortos
• Podar los mensajes viejos del historial

Concurrencia y escalado

La implementación actual crea una instancia nueva de Assistant por cada request del webhook:

assistant = Assistant()

Es seguro para requests concurrentes porque cada instancia maneja sus propias conexiones con ExitStack. Igual, el enfoque tiene trade-offs:

• Pro: no hay estado compartido entre requests
• Pro: el aislamiento de errores es simple
• Contra: se paga la inicialización de PostgresSaver en cada request
• Contra: se lee el archivo del system prompt una y otra vez

Para más tráfico, mirá:

• Usar inyección de dependencias para compartir una sola instancia de Assistant
• Implementar connection pooling para PostgresSaver
• Cachear el system prompt en memoria

Features avanzadas y personalización

Sumar tools propias

Extendé el asistente sumando tools nuevas. Ejemplo de una tool de clima:

def get_weather(location: str) -> dict:
    """
    Get current weather for a location.
    
    Args:
        location: City name or ZIP code
    
    Returns:
        Dictionary with weather information
    """
    # Implement weather API call
    return {
        "location": location,
        "temperature": 72,
        "condition": "Sunny"
    }

Sumala a la lista de tools en assistant.py:

self.tools = [
    TavilySearch(max_results=5),
    create_calendar_event,
    get_calendar_events,
    delete_calendar_event,
    get_weather,  # New tool
]

El modelo aprende solo a usar las tools nuevas a partir de sus docstrings y type hints. Asegurate de que los docstrings expliquen claro cuándo usar cada tool y qué parámetros necesita.

Personalizar el system prompt

El system prompt se carga desde prompts/_evo_001. Editá ese archivo para:

• Cambiar la personalidad del asistente
• Sumar conocimiento del dominio
• Implementar reglas de negocio
• Definir formatos de respuesta

La implementación le suma la fecha y la timezone actuales al prompt:

content = self.assistant_instructions + 
         f"\n\nCurrent date is {date.today().isoformat()} and default timezone is UTC -3 (ART)."

Ajustá la timezone según tus usuarios.

Problemas comunes

Error de context manager en PostgresSaver

Problema: AttributeError: '_GeneratorContextManager' object has no attribute 'get_next_version'

Solución: pasa cuando le pasás el context manager directo en vez de entrar en él. La implementación usa bien ExitStack.enter_context():

cm = PostgresSaver.from_conn_string(candidate) self.memory = self._exit_stack.enter_context(cm)

Errores de conexión a la base

Problema: psycopg2.OperationalError: could not connect to server

Solución: verificá que PostgreSQL esté corriendo y que las credenciales sean correctas. La implementación prueba los formatos URL y DSN:

connection_string = f"postgresql://{_db_user}:{_db_pass}@{_db_host}:{_db_port}/{_db_name}" dsn_fallback = f"host={_db_host} port={_db_port} dbname={_db_name} user={_db_user} password={_db_pass}"

Problemas de OAuth con Google Calendar

Problema: las operaciones de calendario fallan con errores de autenticación

Solución: borrá token.json y volvé a autenticarte. Verificá que la Calendar API esté habilitada en Google Cloud Console. La implementación cae a OAuth por consola si falla el flujo con server local.

Timeout del webhook de Twilio

Problema: Twilio muestra errores de timeout del webhook

Solución: los webhooks de Twilio tienen un timeout de 10 segundos. La implementación hace streaming de la respuesta pero no procesa en async. Para operaciones largas, mirá:

@app.post("/message")
async def receive_message(request: Request):
    # Send immediate acknowledgment
    threading.Thread(
        target=process_message_async,
        args=(from_number, to_number, body)
    ).start()
    
    return {"status": "Processing"}

Los mensajes de voz no se transcriben

Problema: la transcripción falla o devuelve texto vacío

Solución: revisá que:

• La API key de OpenAI tenga acceso a Whisper
• El formato de audio esté soportado (Twilio manda OGG/Opus)
• El archivo pese menos de 25MB
• El parámetro filename tenga la extensión correcta
• La implementación maneja los errores devolviendo un string vacío:

except Exception as e:
    print(f"[transcribe_audio] Failed to transcribe audio: {e}")
    return ""

Buenas prácticas de seguridad

1. Validá los requests de Twilio: verificá la firma del webhook para evitar spoofing
2. Rate limiting: poné límites por usuario
3. Sanitización de entrada: la implementación le pasa el input del usuario directo al LLM, lo cual suele ser seguro, pero en producción conviene validar más
4. Credenciales seguras: usá variables de entorno para todo dato sensible
5. Control de acceso: sumá autenticación si manejás operaciones sensibles
6. Auditoría: la tabla de historial ya te da una capacidad básica de auditoría
7. Cifrado: evaluá cifrar los datos sensibles en PostgreSQL

Casos de uso reales

Automatizar el soporte al cliente

Desplegá este asistente como primera línea de soporte al cliente:

• Responder preguntas frecuentes automáticamente con búsqueda web
• Agendar llamadas de soporte con la integración de calendario
• Mantener el contexto de la conversación entre sesiones
• Escalar los casos complejos a agentes humanos

Asistente de productividad personal

Convertí el sistema en un asistente personal:

• Manejar el calendario y agendar reuniones
• Poner recordatorios y seguir tareas
• Buscar información a demanda
• Procesar mensajes de voz mientras hacés otra cosa

Automatización de procesos de negocio

Adaptalo a flujos de negocio:

• Calificar leads con preguntas conversacionales
• Agendar demos y reuniones automáticamente
• Responder preguntas de producto buscando en la base de conocimiento
• Juntar y estructurar información de las conversaciones

Conclusión

Esta implementación muestra un asistente de WhatsApp con IA capaz de ir a producción. Está construido sobre GPT-4 de OpenAI, los workflows agénticos de LangGraph y la infraestructura de mensajería de Twilio. La arquitectura resuelve varios requerimientos técnicos:

• Comportamiento agéntico: el sistema razona qué tools usar en vez de seguir flujos predefinidos
• Persistencia de estado: el enfoque de dos bases mantiene el estado y el historial
• Procesamiento multimodal: texto y voz se manejan igual, gracias a la transcripción automática
• Integración de tools: el diseño modular habilita calendario, búsqueda web y extensiones
• Consideraciones de producción: manejo de errores, de conexiones y degradación elegante

Las decisiones de arquitectura clave tienen impactos medibles. El checkpointing de LangGraph cambia almacenamiento por confiabilidad. GPT-4o-mini equilibra costo y capacidad. Manejar los mensajes por webhook mete restricciones de latencia que afectan la experiencia.

Entender esos trade-offs te deja decidir con información cuando adaptes esta arquitectura a tus requerimientos. La implementación es una base que podés extender con más tools, otros modelos, o integrar en sistemas más grandes.

Próximos pasos

Para extender este asistente:

1. Sumar tools: integrar CRMs, sistemas de mail o plataformas de gestión de proyectos
2. Implementar autenticación: verificar al usuario en las operaciones sensibles
3. Analítica: seguir patrones de conversación y métricas de uso de tools
4. Testing: tests unitarios para las tools y de integración para los flujos
5. Síntesis de voz: generar respuestas habladas con text-to-speech
6. Visión: sumar GPT-4 Vision para analizar imágenes
7. Multi-idioma: soportar varios idiomas con detección automática
8. Modelos propios: hacer fine-tuning para casos de uso del dominio

Recursos adicionales

• Documentación de LangGraph
• Guía de la API de WhatsApp de Twilio
• Referencia de la API de OpenAI
• Documentación de FastAPI
• Guía de la API de Google Calendar
• Código del repositorio