export session state as db and compute delta for sync

examples/voice_agents/sms_agent.py

@@ -104,7 +104,7 @@ async def text_handler(ctx: TextMessageContext):
 
     session = AgentSession(
         llm="openai/gpt-4.1-mini",
-        state_passphrase="my-secret-passphrase",
+        # state_passphrase="my-secret-passphrase",
     )
     if ctx.session_data:
         await session.rehydrate(ctx.session_data)

livekit-agents/livekit/agents/beta/workflows/address.py

@@ -84,11 +84,11 @@ def __init__(
     def get_init_kwargs(self) -> dict[str, Any]:
         return self._init_kwargs
 
-    def __getstate__(self) -> dict[str, Any]:
-        return super().__getstate__() | {"current_address": self._current_address}
+    def get_state(self) -> dict[str, Any]:
+        return super().get_state() | {"current_address": self._current_address}
 
-    def __setstate__(self, state: dict[str, Any]) -> None:
-        super().__setstate__(state)
+    def set_state(self, state: dict[str, Any]) -> None:
+        super().set_state(state)
         self._current_address = state["current_address"]
 
     async def on_enter(self) -> None:

livekit-agents/livekit/agents/beta/workflows/email_address.py

@@ -136,9 +136,9 @@ async def decline_email_capture(self, reason: str) -> None:
         if not self.done():
             self.complete(ToolError(f"couldn't get the email address: {reason}"))
 
-    def __getstate__(self) -> dict[str, Any]:
-        return super().__getstate__() | {"current_email": self._current_email}
+    def get_state(self) -> dict[str, Any]:
+        return super().get_state() | {"current_email": self._current_email}
 
-    def __setstate__(self, state: dict[str, Any]) -> None:
-        super().__setstate__(state)
+    def set_state(self, state: dict[str, Any]) -> None:
+        super().set_state(state)
         self._current_email = state["current_email"]

livekit-agents/livekit/agents/cli/cli.py

@@ -1168,6 +1168,8 @@ def _sms_text_mode(
     sess_data_file: str,
     loop: asyncio.AbstractEventLoop,
 ) -> None:
+    from ..utils.session_store import SessionStore
+
     while True:
         try:
             text = prompt(Text.from_markup("  [bold]User input[/bold]: "), console=c.console)
@@ -1178,6 +1180,7 @@ def _sms_text_mode(
             c.console.bell()
             continue
 
+        # TODO: use SessionStore to load the session data
         if os.path.exists(sess_data_file):
             with open(sess_data_file, "rb") as rf:
                 session_data = rf.read()
@@ -1206,7 +1209,7 @@ def _sms_text_mode(
         c.print(text, tag="You")
         c.wait_for_io_acquisition()
 
-        def _collect_responses(output_queue: queue.Queue[str | bytes]) -> None:
+        def _collect_responses(output_queue: queue.Queue[str | dict[str, Any]]) -> None:
             async def _collect() -> None:
                 text_ctx = get_job_context().text_message_context
                 if text_ctx is None:
@@ -1222,16 +1225,17 @@ def _done_callback(_: asyncio.Task[None]) -> None:
                     logger.warning("no session data available")
                     output_queue.put(b"", block=False)
                 else:
-                    output_queue.put(session.serialize(), block=False)
+                    output_queue.put(session.get_state(), block=False)
 
             task = asyncio.create_task(_collect())
             task.add_done_callback(_done_callback)
 
-        response_queue = queue.Queue[str | bytes]()
+        response_queue = queue.Queue[str | dict[str, Any]]()
         c.io_loop.call_soon_threadsafe(_collect_responses, response_queue, context=c.io_context)
 
+        new_state: dict[str, Any] | None = None
         while True:
-            resp: str | bytes = ""
+            resp: str | dict[str, Any] = ""
             with live_status(c.console, Text.from_markup("   [bold]Generating...[/bold]")):
                 while True:
                     try:
@@ -1241,14 +1245,27 @@ def _done_callback(_: asyncio.Task[None]) -> None:
                         pass
             if isinstance(resp, str) and resp:
                 c.print(resp, tag="Agent", tag_style=Style.parse("black on #B11FF9"))
-            elif isinstance(resp, bytes):
-                session_data = resp
+            elif isinstance(resp, dict):
+                new_state = resp
                 break
 
         # save the session data
-        if session_data:
-            with open(sess_data_file, "wb") as wf:
-                wf.write(session_data)
+        if new_state:
+            chnageset_dir = pathlib.Path(sess_data_file).with_suffix(".changesets")
+            chnageset_dir.mkdir(parents=True, exist_ok=True)
+            with SessionStore.from_session_state(new_state) as store:
+                # compute the changeset
+                if os.path.exists(sess_data_file):
+                    with SessionStore(db_file=sess_data_file) as old_store:
+                        delta = old_store.compute_delta(store)
+
+                    name = f"{delta.base_version[:8]}-{delta.new_version[:8]}.changeset"
+                    with open(chnageset_dir / name, "wb") as wf:
+                        wf.write(delta.dumps())
+
+                with open(sess_data_file, "wb") as wf:
+                    wf.write(store.export_database())
+
             logger.debug("session data saved", extra={"session_data_file": sess_data_file})
 
         # release the console for next run

livekit-agents/livekit/agents/llm/utils.py

@@ -81,11 +81,11 @@ class DiffOps:
     ]  # (previous_item_id, id), the items with the same id but different content
 
 
-def compute_chat_ctx_diff(old_ctx: ChatContext, new_ctx: ChatContext) -> DiffOps:
+def compute_chat_ctx_diff(old_ctx: ChatContext | list[str], new_ctx: ChatContext) -> DiffOps:
     """Computes the minimal list of create/remove operations to transform old_ctx into new_ctx."""
     # TODO(theomonnom): Make ChatMessage hashable and also add update ops
 
-    old_ids = [m.id for m in old_ctx.items]
+    old_ids = [m.id for m in old_ctx.items] if isinstance(old_ctx, ChatContext) else old_ctx
     new_ids = [m.id for m in new_ctx.items]
 
     lcs_ids = set(_compute_lcs(old_ids, new_ids))

livekit-agents/livekit/agents/utils/__init__.py

@@ -38,6 +38,7 @@
     "encryption",
 ]
 
+
 # Cleanup docs of unexported modules
 _module = dir()
 NOT_IN_ALL = [m for m in _module if m not in __all__]