diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml
index a1086347..6d730c98 100644
--- a/.github/workflows/ci-python.yml
+++ b/.github/workflows/ci-python.yml
@@ -21,13 +21,19 @@ env:
jobs:
agent-framework-python:
- name: agent-framework-python (Python ${{ matrix.python-version }})
+ name: agent-framework-python (${{ matrix.dependency-lane }}, Python ${{ matrix.python-version }})
runs-on: ubuntu-latest
timeout-minutes: 15
strategy:
fail-fast: false
matrix:
- python-version: ["3.10", "3.13"]
+ include:
+ - python-version: "3.10"
+ dependency-lane: minimum-supermemory
+ supermemory-version: "3.16.0"
+ - python-version: "3.13"
+ dependency-lane: current-supermemory
+ supermemory-version: "3.59.0"
defaults:
run:
working-directory: packages/agent-framework-python
@@ -50,15 +56,19 @@ jobs:
- name: Build wheel
run: python -m build --wheel --outdir "$RUNNER_TEMP/wheels"
- - name: Install wheel and runtime dependencies
- run: python -m pip install "$RUNNER_TEMP"/wheels/*.whl
+ - name: Install wheel and tested Supermemory SDK
+ run: >-
+ python -m pip install "$RUNNER_TEMP"/wheels/*.whl
+ "supermemory==${{ matrix.supermemory-version }}"
- name: Check dependency compatibility
run: python -m pip check
- - name: Verify installed wheel
+ - name: Verify installed wheel and SDK version
run: >-
- python -c "from pathlib import Path; import supermemory_agent_framework;
+ python -c "from importlib.metadata import version; from pathlib import Path;
+ import supermemory_agent_framework;
+ assert version('supermemory') == '${{ matrix.supermemory-version }}';
assert 'site-packages' in Path(supermemory_agent_framework.__file__).parts"
- name: Run tests
@@ -141,13 +151,13 @@ jobs:
matrix:
include:
- python-version: "3.10"
- dependency-lane: minimum-supermemory
- supermemory-spec: "supermemory==3.16.0"
+ dependency-lane: minimum-dependencies
supermemory-version: "3.16.0"
+ cartesia-line-version: "0.2.0"
- python-version: "3.12"
- dependency-lane: current-supermemory
- supermemory-spec: "supermemory==3.59.0"
+ dependency-lane: current-dependencies
supermemory-version: "3.59.0"
+ cartesia-line-version: "0.2.17"
defaults:
run:
working-directory: packages/cartesia-sdk-python
@@ -164,25 +174,29 @@ jobs:
cache: pip
cache-dependency-path: packages/cartesia-sdk-python/pyproject.toml
- - name: Install build tools and lightweight test dependencies
- run: >-
- python -m pip install build pytest "loguru>=0.7.3" "pydantic>=2.10.0"
- "${{ matrix.supermemory-spec }}"
+ - 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 without the voice framework
- run: python -m pip install --no-deps "$RUNNER_TEMP"/wheels/*.whl
+ - name: Install wheel and tested runtime dependencies
+ run: >-
+ python -m pip install "$RUNNER_TEMP"/wheels/*.whl
+ "supermemory==${{ matrix.supermemory-version }}"
+ "cartesia-line==${{ matrix.cartesia-line-version }}"
- - name: Run tests against the installed wheel and real lightweight dependencies
+ - name: Check dependency compatibility
+ run: python -m pip check
+
+ - name: Verify real Cartesia Line integration and run tests
run: >-
python -c "from importlib.metadata import version;
- from pathlib import Path; import loguru, pydantic, supermemory, pytest;
+ from pathlib import Path; import line, pytest, supermemory_cartesia;
assert version('supermemory') == '${{ matrix.supermemory-version }}';
- result = pytest.main(['-W', 'ignore::pytest.PytestAssertRewriteWarning', 'tests']);
- import supermemory_cartesia;
+ assert version('cartesia-line') == '${{ matrix.cartesia-line-version }}';
assert 'site-packages' in Path(supermemory_cartesia.__file__).parts;
+ result = pytest.main(['-W', 'ignore::pytest.PytestAssertRewriteWarning', 'tests']);
raise SystemExit(result)"
pipecat-sdk-python:
@@ -194,13 +208,13 @@ jobs:
matrix:
include:
- python-version: "3.10"
- dependency-lane: minimum-supermemory
- supermemory-spec: "supermemory==3.16.0"
+ dependency-lane: minimum-dependencies
supermemory-version: "3.16.0"
+ pipecat-version: "0.0.98"
- python-version: "3.12"
- dependency-lane: current-supermemory
- supermemory-spec: "supermemory==3.59.0"
+ dependency-lane: current-dependencies
supermemory-version: "3.59.0"
+ pipecat-version: "1.7.0"
defaults:
run:
working-directory: packages/pipecat-sdk-python
@@ -217,23 +231,27 @@ jobs:
cache: pip
cache-dependency-path: packages/pipecat-sdk-python/pyproject.toml
- - name: Install build tools and lightweight test dependencies
- run: >-
- python -m pip install build pytest "loguru>=0.7.3" "pydantic>=2.10.0"
- "${{ matrix.supermemory-spec }}"
+ - 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 without the voice framework
- run: python -m pip install --no-deps "$RUNNER_TEMP"/wheels/*.whl
+ - name: Install wheel and tested runtime dependencies
+ run: >-
+ python -m pip install "$RUNNER_TEMP"/wheels/*.whl
+ "supermemory==${{ matrix.supermemory-version }}"
+ "pipecat-ai==${{ matrix.pipecat-version }}"
- - name: Run tests against the installed wheel and real lightweight dependencies
+ - name: Check dependency compatibility
+ run: python -m pip check
+
+ - name: Verify real Pipecat integration and run tests
run: >-
python -c "from importlib.metadata import version;
- from pathlib import Path; import loguru, pydantic, supermemory, pytest;
+ from pathlib import Path; import pipecat, pytest, supermemory_pipecat;
assert version('supermemory') == '${{ matrix.supermemory-version }}';
- result = pytest.main(['-W', 'ignore::pytest.PytestAssertRewriteWarning', 'tests']);
- import supermemory_pipecat;
+ assert version('pipecat-ai') == '${{ matrix.pipecat-version }}';
assert 'site-packages' in Path(supermemory_pipecat.__file__).parts;
+ result = pytest.main(['-W', 'ignore::pytest.PytestAssertRewriteWarning', 'tests']);
raise SystemExit(result)"
diff --git a/.github/workflows/publish-agent-framework-python.yml b/.github/workflows/publish-agent-framework-python.yml
index ef74637d..fe678968 100644
--- a/.github/workflows/publish-agent-framework-python.yml
+++ b/.github/workflows/publish-agent-framework-python.yml
@@ -23,20 +23,22 @@ jobs:
working-directory: ./packages/agent-framework-python
steps:
- name: Checkout
- uses: actions/checkout@v4
+ uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+ with:
+ persist-credentials: false
- name: Setup Python
- uses: actions/setup-python@v5
+ uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
with:
python-version: "3.12"
- name: Install build dependencies
- run: pip install hatchling build
+ run: python -m pip install hatchling build
- name: Build package
run: python -m build
- name: Publish to PyPI
- uses: pypa/gh-action-pypi-publish@release/v1
+ uses: pypa/gh-action-pypi-publish@dc37677b2e1c63e2034f94d8a5b11f265b73ba33 # v1.14.2
with:
packages-dir: packages/agent-framework-python/dist/
diff --git a/.github/workflows/publish-cartesia-sdk-python.yml b/.github/workflows/publish-cartesia-sdk-python.yml
index aa65eb4f..1da4423a 100644
--- a/.github/workflows/publish-cartesia-sdk-python.yml
+++ b/.github/workflows/publish-cartesia-sdk-python.yml
@@ -23,20 +23,22 @@ jobs:
working-directory: ./packages/cartesia-sdk-python
steps:
- name: Checkout
- uses: actions/checkout@v4
+ uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+ with:
+ persist-credentials: false
- name: Setup Python
- uses: actions/setup-python@v5
+ uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
with:
python-version: "3.12"
- name: Install build dependencies
- run: pip install hatchling build
+ run: python -m pip install hatchling build
- name: Build package
run: python -m build
- name: Publish to PyPI
- uses: pypa/gh-action-pypi-publish@release/v1
+ uses: pypa/gh-action-pypi-publish@dc37677b2e1c63e2034f94d8a5b11f265b73ba33 # v1.14.2
with:
packages-dir: packages/cartesia-sdk-python/dist/
diff --git a/.github/workflows/publish-pipecat-sdk-python.yml b/.github/workflows/publish-pipecat-sdk-python.yml
index f98ab1c8..1182b98d 100644
--- a/.github/workflows/publish-pipecat-sdk-python.yml
+++ b/.github/workflows/publish-pipecat-sdk-python.yml
@@ -23,20 +23,22 @@ jobs:
working-directory: ./packages/pipecat-sdk-python
steps:
- name: Checkout
- uses: actions/checkout@v4
+ uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+ with:
+ persist-credentials: false
- name: Setup Python
- uses: actions/setup-python@v5
+ uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
with:
python-version: "3.12"
- name: Install build dependencies
- run: pip install hatchling build
+ run: python -m pip install hatchling build
- name: Build package
run: python -m build
- name: Publish to PyPI
- uses: pypa/gh-action-pypi-publish@release/v1
+ uses: pypa/gh-action-pypi-publish@dc37677b2e1c63e2034f94d8a5b11f265b73ba33 # v1.14.2
with:
packages-dir: packages/pipecat-sdk-python/dist/
diff --git a/packages/agent-framework-python/README.md b/packages/agent-framework-python/README.md
index 57b5d25b..3908692d 100644
--- a/packages/agent-framework-python/README.md
+++ b/packages/agent-framework-python/README.md
@@ -9,23 +9,13 @@ This package provides both **automatic memory injection middleware** and **manua
Install using uv (recommended):
```bash
-uv add --prerelease=allow supermemory-agent-framework
+uv add supermemory-agent-framework
```
Or with pip:
```bash
-pip install --pre supermemory-agent-framework
-```
-
-> **Note:** The `--prerelease=allow` / `--pre` flag is required because `agent-framework-core` depends on pre-release versions of Azure packages.
-
-For async HTTP support (recommended):
-
-```bash
-uv add supermemory-agent-framework[async]
-# or
-pip install 'supermemory-agent-framework[async]'
+pip install supermemory-agent-framework
```
## Quick Start
@@ -38,14 +28,19 @@ The easiest way to add memory capabilities is using the `SupermemoryChatMiddlewa
import asyncio
from agent_framework.openai import OpenAIResponsesClient
from supermemory_agent_framework import (
+ AgentSupermemory,
SupermemoryChatMiddleware,
SupermemoryMiddlewareOptions,
)
async def main():
- # Create Supermemory middleware
- middleware = SupermemoryChatMiddleware(
+ connection = AgentSupermemory(
+ api_key="your-supermemory-api-key",
container_tag="user-123",
+ )
+
+ middleware = SupermemoryChatMiddleware(
+ connection,
options=SupermemoryMiddlewareOptions(
mode="full", # "profile", "query", or "full"
verbose=True, # Enable logging
@@ -77,13 +72,16 @@ The most idiomatic way to add memory in Agent Framework, using the same pattern
import asyncio
from agent_framework import AgentSession
from agent_framework.openai import OpenAIResponsesClient
-from supermemory_agent_framework import SupermemoryContextProvider
+from supermemory_agent_framework import AgentSupermemory, SupermemoryContextProvider
async def main():
- # Create context provider
- provider = SupermemoryContextProvider(
- container_tag="user-123",
+ connection = AgentSupermemory(
api_key="your-supermemory-api-key",
+ container_tag="user-123",
+ )
+
+ provider = SupermemoryContextProvider(
+ connection,
mode="full",
store_conversations=True,
)
@@ -113,14 +111,14 @@ For explicit tool-based memory access:
```python
import asyncio
from agent_framework.openai import OpenAIResponsesClient
-from supermemory_agent_framework import SupermemoryTools
+from supermemory_agent_framework import AgentSupermemory, SupermemoryTools
async def main():
- # Create memory tools
- tools = SupermemoryTools(
+ connection = AgentSupermemory(
api_key="your-supermemory-api-key",
- config={"project_id": "my-project"},
+ container_tag="user-123",
)
+ tools = SupermemoryTools(connection)
# Create agent
agent = OpenAIResponsesClient().as_agent(
@@ -146,6 +144,7 @@ For maximum flexibility, use both middleware (automatic context injection) and t
import asyncio
from agent_framework.openai import OpenAIResponsesClient
from supermemory_agent_framework import (
+ AgentSupermemory,
SupermemoryChatMiddleware,
SupermemoryMiddlewareOptions,
SupermemoryTools,
@@ -153,14 +152,17 @@ from supermemory_agent_framework import (
async def main():
api_key = "your-supermemory-api-key"
-
- middleware = SupermemoryChatMiddleware(
- container_tag="user-123",
- options=SupermemoryMiddlewareOptions(mode="full"),
+ connection = AgentSupermemory(
api_key=api_key,
+ container_tag="user-123",
)
- tools = SupermemoryTools(api_key=api_key)
+ middleware = SupermemoryChatMiddleware(
+ connection,
+ options=SupermemoryMiddlewareOptions(mode="full"),
+ )
+
+ tools = SupermemoryTools(connection)
agent = OpenAIResponsesClient().as_agent(
name="MemoryAgent",
@@ -217,11 +219,20 @@ SupermemoryMiddlewareOptions(add_memory="never")
### Complete Configuration
```python
-SupermemoryMiddlewareOptions(
- conversation_id="chat-session-456", # Group messages into conversations
- verbose=True, # Enable detailed logging
- mode="full", # Use both profile and query
- add_memory="always" # Auto-save conversations
+connection = AgentSupermemory(
+ api_key="your-supermemory-api-key",
+ container_tag="user-123", # Memory scope
+ conversation_id="chat-session-456", # Groups stored conversations
+ entity_context="User is on the pro plan", # Optional fixed context
+)
+
+middleware = SupermemoryChatMiddleware(
+ connection,
+ options=SupermemoryMiddlewareOptions(
+ verbose=True,
+ mode="full",
+ add_memory="always",
+ ),
)
```
@@ -232,13 +243,11 @@ SupermemoryMiddlewareOptions(
Memory tools that integrate with Agent Framework's tool system.
```python
-tools = SupermemoryTools(
+connection = AgentSupermemory(
api_key="your-api-key",
- config={
- "project_id": "my-project", # or use container_tags
- "base_url": "https://custom.com", # optional
- }
+ container_tag="user-123",
)
+tools = SupermemoryTools(connection)
# Get FunctionTool instances for Agent.run()
agent_tools = tools.get_tools()
@@ -249,26 +258,19 @@ result = await tools.add_memory("User prefers dark mode")
result = await tools.get_profile()
```
+`search_memories` uses v4 hybrid search, so results can contain either a
+structured memory or a source chunk. The old Python-only `include_full_docs`
+argument is deprecated and ignored because v4 search does not return full
+source documents; it is not exposed to the model as a tool parameter.
+
### SupermemoryChatMiddleware
Chat middleware for automatic memory injection.
```python
middleware = SupermemoryChatMiddleware(
- container_tag="user-123", # Memory scope identifier
+ connection, # Shared AgentSupermemory connection
options=SupermemoryMiddlewareOptions(...),
- api_key="your-api-key", # Or set SUPERMEMORY_API_KEY env var
-)
-```
-
-### with_supermemory_middleware()
-
-Convenience function for creating middleware:
-
-```python
-middleware = with_supermemory_middleware(
- "user-123",
- SupermemoryMiddlewareOptions(mode="full"),
)
```
@@ -278,11 +280,9 @@ Context provider for the Agent Framework session pipeline (like Mem0):
```python
provider = SupermemoryContextProvider(
- container_tag="user-123",
- api_key="your-api-key", # Or set SUPERMEMORY_API_KEY env var
+ connection, # Shared AgentSupermemory connection
mode="full", # "profile", "query", or "full"
store_conversations=True, # Save conversations after each run
- conversation_id="chat-456", # Optional grouping ID
context_prompt="## Memories\n...", # Custom header for injected memories
verbose=True, # Enable logging
)
@@ -292,6 +292,7 @@ provider = SupermemoryContextProvider(
```python
from supermemory_agent_framework import (
+ AgentSupermemory,
SupermemoryConfigurationError,
SupermemoryAPIError,
SupermemoryNetworkError,
@@ -299,7 +300,7 @@ from supermemory_agent_framework import (
)
try:
- middleware = SupermemoryChatMiddleware("user-123")
+ connection = AgentSupermemory(container_tag="user-123")
except SupermemoryConfigurationError as e:
print(f"Configuration issue: {e}")
```
@@ -322,11 +323,8 @@ except SupermemoryConfigurationError as e:
### Required
- `agent-framework-core>=1.0.0rc3` - Microsoft Agent Framework
-- `supermemory>=3.1.0` - Supermemory client
-- `requests>=2.25.0` - HTTP requests (fallback)
-
-### Optional
-- `aiohttp>=3.8.0` - Async HTTP requests (recommended)
+- `supermemory>=3.16.0` - Supermemory client with v4 hybrid search support
+- `typing-extensions>=4.0.0` - Typing compatibility helpers
## Development
diff --git a/packages/agent-framework-python/pyproject.toml b/packages/agent-framework-python/pyproject.toml
index 65983083..d308aa57 100644
--- a/packages/agent-framework-python/pyproject.toml
+++ b/packages/agent-framework-python/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "supermemory-agent-framework"
-version = "1.0.0"
+version = "1.0.1"
description = "Memory tools and middleware for Microsoft Agent Framework with supermemory"
readme = "README.md"
license = "MIT"
@@ -25,7 +25,7 @@ classifiers = [
requires-python = ">=3.10"
dependencies = [
"agent-framework-core>=1.0.0rc3",
- "supermemory>=3.1.0",
+ "supermemory>=3.16.0",
"typing-extensions>=4.0.0",
]
diff --git a/packages/agent-framework-python/src/supermemory_agent_framework/context_provider.py b/packages/agent-framework-python/src/supermemory_agent_framework/context_provider.py
index 5359426e..a1d7b161 100644
--- a/packages/agent-framework-python/src/supermemory_agent_framework/context_provider.py
+++ b/packages/agent-framework-python/src/supermemory_agent_framework/context_provider.py
@@ -7,10 +7,10 @@ This is the idiomatic way to integrate persistent memory in Agent Framework,
following the same pattern as the built-in Mem0 integration.
"""
-from typing import Any, Literal, Optional
+from typing import Any, Literal
try:
- from agent_framework import BaseContextProvider
+ from agent_framework import BaseContextProvider # type: ignore[attr-defined]
except ImportError:
# Renamed in agent-framework-core 1.0.0 stable; the interface is
# unchanged (source_id __init__, before_run/after_run hooks with
diff --git a/packages/agent-framework-python/src/supermemory_agent_framework/tools.py b/packages/agent-framework-python/src/supermemory_agent_framework/tools.py
index c59ca6e1..2b70bdd7 100644
--- a/packages/agent-framework-python/src/supermemory_agent_framework/tools.py
+++ b/packages/agent-framework-python/src/supermemory_agent_framework/tools.py
@@ -4,7 +4,8 @@ Provides FunctionTool-compatible tools that can be passed to Agent.run(tools=[..
"""
import json
-from typing import Annotated, Any, TypedDict
+import warnings
+from typing import Annotated, Any, Optional, TypedDict
from agent_framework import FunctionTool, tool
@@ -37,6 +38,25 @@ class ProfileResult(TypedDict, total=False):
error: str | None
+def _to_jsonable(value: Any) -> Any:
+ """Convert generated SDK models into JSON-compatible structures."""
+ if isinstance(value, dict):
+ return {key: _to_jsonable(item) for key, item in value.items()}
+ if isinstance(value, (list, tuple)):
+ return [_to_jsonable(item) for item in value]
+
+ model_dump = getattr(value, "model_dump", None)
+ if callable(model_dump):
+ try:
+ return _to_jsonable(model_dump(mode="json"))
+ except TypeError:
+ # Compatibility with pydantic-like models whose model_dump does not
+ # accept Pydantic v2's ``mode`` argument.
+ return _to_jsonable(model_dump())
+
+ return value
+
+
class SupermemoryTools:
"""Memory tools for Microsoft Agent Framework.
@@ -64,27 +84,37 @@ class SupermemoryTools:
async def search_memories(
self,
information_to_get: Annotated[
- str, "Terms to search for in the user's memories"
+ str, "Terms to search for in stored memories and source content"
],
- include_full_docs: Annotated[
- bool,
- "Whether to include full document content. Defaults to true for better AI context.",
- ] = True,
+ include_full_docs: Optional[bool] = None,
limit: Annotated[int, "Maximum number of results to return"] = 10,
) -> str:
- """Search (recall) memories/details/information about the user or other facts or entities. Run when explicitly asked or when context about user's past choices would be helpful."""
- try:
- response = await self._client.search.execute(
- q=information_to_get,
- container_tags=[self._connection.container_tag],
- limit=limit,
- chunk_threshold=0.6,
- include_full_docs=include_full_docs,
+ """Search stored memories and source chunks.
+
+ ``include_full_docs`` remains a deprecated Python-only argument for
+ source compatibility. V4 search cannot return full source documents.
+ """
+ if include_full_docs is not None:
+ warnings.warn(
+ "include_full_docs is deprecated and ignored because v4 search "
+ "does not return full source documents",
+ DeprecationWarning,
+ stacklevel=2,
)
+
+ try:
+ response = await self._client.search.memories(
+ q=information_to_get,
+ container_tag=self._connection.container_tag,
+ limit=limit,
+ threshold=0.6,
+ search_mode="hybrid",
+ )
+ results = response.results or []
result: MemorySearchResult = {
"success": True,
- "results": response.results,
- "count": len(response.results) if response.results else 0,
+ "results": [_to_jsonable(item) for item in results],
+ "count": len(results),
}
return json.dumps(result, default=str)
except Exception as error:
@@ -107,7 +137,7 @@ class SupermemoryTools:
)
result: MemoryAddResult = {
"success": True,
- "memory": response,
+ "memory": _to_jsonable(response),
}
return json.dumps(result, default=str)
except Exception as error:
@@ -130,9 +160,13 @@ class SupermemoryTools:
response = await self._client.profile(**kwargs)
result: dict[str, Any] = {
"success": True,
- "profile": response.profile if hasattr(response, "profile") else None,
+ "profile": (
+ _to_jsonable(response.profile)
+ if hasattr(response, "profile")
+ else None
+ ),
"search_results": (
- response.search_results
+ _to_jsonable(response.search_results)
if hasattr(response, "search_results")
else None
),
@@ -152,11 +186,11 @@ class SupermemoryTools:
tool(
name="search_memories",
description=(
- "Search (recall) memories/details/information about the user or other "
- "facts or entities. Run when explicitly asked or when context about "
- "user's past choices would be helpful."
+ "Search stored memories and source chunks for relevant facts, preferences, "
+ "history, and context. Use proactively whenever prior context could help; "
+ "hybrid results can contain either a memory or a source chunk."
),
- )(self.search_memories),
+ )(self._search_memories_tool),
tool(
name="add_memory",
description=(
@@ -174,3 +208,16 @@ class SupermemoryTools:
),
)(self.get_profile),
]
+
+ async def _search_memories_tool(
+ self,
+ information_to_get: Annotated[
+ str, "Terms to search for in stored memories and source content"
+ ],
+ limit: Annotated[int, "Maximum number of results to return"] = 10,
+ ) -> str:
+ """Model-facing search wrapper that omits deprecated arguments."""
+ return await self.search_memories(
+ information_to_get=information_to_get,
+ limit=limit,
+ )
diff --git a/packages/agent-framework-python/src/supermemory_agent_framework/utils.py b/packages/agent-framework-python/src/supermemory_agent_framework/utils.py
index 8b8c9be0..1e4ee56a 100644
--- a/packages/agent-framework-python/src/supermemory_agent_framework/utils.py
+++ b/packages/agent-framework-python/src/supermemory_agent_framework/utils.py
@@ -1,6 +1,7 @@
"""Utility functions for Supermemory Agent Framework integration."""
import json
+import re
from typing import Any, Optional, Protocol
DEFAULT_CONTEXT_PROMPT = "The following are retrieved memories about the user."
@@ -92,17 +93,31 @@ def deduplicate_memories(
def extract_memory_text(item: Any) -> Optional[str]:
if item is None:
return None
- if isinstance(item, dict):
- memory = item.get("memory")
- if isinstance(memory, str):
- trimmed = memory.strip()
- return trimmed if trimmed else None
- return None
if isinstance(item, str):
trimmed = item.strip()
return trimmed if trimmed else None
+ if isinstance(item, dict):
+ for field in ("memory", "chunk", "content"):
+ value = item.get(field)
+ if isinstance(value, str) and value.strip():
+ return value.strip()
+ return None
+ # Stainless SDK returns pydantic models (attribute access, snake_case).
+ for field in ("memory", "chunk", "content"):
+ value = getattr(item, field, None)
+ if isinstance(value, str) and value.strip():
+ return value.strip()
return None
+ def comparison_key(memory: str) -> str:
+ """Remove Mono's dynamic-profile date decoration for comparison only."""
+ return re.sub(
+ r"^(?:\[Recent\]\s*)?\[\d{4}-\d{2}-\d{2}\]\s*",
+ "",
+ memory,
+ count=1,
+ ).strip()
+
static_memories: list[str] = []
seen_memories: set[str] = set()
@@ -110,21 +125,21 @@ def deduplicate_memories(
memory = extract_memory_text(item)
if memory is not None:
static_memories.append(memory)
- seen_memories.add(memory)
+ seen_memories.add(comparison_key(memory))
dynamic_memories: list[str] = []
for item in dynamic_items:
memory = extract_memory_text(item)
- if memory is not None and memory not in seen_memories:
+ if memory is not None and comparison_key(memory) not in seen_memories:
dynamic_memories.append(memory)
- seen_memories.add(memory)
+ seen_memories.add(comparison_key(memory))
search_memories: list[str] = []
for item in search_items:
memory = extract_memory_text(item)
- if memory is not None and memory not in seen_memories:
+ if memory is not None and comparison_key(memory) not in seen_memories:
search_memories.append(memory)
- seen_memories.add(memory)
+ seen_memories.add(comparison_key(memory))
return DeduplicatedMemories(
static=static_memories,
diff --git a/packages/cartesia-sdk-python/README.md b/packages/cartesia-sdk-python/README.md
index bb446a7c..978ec23f 100644
--- a/packages/cartesia-sdk-python/README.md
+++ b/packages/cartesia-sdk-python/README.md
@@ -23,6 +23,7 @@ async def get_agent(env, call_request):
# 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!"
@@ -101,7 +102,7 @@ read_only_agent = SupermemoryCartesiaAgent(
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** - Adds memories to event history as system 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
@@ -135,7 +136,7 @@ UserTurnEnded Event {content: "user message", history: [...]}
│ 1. Intercept UserTurnEnded │
│ 2. Extract user message │
│ 3. Query Supermemory API │
-│ 4. Enrich event.history with memories │
+│ 4. Add memories as per-turn context │
│ 5. Pass to wrapped LlmAgent │
│ 6. Store conversation (async background) │
└──────────────────────────────────────────────┘
@@ -155,7 +156,7 @@ Audio Output
| **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()` | Modify `event.history` |
+| **Memory Injection** | Modify `context.add_message()` | Pass per-turn `context` |
## Full Example with Tools
@@ -183,6 +184,7 @@ async def get_agent(env, call_request):
# 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.",
diff --git a/packages/cartesia-sdk-python/pyproject.toml b/packages/cartesia-sdk-python/pyproject.toml
index 81dac7fd..0de043a4 100644
--- a/packages/cartesia-sdk-python/pyproject.toml
+++ b/packages/cartesia-sdk-python/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "supermemory-cartesia"
-version = "0.1.1"
+version = "0.1.2"
description = "Supermemory integration for Cartesia Line - memory-enhanced voice agents"
readme = "README.md"
license = "MIT"
@@ -33,7 +33,7 @@ classifiers = [
]
dependencies = [
"supermemory>=3.16.0",
- "cartesia-line>=0.2.0",
+ "cartesia-line>=0.2.0,<0.3.0",
"pydantic>=2.10.0",
"loguru>=0.7.3",
]
diff --git a/packages/cartesia-sdk-python/requirements.txt b/packages/cartesia-sdk-python/requirements.txt
index dd4b8fe5..3cb93560 100644
--- a/packages/cartesia-sdk-python/requirements.txt
+++ b/packages/cartesia-sdk-python/requirements.txt
@@ -2,4 +2,4 @@
supermemory>=3.16.0
pydantic>=2.10.0
loguru>=0.7.3
-cartesia-line>=0.2.0
+cartesia-line>=0.2.0,<0.3.0
diff --git a/packages/cartesia-sdk-python/src/supermemory_cartesia/__init__.py b/packages/cartesia-sdk-python/src/supermemory_cartesia/__init__.py
index b21511d9..c1d27e29 100644
--- a/packages/cartesia-sdk-python/src/supermemory_cartesia/__init__.py
+++ b/packages/cartesia-sdk-python/src/supermemory_cartesia/__init__.py
@@ -5,12 +5,15 @@ enabling persistent memory and context enhancement for voice AI applications.
Example:
```python
+ import os
+
from supermemory_cartesia import SupermemoryCartesiaAgent, MemoryConfig
from line.llm_agent import LlmAgent, LlmConfig
# 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 assistant.",
introduction="Hello!"
@@ -22,6 +25,7 @@ Example:
agent=base_agent,
api_key=os.getenv("SUPERMEMORY_API_KEY"),
container_tag="user-123",
+ custom_id="conversation-456",
)
```
"""
@@ -46,7 +50,13 @@ from .utils import (
get_last_user_message,
)
-__version__ = "0.1.0"
+try:
+ from importlib.metadata import PackageNotFoundError, version
+
+ __version__ = version("supermemory-cartesia")
+except PackageNotFoundError:
+ # Source checkouts do not have installed distribution metadata.
+ __version__ = "0.1.2"
__all__ = [
# Main agent
diff --git a/packages/cartesia-sdk-python/src/supermemory_cartesia/agent.py b/packages/cartesia-sdk-python/src/supermemory_cartesia/agent.py
index 9019193e..27aba07b 100644
--- a/packages/cartesia-sdk-python/src/supermemory_cartesia/agent.py
+++ b/packages/cartesia-sdk-python/src/supermemory_cartesia/agent.py
@@ -5,6 +5,7 @@ Cartesia Line voice agents, adding persistent memory and context enrichment.
"""
import asyncio
+import inspect
import os
import re
from typing import Any, AsyncGenerator, Dict, List, Literal, Optional
@@ -13,7 +14,7 @@ from loguru import logger
from pydantic import BaseModel, Field
from .exceptions import ConfigurationError, MemoryRetrievalError
-from .utils import deduplicate_memories, format_memories_to_text
+from .utils import _field, deduplicate_memories, format_memories_to_text
try:
import supermemory
@@ -34,8 +35,7 @@ class SupermemoryCartesiaAgent:
"""Memory-enhanced wrapper for Cartesia Line agents.
This wrapper intercepts UserTurnEnded events, retrieves relevant memories
- from Supermemory, and enriches the conversation history before passing to
- the wrapped agent.
+ from Supermemory, and passes them as per-turn context to the wrapped agent.
Example:
```python
@@ -44,6 +44,7 @@ class SupermemoryCartesiaAgent:
base_agent = LlmAgent(
model="anthropic/claude-haiku-4-5-20251001",
+ api_key=os.getenv("ANTHROPIC_API_KEY"),
config=LlmConfig(
system_prompt="You are a helpful assistant.",
introduction="Hello! How can I help you today?"
@@ -141,8 +142,8 @@ class SupermemoryCartesiaAgent:
except Exception as e:
logger.error(f"[Supermemory] Failed to initialize client: {e}")
- self._messages_sent_count: int = 0
- self._last_query: Optional[str] = None
+ self._history_cursor: List[Dict[str, str]] = []
+ self._last_retrieval_event: Optional[str] = None
self._background_tasks: set = set() # Track background tasks to prevent GC
async def _retrieve_memories(self, query: str) -> Dict[str, Any]:
@@ -151,31 +152,31 @@ class SupermemoryCartesiaAgent:
raise MemoryRetrievalError("Supermemory client not initialized")
try:
- # Use primary container tag for profile retrieval
- kwargs: Dict[str, Any] = {"container_tag": self.container_tags[0]}
+ logger.info(f"[Supermemory] Retrieving memories for query: {query[:50]}...")
+ # One profile call: static + dynamic, and (when mode/query allow)
+ # search_results via `q` — keeps a single round trip for latency.
+ kwargs: Dict[str, Any] = {"container_tag": self.container_tags[0]}
if self.config.mode != "profile" and query:
kwargs["q"] = query
kwargs["threshold"] = self.config.search_threshold
- kwargs["extra_body"] = {"limit": self.config.search_limit}
-
- logger.info(f"[Supermemory] Retrieving memories for query: {query[:50]}...")
response = await asyncio.wait_for(
self._supermemory_client.profile(**kwargs),
- timeout=10.0
+ timeout=10.0,
)
# A user with no stored memories yet gets a null profile back, which
# is a normal case, not an error. Guard against it so we return an
# empty profile instead of raising AttributeError on response.profile.
- profile = getattr(response, "profile", None)
- profile_static = profile.static if profile is not None and profile.static else []
- profile_dynamic = profile.dynamic if profile is not None and profile.dynamic else []
+ profile = _field(response, "profile")
+ profile_static = list(_field(profile, "static", default=[]) or [])
+ profile_dynamic = list(_field(profile, "dynamic", default=[]) or [])
- search_results = []
- if response.search_results and response.search_results.results:
- search_results = response.search_results.results
+ search_results: List[Any] = []
+ search_response = _field(response, "search_results", "searchResults")
+ raw_search_results = _field(search_response, "results", default=[]) or []
+ search_results = list(raw_search_results)[: self.config.search_limit]
logger.info(
f"[Supermemory] Retrieved memories - static: {len(profile_static)}, "
@@ -292,54 +293,73 @@ class SupermemoryCartesiaAgent:
return str(content)
def _extract_conversation_from_history(self, history: list) -> List[Dict[str, str]]:
- """Extract messages from Cartesia event history."""
- messages = []
- seen = set()
+ """Extract messages, suppressing only adjacent duplicate representations."""
+ messages: List[Dict[str, str]] = []
+
+ def append_message(role: str, content: Any) -> None:
+ if role not in ("user", "assistant") or not isinstance(content, str) or not content:
+ return
+ message = {"role": role, "content": content}
+ if not messages or messages[-1] != message:
+ messages.append(message)
for item in history:
if isinstance(item, dict):
if item.get("role") in ("user", "assistant"):
- content = item.get("content", "")
- if content and content not in seen:
- messages.append(item)
- seen.add(content)
+ append_message(item["role"], item.get("content", ""))
continue
- event_type = getattr(item, 'type', None) or type(item).__name__
+ event_type = getattr(item, "type", None) or type(item).__name__
- if event_type in ('user_turn_ended', 'UserTurnEnded'):
- nested = getattr(item, 'content', [])
+ if event_type in ("user_turn_ended", "UserTurnEnded"):
+ nested = getattr(item, "content", [])
if isinstance(nested, list):
- for n in nested:
- if hasattr(n, 'content') and isinstance(n.content, str):
- if n.content not in seen:
- messages.append({"role": "user", "content": n.content})
- seen.add(n.content)
+ for nested_item in nested:
+ if hasattr(nested_item, "content"):
+ append_message("user", nested_item.content)
- elif event_type in ('agent_turn_ended', 'AgentTurnEnded'):
- nested = getattr(item, 'content', [])
+ elif event_type in ("agent_turn_ended", "AgentTurnEnded"):
+ nested = getattr(item, "content", [])
if isinstance(nested, list):
- texts = [n.content for n in nested if hasattr(n, 'content') and isinstance(n.content, str)]
+ texts = [
+ nested_item.content
+ for nested_item in nested
+ if hasattr(nested_item, "content") and isinstance(nested_item.content, str)
+ ]
if texts:
- content = " ".join(texts)
- if content not in seen:
- messages.append({"role": "assistant", "content": content})
- seen.add(content)
+ append_message("assistant", " ".join(texts))
- elif event_type in ('user_text_sent', 'UserTextSent'):
- content = getattr(item, 'content', '')
- if content and isinstance(content, str) and content not in seen:
- messages.append({"role": "user", "content": content})
- seen.add(content)
+ elif event_type in ("user_text_sent", "UserTextSent"):
+ append_message("user", getattr(item, "content", ""))
- elif event_type in ('agent_text_sent', 'AgentTextSent'):
- content = getattr(item, 'content', '')
- if content and isinstance(content, str) and content not in seen:
- messages.append({"role": "assistant", "content": content})
- seen.add(content)
+ elif event_type in ("agent_text_sent", "AgentTextSent"):
+ append_message("assistant", getattr(item, "content", ""))
return messages
+ def _new_messages_from_sequence(
+ self,
+ current_messages: List[Dict[str, str]],
+ ) -> List[Dict[str, str]]:
+ """Return the append after the longest previous-suffix/current-prefix overlap."""
+ if current_messages and len(current_messages) <= len(self._history_cursor):
+ for start in range(len(self._history_cursor) - len(current_messages) + 1):
+ if self._history_cursor[start : start + len(current_messages)] == current_messages:
+ return []
+
+ overlap = 0
+ for size in range(min(len(self._history_cursor), len(current_messages)), 0, -1):
+ if self._history_cursor[-size:] == current_messages[:size]:
+ overlap = size
+ break
+
+ self._history_cursor = current_messages
+ return current_messages[overlap:]
+
+ def _new_history_messages(self, history: list) -> List[Dict[str, str]]:
+ """Return only messages appended to a cumulative or front-truncated history."""
+ return self._new_messages_from_sequence(self._extract_conversation_from_history(history))
+
async def _enrich_event_with_memories(self, event: Any) -> tuple[Any, Optional[str]]:
"""Enrich event by retrieving memories.
@@ -353,14 +373,16 @@ class SupermemoryCartesiaAgent:
logger.warning("[Supermemory] Could not extract user message from event")
return event, None
- if user_message == self._last_query:
+ event_id = _field(event, "event_id", "eventId")
+ event_marker = f"event:{event_id}" if event_id else f"object:{id(event)}"
+ if event_marker == self._last_retrieval_event:
return event, None
- self._last_query = user_message
logger.info(f"[Supermemory] Processing user message: {user_message[:50]}...")
try:
memories_data = await self._retrieve_memories(user_message)
+ self._last_retrieval_event = event_marker
memory_context = self._build_memory_message(memories_data)
if not memory_context:
@@ -377,6 +399,69 @@ class SupermemoryCartesiaAgent:
logger.error(f"[Supermemory] Error in memory enrichment: {e}")
return event, None
+ def _agent_accepts_context(self) -> bool:
+ """Return whether the wrapped agent supports per-call context."""
+ try:
+ parameters = inspect.signature(self.agent.process).parameters.values()
+ except (TypeError, ValueError):
+ return False
+
+ return any(
+ parameter.name == "context" or parameter.kind is inspect.Parameter.VAR_KEYWORD
+ for parameter in parameters
+ )
+
+ @staticmethod
+ def _without_memory_context(prompt: str) -> str:
+ """Remove context previously injected by this wrapper."""
+ return re.sub(
+ rf"{re.escape(MEMORY_TAG_START)}.*?{re.escape(MEMORY_TAG_END)}\s*",
+ "",
+ prompt,
+ flags=re.DOTALL,
+ ).strip()
+
+ async def _process_agent(
+ self,
+ env: Any,
+ event: Event,
+ memory_context: Optional[str],
+ ) -> AsyncGenerator[Event, None]:
+ """Call modern Line agents with per-turn context, with a legacy fallback."""
+ if self._agent_accepts_context():
+ process_kwargs = {"context": memory_context} if memory_context else {}
+ async for output in self.agent.process(env, event, **process_kwargs):
+ yield output
+ return
+
+ # Cartesia Line 0.2.0-0.2.2 exposed a mutable ``config`` property and
+ # did not yet support per-call context. Keep that narrow compatibility
+ # path while avoiding persistent memory text in the base prompt.
+ legacy_config = getattr(self.agent, "config", None)
+ if legacy_config is None:
+ if memory_context:
+ logger.warning(
+ "[Supermemory] Wrapped agent cannot accept memory context; "
+ "forwarding the event unchanged"
+ )
+ async for output in self.agent.process(env, event):
+ yield output
+ return
+
+ original_prompt = getattr(legacy_config, "system_prompt", "") or ""
+ clean_prompt = self._without_memory_context(str(original_prompt))
+ prompt_for_call = (
+ f"{memory_context}\n\n{clean_prompt}"
+ if memory_context and clean_prompt
+ else memory_context or clean_prompt
+ )
+ legacy_config.system_prompt = prompt_for_call
+ try:
+ async for output in self.agent.process(env, event):
+ yield output
+ finally:
+ legacy_config.system_prompt = clean_prompt
+
async def process(self, env: Any, event: Event) -> AsyncGenerator[Event, None]:
"""Process events with memory enrichment.
@@ -392,49 +477,31 @@ class SupermemoryCartesiaAgent:
logger.info("[Supermemory] Processing UserTurnEnded event")
event, memory_context = await self._enrich_event_with_memories(event)
- # Clean up old memory context and inject new one if available
- if hasattr(self.agent, 'config'):
- original_prompt = getattr(self.agent.config, 'system_prompt', '')
- # Always remove old memory context if present to prevent stale data
- if MEMORY_TAG_START in original_prompt:
- original_prompt = re.sub(
- rf'{re.escape(MEMORY_TAG_START)}.*?{re.escape(MEMORY_TAG_END)}\s*',
- '',
- original_prompt,
- flags=re.DOTALL
- )
- logger.debug("[Supermemory] Removed old memory context from system prompt")
-
- # Inject new memory context if available
- if memory_context:
- self.agent.config.system_prompt = f"{memory_context}\n\n{original_prompt}"
- logger.info("[Supermemory] Injected new memory context into system prompt")
- else:
- # No new memories, but we cleaned up old ones
- self.agent.config.system_prompt = original_prompt
- logger.debug("[Supermemory] No new memories to inject, using clean prompt")
-
# Store conversation in background
if hasattr(event, 'history') and event.history:
- messages = self._extract_conversation_from_history(event.history)
- unsent = messages[self._messages_sent_count:]
- if unsent:
- logger.info(f"[Supermemory] Queuing {len(unsent)} messages for storage")
- task = asyncio.create_task(self._store_messages(unsent))
+ new_messages = self._new_history_messages(event.history)
+ if new_messages:
+ logger.info(
+ f"[Supermemory] Queuing {len(new_messages)} messages for storage"
+ )
+ task = asyncio.create_task(self._store_messages(new_messages))
self._background_tasks.add(task)
task.add_done_callback(self._background_tasks.discard)
- self._messages_sent_count = len(messages)
else:
# No history yet, store just the current user message
- user_content = self._extract_user_message(event)
- if user_content:
- logger.info(f"[Supermemory] No history, storing current user message: {user_content[:50]}...")
- task = asyncio.create_task(self._store_messages([{"role": "user", "content": user_content}]))
+ current_messages = self._extract_conversation_from_history([event])
+ if not current_messages:
+ user_content = self._extract_user_message(event)
+ if user_content:
+ current_messages = [{"role": "user", "content": user_content}]
+ new_messages = self._new_messages_from_sequence(current_messages)
+ if new_messages:
+ logger.info("[Supermemory] No history, storing current user message")
+ task = asyncio.create_task(self._store_messages(new_messages))
self._background_tasks.add(task)
task.add_done_callback(self._background_tasks.discard)
- self._messages_sent_count = 1 # CRITICAL: Increment counter to prevent duplicate storage
- async for output in self.agent.process(env, event):
+ async for output in self._process_agent(env, event, memory_context):
yield output
else:
async for output in self.agent.process(env, event):
@@ -447,6 +514,6 @@ class SupermemoryCartesiaAgent:
def reset_memory_tracking(self) -> None:
"""Reset memory tracking for a new conversation."""
- self._messages_sent_count = 0
- self._last_query = None
+ self._history_cursor = []
+ self._last_retrieval_event = None
logger.info("[Supermemory] Reset memory tracking state")
diff --git a/packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py b/packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py
index eb366426..1e4a6e00 100644
--- a/packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py
+++ b/packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py
@@ -1,5 +1,6 @@
"""Utility functions for Supermemory Cartesia integration."""
+import re
from datetime import datetime, timezone
from typing import Any, Dict, List, Union
@@ -49,34 +50,74 @@ def format_relative_time(iso_timestamp: str) -> str:
return ""
+def _field(item: Any, *names: str, default: Any = None) -> Any:
+ """Read a field from a dict or pydantic/SDK model.
+
+ Accepts camelCase and snake_case names so helpers work with both raw JSON
+ dicts and Stainless-generated response models.
+ """
+ if item is None:
+ return default
+ if isinstance(item, dict):
+ for name in names:
+ if name in item and item[name] is not None:
+ return item[name]
+ return default
+ for name in names:
+ value = getattr(item, name, None)
+ if value is not None:
+ return value
+ return default
+
+
+_MEMORY_DATE_PREFIX = re.compile(
+ r"^\s*(?:\[recent\]\s*)?(?:\[\d{4}-\d{2}-\d{2}\]\s*)?",
+ re.IGNORECASE,
+)
+
+
+def _memory_key(memory: str) -> str:
+ """Normalize display-only profile prefixes for duplicate comparison."""
+ without_prefix = _MEMORY_DATE_PREFIX.sub("", memory)
+ return " ".join(without_prefix.split()).casefold()
+
+
def deduplicate_memories(
static: List[str],
dynamic: List[str],
- search_results: List[Dict[str, Any]],
-) -> Dict[str, Union[List[str], List[Dict[str, Any]]]]:
+ search_results: List[Any],
+) -> Dict[str, Union[List[str], List[Any]]]:
"""Deduplicate memories. Priority: static > dynamic > search.
Args:
static: List of static memory strings.
dynamic: List of dynamic memory strings.
- search_results: List of search result dicts with 'memory' and 'updatedAt'.
+ search_results: Search result dicts or pydantic models with a memory field.
"""
seen = set()
def unique_strings(memories: List[str]) -> List[str]:
out = []
for m in memories:
- if m not in seen:
- seen.add(m)
+ if not isinstance(m, str):
+ continue
+ key = _memory_key(m)
+ if key and key not in seen:
+ seen.add(key)
out.append(m)
return out
- def unique_search(results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+ def unique_search(results: List[Any]) -> List[Any]:
out = []
for r in results:
- memory = r.get("memory", "")
- if memory and memory not in seen:
- seen.add(memory)
+ # v4 search.memories/hybrid uses `memory` or `chunk`.
+ memory = _field(r, "memory", "chunk", "content", default="")
+ if not isinstance(memory, str):
+ memory = ""
+ memory = memory.strip()
+ key = _memory_key(memory)
+ if key and key not in seen:
+ seen.add(key)
out.append(r)
return out
@@ -88,7 +129,7 @@ def deduplicate_memories(
def format_memories_to_text(
- memories: Dict[str, Union[List[str], List[Dict[str, Any]]]],
+ memories: Dict[str, Union[List[str], List[Any]]],
system_prompt: str = "Based on previous conversations, I recall:\n\n",
include_static: bool = True,
include_dynamic: bool = True,
@@ -116,16 +157,17 @@ def format_memories_to_text(
sections.append("## Relevant Memories")
lines = []
for item in search_results:
- if isinstance(item, dict):
- memory = item.get("memory", "")
- updated_at = item.get("updatedAt", "")
- time_str = format_relative_time(updated_at) if updated_at else ""
- if time_str:
- lines.append(f"- [{time_str}] {memory}")
- else:
- lines.append(f"- {memory}")
- else:
+ 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 updated_at else ""
+ if time_str:
+ lines.append(f"- [{time_str}] {memory}")
+ else:
+ lines.append(f"- {memory}")
sections.append("\n".join(lines))
if not sections:
diff --git a/packages/openai-sdk-python/README.md b/packages/openai-sdk-python/README.md
index 9b455b3b..cc13c9c5 100644
--- a/packages/openai-sdk-python/README.md
+++ b/packages/openai-sdk-python/README.md
@@ -49,7 +49,9 @@ async def main():
custom_id="chat-123", # Required: groups messages into documents
mode="full", # "profile", "query", or "full"
verbose=True, # Enable logging
- add_memory="always" # Automatically save conversations (default)
+ add_memory="always", # Automatically save conversations (default)
+ api_key="your-supermemory-api-key", # Or use SUPERMEMORY_API_KEY
+ # base_url="https://api.supermemory.ai", # Optional custom endpoint
)
)
@@ -357,6 +359,8 @@ class OpenAIMiddlewareOptions:
verbose: bool = False # Enable detailed logging
mode: Literal["profile", "query", "full"] = "profile" # Memory injection mode
add_memory: Literal["always", "never"] = "always" # Auto-save behavior
+ api_key: Optional[str] = None # Falls back to SUPERMEMORY_API_KEY
+ base_url: Optional[str] = None # Falls back to SUPERMEMORY_BASE_URL
```
### SupermemoryTools
@@ -436,7 +440,7 @@ All exceptions include the original error for debugging and have descriptive err
Set these environment variables:
-- `SUPERMEMORY_API_KEY` - Your Supermemory API key (required)
+- `SUPERMEMORY_API_KEY` - Your Supermemory API key (unless passed in middleware options)
- `OPENAI_API_KEY` - Your OpenAI API key (required for examples)
Optional for testing:
diff --git a/packages/openai-sdk-python/src/supermemory_openai/middleware.py b/packages/openai-sdk-python/src/supermemory_openai/middleware.py
index 9cbad1b0..5191a6c8 100644
--- a/packages/openai-sdk-python/src/supermemory_openai/middleware.py
+++ b/packages/openai-sdk-python/src/supermemory_openai/middleware.py
@@ -28,6 +28,9 @@ from .utils import (
get_last_user_message,
)
+DEFAULT_SUPERMEMORY_BASE_URL = "https://api.supermemory.ai"
+PROFILE_REQUEST_TIMEOUT_SECONDS = 30.0
+
@dataclass
class OpenAIMiddlewareOptions:
@@ -38,6 +41,8 @@ class OpenAIMiddlewareOptions:
verbose: bool = False
mode: Literal["profile", "query", "full"] = "profile"
add_memory: Literal["always", "never"] = "always"
+ api_key: Optional[str] = None
+ base_url: Optional[str] = None
class SupermemoryProfileSearch:
@@ -52,6 +57,7 @@ async def supermemory_profile_search(
container_tag: str,
query_text: str,
api_key: str,
+ base_url: str,
) -> SupermemoryProfileSearch:
"""Search for memories using the SuperMemory profile API."""
payload = {
@@ -59,20 +65,23 @@ async def supermemory_profile_search(
}
if query_text:
payload["q"] = query_text
+ profile_url = f"{base_url.rstrip('/')}/v4/profile"
try:
import aiohttp
- async with aiohttp.ClientSession() as session:
+ timeout = aiohttp.ClientTimeout(total=PROFILE_REQUEST_TIMEOUT_SECONDS)
+ async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.post(
- "https://api.supermemory.ai/v4/profile",
+ profile_url,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
},
json=payload,
+ allow_redirects=False,
) as response:
- if not response.ok:
+ if not 200 <= response.status < 300:
error_text = await response.text()
raise SupermemoryAPIError(
"Supermemory profile search failed",
@@ -88,15 +97,17 @@ async def supermemory_profile_search(
import requests
response = requests.post(
- "https://api.supermemory.ai/v4/profile",
+ profile_url,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
},
json=payload,
+ timeout=PROFILE_REQUEST_TIMEOUT_SECONDS,
+ allow_redirects=False,
)
- if not response.ok:
+ if not 200 <= response.status_code < 300:
raise SupermemoryAPIError(
"Supermemory profile search failed",
status_code=response.status_code,
@@ -112,6 +123,7 @@ async def add_system_prompt(
logger: Logger,
mode: Literal["profile", "query", "full"],
api_key: str,
+ base_url: str,
) -> list[ChatCompletionMessageParam]:
"""Add memory-enhanced system prompts to chat completion messages."""
system_prompt_exists = any(msg.get("role") == "system" for msg in messages)
@@ -119,7 +131,10 @@ async def add_system_prompt(
query_text = get_last_user_message(messages) if mode != "profile" else ""
memories_response = await supermemory_profile_search(
- container_tag, query_text, api_key
+ container_tag,
+ query_text,
+ api_key,
+ base_url,
)
profile = memories_response.profile or {}
@@ -199,9 +214,11 @@ async def add_system_prompt(
if system_prompt_exists:
logger.debug("Added memories to existing system prompt")
return [
- {**msg, "content": f"{msg.get('content', '')} \n {memories}"}
- if msg.get("role") == "system"
- else msg
+ (
+ {**msg, "content": f"{msg.get('content', '')} \n {memories}"}
+ if msg.get("role") == "system"
+ else msg
+ )
for msg in messages
]
@@ -222,15 +239,17 @@ async def add_memory_tool(
) -> None:
"""Add a new memory to the SuperMemory system."""
try:
- # Handle both sync and async supermemory clients
- if custom_id is None:
- result = client.add(content=content, container_tag=container_tag)
+ kwargs = {"content": content, "container_tag": container_tag}
+ if custom_id is not None:
+ kwargs["custom_id"] = custom_id
+
+ # The wrapper currently constructs the synchronous Supermemory client for
+ # both OpenAI variants. Never execute that network call on an async event
+ # loop; mocks or future async clients can still return an awaitable.
+ if inspect.iscoroutinefunction(client.add):
+ result = client.add(**kwargs)
else:
- result = client.add(
- content=content,
- container_tag=container_tag,
- custom_id=custom_id,
- )
+ result = await asyncio.to_thread(client.add, **kwargs)
if inspect.isawaitable(result):
response = await result
else:
@@ -273,6 +292,8 @@ class SupermemoryOpenAIWrapper:
self._container_tag: str = options.container_tag
self._options: OpenAIMiddlewareOptions = options
self._logger: Logger = create_logger(self._options.verbose)
+ self._api_key = self._resolve_api_key(options.api_key)
+ self._base_url = self._resolve_base_url(options.base_url)
# Track background tasks to ensure they complete
self._background_tasks: set[asyncio.Task] = set()
@@ -283,10 +304,10 @@ class SupermemoryOpenAIWrapper:
ImportError("supermemory package not installed"),
)
- api_key = self._get_api_key()
try:
self._supermemory_client: supermemory.Supermemory = supermemory.Supermemory(
- api_key=api_key
+ api_key=self._api_key,
+ base_url=self._base_url,
)
except Exception as e:
raise SupermemoryConfigurationError(
@@ -296,16 +317,28 @@ class SupermemoryOpenAIWrapper:
# Wrap the chat completions create method
self._wrap_chat_completions()
- def _get_api_key(self) -> str:
- """Get Supermemory API key from environment."""
- import os
-
- api_key = os.getenv("SUPERMEMORY_API_KEY")
+ @staticmethod
+ def _resolve_api_key(configured_api_key: Optional[str]) -> str:
+ """Resolve the API key once when the middleware is constructed."""
+ api_key = (configured_api_key or "").strip() or (
+ os.getenv("SUPERMEMORY_API_KEY") or ""
+ ).strip()
if not api_key:
raise SupermemoryConfigurationError(
- "SUPERMEMORY_API_KEY environment variable is required but not set"
+ "A Supermemory API key is required. Pass api_key to "
+ "OpenAIMiddlewareOptions or set SUPERMEMORY_API_KEY."
)
- return api_key
+ return api_key.strip()
+
+ @staticmethod
+ def _resolve_base_url(configured_base_url: Optional[str]) -> str:
+ """Resolve and normalize the API base URL once."""
+ base_url = (
+ (configured_base_url or "").strip()
+ or (os.getenv("SUPERMEMORY_BASE_URL") or "").strip()
+ or DEFAULT_SUPERMEMORY_BASE_URL
+ )
+ return base_url.rstrip("/")
def _wrap_chat_completions(self) -> None:
"""Wrap the chat completions create method with memory injection."""
@@ -317,6 +350,7 @@ class SupermemoryOpenAIWrapper:
**kwargs: Any,
) -> Any:
return await self._create_with_memory_async(original_create, **kwargs)
+
else:
def create_with_memory(
@@ -413,7 +447,8 @@ class SupermemoryOpenAIWrapper:
self._container_tag,
self._logger,
self._options.mode,
- self._get_api_key(),
+ self._api_key,
+ self._base_url,
)
kwargs["messages"] = enhanced_messages
@@ -500,7 +535,8 @@ class SupermemoryOpenAIWrapper:
self._container_tag,
self._logger,
self._options.mode,
- self._get_api_key(),
+ self._api_key,
+ self._base_url,
)
)
except RuntimeError as e:
@@ -516,7 +552,8 @@ class SupermemoryOpenAIWrapper:
self._container_tag,
self._logger,
self._options.mode,
- self._get_api_key(),
+ self._api_key,
+ self._base_url,
),
)
enhanced_messages = future.result()
@@ -558,7 +595,9 @@ class SupermemoryOpenAIWrapper:
f"Background tasks did not complete within {timeout}s timeout"
)
# Cancel remaining tasks
- tasks_to_cancel = [task for task in self._background_tasks if not task.done()]
+ tasks_to_cancel = [
+ task for task in self._background_tasks if not task.done()
+ ]
for task in tasks_to_cancel:
task.cancel()
diff --git a/packages/pipecat-sdk-python/README.md b/packages/pipecat-sdk-python/README.md
index bb8e26e1..f885ed71 100644
--- a/packages/pipecat-sdk-python/README.md
+++ b/packages/pipecat-sdk-python/README.md
@@ -13,7 +13,7 @@ pip install supermemory-pipecat
```python
import os
from pipecat.pipeline.pipeline import Pipeline
-from pipecat.services.openai import OpenAILLMService, OpenAIUserContextAggregator
+from pipecat.processors.aggregators.llm_context import LLMContext
from supermemory_pipecat import SupermemoryPipecatService
# Create memory service
@@ -23,14 +23,19 @@ memory = SupermemoryPipecatService(
session_id="conversation-456", # Optional: groups memories by session
)
+# Use the universal LLM context supported by current Pipecat releases.
+context = LLMContext([{"role": "system", "content": "You are a helpful assistant."}])
+context_aggregator = llm.create_context_aggregator(context)
+
# Create pipeline with memory
pipeline = Pipeline([
transport.input(),
stt,
- user_context,
+ context_aggregator.user(),
memory, # Automatically retrieves and injects relevant memories
llm,
transport.output(),
+ context_aggregator.assistant(),
])
```
@@ -58,6 +63,7 @@ memory = SupermemoryPipecatService(
search_limit=10, # Max memories to retrieve
search_threshold=0.1, # Similarity threshold
mode="full", # "profile", "query", or "full"
+ inject_mode="auto", # "auto", "system", or "user"
system_prompt="Based on previous conversations, I recall:\n\n",
),
)
@@ -73,25 +79,30 @@ memory = SupermemoryPipecatService(
## How It Works
-1. **Intercepts context frames** - Listens for `LLMContextFrame` in the pipeline
-2. **Tracks conversation** - Maintains clean conversation history (no injected memories)
+1. **Intercepts context frames** - Listens for Pipecat's universal `LLMContextFrame` (and legacy 0.x frames)
+2. **Tracks conversation** - Separates real conversation messages from tagged memory context
3. **Retrieves memories** - Queries `/v4/profile` API with user's message
-4. **Injects memories** - Formats and adds to LLM context as system message
-5. **Stores messages** - Sends last user message to Supermemory (background, non-blocking)
+4. **Injects memories** - Uses a system message for audio/system mode and a tagged user message otherwise
+5. **Stores messages** - Serializes newly observed user and assistant messages through a background queue that drains during cleanup
### What Gets Stored
-Only the last user message is sent to Supermemory:
+New user and assistant messages are stored as a JSON conversation segment. The
+injected `` message is filtered out before storage and does not
+advance the storage cursor.
+
+For example, this conversation segment:
```
User: What's the weather like today?
+Assistant: It's sunny today.
```
-Stored as:
+is sent to Supermemory as:
```json
{
- "content": "User: What's the weather like today?",
+ "content": "[{\"role\": \"user\", \"content\": \"What's the weather like today?\"}, {\"role\": \"assistant\", \"content\": \"It's sunny today.\"}]",
"container_tags": ["user-123"],
"custom_id": "conversation-456",
"metadata": { "platform": "pipecat" }
@@ -107,7 +118,7 @@ from fastapi import FastAPI, WebSocket
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.task import PipelineTask
from pipecat.pipeline.runner import PipelineRunner
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService
from pipecat.transports.websocket.fastapi import (
FastAPIWebsocketTransport,
@@ -132,7 +143,7 @@ async def websocket_endpoint(websocket: WebSocket):
model="models/gemini-2.5-flash-native-audio-preview-12-2025",
)
- context = OpenAILLMContext([{"role": "system", "content": "You are a helpful assistant."}])
+ context = LLMContext([{"role": "system", "content": "You are a helpful assistant."}])
context_aggregator = llm.create_context_aggregator(context)
# Supermemory memory service
diff --git a/packages/pipecat-sdk-python/pyproject.toml b/packages/pipecat-sdk-python/pyproject.toml
index 1c25b6a2..5f1f405e 100644
--- a/packages/pipecat-sdk-python/pyproject.toml
+++ b/packages/pipecat-sdk-python/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "supermemory-pipecat"
-version = "0.1.1"
+version = "0.1.2"
description = "Supermemory integration for Pipecat - memory-enhanced conversational AI pipelines"
readme = "README.md"
license = "MIT"
@@ -31,7 +31,7 @@ classifiers = [
"Topic :: Scientific/Engineering :: Artificial Intelligence",
]
dependencies = [
- "pipecat-ai>=0.0.98",
+ "pipecat-ai>=0.0.98,<2.0.0",
"supermemory>=3.16.0",
"pydantic>=2.10.0",
"loguru>=0.7.3",
diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/__init__.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/__init__.py
index aeb190c9..234bfa7e 100644
--- a/packages/pipecat-sdk-python/src/supermemory_pipecat/__init__.py
+++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/__init__.py
@@ -25,6 +25,8 @@ Example:
```
"""
+from importlib.metadata import PackageNotFoundError, version
+
from .exceptions import (
APIError,
ConfigurationError,
@@ -40,7 +42,11 @@ from .utils import (
get_last_user_message,
)
-__version__ = "0.1.1"
+try:
+ __version__ = version("supermemory-pipecat")
+except PackageNotFoundError:
+ # Source-tree fallback; built wheels always use package metadata above.
+ __version__ = "0.1.2"
__all__ = [
# Main service
diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py
index eb9d5fb6..d3b000b5 100644
--- a/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py
+++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/service.py
@@ -6,22 +6,37 @@ historical information.
"""
import asyncio
+import copy
import json
import os
import re
+from collections import deque
from typing import Any, Dict, List, Literal, Optional
from loguru import logger
from pydantic import BaseModel, Field
-from pipecat.frames.frames import Frame, InputAudioRawFrame, LLMContextFrame, LLMMessagesFrame
+from pipecat.frames.frames import Frame, InputAudioRawFrame, LLMContextFrame
from pipecat.processors.aggregators.llm_context import LLMContext
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pydantic import BaseModel, Field
-from .exceptions import ConfigurationError, MemoryRetrievalError
-from .utils import deduplicate_memories, format_memories_to_text, get_last_user_message
+from .exceptions import ConfigurationError, MemoryRetrievalError, MemoryStorageError
+from .utils import _field, deduplicate_memories, format_memories_to_text
+
+# Pipecat 1.0 removed the legacy message and OpenAI-specific context frames.
+# Keep them optional so the integration supports both the declared 0.0.98
+# minimum and the universal LLMContextFrame used by current Pipecat releases.
+try:
+ from pipecat.frames.frames import LLMMessagesFrame as _LegacyLLMMessagesFrame
+except ImportError: # Pipecat >= 1.0
+ _LegacyLLMMessagesFrame = None # type: ignore[assignment]
+
+try:
+ from pipecat.processors.aggregators.openai_llm_context import (
+ OpenAILLMContextFrame as _LegacyOpenAILLMContextFrame,
+ )
+except (ImportError, ModuleNotFoundError): # Pipecat >= 1.0
+ _LegacyOpenAILLMContextFrame = None # type: ignore[assignment]
try:
import supermemory
@@ -34,6 +49,72 @@ MEMORY_TAG_END = ""
MEMORY_TAG_PATTERN = re.compile(r".*?", re.DOTALL)
+def _is_legacy_openai_context_frame(frame: Frame) -> bool:
+ return _LegacyOpenAILLMContextFrame is not None and isinstance(
+ frame, _LegacyOpenAILLMContextFrame
+ )
+
+
+def _is_legacy_messages_frame(frame: Frame) -> bool:
+ return _LegacyLLMMessagesFrame is not None and isinstance(frame, _LegacyLLMMessagesFrame)
+
+
+def _snapshot_storable_messages(messages: List[Any]) -> List[Dict[str, Any]]:
+ """Copy real conversation messages, excluding injected memory context."""
+ storable: List[Dict[str, Any]] = []
+ for message in messages:
+ if not isinstance(message, dict) or message.get("role") not in ("user", "assistant"):
+ continue
+
+ if _is_injected_user_memory(message):
+ continue
+
+ storable.append(copy.deepcopy(message))
+
+ return storable
+
+
+def _is_injected_user_memory(message: Any) -> bool:
+ """Return whether a message is the wrapper's standalone memory message."""
+ return _is_standalone_memory_message(message, roles=("user",))
+
+
+def _is_standalone_memory_message(message: Any, *, roles: tuple[str, ...]) -> bool:
+ if not isinstance(message, dict) or message.get("role") not in roles:
+ return False
+ content = message.get("content")
+ return isinstance(content, str) and MEMORY_TAG_PATTERN.fullmatch(content.strip()) is not None
+
+
+def _messages_after_overlap(
+ previous: List[Dict[str, Any]],
+ current: List[Dict[str, Any]],
+) -> List[Dict[str, Any]]:
+ """Return messages after the largest previous-suffix/current-prefix overlap.
+
+ Comparing ordered occurrences instead of only list lengths handles appended,
+ replaced, and front-truncated contexts. Identical repeated messages remain
+ distinct because overlap is positional rather than set-based.
+ """
+ max_overlap = min(len(previous), len(current))
+ for overlap in range(max_overlap, 0, -1):
+ if previous[-overlap:] == current[:overlap]:
+ return copy.deepcopy(current[overlap:])
+ return copy.deepcopy(current)
+
+
+def _latest_user_occurrence(
+ messages: List[Dict[str, Any]],
+) -> tuple[Optional[str], Optional[List[Dict[str, Any]]]]:
+ """Return the last text user query and its ordered conversation prefix."""
+ for index in range(len(messages) - 1, -1, -1):
+ message = messages[index]
+ content = message.get("content")
+ if message.get("role") == "user" and isinstance(content, str):
+ return content, copy.deepcopy(messages[: index + 1])
+ return None, None
+
+
class SupermemoryPipecatService(FrameProcessor):
"""Memory service that integrates Supermemory with Pipecat pipelines.
@@ -115,9 +196,12 @@ class SupermemoryPipecatService(FrameProcessor):
except Exception as e:
logger.warning(f"Failed to initialize Supermemory client: {e}")
- self._messages_sent_count: int = 0
self._last_query: Optional[str] = None
+ self._last_recalled_user_prefix: Optional[List[Dict[str, Any]]] = None
self._audio_frames_detected: bool = False
+ self._latest_storable_context: List[Dict[str, Any]] = []
+ self._storage_queue: deque[List[Dict[str, Any]]] = deque()
+ self._storage_worker_task: Optional[asyncio.Task[None]] = None
async def _retrieve_memories(self, query: str) -> Dict[str, Any]:
"""Retrieve relevant memories from Supermemory.
@@ -137,26 +221,25 @@ class SupermemoryPipecatService(FrameProcessor):
)
try:
+ # One profile call: static + dynamic, and (when mode/query allow)
+ # search_results via `q`. This is the intended profile API shape.
kwargs: Dict[str, Any] = {"container_tag": self.container_tag}
-
if self.params.mode != "profile" and query:
kwargs["q"] = query
kwargs["threshold"] = self.params.search_threshold
- kwargs["extra_body"] = {"limit": self.params.search_limit}
response = await self._supermemory_client.profile(**kwargs)
- profile = getattr(response, "profile", None)
- search_results_response = getattr(response, "search_results", None)
+ profile = _field(response, "profile")
+ search_results_response = _field(response, "search_results", "searchResults")
- search_results = []
- if search_results_response and search_results_response.results:
- search_results = search_results_response.results
+ raw_search_results = _field(search_results_response, "results", default=[]) or []
+ search_results = list(raw_search_results)[: self.params.search_limit]
return {
"profile": {
- "static": profile.static if profile is not None else [],
- "dynamic": profile.dynamic if profile is not None else [],
+ "static": list(_field(profile, "static", default=[]) or []),
+ "dynamic": list(_field(profile, "dynamic", default=[]) or []),
},
"search_results": search_results,
}
@@ -166,10 +249,13 @@ class SupermemoryPipecatService(FrameProcessor):
raise MemoryRetrievalError("Failed to retrieve memories", e)
async def _store_messages(self, messages: List[Dict[str, Any]]) -> None:
- """Store messages in Supermemory (non-blocking, fire-and-forget)."""
- if self._supermemory_client is None or not messages:
+ """Store one ordered message batch in Supermemory."""
+ if not messages:
return
+ if self._supermemory_client is None:
+ raise MemoryStorageError("Supermemory client is not initialized")
+
try:
add_params: Dict[str, Any] = {
"content": json.dumps(messages),
@@ -179,10 +265,82 @@ class SupermemoryPipecatService(FrameProcessor):
if self.session_id:
add_params["custom_id"] = self.session_id
- await self._supermemory_client.memories.add(**add_params)
+ await self._supermemory_client.add(**add_params)
except Exception as e:
- logger.error(f"Error storing messages: {e}")
+ raise MemoryStorageError("Failed to store messages", e) from e
+
+ def _queue_context_for_storage(self, messages: List[Dict[str, Any]]) -> None:
+ """Queue only newly observed message occurrences, preserving order."""
+ current = copy.deepcopy(messages)
+ new_messages = _messages_after_overlap(self._latest_storable_context, current)
+ self._latest_storable_context = current
+
+ if new_messages:
+ self._storage_queue.append(new_messages)
+
+ # A new frame also retries a previously failed head batch.
+ self._start_storage_worker()
+
+ def _start_storage_worker(self) -> None:
+ """Start the single serial storage worker when work is pending."""
+ if not self._storage_queue:
+ return
+ if self._storage_worker_task is not None and not self._storage_worker_task.done():
+ return
+ self._storage_worker_task = asyncio.create_task(self._run_storage_queue())
+
+ async def _run_storage_queue(self) -> None:
+ """Write queued batches serially, retaining the head batch on failure."""
+ while self._storage_queue:
+ messages = self._storage_queue[0]
+ try:
+ await self._store_messages(messages)
+ except Exception as e:
+ logger.error(f"Error storing messages; batch remains queued for retry: {e}")
+ return
+ self._storage_queue.popleft()
+
+ async def _drain_storage_queue(self) -> None:
+ """Wait for queued writes and retry a failed head batch once at teardown."""
+ task = self._storage_worker_task
+ if task is not None and not task.done():
+ await task
+
+ if self._storage_queue:
+ self._start_storage_worker()
+ retry_task = self._storage_worker_task
+ if retry_task is not None:
+ await retry_task
+
+ if self._storage_queue:
+ logger.error(
+ f"Unable to drain {len(self._storage_queue)} Supermemory storage batch(es)"
+ )
+
+ @staticmethod
+ def _clear_injected_memories(context: LLMContext) -> None:
+ """Remove memory tags owned by this wrapper while preserving other entries."""
+ messages = context.get_messages()
+ cleaned_messages: List[Any] = []
+
+ for message in messages:
+ if _is_standalone_memory_message(message, roles=("system", "user")):
+ continue
+ if not isinstance(message, dict):
+ cleaned_messages.append(message)
+ continue
+
+ role = message.get("role")
+ content = message.get("content")
+ if role in ("system", "user") and isinstance(content, str):
+ cleaned_content = MEMORY_TAG_PATTERN.sub("", content)
+ if cleaned_content != content:
+ message["content"] = cleaned_content.strip()
+
+ cleaned_messages.append(message)
+
+ messages[:] = cleaned_messages
def _enhance_context_with_memories(
self,
@@ -200,11 +358,6 @@ class SupermemoryPipecatService(FrameProcessor):
query: The query used for retrieval.
memories_data: Memory data from Supermemory API.
"""
- if self._last_query == query:
- return
-
- self._last_query = query
-
profile = memories_data["profile"]
deduplicated = deduplicate_memories(
static=profile["static"],
@@ -246,7 +399,11 @@ class SupermemoryPipecatService(FrameProcessor):
if inject_to_system:
system_idx = None
for i, msg in enumerate(messages):
- if msg.get("role") == "system":
+ if (
+ isinstance(msg, dict)
+ and msg.get("role") == "system"
+ and isinstance(msg.get("content"), str)
+ ):
system_idx = i
break
@@ -264,7 +421,7 @@ class SupermemoryPipecatService(FrameProcessor):
# Remove previous memory message if exists
for i in range(len(messages) - 1, -1, -1):
msg = messages[i]
- if msg.get("role") == "user" and MEMORY_TAG_START in msg.get("content", ""):
+ if _is_injected_user_memory(msg):
messages.pop(i)
break
@@ -282,51 +439,66 @@ class SupermemoryPipecatService(FrameProcessor):
return
context = None
- messages = None
- if isinstance(frame, (LLMContextFrame, OpenAILLMContextFrame)):
+ legacy_messages_frame = False
+
+ if isinstance(frame, LLMContextFrame) or _is_legacy_openai_context_frame(frame):
context = frame.context
- elif isinstance(frame, LLMMessagesFrame):
- messages = frame.messages
- context = LLMContext(messages)
+ elif _is_legacy_messages_frame(frame):
+ legacy_messages_frame = True
+ context = LLMContext(frame.messages)
- if context:
+ if context is not None:
try:
context_messages = context.get_messages()
- latest_user_message = get_last_user_message(context_messages)
+ # Snapshot the real conversation before adding memory context.
+ # Injected messages must never be persisted or
+ # included in the sent-message cursor.
+ storable_messages = _snapshot_storable_messages(context_messages)
+ latest_user_message, user_prefix = _latest_user_occurrence(storable_messages)
- if latest_user_message:
+ if (
+ latest_user_message
+ and user_prefix is not None
+ and user_prefix != self._last_recalled_user_prefix
+ ):
+ # Clear stale recall before a new lookup. If retrieval
+ # fails or returns no memories, old context cannot leak
+ # into the new turn.
+ self._clear_injected_memories(context)
try:
memories_data = await self._retrieve_memories(latest_user_message)
self._enhance_context_with_memories(
context, latest_user_message, memories_data
)
+ # Mark only successful recalls (including empty ones).
+ # Failures stay retryable on the next repeated frame.
+ self._last_query = latest_user_message
+ self._last_recalled_user_prefix = user_prefix
except MemoryRetrievalError as e:
logger.warning(f"Memory retrieval failed: {e}")
- # Store unsent messages (user and assistant only)
- storable_messages = [
- msg for msg in context_messages if msg["role"] in ("user", "assistant")
- ]
- unsent_messages = storable_messages[self._messages_sent_count :]
+ self._queue_context_for_storage(storable_messages)
- if unsent_messages:
- asyncio.create_task(self._store_messages(unsent_messages))
- self._messages_sent_count = len(storable_messages)
-
- if messages is not None:
- await self.push_frame(LLMMessagesFrame(context.get_messages()))
+ if legacy_messages_frame:
+ await self.push_frame(frame.__class__(context.get_messages()), direction)
else:
- await self.push_frame(frame)
+ await self.push_frame(frame, direction)
except Exception as e:
logger.error(f"Error processing frame: {e}")
- await self.push_frame(frame)
+ await self.push_frame(frame, direction)
else:
await self.push_frame(frame, direction)
+ async def cleanup(self) -> None:
+ """Drain pending Supermemory writes before Pipecat tears down the processor."""
+ await self._drain_storage_queue()
+ await super().cleanup()
+
def reset_memory_tracking(self) -> None:
"""Reset memory tracking state for a new conversation."""
- self._messages_sent_count = 0
+ self._latest_storable_context = []
self._last_query = None
+ self._last_recalled_user_prefix = None
self._audio_frames_detected = False
diff --git a/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py b/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py
index a27da256..743bb4c0 100644
--- a/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py
+++ b/packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py
@@ -1,14 +1,19 @@
"""Utility functions for Supermemory Pipecat integration."""
+import re
from datetime import datetime, timezone
from typing import Any, Dict, List, Union
-def get_last_user_message(messages: List[Dict[str, str]]) -> str | None:
+_DYNAMIC_DATE_PREFIX = re.compile(r"^\s*(?:\[Recent\]\s*)?\[\d{4}-\d{2}-\d{2}\]\s*")
+
+
+def get_last_user_message(messages: List[Dict[str, Any]]) -> str | None:
"""Extract the last user message content from a list of messages."""
for msg in reversed(messages):
- if msg["role"] == "user":
- return msg["content"]
+ content = msg.get("content")
+ if msg.get("role") == "user" and isinstance(content, str):
+ return content
return None
@@ -49,34 +54,69 @@ def format_relative_time(iso_timestamp: str) -> str:
return ""
+def _field(item: Any, *names: str, default: Any = None) -> Any:
+ """Read a field from a dict or pydantic/SDK model.
+
+ Accepts camelCase and snake_case names so helpers work with both raw JSON
+ dicts and Stainless-generated response models.
+ """
+ if item is None:
+ return default
+ if isinstance(item, dict):
+ for name in names:
+ if name in item and item[name] is not None:
+ return item[name]
+ return default
+ for name in names:
+ value = getattr(item, name, None)
+ if value is not None:
+ return value
+ return default
+
+
def deduplicate_memories(
static: List[str],
dynamic: List[str],
- search_results: List[Dict[str, Any]],
-) -> Dict[str, Union[List[str], List[Dict[str, Any]]]]:
+ search_results: List[Any],
+) -> Dict[str, Union[List[str], List[Any]]]:
"""Deduplicate memories. Priority: static > dynamic > search.
Args:
static: List of static memory strings.
dynamic: List of dynamic memory strings.
- search_results: List of search result dicts with 'memory' and 'updatedAt'.
+ search_results: Search result dicts or pydantic models with a memory field.
"""
- seen = set()
+ seen: set[str] = set()
+
+ def comparison_key(memory: str) -> str:
+ # Dynamic profile entries are date-labelled by the API while search
+ # results contain the same memory without that presentation prefix.
+ return _DYNAMIC_DATE_PREFIX.sub("", memory.strip())
def unique_strings(memories: List[str]) -> List[str]:
- out = []
+ out: List[str] = []
for m in memories:
- if m not in seen:
- seen.add(m)
+ if not isinstance(m, str):
+ continue
+ key = comparison_key(m)
+ if key and key not in seen:
+ seen.add(key)
out.append(m)
return out
- def unique_search(results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
- out = []
+ def unique_search(results: List[Any]) -> List[Any]:
+ out: List[Any] = []
for r in results:
- memory = r.get("memory", "")
- if memory and memory not in seen:
- seen.add(memory)
+ # v4 search.memories/hybrid uses `memory` or `chunk`.
+ memory = (
+ r if isinstance(r, str) else _field(r, "memory", "chunk", "content", default="")
+ )
+ if not isinstance(memory, str):
+ memory = ""
+ memory = memory.strip()
+ key = comparison_key(memory)
+ if key and key not in seen:
+ seen.add(key)
out.append(r)
return out
@@ -88,7 +128,7 @@ def deduplicate_memories(
def format_memories_to_text(
- memories: Dict[str, Union[List[str], List[Dict[str, Any]]]],
+ memories: Dict[str, Union[List[str], List[Any]]],
system_prompt: str = "Based on previous conversations, I recall:\n\n",
include_static: bool = True,
include_dynamic: bool = True,
@@ -116,16 +156,17 @@ def format_memories_to_text(
sections.append("## Relevant Memories")
lines = []
for item in search_results:
- if isinstance(item, dict):
- memory = item.get("memory", "")
- updated_at = item.get("updatedAt", "")
- time_str = format_relative_time(updated_at) if updated_at else ""
- if time_str:
- lines.append(f"- [{time_str}] {memory}")
- else:
- lines.append(f"- {memory}")
- else:
+ 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 updated_at else ""
+ if time_str:
+ lines.append(f"- [{time_str}] {memory}")
+ else:
+ lines.append(f"- {memory}")
sections.append("\n".join(lines))
if not sections: