supermemory/packages/cartesia-sdk-python
Dhravya 03773c4f2e
feat(python-sdks): SDK-level cross-source memory deduplication (#1532)
## Stack Context

Part 2 of a 3-PR stack moving memory deduplication into the SDKs. See `sdk-dedup/tools-ts` (parent) for the full context and the TypeScript implementation this mirrors.

## What?

Port the normalized, priority-ordered (`static > dynamic > search`) profile deduplication into the Python SDKs.

- Each request injects one **owned memory block that replaces** the prior block rather than accumulating.
- Dedup is **request-local** (no shared state), so it stays correct under concurrency.

Covers OpenAI, Agent Framework (middleware + context provider), Cartesia, and Pipecat.

## Why?

Keeps the Python SDKs at behavioral parity with the TypeScript SDK so all integrations deduplicate memory the same way.

## Testing

- OpenAI: 31 passed, 11 skipped (live)
- Agent Framework: 59 passed
- Cartesia: 8 passed
- Pipecat: 8 passed

🤖 Generated with [Claude Code](https://claude.com/claude-code)

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Changes memory formatting and system-prompt injection across multiple SDK integrations; incorrect dedup or replacement could alter LLM context, but there is no auth or data-store risk.
>
> **Overview**
> Ports **normalized cross-source memory deduplication** and **replace-not-append injection** into the Python OpenAI, Agent Framework, Cartesia, and Pipecat packages so they match the TypeScript SDK behavior.
>
> **Deduplication** uses request-local keys: strip optional `[YYYY-MM-DD]` prefixes, normalize whitespace, and compare with `casefold`, with priority **static → dynamic → search**. In **`query` mode**, profile static/dynamic are excluded from dedup input so facts that only appear in search (or overlap profile) are not dropped before formatting.
>
> **Injection** no longer appends memory text every turn. OpenAI and Agent Framework middleware **strip prior owned `<supermemory context="user-memories" readonly>` blocks** and **replace** them once per request while keeping the caller’s system instructions; extra system messages lose stale blocks only. New helpers (`strip`/`replace`/`wrap`) live in each package’s utils.
>
> Tests cover normalized fact variants, query-mode search retention, and stale block replacement.
>
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit 42f308b224. Bugbot is set up for automated code reviews on this repo. Configure [here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->
2026-09-01 06:10:36 +00:00
..
src/supermemory_cartesia feat(python-sdks): SDK-level cross-source memory deduplication (#1532) 2026-09-01 06:10:36 +00:00
tests feat(python-sdks): SDK-level cross-source memory deduplication (#1532) 2026-09-01 06:10:36 +00:00
pyproject.toml feat(python-sdks): SDK-level cross-source memory deduplication (#1532) 2026-09-01 06:10:36 +00:00
README.md fix(python-sdks): v4 API migration for integration packages (#1434) 2026-09-01 06:10:36 +00:00
requirements.txt fix(python-sdks): v4 API migration for integration packages (#1434) 2026-09-01 06:10:36 +00:00

Supermemory Cartesia SDK

Memory-enhanced voice agents with Supermemory and Cartesia Line.

Installation

pip install supermemory-cartesia

Quick Start

import os
from line.llm_agent import LlmAgent, LlmConfig
from line.voice_agent_app import VoiceAgentApp
from supermemory_cartesia import SupermemoryCartesiaAgent

async def get_agent(env, call_request):
    # Extract container_tag from call metadata (typically user ID)
    container_tag = call_request.metadata.get("user_id", "default-user")

    # Create base LLM agent
    base_agent = LlmAgent(
        model="gemini/gemini-2.5-flash-preview-09-2025",
        api_key=os.getenv("GEMINI_API_KEY"),
        config=LlmConfig(
            system_prompt="You are a helpful voice assistant with memory.",
            introduction="Hello! Great to talk with you again!"
        )
    )

    # Wrap with Supermemory
    memory_agent = SupermemoryCartesiaAgent(
        agent=base_agent,
        api_key=os.getenv("SUPERMEMORY_API_KEY"),
        container_tag=container_tag,
        custom_id=call_request.call_id,
    )

    return memory_agent

# Create voice agent app
app = VoiceAgentApp(get_agent=get_agent)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

Configuration

Parameters

Parameter Type Required Description
agent LlmAgent Yes The Cartesia Line agent to wrap
container_tag str Yes Primary container tag for memory scoping (e.g., user ID)
custom_id str Yes Custom ID for grouping conversation messages into a single document
add_memory Literal No Memory persistence mode: "always" (default) or "never"
container_tags List[str] No Additional container tags for organization (e.g., ["org", "prod"])
api_key str No Supermemory API key (or set SUPERMEMORY_API_KEY env var)
config MemoryConfig No Advanced configuration
base_url str No Custom API endpoint

Advanced Configuration

from supermemory_cartesia import SupermemoryCartesiaAgent

memory_agent = SupermemoryCartesiaAgent(
    agent=base_agent,
    container_tag="user-123",
    custom_id="conversation-456",
    add_memory="always",           # "always" (default) or "never"
    container_tags=["org-acme", "prod"],  # Optional: additional tags
    config=SupermemoryCartesiaAgent.MemoryConfig(
        search_limit=10,           # Max memories to retrieve
        search_threshold=0.1,      # Similarity threshold
        mode="full",               # "profile", "query", or "full"
        system_prompt="Based on previous conversations, I recall:\n\n",
    ),
)

# Read-only mode - retrieve memories but don't save new ones
read_only_agent = SupermemoryCartesiaAgent(
    agent=base_agent,
    container_tag="user-123",
    custom_id="conversation-456",
    add_memory="never",  # Only retrieve, don't save
)

Memory Modes

Mode Static Profile Dynamic Profile Search Results
"profile" Yes Yes No
"query" No No Yes
"full" Yes Yes Yes

How It Works

  1. Intercepts events - Listens for UserTurnEnded events from Cartesia Line
  2. Retrieves memories - Queries Supermemory /v4/profile API with user's message
  3. Enriches context - Passes memories as non-persistent context for the current turn
  4. Stores messages - Sends conversation to Supermemory (background, non-blocking)
  5. Passes to agent - Forwards enriched event to wrapped LlmAgent

What Gets Stored

User and assistant messages are sent to Supermemory:

{
  "content": "User: What's the weather?\nAssistant: It's sunny today!",
  "container_tags": ["user-123", "org-acme", "prod"],
  "metadata": { "platform": "cartesia" }
}

Architecture

Cartesia Line uses an event-driven architecture:

User Speaks (Audio)
    ↓
[Ink STT] → Automatic speech recognition
    ↓
UserTurnEnded Event {content: "user message", history: [...]}
    ↓
┌──────────────────────────────────────────────┐
│   SUPERMEMORY CARTESIA AGENT (Wrapper)       │
│                                              │
│  process(env, event):                        │
│    1. Intercept UserTurnEnded                │
│    2. Extract user message                   │
│    3. Query Supermemory API                  │
│    4. Add memories as per-turn context       │
│    5. Pass to wrapped LlmAgent               │
│    6. Store conversation (async background)  │
└──────────────────────────────────────────────┘
    ↓
AgentSendText Event {text: "response"}
    ↓
[Sonic TTS] → Ultra-fast speech synthesis
    ↓
Audio Output

Comparison with Pipecat SDK

Aspect Pipecat Cartesia Line
Integration Pattern Extends FrameProcessor Wrapper around LlmAgent
Event Handling process_frame() method process() method
Events LLMContextFrame, LLMMessagesFrame UserTurnEnded, CallStarted
Context Object LLMContext.get_messages() event.history
Memory Injection Modify context.add_message() Pass per-turn context

Full Example with Tools

import os
from line.llm_agent import LlmAgent, LlmConfig
from line.tools import LoopbackTool
from line.voice_agent_app import VoiceAgentApp
from supermemory_cartesia import SupermemoryCartesiaAgent

# Define custom tools
async def get_weather(location: str) -> str:
    return f"The weather in {location} is sunny, 72°F"

weather_tool = LoopbackTool(
    name="get_weather",
    description="Get current weather for a location",
    function=get_weather
)

async def get_agent(env, call_request):
    container_tag = call_request.metadata.get("user_id", "default-user")
    org_id = call_request.metadata.get("org_id")

    # Create LLM agent with tools
    base_agent = LlmAgent(
        model="gemini/gemini-2.5-flash-preview-09-2025",
        api_key=os.getenv("GEMINI_API_KEY"),
        tools=[weather_tool],
        config=LlmConfig(
            system_prompt="You are a personal assistant with memory and tools.",
            introduction="Hi! How can I help you today?"
        )
    )

    # Wrap with Supermemory
    memory_agent = SupermemoryCartesiaAgent(
        agent=base_agent,
        api_key=os.getenv("SUPERMEMORY_API_KEY"),
        container_tag=container_tag,
        custom_id=call_request.call_id,
        container_tags=[org_id] if org_id else None,
        config=SupermemoryCartesiaAgent.MemoryConfig(
            mode="full",
            search_limit=15,
            search_threshold=0.15,
        )
    )

    return memory_agent

app = VoiceAgentApp(get_agent=get_agent)

Development

# Clone repository
git clone https://github.com/supermemoryai/supermemory
cd supermemory/packages/cartesia-sdk-python

# Install in development mode
pip install -e ".[dev]"

# Run tests
pytest

# Format code
black .
isort .

License

MIT