parlot.instrumentation.livekit

parlot-instrumentation-livekit: OTel instrumentation for LiveKit Agents.

View Source

 1"""parlot-instrumentation-livekit: OTel instrumentation for LiveKit Agents."""
 2
 3from parlot.core import (
 4    add_platform_ref,
 5    human_escalation,
 6    record_human_rep,
 7    set_session_attribute,
 8    set_session_metadata,
 9    stamp_platform_refs,
10)
11
12from ._auto import parlotize
13from ._events import install_session_hooks
14from ._processor import LiveKitGenAIProcessor
15
16__all__ = [
17    "parlotize",
18    "install_session_hooks",
19    "LiveKitGenAIProcessor",
20    "add_platform_ref",
21    "human_escalation",
22    "record_human_rep",
23    "set_session_attribute",
24    "set_session_metadata",
25    "stamp_platform_refs",
26]

def parlotize( agent_id: str, *, endpoint: Optional[str] = None, api_key: Optional[str] = None, capture_genai_content: Optional[bool] = None, service_name: Optional[str] = None, tracer_provider: opentelemetry.trace.TracerProvider | None = None, auto_escalate_sip: bool = False, escalation_metadata_match: dict[str, str] | None = None, version: Optional[str] = None, record: bool | list[str] | None = None, capture_logs: bool | list[str] | None = None, log_level: Optional[str] = None) -> parlot.core.ParlotContext: View Source

 66def parlotize(
 67    agent_id: str,
 68    *,
 69    endpoint: Optional[str] = None,
 70    api_key: Optional[str] = None,
 71    capture_genai_content: Optional[bool] = None,
 72    service_name: Optional[str] = None,
 73    tracer_provider: TracerProvider | None = None,
 74    auto_escalate_sip: bool = False,
 75    escalation_metadata_match: dict[str, str] | None = None,
 76    version: Optional[str] = None,
 77    record: bool | list[str] | None = None,
 78    capture_logs: bool | list[str] | None = None,
 79    log_level: Optional[str] = None,
 80) -> ParlotContext:
 81    """Parlotize LiveKit instrumentation and OTLP export.
 82
 83    Call before constructing ``AgentSession``. Builds a ``TracerProvider`` with
 84    an OTLP exporter, registers it with ``livekit.agents.telemetry``, fetches
 85    telemetry bootstrap when ``PARLOT_API_KEY`` is set, patches
 86    ``JobContext.connect`` / ``AgentSession.__init__``, and installs session
 87    event hooks.
 88
 89    Shared parameters (``agent_id``, plus keyword-only ``endpoint``, ``api_key``,
 90    ``version``, ``capture_genai_content``, ``capture_logs``, ``log_level``,
 91    ``service_name``, ``tracer_provider``) match every adapter — see
 92    ``parlot.core.ParlotizeProtocol``.
 93
 94    Under LiveKit ``dev`` / job workers, ``__main__`` is often LiveKit's IPC
 95    entrypoint, not your agent file — pass ``agent_id`` / ``version=``
 96    explicitly when you care about stable deployment identity.
 97
 98    Args:
 99        agent_id: Shared — required canonical ``session.agent_id``. Must be
100            non-empty after stripping whitespace.
101        endpoint: Shared — Parlot OTLP base URL (or ``PARLOT_ENDPOINT``).
102        api_key: Shared — org API key (or ``PARLOT_API_KEY``).
103        capture_genai_content: Shared — GenAI payload capture override.
104            Precedence: job metadata → this kwarg → Settings → Generative AI → on.
105        service_name: Shared — OTel ``service.name`` (defaults to ``agent_id``
106            or ``"unknown"``).
107        tracer_provider: Shared — existing ``TracerProvider``, or build one with
108            Parlot's OTLP exporter and ``LiveKitGenAIProcessor``.
109        auto_escalate_sip: When ``True``, mark the session escalated when a SIP
110            participant joins the room.
111        escalation_metadata_match: Participant metadata key/value pairs that
112            classify joining participants as human representatives.
113        version: Shared — ``gen_ai.agent.version``. Pass ``version=`` to set it;
114            otherwise stamped as ``\"unknown\"``.
115        record: Audio recording policy. Boolean or agent-id glob patterns
116            (e.g. ``["support-*", "billing"]``). Precedence: LiveKit job
117            metadata ``record`` → this kwarg → Settings → Recording.
118        capture_logs: Shared — session log capture (bool or globs). Precedence:
119            job metadata → this kwarg → Settings → Logs → on.
120        log_level: Shared — minimum level for session log capture.
121
122    Returns:
123        The ``ParlotContext`` created for this process (or the prior one if
124        already configured).
125    """
126    global _configured, _auto_escalate_sip, _escalation_metadata_match
127    global _configured_agent_id, _configured_agent_version
128    global _configured_record
129    global _configured_capture_genai_content, _configured_capture_logs, _configured_log_level
130    global _parlot_context
131    if _configured:
132        logger.debug("parlot-instrumentation.livekit already configured — skipping")
133        assert _parlot_context is not None
134        return _parlot_context
135
136    if _is_livekit_dev_watch_parent():
137        logger.debug(
138            "Skipping parlot-instrumentation.livekit parlotize in LiveKit dev "
139            "watcher parent (worker child will parlotize)"
140        )
141        return ParlotContext()
142
143    _auto_escalate_sip = auto_escalate_sip
144    _escalation_metadata_match = (
145        dict(escalation_metadata_match) if escalation_metadata_match else None
146    )
147    if isinstance(record, list):
148        _configured_record = [str(item).strip() for item in record if str(item).strip()]
149    else:
150        _configured_record = record
151    from parlot.core.provider import resolve_capture_genai_content
152
153    _configured_capture_genai_content = resolve_capture_genai_content(capture_genai_content)
154    if isinstance(capture_logs, list):
155        _configured_capture_logs = [
156            str(item).strip() for item in capture_logs if str(item).strip()
157        ]
158    else:
159        _configured_capture_logs = capture_logs
160    _configured_log_level = log_level.strip().upper() if log_level else None
161
162    from ._agent_version import resolve_agent_version
163    _configured_agent_version = resolve_agent_version(version)
164
165    from parlot.core import base_parlotize
166
167    res = base_parlotize(
168        agent_id,
169        endpoint=endpoint,
170        api_key=api_key,
171        capture_genai_content=capture_genai_content,
172        tracer_provider=tracer_provider,
173        version=_configured_agent_version,
174        capture_logs=capture_logs,
175        log_level=log_level,
176    )
177    _parlot_context = res.context
178    _configured_agent_id = res.agent_id
179
180    from ._recording_guard import (
181        current_job_capture_logs_metadata,
182        livekit_session_log_fields,
183        recording_agent_id_from_ctx,
184    )
185
186    def _agent_id_resolver() -> str:
187        try:
188            from livekit.agents.job import get_job_context
189
190            ctx = get_job_context()
191            if ctx is not None:
192                return recording_agent_id_from_ctx(ctx)
193        except Exception:
194            pass
195        return _configured_agent_id or ""
196
197    res.context.session_logs.set_resolvers(
198        session_resolver=livekit_session_log_fields,
199        agent_id_resolver=_agent_id_resolver,
200        metadata_resolver=current_job_capture_logs_metadata,
201    )
202
203    tracer_provider = res.tracer_provider
204    if tracer_provider is None:
205        tracer_provider = _build_provider(
206            endpoint=res.endpoint,
207            api_key=res.api_key,
208            service_name=service_name,
209            service_version=_configured_agent_version or None,
210            context=res.context,
211        )
212
213    from parlot.core.processor import assert_sync_span_processors
214
215    from ._session import set_span_context_attach_enabled
216
217    attach_ok = assert_sync_span_processors(tracer_provider)
218    set_span_context_attach_enabled(attach_ok)
219
220    _register_with_livekit(tracer_provider)
221    _patch_agent_session_with_tracer(tracer_provider)
222    _patch_job_context_connect()
223    _install_telemetry_compare()
224
225    _configured = True
226    logger.debug("parlot-instrumentation.livekit configured (endpoint=%s)", res.endpoint)
227    return res.context

