feat(voice): opt-in PII redaction at the LLM egress boundary

livekit-agents/livekit/agents/__init__.py

@@ -90,7 +90,15 @@
     FunctionToolsExecutedEvent,
     MetricsCollectedEvent,
     ModelSettings,
+    NoopRedactor,
     RecordingOptions,
+    RedactedEntity,
+    RedactionContext,
+    RedactionOptions,
+    RedactionResult,
+    RedactionSink,
+    Redactor,
+    RegexRedactor,
     RunContext,
     SessionUsageUpdatedEvent,
     SpeechCreatedEvent,
@@ -202,6 +210,14 @@ def __getattr__(name: str) -> typing.Any:
     "SimulationVerdict",
     "AgentSession",
     "RecordingOptions",
+    "Redactor",
+    "RedactionOptions",
+    "RedactionSink",
+    "RedactionContext",
+    "RedactionResult",
+    "RedactedEntity",
+    "NoopRedactor",
+    "RegexRedactor",
     "text_transforms",
     "AgentEvent",
     "ModelSettings",

livekit-agents/livekit/agents/voice/__init__.py

@@ -18,6 +18,16 @@
     UserStateChangedEvent,
     UserTurnExceededEvent,
 )
+from .redaction import (
+    NoopRedactor,
+    RedactedEntity,
+    RedactionContext,
+    RedactionOptions,
+    RedactionResult,
+    RedactionSink,
+    Redactor,
+    RegexRedactor,
+)
 from .remote_session import RemoteSession
 from .room_io import (
     _ParticipantAudioOutput,
@@ -51,6 +61,14 @@
     "AgentFalseInterruptionEvent",
     "RemoteSession",
     "UserTurnExceededEvent",
+    "Redactor",
+    "RedactionOptions",
+    "RedactionSink",
+    "RedactionContext",
+    "RedactionResult",
+    "RedactedEntity",
+    "NoopRedactor",
+    "RegexRedactor",
     "TranscriptSynchronizer",
     "io",
     "room_io",

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

@@ -80,6 +80,7 @@
     remove_instructions,
     update_instructions,
 )
