Saltar al contenido principal

FastAPI + OpenAI chat

Construye un servicio FastAPI mínimo que expone POST /chat, hidrata contexto desde iLAB Memory para el usuario que pide, llama a gpt-4o-mini y persiste el intercambio — todo en un único archivo Python.

Lo que necesitás​

Python 3.11+

from __future__ import annotations + type hints modernos en todo.

ilab-memory

pip install ilab-memory — librería embebida, sin servidor remoto.

openai

pip install openai — SDK oficial de Python (>=1.30).

fastapi + uvicorn

pip install "fastapi[standard]" — trae uvicorn para fastapi dev.

Arquitectura​

 ┌───────────┐   POST /chat   ┌────────────────┐
 │  cliente  │ ─────────────▶ │  app FastAPI   │
 └───────────┘                │  (chat_app.py) │
                              └──┬─────────┬───┘
                                 │         │
                hidrata contexto │         │  guarda turno
                                 ▼         ▼
                         ┌────────────────────┐
                         │   ILabMemory       │
                         │   (./memory.db)    │
                         └────────────────────┘
                                 ▲
                                 │ system prompt + historial
                                 │
                         ┌────────────────────┐
                         │   OpenAI gpt-4o-mini│
                         └────────────────────┘

La app es la dueña del loop. iLAB Memory se invoca de forma determinista — el LLM nunca decide cuándo leer o escribir memoria.

Implementación​

1. Bootstrap de la app​

Construí un ILabMemory singleton al arranque usando from_path y el hook lifespan. La ruta de la DB sale de ILAB_MEMORY_DB_PATH para que en producción puedas montar un volumen persistente.

from contextlib import asynccontextmanager
from fastapi import FastAPI
from ilab_memory import ILabMemory

DB_PATH = os.environ.get("ILAB_MEMORY_DB_PATH", "./memory.db")

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.mem = ILabMemory.from_path(DB_PATH)
    try:
        yield
    finally:
        app.state.mem.close()

app = FastAPI(lifespan=lifespan)

2. Definí el schema de request​

Un ChatRequest lleva el user_id (la tenant key — ver Receta 3) y el mensaje del usuario. Pydantic valida ambos lados gratis.

from pydantic import BaseModel, Field

class ChatRequest(BaseModel):
    user_id: str = Field(..., min_length=1)
    message: str = Field(..., min_length=1)

3. Hidratá el contexto desde memoria​

mem_session_start es el punto de entrada de hidratación. Reusa una sesión activa, devuelve los últimos 5 resúmenes de sesión y las observaciones más recientes scoreadas con ContextScore.

session = mem.mem_session_start(user_id=req.user_id)
history_lines = [f"- ({m.type}) {m.title}" for m in session.memories]

4. Componé el system prompt​

Inyectá la memoria hidratada en el system prompt. Mantenelo corto — el LLM ve el resumen, no el corpus completo.

system_prompt = (
    "Sos un asistente útil. Usá contexto previo solo si es relevante.\n"
    f"Perfil del usuario / memoria pasada:\n{chr(10).join(history_lines) or '(ninguna aún)'}"
)

5. Llamá a OpenAI​

Usá el cliente async. El SDK lee OPENAI_API_KEY del entorno automáticamente.

from openai import AsyncOpenAI

client = AsyncOpenAI()

completion = await client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": req.message},
    ],
)
reply = completion.choices[0].message.content or ""

6. Persistí el turno​

Guardá el intercambio como observación discovery. El tag <private> strippea secretos ANTES de que el contenido se hashee y persista.

mem.mem_save(
    user_id=req.user_id,
    type="discovery",
    title=f"Turno: {req.message[:60]}",
    content=f"Usuario: {req.message}\nAsistente: {reply}",
    topic_key=f"chat/{req.user_id}/turn",
)
tip

Trade-off — topic_key por usuario vs por turno:

  • Mismo topic_key (p.ej. chat/{user_id}/turn) → una observación rolling que se actualiza in-place. Almacenamiento más barato, menos filas, pero perdés el historial individual de cada turno.
  • Sin topic_key → una observación por turno. Historial completo, recall completo de search, pero la tabla crece linealmente con el tráfico.

Elegí por turno cuando necesitás replay del timeline; por usuario cuando solo te importa el último estado.

7. Devolvé la respuesta​

Envolvé la respuesta en un JSON. Incluí el session_id para que el cliente correlacione logs.

return {"session_id": session.session_id, "reply": reply}
nota

mem_session_start y mem_save son síncronos (las escrituras subyacentes a SQLite son bloqueantes). FastAPI lo maneja bien dentro de un endpoint async para tráfico bajo. Para alto throughput, ejecutalos en run_in_threadpool para no bloquear el event loop.

aviso

Nunca loggees el req.user_id o req.message crudo sin redactar primero. La memoria strippea bloques <private>, pero los logs no.

Archivo completo​

Código completo (listo para copiar y pegar)
from __future__ import annotations

import os
from contextlib import asynccontextmanager
from typing import AsyncIterator

from fastapi import FastAPI
from openai import AsyncOpenAI
from pydantic import BaseModel, Field

from ilab_memory import ILabMemory

DB_PATH = os.environ.get("ILAB_MEMORY_DB_PATH", "./memory.db")
MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    app.state.mem = ILabMemory.from_path(DB_PATH)
    app.state.openai = AsyncOpenAI()  # lee OPENAI_API_KEY del entorno
    try:
        yield
    finally:
        app.state.mem.close()


app = FastAPI(lifespan=lifespan, title="iLAB Memory Chat")


class ChatRequest(BaseModel):
    user_id: str = Field(..., min_length=1)
    message: str = Field(..., min_length=1)


class ChatResponse(BaseModel):
    session_id: str
    reply: str


@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest) -> ChatResponse:
    mem: ILabMemory = app.state.mem
    client: AsyncOpenAI = app.state.openai

    # 1. Hidratar memoria por usuario
    session = mem.mem_session_start(user_id=req.user_id)
    history_lines = [f"- ({m.type}) {m.title}" for m in session.memories]

    # 2. Componer system prompt con contexto previo
    system_prompt = (
        "Sos un asistente útil. Usá contexto previo solo si es relevante.\n"
        f"Perfil del usuario / memoria pasada:\n{chr(10).join(history_lines) or '(ninguna aún)'}"
    )

    # 3. Llamar al LLM
    completion = await client.chat.completions.create(
        model=MODEL,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": req.message},
        ],
    )
    reply = completion.choices[0].message.content or ""

    # 4. Persistir el turno (type='discovery', sin topic_key — una fila por turno)
    mem.mem_save(
        user_id=req.user_id,
        type="discovery",
        title=f"Turno: {req.message[:60]}",
        content=f"Usuario: {req.message}\nAsistente: {reply}",
    )

    return ChatResponse(session_id=session.session_id, reply=reply)


# Correr con: fastapi dev chat_app.py

Probalo desde otra terminal:

curl -X POST http://127.0.0.1:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"user_id":"alice","message":"Hola, soy Alice y me encanta el senderismo."}'

La próxima llamada del mismo user_id verá Hola, soy Alice en memories[] automáticamente.

¿Y ahora?​

Agente de larga duración

Manejá contexto a través de 50+ turnos con summarization periódica y filtros por type.

Aislamiento multi-usuario

Acotá memoria por tenant con patrones estables de user_id.

Sesiones

Cómo mem_session_start reusa, auto-cierra e hidrata contexto.

Privacidad

Qué pasa con <private>...</private> antes de que el contenido toque disco.