From 2d3c706a3e9c2126271f22746fa63c552a1c742b Mon Sep 17 00:00:00 2001 From: Dhravya <63950637+Dhravya@users.noreply.github.com> Date: Thu, 24 Sep 2026 06:43:45 +0000 Subject: [PATCH] feat(livekit): add persistent memory for LiveKit Agents --- .github/workflows/ci-python.yml | 58 +++ .../workflows/publish-livekit-sdk-python.yml | 44 ++ apps/docs/docs.json | 6 + apps/docs/images/livekit-dark.svg | 18 + apps/docs/images/livekit.svg | 18 + apps/docs/integrations/livekit.mdx | 173 ++++++++ apps/docs/style.css | 4 + packages/livekit-sdk-python/README.md | 107 +++++ packages/livekit-sdk-python/pyproject.toml | 56 +++ .../src/supermemory_livekit/__init__.py | 29 ++ .../src/supermemory_livekit/agent.py | 25 ++ .../src/supermemory_livekit/exceptions.py | 19 + .../src/supermemory_livekit/identifiers.py | 22 + .../src/supermemory_livekit/memory.py | 414 ++++++++++++++++++ .../src/supermemory_livekit/tools.py | 43 ++ .../src/supermemory_livekit/utils.py | 215 +++++++++ .../livekit-sdk-python/tests/test_memory.py | 391 +++++++++++++++++ 17 files changed, 1642 insertions(+) create mode 100644 .github/workflows/publish-livekit-sdk-python.yml create mode 100644 apps/docs/images/livekit-dark.svg create mode 100644 apps/docs/images/livekit.svg create mode 100644 apps/docs/integrations/livekit.mdx create mode 100644 packages/livekit-sdk-python/README.md create mode 100644 packages/livekit-sdk-python/pyproject.toml create mode 100644 packages/livekit-sdk-python/src/supermemory_livekit/__init__.py create mode 100644 packages/livekit-sdk-python/src/supermemory_livekit/agent.py create mode 100644 packages/livekit-sdk-python/src/supermemory_livekit/exceptions.py create mode 100644 packages/livekit-sdk-python/src/supermemory_livekit/identifiers.py create mode 100644 packages/livekit-sdk-python/src/supermemory_livekit/memory.py create mode 100644 packages/livekit-sdk-python/src/supermemory_livekit/tools.py create mode 100644 packages/livekit-sdk-python/src/supermemory_livekit/utils.py create mode 100644 packages/livekit-sdk-python/tests/test_memory.py diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml index 6d730c98..4ead1e2d 100644 --- a/.github/workflows/ci-python.yml +++ b/.github/workflows/ci-python.yml @@ -7,6 +7,7 @@ on: - "packages/cartesia-sdk-python/**" - "packages/openai-sdk-python/**" - "packages/pipecat-sdk-python/**" + - "packages/livekit-sdk-python/**" - ".github/workflows/ci-python.yml" permissions: @@ -255,3 +256,60 @@ jobs: assert 'site-packages' in Path(supermemory_pipecat.__file__).parts; result = pytest.main(['-W', 'ignore::pytest.PytestAssertRewriteWarning', 'tests']); raise SystemExit(result)" + + livekit-sdk-python: + name: livekit-sdk-python (${{ matrix.dependency-lane }}, Python ${{ matrix.python-version }}) + runs-on: ubuntu-latest + timeout-minutes: 15 + strategy: + fail-fast: false + matrix: + include: + - python-version: "3.10" + dependency-lane: minimum-dependencies + supermemory-version: "3.50.0" + livekit-agents-version: "1.2.0" + - python-version: "3.12" + dependency-lane: current-dependencies + supermemory-version: "3.59.0" + livekit-agents-version: "1.8.3" + defaults: + run: + working-directory: packages/livekit-sdk-python + steps: + - name: Checkout code + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - name: Setup Python + uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: ${{ matrix.python-version }} + cache: pip + cache-dependency-path: packages/livekit-sdk-python/pyproject.toml + + - name: Install build and test tools + run: python -m pip install build pytest + + - name: Build wheel + run: python -m build --wheel --outdir "$RUNNER_TEMP/wheels" + + - name: Install wheel and tested runtime dependencies + run: >- + python -m pip install "$RUNNER_TEMP"/wheels/*.whl + "supermemory==${{ matrix.supermemory-version }}" + "livekit-agents==${{ matrix.livekit-agents-version }}" + + - name: Check dependency compatibility + run: python -m pip check + + - name: Verify real LiveKit integration and run tests + run: >- + python -c "from importlib.metadata import version; + from pathlib import Path; import livekit.agents, pytest, supermemory_livekit; + assert version('supermemory') == '${{ matrix.supermemory-version }}'; + assert version('livekit-agents') == '${{ matrix.livekit-agents-version }}'; + assert 'site-packages' in Path(supermemory_livekit.__file__).parts; + result = pytest.main(['-W', 'ignore::pytest.PytestAssertRewriteWarning', 'tests']); + raise SystemExit(result)" diff --git a/.github/workflows/publish-livekit-sdk-python.yml b/.github/workflows/publish-livekit-sdk-python.yml new file mode 100644 index 00000000..a1b4f27e --- /dev/null +++ b/.github/workflows/publish-livekit-sdk-python.yml @@ -0,0 +1,44 @@ +name: Publish LiveKit SDK Python + +on: + push: + branches: + - main + paths: + - "packages/livekit-sdk-python/pyproject.toml" + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: false + +jobs: + publish: + runs-on: ubuntu-latest + timeout-minutes: 15 + permissions: + contents: read + id-token: write + defaults: + run: + working-directory: ./packages/livekit-sdk-python + steps: + - name: Checkout + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - name: Setup Python + uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.12" + + - name: Install build dependencies + run: python -m pip install hatchling build + + - name: Build package + run: python -m build + + - name: Publish to PyPI + uses: pypa/gh-action-pypi-publish@dc37677b2e1c63e2034f94d8a5b11f265b73ba33 # v1.14.2 + with: + packages-dir: packages/livekit-sdk-python/dist/ diff --git a/apps/docs/docs.json b/apps/docs/docs.json index 0d41141c..43ddf02e 100644 --- a/apps/docs/docs.json +++ b/apps/docs/docs.json @@ -248,6 +248,7 @@ "integrations/claude-memory", "integrations/pipecat", "integrations/cartesia", + "integrations/livekit", "integrations/n8n", "integrations/viasocket", "integrations/zapier", @@ -515,6 +516,11 @@ "permanent": true, "source": "/voice-realtime/pipecat" }, + { + "destination": "/integrations/livekit", + "permanent": true, + "source": "/voice-realtime/livekit" + }, { "destination": "/integrations/n8n", "permanent": true, diff --git a/apps/docs/images/livekit-dark.svg b/apps/docs/images/livekit-dark.svg new file mode 100644 index 00000000..00738bff --- /dev/null +++ b/apps/docs/images/livekit-dark.svg @@ -0,0 +1,18 @@ + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/apps/docs/images/livekit.svg b/apps/docs/images/livekit.svg new file mode 100644 index 00000000..cf1d004f --- /dev/null +++ b/apps/docs/images/livekit.svg @@ -0,0 +1,18 @@ + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/apps/docs/integrations/livekit.mdx b/apps/docs/integrations/livekit.mdx new file mode 100644 index 00000000..7b397aa3 --- /dev/null +++ b/apps/docs/integrations/livekit.mdx @@ -0,0 +1,173 @@ +--- +title: "LiveKit" +sidebarTitle: "LiveKit (Voice)" +description: "Add persistent memory to LiveKit Agents voice sessions" +icon: "/images/livekit.svg" +--- + +Supermemory integrates with [LiveKit Agents](https://docs.livekit.io/agents/), so a voice agent can remember a caller across sessions and use that context on the next turn. Recall runs before the model replies. Completed turns are stored automatically. The model can also search, remember, and forget explicitly. + +A Supermemory timeout or outage is skipped. It does not end the call. + +## Installation + +```bash +pip install supermemory-livekit +``` + +```bash +export SUPERMEMORY_API_KEY=your_supermemory_api_key +``` + +Create a key at [console.supermemory.ai](https://console.supermemory.ai). For a self-hosted API, pass `base_url` to `SupermemoryLiveKit`. + +## Scope + +Memory is isolated by container tag. Use a stable caller id, the same one you use in the rest of your app, so a LiveKit call and a chat session share one profile. + +Container tags may only contain letters, numbers, `_`, `-`, and `:`, and must be 100 characters or fewer. Two ways to set it: + +- Pass `container_tag` yourself, usually from [job metadata](https://docs.livekit.io/agents/build/external-data/). +- Let the plugin read the participant attribute `supermemory_container_tag`, or fall back to the participant identity. + +Identities that are not valid container tags are sanitized to a stable tag. Set the attribute when the identity is an email, phone number, or SIP address and you already have memories under a different id. + +One call is stored as a single document. Pass the LiveKit room name as `session_id`. The document id is `lk-`, so a later update with the same id appends to that call instead of creating another. + +## Quick start + +Create one `SupermemoryLiveKit` inside the job. A worker handles many calls, and each call needs its own scope. + +```python +import json +import os + +from livekit import agents +from livekit.agents import AgentServer, AgentSession, ChatContext, JobContext +from supermemory_livekit import SupermemoryAgent, SupermemoryLiveKit + +server = AgentServer() + + +@server.rtc_session(agent_name="memory-agent") +async def entrypoint(ctx: JobContext): + memory = SupermemoryLiveKit(api_key=os.environ["SUPERMEMORY_API_KEY"]) + metadata = json.loads(ctx.job.metadata or "{}") + container_tag = metadata.get("container_tag") + if container_tag: + memory.bind(container_tag=container_tag, session_id=ctx.room.name) + + chat_ctx = ChatContext() + if container_tag: + await memory.preload(chat_ctx) + + await ctx.connect() + if not container_tag: + participant = await ctx.wait_for_participant() + memory.bind(participant=participant, session_id=ctx.room.name) + + session = AgentSession( + stt="deepgram/nova-3:en", + llm="openai/gpt-4.1-mini", + tts="cartesia/sonic-3", + ) + memory.attach(session) + await session.start( + room=ctx.room, + agent=SupermemoryAgent( + memory, + chat_ctx=chat_ctx, + instructions=( + "You are a helpful voice assistant. You remember this caller across calls. " + "Use that naturally, and do not mention the memory system." + ), + ), + ) + + +if __name__ == "__main__": + agents.cli.run_app(server) +``` + +`preload` puts the caller's profile into the first turn, so the greeting can use it. `attach` stores user and assistant turns when the session closes, including a caller who hangs up mid-turn. `SupermemoryAgent` recalls before each reply and adds the memory tools. + +## Your own agent + +If you already subclass `Agent`, keep that class. Pass the tools in, and recall from the turn hook. + +```python +from livekit.agents import Agent +from supermemory_livekit import SupermemoryLiveKit + +memory = SupermemoryLiveKit(container_tag="user_123", session_id="room_123") + + +class Assistant(Agent): + def __init__(self) -> None: + super().__init__( + instructions="You are a helpful voice assistant.", + tools=memory.tools(), + ) + + async def on_user_turn_completed(self, turn_ctx, new_message) -> None: + await memory.on_user_turn_completed(turn_ctx, new_message) +``` + +Call `memory.attach(session)` before `session.start`. + +`on_user_turn_completed` runs for STT-LLM-TTS pipelines. A realtime model only hits that hook when turn detection runs in your agent, not inside the model. See [LiveKit's external data guide](https://docs.livekit.io/agents/build/external-data/). Turn capture still listens for `conversation_item_added`. + +## What gets recalled + +| Mode | Static profile | Dynamic profile | Search | Use when | +| --- | --- | --- | --- | --- | +| `profile` | Yes | Yes | No | Durable facts are enough | +| `query` | No | No | Yes | Only this turn's related memories | +| `full` | Yes | Yes | Yes | Default | + +```python +from supermemory_livekit import InputParams, SupermemoryLiveKit + +memory = SupermemoryLiveKit( + container_tag="user_123", + params=InputParams( + mode="full", + search_limit=10, + search_threshold=0.1, + recall_timeout=1.5, + capture="always", + ), +) +``` + +Recall waits at most `recall_timeout` seconds (default 1.5). If the profile call is slower than that, the turn proceeds without memory. Retrieved text is inserted immediately before the user message and is not stored back as something the agent said. + +| Parameter | Default | Description | +| --- | --- | --- | +| `search_limit` | `10` | Maximum search results merged into the turn | +| `search_threshold` | `0.1` | Minimum similarity, from 0 to 1 | +| `mode` | `"full"` | `profile`, `query`, or `full` | +| `recall_timeout` | `1.5` | Seconds to wait before skipping recall | +| `capture` | `"always"` | `never` disables storing the call | + +## Tools + +`SupermemoryAgent` adds these tools. They are scoped to the bound container tag, so the model cannot read or write another caller. + +| Tool | Used for | +| --- | --- | +| `search_memories` | Look up facts and past calls when the automatic recall is not enough | +| `remember` | Save one explicit fact, preference, or correction | +| `forget` | Forget one fact by id from `search_memories`, or by exact text | + +`remember` stores a standalone fact. It is not appended to the call transcript. Automatic capture is what records the conversation. + +## Self-hosting + +```python +memory = SupermemoryLiveKit( + api_key=os.environ["SUPERMEMORY_API_KEY"], + base_url="http://localhost:6767", + container_tag="user_123", +) +``` diff --git a/apps/docs/style.css b/apps/docs/style.css index 9aad7eee..9693a3af 100644 --- a/apps/docs/style.css +++ b/apps/docs/style.css @@ -9,6 +9,10 @@ content: url("/images/opencode-logo-dark.svg"); } +.dark img[src*="livekit.svg"] { + content: url("/images/livekit-dark.svg"); +} + .eve-memory-video { display: block; width: 100%; diff --git a/packages/livekit-sdk-python/README.md b/packages/livekit-sdk-python/README.md new file mode 100644 index 00000000..dfd116b8 --- /dev/null +++ b/packages/livekit-sdk-python/README.md @@ -0,0 +1,107 @@ +# Supermemory LiveKit SDK + +Persistent memory for [LiveKit Agents](https://docs.livekit.io/agents/) voice sessions, powered by [Supermemory](https://supermemory.ai). + +The plugin recalls the caller's profile and relevant memories before each reply, stores the call as one conversation, and gives the model tools to search, remember, and forget. A Supermemory outage does not end the call. + +## Installation + +```bash +pip install supermemory-livekit +``` + +```bash +export SUPERMEMORY_API_KEY=your_supermemory_api_key +``` + +Create a key at [console.supermemory.ai](https://console.supermemory.ai). + +## Quick start + +Scope memory with a stable caller id. A LiveKit participant identity works when it is already a container tag (`letters`, `numbers`, `_`, `-`, `:`). Otherwise set the participant attribute `supermemory_container_tag`. + +```python +import json +import os + +from livekit import agents +from livekit.agents import AgentServer, AgentSession, ChatContext, JobContext +from supermemory_livekit import SupermemoryAgent, SupermemoryLiveKit + +server = AgentServer() + + +@server.rtc_session(agent_name="memory-agent") +async def entrypoint(ctx: JobContext): + memory = SupermemoryLiveKit(api_key=os.getenv("SUPERMEMORY_API_KEY")) + metadata = json.loads(ctx.job.metadata or "{}") + container_tag = metadata.get("container_tag") + if container_tag: + memory.bind(container_tag=container_tag, session_id=ctx.room.name) + + chat_ctx = ChatContext() + if container_tag: + await memory.preload(chat_ctx) + + await ctx.connect() + if not container_tag: + participant = await ctx.wait_for_participant() + memory.bind(participant=participant, session_id=ctx.room.name) + + session = AgentSession( + stt="deepgram/nova-3:en", + llm="openai/gpt-4.1-mini", + tts="cartesia/sonic-3", + ) + memory.attach(session) + await session.start( + room=ctx.room, + agent=SupermemoryAgent( + memory, + chat_ctx=chat_ctx, + instructions=( + "You are a helpful voice assistant. You remember this caller across calls. " + "Use that naturally, and do not mention the memory system." + ), + ), + ) + + +if __name__ == "__main__": + agents.cli.run_app(server) +``` + +If you already have an `Agent` subclass, pass `tools=memory.tools()` and call `await memory.on_user_turn_completed(turn_ctx, new_message)` from `on_user_turn_completed`. + +`on_user_turn_completed` runs for STT-LLM-TTS pipelines. Realtime models only hit that hook when turn detection runs in the agent, not inside the model. Call capture still listens to `conversation_item_added`. + +## Configuration + +```python +from supermemory_livekit import InputParams, SupermemoryLiveKit + +memory = SupermemoryLiveKit( + container_tag="user_123", + session_id="room-123", + params=InputParams( + mode="full", # "profile" | "query" | "full" + search_limit=10, + search_threshold=0.1, + recall_timeout=1.5, # seconds; a slow recall is skipped + capture="always", # "always" | "never" + ), +) +``` + +| Mode | Profile | Search | Use when | +| --- | --- | --- | --- | +| `profile` | Yes | No | You only need durable facts | +| `query` | No | Yes | You only need memories related to this turn | +| `full` | Yes | Yes | Default | + +One call is stored as a single document under custom id `lk-`, so a reconnect with the same session id updates that document instead of creating another. Explicit `remember` calls are separate facts and are not tied to the call document. + +## Links + +- [Docs](https://supermemory.ai/docs/integrations/livekit) +- [LiveKit Agents](https://docs.livekit.io/agents/) diff --git a/packages/livekit-sdk-python/pyproject.toml b/packages/livekit-sdk-python/pyproject.toml new file mode 100644 index 00000000..eaf142a3 --- /dev/null +++ b/packages/livekit-sdk-python/pyproject.toml @@ -0,0 +1,56 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "supermemory-livekit" +version = "0.1.0" +description = "Supermemory integration for LiveKit Agents — persistent memory for voice AI" +readme = "README.md" +license = "MIT" +requires-python = ">=3.10" +authors = [ + { name = "Supermemory", email = "support@supermemory.ai" } +] +keywords = [ + "supermemory", + "livekit", + "memory", + "voice-ai", + "agents", +] +classifiers = [ + "Development Status :: 4 - Beta", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Topic :: Scientific/Engineering :: Artificial Intelligence", +] +dependencies = [ + "livekit-agents>=1.2.0,<2", + "pydantic>=2.10.0", + "supermemory>=3.50.0", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.3.5", +] + +[project.urls] +Homepage = "https://supermemory.ai" +Documentation = "https://supermemory.ai/docs/integrations/livekit" +Repository = "https://github.com/supermemoryai/supermemory" + +[tool.hatch.build.targets.wheel] +packages = ["src/supermemory_livekit"] + +[tool.hatch.build.targets.sdist] +include = [ + "/src", + "/tests", + "/README.md", +] diff --git a/packages/livekit-sdk-python/src/supermemory_livekit/__init__.py b/packages/livekit-sdk-python/src/supermemory_livekit/__init__.py new file mode 100644 index 00000000..b2a1894a --- /dev/null +++ b/packages/livekit-sdk-python/src/supermemory_livekit/__init__.py @@ -0,0 +1,29 @@ +"""Supermemory for LiveKit Agents.""" + +from importlib.metadata import PackageNotFoundError, version +from typing import Any + +from .exceptions import ConfigurationError, SupermemoryLiveKitError +from .memory import InputParams, SupermemoryLiveKit + +try: + __version__ = version("supermemory-livekit") +except PackageNotFoundError: + __version__ = "0.1.0" + + +def __getattr__(name: str) -> Any: + if name == "SupermemoryAgent": + from .agent import SupermemoryAgent + + return SupermemoryAgent + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + + +__all__ = [ + "ConfigurationError", + "InputParams", + "SupermemoryAgent", + "SupermemoryLiveKit", + "SupermemoryLiveKitError", +] diff --git a/packages/livekit-sdk-python/src/supermemory_livekit/agent.py b/packages/livekit-sdk-python/src/supermemory_livekit/agent.py new file mode 100644 index 00000000..a815088b --- /dev/null +++ b/packages/livekit-sdk-python/src/supermemory_livekit/agent.py @@ -0,0 +1,25 @@ +"""Agent subclass that recalls memory before each reply.""" + +from typing import Any + +from livekit.agents import Agent + +from .memory import SupermemoryLiveKit + + +class SupermemoryAgent(Agent): + """LiveKit agent that injects Supermemory before each user turn. + + Pass any extra tools through ``tools``. Memory tools are added for you. + Override ``on_user_turn_completed`` and call ``super()`` if you also need + the hook. + """ + + def __init__(self, memory: SupermemoryLiveKit, **kwargs: Any) -> None: + tools = list(kwargs.pop("tools", None) or []) + tools.extend(memory.tools()) + super().__init__(tools=tools, **kwargs) + self.memory = memory + + async def on_user_turn_completed(self, turn_ctx: Any, new_message: Any) -> None: + await self.memory.on_user_turn_completed(turn_ctx, new_message) diff --git a/packages/livekit-sdk-python/src/supermemory_livekit/exceptions.py b/packages/livekit-sdk-python/src/supermemory_livekit/exceptions.py new file mode 100644 index 00000000..44dbc635 --- /dev/null +++ b/packages/livekit-sdk-python/src/supermemory_livekit/exceptions.py @@ -0,0 +1,19 @@ +"""Errors raised while configuring the LiveKit integration.""" + +from typing import Optional + + +class SupermemoryLiveKitError(Exception): + def __init__(self, message: str, original_error: Optional[Exception] = None): + super().__init__(message) + self.message = message + self.original_error = original_error + + def __str__(self) -> str: + if self.original_error: + return f"{self.message}: {self.original_error}" + return self.message + + +class ConfigurationError(SupermemoryLiveKitError): + """Raised when the integration cannot start, for example a missing API key.""" diff --git a/packages/livekit-sdk-python/src/supermemory_livekit/identifiers.py b/packages/livekit-sdk-python/src/supermemory_livekit/identifiers.py new file mode 100644 index 00000000..24b6f4d2 --- /dev/null +++ b/packages/livekit-sdk-python/src/supermemory_livekit/identifiers.py @@ -0,0 +1,22 @@ +"""Map LiveKit identities onto Supermemory container tags and custom ids.""" + +import hashlib +import re + +_INVALID = re.compile(r"[^a-zA-Z0-9_:-]+") +_MAX_LEN = 100 + + +def to_identifier(value: str) -> str: + """Return a stable container tag or custom id, hashing only when sanitizing.""" + raw = value.strip() + if not raw: + raise ValueError("identifier is empty") + + cleaned = _INVALID.sub("_", raw).strip("_") or "id" + if cleaned == raw and len(cleaned) <= _MAX_LEN: + return cleaned + + digest = hashlib.sha256(raw.encode()).hexdigest()[:8] + head = cleaned[: _MAX_LEN - len(digest) - 1].strip("_") or "id" + return f"{head}_{digest}"[:_MAX_LEN] diff --git a/packages/livekit-sdk-python/src/supermemory_livekit/memory.py b/packages/livekit-sdk-python/src/supermemory_livekit/memory.py new file mode 100644 index 00000000..023cf5ab --- /dev/null +++ b/packages/livekit-sdk-python/src/supermemory_livekit/memory.py @@ -0,0 +1,414 @@ +"""Recall and capture memory for a LiveKit voice session.""" + +from __future__ import annotations + +import asyncio +import logging +import os +from typing import Any, Literal, Optional +from uuid import uuid4 + +from pydantic import BaseModel, Field + +from .exceptions import ConfigurationError +from .identifiers import to_identifier +from .utils import ( + _field, + deduplicate_memories, + format_memories_to_text, + format_tool_results, + is_injected_memory, + message_role, + message_text, + wrap_memory, +) + +logger = logging.getLogger("supermemory_livekit") + +_UNAVAILABLE = "I couldn't reach memory just now." +_ATTRIBUTE = "supermemory_container_tag" + + +class InputParams(BaseModel): + search_limit: int = Field(default=10, ge=1, le=20) + search_threshold: float = Field(default=0.1, ge=0.0, le=1.0) + system_prompt: str = Field(default="Relevant memory for this caller:\n\n") + mode: Literal["profile", "query", "full"] = "full" + recall_timeout: float = Field(default=1.5, gt=0.0, le=8.0) + capture: Literal["always", "never"] = "always" + + +class SupermemoryLiveKit: + """Persistent memory for a LiveKit Agents session. + + Call ``on_user_turn_completed`` before the LLM replies, and ``attach`` the + session so completed turns are stored. ``tools()`` exposes search, remember, + and forget to the model. A Supermemory outage never fails the call. + """ + + InputParams = InputParams + + def __init__( + self, + *, + api_key: Optional[str] = None, + container_tag: Optional[str] = None, + session_id: Optional[str] = None, + params: Optional[InputParams] = None, + base_url: Optional[str] = None, + client: Any = None, + ) -> None: + self.api_key = api_key or os.getenv("SUPERMEMORY_API_KEY") + if not self.api_key and client is None: + raise ConfigurationError( + "API key is required. Pass api_key or set SUPERMEMORY_API_KEY." + ) + + self.params = params or InputParams() + self.container_tag: Optional[str] = None + self.session_id: Optional[str] = None + self._generated_session = f"session-{uuid4().hex[:12]}" + self._client = client if client is not None else self._build_client(base_url) + self._seen: set[str] = set() + self._buffer: list[dict[str, str]] = [] + self._lock = asyncio.Lock() + self._flush_task: Optional[asyncio.Task[None]] = None + self._session: Any = None + self._shutdown_registered = False + + if container_tag or session_id: + self.bind(container_tag=container_tag, session_id=session_id) + + def _build_client(self, base_url: Optional[str]) -> Any: + try: + import supermemory + except ImportError as exc: + raise ConfigurationError( + "supermemory is not installed. Install with: pip install supermemory-livekit" + ) from exc + kwargs: dict[str, Any] = {"api_key": self.api_key} + if base_url: + kwargs["base_url"] = base_url + return supermemory.AsyncSupermemory(**kwargs) + + def bind( + self, + *, + container_tag: Optional[str] = None, + session_id: Optional[str] = None, + participant: Any = None, + ) -> None: + """Scope memory to a container tag, or to a LiveKit participant identity.""" + if participant is not None and not container_tag: + attributes = getattr(participant, "attributes", None) or {} + if isinstance(attributes, dict) and attributes.get(_ATTRIBUTE): + container_tag = str(attributes[_ATTRIBUTE]) + else: + container_tag = getattr(participant, "identity", None) + + if container_tag is not None: + raw = str(container_tag).strip() + if not raw: + if participant is None: + raise ConfigurationError("container_tag is empty") + else: + scoped = to_identifier(raw) + if scoped != raw: + logger.info("scoped memory to container tag %s", scoped) + self.container_tag = scoped + if session_id: + self.session_id = str(session_id) + if self._buffer and self.container_tag: + self._schedule_flush() + + def tools(self) -> list[Any]: + from .tools import build_tools + + return build_tools(self) + + async def preload(self, chat_ctx: Any) -> bool: + """Inject the caller profile into the initial chat context. Returns whether anything was added.""" + text = await self._recall_text(query=None) + if not text: + return False + self._inject(chat_ctx, text, created_at=None) + return True + + async def on_user_turn_completed(self, turn_ctx: Any, new_message: Any) -> None: + """Retrieve memory for this turn and insert it just before the user message.""" + query = message_text(new_message) + if not query: + return + try: + text = await self._recall_text(query=query) + if not text: + return + created_at = getattr(new_message, "created_at", None) + before = created_at - 0.001 if isinstance(created_at, (int, float)) else None + self._strip_injected(turn_ctx) + self._inject(turn_ctx, text, created_at=before) + except Exception: + logger.warning("memory inject failed", exc_info=True) + + def attach( + self, + session: Any, + *, + container_tag: Optional[str] = None, + session_id: Optional[str] = None, + ) -> None: + """Store completed user and assistant turns from this session.""" + if container_tag or session_id: + self.bind(container_tag=container_tag, session_id=session_id) + if self._session is not session: + self._detach() + session.on("conversation_item_added", self._on_conversation_item) + session.on("close", self._on_close) + self._session = session + self._register_shutdown() + + async def aclose(self) -> None: + """Flush captured turns. Safe to call more than once.""" + task = self._flush_task + if task is not None and not task.done(): + try: + await task + except Exception: + logger.warning("memory capture failed while closing", exc_info=True) + for _ in range(2): + if not self._buffer: + return + try: + await self._flush() + return + except Exception: + logger.warning("memory capture failed while closing", exc_info=True) + + async def search(self, query: str) -> str: + tag = self.container_tag + if not tag: + return "Memory is not scoped to a caller yet." + if not query.strip(): + return "No matching memories." + try: + response = await asyncio.wait_for( + self._client.search.memories( + q=query, + container_tag=tag, + limit=self.params.search_limit, + threshold=self.params.search_threshold, + search_mode="hybrid", + ), + timeout=4.0, + ) + except Exception: + logger.warning("memory search failed", exc_info=True) + return _UNAVAILABLE + results = list(_field(response, "results", default=[]) or []) + return format_tool_results(results[: self.params.search_limit]) + + async def remember(self, memory_text: str) -> str: + tag = self.container_tag + text = memory_text.strip() + if not tag: + return "Memory is not scoped to a caller yet." + if not text: + return "Nothing to remember." + try: + await asyncio.wait_for( + self._client.add( + content=text, + container_tag=tag, + metadata={"source": "livekit", "kind": "explicit"}, + ), + timeout=4.0, + ) + except Exception: + logger.warning("memory remember failed", exc_info=True) + return _UNAVAILABLE + return "Saved." + + async def forget(self, *, memory_id: str = "", memory_text: str = "") -> str: + tag = self.container_tag + if not tag: + return "Memory is not scoped to a caller yet." + memory_id = memory_id.strip() + memory_text = memory_text.strip() + if not memory_id and not memory_text: + return "Pass a memory id from search_memories, or the exact memory text." + kwargs: dict[str, Any] = {"container_tag": tag} + if memory_id: + kwargs["id"] = memory_id + if memory_text: + kwargs["content"] = memory_text + try: + await asyncio.wait_for(self._client.memories.forget(**kwargs), timeout=4.0) + except Exception: + logger.warning("memory forget failed", exc_info=True) + return _UNAVAILABLE + return "Forgotten." + + async def _recall_text(self, *, query: Optional[str]) -> Optional[str]: + if not self.container_tag or self._client is None: + return None + include_profile = self.params.mode in ("profile", "full") + include_search = self.params.mode in ("query", "full") and bool(query) + if self.params.mode == "profile": + include_search = False + if not include_profile and not include_search: + return None + try: + memories = await asyncio.wait_for( + self._retrieve(query if include_search else None), + timeout=self.params.recall_timeout, + ) + except TimeoutError: + logger.warning("memory recall timed out after %.2fs", self.params.recall_timeout) + return None + except Exception: + logger.warning("memory recall failed", exc_info=True) + return None + + profile = memories["profile"] + deduped = deduplicate_memories( + static=profile["static"] if include_profile else [], + dynamic=profile["dynamic"] if include_profile else [], + search_results=memories["search_results"] if include_search else [], + ) + text = format_memories_to_text( + deduped, + system_prompt=self.params.system_prompt, + include_static=include_profile, + include_dynamic=include_profile, + include_search=include_search, + ) + return text or None + + async def _retrieve(self, query: Optional[str]) -> dict[str, Any]: + kwargs: dict[str, Any] = { + "container_tag": self.container_tag, + "timeout": self.params.recall_timeout, + } + if query: + kwargs["q"] = query + kwargs["threshold"] = self.params.search_threshold + response = await self._client.profile(**kwargs) + profile = _field(response, "profile") + search_response = _field(response, "search_results", "searchResults") + raw_results = list(_field(search_response, "results", default=[]) or []) + return { + "profile": { + "static": list(_field(profile, "static", default=[]) or []), + "dynamic": list(_field(profile, "dynamic", default=[]) or []), + }, + "search_results": raw_results[: self.params.search_limit], + } + + # LiveKit orders the user turn by timestamp, so a late add would land after it. + def _inject(self, chat_ctx: Any, text: str, *, created_at: Optional[float]) -> None: + wrapped = wrap_memory(text) + if created_at is None: + chat_ctx.add_message(role="assistant", content=wrapped) + return + try: + chat_ctx.add_message(role="assistant", content=wrapped, created_at=created_at) + except TypeError: + chat_ctx.add_message(role="assistant", content=wrapped) + + def _strip_injected(self, chat_ctx: Any) -> None: + items = list(getattr(chat_ctx, "items", []) or []) + for item in items: + text = message_text(item) + if message_role(item) != "assistant" or not text or not is_injected_memory(text): + continue + item_id = getattr(item, "id", None) + if item_id is not None and hasattr(chat_ctx, "remove"): + chat_ctx.remove(item_id) + elif hasattr(chat_ctx, "items") and item in chat_ctx.items: + chat_ctx.items.remove(item) + + def _on_conversation_item(self, event: Any) -> None: + item = getattr(event, "item", event) + role = message_role(item) + if role not in ("user", "assistant"): + return + text = message_text(item) + if not text or is_injected_memory(text): + return + item_id = getattr(item, "id", None) or f"anon-{id(item)}" + if item_id in self._seen: + return + self._seen.add(item_id) + self._buffer.append({"role": role, "content": text}) + if role == "assistant": + self._schedule_flush() + + def _on_close(self, _event: Any = None) -> None: + self._schedule_flush() + + def _schedule_flush(self) -> None: + if self.params.capture != "always" or not self._buffer or not self.container_tag: + return + try: + loop = asyncio.get_running_loop() + except RuntimeError: + return + if self._flush_task is not None and not self._flush_task.done(): + return + self._flush_task = loop.create_task(self._flush()) + + async def _flush(self) -> None: + while True: + async with self._lock: + if self.params.capture != "always" or not self._buffer or not self.container_tag: + return + batch = self._buffer + self._buffer = [] + try: + await self._store(batch) + except Exception: + async with self._lock: + self._buffer = batch + self._buffer + raise + + async def _store(self, messages: list[dict[str, str]]) -> None: + lines = [ + f"{'User' if message['role'] == 'user' else 'Assistant'}: {message['content']}" + for message in messages + ] + await self._client.add( + content="\n".join(lines), + container_tag=self.container_tag, + custom_id=self._custom_id(), + metadata={"source": "livekit", "kind": "conversation"}, + ) + + def _custom_id(self) -> str: + raw = self.session_id or self._generated_session + return to_identifier(f"lk-{raw}") + + def _detach(self) -> None: + session = self._session + if session is None or not hasattr(session, "off"): + self._session = None + return + session.off("conversation_item_added", self._on_conversation_item) + session.off("close", self._on_close) + self._session = None + + def _register_shutdown(self) -> None: + if self._shutdown_registered: + return + try: + from livekit.agents import get_job_context + + try: + ctx = get_job_context(required=False) + except TypeError: + ctx = get_job_context() + except Exception: + return + if ctx is None or not hasattr(ctx, "add_shutdown_callback"): + return + ctx.add_shutdown_callback(self.aclose) + self._shutdown_registered = True diff --git a/packages/livekit-sdk-python/src/supermemory_livekit/tools.py b/packages/livekit-sdk-python/src/supermemory_livekit/tools.py new file mode 100644 index 00000000..e07f1189 --- /dev/null +++ b/packages/livekit-sdk-python/src/supermemory_livekit/tools.py @@ -0,0 +1,43 @@ +"""LiveKit function tools bound to one memory scope.""" + +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from .memory import SupermemoryLiveKit + + +def build_tools(memory: "SupermemoryLiveKit") -> list[Any]: + from livekit.agents import RunContext, function_tool + + @function_tool() + async def search_memories(context: RunContext, query: str) -> str: + """Search long-term memory for facts, preferences, and past calls. + + Args: + query: What to look up, using the caller's words, names, or topic. + """ + del context + return await memory.search(query) + + @function_tool() + async def remember(context: RunContext, fact: str) -> str: + """Save one durable fact the caller asked you to remember. + + Args: + fact: A single sentence fact, preference, decision, or correction. + """ + del context + return await memory.remember(fact) + + @function_tool() + async def forget(context: RunContext, memory_id: str = "", fact: str = "") -> str: + """Forget one learned fact. Pass a memory id from search_memories, or the exact text. + + Args: + memory_id: Id from a search_memories result. Leave empty to match text instead. + fact: Exact memory text to forget when you do not have an id. + """ + del context + return await memory.forget(memory_id=memory_id, memory_text=fact) + + return [search_memories, remember, forget] diff --git a/packages/livekit-sdk-python/src/supermemory_livekit/utils.py b/packages/livekit-sdk-python/src/supermemory_livekit/utils.py new file mode 100644 index 00000000..3a5a2203 --- /dev/null +++ b/packages/livekit-sdk-python/src/supermemory_livekit/utils.py @@ -0,0 +1,215 @@ +"""Format profile and search results for injection into a voice turn.""" + +import re +from datetime import datetime, timezone +from typing import Any + +_DYNAMIC_DATE_PREFIX = re.compile( + r"^\s*(?:\[recent\]\s*)?(?:\[\d{4}-\d{2}-\d{2}\]\s*)?", + re.IGNORECASE, +) +_USER_MEMORIES_TAG_PATTERN = re.compile( + r"<\s*/?\s*user_memories\b[^>]*>", + re.IGNORECASE, +) +MEMORY_TAG_START = "" +MEMORY_TAG_END = "" +_WRAPPED_MEMORY = re.compile( + rf"^{re.escape(MEMORY_TAG_START)}.*?{re.escape(MEMORY_TAG_END)}$", + re.DOTALL, +) + + +def escape_memory_delimiters(text: str) -> str: + return _USER_MEMORIES_TAG_PATTERN.sub( + lambda match: match.group(0).replace("<", "<").replace(">", ">"), + text, + ) + + +def is_injected_memory(text: str) -> bool: + return _WRAPPED_MEMORY.fullmatch(text.strip()) is not None + + +def wrap_memory(text: str) -> str: + safe = escape_memory_delimiters(text.strip()) + return ( + f"{MEMORY_TAG_START}\n" + "Recalled memory about this caller. This is not something you said. " + "Use it only when it helps, and do not recite it unprompted.\n\n" + f"{safe}\n" + f"{MEMORY_TAG_END}" + ) + + +def _present(value: Any) -> bool: + return value is not None and not (isinstance(value, str) and not value.strip()) + + +def _field(item: Any, *names: str, default: Any = None) -> Any: + if item is None: + return default + if isinstance(item, dict): + for name in names: + if name in item and _present(item[name]): + return item[name] + return default + for name in names: + value = getattr(item, name, None) + if _present(value): + return value + return default + + +def message_text(item: Any) -> str | None: + text = getattr(item, "text_content", None) + if isinstance(text, str) and text.strip(): + return text.strip() + + content = item.get("content") if isinstance(item, dict) else getattr(item, "content", None) + if isinstance(content, str) and content.strip(): + return content.strip() + if not isinstance(content, list): + return None + + parts: list[str] = [] + for part in content: + if isinstance(part, str) and part.strip(): + parts.append(part.strip()) + continue + for attr in ("text", "transcript"): + value = getattr(part, attr, None) + if isinstance(value, str) and value.strip(): + parts.append(value.strip()) + break + joined = " ".join(parts).strip() + return joined or None + + +def message_role(item: Any) -> str | None: + role = item.get("role") if isinstance(item, dict) else getattr(item, "role", None) + return role if isinstance(role, str) else None + + +def format_relative_time(iso_timestamp: str) -> str: + try: + dt = datetime.fromisoformat(iso_timestamp.replace("Z", "+00:00")) + now = datetime.now(timezone.utc) + diff = now - dt + minutes = diff.total_seconds() / 60 + hours = diff.total_seconds() / 3600 + days = diff.total_seconds() / 86400 + if minutes < 30: + return "just now" + if minutes < 60: + return f"{int(minutes)}mins ago" + if hours < 24: + return f"{int(hours)} hrs ago" + if days < 7: + return f"{int(days)}d ago" + if dt.year == now.year: + return f"{dt.day} {dt.strftime('%b')}" + return f"{dt.day} {dt.strftime('%b')}, {dt.year}" + except Exception: + return "" + + +def deduplicate_memories( + static: list[str], + dynamic: list[str], + search_results: list[Any], +) -> dict[str, list[Any]]: + seen: set[str] = set() + + def comparison_key(memory: str) -> str: + without_prefix = _DYNAMIC_DATE_PREFIX.sub("", memory.strip()) + return " ".join(without_prefix.split()).casefold() + + def unique_strings(memories: list[str]) -> list[str]: + out: list[str] = [] + for memory in memories: + if not isinstance(memory, str): + continue + key = comparison_key(memory) + if key and key not in seen: + seen.add(key) + out.append(memory) + return out + + def unique_search(results: list[Any]) -> list[Any]: + out: list[Any] = [] + for result in results: + memory = ( + result + if isinstance(result, str) + else _field(result, "memory", "chunk", "content", default="") + ) + if not isinstance(memory, str): + memory = "" + key = comparison_key(memory) + if key and key not in seen: + seen.add(key) + out.append(result) + return out + + return { + "static": unique_strings(static), + "dynamic": unique_strings(dynamic), + "search_results": unique_search(search_results), + } + + +def format_memories_to_text( + memories: dict[str, list[Any]], + *, + system_prompt: str, + include_static: bool, + include_dynamic: bool, + include_search: bool, +) -> str: + sections: list[str] = [] + static = memories["static"] + dynamic = memories["dynamic"] + search_results = memories["search_results"] + + if include_static and static: + sections.append("## User Profile\n" + "\n".join(f"- {item}" for item in static)) + if include_dynamic and dynamic: + sections.append("## Recent Context\n" + "\n".join(f"- {item}" for item in dynamic)) + if include_search and search_results: + lines: list[str] = [] + for item in search_results: + if isinstance(item, str): + lines.append(f"- {item}") + continue + memory = _field(item, "memory", "chunk", "content", default="") + updated_at = _field(item, "updatedAt", "updated_at", default="") + time_str = format_relative_time(updated_at) if isinstance(updated_at, str) else "" + prefix = f"[{time_str}] " if time_str else "" + if isinstance(memory, str) and memory.strip(): + lines.append(f"- {prefix}{memory.strip()}") + if lines: + sections.append("## Relevant Memories\n" + "\n".join(lines)) + + if not sections: + return "" + return f"{system_prompt}\n" + "\n\n".join(sections) + + +def format_tool_results(results: list[Any]) -> str: + lines: list[str] = [] + for item in results: + if isinstance(item, str) and item.strip(): + lines.append(f"- {item.strip()}") + continue + memory = _field(item, "memory") + chunk = _field(item, "chunk", "content", default="") + text = memory if isinstance(memory, str) and memory.strip() else chunk + if not isinstance(text, str) or not text.strip(): + continue + memory_id = _field(item, "id") + if isinstance(memory, str) and memory.strip() and isinstance(memory_id, str) and memory_id: + lines.append(f"- {memory.strip()} (id: {memory_id})") + else: + lines.append(f"- {text.strip()}") + return "\n".join(lines) if lines else "No matching memories." diff --git a/packages/livekit-sdk-python/tests/test_memory.py b/packages/livekit-sdk-python/tests/test_memory.py new file mode 100644 index 00000000..f1e0e2c5 --- /dev/null +++ b/packages/livekit-sdk-python/tests/test_memory.py @@ -0,0 +1,391 @@ +from __future__ import annotations + +import asyncio +import sys +import types +import unittest +from types import SimpleNamespace + +from supermemory_livekit import ConfigurationError, InputParams, SupermemoryLiveKit +from supermemory_livekit.identifiers import to_identifier +from supermemory_livekit.utils import format_tool_results, is_injected_memory, wrap_memory + + +class FakeProfile: + def __init__(self, static=None, dynamic=None, results=None, delay=0): + self.static = static + self.dynamic = dynamic + self.results = results or [] + self.delay = delay + self.calls = [] + + async def __call__(self, **kwargs): + self.calls.append(kwargs) + if self.delay: + await asyncio.sleep(self.delay) + return SimpleNamespace( + profile=None + if self.static is None and self.dynamic is None + else SimpleNamespace(static=self.static or [], dynamic=self.dynamic or []), + search_results=SimpleNamespace(results=self.results), + ) + + +class FakeSearch: + def __init__(self): + self.calls = [] + + async def memories(self, **kwargs): + self.calls.append(kwargs) + return SimpleNamespace( + results=[ + SimpleNamespace(id="mem_1", memory="Likes short answers"), + SimpleNamespace(id="chunk_1", chunk="raw transcript", memory=None), + ] + ) + + +class FakeMemories: + def __init__(self): + self.calls = [] + + async def forget(self, **kwargs): + self.calls.append(kwargs) + + +class FakeClient: + def __init__(self, profile=None): + self.profile = profile or FakeProfile(static=[], dynamic=[]) + self.search = FakeSearch() + self.memories = FakeMemories() + self.added = [] + + async def add(self, **kwargs): + self.added.append(kwargs) + return SimpleNamespace(id="doc_1") + + +class ChatCtx: + def __init__(self): + self.items = [] + + def add_message(self, *, role, content, created_at=None): + item = SimpleNamespace( + id=f"item_{len(self.items)}", + role=role, + content=content, + text_content=content, + created_at=created_at, + ) + self.items.append(item) + return item + + def remove(self, item_id): + self.items = [item for item in self.items if item.id != item_id] + + +class Session: + def __init__(self): + self.handlers = {} + + def on(self, event, callback): + self.handlers.setdefault(event, []).append(callback) + + def off(self, event, callback): + self.handlers[event] = [cb for cb in self.handlers.get(event, []) if cb is not callback] + + def emit(self, event, payload): + for callback in list(self.handlers.get(event, [])): + callback(payload) + + +def memory(client, **kwargs): + return SupermemoryLiveKit(api_key="sm_test", client=client, **kwargs) + + +class MemoryTests(unittest.TestCase): + def test_missing_api_key(self): + with self.assertRaises(ConfigurationError): + SupermemoryLiveKit(api_key="", client=None) + + def test_empty_container_tag(self): + with self.assertRaises(ConfigurationError): + memory(FakeClient(), container_tag=" ") + + def test_empty_profile_does_not_inject(self): + client = FakeClient(FakeProfile(static=[], dynamic=[])) + plugin = memory(client, container_tag="user_1") + ctx = ChatCtx() + + added = asyncio.run(plugin.on_user_turn_completed(ctx, SimpleNamespace(text_content="hi", created_at=10))) + + self.assertIsNone(added) + self.assertEqual(ctx.items, []) + self.assertEqual(client.profile.calls[0]["q"], "hi") + + def test_null_profile_is_empty(self): + client = FakeClient(FakeProfile()) + plugin = memory(client, container_tag="user_1") + ctx = ChatCtx() + + asyncio.run(plugin.preload(ctx)) + + self.assertEqual(ctx.items, []) + + def test_recall_inserts_before_user_timestamp_and_replaces_previous(self): + client = FakeClient( + FakeProfile( + static=["Name is Ada"], + dynamic=["Working on voice agents"], + results=[SimpleNamespace(memory="Prefers concise replies", updatedAt="2026-09-20T00:00:00Z")], + ) + ) + plugin = memory(client, container_tag="user_1") + ctx = ChatCtx() + ctx.add_message(role="assistant", content=wrap_memory("stale"), created_at=1) + user = SimpleNamespace(text_content="what do you remember?", created_at=20.0) + + asyncio.run(plugin.on_user_turn_completed(ctx, user)) + + self.assertEqual(len(ctx.items), 1) + injected = ctx.items[0] + self.assertEqual(injected.role, "assistant") + self.assertAlmostEqual(injected.created_at, 19.999) + self.assertIn("Name is Ada", injected.content) + self.assertIn("Prefers concise replies", injected.content) + self.assertNotIn("stale", injected.content) + self.assertTrue(is_injected_memory(injected.content)) + self.assertEqual(client.profile.calls[0]["q"], "what do you remember?") + self.assertEqual(client.profile.calls[0]["container_tag"], "user_1") + + def test_query_mode_skips_profile_sections(self): + client = FakeClient( + FakeProfile(static=["Name is Ada"], results=[SimpleNamespace(memory="Lives in Lisbon")]) + ) + plugin = memory( + client, + container_tag="user_1", + params=InputParams(mode="query"), + ) + ctx = ChatCtx() + + asyncio.run(plugin.on_user_turn_completed(ctx, SimpleNamespace(text_content="where?", created_at=2))) + + self.assertIn("Lives in Lisbon", ctx.items[0].content) + self.assertNotIn("Name is Ada", ctx.items[0].content) + + def test_timeout_and_errors_do_not_fail_the_turn(self): + slow = FakeClient(FakeProfile(static=["Name is Ada"], delay=0.05)) + plugin = memory(slow, container_tag="user_1", params=InputParams(recall_timeout=0.01)) + ctx = ChatCtx() + asyncio.run(plugin.on_user_turn_completed(ctx, SimpleNamespace(text_content="hi", created_at=1))) + self.assertEqual(ctx.items, []) + + class Boom: + async def __call__(self, **kwargs): + raise RuntimeError("down") + + broken = FakeClient() + broken.profile = Boom() + plugin = memory(broken, container_tag="user_1") + asyncio.run(plugin.on_user_turn_completed(ctx, SimpleNamespace(text_content="hi", created_at=1))) + self.assertEqual(ctx.items, []) + + def test_participant_attribute_beats_identity_and_sanitizes(self): + plugin = memory(FakeClient()) + plugin.bind( + participant=SimpleNamespace( + identity="sip:+15551212", + attributes={"supermemory_container_tag": "user_42"}, + ) + ) + self.assertEqual(plugin.container_tag, "user_42") + + plugin.bind(participant=SimpleNamespace(identity="user@example.com", attributes={})) + self.assertNotEqual(plugin.container_tag, "user@example.com") + self.assertEqual(plugin.container_tag, to_identifier("user@example.com")) + self.assertLessEqual(len(plugin.container_tag), 100) + + def test_explicit_tag_beats_participant(self): + plugin = memory(FakeClient()) + plugin.bind( + container_tag="user_9", + participant=SimpleNamespace(identity="other", attributes={}), + ) + self.assertEqual(plugin.container_tag, "user_9") + + def test_capture_groups_new_turns_and_skips_injection(self): + client = FakeClient() + plugin = memory(client, container_tag="user_1", session_id="room 1") + session = Session() + plugin.attach(session) + + session.emit( + "conversation_item_added", + SimpleNamespace(item=SimpleNamespace(id="u1", role="user", text_content="I like tea")), + ) + session.emit( + "conversation_item_added", + SimpleNamespace( + item=SimpleNamespace(id="a0", role="assistant", text_content=wrap_memory("secret")) + ), + ) + session.emit( + "conversation_item_added", + SimpleNamespace(item=SimpleNamespace(id="a1", role="assistant", text_content="Noted.")), + ) + session.emit( + "conversation_item_added", + SimpleNamespace(item=SimpleNamespace(id="a1", role="assistant", text_content="Noted.")), + ) + asyncio.run(plugin.aclose()) + + self.assertEqual(len(client.added), 1) + stored = client.added[0] + self.assertEqual(stored["content"], "User: I like tea\nAssistant: Noted.") + self.assertEqual(stored["container_tag"], "user_1") + self.assertEqual(stored["custom_id"], to_identifier("lk-room 1")) + self.assertNotIn("secret", stored["content"]) + self.assertEqual(stored["metadata"]["source"], "livekit") + + def test_close_flushes_a_trailing_user_turn(self): + client = FakeClient() + plugin = memory(client, container_tag="user_1", session_id="room-2") + session = Session() + + async def scenario(): + plugin.attach(session) + session.emit( + "conversation_item_added", + SimpleNamespace(item=SimpleNamespace(id="u1", role="user", text_content="still here")), + ) + session.emit("close", SimpleNamespace()) + await asyncio.sleep(0) + + asyncio.run(scenario()) + self.assertEqual(client.added[0]["content"], "User: still here") + + def test_capture_never_does_not_store(self): + client = FakeClient() + plugin = memory( + client, + container_tag="user_1", + params=InputParams(capture="never"), + ) + session = Session() + plugin.attach(session) + session.emit( + "conversation_item_added", + SimpleNamespace(item=SimpleNamespace(id="u1", role="user", text_content="hi")), + ) + session.emit( + "conversation_item_added", + SimpleNamespace(item=SimpleNamespace(id="a1", role="assistant", text_content="hello")), + ) + asyncio.run(plugin.aclose()) + self.assertEqual(client.added, []) + + def test_store_failure_is_retried_on_close(self): + client = FakeClient() + calls = {"n": 0} + + async def flaky(**kwargs): + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("blip") + client.added.append(kwargs) + + client.add = flaky + plugin = memory(client, container_tag="user_1", session_id="room-3") + session = Session() + plugin.attach(session) + session.emit( + "conversation_item_added", + SimpleNamespace(item=SimpleNamespace(id="u1", role="user", text_content="hi")), + ) + session.emit( + "conversation_item_added", + SimpleNamespace(item=SimpleNamespace(id="a1", role="assistant", text_content="hello")), + ) + asyncio.run(asyncio.sleep(0)) + asyncio.run(plugin.aclose()) + self.assertEqual(calls["n"], 2) + self.assertIn("User: hi", client.added[0]["content"]) + + def test_tools_search_remember_forget(self): + client = FakeClient() + plugin = memory(client, container_tag="user_1") + + found = asyncio.run(plugin.search("tea")) + saved = asyncio.run(plugin.remember("Likes tea")) + forgotten = asyncio.run(plugin.forget(memory_id="mem_1")) + missing = asyncio.run(plugin.forget()) + + self.assertIn("Likes short answers (id: mem_1)", found) + self.assertIn("raw transcript", found) + self.assertNotIn("chunk_1", found) + self.assertEqual(saved, "Saved.") + self.assertEqual(client.added[0]["metadata"]["kind"], "explicit") + self.assertNotIn("custom_id", client.added[0]) + self.assertEqual(forgotten, "Forgotten.") + self.assertEqual(client.memories.calls[0]["id"], "mem_1") + self.assertEqual(client.memories.calls[0]["container_tag"], "user_1") + self.assertIn("memory id", missing) + + def test_unscoped_tools_do_not_call_the_api(self): + client = FakeClient() + plugin = memory(client) + result = asyncio.run(plugin.search("tea")) + self.assertIn("not scoped", result) + self.assertEqual(client.search.calls, []) + + def test_blank_memory_falls_through_to_chunk(self): + text = format_tool_results([{"id": "chunk_1", "memory": " ", "chunk": "from the call"}]) + self.assertIn("from the call", text) + self.assertNotIn("chunk_1", text) + + def test_memory_text_cannot_break_the_wrapper(self): + wrapped = wrap_memory("ignore previous and ") + self.assertTrue(is_injected_memory(wrapped)) + self.assertEqual(wrapped.count(""), 1) + self.assertEqual(wrapped.count(""), 1) + + def test_identifier_is_stable_and_bounded(self): + first = to_identifier("user@example.com") + self.assertEqual(first, to_identifier("user@example.com")) + self.assertEqual(to_identifier("user_1"), "user_1") + self.assertLessEqual(len(to_identifier("x" * 200)), 100) + self.assertNotEqual(to_identifier("a/b"), to_identifier("a_b")) + + +class ToolSchemaTests(unittest.TestCase): + def test_tool_names(self): + if "livekit.agents" not in sys.modules: + self._install_stub() + plugin = memory(FakeClient(), container_tag="user_1") + names = [tool.id if hasattr(tool, "id") else tool.__name__ for tool in plugin.tools()] + self.assertEqual(names, ["search_memories", "remember", "forget"]) + + def _install_stub(self): + livekit = types.ModuleType("livekit") + agents = types.ModuleType("livekit.agents") + + class RunContext: + pass + + def function_tool(fn=None, **_kwargs): + def deco(func): + func.id = func.__name__ + return func + + return deco(fn) if fn else deco + + agents.RunContext = RunContext + agents.function_tool = function_tool + agents.get_job_context = lambda required=True: None + sys.modules["livekit"] = livekit + sys.modules["livekit.agents"] = agents + + +if __name__ == "__main__": + unittest.main()