Parlotize LiveKit instrumentation and OTLP export.

Call before constructing AgentSession. Builds a TracerProvider with an OTLP exporter, registers it with livekit.agents.telemetry, fetches telemetry bootstrap when PARLOT_API_KEY is set, patches JobContext.connect / AgentSession.__init__, and installs session event hooks.

Shared parameters (agent_id, plus keyword-only endpoint, api_key, version, capture_genai_content, capture_logs, log_level, service_name, tracer_provider) match every adapter — see parlot.core.ParlotizeProtocol.

Under LiveKit dev / job workers, __main__ is often LiveKit's IPC entrypoint, not your agent file — pass agent_id / version= explicitly when you care about stable deployment identity.

Arguments:

agent_id: Shared — required canonical session.agent_id. Must be non-empty after stripping whitespace.
endpoint: Shared — Parlot OTLP base URL (or PARLOT_ENDPOINT).
api_key: Shared — org API key (or PARLOT_API_KEY).
capture_genai_content: Shared — GenAI payload capture override. Precedence: job metadata → this kwarg → Settings → Generative AI → on.
service_name: Shared — OTel service.name (defaults to agent_id or "unknown").
tracer_provider: Shared — existing TracerProvider, or build one with Parlot's OTLP exporter and LiveKitGenAIProcessor.
auto_escalate_sip: When True, mark the session escalated when a SIP participant joins the room.
escalation_metadata_match: Participant metadata key/value pairs that classify joining participants as human representatives.
version: Shared — gen_ai.agent.version. Pass version= to set it; otherwise stamped as "unknown".
record: Audio recording policy. Boolean or agent-id glob patterns (e.g. ["support-*", "billing"]). Precedence: LiveKit job metadata record → this kwarg → Settings → Recording.
capture_logs: Shared — session log capture (bool or globs). Precedence: job metadata → this kwarg → Settings → Logs → on.
log_level: Shared — minimum level for session log capture.

