Async Patterns

Memori works with Python's async/await out of the box. Use AsyncOpenAI or AsyncAnthropic instead of their sync counterparts — everything else stays the same.

When to Use Async

Scenario	Async?	Why
Web servers (FastAPI)	Yes	Concurrent request handling
Chatbots with many users	Yes	Non-blocking I/O
CLI scripts	No	Sync is simpler
Jupyter notebooks	No	Event loop already running

Basic Async Setup

import os
import asyncio
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from memori import Memori
from openai import AsyncOpenAI

engine = create_engine("sqlite:///memori.db")
SessionLocal = sessionmaker(bind=engine)

async def main():
    client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    mem = Memori(conn=SessionLocal).llm.register(client)
    mem.attribution(entity_id="user_123", process_id="async_agent")
    mem.config.storage.build()

    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "I prefer async Python."}]
    )
    print(response.choices[0].message.content)
    mem.augmentation.wait()

asyncio.run(main())

Works identically with AsyncAnthropic — just swap the client.

FastAPI Example

import os
from fastapi import FastAPI
from pydantic import BaseModel
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from memori import Memori
from openai import AsyncOpenAI

app = FastAPI()
engine = create_engine("sqlite:///memori.db", connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)
Memori(conn=SessionLocal).config.storage.build()

class ChatRequest(BaseModel):
    message: str

@app.post("/chat/{user_id}")
async def chat(user_id: str, req: ChatRequest):
    client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    mem = Memori(conn=SessionLocal).llm.register(client)
    mem.attribution(entity_id=user_id, process_id="fastapi_async")

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

Thread Safety

Pattern	Safe?	Why
`conn=SessionLocal` (factory)	Yes	New session per operation
`conn=lambda: existing_session`	No	Shares one session

For production async apps, use PostgreSQL with larger pools:

engine = create_engine(
    "postgresql+psycopg://user:pass@host/db",
    pool_pre_ping=True,
    pool_size=20,
    max_overflow=40,
    pool_recycle=300
)