cerebe

Python SDK for the Cerebe Cognitive Services Platform — memory, knowledge graphs, meta-learning, and agent tooling.

Installation

pip install cerebe

Quick Start

from cerebe import Cerebe

client = Cerebe(api_key="ck_live_xxx", project="proj_xxx")

# Store a memory
client.memory.add(
    "User prefers dark mode",
    "sess_123",
    type="semantic",
    importance=0.8,
)

# Search memories
results = client.memory.search("user preferences", "sess_123")
print(results.data)

Async

from cerebe import AsyncCerebe

async with AsyncCerebe(api_key="ck_live_xxx") as client:
    results = await client.memory.search("user preferences", "sess_123")

Every method available on Cerebe has an identical async counterpart on AsyncCerebe.

Configuration

Environment variable fallbacks: CEREBE_API_KEY, CEREBE_PROJECT, CEREBE_BASE_URL.

API Reference

Memory — client.memory

Memory types: episodic, semantic, procedural, sequential, execution_history, plan, tool_reliability, working, declarative

# Store with full options
client.memory.add(
    "Learned quadratic formula today",
    "sess_123",
    type="episodic",
    importance=0.9,
    entity_id="user_42",
    metadata={"subject": "math"},
)

# Search with filters
results = client.memory.search(
    "math concepts",
    "sess_123",
    types=["episodic", "semantic"],
    min_importance=0.5,
    limit=10,
)

# Harvest memories from conversation
client.memory.harvest(
    "sess_123",
    "User: I find visual explanations helpful...",
    entity_id="user_42",
)

Knowledge — client.knowledge

client.knowledge.ingest(
    "Photosynthesis converts light energy to chemical energy",
    entity_id="biology_101",
    source="textbook",
)

graph = client.knowledge.query("photosynthesis", depth=3)

Storage — client.storage

# Get presigned upload URL
result = client.storage.presigned_upload(
    file_name="essay.pdf",
    file_type="application/pdf",
    file_size=102400,
    content_hash="sha256_abc123",
    tenant_id="tenant_1",
)

# Analyze uploaded content
client.storage.analyze_content("up_123", context="student_homework")

Meta-Learning — client.meta_learning

profile = client.meta_learning.profile("user_42")

state = client.meta_learning.plre_state("user_42", session_id="sess_123")

Agents — client.agents

# Ingest an agent trace
client.agents.ingest_trace(
    "Called search tool with query 'quadratic formula'",
    "sess_123",
    metadata={"tool": "search", "latency_ms": 120},
)

# Set working memory with TTL
client.agents.set_working_memory(
    "Current task: help user with algebra homework",
    "sess_123",
    ttl_seconds=3600,
)

Sessions — client.sessions

Graph — client.graph

RAG — client.rag

Retrieval-Augmented Generation: embed documents into a tenant-scoped vector space and retrieve semantically relevant chunks.

from cerebe import Cerebe

client = Cerebe(api_key="ck_live_xxx")

# Embed a document
client.rag.embed(source="docs/auth.md", content=open("docs/auth.md").read())

# Semantic search
results = client.rag.search("how does authentication work?", k=3)
for r in results.data["results"]:
    print(r["source"], r["score"])

# Hybrid search (semantic + keyword)
results = client.rag.hybrid_search("auth middleware", semantic_weight=0.8)

# Find similar documents
results = client.rag.find_similar("JWT token validation flow")

# Collection stats and cleanup
stats = client.rag.stats()
client.rag.delete_document("docs/auth.md")

Every document and query is isolated by organization — results never leak across tenants.

Error Handling

from cerebe import Cerebe
from cerebe._errors import (
    AuthenticationError,
    NotFoundError,
    RateLimitError,
    ValidationError,
    ServerError,
)

client = Cerebe(api_key="ck_live_xxx")

try:
    client.memory.get("mem_nonexistent")
except NotFoundError:
    print("Memory not found")
except RateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")
except AuthenticationError:
    print("Invalid API key")

Retries

The SDK automatically retries on:

• 429 (Rate Limited) — respects Retry-After header
• 5xx (Server Error) — exponential backoff

Set max_retries=0 to disable.

Type Safety

The SDK is fully typed with py.typed marker (PEP 561). Works with mypy, pyright, and IDE autocomplete.

from cerebe.resources.memory import MemoryType

memory_type: MemoryType = "semantic"  # type-checked literal

Requirements

• Python >= 3.10
• httpx >= 0.25.0
• pydantic >= 2.0.0

License

MIT