Returns:

The ParlotContext created for this process (or the prior one if already configured).

def install_session_hooks( session: Any, processor: LiveKitGenAIProcessor, tracer: opentelemetry.trace.Tracer) -> None: View Source

483def install_session_hooks(
484    session: Any,
485    processor: "LiveKitGenAIProcessor",
486    tracer: "Tracer",
487) -> None:
488    """Install all AgentSession event listeners for Parlot instrumentation.
489
490    ``parlotize()`` patches ``AgentSession.__init__`` to invoke this
491    automatically. Call directly only when manually instantiating unpatched
492    sessions.
493
494    Args:
495        session: LiveKit ``AgentSession`` instance to instrument.
496        processor: Span processor that owns session/turn enrichment state.
497        tracer: OpenTelemetry tracer used for Parlot contract spans.
498    """
499    LiveKitEventBridge(processor, tracer).install(session)

Install all AgentSession event listeners for Parlot instrumentation.

parlotize() patches AgentSession.__init__ to invoke this automatically. Call directly only when manually instantiating unpatched sessions.

Arguments:

session: LiveKit AgentSession instance to instrument.
processor: Span processor that owns session/turn enrichment state.
tracer: OpenTelemetry tracer used for Parlot contract spans.

def add_platform_ref(kind: str, value: str, *, framework: str = 'custom') -> None: View Source

 71def add_platform_ref(
 72    kind: str,
 73    value: str,
 74    *,
 75    framework: str = "custom",
 76) -> None:
 77    """
 78    Attach a searchable external ID to the active Parlot session.
 79
 80    Stamps ``platform.ref.{kind}`` (and the primary triple when this is the
 81    first ref) on the live session span so the session can be found by that
 82    value in Parlot search / resolve.
 83
 84    Args:
 85        kind: Identifier type (e.g. ``"crm_ticket"``, ``"order_number"``,
 86            ``"call_sid"``).
 87        value: Unique identifier value (e.g. ``"TKT-9921"``).
 88        framework: Originating framework name. Defaults to ``"custom"``.
 89
 90    Example::
 91
 92        from parlot.instrumentation.livekit import add_platform_ref
 93
 94        add_platform_ref("crm_ticket", "TKT-9")
 95    """
 96    kind = str(kind or "").strip()
 97    value = str(value or "").strip()
 98    framework = str(framework or "custom").strip() or "custom"
 99    if not kind or not value:
