(expressive mode): add to update_options

livekit-agents/livekit/agents/voice/agent_activity.py

@@ -70,6 +70,7 @@
     _AudioOutput,
     _ForwardOutput,
     _inject_running_tool_calls,
+    _strip_assistant_markup,
     _strip_running_tool_calls,
     _TextOutput,
     _TTSGenerationData,
@@ -79,7 +80,9 @@
     perform_text_forwarding,
     perform_tool_executions,
     perform_tts_inference,
+    remove_expressive_instructions,
     remove_instructions,
+    update_expressive_instructions,
     update_instructions,
 )
 from .speech_handle import DEFAULT_INPUT_DETAILS, InputDetails, SpeechHandle
@@ -2500,7 +2503,9 @@ def _to_instructions(v: Instructions | str) -> Instructions:
                 },
             )
             if text.strip():
-                chat_ctx.add_message(role="system", content=text)
+                # keyed message: re-injection replaces last turn's guide instead of
+                # stacking copies, and an expressive-off turn removes it again
+                update_expressive_instructions(chat_ctx, text=text)
 
     def _on_pipeline_reply_done(self, _: asyncio.Task[None]) -> None:
         if not self._speech_q and (not self._current_speech or self._current_speech.done()):
@@ -2839,6 +2844,20 @@ async def _pipeline_reply_task_impl(
         _expr_opts = self._resolve_expressive_options()
         if _expr_opts is not None:
             self._inject_expressive_instructions(chat_ctx, _expr_opts, speech_handle)
+        else:
+            # expressive is off for this turn (toggled off via update_options, an agent
+            # override, or a handoff to a TTS without a markup dialect): remove the
+            # injected markup guide and scrub markup left in past assistant turns so
+            # the LLM isn't instructed or few-shotted into emitting tags nothing
+            # downstream converts or strips — an unsupported tag would reach the TTS
+            # as literal text and be spoken.
+            remove_expressive_instructions(chat_ctx)
+            _strip_assistant_markup(chat_ctx)
+            if chat_ctx is not self._agent._chat_ctx:
+                # user turns run on a copy of the agent's history; clean the stored
+                # history too so stale markup doesn't survive into future snapshots
+                remove_expressive_instructions(self._agent._chat_ctx)
+                _strip_assistant_markup(self._agent._chat_ctx)
 
         # TODO(theomonnom): since pause is closing STT/LLM/TTS, we have issues for SpeechHandle still in queue  # noqa: E501
         # I should implement a retry mechanism?

livekit-agents/livekit/agents/voice/agent_session.py

@@ -15,6 +15,7 @@
     Literal,
     Protocol,
     TypeVar,
+    cast,
     overload,
     runtime_checkable,
 )
