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, await_recording: bool = False, capture_logs: bool | list[str] | None = None, log_level: Optional[str] = None) -> parlot.core.ParlotContext: View Source

 30def parlotize(
 31    agent_id: str,
 32    *,
 33    endpoint: Optional[str] = None,
 34    api_key: Optional[str] = None,
 35    capture_genai_content: Optional[bool] = None,
 36    service_name: Optional[str] = None,
 37    tracer_provider: TracerProvider | None = None,
 38    auto_escalate_sip: bool = False,
 39    escalation_metadata_match: dict[str, str] | None = None,
 40    version: Optional[str] = None,
 41    record: bool | list[str] | None = None,
 42    await_recording: bool = False,
 43    capture_logs: bool | list[str] | None = None,
 44    log_level: Optional[str] = None,
 45) -> ParlotContext:
 46    """Parlotize LiveKit instrumentation and OTLP export.
 47
 48    Call before constructing ``AgentSession``. Builds a ``TracerProvider`` with
 49    an OTLP exporter, registers it with ``livekit.agents.telemetry``, fetches
 50    telemetry bootstrap when ``PARLOT_API_KEY`` is set, patches
 51    ``JobContext.connect`` / ``AgentSession.__init__``, and installs session
 52    event hooks.
 53
 54    Shared parameters (``agent_id``, plus keyword-only ``endpoint``, ``api_key``,
 55    ``version``, ``capture_genai_content``, ``capture_logs``, ``log_level``,
 56    ``service_name``, ``tracer_provider``) match every adapter — see
 57    ``parlot.core.ParlotizeProtocol``.
 58
 59    Under LiveKit ``dev`` / job workers, ``__main__`` is often LiveKit's IPC
 60    entrypoint, not your agent file — pass ``agent_id`` / ``version=``
 61    explicitly when you care about stable deployment identity.
 62
 63    Args:
 64        agent_id: Shared — required canonical ``session.agent_id``. Must be
 65            non-empty after stripping whitespace.
 66        endpoint: Shared — Parlot OTLP base URL (or ``PARLOT_ENDPOINT``).
 67        api_key: Shared — org API key (or ``PARLOT_API_KEY``).
 68        capture_genai_content: Shared — GenAI payload capture override.
 69            Precedence: job metadata → this kwarg → Settings → Generative AI → on.
 70        service_name: Shared — OTel ``service.name`` (defaults to ``agent_id``
 71            or ``"unknown"``).
 72        tracer_provider: Shared — existing ``TracerProvider``, or build one with
 73            Parlot's OTLP exporter and ``LiveKitGenAIProcessor``.
 74        auto_escalate_sip: When ``True``, mark the session escalated when a SIP
 75            participant joins the room.
 76        escalation_metadata_match: Participant metadata key/value pairs that
 77            classify joining participants as human representatives.
 78        version: Shared — ``gen_ai.agent.version``. Pass ``version=`` to set it;
 79            otherwise stamped as ``\"unknown\"``.
 80        record: Audio recording policy. Boolean or agent-id glob patterns
 81            (e.g. ``["support-*", "billing"]``). Precedence: LiveKit job
 82            metadata ``record`` → this kwarg → Settings → Recording.
 83        await_recording: When ``True`` and recording is enabled, ``ctx.connect()``
 84            bootstraps the Parlot session if needed and waits until Room Composite
 85            egress has started (recording timeline t=0) before returning. Use with
 86            ``await ctx.connect()`` before ``session.start()`` so the first agent
 87            greeting is not cut from the session recording. Default ``False``
 88            preserves low connect latency (egress may start after the greeting).
 89        capture_logs: Shared — session log capture (bool or globs). Precedence:
 90            job metadata → this kwarg → Settings → Logs → on.
 91        log_level: Shared — minimum level for session log capture.
 92
 93    Returns:
 94        The ``ParlotContext`` created for this process (or the prior one if
 95        already configured).
 96    """
 97    global _configured, _auto_escalate_sip, _escalation_metadata_match
 98    global _configured_agent_id, _configured_agent_version
 99    global _configured_record, _configured_await_recording