+from .redaction import RedactionSink, redact_chat_ctx
 from .speech_handle import DEFAULT_INPUT_DETAILS, InputDetails, SpeechHandle
 from .tool_executor import _resolve_async_tool_options, _ToolExecutor
 from .turn import (
@@ -493,6 +494,13 @@ async def update_tools(self, tools: list[llm.Tool | llm.Toolset]) -> None:
             # for realtime LLM, we assume the server will remove unvalid tool messages
             await self.update_chat_ctx(self._agent._chat_ctx.copy(tools=tools))
 
+    async def _redact_for_llm_egress(self, chat_ctx: llm.ChatContext) -> llm.ChatContext:
+        """Return a redacted copy of chat_ctx when the LLM redaction sink is enabled."""
+        redaction = self._session.options.redaction
+        if redaction is None or RedactionSink.LLM not in redaction.sinks:
+            return chat_ctx
+        return await redact_chat_ctx(chat_ctx, redaction, sink=RedactionSink.LLM)
+
     async def update_chat_ctx(
         self, chat_ctx: llm.ChatContext, *, exclude_invalid_function_calls: bool = True
     ) -> None:
@@ -501,7 +509,7 @@ async def update_chat_ctx(
 
         if self._rt_session is not None:
             remove_instructions(chat_ctx)
-            await self._rt_session.update_chat_ctx(chat_ctx)
+            await self._rt_session.update_chat_ctx(await self._redact_for_llm_egress(chat_ctx))
         else:
             update_instructions(
                 chat_ctx, instructions=self._agent.instructions, add_if_missing=True
@@ -2696,6 +2704,10 @@ async def _pipeline_reply_task_impl(
         # apply the correct variant of the instructions for the turn's input modality
         apply_instructions_modality(chat_ctx, modality=speech_handle.input_details.modality)
 
+        # redact a copy of the full context at the LLM egress boundary; placed before
+        # the llm_node call so overridden llm_node implementations are covered too
+        chat_ctx = await self._redact_for_llm_egress(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?
 
@@ -3210,7 +3222,7 @@ async def _realtime_reply_task(
         if user_input is not None:
             chat_ctx = self._rt_session.chat_ctx.copy()
             msg = chat_ctx.add_message(role="user", content=user_input)
-            await self._rt_session.update_chat_ctx(chat_ctx)
+            await self._rt_session.update_chat_ctx(await self._redact_for_llm_egress(chat_ctx))
             self._agent._chat_ctx._upsert_item(msg)
             self._session._conversation_item_added(msg)
 
@@ -3636,7 +3648,9 @@ def _create_assistant_message(
         # them, or message_outputs entries left in "skipped")
         if speech_handle.interrupted and any_skipped and self.llm.capabilities.mutable_chat_context:
             try:
-                await self._rt_session.update_chat_ctx(self._agent._chat_ctx)
+                await self._rt_session.update_chat_ctx(
+                    await self._redact_for_llm_egress(self._agent._chat_ctx)
+                )
             except llm.RealtimeError as e:
                 logger.warning(
                     "failed to sync chat context to remove never-played messages",
@@ -3757,7 +3771,9 @@ async def _wait_for_auto_tool_reply() -> None:
                 chat_ctx = self._rt_session.chat_ctx.copy()
                 chat_ctx.items.extend(new_fnc_outputs)
                 try:
-                    await self._rt_session.update_chat_ctx(chat_ctx)
+                    await self._rt_session.update_chat_ctx(
+                        await self._redact_for_llm_egress(chat_ctx)
+                    )
                 except llm.RealtimeError as e:
                     logger.warning(
                         "failed to update chat context before generating the function calls results",  # noqa: E501

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

@@ -62,6 +62,7 @@
 )
 from .ivr import IVRActivity
 from .recorder_io import RecorderIO
+from .redaction import RedactionOptions
 from .remote_session import RoomSessionTransport, SessionHost, SessionTransport
 from .run_result import RunResult
 from .speech_handle import InputDetails, SpeechHandle
@@ -152,6 +153,7 @@ class AgentSessionOptions:
     ivr_detection: bool
     aec_warmup_duration: float | None
     session_close_transcript_timeout: float
+    redaction: RedactionOptions | None
 
     @property
     def endpointing(self) -> EndpointingOptions:
@@ -246,6 +248,7 @@ def __init__(
         ivr_detection: bool = False,
         user_away_timeout: float | None = 15.0,
         session_close_transcript_timeout: float = 2.0,
+        redaction: NotGivenOr[RedactionOptions] = NOT_GIVEN,
         # Runtime settings
         conn_options: NotGivenOr[SessionConnectOptions] = NOT_GIVEN,
         loop: asyncio.AbstractEventLoop | None = None,
@@ -324,6 +327,10 @@ def __init__(
             session_close_transcript_timeout (float, optional): Seconds to wait for the
                 final STT transcript when closing the session (after audio is detached).
                 Default ``2.0`` s (independent of ``commit_user_turn``'s ``transcript_timeout``).
+            redaction (RedactionOptions, optional): Opt-in PII redaction applied to a copy
+                of the chat content at the enabled egress sinks (currently the chat context
+                sent to the LLM); the in-memory history keeps the raw values. Disabled when
+                NOT_GIVEN.
             preemptive_generation (NotGivenOr[bool | PreemptiveGenerationOptions]): Deprecated, use turn_handling=TurnHandlingOptions(...) instead.
             min_endpointing_delay (NotGivenOr[float]): Deprecated, use turn_handling=TurnHandlingOptions(...) instead.
             max_endpointing_delay (NotGivenOr[float]): Deprecated, use turn_handling=TurnHandlingOptions(...) instead.
@@ -396,6 +403,7 @@ def __init__(
             ),
             aec_warmup_duration=aec_warmup_duration,
             session_close_transcript_timeout=session_close_transcript_timeout,
+            redaction=redaction if is_given(redaction) else None,
         )
         self._conn_options = conn_options or SessionConnectOptions()
         self._started = False

livekit-agents/livekit/agents/voice/redaction.py

@@ -0,0 +1,191 @@
+from __future__ import annotations
+
+import asyncio
+import enum
+import re
+from dataclasses import dataclass, field
+from typing import Protocol, runtime_checkable
+
+from ..llm import ChatContext
+from ..log import logger
+
+REDACTION_FAILURE_MARKER = "[REDACTED]"
+"""Replacement applied to a text item when the redactor fails or times out (fail-closed)."""
+
+
+class RedactionSink(enum.Enum):
+    """A trust-boundary egress where text leaves the agent process."""
+
+    LLM = "llm"
+    """Chat context sent to a (typically third-party) LLM."""
+    TELEMETRY = "telemetry"
+    """Chat content written to telemetry spans/logs. Not wired yet."""
+    TRANSCRIPT = "transcript"
+    """Chat history serialized or uploaded at session end. Not wired yet."""
+
+
+@dataclass
+class RedactedEntity:
+    """Audit metadata for a single redacted span.
+
+    Carries only the entity type and the span in the *original* text —
+    never the raw value.
+    """
+
+    type: str
+    start: int
+    end: int
+
+
+@dataclass
+class RedactionContext:
+    """Context passed to a redactor so it can apply per-boundary policy."""
+
+    sink: RedactionSink
+    role: str | None = None
+    """Role of the chat item being redacted ("user", "assistant", "system", "tool")."""
+
+
+@dataclass
+class RedactionResult:
+    text: str
+    entities: list[RedactedEntity] = field(default_factory=list)
+
+
+@runtime_checkable
+class Redactor(Protocol):
+    async def redact(self, text: str, ctx: RedactionContext) -> RedactionResult: ...
+
+
+@dataclass
+class RedactionOptions:
+    """Opt-in redaction configuration for an ``AgentSession``.
+
+    Redaction is applied to a copy of the content at each enabled sink
+    (redact-on-egress): the in-memory chat history keeps the raw values.
+    """
+
+    redactor: Redactor
+    sinks: set[RedactionSink] = field(default_factory=lambda: {RedactionSink.LLM})
+    timeout: float = 3.0
+    """Max seconds a single redact() call may take before failing closed."""
+
+
+class NoopRedactor:
+    """Redactor that returns text unchanged."""
+
+    async def redact(self, text: str, ctx: RedactionContext) -> RedactionResult:
+        return RedactionResult(text=text)
+
+
+def _luhn_valid(digits: str) -> bool:
+    total = 0
+    for i, ch in enumerate(reversed(digits)):
+        d = int(ch)
+        if i % 2 == 1:
+            d *= 2
+            if d > 9:
+                d -= 9
+        total += d
+    return total % 10 == 0
+
+
+def _iban_valid(iban: str) -> bool:
+    rearranged = iban[4:] + iban[:4]
+    numeric = "".join(str(int(ch, 36)) for ch in rearranged)
+    return int(numeric) % 97 == 1
+
+
+def _ssn_valid(area: str, group: str, serial: str) -> bool:
+    if area in ("000", "666") or area.startswith("9"):
+        return False
+    return group != "00" and serial != "0000"
+
+
+class RegexRedactor:
+    """Best-effort redaction of structured PII using validated pattern matching.
+
+    Detects credit card numbers (Luhn-checked), US SSNs (format-checked), and
+    IBANs (mod-97-checked), replacing each with a ``[TYPE]`` placeholder. The
+    validators keep false positives low on what it claims to catch, but this is
+    pattern matching, not a compliance guarantee: it makes no attempt to detect
+    free-form PII such as names, addresses, or organizations. Layer a stronger
+    detector (e.g. Presidio) or vendor-side redaction for higher recall.
+    """
+
+    _CREDIT_CARD_RE = re.compile(r"\b(?:\d[ -]?){12,18}\d\b")
+    _SSN_RE = re.compile(r"\b(\d{3})-(\d{2})-(\d{4})\b")
+    _IBAN_RE = re.compile(r"\b[A-Z]{2}\d{2}[A-Z0-9]{11,30}\b")
+
+    async def redact(self, text: str, ctx: RedactionContext) -> RedactionResult:
+        spans: list[RedactedEntity] = []
+
+        for m in self._CREDIT_CARD_RE.finditer(text):
+            digits = re.sub(r"[ -]", "", m.group())
+            if 13 <= len(digits) <= 19 and _luhn_valid(digits):
+                spans.append(RedactedEntity(type="CREDIT_CARD", start=m.start(), end=m.end()))
+
+        for m in self._SSN_RE.finditer(text):
+            if _ssn_valid(m.group(1), m.group(2), m.group(3)):
+                spans.append(RedactedEntity(type="US_SSN", start=m.start(), end=m.end()))
+
+        for m in self._IBAN_RE.finditer(text):
+            if _iban_valid(m.group()):
+                spans.append(RedactedEntity(type="IBAN", start=m.start(), end=m.end()))
+
+        spans.sort(key=lambda e: e.start)
+        entities: list[RedactedEntity] = []
+        pieces: list[str] = []
+        cursor = 0
+        for entity in spans:
+            if entity.start < cursor:  # overlaps a previous match
+                continue
+            pieces.append(text[cursor : entity.start])
+            pieces.append(f"[{entity.type}]")
+            cursor = entity.end
+            entities.append(entity)
+        pieces.append(text[cursor:])
+
+        return RedactionResult(text="".join(pieces), entities=entities)
+
+
+async def _redact_text(text: str, opts: RedactionOptions, ctx: RedactionContext) -> str:
+    try:
+        result = await asyncio.wait_for(opts.redactor.redact(text, ctx), timeout=opts.timeout)
+        return result.text
+    except Exception:
+        # fail closed: never let raw text through on a redactor failure
+        logger.exception(
+            "redactor failed, replacing text with redaction marker",
+            extra={"sink": ctx.sink.value},
+        )
+        return REDACTION_FAILURE_MARKER
+
+
+async def redact_chat_ctx(
+    chat_ctx: ChatContext, opts: RedactionOptions, *, sink: RedactionSink
+) -> ChatContext:
+    """Return a redacted copy of ``chat_ctx``; the input context and its items are untouched.
+
+    Redacts the text content of messages and function call outputs. If the
+    redactor raises or exceeds ``opts.timeout``, the affected text is replaced
+    with :data:`REDACTION_FAILURE_MARKER`.
+    """
+    items = []
+    for item in chat_ctx.items:
+        if item.type == "message":
+            ctx = RedactionContext(sink=sink, role=item.role)
+            new_content = [
+                await _redact_text(c, opts, ctx) if isinstance(c, str) else c for c in item.content
+            ]
+            if new_content != item.content:
+                item = item.model_copy(update={"content": new_content})
+        elif item.type == "function_call_output":
+            ctx = RedactionContext(sink=sink, role="tool")
+            new_output = await _redact_text(item.output, opts, ctx)
+            if new_output != item.output:
+                item = item.model_copy(update={"output": new_output})
+
+        items.append(item)
+
+    return ChatContext(items)