100        return
101    span = get_active_session_span()
102    if span is None:
103        return
104    stamp_platform_refs(span, [(framework, kind, value)])

Attach a searchable external ID to the active Parlot session.

Stamps platform.ref.{kind} (and the primary triple when this is the first ref) on the live session span so the session can be found by that value in Parlot search / resolve.

Arguments:

kind: Identifier type (e.g. "crm_ticket", "order_number", "call_sid").
value: Unique identifier value (e.g. "TKT-9921").
framework: Originating framework name. Defaults to "custom".

Example::

from parlot.instrumentation.livekit import add_platform_ref

add_platform_ref("crm_ticket", "TKT-9")

@contextmanager

def human_escalation(label: str | None = None) -> Iterator[NoneType]: View Source

50@contextmanager
51def human_escalation(label: str | None = None) -> Iterator[None]:
52    """Mark the next participant who joins the active session as a human rep.
53
54    Args:
55        label: Optional role or team label for the incoming human representative.
56    """
57    token = _pending_escalation_label.set(label)
58    try:
59        yield
60    finally:
61        _pending_escalation_label.reset(token)

Mark the next participant who joins the active session as a human rep.

Arguments:

label: Optional role or team label for the incoming human representative.

def record_human_rep(participant_id: str, *, label: str | None = None) -> None: View Source

19def record_human_rep(participant_id: str, *, label: str | None = None) -> None:
20    """
21    Mark a participant as a human representative.
22
23    Stamps ``session.topology.agents`` on the active session span and registers
24    the participant for ``turn.participant_role=human_rep`` on future turns.
25
26    Args:
27        participant_id: Participant identifier within the room/session.
28        label: Optional human-readable name (e.g. ``"Tier 2 Escalation Desk"``).
29    """
30    participant_id = str(participant_id or "").strip()
31    if not participant_id:
32        return
33
34    state = get_active_session()
35    if state is not None:
36        state.human_rep_participant_ids.add(participant_id)
37        entry: dict[str, str] = {"id": participant_id, "role": "human_rep"}
38        if label:
39            entry["label"] = label
40        if not any(a.get("id") == participant_id for a in state.topology_agents):
41            state.topology_agents.append(entry)
42
43    span = get_active_session_span()
44    if span is not None and state is not None and state.topology_agents:
45        span.set_attribute(
46            ATTR_SESSION_TOPOLOGY_AGENTS, json.dumps(state.topology_agents)
47        )

Mark a participant as a human representative.

Stamps session.topology.agents on the active session span and registers the participant for turn.participant_role=human_rep on future turns.

Arguments:

participant_id: Participant identifier within the room/session.
label: Optional human-readable name (e.g. "Tier 2 Escalation Desk").

def set_session_attribute(key: str, value: str | int | float | bool) -> None: View Source

20def set_session_attribute(key: str, value: str | int | float | bool) -> None:
21    """
22    Stamp one custom attribute on the active session under ``session.metadata.*``.
23
24    Keys are normalized to ``session.metadata.<key>``. Values are stored as
25    strings on the live session span and remembered on session state so they
26    are also present on ``parlot.session.close``.
27
28    Args:
29        key: Attribute name. If not prefixed with ``session.metadata.``, the
30            prefix is added automatically.
31        value: Value to record (``str``, ``int``, ``float``, or ``bool``).
32    """
33    full_key = session_metadata_key(key)
34    if not full_key or full_key == ATTR_SESSION_METADATA_PREFIX:
35        return
36    str_value = value if isinstance(value, str) else str(value)
37
38    state = get_active_session()
39    if state is not None:
40        state.custom_metadata[full_key] = str_value
41
42    span = get_active_session_span()
43    if span is not None and hasattr(span, "set_attribute"):
44        span.set_attribute(full_key, str_value)

