scriptoriumadmin/supermemory
1
0
Fork 0
mirror of https://github.com/supermemoryai/supermemory.git synced 2026-09-30 01:51:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

fix(livekit): recall once per turn in llm_node and remember instantly

Browse source 
Found in live LiveKit Cloud voice calls:

- Recalling in on_user_turn_completed changed the turn context, so
  LiveKit discarded its preemptive generation on every turn with memory.
  SupermemoryAgent now recalls only in llm_node, and uses the turn hook
  only for realtime models, which skip llm_node.
- A turn with no memory recalled twice (hook, then llm_node), and tool
  follow-ups recalled again. Each user message is now recalled once;
  retries and follow-ups share the result.
- A preloaded profile made llm_node skip recall for the rest of the
  call. Fresh recall now replaces the earlier injection.
- remember used the default dynamic dreaming, so an explicit fact took
  about 18 minutes to become recallable. It now uses instant (~40s).
  The call transcript mode is configurable via capture_dreaming.
- recall_timeout defaults to 2s; profile calls measured 0.5-1.4s.
- Require livekit-agents>=1.3.6: AgentServer (used in the quick start)
  arrived in 1.3.1, and 1.3.1-1.3.5 no longer import with current
  opentelemetry-sdk.
This commit is contained in:
Ishaan Gupta 2026-09-28 20:56:25 +05:30
parent 9b8308beee
commit 2d281f8cbc
7 changed files with 136 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -268,7 +268,7 @@ jobs:
- python-version: "3.10"
dependency-lane: minimum-dependencies
supermemory-version: "3.50.0"
livekit-agents-version: "1.2.0"
livekit-agents-version: "1.3.6"
- python-version: "3.12"
dependency-lane: current-dependencies
supermemory-version: "3.59.0"

32
apps/docs/integrations/livekit.mdx
View file

@ -32,6 +32,8 @@ Container tags may only contain letters, numbers, `_`, `-`, and `:`, and must be
Identities that are not valid container tags are sanitized to a stable tag. Set the attribute when the identity is an email, phone number, or SIP address and you already have memories under a different id.
Set the attribute in the access token your backend issues. Do not grant the caller `canUpdateOwnMetadata`, or they could change the attribute and read another caller's memory.
One call is stored as a single document. Pass the LiveKit room name as `session_id`. The document id is `lk-<session_id>`, so a later update with the same id appends to that call instead of creating another.
## Quick start
@ -89,33 +91,37 @@ if __name__ == "__main__":
agents.cli.run_app(server)
```
`preload` puts the caller's profile into the first turn, so the greeting can use it. `attach` stores user and assistant turns when the session closes, including a caller who hangs up mid-turn. `SupermemoryAgent` recalls before each reply and adds the memory tools. Voice turns use `on_user_turn_completed`. Text turns from `generate_reply` or `session.run` skip that hook, so the agent also recalls inside `llm_node`. A turn that already has memory is not fetched again.
`preload` puts the caller's profile into the first turn, so the greeting can use it. `attach` stores the conversation after each agent reply, and stores anything left when the session closes, including a caller who hangs up mid-turn. `SupermemoryAgent` recalls inside `llm_node` before each reply and adds the memory tools. That covers voice turns, text turns from `generate_reply` or `session.run`, and LiveKit's preemptive generation. Each user message is recalled once, and tool follow-ups reuse it.
`agent_name` turns on explicit dispatch, which is how job metadata reaches the agent. Remove it to join every new room automatically, for example when testing in the [Agents Playground](https://agents-playground.livekit.io).
## Your own agent
If you already subclass `Agent`, keep that class. Pass the tools in, and recall from the turn hook.
If you already subclass `Agent`, keep that class. Pass the tools in, and recall from `llm_node`.
```python
from livekit.agents import Agent
from supermemory_livekit import SupermemoryLiveKit
memory = SupermemoryLiveKit(container_tag="user_123", session_id="room_123")
class Assistant(Agent):
def __init__(self) -> None:
def __init__(self, memory: SupermemoryLiveKit) -> None:
super().__init__(
instructions="You are a helpful voice assistant.",
tools=memory.tools(),
)
self.memory = memory
async def on_user_turn_completed(self, turn_ctx, new_message) -> None:
await memory.on_user_turn_completed(turn_ctx, new_message)
async def llm_node(self, chat_ctx, tools, model_settings):
await self.memory.enrich(chat_ctx)
return Agent.default.llm_node(self, chat_ctx, tools, model_settings)
```
Call `memory.attach(session)` before `session.start`.
`on_user_turn_completed` runs for STT-LLM-TTS pipelines. A realtime model only hits that hook when turn detection runs in your agent, not inside the model. See [LiveKit's external data guide](https://docs.livekit.io/agents/build/external-data/). Turn capture still listens for `conversation_item_added`.
Recall in `llm_node` rather than `on_user_turn_completed`. Changing the turn context in that hook makes LiveKit discard its preemptive generation, which adds latency to every turn.
Realtime models skip `llm_node`. With a realtime model, call `await self.memory.on_user_turn_completed(turn_ctx, new_message)` from `on_user_turn_completed` instead. That hook only runs when turn detection runs in your agent, not inside the model. See [LiveKit's external data guide](https://docs.livekit.io/agents/build/external-data/). `SupermemoryAgent` picks the right hook for you. Turn capture still listens for `conversation_item_added`.
## What gets recalled
@ -134,21 +140,23 @@ memory = SupermemoryLiveKit(
mode="full",
search_limit=10,
search_threshold=0.1,
recall_timeout=1.5,
recall_timeout=2.0,
capture="always",
capture_dreaming="dynamic",
),
)
```
Recall waits at most `recall_timeout` seconds (default 1.5). If the profile call is slower than that, the turn proceeds without memory. Retrieved text is inserted immediately before the user message and is not stored back as something the agent said.
Recall waits at most `recall_timeout` seconds (default 2). If the profile call is slower than that, the turn proceeds without memory. Retrieved text is inserted immediately before the user message and is not stored back as something the agent said.
| Parameter | Default | Description |
| --- | --- | --- |
| `search_limit` | `10` | Maximum search results merged into the turn |
| `search_threshold` | `0.1` | Minimum similarity, from 0 to 1 |
| `mode` | `"full"` | `profile`, `query`, or `full` |
| `recall_timeout` | `1.5` | Seconds to wait before skipping recall |
| `recall_timeout` | `2.0` | Seconds to wait before skipping recall |
| `capture` | `"always"` | `never` disables storing the call |
| `capture_dreaming` | `"dynamic"` | How the call transcript becomes memories. `dynamic` batches related documents and can take several minutes. `instant` is usually ready within a minute and bills one extra operation per document. Until then, `search_memories` still finds the raw call text. |
## Tools
@ -160,7 +168,7 @@ Recall waits at most `recall_timeout` seconds (default 1.5). If the profile call
| `remember` | Save one explicit fact, preference, or correction |
| `forget` | Forget one fact by id from `search_memories`, or by exact text |
`remember` stores a standalone fact. It is not appended to the call transcript. Automatic capture is what records the conversation.
`remember` stores a standalone fact and processes it right away, so the next call can recall it. It is not appended to the call transcript. Automatic capture is what records the conversation.
## Self-hosting

15
packages/livekit-sdk-python/README.md
View file

