From 2d281f8cbc8d019c0d3bb376b8ccd8cadca4fe4f Mon Sep 17 00:00:00 2001 From: Ishaan Gupta Date: Mon, 28 Sep 2026 20:56:25 +0530 Subject: [PATCH] fix(livekit): recall once per turn in llm_node and remember instantly 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. --- .github/workflows/ci-python.yml | 2 +- apps/docs/integrations/livekit.mdx | 32 +++++---- packages/livekit-sdk-python/README.md | 15 +++-- packages/livekit-sdk-python/pyproject.toml | 2 +- .../src/supermemory_livekit/agent.py | 24 +++++-- .../src/supermemory_livekit/memory.py | 67 +++++++++---------- .../livekit-sdk-python/tests/test_memory.py | 51 ++++++++++++++ 7 files changed, 136 insertions(+), 57 deletions(-) diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml index 4ead1e2d..ee8f8179 100644 --- a/.github/workflows/ci-python.yml +++ b/.github/workflows/ci-python.yml @@ -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" diff --git a/apps/docs/integrations/livekit.mdx b/apps/docs/integrations/livekit.mdx index f8e8e7ee..8a85a4d3 100644 --- a/apps/docs/integrations/livekit.mdx +++ b/apps/docs/integrations/livekit.mdx @@ -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-`, 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 diff --git a/packages/livekit-sdk-python/README.md b/packages/livekit-sdk-python/README.md index dfd116b8..b88bd531 100644 --- a/packages/livekit-sdk-python/README.md +++ b/packages/livekit-sdk-python/README.md @@ -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-`, 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-`, 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 diff --git a/packages/livekit-sdk-python/pyproject.toml b/packages/livekit-sdk-python/pyproject.toml index eaf142a3..f699f894 100644 --- a/packages/livekit-sdk-python/pyproject.toml +++ b/packages/livekit-sdk-python/pyproject.toml @@ -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", ] diff --git a/packages/livekit-sdk-python/src/supermemory_livekit/agent.py b/packages/livekit-sdk-python/src/supermemory_livekit/agent.py index 68183dc5..cab249bc 100644 --- a/packages/livekit-sdk-python/src/supermemory_livekit/agent.py +++ b/packages/livekit-sdk-python/src/supermemory_livekit/agent.py @@ -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) diff --git a/packages/livekit-sdk-python/src/supermemory_livekit/memory.py b/packages/livekit-sdk-python/src/supermemory_livekit/memory.py index 0cab7b2d..edbae245 100644 --- a/packages/livekit-sdk-python/src/supermemory_livekit/memory.py +++ b/packages/livekit-sdk-python/src/supermemory_livekit/memory.py @@ -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: diff --git a/packages/livekit-sdk-python/tests/test_memory.py b/packages/livekit-sdk-python/tests/test_memory.py index f07fef71..237ec291 100644 --- a/packages/livekit-sdk-python/tests/test_memory.py +++ b/packages/livekit-sdk-python/tests/test_memory.py @@ -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")