Stamp one custom attribute on the active session under session.metadata.*.

Keys are normalized to session.metadata.<key>. Values are stored as strings on the live session span and remembered on session state so they are also present on parlot.session.close.

Arguments:

key: Attribute name. If not prefixed with session.metadata., the prefix is added automatically.
value: Value to record (str, int, float, or bool).

def set_session_metadata(**pairs: str | int | float | bool) -> None: View Source

47def set_session_metadata(**pairs: str | int | float | bool) -> None:
48    """
49    Attach custom key/value metadata to the active Parlot session.
50
51    Each keyword becomes ``session.metadata.<name>`` on the session span and
52    appears in session detail in the Parlot UI.
53
54    Args:
55        **pairs: Keyword metadata pairs (values ``str``, ``int``, ``float``,
56            or ``bool``).
57
58    Example::
59
60        from parlot.instrumentation.livekit import set_session_metadata
61
62        set_session_metadata(order_id="12345", crm_ticket="TKT-9")
63    """
64    for key, value in pairs.items():
65        set_session_attribute(key, value)

Attach custom key/value metadata to the active Parlot session.

Each keyword becomes session.metadata.<name> on the session span and appears in session detail in the Parlot UI.

Arguments:

**pairs: Keyword metadata pairs (values str, int, float, or bool).

Example::

from parlot.instrumentation.livekit import set_session_metadata

set_session_metadata(order_id="12345", crm_ticket="TKT-9")

def stamp_platform_refs(span: Any, refs: list[tuple[str, str, str]]) -> None: View Source

28def stamp_platform_refs(
29    span: Any,
30    refs: list[tuple[str, str, str]],
31) -> None:
32    """Stamp ``platform.ref.*`` triples onto a span.
33
34    ``refs`` is a list of ``(framework, kind, value)`` tuples, e.g.
35    ``("livekit", "room_sid", "RM_abc")``. The first tuple is also written
36    to the canonical triple attributes (``platform.ref.framework/kind/value``)
37    so backends can pivot on a single primary ref.
38
39    Every tuple is also written as a flat ``platform.ref.{kind} = value`` key
40    for ingestion fallbacks.
41
42    Args:
43        span: A live OTel span (``set_attribute``) or a ``ReadableSpan`` with a
44            mutable ``_attributes`` dict.
45        refs: Non-empty list of reference triples.
46    """
47    if not refs:
48        return
49
50    def _write(key: str, value: str) -> None:
51        if hasattr(span, "set_attribute") and callable(span.set_attribute):
52            try:
53                span.set_attribute(key, value)
54                return
55            except Exception:
56                pass
57        try:
58            ParlotBaseProcessor._set(span, key, value)
59        except Exception:
60            return
61
62    fw, kind, val = refs[0]
63    _write(ATTR_PLATFORM_FRAMEWORK, fw)
64    _write(ATTR_PLATFORM_KIND, kind)
65    _write(ATTR_PLATFORM_VALUE, val)
66
67    for _fw_i, kind_i, val_i in refs:
68        _write(platform_ref_flat_key(kind_i), val_i)

Stamp platform.ref.* triples onto a span.

refs is a list of (framework, kind, value) tuples, e.g. ("livekit", "room_sid", "RM_abc"). The first tuple is also written to the canonical triple attributes (platform.ref.framework/kind/value) so backends can pivot on a single primary ref.

Every tuple is also written as a flat platform.ref.{kind} = value key for ingestion fallbacks.

Arguments:

span: A live OTel span (set_attribute) or a ReadableSpan with a mutable _attributes dict.
refs: Non-empty list of reference triples.