episodial 0.1.1

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",
    chat_id="conv_1",        # optional — scope to a conversation
    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...",
    chat_id="conv_1",        # optional — scope to a conversation
    salience=0.7,
)

Retrieve

result = ep.retrieve(
    user_id="user_123",
    project_id="my-app",
    chat_id="conv_1",        # optional — filter by conversation
    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", chat_id="conv_1")

Conversation Scoping

Use chat_id to isolate memories per conversation. Omit it (or pass None) for cross-conversation global memories.

# Retrieve memories from this chat + global memories (default)
result = ep.retrieve(
    user_id="user_123",
    project_id="my-app",
    chat_id="conv_1",
    include_global=True,
    task_goal="deployment help",
)

# Only memories from this specific chat
result = ep.retrieve(
    user_id="user_123",
    project_id="my-app",
    chat_id="conv_1",
    chat_only=True,
    task_goal="deployment help",
)

# All memories for the user (no chat filter)
result = ep.retrieve(
    user_id="user_123",
    project_id="my-app",
    task_goal="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