scriptoriumadmin/supermemory
1
0
Fork 0
mirror of https://github.com/supermemoryai/supermemory.git synced 2026-09-05 08:06:19 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

fix(python-sdks): v4 API migration for integration packages (#1434)

Browse source 
## Summary
- **agent-framework**: proactive search tool descriptions
- **cartesia / pipecat**: v4 `client.add` + hybrid search, dedupe fixes, tests

Stacked on #1433

## Test plan
- [ ] pytest in agent-framework, cartesia, pipecat packages

Made with [Cursor](https://cursor.com)
This commit is contained in:
Dhravya 2026-09-01 06:10:35 +00:00
parent 879ddd5c95
commit c262cc9953
No known key found for this signature in database
GPG key ID: 135A27003CF4F6CB
22 changed files with 853 additions and 375 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

86
.github/workflows/ci-python.yml vendored
View file

@ -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)"

10
.github/workflows/publish-agent-framework-python.yml vendored
View file

@ -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/

10
.github/workflows/publish-cartesia-sdk-python.yml vendored
View file

@ -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/

10
.github/workflows/publish-pipecat-sdk-python.yml vendored
View file

@ -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/

116
packages/agent-framework-python/README.md
View file

@ -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

4
packages/agent-framework-python/pyproject.toml
View file

@ -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",
]

4
packages/agent-framework-python/src/supermemory_agent_framework/context_provider.py
View file

@ -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

93
packages/agent-framework-python/src/supermemory_agent_framework/tools.py
View file

@ -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,
)

37
packages/agent-framework-python/src/supermemory_agent_framework/utils.py
View file

@ -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,

8
packages/cartesia-sdk-python/README.md
View file

@ -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.",

4
packages/cartesia-sdk-python/pyproject.toml
View file

@ -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",
]

2
packages/cartesia-sdk-python/requirements.txt
View file

@ -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

12
packages/cartesia-sdk-python/src/supermemory_cartesia/__init__.py
View file

@ -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

241
packages/cartesia-sdk-python/src/supermemory_cartesia/agent.py
View file

@ -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")

80
packages/cartesia-sdk-python/src/supermemory_cartesia/utils.py
View file

@ -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:

8
packages/openai-sdk-python/README.md
View file

@ -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:

99
packages/openai-sdk-python/src/supermemory_openai/middleware.py
View file

@ -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()

33
packages/pipecat-sdk-python/README.md
View file

@ -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 `<user_memories>` 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

4
packages/pipecat-sdk-python/pyproject.toml
View file

@ -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",

8
packages/pipecat-sdk-python/src/supermemory_pipecat/__init__.py
View file

@ -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

268
packages/pipecat-sdk-python/src/supermemory_pipecat/service.py
View file

@ -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 = "</user_memories>"
MEMORY_TAG_PATTERN = re.compile(r"<user_memories>.*?</user_memories>", 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 <user_memories> 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

91
packages/pipecat-sdk-python/src/supermemory_pipecat/utils.py
View file

@ -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: