episodial 0.1.0

Python SDK for the Episodial cognitive memory API

episodial

Python SDK for the Episodial cognitive memory API. Includes both synchronous and async clients.

Install

pip install episodial

Quick Start

from episodial import Episodial

ep = Episodial(
    base_url="https://api.episodial.ai",
    api_key="ep_live_...",
    project_id="my-app",
)

# Store a memory
ep.remember(
    user_id="user_123",
    user_message="How do I reset my password?",
    assistant_message="Go to Settings > Security > Reset password.",
)

# Recall relevant memories
result = ep.recall("user_123", "password help")
for mem in result.memories:
    print(mem.content, mem.salience)

Async Client

import asyncio
from episodial import AsyncEpisodial

async def main():
    async with AsyncEpisodial(
        base_url="https://api.episodial.ai",
        api_key="ep_live_...",
    ) as ep:
        result = await ep.recall("user_123", "password help")
        print(result.memories)

asyncio.run(main())

Configuration

ep = Episodial(
    base_url="http://localhost:8000",  # Memory Core URL
    api_key="ep_live_...",             # API key (optional in dev)
    organization_id="org_...",         # For billing endpoints
    project_id="default",             # Default project for all calls
    timeout=30.0,                     # Request timeout in seconds
)

Memory Operations

Ingest

ep.ingest(
    user_id="user_123",
    project_id="my-app",
    event_type="llm_interaction",
    payload={
        "user_message": "What is Kubernetes?",
        "assistant_message": "Kubernetes is a container orchestration platform...",
    },
    initial_salience=0.7,
)

Or use the convenience method:

ep.remember(
    user_id="user_123",
    user_message="What is Kubernetes?",
    assistant_message="Kubernetes is a container orchestration platform...",
    salience=0.7,
)

Retrieve

result = ep.retrieve(
    user_id="user_123",
    project_id="my-app",
    task_goal="Help user with deployment",
    token_budget=2000,
)

for memory in result.memories:
    print(memory.content, memory.memory_type, memory.salience)

Or use the shorthand:

result = ep.recall("user_123", "deployment help")

Feedback

ep.helpful("memory_id_123")    # positive feedback
ep.unhelpful("memory_id_456")  # negative feedback

# Fine-grained control
ep.feedback(
    memory_id="memory_id_123",
    feedback_type="success",
    value=0.8,
    context="User confirmed this solved their issue",
)

Forget

# Forget specific memories
ep.forget(
    memory_ids=["mem_1", "mem_2"],
    reason="User requested data deletion",
)

# Forget all memories for a user
ep.forget(
    user_id="user_123",
    project_id="my-app",
    reason="Account deletion",
)

Consolidation

ep.consolidate("user_123", "my-app")

Procedural Learning

ep.mine_procedures("user_123", "my-app")

result = ep.match_procedures(
    "user_123",
    "user wants to deploy to production",
)
for proc in result.procedures:
    print(proc.readable, proc.confidence)

Billing

balance = ep.get_balance()
print(f"${balance.balance_dollars} remaining")

pricing = ep.get_pricing()

ep.topup("growth")  # $50 + 20% bonus

Error Handling

from episodial import Episodial, EpisodialError, InsufficientBalanceError

try:
    ep.recall("user_123", "hello")
except InsufficientBalanceError:
    print("Balance too low — top up!")
except EpisodialError as e:
    print(f"API error {e.status}: {e}")

Context Manager

Both clients support context managers for automatic cleanup:

with Episodial(api_key="ep_live_...") as ep:
    ep.remember(
        user_id="u1",
        user_message="hi",
        assistant_message="hello!",
    )
# httpx client closed automatically

async with AsyncEpisodial(api_key="ep_live_...") as ep:
    await ep.remember(
        user_id="u1",
        user_message="hi",
        assistant_message="hello!",
    )

License

MIT