100    global _configured_capture_genai_content, _configured_capture_logs, _configured_log_level
101    global _parlot_context
102    if _configured:
103        logger.debug("parlot-instrumentation.livekit already configured — skipping")
104        assert _parlot_context is not None
105        return _parlot_context
106
107    # Note: do not skip under ``lk agent dev`` / ``start --dev``. Hot-reload is
108    # owned by the Go CLI; this Python process *is* the worker (LiveKit removed
109    # the in-process watch parent that used to spawn a child).
110
111    _auto_escalate_sip = auto_escalate_sip
112    _escalation_metadata_match = (
113        dict(escalation_metadata_match) if escalation_metadata_match else None
114    )
115    from parlot.core.parlotize import normalize_glob_or_bool
116    from parlot.core.provider import resolve_capture_genai_content
117
118    _configured_record = normalize_glob_or_bool(record)
119    _configured_await_recording = bool(await_recording)
120    _configured_capture_genai_content = resolve_capture_genai_content(capture_genai_content)
121    _configured_capture_logs = normalize_glob_or_bool(capture_logs)
122    _configured_log_level = log_level.strip().upper() if log_level else None
123
124    from ._agent_version import resolve_agent_version
125    _configured_agent_version = resolve_agent_version(version)
126
127    from parlot.core import base_parlotize
128
129    res = base_parlotize(
130        agent_id,
131        endpoint=endpoint,
132        api_key=api_key,
133        capture_genai_content=capture_genai_content,
134        tracer_provider=tracer_provider,
135        version=_configured_agent_version,
136        capture_logs=capture_logs,
137        log_level=log_level,
138    )
139    _parlot_context = res.context
140    _configured_agent_id = res.agent_id
141
142    from ._recording_guard import (
143        current_job_capture_logs_metadata,
144        livekit_session_log_fields,
145        recording_agent_id_from_ctx,
146    )
147
148    def _agent_id_resolver() -> str:
149        try:
150            from livekit.agents.job import get_job_context
151
152            ctx = get_job_context()
153            if ctx is not None:
154                return recording_agent_id_from_ctx(ctx)
155        except Exception:
156            pass
157        return _configured_agent_id or ""
158
159    res.context.session_logs.set_resolvers(
160        session_resolver=livekit_session_log_fields,
161        agent_id_resolver=_agent_id_resolver,
162        metadata_resolver=current_job_capture_logs_metadata,
163    )
164
165    tracer_provider = res.tracer_provider
166    if tracer_provider is None:
167        tracer_provider = _build_provider(
168            endpoint=res.endpoint,
169            api_key=res.api_key,
170            service_name=service_name,
171            service_version=_configured_agent_version or None,
172            context=res.context,
173        )
174
175    from parlot.core.processor import assert_sync_span_processors
176
177    from ._session import set_span_context_attach_enabled
178
179    attach_ok = assert_sync_span_processors(tracer_provider)
180    set_span_context_attach_enabled(attach_ok)
181
182    _register_with_livekit(tracer_provider)
183    _patch_agent_session_with_tracer(tracer_provider)
184    _patch_job_context_connect()
185    _install_telemetry_compare()
186
187    _configured = True
188    from parlot.core.parlotize import log_parlotize_ready
189
190    bootstrap_ok = res.context.runtime is not None
191    # Treat missing endpoint/key as not-ready (same as prior warning path).
192    if not res.endpoint or not res.api_key:
193        bootstrap_ok = False
194    log_parlotize_ready(
195        "livekit",
196        agent_id=res.agent_id,
197        endpoint=res.endpoint or "",
198        api_key=res.api_key or "",
199        bootstrap_ok=bootstrap_ok,
200        log=logger,
201    )
202    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.
await_recording: When True and recording is enabled, ctx.connect() bootstraps the Parlot session if needed and waits until Room Composite egress has started (recording timeline t=0) before returning. Use with await ctx.connect() before session.start() so the first agent greeting is not cut from the session recording. Default False preserves low connect latency (egress may start after the greeting).
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

491def install_session_hooks(
492    session: Any,
493    processor: "LiveKitGenAIProcessor",
494    tracer: "Tracer",
495) -> None:
496    """Install all AgentSession event listeners for Parlot instrumentation.
497
498    ``parlotize()`` patches ``AgentSession.__init__`` to invoke this
499    automatically. Call directly only when manually instantiating unpatched
500    sessions.
501
502    Args:
503        session: LiveKit ``AgentSession`` instance to instrument.
504        processor: Span processor that owns session/turn enrichment state.
505        tracer: OpenTelemetry tracer used for Parlot contract spans.
506    """
507    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

 89def add_platform_ref(
 90    kind: str,
 91    value: str,
 92    *,
 93    framework: str = "custom",
 94) -> None:
 95    """
 96    Attach a searchable external ID to the active Parlot session.
 97
 98    Stamps ``platform.ref.{kind}`` (and the primary triple when this is the
 99    first ref) on the live session span so the session can be found by that
100    value in Parlot search / resolve.
101
102    Args:
103        kind: Identifier type (e.g. ``"crm_ticket"``, ``"order_number"``,
104            ``"call_sid"``).
105        value: Unique identifier value (e.g. ``"TKT-9921"``).
106        framework: Originating framework name. Defaults to ``"custom"``.
107
108    Example::
109
110        from parlot.instrumentation.livekit import add_platform_ref
111
112        add_platform_ref("crm_ticket", "TKT-9")
113    """
114    kind = str(kind or "").strip()
115    value = str(value or "").strip()
116    framework = str(framework or "custom").strip() or "custom"
117    if not kind or not value:
118        return
119    span = get_active_session_span()
120    if span is None:
121        return
122    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

46def stamp_platform_refs(
47    span: Any,
48    refs: list[tuple[str, str, str]],
49) -> None:
50    """Stamp ``platform.ref.*`` triples onto a span.
51
52    ``refs`` is a list of ``(framework, kind, value)`` tuples, e.g.
53    ``("livekit", "room_sid", "RM_abc")``. The first tuple is also written
54    to the canonical triple attributes (``platform.ref.framework/kind/value``)
55    so backends can pivot on a single primary ref.
56
57    Every tuple is also written as a flat ``platform.ref.{kind} = value`` key
58    for ingestion fallbacks.
59
60    Args:
61        span: A live OTel span (``set_attribute``) or a ``ReadableSpan`` with a
62            mutable ``_attributes`` dict.
63        refs: Non-empty list of reference triples.
64    """
65    if not refs:
66        return
67
68    def _write(key: str, value: str) -> None:
69        if hasattr(span, "set_attribute") and callable(span.set_attribute):
70            try:
71                span.set_attribute(key, value)
72                return
73            except Exception:
74                pass
75        try:
76            ParlotBaseProcessor._set(span, key, value)
77        except Exception:
78            return
79
80    fw, kind, val = refs[0]
81    _write(ATTR_PLATFORM_FRAMEWORK, fw)
82    _write(ATTR_PLATFORM_KIND, kind)
83    _write(ATTR_PLATFORM_VALUE, val)
84
85    for _fw_i, kind_i, val_i in refs:
86        _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.