@ -71,9 +71,15 @@ if __name__ == "__main__":
agents.cli.run_app(server)
```
If you already have an `Agent` subclass, pass `tools=memory.tools()` and call `await memory.on_user_turn_completed(turn_ctx, new_message)` from `on_user_turn_completed`.
If you already have an `Agent` subclass, pass `tools=memory.tools()` and recall from `llm_node`:
`on_user_turn_completed` runs for STT-LLM-TTS pipelines. Realtime models only hit that hook when turn detection runs in the agent, not inside the model. Call capture still listens to `conversation_item_added`.
```python
async def llm_node(self, chat_ctx, tools, model_settings):
await memory.enrich(chat_ctx)
return Agent.default.llm_node(self, chat_ctx, tools, model_settings)
```
Recall in `llm_node`, not `on_user_turn_completed`: changing the turn context in that hook makes LiveKit discard its preemptive generation. Realtime models skip `llm_node`, so with one call `await memory.on_user_turn_completed(turn_ctx, new_message)` from `on_user_turn_completed`. `SupermemoryAgent` picks the right hook for you. Call capture listens to `conversation_item_added`.
## Configuration
@ -87,8 +93,9 @@ memory = SupermemoryLiveKit(
mode="full", # "profile" | "query" | "full"
search_limit=10,
search_threshold=0.1,
recall_timeout=1.5, # seconds; a slow recall is skipped
recall_timeout=2.0, # seconds; a slow recall is skipped
capture="always", # "always" | "never"
capture_dreaming="dynamic", # "dynamic" | "instant" (ready within a minute, extra operation)
),
)
```
@ -99,7 +106,7 @@ memory = SupermemoryLiveKit(
| `query` | No | Yes | You only need memories related to this turn |
| `full` | Yes | Yes | Default |
One call is stored as a single document under custom id `lk-<session_id>`, so a reconnect with the same session id updates that document instead of creating another. Explicit `remember` calls are separate facts and are not tied to the call document.
One call is stored as a single document under custom id `lk-<session_id>`, so a reconnect with the same session id updates that document instead of creating another. Explicit `remember` calls are separate facts, processed right away so the next call can recall them, and are not tied to the call document.
## Links

2
packages/livekit-sdk-python/pyproject.toml
View file

@ -30,7 +30,7 @@ classifiers = [
"Topic :: Scientific/Engineering :: Artificial Intelligence",
]
dependencies = [
"livekit-agents>=1.2.0,<2",
"livekit-agents>=1.3.6,<2",
"pydantic>=2.10.0",
"supermemory>=3.50.0",
]

24
packages/livekit-sdk-python/src/supermemory_livekit/agent.py
View file

@ -2,17 +2,21 @@
from typing import Any
from livekit.agents import Agent
from livekit.agents import Agent, llm
from .memory import SupermemoryLiveKit
class SupermemoryAgent(Agent):
"""LiveKit agent that injects Supermemory before each user turn.
"""LiveKit agent that injects Supermemory before each reply.
Recall runs in ``llm_node``, so voice and text turns both get memory and
LiveKit's preemptive generation is kept. Realtime models skip ``llm_node``,
so for those it runs in ``on_user_turn_completed``.
Pass any extra tools through ``tools``. Memory tools are added for you.
Override ``on_user_turn_completed`` and call ``super()`` if you also need
the hook.
Override ``llm_node`` or ``on_user_turn_completed`` and call ``super()`` if you
also need those hooks.
"""
def __init__(self, memory: SupermemoryLiveKit, **kwargs: Any) -> None:
@ -22,8 +26,18 @@ class SupermemoryAgent(Agent):
self.memory = memory
async def on_user_turn_completed(self, turn_ctx: Any, new_message: Any) -> None:
await self.memory.on_user_turn_completed(turn_ctx, new_message)
if self._uses_realtime_model():
await self.memory.on_user_turn_completed(turn_ctx, new_message)
async def llm_node(self, chat_ctx: Any, tools: list[Any], model_settings: Any) -> Any:
await self.memory.enrich(chat_ctx)
return Agent.default.llm_node(self, chat_ctx, tools, model_settings)
def _uses_realtime_model(self) -> bool:
model = self.llm
if not isinstance(model, (llm.LLM, llm.RealtimeModel)):
try:
model = self.session.llm
except RuntimeError:
return False
return isinstance(model, llm.RealtimeModel)

67
packages/livekit-sdk-python/src/supermemory_livekit/memory.py
View file

@ -34,14 +34,15 @@ class InputParams(BaseModel):
search_threshold: float = Field(default=0.1, ge=0.0, le=1.0)
system_prompt: str = Field(default="Relevant memory for this caller:\n\n")
mode: Literal["profile", "query", "full"] = "full"
recall_timeout: float = Field(default=1.5, gt=0.0, le=8.0)
recall_timeout: float = Field(default=2.0, gt=0.0, le=8.0)
capture: Literal["always", "never"] = "always"
capture_dreaming: Literal["dynamic", "instant"] = "dynamic"
class SupermemoryLiveKit:
"""Persistent memory for a LiveKit Agents session.
Call ``on_user_turn_completed`` before the LLM replies, and ``attach`` the
Call ``enrich`` from ``llm_node`` before the LLM replies, and ``attach`` the
session so completed turns are stored. ``tools()`` exposes search, remember,
and forget to the model. A Supermemory outage never fails the call.
"""
@ -73,6 +74,7 @@ class SupermemoryLiveKit:
self._buffer: list[dict[str, str]] = []
self._lock = asyncio.Lock()
self._flush_task: Optional[asyncio.Task[None]] = None
self._recall: Optional[tuple[tuple[Optional[str], str], asyncio.Future[Optional[str]]]] = None
self._session: Any = None
self._shutdown_registered = False
@ -134,45 +136,47 @@ class SupermemoryLiveKit:
self._inject(chat_ctx, text, created_at=None)
return True
async def on_user_turn_completed(self, turn_ctx: Any, new_message: Any) -> None:
"""Retrieve memory for this turn and insert it just before the user message."""
query = message_text(new_message)
if not query:
return
try:
text = await self._recall_text(query=query)
if not text:
return
created_at = getattr(new_message, "created_at", None)
before = created_at - 0.001 if isinstance(created_at, (int, float)) else None
self._strip_injected(turn_ctx)
self._inject(turn_ctx, text, created_at=before)
except Exception:
logger.warning("memory inject failed", exc_info=True)
async def enrich(self, chat_ctx: Any) -> None:
"""Recall for a context that already contains the user message.
"""Recall for the latest user message and insert it just before that message.
``session.run`` and ``generate_reply`` skip ``on_user_turn_completed``, so the
agent calls this from ``llm_node``. A voice turn that already injected memory
is left alone.
Call from ``llm_node``. That covers voice and text turns, and runs during
LiveKit's preemptive generation instead of invalidating it. Each user message
is recalled once; tool follow-ups and preemptive retries reuse the result.
"""
if self._has_injection(chat_ctx):
return
user = self._last_user_item(chat_ctx)
query = message_text(user) if user is not None else None
if user is not None:
await self._recall_into(chat_ctx, user)
async def on_user_turn_completed(self, turn_ctx: Any, new_message: Any) -> None:
"""Recall from the turn hook, for realtime models that skip ``llm_node``.
With an STT-LLM-TTS pipeline use ``enrich`` from ``llm_node`` instead. Editing
the turn context here discards LiveKit's preemptive generation.
"""
await self._recall_into(turn_ctx, new_message)
async def _recall_into(self, chat_ctx: Any, user: Any) -> None:
query = message_text(user)
if not query:
return
try:
text = await self._recall_text(query=query)
text = await self._recall_once(query)
if not text:
return
created_at = getattr(user, "created_at", None)
before = created_at - 0.001 if isinstance(created_at, (int, float)) else None
self._strip_injected(chat_ctx)
self._inject(chat_ctx, text, created_at=before)
except Exception:
logger.warning("memory inject failed", exc_info=True)
async def _recall_once(self, query: str) -> Optional[str]:
key = (self.container_tag, query)
if self._recall is None or self._recall[0] != key:
self._recall = (key, asyncio.ensure_future(self._recall_text(query=query)))
# A cancelled preemptive generation must not cancel a recall the next attempt reuses.
return await asyncio.shield(self._recall[1])
def attach(
self,
session: Any,
@ -243,6 +247,7 @@ class SupermemoryLiveKit:
content=text,
container_tag=tag,
metadata={"source": "livekit", "kind": "explicit"},
dreaming="instant",
),
timeout=4.0,
)
@ -285,7 +290,7 @@ class SupermemoryLiveKit:
self._retrieve(query if include_search else None),
timeout=self.params.recall_timeout,
)
except TimeoutError:
except asyncio.TimeoutError:
logger.warning("memory recall timed out after %.2fs", self.params.recall_timeout)
return None
except Exception:
@ -344,13 +349,6 @@ class SupermemoryLiveKit:
return item
return None
def _has_injection(self, chat_ctx: Any) -> bool:
for item in getattr(chat_ctx, "items", []) or []:
text = message_text(item)
if text and is_injected_memory(text):
return True
return False
def _strip_injected(self, chat_ctx: Any) -> None:
items = list(getattr(chat_ctx, "items", []) or [])
for item in items:
@ -417,6 +415,7 @@ class SupermemoryLiveKit:
container_tag=self.container_tag,
custom_id=self._custom_id(),
metadata={"source": "livekit", "kind": "conversation"},
dreaming=self.params.capture_dreaming,
)
def _custom_id(self) -> str:

