Primeiros Passos¶

Este guia te leva do zero ao funcionamento com arandu: instalando dependências, configurando PostgreSQL com pgvector, escrevendo seus primeiros fatos e recuperando-os.

Pré-requisitos¶

Python 3.11+
PostgreSQL 15+ com a extensão pgvector instalada
Uma chave de API da OpenAI (ou qualquer LLM/embedding provider - veja Custom Providers)

Passo 1: Instalação¶

pip install arandu[openai]

Isso instala o SDK core mais o provider OpenAI incluído. Se você usa um LLM provider diferente, instale apenas o core:

pip install arandu

Passo 2: Configurar PostgreSQL + pgvector¶

O arandu armazena fatos, entidades e embeddings no PostgreSQL usando a extensão pgvector para busca por similaridade vetorial.

Opção A: Docker (recomendado para desenvolvimento)¶

docker run -d \
  --name memory-db \
  -e POSTGRES_USER=memory \
  -e POSTGRES_PASSWORD=memory \
  -e POSTGRES_DB=memory \
  -p 5432:5432 \
  pgvector/pgvector:pg16

A imagem pgvector/pgvector já vem com a extensão pré-instalada. Sua connection string será: postgresql+psycopg://memory:memory@localhost:5432/memory

psycopg vs psycopg2

O Arandu usa psycopg (driver async), não psycopg2 (sync). Sua connection string deve começar com postgresql+psycopg://, não postgresql+psycopg2://. Muitos tutoriais de Django/Flask usam psycopg2 - certifique-se de usar o correto.

Opção B: PostgreSQL existente¶

Se você já tem PostgreSQL rodando, habilite a extensão pgvector:

CREATE EXTENSION IF NOT EXISTS vector;

Instalação do pgvector

Se você não tem o pgvector instalado no seu servidor, siga o guia de instalação do pgvector.

Entendendo Agent, Session e Speaker¶

Todo write() usa três identificadores. Aqui vai o que cada um faz:

agent_id identifica de quem é a memória. Pense como um cérebro. Um agente = um espaço de memória. Todos os fatos, entidades e relacionamentos vivem dentro da memória daquele agente. Se você tem dois chatbots, cada um tem seu próprio agent_id e eles não compartilham memórias.

speaker_name identifica quem está falando. Quando alguém diz "Eu moro em São Paulo", o SDK precisa saber quem é "Eu". Se o speaker é Rafael, "Eu" vira "Rafael mora em São Paulo". Sem speaker_name, o SDK não sabe a quem atribuir o "Eu" e vai levantar um ValueError.

session_id marca o contexto da conversa. É opcional (default "default"). Use quando quiser rastrear de qual conversa uma mensagem veio. Por exemplo: um ticket de suporte, uma sessão de terapia, ou uma reunião.

A memória NÃO é separada por sessão

Mudar o session_id não cria uma memória separada. Todos os fatos vão pra mesma memória do agente, independente da sessão. Quando você chama retrieve(), ele busca tudo que o agente sabe, de todas as sessões. O session_id é metadata no evento, não uma chave de partição.

await memory.write(
    agent_id="meu-assistente",      # de quem é a memória
    message="Eu moro em São Paulo",
    speaker_name="Rafael",          # quem está falando
    session_id="ticket-suporte-42", # tag de contexto (opcional)
)

Pra um chatbot simples com um usuário, você só precisa de agent_id e speaker_name. Adicione session_id quando quiser rastrear de onde uma conversa veio.

Por que três campos separados?

Um sistema de memória precisa responder três perguntas: de quem é o cérebro que armazena? (agent), quem falou? (speaker), e em que contexto? (session). Misturar tudo num único identificador quebra quando duas pessoas falam com o mesmo agente, ou a mesma pessoa tem várias conversas. Separar mantém o modelo limpo e flexível.

Passo 3: Inicializar o Client¶

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

async def main():
    # Criar o provider de LLM + embedding
    provider = OpenAIProvider(api_key="sk-...")

    # Criar o memory client
    memory = MemoryClient(
        database_url="postgresql+psycopg://memory:memory@localhost:5432/memory",
        llm=provider,
        embeddings=provider,
    )

    # Criar tabelas (seguro chamar múltiplas vezes)
    await memory.initialize()

    print("Memory initialized!")
    await memory.close()

asyncio.run(main())

Usando Anthropic (Claude) em vez de OpenAI

from arandu.providers.anthropic import AnthropicProvider
from arandu.providers.openai import OpenAIProvider

llm = AnthropicProvider(api_key="sk-ant-...")  # Claude para LLM
embeddings = OpenAIProvider(api_key="sk-...")   # OpenAI apenas para embeddings

memory = MemoryClient(
    database_url="postgresql+psycopg://...",
    llm=llm,
    embeddings=embeddings,  # Anthropic não oferece embeddings
)

Instale com: pip install arandu[anthropic]

Usando DeepSeek, Groq ou modelos locais

Qualquer provider compatível com OpenAI funciona com OpenAIProvider. Basta definir base_url:

llm = OpenAIProvider(api_key="sk-...", model="deepseek-chat", base_url="https://api.deepseek.com/v1")

Veja o Cookbook para mais exemplos.

O initialize() cria todas as tabelas e índices necessários (incluindo índices HNSW do pgvector). É idempotente - seguro para chamar a cada startup.

Sobre o agent_id

O agent_id é sua chave de particionamento. Cada agent_id tem seu próprio espaço de memória isolado - fatos escritos para um agente nunca são retornados para outro. Pense nele como um cérebro: um agente, uma memória. Use qualquer string (ID do banco, UUID, slug). O mesmo agent_id deve ser usado em write() e retrieve() para o mesmo agente.

Sobre o session_id

O session_id identifica o contexto da conversa (default: "default"). Pense nele como uma thread de conversa no WhatsApp - mesmo agente, conversas diferentes. Quando não fornecido, todas as escritas vão para a sessão "default".

Sobre o speaker_name

O speaker_name identifica quem está falando a mensagem. É um parâmetro obrigatório de write(). Pronomes como "eu", "I", "me", "myself" são automaticamente resolvidos para a entidade do speaker (person:{speaker_slug}). Por exemplo, se speaker_name="Rafael" e a mensagem diz "I live in São Paulo", o sistema entende que Rafael mora em São Paulo. O agent_id é o agente/cérebro que armazena a memória; o speaker_name é quem está falando.

Passo 4: Escrever Seus Primeiros Fatos¶

O método write() recebe uma mensagem em linguagem natural e automaticamente:

Extrai entidades, fatos e relacionamentos usando um LLM
Resolve entidades para registros canônicos (deduplicação)
Reconcilia novos fatos contra o conhecimento existente
Faz upsert dos resultados no banco de dados

async def write_example(memory: MemoryClient):
    # Primeira mensagem
    result = await memory.write(
        agent_id="user_123",
        message="My name is Rafael and I live in São Paulo. I work at Acme Corp as a backend engineer.",
        speaker_name="Rafael",
    )
    print(f"Facts added: {len(result.facts_added)}")
    for fact in result.facts_added:
        print(f"  [{fact.entity_name}] {fact.fact_text} (confidence: {fact.confidence})")
    # Output:
    #   [Rafael] Lives in São Paulo (confidence: 0.95)
    #   [Rafael] Works at Acme Corp as a backend engineer (confidence: 0.95)
    #   [Acme Corp] Rafael works at Acme Corp (confidence: 0.95)
    print(f"Entities resolved: {len(result.entities_resolved)}")
    print(f"Duration: {result.duration_ms:.0f}ms")

    # Segunda mensagem — o sistema reconhece "Rafael" e atualiza o conhecimento
    result = await memory.write(
        agent_id="user_123",
        message="I just moved to Rio de Janeiro. Still working at Acme though.",
        speaker_name="Rafael",
        session_id="onboarding",  # opcional — default é "default"
    )
    print(f"Facts added: {len(result.facts_added)}")
    print(f"Facts updated: {len(result.facts_updated)}")  # "lives in São Paulo" → "lives in Rio"

Entendendo o WriteResult¶

O objeto WriteResult te diz exatamente o que aconteceu:

Campo	Tipo	Descrição
`event_id`	`str`	ID único deste evento de escrita
`facts_added`	`list`	Novos fatos criados (decisões ADD)
`facts_updated`	`list`	Fatos existentes substituídos (decisões UPDATE)
`facts_unchanged`	`list`	Fatos confirmados mas não alterados (decisões NOOP)
`facts_deleted`	`list`	Fatos retratados (decisões DELETE)
`entities_resolved`	`list`	Entidades identificadas e resolvidas
`duration_ms`	`float`	Duração total do pipeline
`success`	`bool`	Se o pipeline completou sem erros
`error`	`str \\| None`	Mensagem de erro se o pipeline falhou internamente

