Cookbook: general AI memory

Use this for: AI assistants, productivity tools, coding agents, SaaS personalization, onboarding and support chat that doesn’t need the Support schema. The General Engine stores facts, preferences, goals, procedures, relationships, and expertise per external_user_id inside your tenant.

Flow

User message → retrieve governed context → customer LLM call → queue completed turn

import os
from memoryos import Memory, MemoryOSError
from openai import OpenAI

BASE_PROMPT = "You are a helpful product assistant."

def answer_user(external_user_id: str, user_message: str) -> str:
    memory = Memory(api_key=os.environ["MEMORYOS_API_KEY"])
    openai = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

    system_prompt = BASE_PROMPT
    try:
        memories = memory.get(
            query=user_message,
            external_user_id=external_user_id,
            limit=5,
            context_max_tokens=500,
        )
        if memories.has_context:
            system_prompt = f"{BASE_PROMPT}\n\n{memories.system_prompt_addition}"
    except MemoryOSError:
        pass  # Log in server telemetry; continue the customer conversation.

    response = openai.responses.create(
        model="gpt-4.1-mini",
        input=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_message},
        ],
    )

    try:
        memory.add(
            messages=[
                {"role": "user", "content": user_message},
                {"role": "assistant", "content": response.output_text},
            ],
            external_user_id=external_user_id,
            metadata={"source": "chat"},
        )
    except MemoryOSError:
        pass  # The answer is already available; retry ingestion separately if desired.
    finally:
        memory.close()
    return response.output_text

import OpenAI from "openai";
import { MemoryOS, MemoryOSError } from "memoryo-sdk";

const BASE_PROMPT = "You are a helpful product assistant.";

export async function answerUser(externalUserId: string, userMessage: string): Promise<string> {
  const memory = new MemoryOS(process.env.MEMORYOS_API_KEY!);
  const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY! });

  let systemPrompt = BASE_PROMPT;
  try {
    const memories = await memory.get({
      query: userMessage,
      externalUserId,
      limit: 5,
      contextMaxTokens: 500,
    });
    if (memories.hasContext) {
      systemPrompt = `${BASE_PROMPT}\n\n${memories.systemPromptAddition}`;
    }
  } catch (error) {
    if (!(error instanceof MemoryOSError)) throw error;
    // Log in server telemetry; continue the customer conversation.
  }

  const response = await openai.responses.create({
    model: "gpt-4.1-mini",
    input: [
      { role: "system", content: systemPrompt },
      { role: "user", content: userMessage },
    ],
  });

  try {
    await memory.add(
      [
        { role: "user", content: userMessage },
        { role: "assistant", content: response.output_text },
      ],
      externalUserId,
      undefined,
      { source: "chat" },
    );
  } catch (error) {
    if (!(error instanceof MemoryOSError)) throw error;
    // The answer is already available; retry ingestion separately if desired.
  }

  return response.output_text;
}

Production handling

Always treat MemoryOS as additive. Your app should answer even when memory is empty, passthrough, or unavailable.

memories = memory.get(query=user_message, external_user_id=external_user_id)

prompt_addition = ""
if memories.has_context:
    prompt_addition = memories.system_prompt_addition

For writes, check status before assuming a memory was stored:

result = memory.add(messages=messages, external_user_id=external_user_id)

if result.was_queued:
    pass  # queued for extraction
elif result.nothing_to_extract:
    pass  # valid conversation, nothing durable to store
elif result.status == "passthrough":
    pass  # write skipped by quota mode

What to send to `add()`

Send meaningful turns. The quality gate blocks filler. Good: "I prefer short technical answers with Python examples." or "My main goal is to prepare for the AWS Solutions Architect exam." Poor: "hi", "ok", empty messages, repeated copies of the same turn.

Retrieval query design

Use the current task as the query, not a generic term.

# Good
memory.get(query="Write a follow-up email for this customer", external_user_id="customer-123")

# Poor
memory.get(query="preferences", external_user_id="customer-123")

Getting Started

Core Memory

Domain Memory

Reliability

Legal and Trust

Cookbook: general AI memory

Flow

Production handling

What to send to `add()`

Retrieval query design

​Flow

​Production handling

​What to send to add()

​Retrieval query design

​Related pages

Flow

Production handling

What to send to `add()`

Retrieval query design

Related pages