51
packages/livekit-sdk-python/tests/test_memory.py
View file

@ -187,6 +187,55 @@ class MemoryTests(unittest.TestCase):
self.assertAlmostEqual(ctx.items[1].created_at, 7.999)
asyncio.run(plugin.enrich(ctx))
self.assertEqual(len(client.profile.calls), 1)
self.assertEqual(len(ctx.items), 2)
def test_enrich_replaces_preloaded_profile_each_turn(self):
client = FakeClient(
FakeProfile(static=["Name is Ada"], results=[SimpleNamespace(memory="Owns a dog named Biscuit")])
)
plugin = memory(client, container_tag="user_1")
ctx = ChatCtx()
asyncio.run(plugin.preload(ctx))
ctx.add_message(role="user", content="what is my dog called?", created_at=30.0)
asyncio.run(plugin.enrich(ctx))
injected = [item for item in ctx.items if is_injected_memory(item.content)]
self.assertEqual(len(injected), 1)
self.assertIn("Biscuit", injected[0].content)
self.assertEqual(client.profile.calls[-1]["q"], "what is my dog called?")
def test_one_recall_per_user_message(self):
client = FakeClient(FakeProfile(static=[], dynamic=[], delay=0.05))
plugin = memory(client, container_tag="user_1")
async def run():
first, second = ChatCtx(), ChatCtx()
for ctx in (first, second):
ctx.add_message(role="user", content="hi, first time calling", created_at=5.0)
# A preemptive attempt is cancelled while the retry reuses its recall.
preemptive = asyncio.ensure_future(plugin.enrich(first))
await asyncio.sleep(0.01)
preemptive.cancel()
await plugin.enrich(second)
await plugin.enrich(second)
asyncio.run(run())
self.assertEqual(len(client.profile.calls), 1)
def test_new_user_message_recalls_again(self):
client = FakeClient(FakeProfile(static=["Name is Ada"]))
plugin = memory(client, container_tag="user_1")
ctx = ChatCtx()
ctx.add_message(role="user", content="hello", created_at=1.0)
asyncio.run(plugin.enrich(ctx))
ctx.add_message(role="user", content="what's my name?", created_at=2.0)
asyncio.run(plugin.enrich(ctx))
self.assertEqual([call["q"] for call in client.profile.calls], ["hello", "what's my name?"])
self.assertEqual(sum(is_injected_memory(item.content) for item in ctx.items), 1)
def test_timeout_and_errors_do_not_fail_the_turn(self):
slow = FakeClient(FakeProfile(static=["Name is Ada"], delay=0.05))
@ -261,6 +310,7 @@ class MemoryTests(unittest.TestCase):
self.assertEqual(stored["custom_id"], to_identifier("lk-room 1"))
self.assertNotIn("secret", stored["content"])
self.assertEqual(stored["metadata"]["source"], "livekit")
self.assertEqual(stored["dreaming"], "dynamic")
def test_close_flushes_a_trailing_user_turn(self):
client = FakeClient()
@ -340,6 +390,7 @@ class MemoryTests(unittest.TestCase):
self.assertNotIn("chunk_1", found)
self.assertEqual(saved, "Saved.")
self.assertEqual(client.added[0]["metadata"]["kind"], "explicit")
self.assertEqual(client.added[0]["dreaming"], "instant")
self.assertNotIn("custom_id", client.added[0])
self.assertEqual(forgotten, "Forgotten.")
self.assertEqual(client.memories.calls[0]["id"], "mem_1")