Passo 5: Recuperar Contexto¶

O método retrieve() encontra fatos relevantes para uma query usando múltiplos sinais:

async def retrieve_example(memory: MemoryClient):
    result = await memory.retrieve(
        agent_id="user_123",
        query="where does Rafael live and what does he do?",
    )

    # Opção 1: String pré-formatada — cole direto no prompt do LLM
    print(result.context)

    # Opção 2: Fatos individuais — para acesso programático
    for fact in result.facts:
        print(f"  [{fact.score:.2f}] {fact.entity_name}: {fact.fact_text}")

    # Retrieve dentro de uma sessão específica (opcional — omita para buscar em todas)
    session_result = await memory.retrieve(
        agent_id="user_123",
        query="onde o Rafael mora?",
        session_id="onboarding",  # opcional — default busca em todas as sessões
    )

    # Com ajustes de config (ex: desabilitar reranker para resultados mais rápidos)
    fast_result = await memory.retrieve(
        agent_id="user_123",
        query="onde o Rafael mora?",
        config_overrides={"enable_reranker": False, "topk_facts": 5},
    )

    print(f"Total candidates evaluated: {result.total_candidates}")
    print(f"Duration: {result.duration_ms:.0f}ms")

.context vs .facts

Use result.context quando precisa apenas de uma string para injetar no prompt do LLM - já vem formatada com labels de tier (CORE MEMORY, EXTENDED CONTEXT, etc.). Use result.facts quando precisa de acesso programático aos fatos individuais, scores e metadados.

Overrides de Config por Request¶

Você pode sobrescrever qualquer campo do MemoryConfig para uma única request, sem alterar o config padrão do client:

result = await memory.retrieve(
    agent_id="user_123",
    query="onde o Rafael mora?",
    config_overrides={
        "enable_reranker": False,
        "topk_facts": 5,
        "spreading_activation_hops": 0,
    },
)

# config_effective mostra o config efetivo usado nesta request
print(result.config_effective)

Apenas as chaves fornecidas são sobrescritas; todos os outros campos herdam do MemoryConfig do client.

Entendendo o RetrieveResult¶

Campo	Tipo	Descrição
`facts`	`list[ScoredFact]`	Fatos ranqueados com scores
`context`	`str`	String de contexto pré-formatada para prompts LLM
`total_candidates`	`int`	Total de fatos avaliados antes do ranking
`duration_ms`	`float`	Duração total do pipeline
`config_effective`	`dict`	Valores de config efetivos usados nesta request

Cada ScoredFact contém:

Campo	Tipo	Descrição
`fact_id`	`str`	Identificador único do fato
`entity_name`	`str`	Nome legível da entidade
`attribute_key`	`str`	Categoria/atributo do fato
`fact_text`	`str`	O conteúdo do fato
`score`	`float`	Score combinado de relevância (0-1)
`scores`	`dict`	Detalhamento por sinal (semantic, recency, etc.)
`speaker`	`str \\| None`	Quem falou a mensagem de onde este fato foi extraído

Passo 6: Gerenciando Fatos¶

Além de write() e retrieve(), o SDK oferece operações CRUD para gerenciar fatos individualmente: buscar por ID, listar todos e deletar.

Buscar um fato específico¶

detail = await memory.get(agent_id="user_123", fact_id="algum-uuid-aqui")
if detail:
    print(f"[{detail.entity_name}] {detail.fact_text}")
    print(f"  confidence: {detail.confidence}, importance: {detail.importance}")
    print(f"  criado em: {detail.created_at}")
else:
    print("Fato não encontrado")

get() retorna um FactDetail ou None. Busca qualquer fato pelo ID — incluindo fatos que foram soft-deleted pelo pipeline de reconciliação (ou seja, substituídos por uma versão mais nova). Use para lookups diretos quando você tem o ID.

Listar todos os fatos¶

# Primeira página (mais recentes primeiro)
facts = await memory.get_all(agent_id="user_123", limit=50, offset=0)
for fact in facts:
    print(f"[{fact.fact_id}] {fact.entity_name}: {fact.fact_text}")

# Próxima página
page2 = await memory.get_all(agent_id="user_123", limit=50, offset=50)

get_all() retorna apenas fatos ativos (valid_to IS NULL), ordenados por created_at decrescente. Use limit e offset para paginação.

Filtrando por entidade¶

# Só fatos sobre a Ana
ana_facts = await memory.get_all(agent_id="user_123", entity_keys=["person:ana"])

# Fatos sobre Pedro OU Ana
facts = await memory.get_all(agent_id="user_123", entity_keys=["person:pedro", "person:ana"])

O filtro entity_keys também funciona com retrieve():

# Busca semântica restrita a fatos sobre a Ana
result = await memory.retrieve(
    agent_id="user_123",
    query="o que ela faz de trabalho?",
    entity_keys=["person:ana"],
)

Quando entity_keys é fornecido, apenas fatos linkados a pelo menos uma das entidades especificadas são retornados (lógica OR). Sem entity_keys, todos os fatos são buscados como antes.

Aliases são resolvidos automaticamente

entity_keys aceita tanto chaves canônicas (person:pedro_menezes) quanto aliases (person:pedro ou só pedro). O SDK resolve aliases via memory_entity_aliases antes de filtrar, então você não precisa saber a forma canônica. Qualquer chave que não resolva aparece em result.warnings — retrieve() nunca retorna "zero silencioso" por causa de uma chave errada.

result = await memory.retrieve(
    agent_id="user_123",
    query="o que ela faz?",
    entity_keys=["person:pedro", "person:unknown"],
)
# result.facts → filtrado pela canônica do Pedro (alias resolvido)
# result.warnings → ["entity_key 'person:unknown' not found (not canonical, no matching alias)"]

Deletar um fato¶

deleted = await memory.delete(agent_id="user_123", fact_id="algum-uuid-aqui")
print(f"Deletado: {deleted}")  # True se encontrado e removido, False caso contrário

delete() faz um hard delete — a row é fisicamente removida do banco de dados. Entity links associados são removidos automaticamente via cascade. Esta é a ação explícita do usuário ("quero que isso suma"); o soft-delete do pipeline via valid_to é um mecanismo separado para reconciliação.

Deletar todos os fatos¶

count = await memory.delete_all(agent_id="user_123")
print(f"{count} fatos deletados")

delete_all() remove todos os fatos do agente. Use com cuidado — é irreversível. Destinado a cenários de reset/debug.

Listar entidades¶

entities = await memory.entities(agent_id="user_123", limit=50)
for entity in entities:
    print(f"[{entity.entity_type}] {entity.display_name} ({entity.fact_count} fatos)")
    if entity.summary_text:
        print(f"  Resumo: {entity.summary_text}")

entities() retorna entidades ativas ordenadas por last_seen_at decrescente. Use pra ver o que o agente conhece — pessoas, lugares, organizações, etc.

Entendendo o FactDetail¶

Campo	Tipo	Descrição
`fact_id`	`str`	Identificador único do fato
`entity_name`	`str`	Nome legível da entidade
`entity_key`	`str`	Chave canônica da entidade
`entity_type`	`str`	Tipo da entidade (ex: "person", "organization")
`attribute_key`	`str \\| None`	Categoria/atributo do fato
`fact_text`	`str`	O conteúdo do fato
`category`	`str \\| None`	Categoria do fato
`confidence`	`float`	Score de confiança (0-1)
`importance`	`float`	Score de importância (0-1)
`valid_from`	`datetime \\| None`	Quando o fato se tornou válido
`created_at`	`datetime \\| None`	Quando o fato foi criado
`source_context`	`str \\| None`	Trecho do contexto original
`speaker`	`str \\| None`	Quem falou a mensagem de onde este fato foi extraído

Entendendo o EntityDetail¶

Campo	Tipo	Descrição
`entity_id`	`str`	Identificador único da entidade
`canonical_key`	`str`	Chave canônica da entidade (ex: "person::rafael")
`display_name`	`str`	Nome legível da entidade
`entity_type`	`str`	Tipo da entidade (ex: "person", "organization")
`summary_text`	`str \\| None`	Resumo auto-gerado da entidade
`fact_count`	`int`	Número de fatos ligados a esta entidade
`importance_score`	`float \\| None`	Score de importância computado
`first_seen_at`	`datetime \\| None`	Quando a entidade foi mencionada pela primeira vez
`last_seen_at`	`datetime \\| None`	Quando a entidade foi mencionada pela última vez
`profile_text`	`str \\| None`	Perfil consolidado da entidade

Passo 7: Configurar (Opcional)¶

Todos os aspectos do pipeline são configuráveis via MemoryConfig:

from arandu import MemoryConfig
from arandu.providers.openai import OpenAIProvider

# Provider único para todas as operações LLM (extração, reranker, etc.)
llm = OpenAIProvider(api_key="sk-...", model="gpt-4o")

config = MemoryConfig(
    # Timeout curto para chat em tempo real
    extraction_timeout_sec=15.0,

    # Ajustar retrieval
    topk_facts=30,
    min_similarity=0.25,
    enable_reranker=True,

    # Pesos customizados (default: semantic=0.70, recency=0.20, importance=0.10)
    score_weights={
        "semantic": 0.60,
        "recency": 0.25,
        "importance": 0.15,
    },

    # Definir timezone para cálculos de recência
    timezone="America/Sao_Paulo",
)

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

Todos os parâmetros têm defaults sensatos - você só precisa sobrescrever o que importa para o seu caso de uso.

Passo 8: Debug com Verbose Mode¶

Passe verbose=True para write() ou retrieve() para obter um trace detalhado de cada step do pipeline:

result = await memory.write(agent_id="user_123", message="...", speaker_name="Rafael", verbose=True)

# Acessar o trace do pipeline
if result.pipeline:
    for step in result.pipeline.steps:
        print(f"  {step.name}: {step.duration_ms:.1f}ms")
        print(f"    data: {step.data}")

O trace inclui steps como extraction, entity_resolution, reconciliation e upsert, cada um com timing e dados intermediários. Se o pipeline falhar internamente, um step error é adicionado com os detalhes da exceção - útil para diagnosticar falhas silenciosas.

Você pode serializar o trace completo com result.pipeline.to_dict().

Passo 9: Cleanup¶

Sempre feche o client ao terminar para liberar conexões do banco:

await memory.close()

Ou use como um padrão de contexto async:

memory = MemoryClient(...)
await memory.initialize()
try:
    # ... usar memory
finally:
    await memory.close()

Exemplo Completo¶

Aqui está um exemplo completo funcional juntando tudo:

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:
        # Escrever alguns fatos
        await memory.write(
            agent_id="user_123",
            message="I'm a software engineer living in Berlin. I love cycling and craft coffee.",
            speaker_name="Rafael",
        )
        await memory.write(
            agent_id="user_123",
            message="My girlfriend Ana is a designer. We adopted a cat named Pixel last month.",
            speaker_name="Rafael",
        )

        # Recuperar contexto
        result = await memory.retrieve(agent_id="user_123", query="tell me about this person")
        print(result.context)

        # Retrieval direcionado
        result = await memory.retrieve(agent_id="user_123", query="who is Ana?")
        for fact in result.facts:
            print(f"  [{fact.score:.2f}] {fact.entity_name}: {fact.fact_text}")
    finally:
        await memory.close()


asyncio.run(main())

Custom Providers¶

O arandu usa protocolos Python para injeção de dependência. Você pode trazer qualquer LLM ou embedding provider implementando duas interfaces simples:

from arandu.protocols import LLMProvider, LLMResult, EmbeddingProvider

class MyLLMProvider:
    async def complete(
        self,
        messages: list[dict],
        temperature: float = 0,
        response_format: dict | None = None,
        max_tokens: int | None = None,
    ) -> LLMResult:
        # Chame seu LLM aqui
        text = ...  # obtenha o texto de resposta do seu LLM
        return LLMResult(text=text, usage=None)

class MyEmbeddingProvider:
    async def embed(self, texts: list[str]) -> list[list[float]]:
        # Retorne embeddings para um batch
        ...

    async def embed_one(self, text: str) -> list[float] | None:
        # Retorne embedding para um único texto
        ...

Sem herança necessária - apenas implemente os métodos com as assinaturas corretas.

Próximos Passos¶

Write Pipeline - Entenda como fatos são extraídos, entidades resolvidas e conhecimento reconciliado
Read Pipeline - Aprenda como o retrieval multi-signal encontra os fatos mais relevantes
Tipos de Dados & Schema - Referência do schema do banco (tabelas, colunas, tipos) para queries SQL diretas
Background Jobs - Configure clustering, consolidation e importance scoring
Filosofia de Design - Explore a arquitetura inspirada em neurociência