@@ -1150,6 +1151,7 @@ def update_options(
         endpointing_opts: NotGivenOr[EndpointingOptions] = NOT_GIVEN,
         turn_detection: NotGivenOr[TurnDetectionMode | None] = NOT_GIVEN,
         keyterms: NotGivenOr[list[str]] = NOT_GIVEN,
+        expressive: NotGivenOr[bool | ExpressiveOptions] = NOT_GIVEN,
         # deprecated
         min_endpointing_delay: NotGivenOr[float] = NOT_GIVEN,
         max_endpointing_delay: NotGivenOr[float] = NOT_GIVEN,
@@ -1163,9 +1165,18 @@ def update_options(
                 when the user has finished speaking. ``None`` reverts to automatic selection.
             keyterms (NotGivenOr[list[str]], optional): Replace the user-defined keyterms applied
                 to the STT. Auto-detected keyterms are left untouched.
+            expressive (NotGivenOr[bool | ExpressiveOptions], optional): Turn expressive TTS
+                delivery on/off or switch its preset/options mid-session. Takes effect on the
+                next reply. An ``expressive`` set on the active :class:`Agent` overrides the
+                session value. When a turn runs with expressive off, markup left in past
+                assistant messages is stripped from the chat history so the LLM doesn't
+                imitate tags nothing downstream converts.
             min_endpointing_delay: Deprecated, use ``endpointing_opts`` instead.
             max_endpointing_delay: Deprecated, use ``endpointing_opts`` instead.
         """
+        if is_given(expressive):
+            # mypy can't narrow NotGiven out of the TypedDict union here
+            self._opts.expressive = cast("bool | ExpressiveOptions", expressive)
         if is_given(keyterms):
             self._keyterm_detector.set_static_keyterms(keyterms)
         if is_given(min_endpointing_delay) or is_given(max_endpointing_delay):

livekit-agents/livekit/agents/voice/generation.py

@@ -103,6 +103,28 @@ def _inject_running_tool_calls(
         )
 
 
+def _strip_assistant_markup(chat_ctx: ChatContext) -> None:
+    """Remove expressive TTS markup from past assistant messages, in place.
+
+    Called when a turn runs with expressive off (toggled off via
+    ``session.update_options``, an agent-level override, or a handoff to a TTS
+    without a markup dialect): tags left in history would few-shot the LLM into
+    emitting markup that nothing downstream converts or strips, so an unsupported
+    tag would reach the TTS as literal text and be spoken. Mutates the stored
+    history: once a turn runs with expressive off, prior turns' markup is gone
+    even if expressive is re-enabled later (the re-injected instructions carry
+    the style examples instead).
+    """
+    from ..tts._provider_format import strip_all_markup
+
+    for item in chat_ctx.items:
+        if item.type != "message" or item.role != "assistant":
+            continue
+        if not any(isinstance(c, str) and ("<" in c or "[" in c) for c in item.content):
+            continue
+        item.content = [strip_all_markup(c) if isinstance(c, str) else c for c in item.content]
+
+
 def _strip_running_tool_calls(chat_ctx: ChatContext) -> None:
     """Remove the pairs added by :func:`_inject_running_tool_calls`, keeping everything
     else (e.g. items a custom ``llm_node`` added)."""
@@ -1016,3 +1038,39 @@ def remove_instructions(chat_ctx: ChatContext) -> None:
             chat_ctx.items.remove(msg)
         else:
             break
+
+
+EXPRESSIVE_INSTRUCTIONS_MESSAGE_ID = "lk.expressive.instructions"  #  value must not change
+"""
+The ID of the expressive TTS markup-guide message in the chat context.
+"""
+
+
+def update_expressive_instructions(chat_ctx: ChatContext, *, text: str) -> None:
+    """Insert or replace the expressive markup-guide system message.
+
+    Keyed by :data:`EXPRESSIVE_INSTRUCTIONS_MESSAGE_ID` so per-turn re-injection
+    replaces the previous guide instead of accumulating one copy per turn, and a
+    turn that runs with expressive off can remove it again
+    (:func:`remove_expressive_instructions`).
+    """
+    idx = chat_ctx.index_by_id(EXPRESSIVE_INSTRUCTIONS_MESSAGE_ID)
+    if idx is not None:
+        chat_ctx.items[idx] = llm.ChatMessage(
+            id=EXPRESSIVE_INSTRUCTIONS_MESSAGE_ID,
+            role="system",
+            content=[text],
+            created_at=chat_ctx.items[idx].created_at,
+        )
+    else:
+        chat_ctx.add_message(role="system", content=text, id=EXPRESSIVE_INSTRUCTIONS_MESSAGE_ID)
+
+
+def remove_expressive_instructions(chat_ctx: ChatContext) -> None:
+    """Remove the expressive markup-guide message added by
+    :func:`update_expressive_instructions`, if present."""
+    while True:
+        if msg := chat_ctx.get_by_id(EXPRESSIVE_INSTRUCTIONS_MESSAGE_ID):
+            chat_ctx.items.remove(msg)
+        else:
+            break

tests/test_expressive_toggle.py

@@ -0,0 +1,122 @@
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from livekit.agents import Agent, AgentSession
+from livekit.agents.llm.chat_context import ChatContext
+from livekit.agents.voice import presets
+from livekit.agents.voice.generation import (
+    EXPRESSIVE_INSTRUCTIONS_MESSAGE_ID,
+    _strip_assistant_markup,
+    remove_expressive_instructions,
+    update_expressive_instructions,
+)
+
+from .fake_session import FakeActions, create_session, run_session
+
+pytestmark = [pytest.mark.unit, pytest.mark.virtual_time, pytest.mark.no_concurrent]
+
+SESSION_TIMEOUT = 60
+
+
+MARKED_UP = '<expression value="happy"/> Welcome back! [chuckling] Glad you called again.'
+
+
+def test_strip_assistant_markup() -> None:
+    ctx = ChatContext.empty()
+    ctx.add_message(role="assistant", content=MARKED_UP)
+    ctx.add_message(role="user", content='I typed <expression value="happy"/> literally')
+    plain = ctx.add_message(role="assistant", content="No tags here.")
+    plain_content = plain.content
+
+    _strip_assistant_markup(ctx)
+
+    assistant_texts = [
+        item.text_content
+        for item in ctx.items
+        if item.type == "message" and item.role == "assistant"
+    ]
+    assert "<expression" not in (assistant_texts[0] or "")
+    assert "[chuckling]" not in (assistant_texts[0] or "")
+    assert "Welcome back!" in (assistant_texts[0] or "")
+    assert "Glad you called again." in (assistant_texts[0] or "")
+
+    # user content is never touched, tag-shaped or not
+    user_text = next(
+        item.text_content for item in ctx.items if item.type == "message" and item.role == "user"
+    )
+    assert '<expression value="happy"/>' in (user_text or "")
+
+    # tag-free assistant content is left as-is (fast path)
+    assert plain.content is plain_content
+
+
+def test_update_options_expressive() -> None:
+    session = AgentSession(expressive=presets.CUSTOMER_SERVICE)
+    assert session.options.expressive == presets.CUSTOMER_SERVICE
+
+    # turn off
+    session.update_options(expressive=False)
+    assert session.options.expressive is False
+
+    # turn back on with a different preset
+    session.update_options(expressive=presets.CASUAL)
+    assert session.options.expressive == presets.CASUAL
+
+    # untouched when not given
+    session.update_options()
+    assert session.options.expressive == presets.CASUAL
+
+
+def test_update_and_remove_expressive_instructions() -> None:
+    ctx = ChatContext.empty()
+    update_expressive_instructions(ctx, text="markup guide v1")
+    update_expressive_instructions(ctx, text="markup guide v2")
+
+    guides = [item for item in ctx.items if item.id == EXPRESSIVE_INSTRUCTIONS_MESSAGE_ID]
+    assert len(guides) == 1, "re-injection must replace, not stack"
+    assert guides[0].text_content == "markup guide v2"
+
+    remove_expressive_instructions(ctx)
+    assert all(item.id != EXPRESSIVE_INSTRUCTIONS_MESSAGE_ID for item in ctx.items)
+
+
+async def test_expressive_off_turn_scrubs_history() -> None:
+    """A turn that runs with expressive off removes the injected markup guide and
+    scrubs markup from past assistant turns."""
+    actions = FakeActions()
+    actions.add_user_speech(0.5, 2.5, "Hello, how are you?", stt_delay=0.2)
+    actions.add_llm("I'm doing well, thank you!", ttft=0.1, duration=0.3)
+    actions.add_tts(2.0, ttfb=0.2, duration=0.3)
+
+    # FakeTTS has no markup dialect, so expressive resolves to off even though the
+    # session asks for it — same situation as a handoff to a non-expressive TTS.
+    session = create_session(actions, extra_kwargs={"expressive": presets.CUSTOMER_SERVICE})
+
+    # seed history as if previous turns ran expressive: marked-up assistant text +
+    # the injected markup guide
+    seeded = ChatContext.empty()
+    seeded.add_message(role="assistant", content=MARKED_UP)
+    update_expressive_instructions(seeded, text="Use the formatting tags below…")
+    agent = Agent(instructions="You are a helpful assistant.", chat_ctx=seeded)
+
+    await asyncio.wait_for(run_session(session, agent), timeout=SESSION_TIMEOUT)
+
+    # the injected guide is gone
+    assert all(item.id != EXPRESSIVE_INSTRUCTIONS_MESSAGE_ID for item in agent.chat_ctx.items)
+
+    assistant_texts = [
+        item.text_content
+        for item in agent.chat_ctx.items
+        if item.type == "message" and item.role == "assistant"
+    ]
+    assert assistant_texts, "expected assistant messages in history"
+    for text in assistant_texts:
+        assert "<expression" not in (text or "")
+        assert "[chuckling]" not in (text or "")
+    # the seeded message's visible text survives the scrub
+    assert any("Welcome back!" in (t or "") for t in assistant_texts)
+    # and the new reply went through normally
+    assert any("I'm doing well" in (t or "") for t in assistant_texts)