Cookbook

Complete, copy-paste-ready examples for common use cases.

---

Basic Usage

The simplest integration: write facts from user messages and retrieve context for responses.

import asyncio
from arandu import MemoryClient
from arandu.providers.openai import OpenAIProvider


async def main():
    provider = OpenAIProvider(api_key="sk-...")
    memory = MemoryClient(
        database_url="postgresql+psycopg://memory:memory@localhost:5432/memory",
        llm=provider,
        embeddings=provider,
    )
    await memory.initialize()

    try:
        # Simulate a conversation
        messages = [
            "Hi, I'm Rafael. I'm a backend engineer at Acme Corp in São Paulo.",
            "My girlfriend Ana is a UX designer. We have a cat named Pixel.",
            "I've been learning Rust lately, mostly on weekends.",
            "Actually, I just moved to Rio de Janeiro. Still remote at Acme.",
        ]

        for msg in messages:
            result = await memory.write(agent_id="rafael", message=msg, speaker_name="Rafael")
            added = len(result.facts_added)
            updated = len(result.facts_updated)
            print(f"Write: +{added} facts, ~{updated} updates ({result.duration_ms:.0f}ms)")

        # Retrieve context for different queries
        queries = [
            "where does Rafael live?",
            "tell me about Rafael's relationships",
            "what are Rafael's hobbies?",
        ]

        for query in queries:
            result = await memory.retrieve(agent_id="rafael", query=query)
            print(f"\nQuery: {query}")
            print(f"Found {len(result.facts)} facts ({result.duration_ms:.0f}ms)")
            for fact in result.facts[:5]:
                print(f"  [{fact.score:.2f}] {fact.entity_name}: {fact.fact_text}")
    finally:
        await memory.close()


asyncio.run(main())

---

Using Anthropic (Claude)

Use Claude as your LLM while keeping OpenAI for embeddings (Anthropic doesn't offer an embeddings API):

import asyncio
from arandu import MemoryClient, MemoryConfig
from arandu.providers.anthropic import AnthropicProvider
from arandu.providers.openai import OpenAIProvider


async def main():
    # Claude for reasoning, OpenAI for embeddings
    llm = AnthropicProvider(api_key="sk-ant-...", model="claude-sonnet-4-20250514")
    embeddings = OpenAIProvider(api_key="sk-...")

    memory = MemoryClient(
        database_url="postgresql+psycopg://memory:memory@localhost/memory",
        llm=llm,
        embeddings=embeddings,
    )
    await memory.initialize()

    try:
        result = await memory.write(
            agent_id="demo",
            message="I love hiking in the mountains. Last weekend I went to Serra da Mantiqueira.",
            speaker_name="Rafael",
        )
        print(f"Extracted {len(result.facts_added)} facts using Claude")

        context = await memory.retrieve(agent_id="demo", query="outdoor activities")
        print(context.context)
    finally:
        await memory.close()


asyncio.run(main())

---

Using Other LLM Providers

OpenAIProvider works with any provider that exposes an OpenAI-compatible API. Just change api_key, model, and base_url:

from arandu import MemoryClient
from arandu.providers.openai import OpenAIProvider

# DeepSeek V3 (cheap, high quality)
llm = OpenAIProvider(
    api_key="sk-deepseek-...",
    model="deepseek-chat",
    base_url="https://api.deepseek.com/v1",
)

# Groq (fast inference)
llm = OpenAIProvider(
    api_key="gsk_...",
    model="llama-3.3-70b-versatile",
    base_url="https://api.groq.com/openai/v1",
)

# Together AI
llm = OpenAIProvider(
    api_key="tog_...",
    model="meta-llama/Llama-3.3-70B-Instruct-Turbo",
    base_url="https://api.together.xyz/v1",
)

# Fireworks AI
llm = OpenAIProvider(
    api_key="fw_...",
    model="accounts/fireworks/models/llama-v3p3-70b-instruct",
    base_url="https://api.fireworks.ai/inference/v1",
)

# Local model via Ollama
llm = OpenAIProvider(
    api_key="ollama",  # any non-empty string
    model="llama3.1",
    base_url="http://localhost:11434/v1",
)

# Use with Arandu (same as OpenAI)
memory = MemoryClient(
    database_url="postgresql+psycopg://...",
    llm=llm,
    embeddings=OpenAIProvider(api_key="sk-..."),  # embeddings still need OpenAI
)

Embeddings still require OpenAI (or a custom provider)

Most OpenAI-compatible providers don't offer embedding APIs. Use OpenAIProvider with your real OpenAI key for embeddings, or implement a custom EmbeddingProvider (see Custom Providers).

---

Advanced Configuration (Retrieval Tuning)

Fine-tune retrieval for different use cases:

import asyncio
from arandu import MemoryClient, MemoryConfig
from arandu.providers.openai import OpenAIProvider


async def main():
    # Single provider for all LLM operations (extraction, reranker, etc.)
    llm = OpenAIProvider(api_key="sk-...", model="gpt-4o")

    # Configuration for a chatbot that needs broad, fresh context
    config = MemoryConfig(
        # Extraction: tight timeout for real-time chat
        extraction_timeout_sec=15.0,

        # Retrieval: more results, favor recency
        topk_facts=40,
        min_similarity=0.15,          # cast a wider net
        recency_half_life_days=7,     # favor recent facts more aggressively

        # Score weights: boost recency for a fast-moving conversation
        score_weights={
            "semantic": 0.50,
            "recency": 0.35,
            "importance": 0.15,
        },

        # Reranker
        enable_reranker=True,

        # Context: larger budget for rich responses
        context_max_tokens=3000,

        # Spreading activation: wider context expansion
        spreading_activation_hops=3,
        spreading_max_related_entities=8,

        # Timezone for recency calculations
        timezone="America/Sao_Paulo",
    )

    memory = MemoryClient(
        database_url="postgresql+psycopg://memory:memory@localhost/memory",
        llm=llm,
        embeddings=llm,
        config=config,
    )
    await memory.initialize()

    try:
        # Write a series of messages
        await memory.write(agent_id="demo", message="I started a new job at TechCorp today!", speaker_name="Rafael")
        await memory.write(agent_id="demo", message="My manager's name is Sarah. She seems great.", speaker_name="Rafael")
        await memory.write(agent_id="demo", message="The office is in downtown with a nice view.", speaker_name="Rafael")

        # Retrieve with tuned settings
        result = await memory.retrieve(agent_id="demo", query="what's new with the user?")
        print(f"Retrieved {len(result.facts)} facts")
        print(f"Context ({len(result.context)} chars):")
        print(result.context)

        # Check individual scores to verify tuning
        for fact in result.facts:
            print(f"\n  [{fact.score:.3f}] {fact.fact_text}")
            print(f"    Scores: {fact.scores}")
    finally:
        await memory.close()


asyncio.run(main())

---

Background Jobs Integration

Set up periodic maintenance to keep memory organized:

import asyncio
from arandu import (
    MemoryClient,
    MemoryConfig,
    cluster_user_facts,
    compute_entity_importance,
    detect_communities,
    refresh_entity_summaries,
    run_consolidation,
    run_memify,
)
from arandu.providers.openai import OpenAIProvider
from arandu.db import create_engine, create_session_factory


async def run_maintenance(
    database_url: str,
    agent_ids: list[str],
    provider: OpenAIProvider,
    config: MemoryConfig,
) -> None:
    """Run all background maintenance jobs for a list of users."""
    engine = create_engine(database_url)
    session_factory = create_session_factory(engine)

    try:
        async with session_factory() as session:
            for agent_id in agent_ids:
                print(f"\n--- Maintenance for {agent_id} ---")

                # 1. Importance scoring (cheap, SQL-only)
                importance = await compute_entity_importance(session, agent_id, config)
                print(f"  Importance: scored {importance.entities_scored} entities")

                # 2. Summary refresh (moderate, LLM)
                summaries = await refresh_entity_summaries(
                    session, agent_id, provider, config
                )
                print(f"  Summaries: refreshed {summaries.summaries_refreshed}")

                # 3. Clustering (moderate, LLM)
                clusters = await cluster_user_facts(
                    session, agent_id, provider, provider, config
                )
                print(f"  Clustering: {clusters.clusters_created} clusters")

                # 4. Community detection
                communities = await detect_communities(
                    session, agent_id, provider, provider, config
                )
                print(f"  Communities: {communities.communities_created} created")

                # 5. Consolidation (moderate, LLM)
                consolidation = await run_consolidation(session, agent_id, provider, config)
                print(f"  Consolidation: {consolidation.observations_created} observations")

                # 6. Memify (moderate, LLM)
                memify = await run_memify(session, agent_id, provider, provider, config)
                print(f"  Memify: {memify.facts_scored} facts scored")

            await session.commit()
    finally:
        await engine.dispose()


async def main():
    provider = OpenAIProvider(api_key="sk-...")
    config = MemoryConfig()
    database_url = "postgresql+psycopg://memory:memory@localhost/memory"

    # Run once
    await run_maintenance(database_url, ["user_123", "user_456"], provider, config)

    # Or schedule with asyncio
    # while True:
    #     await run_maintenance(database_url, agent_ids, provider, config)
    #     await asyncio.sleep(4 * 3600)  # every 4 hours


asyncio.run(main())

---

Multi-Agent Setup

Handle multiple agents with isolated memory spaces:

import asyncio
from arandu import MemoryClient
from arandu.providers.openai import OpenAIProvider


async def main():
    provider = OpenAIProvider(api_key="sk-...")
    memory = MemoryClient(
        database_url="postgresql+psycopg://memory:memory@localhost/memory",
        llm=provider,
        embeddings=provider,
    )
    await memory.initialize()

    try:
        # Each agent has completely isolated memory
        await memory.write(
            agent_id="alice",
            message="I work at Google as a PM. I live in Mountain View.",
            speaker_name="Alice",
        )
        await memory.write(
            agent_id="bob",
            message="I'm a freelance designer based in Berlin.",
            speaker_name="Bob",
        )

        # Agent alice's context only shows alice's facts
        alice_ctx = await memory.retrieve(agent_id="alice", query="where do they work?")
        print("Alice:", alice_ctx.context)

        # Agent bob's context only shows bob's facts
        bob_ctx = await memory.retrieve(agent_id="bob", query="where do they work?")
        print("Bob:", bob_ctx.context)
    finally:
        await memory.close()


asyncio.run(main())

---

Multi-Speaker (Therapy Session)

Two speakers writing to the same agent with the same session - pronouns like "Eu" resolve to the correct speaker each time.

import asyncio
from arandu import MemoryClient
from arandu.providers.openai import OpenAIProvider


async def main():
    provider = OpenAIProvider(api_key="sk-...")
    memory = MemoryClient(
        database_url="postgresql+psycopg://memory:memory@localhost/memory",
        llm=provider,
        embeddings=provider,
    )
    await memory.initialize()

    try:
        # Two speakers, same agent, same session
        await memory.write(
            agent_id="therapy_bot",
            message="Eu me sinto ignorada pelo Carlos. Ele nunca me ouve.",
            speaker_name="Ana Silva",
            session_id="sessao_001",
        )
        await memory.write(
            agent_id="therapy_bot",
            message="Eu trabalho 12 horas por dia pra sustentar a família.",
            speaker_name="Carlos Silva",
            session_id="sessao_001",
        )

        # Retrieve — facts attributed to correct speakers
        result = await memory.retrieve(agent_id="therapy_bot", query="Como a Ana se sente?")
        # Returns: "Ana Silva feels ignored by Carlos" (not Carlos's facts)
        for fact in result.facts:
            print(f"  [{fact.score:.2f}] {fact.entity_name}: {fact.fact_text}")
    finally:
        await memory.close()


asyncio.run(main())

---

Multi-Session (Work + Personal)

Same speaker, same agent, different session_ids. Retrieve without session_id searches all sessions.

import asyncio
from arandu import MemoryClient
from arandu.providers.openai import OpenAIProvider


async def main():
    provider = OpenAIProvider(api_key="sk-...")
    memory = MemoryClient(
        database_url="postgresql+psycopg://memory:memory@localhost/memory",
        llm=provider,
        embeddings=provider,
    )
    await memory.initialize()

    try:
        # Work context
        await memory.write(
            agent_id="assistant",
            message="Preciso entregar o relatório até sexta.",
            speaker_name="Marcos",
            session_id="work",
        )

        # Personal context
        await memory.write(
            agent_id="assistant",
            message="Minha mãe tá doente, vou visitar ela no fim de semana.",
            speaker_name="Marcos",
            session_id="personal",
        )

        # Retrieve searches ALL sessions by default
        result = await memory.retrieve(agent_id="assistant", query="O que o Marcos precisa fazer?")
        # Returns facts from BOTH work and personal sessions
        for fact in result.facts:
            print(f"  [{fact.score:.2f}] {fact.entity_name}: {fact.fact_text}")
    finally:
        await memory.close()


asyncio.run(main())

---

Customer Service Bot (Multiple Customers)

Same agent serving different customers - each customer's facts are isolated by speaker_name.

import asyncio
from arandu import MemoryClient
from arandu.providers.openai import OpenAIProvider


async def main():
    provider = OpenAIProvider(api_key="sk-...")
    memory = MemoryClient(
        database_url="postgresql+psycopg://memory:memory@localhost/memory",
        llm=provider,
        embeddings=provider,
    )
    await memory.initialize()

    try:
        # Customer 1
        await memory.write(
            agent_id="support_bot",
            message="Meu pedido #1234 não chegou.",
            speaker_name="Maria Oliveira",
        )

        # Customer 2
        await memory.write(
            agent_id="support_bot",
            message="Quero trocar o produto que comprei ontem.",
            speaker_name="João Santos",
        )

        # Retrieve for specific customer — agent remembers each one separately
        result = await memory.retrieve(agent_id="support_bot", query="Qual o problema da Maria?")
        for fact in result.facts:
            print(f"  [{fact.score:.2f}] {fact.entity_name}: {fact.fact_text}")
    finally:
        await memory.close()


asyncio.run(main())