parlot.core
parlot-core: shared semantic conventions, base processor, and utilities.
1"""parlot-core: shared semantic conventions, base processor, and utilities.""" 2 3from .attrs import * # noqa: F401, F403 — re-export all attribute constants 4from .context import ParlotContext 5from .escalation import human_escalation, record_human_rep 6from .metadata import set_session_attribute, set_session_metadata 7from .platform_refs import add_platform_ref, stamp_framework_refs, stamp_platform_refs 8from .processor import ParlotBaseProcessor, assert_sync_span_processors 9from .provider import ( 10 adopt_existing_tracer_provider, 11 attach_span_processor_if_absent, 12 build_filtered_tracer_provider, 13 build_otlp_http_exporter, 14 build_parlot_client_headers, 15 build_resource, 16 build_tracer_provider, 17 resolve_api_key, 18 resolve_capture_genai_content, 19 resolve_endpoint, 20 try_set_global_tracer_provider, 21 HEADER_SDK_NAME, 22 HEADER_SDK_VERSION, 23 HEADER_INGESTION_VERSION, 24 INGESTION_PROTOCOL_VERSION, 25) 26from .bootstrap import fetch_telemetry_bootstrap 27from .parlotize import ( 28 BaseParlotizeResult, 29 ParlotizeProtocol, 30 base_parlotize, 31 configure_parlot_logging, 32 log_parlotize_ready, 33 normalize_glob_or_bool, 34) 35from .genai_content_capture import should_capture_genai_content 36from .export import ExportFilterSpanExporter 37from .intent import derive_intent 38from .logs_capture import should_capture_logs 39from .recording import should_record 40from .recording_source import RecordingSource, stamp_external_recording 41from .recording_stamp import ( 42 stamp_recording_disabled, 43 stamp_recording_error, 44 stamp_recording_started, 45 stamp_recording_uploaded, 46) 47from .recording_timeline import ( 48 media_segments_from_speech, 49 set_recording_anchor_wall_ms, 50 speech_wall_ms_from_span, 51) 52from .recording_upload import ( 53 complete_recording, 54 fetch_upload_grant, 55 put_bytes_to_presigned_url, 56) 57from .metrics import ( 58 LONG_LIVED_METRICS_EXPORT_INTERVAL_MS, 59 ParlotMetricsRecorder, 60 QuietOTLPMetricExporter, 61 UsageMetricsRecorder, 62 build_parlot_meter_provider, 63) 64from .runtime import ParlotRuntimeContext, runtime_from_bootstrap 65from .sdk_version import resolve_parlot_sdk_version, stamp_session_sdk_version 66from .session import ( 67 SessionState, 68 clear_active_session, 69 get_active_session, 70 get_active_session_span, 71 session_owned, 72 set_active_session, 73) 74from .session_lifecycle import ( 75 DEFAULT_CLOSE_FLUSH_ATTEMPT_TIMEOUT_MS, 76 DEFAULT_CLOSE_FLUSH_BACKOFFS_S, 77 DEFAULT_CLOSE_FLUSH_BUDGET_S, 78 emit_parlot_session_close, 79 end_session_span_ok, 80 force_flush_tracer_provider, 81 force_flush_with_budget, 82 start_parlot_session_span, 83) 84from .session_logs import SessionLogsCollector 85from .session_registry import ConversationSessionRegistry 86from .span_rename import apply_genai_span_rename, remap_native_span_name 87from .topology import SessionTopology 88from .turn_emit import emit_parlot_turn_root, stamp_turn_utterance_text 89 90__all__ = [ 91 "BaseParlotizeResult", 92 "ConversationSessionRegistry", 93 "DEFAULT_CLOSE_FLUSH_ATTEMPT_TIMEOUT_MS", 94 "DEFAULT_CLOSE_FLUSH_BACKOFFS_S", 95 "DEFAULT_CLOSE_FLUSH_BUDGET_S", 96 "ParlotizeProtocol", 97 "ParlotContext", 98 "ParlotRuntimeContext", 99 "SessionLogsCollector", 100 "apply_genai_span_rename", 101 "attach_span_processor_if_absent", 102 "base_parlotize", 103 "build_filtered_tracer_provider", 104 "configure_parlot_logging", 105 "emit_parlot_session_close", 106 "emit_parlot_turn_root", 107 "end_session_span_ok", 108 "ExportFilterSpanExporter", 109 "force_flush_tracer_provider", 110 "force_flush_with_budget", 111 "LONG_LIVED_METRICS_EXPORT_INTERVAL_MS", 112 "log_parlotize_ready", 113 "normalize_glob_or_bool", 114 "ParlotBaseProcessor", 115 "ParlotMetricsRecorder", 116 "QuietOTLPMetricExporter", 117 "UsageMetricsRecorder", 118 "assert_sync_span_processors", 119 "SessionState", 120 "SessionTopology", 121 "add_platform_ref", 122 "adopt_existing_tracer_provider", 123 "build_otlp_http_exporter", 124 "build_parlot_client_headers", 125 "build_parlot_meter_provider", 126 "build_resource", 127 "build_tracer_provider", 128 "HEADER_SDK_NAME", 129 "HEADER_SDK_VERSION", 130 "HEADER_INGESTION_VERSION", 131 "INGESTION_PROTOCOL_VERSION", 132 "clear_active_session", 133 "derive_intent", 134 "fetch_telemetry_bootstrap", 135 "get_active_session", 136 "get_active_session_span", 137 "human_escalation", 138 "record_human_rep", 139 "remap_native_span_name", 140 "resolve_api_key", 141 "resolve_capture_genai_content", 142 "resolve_endpoint", 143 "resolve_parlot_sdk_version", 144 "runtime_from_bootstrap", 145 "session_owned", 146 "set_active_session", 147 "set_session_attribute", 148 "set_session_metadata", 149 "should_capture_genai_content", 150 "should_record", 151 "RecordingSource", 152 "stamp_external_recording", 153 "stamp_recording_disabled", 154 "stamp_recording_error", 155 "stamp_recording_started", 156 "stamp_recording_uploaded", 157 "set_recording_anchor_wall_ms", 158 "media_segments_from_speech", 159 "speech_wall_ms_from_span", 160 "fetch_upload_grant", 161 "complete_recording", 162 "put_bytes_to_presigned_url", 163 "should_capture_logs", 164 "stamp_framework_refs", 165 "stamp_platform_refs", 166 "stamp_session_sdk_version", 167 "stamp_turn_utterance_text", 168 "start_parlot_session_span", 169 "try_set_global_tracer_provider", 170]
@dataclass
class
BaseParlotizeResult:
134@dataclass 135class BaseParlotizeResult: 136 """Resolved context from ``base_parlotize`` for framework adapters.""" 137 138 context: ParlotContext 139 endpoint: str 140 api_key: str 141 tracer_provider: Any 142 agent_id: str 143 agent_version: str 144 capture_genai_content: Optional[bool] 145 capture_logs: bool | list[str] | None 146 log_level: Optional[str]
Resolved context from base_parlotize for framework adapters.
BaseParlotizeResult( context: ParlotContext, endpoint: str, api_key: str, tracer_provider: Any, agent_id: str, agent_version: str, capture_genai_content: Optional[bool], capture_logs: bool | list[str] | None, log_level: Optional[str])
context: ParlotContext
@dataclass
class
ConversationSessionRegistry26@dataclass 27class ConversationSessionRegistry(Generic[S]): 28 """Keyed map of open ``parlot.session`` spans + framework session state. 29 30 Used by conversation/thread-scoped adapters (Pipecat, LangGraph). LiveKit 31 keeps its job ContextVar bootstrap and does not use this registry. 32 """ 33 34 _sessions: dict[str, tuple[Span, S]] = field(default_factory=dict) 35 _atexit_registered: bool = False 36 _atexit_reason: str = "process_exit" 37 38 def get(self, key: str) -> tuple[Span, S] | None: 39 return self._sessions.get(key) 40 41 def get_state(self, key: str) -> S | None: 42 pair = self._sessions.get(key) 43 return pair[1] if pair else None 44 45 def keys(self) -> list[str]: 46 return list(self._sessions) 47 48 def values(self) -> list[tuple[Span, S]]: 49 return list(self._sessions.values()) 50 51 def __len__(self) -> int: 52 return len(self._sessions) 53 54 def __bool__(self) -> bool: 55 return bool(self._sessions) 56 57 def last_opened_state(self) -> S | None: 58 """Insertion-order last session (single-bot local runners).""" 59 if not self._sessions: 60 return None 61 _span, state = next(reversed(list(self._sessions.values()))) 62 return state 63 64 def put(self, key: str, span: Span, state: S) -> None: 65 self._sessions[key] = (span, state) 66 set_active_session(span, state) 67 68 def ensure( 69 self, 70 key: str, 71 factory: Callable[[], tuple[Span, S] | None], 72 ) -> S | None: 73 """Return existing session (rebinding active) or create via ``factory``. 74 75 ``factory`` returns ``(span, state)`` or ``None`` when it cannot open 76 (e.g. no tracer). 77 """ 78 existing = self._sessions.get(key) 79 if existing is not None: 80 span, state = existing 81 set_active_session(span, state) 82 return state 83 created = factory() 84 if created is None: 85 return None 86 span, state = created 87 self.put(key, span, state) 88 return state 89 90 def pop(self, key: str) -> tuple[Span, S] | None: 91 return self._sessions.pop(key, None) 92 93 def close( 94 self, 95 key: str, 96 *, 97 closer: Callable[[Span, S], None], 98 clear_active: bool = True, 99 ) -> bool: 100 """Pop ``key`` and run ``closer(session_span, state)``. 101 102 Returns True when a session was closed. 103 """ 104 pair = self._sessions.pop(key, None) 105 if pair is None: 106 return False 107 session_span, state = pair 108 closer(session_span, state) 109 if clear_active: 110 active = get_active_session() 111 if active is state: 112 clear_active_session() 113 return True 114 115 def end_and_clear(self, session_span: Span, state: S) -> None: 116 """End the session span OK and clear active when it matches ``state``.""" 117 end_session_span_ok(session_span) 118 active = get_active_session() 119 if active is state: 120 clear_active_session() 121 122 def flush_all(self, closer: Callable[[str], None]) -> None: 123 for key in list(self._sessions): 124 closer(key) 125 126 def ensure_atexit(self, flush_all: Callable[[], None]) -> None: 127 if self._atexit_registered: 128 return 129 atexit.register(flush_all) 130 self._atexit_registered = True
Keyed map of open parlot.session spans + framework session state.
Used by conversation/thread-scoped adapters (Pipecat, LangGraph). LiveKit keeps its job ContextVar bootstrap and does not use this registry.
ConversationSessionRegistry( _sessions: dict[str, tuple[opentelemetry.trace.span.Span, ~S]] = <factory>, _atexit_registered: bool = False, _atexit_reason: str = 'process_exit')
def
last_opened_state(self) -> Optional[~S]:
57 def last_opened_state(self) -> S | None: 58 """Insertion-order last session (single-bot local runners).""" 59 if not self._sessions: 60 return None 61 _span, state = next(reversed(list(self._sessions.values()))) 62 return state
Insertion-order last session (single-bot local runners).
def
ensure( self, key: str, factory: Callable[[], tuple[opentelemetry.trace.span.Span, ~S] | None]) -> Optional[~S]:
68 def ensure( 69 self, 70 key: str, 71 factory: Callable[[], tuple[Span, S] | None], 72 ) -> S | None: 73 """Return existing session (rebinding active) or create via ``factory``. 74 75 ``factory`` returns ``(span, state)`` or ``None`` when it cannot open 76 (e.g. no tracer). 77 """ 78 existing = self._sessions.get(key) 79 if existing is not None: 80 span, state = existing 81 set_active_session(span, state) 82 return state 83 created = factory() 84 if created is None: 85 return None 86 span, state = created 87 self.put(key, span, state) 88 return state
Return existing session (rebinding active) or create via factory.
factory returns (span, state) or None when it cannot open
(e.g. no tracer).
def
close( self, key: str, *, closer: Callable[[opentelemetry.trace.span.Span, ~S], NoneType], clear_active: bool = True) -> bool:
93 def close( 94 self, 95 key: str, 96 *, 97 closer: Callable[[Span, S], None], 98 clear_active: bool = True, 99 ) -> bool: 100 """Pop ``key`` and run ``closer(session_span, state)``. 101 102 Returns True when a session was closed. 103 """ 104 pair = self._sessions.pop(key, None) 105 if pair is None: 106 return False 107 session_span, state = pair 108 closer(session_span, state) 109 if clear_active: 110 active = get_active_session() 111 if active is state: 112 clear_active_session() 113 return True
Pop key and run closer(session_span, state).
Returns True when a session was closed.
def
end_and_clear(self, session_span: opentelemetry.trace.span.Span, state: ~S) -> None:
115 def end_and_clear(self, session_span: Span, state: S) -> None: 116 """End the session span OK and clear active when it matches ``state``.""" 117 end_session_span_ok(session_span) 118 active = get_active_session() 119 if active is state: 120 clear_active_session()
End the session span OK and clear active when it matches state.
DEFAULT_CLOSE_FLUSH_ATTEMPT_TIMEOUT_MS =
4000
DEFAULT_CLOSE_FLUSH_BACKOFFS_S =
(1.0, 3.0, 6.0)
DEFAULT_CLOSE_FLUSH_BUDGET_S =
12.0
@runtime_checkable
class
ParlotizeProtocol80@runtime_checkable 81class ParlotizeProtocol(Protocol): 82 """Common parameters every adapter ``parlotize()`` must accept. 83 84 Framework packages may add extra keyword-only parameters (e.g. LiveKit 85 ``record=`` / ``auto_escalate_sip=``). 86 """ 87 88 def __call__( 89 self, 90 agent_id: str, 91 *, 92 endpoint: Optional[str] = None, 93 api_key: Optional[str] = None, 94 capture_genai_content: Optional[bool] = None, 95 service_name: Optional[str] = None, 96 tracer_provider: TracerProvider | None = None, 97 version: Optional[str] = None, 98 capture_logs: bool | list[str] | None = None, 99 log_level: Optional[str] = None, 100 **kwargs: Any, 101 ) -> ParlotContext: 102 """Shared parameter surface for every adapter ``parlotize()``. 103 104 Args: 105 agent_id: Required canonical deployment identity stamped on 106 ``session.agent_id``. Must be non-empty after stripping whitespace. 107 endpoint: Parlot OTLP collector base URL (e.g. 108 ``https://ingest.parlot.ai``). Spans export to 109 ``{endpoint}/v1/traces``. If omitted, reads ``PARLOT_ENDPOINT``. 110 api_key: Org-scoped API key minted in Parlot **Settings → API Keys**. 111 If omitted, reads ``PARLOT_API_KEY``. Required for remote telemetry 112 bootstrap and recording grants. 113 capture_genai_content: Process-wide override for LLM message bodies and 114 tool input/output payloads. If ``False``, payloads are omitted while 115 preserving span durations, tokens, and turn text. Precedence: job 116 metadata → this kwarg → Settings → Generative AI (default: on). 117 service_name: OpenTelemetry resource ``service.name``. Defaults to 118 ``agent_id`` or a framework-specific fallback. 119 tracer_provider: Existing OpenTelemetry ``TracerProvider`` to adopt. If 120 omitted, adapters build one with Parlot's OTLP exporter (or adopt an 121 already-registered provider when another adapter configured first). 122 version: Deployment version stamped on ``gen_ai.agent.version``. 123 Pass ``version=`` to set it; otherwise ``\"unknown\"``. 124 capture_logs: Intercept Python ``logging`` during active sessions and 125 stream to the session Logs tab. Boolean or agent-id glob patterns. 126 Precedence: job metadata → this kwarg → Settings → Logs (default: on). 127 log_level: Minimum level for session log capture (e.g. ``"INFO"``, 128 ``"WARNING"``). Defaults to ``"INFO"``. 129 **kwargs: Framework-specific options (ignored by the shared surface). 130 """ 131 ...
Common parameters every adapter parlotize() must accept.
Framework packages may add extra keyword-only parameters (e.g. LiveKit
record= / auto_escalate_sip=).
ParlotizeProtocol(*args, **kwargs)
1739def _no_init_or_replace_init(self, *args, **kwargs): 1740 cls = type(self) 1741 1742 if cls._is_protocol: 1743 raise TypeError('Protocols cannot be instantiated') 1744 1745 # Already using a custom `__init__`. No need to calculate correct 1746 # `__init__` to call. This can lead to RecursionError. See bpo-45121. 1747 if cls.__init__ is not _no_init_or_replace_init: 1748 return 1749 1750 # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`. 1751 # The first instantiation of the subclass will call `_no_init_or_replace_init` which 1752 # searches for a proper new `__init__` in the MRO. The new `__init__` 1753 # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent 1754 # instantiation of the protocol subclass will thus use the new 1755 # `__init__` and no longer call `_no_init_or_replace_init`. 1756 for base in cls.__mro__: 1757 init = base.__dict__.get('__init__', _no_init_or_replace_init) 1758 if init is not _no_init_or_replace_init: 1759 cls.__init__ = init 1760 break 1761 else: 1762 # should not happen 1763 cls.__init__ = object.__init__ 1764 1765 cls.__init__(self, *args, **kwargs)
@dataclass
class
ParlotContext:
13@dataclass 14class ParlotContext: 15 """Owns bootstrap runtime and process-local collectors for one SDK instance.""" 16 17 runtime: Optional[ParlotRuntimeContext] = None 18 session_logs: SessionLogsCollector = field(default_factory=SessionLogsCollector) 19 20 def __post_init__(self) -> None: 21 self.session_logs.bind_context(self) 22 23 def shutdown(self) -> None: 24 self.session_logs.shutdown() 25 self.runtime = None
Owns bootstrap runtime and process-local collectors for one SDK instance.
ParlotContext( runtime: Optional[ParlotRuntimeContext] = None, session_logs: SessionLogsCollector = <factory>)
session_logs: SessionLogsCollector
@dataclass(frozen=True)
class
ParlotRuntimeContext:
12@dataclass(frozen=True) 13class ParlotRuntimeContext: 14 endpoint: str 15 api_key: str 16 tenant_id: str 17 content_bucket: str 18 r2_endpoint: str 19 recording_globs: tuple[str, ...] = () 20 recording_agents: tuple[tuple[str, bool], ...] = () 21 logs_globs: tuple[str, ...] = () 22 logs_agents: tuple[tuple[str, bool], ...] = () 23 logs_agent_min_levels: tuple[tuple[str, str], ...] = () 24 logs_min_level: str = DEFAULT_LOGS_MIN_LEVEL 25 # True when bootstrap JSON included a ``logs`` object (even empty). 26 logs_policy_present: bool = False 27 capture_genai_content_globs: tuple[str, ...] = () 28 capture_genai_content_agents: tuple[tuple[str, bool], ...] = () 29 # True when bootstrap JSON included a ``capture_genai_content`` object (even empty). 30 capture_genai_content_policy_present: bool = False 31 32 def recording_agents_map(self) -> dict[str, bool]: 33 return dict(self.recording_agents) 34 35 def logs_agents_map(self) -> dict[str, bool]: 36 return dict(self.logs_agents) 37 38 def logs_agent_min_levels_map(self) -> dict[str, str]: 39 return dict(self.logs_agent_min_levels) 40 41 def capture_genai_content_agents_map(self) -> dict[str, bool]: 42 return dict(self.capture_genai_content_agents)
ParlotRuntimeContext( endpoint: str, api_key: str, tenant_id: str, content_bucket: str, r2_endpoint: str, recording_globs: tuple[str, ...] = (), recording_agents: tuple[tuple[str, bool], ...] = (), logs_globs: tuple[str, ...] = (), logs_agents: tuple[tuple[str, bool], ...] = (), logs_agent_min_levels: tuple[tuple[str, str], ...] = (), logs_min_level: str = 'INFO', logs_policy_present: bool = False, capture_genai_content_globs: tuple[str, ...] = (), capture_genai_content_agents: tuple[tuple[str, bool], ...] = (), capture_genai_content_policy_present: bool = False)
class
SessionLogsCollector:
241class SessionLogsCollector: 242 """Per-``ParlotContext`` session log capture pipeline.""" 243 244 def __init__(self) -> None: 245 self._lock = threading.Lock() 246 self._buffer: deque[SessionLogRecord] = deque(maxlen=_MAX_BUFFER) 247 self._consecutive_failures = 0 248 self._circuit_open_until = 0.0 249 self._last_flush_at = 0.0 250 self._endpoint = "" 251 self._api_key = "" 252 self._flush_thread: threading.Thread | None = None 253 self._stop = threading.Event() 254 self._handler: Optional[SessionLogHandler] = None 255 self._installed = False 256 self._atexit_registered = False 257 self._capture_logs_config: CaptureLogsConfig = None 258 self._capture_logs_level: Optional[str] = None 259 self._session_resolver: Optional[SessionResolver] = None 260 self._agent_id_resolver: Optional[Callable[[], str]] = None 261 self._metadata_resolver: Optional[Callable[[], Optional[bool]]] = None 262 self._context_ref: Optional[weakref.ref[ParlotContext]] = None 263 264 def bind_context(self, context: ParlotContext) -> None: 265 self._context_ref = weakref.ref(context) 266 267 def _runtime(self): 268 ctx = self._context_ref() if self._context_ref is not None else None 269 return ctx.runtime if ctx is not None else None 270 271 def set_capture_logs_config( 272 self, 273 capture_logs: CaptureLogsConfig = None, 274 *, 275 log_level: Optional[str] = None, 276 ) -> None: 277 self._capture_logs_config = capture_logs 278 if log_level is not None: 279 self._capture_logs_level = normalize_log_level(log_level) 280 281 def set_resolvers( 282 self, 283 *, 284 session_resolver: Optional[SessionResolver] = None, 285 agent_id_resolver: Optional[Callable[[], str]] = None, 286 metadata_resolver: Optional[Callable[[], Optional[bool]]] = None, 287 ) -> None: 288 """Optional framework hooks (e.g. LiveKit job bootstrap fallback).""" 289 if session_resolver is not None: 290 self._session_resolver = session_resolver 291 if agent_id_resolver is not None: 292 self._agent_id_resolver = agent_id_resolver 293 if metadata_resolver is not None: 294 self._metadata_resolver = metadata_resolver 295 296 def _resolve_min_level(self) -> str: 297 if self._capture_logs_level: 298 return self._capture_logs_level 299 runtime = self._runtime() 300 if runtime is not None: 301 agent_id = "" 302 if self._agent_id_resolver is not None: 303 try: 304 agent_id = self._agent_id_resolver() or "" 305 except Exception: 306 agent_id = "" 307 if agent_id: 308 agent_levels = runtime.logs_agent_min_levels_map() 309 if agent_id in agent_levels: 310 return normalize_log_level(agent_levels[agent_id]) 311 if runtime.logs_min_level: 312 return normalize_log_level(runtime.logs_min_level) 313 return DEFAULT_LOGS_MIN_LEVEL 314 315 def capture_logs_enabled_for_agent(self, agent_name: str = "") -> bool: 316 runtime = self._runtime() 317 metadata = None 318 if self._metadata_resolver is not None: 319 try: 320 metadata = self._metadata_resolver() 321 except Exception: 322 metadata = None 323 name = agent_name 324 if not name and self._agent_id_resolver is not None: 325 try: 326 name = self._agent_id_resolver() or "" 327 except Exception: 328 name = "" 329 return should_capture_logs( 330 name, 331 metadata_capture_logs=metadata, 332 capture_logs_config=self._capture_logs_config, 333 bootstrap_globs=list(runtime.logs_globs) if runtime else None, 334 bootstrap_agents=runtime.logs_agents_map() if runtime else None, 335 bootstrap_present=runtime is not None and runtime.logs_policy_present, 336 ) 337 338 def handle_record( 339 self, handler: logging.Handler, record: logging.LogRecord 340 ) -> None: 341 name = record.name or "" 342 if any(name == p or name.startswith(p + ".") for p in _SKIP_LOGGER_PREFIXES): 343 return 344 level_name = record.levelname or "INFO" 345 if not level_at_least(level_name, self._resolve_min_level()): 346 return 347 if not self.capture_logs_enabled_for_agent(): 348 return 349 session = _resolve_session_fields(self._session_resolver) 350 if not session or not session.get("session_id"): 351 return 352 message = handler.format(record) if handler.formatter else record.getMessage() 353 if not isinstance(message, str): 354 message = str(message) 355 trace_id, span_id = _current_trace_span() 356 ts = datetime.fromtimestamp(record.created, tz=timezone.utc).strftime( 357 "%Y-%m-%d %H:%M:%S.%f" 358 )[:-3] 359 attrs: dict[str, str] = {} 360 if record.pathname: 361 attrs["pathname"] = str(record.pathname)[:512] 362 if record.lineno: 363 attrs["lineno"] = str(record.lineno) 364 if record.funcName: 365 attrs["funcName"] = str(record.funcName)[:128] 366 event = SessionLogRecord( 367 session_id=str(session["session_id"]), 368 conversation_id=str(session.get("conversation_id") or session["session_id"]), 369 ts=ts, 370 level=normalize_log_level(level_name), 371 logger_name=name, 372 message=message, 373 turn_index=int(session.get("turn_index") or 0), 374 trace_id=trace_id, 375 span_id=span_id, 376 attributes=attrs, 377 ) 378 with self._lock: 379 self._buffer.append(event) 380 381 def init(self, *, endpoint: str = "", api_key: str = "") -> None: 382 """Install root handler + start background flusher.""" 383 with self._lock: 384 self._endpoint = (endpoint or "").rstrip("/") 385 self._api_key = api_key or "" 386 if self._handler is None: 387 self._handler = SessionLogHandler(self) 388 self._handler.setLevel(logging.DEBUG) 389 self._handler.setFormatter(logging.Formatter("%(message)s")) 390 if not self._installed: 391 logging.root.addHandler(self._handler) 392 self._installed = True 393 if self._flush_thread is None or not self._flush_thread.is_alive(): 394 self._stop.clear() 395 self._flush_thread = threading.Thread( 396 target=self._flush_loop, 397 name="parlot-session-logs-flush", 398 daemon=True, 399 ) 400 self._flush_thread.start() 401 if not self._atexit_registered: 402 atexit.register(self.shutdown) 403 self._atexit_registered = True 404 405 def drain(self) -> list[SessionLogRecord]: 406 with self._lock: 407 events = list(self._buffer) 408 self._buffer.clear() 409 return events 410 411 def snapshot(self) -> list[SessionLogRecord]: 412 with self._lock: 413 return list(self._buffer) 414 415 def _circuit_open(self) -> bool: 416 return time.time() < self._circuit_open_until 417 418 def _flush_loop(self) -> None: 419 while not self._stop.wait(_FLUSH_INTERVAL_S): 420 try: 421 self._try_flush() 422 except Exception: 423 continue 424 425 def _try_flush(self, *, force: bool = False) -> None: 426 if self._circuit_open(): 427 with self._lock: 428 self._buffer.clear() 429 return 430 if not self._endpoint or not self._api_key: 431 return 432 433 if not force: 434 now = time.time() 435 if now - self._last_flush_at < _FLUSH_INTERVAL_S * 0.5: 436 return 437 438 events = self.drain() 439 if not events: 440 return 441 442 ok = self._post_events(events) 443 self._last_flush_at = time.time() 444 if ok: 445 self._consecutive_failures = 0 446 return 447 448 self._consecutive_failures += 1 449 if self._consecutive_failures >= _CIRCUIT_FAILURES: 450 self._circuit_open_until = time.time() + _CIRCUIT_COOLDOWN_S 451 self._consecutive_failures = 0 452 453 def _post_events(self, events: list[SessionLogRecord]) -> bool: 454 try: 455 import httpx 456 except Exception: 457 return False 458 459 url = f"{self._endpoint}/v1/logs" 460 try: 461 payload = _encode_otlp_logs(events[:_MAX_BATCH]) 462 except Exception: 463 return False 464 if not payload: 465 return True 466 from parlot.core.provider import build_parlot_client_headers 467 468 headers = build_parlot_client_headers(self._api_key) 469 headers["Content-Type"] = "application/x-protobuf" 470 try: 471 with httpx.Client(timeout=3.0) as client: 472 res = client.post( 473 url, 474 content=payload, 475 headers=headers, 476 ) 477 return 200 <= res.status_code < 300 478 except Exception: 479 return False 480 481 def shutdown(self) -> None: 482 """Stop capture and push any remaining buffered logs. Never raises.""" 483 try: 484 self._stop.set() 485 thread = self._flush_thread 486 if ( 487 thread is not None 488 and thread.is_alive() 489 and thread is not threading.current_thread() 490 ): 491 thread.join(timeout=1.0) 492 self._flush_thread = None 493 with self._lock: 494 if self._installed and self._handler is not None: 495 try: 496 logging.root.removeHandler(self._handler) 497 except Exception: 498 pass 499 self._installed = False 500 self._try_flush(force=True) 501 except Exception: 502 return
Per-ParlotContext session log capture pipeline.
def
set_capture_logs_config( self, capture_logs: Union[bool, Sequence[str], NoneType] = None, *, log_level: Optional[str] = None) -> None:
def
set_resolvers( self, *, session_resolver: Optional[Callable[[], Optional[dict[str, Any]]]] = None, agent_id_resolver: Optional[Callable[[], str]] = None, metadata_resolver: Optional[Callable[[], Optional[bool]]] = None) -> None:
281 def set_resolvers( 282 self, 283 *, 284 session_resolver: Optional[SessionResolver] = None, 285 agent_id_resolver: Optional[Callable[[], str]] = None, 286 metadata_resolver: Optional[Callable[[], Optional[bool]]] = None, 287 ) -> None: 288 """Optional framework hooks (e.g. LiveKit job bootstrap fallback).""" 289 if session_resolver is not None: 290 self._session_resolver = session_resolver 291 if agent_id_resolver is not None: 292 self._agent_id_resolver = agent_id_resolver 293 if metadata_resolver is not None: 294 self._metadata_resolver = metadata_resolver
Optional framework hooks (e.g. LiveKit job bootstrap fallback).
def
capture_logs_enabled_for_agent(self, agent_name: str = '') -> bool:
315 def capture_logs_enabled_for_agent(self, agent_name: str = "") -> bool: 316 runtime = self._runtime() 317 metadata = None 318 if self._metadata_resolver is not None: 319 try: 320 metadata = self._metadata_resolver() 321 except Exception: 322 metadata = None 323 name = agent_name 324 if not name and self._agent_id_resolver is not None: 325 try: 326 name = self._agent_id_resolver() or "" 327 except Exception: 328 name = "" 329 return should_capture_logs( 330 name, 331 metadata_capture_logs=metadata, 332 capture_logs_config=self._capture_logs_config, 333 bootstrap_globs=list(runtime.logs_globs) if runtime else None, 334 bootstrap_agents=runtime.logs_agents_map() if runtime else None, 335 bootstrap_present=runtime is not None and runtime.logs_policy_present, 336 )
def
handle_record(self, handler: logging.Handler, record: logging.LogRecord) -> None:
338 def handle_record( 339 self, handler: logging.Handler, record: logging.LogRecord 340 ) -> None: 341 name = record.name or "" 342 if any(name == p or name.startswith(p + ".") for p in _SKIP_LOGGER_PREFIXES): 343 return 344 level_name = record.levelname or "INFO" 345 if not level_at_least(level_name, self._resolve_min_level()): 346 return 347 if not self.capture_logs_enabled_for_agent(): 348 return 349 session = _resolve_session_fields(self._session_resolver) 350 if not session or not session.get("session_id"): 351 return 352 message = handler.format(record) if handler.formatter else record.getMessage() 353 if not isinstance(message, str): 354 message = str(message) 355 trace_id, span_id = _current_trace_span() 356 ts = datetime.fromtimestamp(record.created, tz=timezone.utc).strftime( 357 "%Y-%m-%d %H:%M:%S.%f" 358 )[:-3] 359 attrs: dict[str, str] = {} 360 if record.pathname: 361 attrs["pathname"] = str(record.pathname)[:512] 362 if record.lineno: 363 attrs["lineno"] = str(record.lineno) 364 if record.funcName: 365 attrs["funcName"] = str(record.funcName)[:128] 366 event = SessionLogRecord( 367 session_id=str(session["session_id"]), 368 conversation_id=str(session.get("conversation_id") or session["session_id"]), 369 ts=ts, 370 level=normalize_log_level(level_name), 371 logger_name=name, 372 message=message, 373 turn_index=int(session.get("turn_index") or 0), 374 trace_id=trace_id, 375 span_id=span_id, 376 attributes=attrs, 377 ) 378 with self._lock: 379 self._buffer.append(event)
def
init(self, *, endpoint: str = '', api_key: str = '') -> None:
381 def init(self, *, endpoint: str = "", api_key: str = "") -> None: 382 """Install root handler + start background flusher.""" 383 with self._lock: 384 self._endpoint = (endpoint or "").rstrip("/") 385 self._api_key = api_key or "" 386 if self._handler is None: 387 self._handler = SessionLogHandler(self) 388 self._handler.setLevel(logging.DEBUG) 389 self._handler.setFormatter(logging.Formatter("%(message)s")) 390 if not self._installed: 391 logging.root.addHandler(self._handler) 392 self._installed = True 393 if self._flush_thread is None or not self._flush_thread.is_alive(): 394 self._stop.clear() 395 self._flush_thread = threading.Thread( 396 target=self._flush_loop, 397 name="parlot-session-logs-flush", 398 daemon=True, 399 ) 400 self._flush_thread.start() 401 if not self._atexit_registered: 402 atexit.register(self.shutdown) 403 self._atexit_registered = True
Install root handler + start background flusher.
def
shutdown(self) -> None:
481 def shutdown(self) -> None: 482 """Stop capture and push any remaining buffered logs. Never raises.""" 483 try: 484 self._stop.set() 485 thread = self._flush_thread 486 if ( 487 thread is not None 488 and thread.is_alive() 489 and thread is not threading.current_thread() 490 ): 491 thread.join(timeout=1.0) 492 self._flush_thread = None 493 with self._lock: 494 if self._installed and self._handler is not None: 495 try: 496 logging.root.removeHandler(self._handler) 497 except Exception: 498 pass 499 self._installed = False 500 self._try_flush(force=True) 501 except Exception: 502 return
Stop capture and push any remaining buffered logs. Never raises.
def
apply_genai_span_rename( span: opentelemetry.sdk.trace.ReadableSpan, *, remap_fn: Callable[[str, Optional[Mapping[str, Union[str, bool, int, float, Sequence[str], Sequence[bool], Sequence[int], Sequence[float]]]]], str | None], stage_for_native: Optional[Mapping[str, str]] = None, llm_names: Collection[str] = (), tool_names: Collection[str] = (), set_chat_op: bool = True, set_tool_op: bool = True, preserve_ops: frozenset[str] | None = None, resolve_stage_key: Optional[Callable[[str], str]] = None, after_rename: Optional[Callable[[opentelemetry.sdk.trace.ReadableSpan, str, str, Mapping[str, Union[str, bool, int, float, Sequence[str], Sequence[bool], Sequence[int], Sequence[float]]]], NoneType]] = None) -> None:
82def apply_genai_span_rename( 83 span: ReadableSpan, 84 *, 85 remap_fn: Callable[[str, Mapping[str, AttributeValue] | None], str | None], 86 stage_for_native: Mapping[str, str] | None = None, 87 llm_names: Collection[str] = (), 88 tool_names: Collection[str] = (), 89 set_chat_op: bool = True, 90 set_tool_op: bool = True, 91 preserve_ops: frozenset[str] | None = None, 92 resolve_stage_key: Callable[[str], str] | None = None, 93 after_rename: Callable[ 94 [ReadableSpan, str, str, Mapping[str, AttributeValue]], None 95 ] 96 | None = None, 97) -> None: 98 """Mutate ``span._name`` and GenAI op / stage attrs for export. 99 100 ``after_rename(span, native, exported, attrs)`` runs when the adapter needs 101 extra logic (e.g. LiveKit evaluation op preservation, tool name sync). 102 When provided, it replaces the default chat/tool op stamping. 103 """ 104 native = span.name or "" 105 attrs = span.attributes or {} 106 exported = remap_fn(native, attrs) 107 if exported is None: 108 return 109 if exported != native: 110 span._name = exported # type: ignore[attr-defined] 111 112 stage_key = resolve_stage_key(native) if resolve_stage_key else native 113 if stage_for_native: 114 stage = stage_for_native.get(stage_key) 115 if stage and not attrs.get(ATTR_AGENT_STAGE): 116 ParlotBaseProcessor._set(span, ATTR_AGENT_STAGE, stage) 117 118 if after_rename is not None: 119 after_rename(span, native, exported, attrs) 120 return 121 122 preserve = preserve_ops or frozenset({GEN_AI_OP_GENERATE_CONTENT}) 123 if set_chat_op and native in llm_names: 124 existing_op = str(attrs.get(ATTR_GEN_AI_OP_NAME) or "").strip().lower() 125 if existing_op not in preserve: 126 ParlotBaseProcessor._set(span, ATTR_GEN_AI_OP_NAME, GEN_AI_OP_CHAT) 127 elif set_tool_op and native in tool_names: 128 ParlotBaseProcessor._set(span, ATTR_GEN_AI_OP_NAME, GEN_AI_OP_EXECUTE_TOOL)
Mutate span._name and GenAI op / stage attrs for export.
after_rename(span, native, exported, attrs) runs when the adapter needs
extra logic (e.g. LiveKit evaluation op preservation, tool name sync).
When provided, it replaces the default chat/tool op stamping.
def
attach_span_processor_if_absent( provider: Any, factory: Any, *, processor_type: type | None = None) -> bool:
150def attach_span_processor_if_absent( 151 provider: Any, 152 factory: Any, 153 *, 154 processor_type: type | None = None, 155) -> bool: 156 """Attach a span processor when the SDK provider does not already have one. 157 158 ``factory`` is a zero-arg callable that returns a ``SpanProcessor``. 159 ``processor_type`` defaults to the type of the first factory result check — 160 when omitted, skips the duplicate check and always adds. 161 162 Returns True when a processor was attached. 163 """ 164 from opentelemetry.sdk.trace import TracerProvider as SdkTracerProvider 165 166 if not isinstance(provider, SdkTracerProvider): 167 return False 168 169 if processor_type is not None: 170 try: 171 active = getattr(provider, "_active_span_processor", None) 172 procs = getattr(active, "_span_processors", None) or getattr( 173 active, "_processors", None 174 ) 175 if procs and any(isinstance(p, processor_type) for p in procs): 176 return False 177 except Exception: 178 pass 179 try: 180 provider.add_span_processor(factory()) 181 return True 182 except Exception: 183 logger.debug("Could not attach span processor", exc_info=True) 184 return False
Attach a span processor when the SDK provider does not already have one.
factory is a zero-arg callable that returns a SpanProcessor.
processor_type defaults to the type of the first factory result check —
when omitted, skips the duplicate check and always adds.
Returns True when a processor was attached.
def
base_parlotize( agent_id: str, *, endpoint: Optional[str] = None, api_key: Optional[str] = None, capture_genai_content: Optional[bool] = None, tracer_provider: opentelemetry.trace.TracerProvider | None = None, version: Optional[str] = None, capture_logs: bool | list[str] | None = None, log_level: Optional[str] = None) -> BaseParlotizeResult:
172def base_parlotize( 173 agent_id: str, 174 *, 175 endpoint: Optional[str] = None, 176 api_key: Optional[str] = None, 177 capture_genai_content: Optional[bool] = None, 178 tracer_provider: TracerProvider | None = None, 179 version: Optional[str] = None, 180 capture_logs: bool | list[str] | None = None, 181 log_level: Optional[str] = None, 182) -> BaseParlotizeResult: 183 """Execute shared telemetry configuration common across all adapters. 184 185 Creates a ``ParlotContext``, initializes the session log 186 collector, and fetches remote bootstrap into ``context.runtime``. 187 188 See ``ParlotizeProtocol`` for the shared parameter surface. 189 190 Args: 191 agent_id: Required canonical deployment identity for ``session.agent_id``. 192 endpoint: Parlot OTLP collector base URL. If omitted, reads 193 ``PARLOT_ENDPOINT``. 194 api_key: Org-scoped API key. If omitted, reads ``PARLOT_API_KEY``. 195 capture_genai_content: Process-wide GenAI content capture override. 196 tracer_provider: Existing ``TracerProvider`` to adopt, if any. 197 version: Deployment version for ``gen_ai.agent.version``. Defaults to 198 ``\"unknown\"`` when omitted or blank. 199 capture_logs: Session log capture policy (bool or agent-id globs). 200 log_level: Minimum level for session log capture. 201 202 Returns: 203 Resolved endpoint, credentials, provider, and capture settings for the 204 calling adapter. 205 """ 206 from parlot.core.bootstrap import fetch_telemetry_bootstrap 207 from parlot.core.context import ParlotContext 208 from parlot.core.provider import ( 209 adopt_existing_tracer_provider, 210 resolve_api_key, 211 resolve_capture_genai_content, 212 resolve_endpoint, 213 ) 214 215 configure_parlot_logging() 216 217 resolved_agent_id = agent_id.strip() 218 if not resolved_agent_id: 219 raise ValueError("parlotize(agent_id) is required") 220 resolved_version = (version or "").strip() or "unknown" 221 resolved_endpoint = resolve_endpoint(endpoint) 222 resolved_api_key = resolve_api_key(api_key) 223 resolved_capture_content = resolve_capture_genai_content(capture_genai_content) 224 225 resolved_capture_logs = normalize_glob_or_bool(capture_logs) 226 227 resolved_log_level = log_level.strip().upper() if log_level else None 228 229 context = ParlotContext() 230 context.session_logs.set_capture_logs_config( 231 resolved_capture_logs, 232 log_level=resolved_log_level, 233 ) 234 context.session_logs.init(endpoint=resolved_endpoint, api_key=resolved_api_key) 235 236 if not resolved_endpoint: 237 logger.error("parlot: PARLOT_ENDPOINT is not set") 238 if not resolved_api_key: 239 logger.error( 240 "parlot: PARLOT_API_KEY is not set; telemetry will not authenticate" 241 ) 242 elif resolved_endpoint: 243 fetch_telemetry_bootstrap(resolved_endpoint, resolved_api_key, context) 244 245 if tracer_provider is None: 246 tracer_provider = adopt_existing_tracer_provider() 247 248 return BaseParlotizeResult( 249 context=context, 250 endpoint=resolved_endpoint, 251 api_key=resolved_api_key, 252 tracer_provider=tracer_provider, 253 agent_id=resolved_agent_id, 254 agent_version=resolved_version, 255 capture_genai_content=resolved_capture_content, 256 capture_logs=resolved_capture_logs, 257 log_level=resolved_log_level, 258 )
Execute shared telemetry configuration common across all adapters.
Creates a ParlotContext, initializes the session log
collector, and fetches remote bootstrap into context.runtime.
See ParlotizeProtocol for the shared parameter surface.
Arguments:
- agent_id: Required canonical deployment identity for
session.agent_id. - endpoint: Parlot OTLP collector base URL. If omitted, reads
PARLOT_ENDPOINT. - api_key: Org-scoped API key. If omitted, reads
PARLOT_API_KEY. - capture_genai_content: Process-wide GenAI content capture override.
- tracer_provider: Existing
TracerProviderto adopt, if any. - version: Deployment version for
gen_ai.agent.version. Defaults to"unknown"when omitted or blank. - capture_logs: Session log capture policy (bool or agent-id globs).
- log_level: Minimum level for session log capture.
Returns:
Resolved endpoint, credentials, provider, and capture settings for the calling adapter.
def
build_filtered_tracer_provider( *, endpoint: str, api_key: str, service_name: Optional[str] = None, service_version: Optional[str] = None, span_processors: Sequence[opentelemetry.sdk.trace.SpanProcessor] = ()) -> opentelemetry.sdk.trace.TracerProvider:
187def build_filtered_tracer_provider( 188 *, 189 endpoint: str, 190 api_key: str, 191 service_name: Optional[str] = None, 192 service_version: Optional[str] = None, 193 span_processors: Sequence[SpanProcessor] = (), 194) -> TracerProvider: 195 """Build a TracerProvider with ``ExportFilterSpanExporter`` over OTLP/HTTP. 196 197 Framework processors go in ``span_processors`` (before the batch exporter). 198 """ 199 from parlot.core.export import ExportFilterSpanExporter 200 201 resource = build_resource( 202 service_name=service_name, 203 service_version=service_version, 204 ) 205 otlp = build_otlp_http_exporter(endpoint=endpoint, api_key=api_key) 206 exporter = ExportFilterSpanExporter(otlp) 207 provider = TracerProvider(resource=resource) 208 for processor in span_processors: 209 provider.add_span_processor(processor) 210 provider.add_span_processor(BatchSpanProcessor(exporter)) 211 return provider
Build a TracerProvider with ExportFilterSpanExporter over OTLP/HTTP.
Framework processors go in span_processors (before the batch exporter).
def
configure_parlot_logging() -> None:
149def configure_parlot_logging() -> None: 150 """Apply ``PARLOT_DEBUG_LEVEL`` to ``parlot.*`` loggers only. 151 152 Ensures a root handler exists at INFO so messages reach stdout, without 153 flipping the process root to DEBUG (that floods ``websockets`` / ``aiortc`` 154 / similar when debugging Parlot span export). 155 156 Sets the ``parlot`` parent so future ``parlot.*`` children with ``NOTSET`` 157 inherit, and walks already-created ``parlot.*`` loggers so a prior 158 ``setLevel`` cannot stick — no hardcoded name list to maintain. 159 """ 160 level_name = os.getenv("PARLOT_DEBUG_LEVEL", "INFO").upper() 161 level = getattr(logging, level_name, logging.INFO) 162 if not logging.root.handlers: 163 logging.basicConfig(level=logging.INFO) 164 logging.getLogger("parlot").setLevel(level) 165 for name, obj in list(logging.Logger.manager.loggerDict.items()): 166 if name != "parlot" and not name.startswith("parlot."): 167 continue 168 if isinstance(obj, logging.Logger): 169 obj.setLevel(level)
Apply PARLOT_DEBUG_LEVEL to parlot.* loggers only.
Ensures a root handler exists at INFO so messages reach stdout, without
flipping the process root to DEBUG (that floods websockets / aiortc
/ similar when debugging Parlot span export).
Sets the parlot parent so future parlot.* children with NOTSET
inherit, and walks already-created parlot.* loggers so a prior
setLevel cannot stick — no hardcoded name list to maintain.
def
emit_parlot_session_close( tracer: Any, *, session_id: str, conversation_id: str, reason: str = 'completed', framework: str = '', close_error: str | None = None, extra_attrs: Optional[Mapping[str, Any]] = None, copy_from_session_span: typing.Any | None = None, stamp_fn: Optional[Callable[[Any], NoneType]] = None, end_time: int | None = None, state: typing.Any | None = None) -> None:
153def emit_parlot_session_close( 154 tracer: Any, 155 *, 156 session_id: str, 157 conversation_id: str, 158 reason: str = "completed", 159 framework: str = "", 160 close_error: str | None = None, 161 extra_attrs: Mapping[str, Any] | None = None, 162 copy_from_session_span: Any | None = None, 163 stamp_fn: Callable[[Any], None] | None = None, 164 end_time: int | None = None, 165 state: Any | None = None, 166) -> None: 167 """Mint a ``parlot.session.close`` span for collector finalize. 168 169 ``stamp_fn`` applies framework identity / platform refs after base attrs. 170 ``copy_from_session_span`` copies scalar attrs from the live session span 171 (LiveKit uses this so close carries room/job metadata). 172 """ 173 if tracer is None: 174 return 175 attrs: dict[str, Any] = { 176 ATTR_SESSION_ID: session_id, 177 ATTR_SESSION_CONVERSATION_ID: conversation_id, 178 ATTR_SESSION_CLOSE_REASON: reason, 179 } 180 if conversation_id: 181 attrs[ATTR_GEN_AI_CONVERSATION_ID] = conversation_id 182 if framework: 183 attrs[ATTR_AGENT_FRAMEWORK] = framework 184 if close_error: 185 attrs[ATTR_SESSION_CLOSE_ERROR] = close_error 186 if copy_from_session_span is not None: 187 _copy_scalar_attrs_from_span(copy_from_session_span, attrs) 188 if extra_attrs: 189 for key, value in extra_attrs.items(): 190 if value is None: 191 continue 192 attrs[key] = value 193 194 span = tracer.start_span(SPAN_PARLOT_SESSION_CLOSE, attributes=attrs) 195 try: 196 if stamp_fn is not None: 197 stamp_fn(span) 198 if state is not None: 199 stamp_session_sdk_version(span, state) 200 finally: 201 if end_time is not None and hasattr(span, "end"): 202 span.end(end_time=end_time) 203 else: 204 span.end()
Mint a parlot.session.close span for collector finalize.
stamp_fn applies framework identity / platform refs after base attrs.
copy_from_session_span copies scalar attrs from the live session span
(LiveKit uses this so close carries room/job metadata).
def
emit_parlot_turn_root( tracer: Any, *, session_id: str, conversation_id: str, turn_index: int, participant_role: str = '', utterance_text: str = '', user_text: str = '', agent_text: str = '', prev_trace_id: str = '', start_time_unix_ns: int | None = None, end_time_unix_ns: int | None = None, interrupted: bool = False, e2e_latency_s: float | None = None, llm_ttft_s: float | None = None, tts_ttfb_s: float | None = None, stamp_genai_conversation_id: bool = False, extra_attrs: Optional[Mapping[str, Any]] = None, stamp_fn: Optional[Callable[[Any], NoneType]] = None) -> tuple[str, str]:
51def emit_parlot_turn_root( 52 tracer: Any, 53 *, 54 session_id: str, 55 conversation_id: str, 56 turn_index: int, 57 participant_role: str = "", 58 utterance_text: str = "", 59 user_text: str = "", 60 agent_text: str = "", 61 prev_trace_id: str = "", 62 start_time_unix_ns: int | None = None, 63 end_time_unix_ns: int | None = None, 64 interrupted: bool = False, 65 e2e_latency_s: float | None = None, 66 llm_ttft_s: float | None = None, 67 tts_ttfb_s: float | None = None, 68 stamp_genai_conversation_id: bool = False, 69 extra_attrs: Mapping[str, Any] | None = None, 70 stamp_fn: Callable[[Any], None] | None = None, 71) -> tuple[str, str]: 72 """ 73 Start and end a ``parlot.turn`` root span in a **new** OTel trace. 74 75 When ``start_time_unix_ns`` / ``end_time_unix_ns`` are set, OTLP duration 76 matches the speech / turn window. 77 78 Prefer explicit ``user_text`` / ``agent_text`` when both sides of an exchange 79 are known; otherwise pass ``utterance_text`` with ``participant_role``. 80 81 ``extra_attrs`` carries framework-specific turn attributes (media ms, 82 diarization, ``agent.framework``, etc.). ``stamp_fn`` runs before the span 83 ends (platform refs, framework identity). 84 85 Returns ``(trace_id, span_id)`` as 32- and 16-char hex strings. 86 """ 87 trace_id = random.getrandbits(128) 88 span_id = random.getrandbits(64) 89 ctx = SpanContext( 90 trace_id=trace_id, 91 span_id=span_id, 92 is_remote=False, 93 trace_flags=TraceFlags(0x01), 94 ) 95 parent_ctx = trace.set_span_in_context(NonRecordingSpan(ctx)) 96 97 start_kw: dict[str, Any] = {"context": parent_ctx} 98 if start_time_unix_ns is not None: 99 start_kw["start_time"] = start_time_unix_ns 100 101 span = tracer.start_span(SPAN_PARLOT_TURN, **start_kw) 102 try: 103 span.set_attribute(ATTR_SESSION_ID, session_id) 104 span.set_attribute(ATTR_SESSION_CONVERSATION_ID, conversation_id) 105 if stamp_genai_conversation_id: 106 span.set_attribute(ATTR_GEN_AI_CONVERSATION_ID, conversation_id) 107 span.set_attribute(ATTR_TURN_INDEX, turn_index) 108 if participant_role: 109 span.set_attribute(ATTR_TURN_PARTICIPANT_ROLE, participant_role) 110 if prev_trace_id: 111 span.set_attribute(ATTR_TURN_PREV_TRACE_ID, prev_trace_id) 112 if interrupted: 113 span.set_attribute(ATTR_TURN_INTERRUPTED, True) 114 if e2e_latency_s is not None: 115 span.set_attribute(ATTR_TURN_E2E_LATENCY_S, float(e2e_latency_s)) 116 if llm_ttft_s is not None: 117 span.set_attribute(ATTR_TURN_LLM_TTFT_S, float(llm_ttft_s)) 118 if tts_ttfb_s is not None: 119 span.set_attribute(ATTR_TURN_TTS_TTFB_S, float(tts_ttfb_s)) 120 121 user = (user_text or "").strip() 122 agent = (agent_text or "").strip() 123 if user or agent: 124 if user: 125 stamp_turn_utterance_text( 126 span, participant_role="user", utterance_text=user 127 ) 128 if agent: 129 stamp_turn_utterance_text( 130 span, participant_role="agent", utterance_text=agent 131 ) 132 elif utterance_text: 133 stamp_turn_utterance_text( 134 span, 135 participant_role=participant_role, 136 utterance_text=utterance_text, 137 ) 138 139 if extra_attrs: 140 for key, value in extra_attrs.items(): 141 if value is None or value == "": 142 continue 143 span.set_attribute(key, value) 144 if stamp_fn is not None: 145 stamp_fn(span) 146 finally: 147 if end_time_unix_ns is not None: 148 span.end(end_time=end_time_unix_ns) 149 else: 150 span.end() 151 152 return format(trace_id, "032x"), format(span_id, "016x")
Start and end a parlot.turn root span in a new OTel trace.
When start_time_unix_ns / end_time_unix_ns are set, OTLP duration
matches the speech / turn window.
Prefer explicit user_text / agent_text when both sides of an exchange
are known; otherwise pass utterance_text with participant_role.
extra_attrs carries framework-specific turn attributes (media ms,
diarization, agent.framework, etc.). stamp_fn runs before the span
ends (platform refs, framework identity).
Returns (trace_id, span_id) as 32- and 16-char hex strings.
def
end_session_span_ok(session_span: Any) -> None:
234def end_session_span_ok(session_span: Any) -> None: 235 """Mark a session span OK and end it.""" 236 try: 237 session_span.set_status(Status(StatusCode.OK)) 238 except Exception: 239 pass 240 session_span.end()
Mark a session span OK and end it.
class
ExportFilterSpanExporter(opentelemetry.sdk.trace.export.SpanExporter):
14class ExportFilterSpanExporter(SpanExporter): 15 """Pass through only Conversation Contract + GenAI + voice spans.""" 16 17 def __init__(self, exporter: SpanExporter) -> None: 18 self._exporter = exporter 19 20 def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: 21 filtered = [span for span in spans if is_exportable_span_name(span.name)] 22 if not filtered: 23 return SpanExportResult.SUCCESS 24 return self._exporter.export(filtered) 25 26 def shutdown(self) -> None: 27 self._exporter.shutdown() 28 29 def force_flush(self, timeout_millis: int = 30000) -> bool: 30 return self._exporter.force_flush(timeout_millis)
Pass through only Conversation Contract + GenAI + voice spans.
def
export( self, spans: Sequence[opentelemetry.sdk.trace.ReadableSpan]) -> opentelemetry.sdk.trace.export.SpanExportResult:
20 def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: 21 filtered = [span for span in spans if is_exportable_span_name(span.name)] 22 if not filtered: 23 return SpanExportResult.SUCCESS 24 return self._exporter.export(filtered)
Exports a batch of telemetry data.
Arguments:
- spans: The list of
opentelemetry.trace.Spanobjects to be exported
Returns:
The result of the export
def
force_flush(self, timeout_millis: int = 30000) -> bool:
29 def force_flush(self, timeout_millis: int = 30000) -> bool: 30 return self._exporter.force_flush(timeout_millis)
Hint to ensure that the export of any spans the exporter has received prior to the call to ForceFlush SHOULD be completed as soon as possible, preferably before returning from this method.
def
force_flush_tracer_provider( provider: typing.Any | None = None, *, timeout_millis: int = 10000) -> bool:
33def force_flush_tracer_provider( 34 provider: Any | None = None, 35 *, 36 timeout_millis: int = 10_000, 37) -> bool: 38 """Return True when the tracer provider reports a successful flush. 39 40 When ``provider`` is omitted, uses the global tracer provider. 41 """ 42 if provider is None: 43 from opentelemetry import trace 44 45 provider = trace.get_tracer_provider() 46 if provider is None: 47 return True 48 force_flush = getattr(provider, "force_flush", None) 49 if not callable(force_flush): 50 return True 51 try: 52 result = force_flush(timeout_millis=timeout_millis) 53 except TypeError: 54 try: 55 result = force_flush() 56 except Exception: 57 logger.warning("parlot: force_flush raised", exc_info=True) 58 return False 59 except Exception: 60 logger.warning("parlot: force_flush raised", exc_info=True) 61 return False 62 if result is None: 63 # Some providers return None on success; treat as ok. 64 return True 65 return bool(result)
Return True when the tracer provider reports a successful flush.
When provider is omitted, uses the global tracer provider.
def
force_flush_with_budget( *, budget_s: float = 12.0, backoffs_s: Sequence[float] = (1.0, 3.0, 6.0), attempt_timeout_ms: int = 4000, provider: typing.Any | None = None, flush_fn: Optional[Callable[..., bool]] = None, log: logging.Logger | None = None) -> bool:
68def force_flush_with_budget( 69 *, 70 budget_s: float = DEFAULT_CLOSE_FLUSH_BUDGET_S, 71 backoffs_s: Sequence[float] = DEFAULT_CLOSE_FLUSH_BACKOFFS_S, 72 attempt_timeout_ms: int = DEFAULT_CLOSE_FLUSH_ATTEMPT_TIMEOUT_MS, 73 provider: Any | None = None, 74 flush_fn: Callable[..., bool] | None = None, 75 log: logging.Logger | None = None, 76) -> bool: 77 """Retry OTLP flush within a wall-clock budget. 78 79 Returns True when flush confirmed within the budget; False otherwise. 80 Default policy matches LiveKit's Room Composite close path (1s/3s/6s 81 backoff inside ~12s). 82 83 ``flush_fn`` defaults to :func:`force_flush_tracer_provider`. Adapters may 84 pass a local wrapper so tests can patch the adapter module symbol. 85 """ 86 log = log or logger 87 do_flush = flush_fn or ( 88 lambda *, timeout_millis: force_flush_tracer_provider( 89 provider, timeout_millis=timeout_millis 90 ) 91 ) 92 started = time.monotonic() 93 attempt = 0 94 while True: 95 attempt += 1 96 remaining_s = budget_s - (time.monotonic() - started) 97 if remaining_s <= 0: 98 break 99 timeout_ms = min(attempt_timeout_ms, max(1, int(remaining_s * 1000))) 100 ok = do_flush(timeout_millis=timeout_ms) 101 elapsed_ms = int((time.monotonic() - started) * 1000) 102 if ok: 103 log.info( 104 "parlot: otlp flush before close ok attempts=%s elapsed_ms=%s", 105 attempt, 106 elapsed_ms, 107 ) 108 return True 109 backoff_idx = attempt - 1 110 if backoff_idx >= len(backoffs_s): 111 break 112 backoff_s = backoffs_s[backoff_idx] 113 remaining_s = budget_s - (time.monotonic() - started) 114 if remaining_s <= 0: 115 break 116 sleep_s = min(backoff_s, remaining_s) 117 log.warning( 118 "parlot: otlp flush before close incomplete; retrying " 119 "attempt=%s sleep_s=%.1f remaining_s=%.1f", 120 attempt, 121 sleep_s, 122 remaining_s, 123 ) 124 time.sleep(sleep_s) 125 elapsed_ms = int((time.monotonic() - started) * 1000) 126 log.warning( 127 "parlot: otlp flush before close failed attempts=%s elapsed_ms=%s", 128 attempt, 129 elapsed_ms, 130 ) 131 return False
Retry OTLP flush within a wall-clock budget.
Returns True when flush confirmed within the budget; False otherwise. Default policy matches LiveKit's Room Composite close path (1s/3s/6s backoff inside ~12s).
flush_fn defaults to force_flush_tracer_provider(). Adapters may
pass a local wrapper so tests can patch the adapter module symbol.
LONG_LIVED_METRICS_EXPORT_INTERVAL_MS =
60000
def
log_parlotize_ready( framework: str, *, agent_id: str, endpoint: str, api_key: str = '', bootstrap_ok: bool | None = None, log: logging.Logger | None = None) -> None:
32def log_parlotize_ready( 33 framework: str, 34 *, 35 agent_id: str, 36 endpoint: str, 37 api_key: str = "", 38 bootstrap_ok: bool | None = None, 39 log: logging.Logger | None = None, 40) -> None: 41 """Shared ready / warn log after adapter ``parlotize()`` succeeds.""" 42 log = log or logger 43 auth_ok = bool(endpoint and api_key) 44 if not auth_ok or bootstrap_ok is False: 45 bootstrap_label = ( 46 "failed" 47 if bootstrap_ok is False 48 else ("ok" if bootstrap_ok else "skipped") 49 ) 50 # LiveKit historically used "is not ready"; Pipecat used "may not authenticate". 51 # Prefer the broader "not ready" wording when bootstrap is explicitly False. 52 if bootstrap_ok is False: 53 log.warning( 54 "parlot: %s instrumentation installed but telemetry is not ready " 55 "(agent_id=%s endpoint=%s bootstrap=%s); check PARLOT_ENDPOINT / " 56 "PARLOT_API_KEY", 57 framework, 58 agent_id, 59 endpoint or "(unset)", 60 bootstrap_label, 61 ) 62 else: 63 log.warning( 64 "parlot: %s instrumentation installed but telemetry may not " 65 "authenticate (agent_id=%s endpoint=%s); check PARLOT_ENDPOINT / " 66 "PARLOT_API_KEY", 67 framework, 68 agent_id, 69 endpoint or "(unset)", 70 ) 71 return 72 log.info( 73 "parlot: %s instrumentation ready (agent_id=%s endpoint=%s)", 74 framework, 75 agent_id, 76 endpoint, 77 )
Shared ready / warn log after adapter parlotize() succeeds.
def
normalize_glob_or_bool(value: bool | list[str] | None) -> bool | list[str] | None:
23def normalize_glob_or_bool( 24 value: bool | list[str] | None, 25) -> bool | list[str] | None: 26 """Normalize a bool-or-glob-list config (``record`` / ``capture_logs``).""" 27 if isinstance(value, list): 28 return [str(item).strip() for item in value if str(item).strip()] 29 return value
Normalize a bool-or-glob-list config (record / capture_logs).
class
ParlotBaseProcessor(opentelemetry.sdk.trace.SpanProcessor):
23class ParlotBaseProcessor(SpanProcessor): 24 """ 25 Subclass and override ``on_end`` (or the ``_enrich`` dispatch method) to 26 add framework-specific span enrichment. Call ``super().on_start()`` / 27 ``super().on_end()`` so shared debug logging runs. 28 29 Do NOT wrap a downstream processor — add alongside BatchSpanProcessor: 30 31 provider.add_span_processor(MyFrameworkProcessor()) 32 provider.add_span_processor(BatchSpanProcessor(your_exporter)) 33 """ 34 35 def on_start(self, span, parent_context=None) -> None: 36 log_span_event("on_start", span) 37 38 def on_end(self, span: ReadableSpan) -> None: 39 log_span_event("on_end", span) 40 41 def shutdown(self) -> None: 42 pass 43 44 def force_flush(self, timeout_millis: int = 30_000) -> bool: 45 return True 46 47 # ------------------------------------------------------------------ 48 # Shared helpers — used by all subclasses 49 # ------------------------------------------------------------------ 50 51 @staticmethod 52 def _mutable_attributes(span: ReadableSpan) -> dict: 53 """Return a mutable ``span._attributes`` dict (copy if needed).""" 54 attrs = span._attributes 55 if attrs is None: 56 span._attributes = {} 57 return span._attributes 58 if getattr(attrs, "_immutable", False) or not isinstance(attrs, dict): 59 span._attributes = dict(attrs) 60 return span._attributes 61 return attrs 62 63 @staticmethod 64 def _set(span: ReadableSpan, key: str, value) -> None: 65 """Write an attribute into a ReadableSpan after it has ended. 66 67 OTel SDK ≥1.44 stores attributes in an immutable ``BoundedAttributes`` 68 by default; mutating it raises ``TypeError``. Copy to a plain dict 69 when needed so enrichment can still stamp session/turn fields. 70 """ 71 attrs = ParlotBaseProcessor._mutable_attributes(span) 72 try: 73 attrs[key] = value 74 except TypeError: 75 span._attributes = dict(attrs) 76 span._attributes[key] = value 77 78 @staticmethod 79 def _delete(span: ReadableSpan, key: str) -> None: 80 """Remove an attribute from a ReadableSpan after it has ended.""" 81 attrs = span._attributes 82 if not attrs or key not in attrs: 83 return 84 mutable = ParlotBaseProcessor._mutable_attributes(span) 85 try: 86 mutable.pop(key, None) 87 except TypeError: 88 span._attributes = {k: v for k, v in dict(mutable).items() if k != key} 89 90 @staticmethod 91 def _add_event(span: ReadableSpan, name: str, attributes: dict) -> None: 92 """Append a span event to a ReadableSpan after it has ended.""" 93 from opentelemetry.sdk.trace import Event 94 evt = Event(name=name, attributes=attributes, timestamp=time.time_ns()) 95 if hasattr(span, "_events") and isinstance(span._events, list): 96 span._events.append(evt) 97 98 @staticmethod 99 def _trace_id_hex(span: ReadableSpan) -> str: 100 return format(span.context.trace_id, "032x") 101 102 @staticmethod 103 def _maybe_update(state: SessionState, attr: str, value) -> None: 104 """Set a state attribute only if it is currently falsy.""" 105 if value and not getattr(state, attr, ""): 106 setattr(state, attr, str(value))
Subclass and override on_end (or the _enrich dispatch method) to
add framework-specific span enrichment. Call super().on_start() /
super().on_end() so shared debug logging runs.
Do NOT wrap a downstream processor — add alongside BatchSpanProcessor:
provider.add_span_processor(MyFrameworkProcessor())
provider.add_span_processor(BatchSpanProcessor(your_exporter))
def
on_start(self, span, parent_context=None) -> None:
Called when a opentelemetry.trace.Span is started.
This method is called synchronously on the thread that starts the span, therefore it should not block or throw an exception.
Arguments:
- span: The
opentelemetry.trace.Spanthat just started. - parent_context: The parent context of the span that just started.
def
on_end(self, span: opentelemetry.sdk.trace.ReadableSpan) -> None:
Called when a opentelemetry.trace.Span is ended.
This method is called synchronously on the thread that ends the span, therefore it should not block or throw an exception.
Arguments:
- span: The
opentelemetry.trace.Spanthat just ended.
class
ParlotMetricsRecorder:
131class ParlotMetricsRecorder: 132 """Records Parlot ``usage.*``, ``turn.*``, and ``session.*`` instruments.""" 133 134 def __init__( 135 self, 136 meter_provider: MeterProvider, 137 *, 138 meter_name: str = "parlot.core", 139 framework: str = "", 140 tenant_id: str = "", 141 ) -> None: 142 self._meter_provider = meter_provider 143 self._framework = (framework or "").strip() 144 self._tenant_id = (tenant_id or "").strip() 145 meter = meter_provider.get_meter(meter_name) 146 self._meter = meter 147 148 self._usage_stt = meter.create_counter( 149 METRIC_USAGE_STT_AUDIO_DURATION, 150 unit="s", 151 description="STT audio duration per usage event", 152 ) 153 self._usage_tts_audio = meter.create_counter( 154 METRIC_USAGE_TTS_AUDIO_DURATION, 155 unit="s", 156 description="TTS audio duration per usage event", 157 ) 158 self._usage_tts_chars = meter.create_counter( 159 METRIC_USAGE_TTS_CHARACTERS, 160 description="TTS characters synthesized per usage event", 161 ) 162 self._usage_llm_in = meter.create_counter( 163 METRIC_USAGE_LLM_INPUT_TOKENS, 164 description="LLM input tokens per usage event", 165 ) 166 self._usage_llm_out = meter.create_counter( 167 METRIC_USAGE_LLM_OUTPUT_TOKENS, 168 description="LLM output tokens per usage event", 169 ) 170 171 self._turn_e2e = meter.create_histogram( 172 "turn.e2e_latency_ms", 173 unit="ms", 174 description="End-to-end latency for one agent turn", 175 ) 176 self._turn_llm_ttft = meter.create_histogram( 177 "turn.llm_ttft_ms", 178 unit="ms", 179 description="LLM time to first token for one agent turn", 180 ) 181 self._turn_tts_ttfb = meter.create_histogram( 182 "turn.tts_ttfb_ms", 183 unit="ms", 184 description="TTS time to first byte for one agent turn", 185 ) 186 self._turn_transcription_delay = meter.create_histogram( 187 "turn.transcription_delay_ms", 188 unit="ms", 189 description="Speech end to transcript available for one agent turn", 190 ) 191 self._turn_eou_delay = meter.create_histogram( 192 "turn.eou_delay_ms", 193 unit="ms", 194 description="End-of-utterance detection delay for one agent turn", 195 ) 196 self._turn_playback_latency = meter.create_histogram( 197 "turn.playback_latency_ms", 198 unit="ms", 199 description="Playback latency for one agent turn", 200 ) 201 202 self._session_turns = meter.create_counter( 203 "session.turn_count", 204 description="Total turns in a session", 205 ) 206 self._session_input_tokens = meter.create_counter( 207 "session.total_input_tokens", 208 description="Total input tokens in session", 209 ) 210 self._session_output_tokens = meter.create_counter( 211 "session.total_output_tokens", 212 description="Total output tokens in session", 213 ) 214 215 @property 216 def meter_provider(self) -> MeterProvider: 217 return self._meter_provider 218 219 @property 220 def meter(self): 221 return self._meter 222 223 def _base_attrs( 224 self, 225 *, 226 session_id: str, 227 model: str = "", 228 provider: str = "", 229 framework: str = "", 230 participant_role: str = "", 231 extra: Optional[Mapping[str, AttributeValue]] = None, 232 ) -> dict[str, AttributeValue]: 233 attrs: dict[str, AttributeValue] = { 234 ATTR_SESSION_ID: session_id, 235 } 236 fw = (framework or self._framework).strip() 237 if fw: 238 attrs[ATTR_AGENT_FRAMEWORK] = fw 239 if self._tenant_id: 240 attrs[ATTR_TENANT_ID] = self._tenant_id 241 if model: 242 attrs[ATTR_GEN_AI_MODEL] = model 243 if provider: 244 attrs[ATTR_GEN_AI_PROVIDER] = provider 245 if participant_role: 246 attrs[ATTR_TURN_PARTICIPANT_ROLE] = participant_role 247 if extra: 248 attrs.update(dict(extra)) 249 return attrs 250 251 def record_stt_audio( 252 self, 253 *, 254 session_id: str, 255 audio_seconds: float, 256 model: str = "", 257 provider: str = "", 258 framework: str = "", 259 extra: Optional[Mapping[str, AttributeValue]] = None, 260 ) -> None: 261 if not session_id or audio_seconds <= 0: 262 return 263 self._usage_stt.add( 264 float(audio_seconds), 265 attributes=self._base_attrs( 266 session_id=session_id, 267 model=model, 268 provider=provider, 269 framework=framework, 270 extra=extra, 271 ), 272 ) 273 274 def record_tts_audio( 275 self, 276 *, 277 session_id: str, 278 audio_seconds: float, 279 model: str = "", 280 provider: str = "", 281 framework: str = "", 282 extra: Optional[Mapping[str, AttributeValue]] = None, 283 ) -> None: 284 if not session_id or audio_seconds <= 0: 285 return 286 self._usage_tts_audio.add( 287 float(audio_seconds), 288 attributes=self._base_attrs( 289 session_id=session_id, 290 model=model, 291 provider=provider, 292 framework=framework, 293 extra=extra, 294 ), 295 ) 296 297 def record_tts_characters( 298 self, 299 *, 300 session_id: str, 301 characters: int, 302 model: str = "", 303 provider: str = "", 304 framework: str = "", 305 extra: Optional[Mapping[str, AttributeValue]] = None, 306 ) -> None: 307 if not session_id or characters <= 0: 308 return 309 self._usage_tts_chars.add( 310 int(characters), 311 attributes=self._base_attrs( 312 session_id=session_id, 313 model=model, 314 provider=provider, 315 framework=framework, 316 extra=extra, 317 ), 318 ) 319 320 def record_llm_tokens( 321 self, 322 *, 323 session_id: str, 324 input_tokens: int = 0, 325 output_tokens: int = 0, 326 model: str = "", 327 provider: str = "", 328 framework: str = "", 329 extra: Optional[Mapping[str, AttributeValue]] = None, 330 ) -> None: 331 if not session_id: 332 return 333 attrs = self._base_attrs( 334 session_id=session_id, 335 model=model, 336 provider=provider, 337 framework=framework, 338 extra=extra, 339 ) 340 if input_tokens > 0: 341 self._usage_llm_in.add(int(input_tokens), attributes=attrs) 342 if output_tokens > 0: 343 self._usage_llm_out.add(int(output_tokens), attributes=attrs) 344 345 def record_turn( 346 self, 347 *, 348 session_id: str, 349 turn_index: int, 350 participant_role: str = "agent", 351 e2e_latency_s: Optional[float] = None, 352 llm_ttft_s: Optional[float] = None, 353 tts_ttfb_s: Optional[float] = None, 354 transcription_delay_s: Optional[float] = None, 355 eou_delay_s: Optional[float] = None, 356 playback_latency_s: Optional[float] = None, 357 interrupted: bool = False, 358 framework: str = "", 359 extra: Optional[Mapping[str, AttributeValue]] = None, 360 ) -> None: 361 if not session_id: 362 return 363 attrs = self._base_attrs( 364 session_id=session_id, 365 framework=framework, 366 participant_role=participant_role, 367 extra=extra, 368 ) 369 attrs[ATTR_TURN_INDEX] = int(turn_index) 370 if interrupted: 371 attrs[ATTR_TURN_INTERRUPTED] = True 372 if e2e_latency_s is not None: 373 self._turn_e2e.record(float(e2e_latency_s) * 1000.0, attributes=attrs) 374 if llm_ttft_s is not None: 375 self._turn_llm_ttft.record(float(llm_ttft_s) * 1000.0, attributes=attrs) 376 if tts_ttfb_s is not None: 377 self._turn_tts_ttfb.record(float(tts_ttfb_s) * 1000.0, attributes=attrs) 378 if transcription_delay_s is not None: 379 self._turn_transcription_delay.record( 380 float(transcription_delay_s) * 1000.0, attributes=attrs 381 ) 382 if eou_delay_s is not None: 383 self._turn_eou_delay.record(float(eou_delay_s) * 1000.0, attributes=attrs) 384 if playback_latency_s is not None: 385 self._turn_playback_latency.record( 386 float(playback_latency_s) * 1000.0, attributes=attrs 387 ) 388 389 def record_session_close( 390 self, 391 *, 392 session_id: str, 393 turn_count: int = 0, 394 total_input_tokens: int = 0, 395 total_output_tokens: int = 0, 396 framework: str = "", 397 extra: Optional[Mapping[str, AttributeValue]] = None, 398 ) -> None: 399 if not session_id: 400 return 401 attrs = self._base_attrs( 402 session_id=session_id, framework=framework, extra=extra 403 ) 404 if turn_count: 405 self._session_turns.add(int(turn_count), attributes=attrs) 406 if total_input_tokens: 407 self._session_input_tokens.add(int(total_input_tokens), attributes=attrs) 408 if total_output_tokens: 409 self._session_output_tokens.add(int(total_output_tokens), attributes=attrs) 410 411 def force_flush(self, timeout_millis: int = 5_000) -> bool: 412 try: 413 return bool(self._meter_provider.force_flush(timeout_millis)) 414 except Exception: 415 return False 416 417 def shutdown(self, timeout_millis: int = 5_000) -> None: 418 """Flush and stop periodic export. Safe to call more than once.""" 419 try: 420 self.force_flush(timeout_millis) 421 except Exception: 422 pass 423 try: 424 self._meter_provider.shutdown(timeout_millis=timeout_millis) 425 except Exception: 426 logger.debug("MeterProvider shutdown failed", exc_info=True)
Records Parlot usage.*, turn.*, and session.* instruments.
ParlotMetricsRecorder( meter_provider: opentelemetry.sdk.metrics._internal.MeterProvider, *, meter_name: str = 'parlot.core', framework: str = '', tenant_id: str = '')
134 def __init__( 135 self, 136 meter_provider: MeterProvider, 137 *, 138 meter_name: str = "parlot.core", 139 framework: str = "", 140 tenant_id: str = "", 141 ) -> None: 142 self._meter_provider = meter_provider 143 self._framework = (framework or "").strip() 144 self._tenant_id = (tenant_id or "").strip() 145 meter = meter_provider.get_meter(meter_name) 146 self._meter = meter 147 148 self._usage_stt = meter.create_counter( 149 METRIC_USAGE_STT_AUDIO_DURATION, 150 unit="s", 151 description="STT audio duration per usage event", 152 ) 153 self._usage_tts_audio = meter.create_counter( 154 METRIC_USAGE_TTS_AUDIO_DURATION, 155 unit="s", 156 description="TTS audio duration per usage event", 157 ) 158 self._usage_tts_chars = meter.create_counter( 159 METRIC_USAGE_TTS_CHARACTERS, 160 description="TTS characters synthesized per usage event", 161 ) 162 self._usage_llm_in = meter.create_counter( 163 METRIC_USAGE_LLM_INPUT_TOKENS, 164 description="LLM input tokens per usage event", 165 ) 166 self._usage_llm_out = meter.create_counter( 167 METRIC_USAGE_LLM_OUTPUT_TOKENS, 168 description="LLM output tokens per usage event", 169 ) 170 171 self._turn_e2e = meter.create_histogram( 172 "turn.e2e_latency_ms", 173 unit="ms", 174 description="End-to-end latency for one agent turn", 175 ) 176 self._turn_llm_ttft = meter.create_histogram( 177 "turn.llm_ttft_ms", 178 unit="ms", 179 description="LLM time to first token for one agent turn", 180 ) 181 self._turn_tts_ttfb = meter.create_histogram( 182 "turn.tts_ttfb_ms", 183 unit="ms", 184 description="TTS time to first byte for one agent turn", 185 ) 186 self._turn_transcription_delay = meter.create_histogram( 187 "turn.transcription_delay_ms", 188 unit="ms", 189 description="Speech end to transcript available for one agent turn", 190 ) 191 self._turn_eou_delay = meter.create_histogram( 192 "turn.eou_delay_ms", 193 unit="ms", 194 description="End-of-utterance detection delay for one agent turn", 195 ) 196 self._turn_playback_latency = meter.create_histogram( 197 "turn.playback_latency_ms", 198 unit="ms", 199 description="Playback latency for one agent turn", 200 ) 201 202 self._session_turns = meter.create_counter( 203 "session.turn_count", 204 description="Total turns in a session", 205 ) 206 self._session_input_tokens = meter.create_counter( 207 "session.total_input_tokens", 208 description="Total input tokens in session", 209 ) 210 self._session_output_tokens = meter.create_counter( 211 "session.total_output_tokens", 212 description="Total output tokens in session", 213 )
def
record_stt_audio( self, *, session_id: str, audio_seconds: float, model: str = '', provider: str = '', framework: str = '', extra: Optional[Mapping[str, Union[str, bool, int, float, Sequence[str], Sequence[bool], Sequence[int], Sequence[float]]]] = None) -> None:
251 def record_stt_audio( 252 self, 253 *, 254 session_id: str, 255 audio_seconds: float, 256 model: str = "", 257 provider: str = "", 258 framework: str = "", 259 extra: Optional[Mapping[str, AttributeValue]] = None, 260 ) -> None: 261 if not session_id or audio_seconds <= 0: 262 return 263 self._usage_stt.add( 264 float(audio_seconds), 265 attributes=self._base_attrs( 266 session_id=session_id, 267 model=model, 268 provider=provider, 269 framework=framework, 270 extra=extra, 271 ), 272 )
def
record_tts_audio( self, *, session_id: str, audio_seconds: float, model: str = '', provider: str = '', framework: str = '', extra: Optional[Mapping[str, Union[str, bool, int, float, Sequence[str], Sequence[bool], Sequence[int], Sequence[float]]]] = None) -> None:
274 def record_tts_audio( 275 self, 276 *, 277 session_id: str, 278 audio_seconds: float, 279 model: str = "", 280 provider: str = "", 281 framework: str = "", 282 extra: Optional[Mapping[str, AttributeValue]] = None, 283 ) -> None: 284 if not session_id or audio_seconds <= 0: 285 return 286 self._usage_tts_audio.add( 287 float(audio_seconds), 288 attributes=self._base_attrs( 289 session_id=session_id, 290 model=model, 291 provider=provider, 292 framework=framework, 293 extra=extra, 294 ), 295 )
def
record_tts_characters( self, *, session_id: str, characters: int, model: str = '', provider: str = '', framework: str = '', extra: Optional[Mapping[str, Union[str, bool, int, float, Sequence[str], Sequence[bool], Sequence[int], Sequence[float]]]] = None) -> None:
297 def record_tts_characters( 298 self, 299 *, 300 session_id: str, 301 characters: int, 302 model: str = "", 303 provider: str = "", 304 framework: str = "", 305 extra: Optional[Mapping[str, AttributeValue]] = None, 306 ) -> None: 307 if not session_id or characters <= 0: 308 return 309 self._usage_tts_chars.add( 310 int(characters), 311 attributes=self._base_attrs( 312 session_id=session_id, 313 model=model, 314 provider=provider, 315 framework=framework, 316 extra=extra, 317 ), 318 )
def
record_llm_tokens( self, *, session_id: str, input_tokens: int = 0, output_tokens: int = 0, model: str = '', provider: str = '', framework: str = '', extra: Optional[Mapping[str, Union[str, bool, int, float, Sequence[str], Sequence[bool], Sequence[int], Sequence[float]]]] = None) -> None:
320 def record_llm_tokens( 321 self, 322 *, 323 session_id: str, 324 input_tokens: int = 0, 325 output_tokens: int = 0, 326 model: str = "", 327 provider: str = "", 328 framework: str = "", 329 extra: Optional[Mapping[str, AttributeValue]] = None, 330 ) -> None: 331 if not session_id: 332 return 333 attrs = self._base_attrs( 334 session_id=session_id, 335 model=model, 336 provider=provider, 337 framework=framework, 338 extra=extra, 339 ) 340 if input_tokens > 0: 341 self._usage_llm_in.add(int(input_tokens), attributes=attrs) 342 if output_tokens > 0: 343 self._usage_llm_out.add(int(output_tokens), attributes=attrs)
def
record_turn( self, *, session_id: str, turn_index: int, participant_role: str = 'agent', e2e_latency_s: Optional[float] = None, llm_ttft_s: Optional[float] = None, tts_ttfb_s: Optional[float] = None, transcription_delay_s: Optional[float] = None, eou_delay_s: Optional[float] = None, playback_latency_s: Optional[float] = None, interrupted: bool = False, framework: str = '', extra: Optional[Mapping[str, Union[str, bool, int, float, Sequence[str], Sequence[bool], Sequence[int], Sequence[float]]]] = None) -> None:
345 def record_turn( 346 self, 347 *, 348 session_id: str, 349 turn_index: int, 350 participant_role: str = "agent", 351 e2e_latency_s: Optional[float] = None, 352 llm_ttft_s: Optional[float] = None, 353 tts_ttfb_s: Optional[float] = None, 354 transcription_delay_s: Optional[float] = None, 355 eou_delay_s: Optional[float] = None, 356 playback_latency_s: Optional[float] = None, 357 interrupted: bool = False, 358 framework: str = "", 359 extra: Optional[Mapping[str, AttributeValue]] = None, 360 ) -> None: 361 if not session_id: 362 return 363 attrs = self._base_attrs( 364 session_id=session_id, 365 framework=framework, 366 participant_role=participant_role, 367 extra=extra, 368 ) 369 attrs[ATTR_TURN_INDEX] = int(turn_index) 370 if interrupted: 371 attrs[ATTR_TURN_INTERRUPTED] = True 372 if e2e_latency_s is not None: 373 self._turn_e2e.record(float(e2e_latency_s) * 1000.0, attributes=attrs) 374 if llm_ttft_s is not None: 375 self._turn_llm_ttft.record(float(llm_ttft_s) * 1000.0, attributes=attrs) 376 if tts_ttfb_s is not None: 377 self._turn_tts_ttfb.record(float(tts_ttfb_s) * 1000.0, attributes=attrs) 378 if transcription_delay_s is not None: 379 self._turn_transcription_delay.record( 380 float(transcription_delay_s) * 1000.0, attributes=attrs 381 ) 382 if eou_delay_s is not None: 383 self._turn_eou_delay.record(float(eou_delay_s) * 1000.0, attributes=attrs) 384 if playback_latency_s is not None: 385 self._turn_playback_latency.record( 386 float(playback_latency_s) * 1000.0, attributes=attrs 387 )
def
record_session_close( self, *, session_id: str, turn_count: int = 0, total_input_tokens: int = 0, total_output_tokens: int = 0, framework: str = '', extra: Optional[Mapping[str, Union[str, bool, int, float, Sequence[str], Sequence[bool], Sequence[int], Sequence[float]]]] = None) -> None:
389 def record_session_close( 390 self, 391 *, 392 session_id: str, 393 turn_count: int = 0, 394 total_input_tokens: int = 0, 395 total_output_tokens: int = 0, 396 framework: str = "", 397 extra: Optional[Mapping[str, AttributeValue]] = None, 398 ) -> None: 399 if not session_id: 400 return 401 attrs = self._base_attrs( 402 session_id=session_id, framework=framework, extra=extra 403 ) 404 if turn_count: 405 self._session_turns.add(int(turn_count), attributes=attrs) 406 if total_input_tokens: 407 self._session_input_tokens.add(int(total_input_tokens), attributes=attrs) 408 if total_output_tokens: 409 self._session_output_tokens.add(int(total_output_tokens), attributes=attrs)
def
shutdown(self, timeout_millis: int = 5000) -> None:
417 def shutdown(self, timeout_millis: int = 5_000) -> None: 418 """Flush and stop periodic export. Safe to call more than once.""" 419 try: 420 self.force_flush(timeout_millis) 421 except Exception: 422 pass 423 try: 424 self._meter_provider.shutdown(timeout_millis=timeout_millis) 425 except Exception: 426 logger.debug("MeterProvider shutdown failed", exc_info=True)
Flush and stop periodic export. Safe to call more than once.
class
QuietOTLPMetricExporter(opentelemetry.sdk.metrics._internal.export.MetricExporter):
51class QuietOTLPMetricExporter(MetricExporter): 52 """Wraps an OTLP metric exporter with one-line warning logs on failure.""" 53 54 def __init__( 55 self, 56 exporter: MetricExporter, 57 *, 58 endpoint_label: str, 59 ) -> None: 60 super().__init__( 61 preferred_temporality=exporter._preferred_temporality, 62 preferred_aggregation=exporter._preferred_aggregation, 63 ) 64 self._exporter = exporter 65 self._endpoint_label = endpoint_label 66 67 def export( 68 self, 69 metrics_data: MetricsData, 70 timeout_millis: float = 10_000, 71 **kwargs, 72 ) -> MetricExportResult: 73 try: 74 result = self._exporter.export( 75 metrics_data, timeout_millis=timeout_millis, **kwargs 76 ) 77 if result == MetricExportResult.FAILURE: 78 logger.warning( 79 "Parlot metrics export returned FAILURE (%s)", 80 self._endpoint_label, 81 ) 82 return result 83 except Exception as exc: 84 msg = str(exc).strip() or exc.__class__.__name__ 85 if len(msg) > 240: 86 msg = msg[:237] + "..." 87 logger.warning( 88 "Parlot metrics export failed (%s): %s", 89 self._endpoint_label, 90 msg, 91 ) 92 return MetricExportResult.FAILURE 93 94 def force_flush(self, timeout_millis: float = 10_000) -> bool: 95 return self._exporter.force_flush(timeout_millis=timeout_millis) 96 97 def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None: 98 self._exporter.shutdown(timeout_millis=timeout_millis, **kwargs)
Wraps an OTLP metric exporter with one-line warning logs on failure.
QuietOTLPMetricExporter( exporter: opentelemetry.sdk.metrics._internal.export.MetricExporter, *, endpoint_label: str)
54 def __init__( 55 self, 56 exporter: MetricExporter, 57 *, 58 endpoint_label: str, 59 ) -> None: 60 super().__init__( 61 preferred_temporality=exporter._preferred_temporality, 62 preferred_aggregation=exporter._preferred_aggregation, 63 ) 64 self._exporter = exporter 65 self._endpoint_label = endpoint_label
def
export( self, metrics_data: opentelemetry.sdk.metrics._internal.point.MetricsData, timeout_millis: float = 10000, **kwargs) -> opentelemetry.sdk.metrics._internal.export.MetricExportResult:
67 def export( 68 self, 69 metrics_data: MetricsData, 70 timeout_millis: float = 10_000, 71 **kwargs, 72 ) -> MetricExportResult: 73 try: 74 result = self._exporter.export( 75 metrics_data, timeout_millis=timeout_millis, **kwargs 76 ) 77 if result == MetricExportResult.FAILURE: 78 logger.warning( 79 "Parlot metrics export returned FAILURE (%s)", 80 self._endpoint_label, 81 ) 82 return result 83 except Exception as exc: 84 msg = str(exc).strip() or exc.__class__.__name__ 85 if len(msg) > 240: 86 msg = msg[:237] + "..." 87 logger.warning( 88 "Parlot metrics export failed (%s): %s", 89 self._endpoint_label, 90 msg, 91 ) 92 return MetricExportResult.FAILURE
Exports a batch of telemetry data.
Arguments:
- metrics: The list of
opentelemetry.sdk.metrics.export.Metricobjects to be exported
Returns:
The result of the export
UsageMetricsRecorder =
<class 'ParlotMetricsRecorder'>
def
assert_sync_span_processors(provider) -> bool:
109def assert_sync_span_processors(provider) -> bool: 110 """Require ``SynchronousMultiSpanProcessor`` on the provider. 111 112 Parlot processors rely on synchronous ``on_start`` / ``on_end`` ordering and 113 often on ``contextvars`` copied at task creation. ``ConcurrentMultiSpanProcessor`` 114 runs handlers on worker threads and breaks that model. 115 116 Returns ``False`` when ``ConcurrentMultiSpanProcessor`` is active (callers 117 should disable OTel ``context.attach`` and similar). Raises ``RuntimeError`` 118 for unknown processor layouts. 119 120 Uses OTel SDK-private ``_active_span_processor`` — re-validate on SDK upgrades. 121 """ 122 from opentelemetry.sdk.trace import ( 123 ConcurrentMultiSpanProcessor, 124 SynchronousMultiSpanProcessor, 125 ) 126 127 asp = getattr(provider, "_active_span_processor", None) 128 if asp is None: 129 raise RuntimeError( 130 "parlot: TracerProvider has no _active_span_processor; " 131 "cannot verify synchronous span processing" 132 ) 133 if isinstance(asp, ConcurrentMultiSpanProcessor): 134 logger.error( 135 "parlot: ConcurrentMultiSpanProcessor detected — " 136 "unsupported for Parlot span processors (use synchronous layout)" 137 ) 138 return False 139 if not isinstance(asp, SynchronousMultiSpanProcessor): 140 raise RuntimeError( 141 f"parlot: unsupported active span processor {type(asp).__name__!r}; " 142 "expected SynchronousMultiSpanProcessor" 143 ) 144 return True
Require SynchronousMultiSpanProcessor on the provider.
Parlot processors rely on synchronous on_start / on_end ordering and
often on contextvars copied at task creation. ConcurrentMultiSpanProcessor
runs handlers on worker threads and breaks that model.
Returns False when ConcurrentMultiSpanProcessor is active (callers
should disable OTel context.attach and similar). Raises RuntimeError
for unknown processor layouts.
Uses OTel SDK-private _active_span_processor — re-validate on SDK upgrades.
@dataclass
class
SessionState:
20@dataclass 21class SessionState: 22 """ 23 Per-trace accumulator for cross-span aggregates. 24 25 Framework-specific instrumentation packages subclass this to add their own 26 fields (e.g. LiveKit adds agent_chain, pending_handoff_end_ns). 27 """ 28 29 session_id: str = "" 30 room_name: str = "" 31 room_sid: str = "" 32 agent_label: str = "" 33 34 turn_count: int = 0 35 tool_call_count: int = 0 36 handoff_count: int = 0 37 38 total_input_tokens: int = 0 39 total_output_tokens: int = 0 40 total_cached_tokens: int = 0 41 total_cost_usd: float = 0.0 42 43 human_rep_participant_ids: set[str] = field(default_factory=set) 44 topology_agents: list[dict[str, Any]] = field(default_factory=list) 45 custom_metadata: dict[str, str] = field(default_factory=dict) 46 # Owning instrumentation package, e.g. "livekit" | "langgraph" 47 framework: str = "" 48 # Sample-0 wall ms for media_segment_* (first write wins via set_recording_anchor_wall_ms). 49 recording_anchor_wall_ms: int | None = None
Per-trace accumulator for cross-span aggregates.
Framework-specific instrumentation packages subclass this to add their own fields (e.g. LiveKit adds agent_chain, pending_handoff_end_ns).
SessionState( session_id: str = '', room_name: str = '', room_sid: str = '', agent_label: str = '', turn_count: int = 0, tool_call_count: int = 0, handoff_count: int = 0, total_input_tokens: int = 0, total_output_tokens: int = 0, total_cached_tokens: int = 0, total_cost_usd: float = 0.0, human_rep_participant_ids: set[str] = <factory>, topology_agents: list[dict[str, typing.Any]] = <factory>, custom_metadata: dict[str, str] = <factory>, framework: str = '', recording_anchor_wall_ms: int | None = None)
@dataclass
class
SessionTopology:
111@dataclass 112class SessionTopology: 113 """ 114 Accumulates intent segments and agent instructions during a live session. 115 116 Framework adapters call ``record_instructions`` / ``open_segment_after_handoff`` 117 during the session and ``stamp_session_span`` at close. 118 """ 119 120 prompts_by_agent: dict[str, str] = field(default_factory=dict) 121 intent_by_agent: dict[str, dict[str, str]] = field(default_factory=dict) 122 # Expected to be a small fixed agent roster (not session-/user-suffixed ids). 123 agents_seen: dict[str, dict[str, Any]] = field(default_factory=dict) 124 intent_segments: list[_MutableSegment] = field(default_factory=list) 125 active_segment: _MutableSegment | None = None 126 pending_segment_from_turn: int | None = None 127 first_agent_label: str = "" 128 agent_chain: list[str] = field(default_factory=list) 129 handoff_count: int = 0 130 _handoff_keys: set[str] = field(default_factory=set) 131 _handoff_keys_turn: int | None = None 132 133 def upsert_agent(self, agent_id: str) -> None: 134 aid = agent_id.strip() 135 if not aid: 136 return 137 if not self.first_agent_label: 138 self.first_agent_label = aid 139 if aid not in self.agents_seen: 140 self.agents_seen[aid] = {"id": aid, "role": "agent"} 141 142 def record_instructions(self, agent_id: str, text: str) -> None: 143 # Keep the longer excerpt so a later shorter capture does not shrink the preview. 144 aid = agent_id.strip() 145 if not aid or not text.strip(): 146 return 147 excerpt = _truncate(text.strip(), INSTRUCTIONS_PREVIEW) 148 prev = self.prompts_by_agent.get(aid, "") 149 if len(excerpt) >= len(prev): 150 self.prompts_by_agent[aid] = excerpt 151 self._sync_agent_intent_summary(aid) 152 self._refresh_open_segment_instructions(aid) 153 154 def push_agent_chain(self, agent_id: str) -> None: 155 aid = agent_id.strip() 156 if not aid: 157 return 158 if not self.agent_chain or self.agent_chain[-1] != aid: 159 self.agent_chain.append(aid) 160 trim_agent_chain(self.agent_chain) 161 162 def open_bootstrap_segment(self, agent_id: str, from_turn: int) -> None: 163 if self.active_segment is not None: 164 return 165 aid = agent_id.strip() or "unknown" 166 self.upsert_agent(aid) 167 self.active_segment = self._new_segment(aid, from_turn, 0) 168 169 def open_segment_after_handoff( 170 self, 171 agent_id: str, 172 *, 173 handoff_index: int | None = None, 174 turn_index: int, 175 from_agent: str = "", 176 ) -> None: 177 aid = agent_id.strip() 178 if not aid: 179 return 180 if self._handoff_keys_turn != turn_index: 181 self._handoff_keys.clear() 182 self._handoff_keys_turn = turn_index 183 dedupe_key = f"{from_agent.strip()}->{aid}@{turn_index}" 184 if dedupe_key in self._handoff_keys: 185 return 186 self._handoff_keys.add(dedupe_key) 187 188 self.close_active_segment(turn_index) 189 self.upsert_agent(aid) 190 self.handoff_count += 1 191 effective_handoff = handoff_index if handoff_index is not None else self.handoff_count 192 if not self.intent_segments and self.active_segment is None: 193 effective_handoff = 0 194 self.pending_segment_from_turn = max(1, turn_index + 1) 195 self.active_segment = self._new_segment( 196 aid, 197 self.pending_segment_from_turn, 198 effective_handoff, 199 ) 200 201 def close_active_segment(self, to_turn: int) -> None: 202 if self.active_segment is None: 203 return 204 self.active_segment.to_turn = max(0, to_turn) 205 closed = self.active_segment 206 self.active_segment = None 207 if self.intent_segments and _merge_same_agent_into(self.intent_segments[-1], closed): 208 self._trim_closed_segments() 209 return 210 self.intent_segments.append(closed) 211 self._trim_closed_segments() 212 213 def _trim_closed_segments(self) -> None: 214 """Drop oldest closed segments only; never touches ``active_segment``.""" 215 overflow = len(self.intent_segments) - MAX_INTENT_SEGMENTS 216 if overflow > 0: 217 del self.intent_segments[:overflow] 218 219 def apply_pending_from_turn_on_emit(self, turn_index: int) -> None: 220 if self.active_segment is None: 221 return 222 if self.pending_segment_from_turn is not None: 223 self.active_segment.from_turn = self.pending_segment_from_turn 224 self.pending_segment_from_turn = None 225 elif self.active_segment.from_turn <= 0: 226 self.active_segment.from_turn = turn_index 227 228 def finalize_intent_sequence(self, final_turn: int) -> list[dict[str, Any]]: 229 if self.active_segment is not None: 230 self.close_active_segment(final_turn) 231 return self._collapse_segments( 232 [ 233 { 234 "segment_index": idx, 235 "agent_id": seg.agent_id, 236 "intent_key": seg.intent_key, 237 "intent_label": seg.intent_label, 238 "from_turn": seg.from_turn, 239 "handoff_index": seg.handoff_index, 240 "source": seg.source, 241 **( 242 {"to_turn": seg.to_turn} 243 if seg.to_turn is not None 244 else {} 245 ), 246 **( 247 {"instructions_excerpt": seg.instructions_excerpt} 248 if seg.instructions_excerpt 249 else {} 250 ), 251 } 252 for idx, seg in enumerate(self.intent_segments) 253 ] 254 ) 255 256 def bootstrap_instructions(self) -> str: 257 if self.first_agent_label: 258 text = self.prompts_by_agent.get(self.first_agent_label, "") 259 if text: 260 return text 261 if self.prompts_by_agent: 262 return max(self.prompts_by_agent.values(), key=len) 263 return "" 264 265 def agents_json(self) -> list[dict[str, Any]]: 266 out: list[dict[str, Any]] = [] 267 for node in self.agents_seen.values(): 268 summary = self.intent_by_agent.get(node["id"], {}) 269 entry = dict(node) 270 if summary: 271 entry["intent_key"] = summary.get("intent_key", node["id"]) 272 entry["intent_label"] = summary.get("intent_label", node["id"]) 273 excerpt = summary.get("instructions_excerpt", "") 274 if excerpt: 275 entry["instructions_excerpt"] = excerpt 276 out.append(entry) 277 return out 278 279 def stamp_session_span(self, span: Any, *, final_turn: int) -> None: 280 """Write topology attrs on the live ``parlot.session`` span.""" 281 if span is None or not hasattr(span, "set_attribute"): 282 return 283 sequence = self.finalize_intent_sequence(final_turn) 284 agents = self.agents_json() 285 bootstrap = self.bootstrap_instructions() 286 if agents: 287 agents_json, agents_truncated = _json_dumps_cap(agents) 288 span.set_attribute(ATTR_SESSION_TOPOLOGY_AGENTS, agents_json) 289 if agents_truncated: 290 span.set_attribute(ATTR_SESSION_TOPOLOGY_AGENTS_TRUNCATED, True) 291 if bootstrap: 292 span.set_attribute(ATTR_SESSION_TOPOLOGY_BOOTSTRAP_INSTRUCTIONS, bootstrap) 293 if sequence: 294 seq_json, seq_truncated = _json_dumps_cap(sequence) 295 span.set_attribute(ATTR_SESSION_INTENT_SEQUENCE, seq_json) 296 if seq_truncated: 297 span.set_attribute(ATTR_SESSION_INTENT_SEQUENCE_TRUNCATED, True) 298 if self.agent_chain: 299 chain = " → ".join(self.agent_chain) 300 span.set_attribute(ATTR_SESSION_AGENT_CHAIN, _truncate(chain, MAX_JSON_CHARS)) 301 if len(chain) > MAX_JSON_CHARS: 302 span.set_attribute(ATTR_SESSION_AGENT_CHAIN_TRUNCATED, True) 303 304 def _new_segment(self, agent_id: str, from_turn: int, handoff_index: int) -> _MutableSegment: 305 instructions = self.prompts_by_agent.get(agent_id, "") 306 derived = derive_intent(agent_id, instructions) 307 self._sync_agent_intent_summary(agent_id) 308 return _MutableSegment( 309 segment_index=0, 310 agent_id=agent_id, 311 intent_key=derived["intent_key"], 312 intent_label=derived["intent_label"], 313 from_turn=from_turn, 314 handoff_index=handoff_index, 315 instructions_excerpt=derived.get("instructions_excerpt", ""), 316 ) 317 318 def _sync_agent_intent_summary(self, agent_id: str) -> None: 319 instructions = self.prompts_by_agent.get(agent_id, "") 320 derived = derive_intent(agent_id, instructions) 321 self.intent_by_agent[agent_id] = derived 322 node = self.agents_seen.get(agent_id) 323 if node is not None: 324 node["intent_key"] = derived["intent_key"] 325 node["intent_label"] = derived["intent_label"] 326 if derived.get("instructions_excerpt"): 327 node["instructions_excerpt"] = derived["instructions_excerpt"] 328 329 def _refresh_open_segment_instructions(self, agent_id: str) -> None: 330 if self.active_segment is None or self.active_segment.agent_id != agent_id.strip(): 331 return 332 instructions = self.prompts_by_agent.get(agent_id, "") 333 derived = derive_intent(agent_id, instructions) 334 self.active_segment.intent_key = derived["intent_key"] 335 self.active_segment.intent_label = derived["intent_label"] 336 self.active_segment.instructions_excerpt = derived.get("instructions_excerpt", "") 337 self._sync_agent_intent_summary(agent_id) 338 339 @staticmethod 340 def _collapse_segments(segments: list[dict[str, Any]]) -> list[dict[str, Any]]: 341 if not segments: 342 return [] 343 merged: list[dict[str, Any]] = [] 344 for seg in segments: 345 if not merged: 346 merged.append(dict(seg)) 347 continue 348 prev = merged[-1] 349 if _merge_same_agent_into(prev, seg): 350 continue 351 merged.append(dict(seg)) 352 for idx, seg in enumerate(merged): 353 seg["segment_index"] = idx 354 return merged
Accumulates intent segments and agent instructions during a live session.
Framework adapters call record_instructions / open_segment_after_handoff
during the session and stamp_session_span at close.
SessionTopology( prompts_by_agent: dict[str, str] = <factory>, intent_by_agent: dict[str, dict[str, str]] = <factory>, agents_seen: dict[str, dict[str, typing.Any]] = <factory>, intent_segments: list[parlot.core.topology._MutableSegment] = <factory>, active_segment: parlot.core.topology._MutableSegment | None = None, pending_segment_from_turn: int | None = None, first_agent_label: str = '', agent_chain: list[str] = <factory>, handoff_count: int = 0, _handoff_keys: set[str] = <factory>, _handoff_keys_turn: int | None = None)
def
record_instructions(self, agent_id: str, text: str) -> None:
142 def record_instructions(self, agent_id: str, text: str) -> None: 143 # Keep the longer excerpt so a later shorter capture does not shrink the preview. 144 aid = agent_id.strip() 145 if not aid or not text.strip(): 146 return 147 excerpt = _truncate(text.strip(), INSTRUCTIONS_PREVIEW) 148 prev = self.prompts_by_agent.get(aid, "") 149 if len(excerpt) >= len(prev): 150 self.prompts_by_agent[aid] = excerpt 151 self._sync_agent_intent_summary(aid) 152 self._refresh_open_segment_instructions(aid)
def
open_segment_after_handoff( self, agent_id: str, *, handoff_index: int | None = None, turn_index: int, from_agent: str = '') -> None:
169 def open_segment_after_handoff( 170 self, 171 agent_id: str, 172 *, 173 handoff_index: int | None = None, 174 turn_index: int, 175 from_agent: str = "", 176 ) -> None: 177 aid = agent_id.strip() 178 if not aid: 179 return 180 if self._handoff_keys_turn != turn_index: 181 self._handoff_keys.clear() 182 self._handoff_keys_turn = turn_index 183 dedupe_key = f"{from_agent.strip()}->{aid}@{turn_index}" 184 if dedupe_key in self._handoff_keys: 185 return 186 self._handoff_keys.add(dedupe_key) 187 188 self.close_active_segment(turn_index) 189 self.upsert_agent(aid) 190 self.handoff_count += 1 191 effective_handoff = handoff_index if handoff_index is not None else self.handoff_count 192 if not self.intent_segments and self.active_segment is None: 193 effective_handoff = 0 194 self.pending_segment_from_turn = max(1, turn_index + 1) 195 self.active_segment = self._new_segment( 196 aid, 197 self.pending_segment_from_turn, 198 effective_handoff, 199 )
def
close_active_segment(self, to_turn: int) -> None:
201 def close_active_segment(self, to_turn: int) -> None: 202 if self.active_segment is None: 203 return 204 self.active_segment.to_turn = max(0, to_turn) 205 closed = self.active_segment 206 self.active_segment = None 207 if self.intent_segments and _merge_same_agent_into(self.intent_segments[-1], closed): 208 self._trim_closed_segments() 209 return 210 self.intent_segments.append(closed) 211 self._trim_closed_segments()
def
apply_pending_from_turn_on_emit(self, turn_index: int) -> None:
219 def apply_pending_from_turn_on_emit(self, turn_index: int) -> None: 220 if self.active_segment is None: 221 return 222 if self.pending_segment_from_turn is not None: 223 self.active_segment.from_turn = self.pending_segment_from_turn 224 self.pending_segment_from_turn = None 225 elif self.active_segment.from_turn <= 0: 226 self.active_segment.from_turn = turn_index
def
finalize_intent_sequence(self, final_turn: int) -> list[dict[str, typing.Any]]:
228 def finalize_intent_sequence(self, final_turn: int) -> list[dict[str, Any]]: 229 if self.active_segment is not None: 230 self.close_active_segment(final_turn) 231 return self._collapse_segments( 232 [ 233 { 234 "segment_index": idx, 235 "agent_id": seg.agent_id, 236 "intent_key": seg.intent_key, 237 "intent_label": seg.intent_label, 238 "from_turn": seg.from_turn, 239 "handoff_index": seg.handoff_index, 240 "source": seg.source, 241 **( 242 {"to_turn": seg.to_turn} 243 if seg.to_turn is not None 244 else {} 245 ), 246 **( 247 {"instructions_excerpt": seg.instructions_excerpt} 248 if seg.instructions_excerpt 249 else {} 250 ), 251 } 252 for idx, seg in enumerate(self.intent_segments) 253 ] 254 )
def
agents_json(self) -> list[dict[str, typing.Any]]:
265 def agents_json(self) -> list[dict[str, Any]]: 266 out: list[dict[str, Any]] = [] 267 for node in self.agents_seen.values(): 268 summary = self.intent_by_agent.get(node["id"], {}) 269 entry = dict(node) 270 if summary: 271 entry["intent_key"] = summary.get("intent_key", node["id"]) 272 entry["intent_label"] = summary.get("intent_label", node["id"]) 273 excerpt = summary.get("instructions_excerpt", "") 274 if excerpt: 275 entry["instructions_excerpt"] = excerpt 276 out.append(entry) 277 return out
def
stamp_session_span(self, span: Any, *, final_turn: int) -> None:
279 def stamp_session_span(self, span: Any, *, final_turn: int) -> None: 280 """Write topology attrs on the live ``parlot.session`` span.""" 281 if span is None or not hasattr(span, "set_attribute"): 282 return 283 sequence = self.finalize_intent_sequence(final_turn) 284 agents = self.agents_json() 285 bootstrap = self.bootstrap_instructions() 286 if agents: 287 agents_json, agents_truncated = _json_dumps_cap(agents) 288 span.set_attribute(ATTR_SESSION_TOPOLOGY_AGENTS, agents_json) 289 if agents_truncated: 290 span.set_attribute(ATTR_SESSION_TOPOLOGY_AGENTS_TRUNCATED, True) 291 if bootstrap: 292 span.set_attribute(ATTR_SESSION_TOPOLOGY_BOOTSTRAP_INSTRUCTIONS, bootstrap) 293 if sequence: 294 seq_json, seq_truncated = _json_dumps_cap(sequence) 295 span.set_attribute(ATTR_SESSION_INTENT_SEQUENCE, seq_json) 296 if seq_truncated: 297 span.set_attribute(ATTR_SESSION_INTENT_SEQUENCE_TRUNCATED, True) 298 if self.agent_chain: 299 chain = " → ".join(self.agent_chain) 300 span.set_attribute(ATTR_SESSION_AGENT_CHAIN, _truncate(chain, MAX_JSON_CHARS)) 301 if len(chain) > MAX_JSON_CHARS: 302 span.set_attribute(ATTR_SESSION_AGENT_CHAIN_TRUNCATED, True)
Write topology attrs on the live parlot.session span.
def
add_platform_ref(kind: str, value: str, *, framework: str = 'custom') -> None:
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")
def
adopt_existing_tracer_provider() -> typing.Any | None:
124def adopt_existing_tracer_provider() -> Any | None: 125 """Return the global tracer provider when already set by another Parlot package. 126 127 Returns ``None`` when only the default no-op / proxy provider is installed. 128 """ 129 from opentelemetry import trace 130 from opentelemetry.sdk.trace import TracerProvider as SdkTracerProvider 131 132 provider = trace.get_tracer_provider() 133 if isinstance(provider, SdkTracerProvider): 134 return provider 135 return None
Return the global tracer provider when already set by another Parlot package.
Returns None when only the default no-op / proxy provider is installed.
def
build_otlp_http_exporter( *, endpoint: str, api_key: str = '') -> opentelemetry.sdk.trace.export.SpanExporter:
71def build_otlp_http_exporter( 72 *, 73 endpoint: str, 74 api_key: str = "", 75) -> SpanExporter: 76 """Build an OTLP/HTTP span exporter targeting ``{endpoint}/v1/traces``.""" 77 from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter 78 79 if not endpoint: 80 raise ValueError( 81 "No OTLP endpoint configured. Pass endpoint= or set the " 82 "PARLOT_ENDPOINT environment variable." 83 ) 84 headers = build_parlot_client_headers(api_key) 85 trace_endpoint = endpoint.rstrip("/") + "/v1/traces" 86 return OTLPSpanExporter(endpoint=trace_endpoint, headers=headers)
Build an OTLP/HTTP span exporter targeting {endpoint}/v1/traces.
def
build_parlot_client_headers(api_key: str = '') -> dict[str, str]:
56def build_parlot_client_headers(api_key: str = "") -> dict[str, str]: 57 """Build standard transport headers sent with Parlot telemetry exports.""" 58 from parlot.core.sdk_version import resolve_parlot_sdk_version 59 60 version = resolve_parlot_sdk_version() or "unknown" 61 headers: dict[str, str] = { 62 HEADER_SDK_NAME: "parlot-python", 63 HEADER_SDK_VERSION: version, 64 HEADER_INGESTION_VERSION: INGESTION_PROTOCOL_VERSION, 65 } 66 if api_key: 67 headers["Authorization"] = f"Bearer {api_key}" 68 return headers
Build standard transport headers sent with Parlot telemetry exports.
def
build_parlot_meter_provider( endpoint: str, resource: opentelemetry.sdk.resources.Resource, *, api_key: str = '', headers: Optional[Mapping[str, str]] = None, export_interval_millis: int = 5000, quiet: bool = True) -> opentelemetry.sdk.metrics._internal.MeterProvider:
101def build_parlot_meter_provider( 102 endpoint: str, 103 resource: Resource, 104 *, 105 api_key: str = "", 106 headers: Optional[Mapping[str, str]] = None, 107 export_interval_millis: int = DEFAULT_METRICS_EXPORT_INTERVAL_MS, 108 quiet: bool = True, 109) -> MeterProvider: 110 """Build a dedicated MeterProvider exporting Parlot metrics. 111 112 Do **not** register this as the global meter provider when the host 113 framework also emits its own meters (LiveKit ``lk.agents.usage.*``). 114 """ 115 metrics_endpoint = endpoint.rstrip("/") + "/v1/metrics" 116 resolved_headers = ( 117 dict(headers) if headers is not None else build_parlot_client_headers(api_key) 118 ) 119 raw = OTLPMetricExporter(endpoint=metrics_endpoint, headers=resolved_headers) 120 exporter: MetricExporter = ( 121 QuietOTLPMetricExporter(raw, endpoint_label=metrics_endpoint) 122 if quiet 123 else raw 124 ) 125 reader = PeriodicExportingMetricReader( 126 exporter, export_interval_millis=int(export_interval_millis) 127 ) 128 return MeterProvider(metric_readers=[reader], resource=resource)
Build a dedicated MeterProvider exporting Parlot metrics.
Do not register this as the global meter provider when the host
framework also emits its own meters (LiveKit lk.agents.usage.*).
def
build_resource( *, service_name: Optional[str] = None, service_version: Optional[str] = None) -> opentelemetry.sdk.resources.Resource:
39def build_resource( 40 *, 41 service_name: Optional[str] = None, 42 service_version: Optional[str] = None, 43) -> Resource: 44 attrs = {service_attributes.SERVICE_NAME: service_name or "unknown"} 45 if service_version: 46 attrs[service_attributes.SERVICE_VERSION] = service_version 47 return Resource.create(attrs)
def
build_tracer_provider( *, endpoint: Optional[str] = None, api_key: Optional[str] = None, service_name: Optional[str] = None, service_version: Optional[str] = None, span_processors: Sequence[opentelemetry.sdk.trace.SpanProcessor] = (), span_exporter: opentelemetry.sdk.trace.export.SpanExporter | None = None) -> opentelemetry.sdk.trace.TracerProvider:
89def build_tracer_provider( 90 *, 91 endpoint: Optional[str] = None, 92 api_key: Optional[str] = None, 93 service_name: Optional[str] = None, 94 service_version: Optional[str] = None, 95 span_processors: Sequence[SpanProcessor] = (), 96 span_exporter: SpanExporter | None = None, 97) -> TracerProvider: 98 """Build a ``TracerProvider`` with optional processors and OTLP export. 99 100 When ``span_exporter`` is omitted, builds a default OTLP/HTTP exporter to 101 ``{endpoint}/v1/traces``. Callers typically wrap that exporter (filter, 102 sanitize, quiet) before passing it here. 103 """ 104 resolved_endpoint = resolve_endpoint(endpoint) 105 resolved_api_key = resolve_api_key(api_key) 106 resource = build_resource( 107 service_name=service_name, 108 service_version=service_version, 109 ) 110 exporter = span_exporter 111 if exporter is None: 112 exporter = build_otlp_http_exporter( 113 endpoint=resolved_endpoint, 114 api_key=resolved_api_key, 115 ) 116 117 provider = TracerProvider(resource=resource) 118 for processor in span_processors: 119 provider.add_span_processor(processor) 120 provider.add_span_processor(BatchSpanProcessor(exporter)) 121 return provider
Build a TracerProvider with optional processors and OTLP export.
When span_exporter is omitted, builds a default OTLP/HTTP exporter to
{endpoint}/v1/traces. Callers typically wrap that exporter (filter,
sanitize, quiet) before passing it here.
HEADER_SDK_NAME =
'x-parlot-sdk-name'
HEADER_SDK_VERSION =
'x-parlot-sdk-version'
HEADER_INGESTION_VERSION =
'x-parlot-ingestion-version'
INGESTION_PROTOCOL_VERSION =
'1'
def
clear_active_session() -> None:
71def clear_active_session() -> None: 72 """Clear the active session bindings for this context.""" 73 set_active_session(None, None)
Clear the active session bindings for this context.
def
derive_intent(agent_id: str, instructions: str = '') -> dict[str, str]:
36def derive_intent(agent_id: str, instructions: str = "") -> dict[str, str]: 37 """Port of platform ingest ``deriveIntent``.""" 38 aid = agent_id.strip() 39 intent_key = _normalize_key(aid) if aid else "unknown" 40 excerpt = _truncate(instructions.strip(), INSTRUCTIONS_EXCERPT_MAX) if instructions.strip() else "" 41 42 if excerpt: 43 first = _first_line(excerpt) 44 intent_label = _truncate(first, INTENT_LABEL_MAX) if first else _title_from_id(intent_key) 45 else: 46 intent_label = _title_from_id(intent_key) 47 48 return { 49 "intent_key": intent_key, 50 "intent_label": intent_label, 51 "instructions_excerpt": excerpt, 52 }
Port of platform ingest deriveIntent.
def
fetch_telemetry_bootstrap( endpoint: str, api_key: str, context: ParlotContext, *, timeout: float = 15.0) -> Optional[dict[str, Any]]:
15def fetch_telemetry_bootstrap( 16 endpoint: str, 17 api_key: str, 18 context: ParlotContext, 19 *, 20 timeout: float = 15.0, 21) -> Optional[dict[str, Any]]: 22 """GET ``/v1/telemetry/bootstrap`` and store runtime on ``context``, or None.""" 23 if not endpoint or not api_key: 24 return None 25 26 import httpx 27 28 from parlot.core.runtime import runtime_from_bootstrap 29 30 url = f"{endpoint.rstrip('/')}/v1/telemetry/bootstrap" 31 headers = {"Authorization": f"Bearer {api_key}"} 32 try: 33 with httpx.Client(timeout=timeout) as client: 34 resp = client.get(url, headers=headers) 35 if resp.status_code >= 400: 36 logger.error( 37 "parlot: telemetry bootstrap failed status=%s endpoint=%s", 38 resp.status_code, 39 endpoint.rstrip("/"), 40 ) 41 return None 42 payload = resp.json() 43 if not isinstance(payload, dict): 44 logger.error( 45 "parlot: telemetry bootstrap returned non-object JSON endpoint=%s", 46 endpoint.rstrip("/"), 47 ) 48 return None 49 context.runtime = runtime_from_bootstrap(endpoint, api_key, payload) 50 logger.info( 51 "parlot: telemetry bootstrap ok (endpoint=%s)", 52 endpoint.rstrip("/"), 53 ) 54 return payload 55 except Exception as exc: 56 logger.error( 57 "parlot: telemetry bootstrap request failed — %s (endpoint=%s)", 58 exc, 59 endpoint.rstrip("/"), 60 ) 61 logger.debug("parlot: telemetry bootstrap request failed", exc_info=True) 62 return None
GET /v1/telemetry/bootstrap and store runtime on context, or None.
52def get_active_session() -> SessionState | None: 53 """Return the session state bound to the current context, if any.""" 54 return _active_session_state.get()
Return the session state bound to the current context, if any.
def
get_active_session_span() -> opentelemetry.trace.span.Span | None:
57def get_active_session_span() -> Span | None: 58 """Return the active ``parlot.session`` span for the current context, if any.""" 59 return _active_session_span.get()
Return the active parlot.session span for the current context, if any.
@contextmanager
def
human_escalation(label: str | None = None) -> Iterator[NoneType]:
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:
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
remap_native_span_name( native: str, attrs: Optional[Mapping[str, Union[str, bool, int, float, Sequence[str], Sequence[bool], Sequence[int], Sequence[float]]]] = None, *, llm_names: Collection[str] = (), tool_names: Collection[str] = (), tts_names: Collection[str] = (), stt_names: Collection[str] = (), drop_names: Collection[str] = (), tool_name_attrs: Collection[str] = ('gen_ai.tool.name',), unknown: Literal['drop', 'passthrough'] = 'drop', custom: Optional[Callable[[str, Mapping[str, Union[str, bool, int, float, Sequence[str], Sequence[bool], Sequence[int], Sequence[float]]]], str | None | object]] = None) -> str | None:
30def remap_native_span_name( 31 native: str, 32 attrs: Mapping[str, AttributeValue] | None = None, 33 *, 34 llm_names: Collection[str] = (), 35 tool_names: Collection[str] = (), 36 tts_names: Collection[str] = (), 37 stt_names: Collection[str] = (), 38 drop_names: Collection[str] = (), 39 tool_name_attrs: Collection[str] = (ATTR_GEN_AI_TOOL_NAME,), 40 unknown: UnknownPolicy = "drop", 41 custom: Callable[ 42 [str, Mapping[str, AttributeValue]], str | None | object 43 ] 44 | None = None, 45) -> str | None: 46 """Map a vendor-native span name to a GenAI / voice export name. 47 48 Returns ``None`` to drop the span from the export name path. 49 50 ``custom`` may return a string, ``None`` (drop), or the sentinel 51 ``NotImplemented`` to fall through to the table-driven rules. 52 """ 53 attrs = attrs or {} 54 native_name = native or "" 55 56 if custom is not None: 57 result = custom(native_name, attrs) 58 if result is not NotImplemented: 59 return result # type: ignore[return-value] 60 61 if native_name in drop_names: 62 return None 63 if native_name in llm_names: 64 model = str(attrs.get(ATTR_GEN_AI_MODEL) or "").strip() or None 65 return span_name_chat(model) 66 if native_name in tool_names: 67 tool = "" 68 for key in tool_name_attrs: 69 tool = str(attrs.get(key) or "").strip() 70 if tool: 71 break 72 return span_name_execute_tool(tool) 73 if native_name in tts_names: 74 return SPAN_VOICE_TTS 75 if native_name in stt_names: 76 return SPAN_VOICE_STT 77 if unknown == "passthrough": 78 return native_name 79 return None
Map a vendor-native span name to a GenAI / voice export name.
Returns None to drop the span from the export name path.
custom may return a string, None (drop), or the sentinel
NotImplemented to fall through to the table-driven rules.
def
resolve_api_key(api_key: Optional[str] = None) -> str:
23def resolve_api_key(api_key: Optional[str] = None) -> str: 24 """Resolve API key from kwarg or ``PARLOT_API_KEY``.""" 25 return api_key or os.environ.get("PARLOT_API_KEY", "") or ""
Resolve API key from kwarg or PARLOT_API_KEY.
def
resolve_capture_genai_content(capture_genai_content: Optional[bool] = None) -> Optional[bool]:
28def resolve_capture_genai_content(capture_genai_content: Optional[bool] = None) -> Optional[bool]: 29 """Return the explicit ``parlotize(capture_genai_content=)`` override, or None. 30 31 When None, callers should resolve via ``should_capture_genai_content`` against 32 bootstrap policy (default on). 33 """ 34 if capture_genai_content is not None: 35 return bool(capture_genai_content) 36 return None
Return the explicit parlotize(capture_genai_content=) override, or None.
When None, callers should resolve via should_capture_genai_content against
bootstrap policy (default on).
def
resolve_endpoint(endpoint: Optional[str] = None) -> str:
18def resolve_endpoint(endpoint: Optional[str] = None) -> str: 19 """Resolve OTLP base endpoint from kwarg or ``PARLOT_ENDPOINT``.""" 20 return (endpoint or os.environ.get("PARLOT_ENDPOINT", "") or "").rstrip("/")
Resolve OTLP base endpoint from kwarg or PARLOT_ENDPOINT.
def
resolve_parlot_sdk_version() -> str:
26def resolve_parlot_sdk_version() -> str: 27 """Return installed parlot-core version, cached after first call.""" 28 global _cached_version 29 if _cached_version is not None: 30 return _cached_version 31 32 try: 33 _cached_version = _normalize_version(metadata.version(_PARLOT_CORE_DIST)) 34 except metadata.PackageNotFoundError: 35 _cached_version = "" 36 37 return _cached_version
Return installed parlot-core version, cached after first call.
def
runtime_from_bootstrap( endpoint: str, api_key: str, payload: dict[str, typing.Any]) -> ParlotRuntimeContext:
136def runtime_from_bootstrap( 137 endpoint: str, api_key: str, payload: dict[str, Any] 138) -> ParlotRuntimeContext: 139 globs, agents = _parse_recording_policy(payload) 140 ( 141 logs_globs, 142 logs_agents, 143 logs_agent_min_levels, 144 logs_min_level, 145 logs_present, 146 ) = _parse_logs_policy(payload) 147 ( 148 capture_globs, 149 capture_agents, 150 capture_present, 151 ) = _parse_capture_genai_content_policy(payload) 152 return ParlotRuntimeContext( 153 endpoint=endpoint.rstrip("/"), 154 api_key=api_key, 155 tenant_id=str(payload.get("tenant_id") or ""), 156 content_bucket=str(payload.get("content_bucket") or ""), 157 r2_endpoint=str(payload.get("r2_endpoint") or ""), 158 recording_globs=globs, 159 recording_agents=agents, 160 logs_globs=logs_globs, 161 logs_agents=logs_agents, 162 logs_agent_min_levels=logs_agent_min_levels, 163 logs_min_level=logs_min_level, 164 logs_policy_present=logs_present, 165 capture_genai_content_globs=capture_globs, 166 capture_genai_content_agents=capture_agents, 167 capture_genai_content_policy_present=capture_present, 168 )
def
session_owned(*, framework: str | None = None) -> bool:
76def session_owned(*, framework: str | None = None) -> bool: 77 """True when an active Parlot session with a non-empty session_id is bound. 78 79 Args: 80 framework: If provided, also require ``state.framework`` to match 81 (e.g. ``session_owned(framework="livekit")`` for coexistence checks). 82 """ 83 state = get_active_session() 84 if state is None or not state.session_id: 85 return False 86 if framework is not None and state.framework != framework: 87 return False 88 return True
True when an active Parlot session with a non-empty session_id is bound.
Arguments:
- framework: If provided, also require
state.frameworkto match (e.g.session_owned(framework="livekit")for coexistence checks).
def
set_active_session( span: opentelemetry.trace.span.Span | None, state: SessionState | None) -> None:
62def set_active_session( 63 span: Span | None, 64 state: SessionState | None, 65) -> None: 66 """Bind (or clear) the active session span + state for this context.""" 67 _active_session_span.set(span) 68 _active_session_state.set(state)
Bind (or clear) the active session span + state for this context.
def
set_session_attribute(key: str, value: str | int | float | bool) -> None:
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, orbool).
def
set_session_metadata(**pairs: str | int | float | bool) -> None:
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, orbool).
Example::
from parlot.instrumentation.livekit import set_session_metadata
set_session_metadata(order_id="12345", crm_ticket="TKT-9")
def
should_capture_genai_content( agent_name: str, *, metadata_capture_genai_content: Optional[bool] = None, capture_genai_content_config: Optional[bool] = None, bootstrap_globs: Optional[Sequence[str]] = None, bootstrap_agents: Optional[Mapping[str, bool]] = None, bootstrap_present: bool = False) -> bool:
30def should_capture_genai_content( 31 agent_name: str, 32 *, 33 metadata_capture_genai_content: Optional[bool] = None, 34 capture_genai_content_config: Optional[bool] = None, 35 bootstrap_globs: Optional[Sequence[str]] = None, 36 bootstrap_agents: Optional[Mapping[str, bool]] = None, 37 bootstrap_present: bool = False, 38) -> bool: 39 """Return True when generative AI / tool content bodies should be captured. 40 41 Precedence: job metadata > ``parlotize(capture_genai_content=…)`` > bootstrap (UI). 42 When bootstrap is absent (``bootstrap_present=False``) or globs are unset, 43 default is **on**. Explicit empty globs means off. 44 """ 45 if metadata_capture_genai_content is False: 46 return False 47 if metadata_capture_genai_content is True: 48 return True 49 50 if capture_genai_content_config is not None: 51 return bool(capture_genai_content_config) 52 53 agents = bootstrap_agents or {} 54 if agent_name in agents: 55 return bool(agents[agent_name]) 56 57 if not bootstrap_present: 58 return True 59 60 if bootstrap_globs is None: 61 return True 62 63 globs = list(bootstrap_globs) 64 if not globs: 65 return False 66 return matches_allowlist(agent_name, globs)
Return True when generative AI / tool content bodies should be captured.
Precedence: job metadata > parlotize(capture_genai_content=…) > bootstrap (UI).
When bootstrap is absent (bootstrap_present=False) or globs are unset,
default is on. Explicit empty globs means off.
def
should_record( agent_name: str, *, metadata_record: Optional[bool] = None, record_config: Union[bool, Sequence[str], NoneType] = None, bootstrap_globs: Optional[Sequence[str]] = None, bootstrap_agents: Optional[Mapping[str, bool]] = None) -> bool:
28def should_record( 29 agent_name: str, 30 *, 31 metadata_record: Optional[bool] = None, 32 record_config: RecordConfig = None, 33 bootstrap_globs: Optional[Sequence[str]] = None, 34 bootstrap_agents: Optional[Mapping[str, bool]] = None, 35) -> bool: 36 """Return True when session audio recording should start for this job. 37 38 Precedence: job metadata > ``parlotize(record=…)`` > bootstrap (UI) policy. 39 """ 40 if metadata_record is False: 41 return False 42 if metadata_record is True: 43 return True 44 45 if record_config is not None: 46 if isinstance(record_config, bool): 47 return record_config 48 return matches_allowlist(agent_name, list(record_config)) 49 50 agents = bootstrap_agents or {} 51 if agent_name in agents: 52 return bool(agents[agent_name]) 53 54 globs = list(bootstrap_globs or ()) 55 if not globs: 56 return False 57 return matches_allowlist(agent_name, globs)
Return True when session audio recording should start for this job.
Precedence: job metadata > parlotize(record=…) > bootstrap (UI) policy.
@runtime_checkable
class
RecordingSource19@runtime_checkable 20class RecordingSource(Protocol): 21 """Adapter-facing recording capture / stamp contract.""" 22 23 async def start(self) -> None: ... 24 25 async def finish(self) -> None: ... 26 27 async def fail(self, reason: str) -> None: ...
Adapter-facing recording capture / stamp contract.
RecordingSource(*args, **kwargs)
1739def _no_init_or_replace_init(self, *args, **kwargs): 1740 cls = type(self) 1741 1742 if cls._is_protocol: 1743 raise TypeError('Protocols cannot be instantiated') 1744 1745 # Already using a custom `__init__`. No need to calculate correct 1746 # `__init__` to call. This can lead to RecursionError. See bpo-45121. 1747 if cls.__init__ is not _no_init_or_replace_init: 1748 return 1749 1750 # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`. 1751 # The first instantiation of the subclass will call `_no_init_or_replace_init` which 1752 # searches for a proper new `__init__` in the MRO. The new `__init__` 1753 # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent 1754 # instantiation of the protocol subclass will thus use the new 1755 # `__init__` and no longer call `_no_init_or_replace_init`. 1756 for base in cls.__mro__: 1757 init = base.__dict__.get('__init__', _no_init_or_replace_init) 1758 if init is not _no_init_or_replace_init: 1759 cls.__init__ = init 1760 break 1761 else: 1762 # should not happen 1763 cls.__init__ = object.__init__ 1764 1765 cls.__init__(self, *args, **kwargs)
def
stamp_external_recording( *, audio_uri: str, anchor_wall_ms: int, mode: str = 'link', format: str = 'opus_ogg', sample_count: Optional[int] = None, wall_end_ms: Optional[int] = None, pii_redaction_required: bool = False, span: Any = None) -> bool:
30def stamp_external_recording( 31 *, 32 audio_uri: str, 33 anchor_wall_ms: int, 34 mode: str = "link", 35 format: str = "opus_ogg", 36 sample_count: Optional[int] = None, 37 wall_end_ms: Optional[int] = None, 38 pii_redaction_required: bool = False, 39 span: Any = None, 40) -> bool: 41 """BYO recording stamp. 42 43 * ``mode="link"`` — external URI only. Rejected when *pii_redaction_required*. 44 * ``mode="ingest"`` — caller already uploaded via grant; stamp as uploaded. 45 46 Returns False when rejected or no session span. 47 """ 48 target = span if span is not None else get_active_session_span() 49 if target is None: 50 logger.warning("parlot: stamp_external_recording skipped — no session span") 51 return False 52 53 if mode == "link" and pii_redaction_required: 54 stamp_recording_error( 55 target, 56 "byo_link_rejected:tenant_pii_redaction_enabled", 57 ) 58 return False 59 60 source = "byo_link" if mode == "link" else "byo_ingest" 61 stamp_recording_started( 62 target, 63 anchor_wall_ms=anchor_wall_ms, 64 source=source, 65 format=format, 66 ) 67 stamp_recording_uploaded( 68 target, 69 audio_uri=audio_uri, 70 partial=False, 71 sample_count=sample_count, 72 wall_end_ms=wall_end_ms, 73 format=format, 74 ) 75 return True
BYO recording stamp.
mode="link"— external URI only. Rejected when pii_redaction_required.mode="ingest"— caller already uploaded via grant; stamp as uploaded.
Returns False when rejected or no session span.
def
stamp_recording_disabled(span: Any, reason: str) -> None:
def
stamp_recording_error(span: Any, error: str) -> None:
def
stamp_recording_started( span: Any, *, anchor_wall_ms: int, source: str, format: str = 'pcm_wav') -> None:
55def stamp_recording_started( 56 span: Any, 57 *, 58 anchor_wall_ms: int, 59 source: str, 60 format: str = "pcm_wav", 61) -> None: 62 stamp_recording_attr(span, ATTR_SESSION_RECORDING_ANCHOR_WALL_MS, int(anchor_wall_ms)) 63 stamp_recording_attr(span, ATTR_SESSION_RECORDING_STATUS, "recording") 64 stamp_recording_attr(span, ATTR_SESSION_RECORDING_SOURCE, source) 65 stamp_recording_attr(span, ATTR_SESSION_RECORDING_FORMAT, format)
def
stamp_recording_uploaded( span: Any, *, audio_uri: str, partial: bool = False, sample_count: Optional[int] = None, wall_end_ms: Optional[int] = None, format: Optional[str] = None, egress_id: Optional[str] = None) -> None:
68def stamp_recording_uploaded( 69 span: Any, 70 *, 71 audio_uri: str, 72 partial: bool = False, 73 sample_count: Optional[int] = None, 74 wall_end_ms: Optional[int] = None, 75 format: Optional[str] = None, 76 egress_id: Optional[str] = None, 77) -> None: 78 stamp_recording_attr(span, ATTR_SESSION_RECORDING_AUDIO_URI, audio_uri) 79 stamp_recording_attr( 80 span, 81 ATTR_SESSION_RECORDING_STATUS, 82 "partial" if partial else "uploaded", 83 ) 84 stamp_recording_attr(span, ATTR_SESSION_RECORDING_PARTIAL, bool(partial)) 85 if sample_count is not None: 86 stamp_recording_attr(span, ATTR_SESSION_RECORDING_SAMPLE_COUNT, int(sample_count)) 87 if wall_end_ms is not None: 88 stamp_recording_attr(span, ATTR_SESSION_RECORDING_WALL_END_MS, int(wall_end_ms)) 89 if format: 90 stamp_recording_attr(span, ATTR_SESSION_RECORDING_FORMAT, format) 91 if egress_id: 92 stamp_recording_attr(span, ATTR_SESSION_RECORDING_EGRESS_ID, egress_id)
def
set_recording_anchor_wall_ms(state: Any, anchor_wall_ms: int) -> None:
13def set_recording_anchor_wall_ms(state: Any, anchor_wall_ms: int) -> None: 14 """Freeze the sample-0 wall clock on *state* (first write wins).""" 15 current = getattr(state, "recording_anchor_wall_ms", None) 16 if current is not None: 17 return 18 state.recording_anchor_wall_ms = int(anchor_wall_ms)
Freeze the sample-0 wall clock on state (first write wins).
def
media_segments_from_speech( state: Any, speech_start_wall_ms: int, speech_end_wall_ms: int) -> tuple[int, int]:
21def media_segments_from_speech( 22 state: Any, 23 speech_start_wall_ms: int, 24 speech_end_wall_ms: int, 25) -> tuple[int, int]: 26 """Return ``(start_ms, end_ms)`` offsets from the frozen recording anchor.""" 27 anchor = getattr(state, "recording_anchor_wall_ms", None) 28 if anchor is None: 29 return (0, max(0, int(speech_end_wall_ms) - int(speech_start_wall_ms))) 30 start = max(0, int(speech_start_wall_ms) - int(anchor)) 31 end = max(start, int(speech_end_wall_ms) - int(anchor)) 32 return (start, end)
Return (start_ms, end_ms) offsets from the frozen recording anchor.
def
speech_wall_ms_from_span(span: Any) -> tuple[int, int] | None:
35def speech_wall_ms_from_span(span: Any) -> tuple[int, int] | None: 36 """Derive speech wall bounds from an OTel ReadableSpan start/end.""" 37 start_ns = getattr(span, "start_time", None) 38 end_ns = getattr(span, "end_time", None) 39 if start_ns is None or end_ns is None: 40 return None 41 try: 42 start_ms = int(start_ns) // 1_000_000 43 end_ms = int(end_ns) // 1_000_000 44 except (TypeError, ValueError): 45 return None 46 if end_ms < start_ms: 47 return None 48 return (start_ms, end_ms)
Derive speech wall bounds from an OTel ReadableSpan start/end.
async def
fetch_upload_grant( runtime: ParlotRuntimeContext, *, session_id: str, room_name: Optional[str] = None, segment_index: Optional[int] = None, sample_offset: Optional[int] = None, content_type: Optional[str] = None) -> Optional[dict[str, Any]]:
16async def fetch_upload_grant( 17 runtime: ParlotRuntimeContext, 18 *, 19 session_id: str, 20 room_name: Optional[str] = None, 21 segment_index: Optional[int] = None, 22 sample_offset: Optional[int] = None, 23 content_type: Optional[str] = None, 24) -> Optional[dict[str, Any]]: 25 """POST ``/v1/recordings/upload-grant``; return JSON body or ``None`` on failure.""" 26 url = f"{runtime.endpoint.rstrip('/')}/v1/recordings/upload-grant" 27 headers = {"Authorization": f"Bearer {runtime.api_key}"} 28 payload: dict[str, Any] = {"session_id": session_id} 29 if room_name: 30 payload["room_name"] = room_name 31 if segment_index is not None: 32 payload["segment_index"] = int(segment_index) 33 if sample_offset is not None: 34 payload["sample_offset"] = int(sample_offset) 35 if content_type: 36 payload["content_type"] = content_type 37 try: 38 async with httpx.AsyncClient(timeout=30.0) as client: 39 resp = await client.post(url, json=payload, headers=headers) 40 if resp.status_code >= 400: 41 logger.error( 42 "parlot: upload grant failed status=%s body=%s", 43 resp.status_code, 44 resp.text[:500], 45 ) 46 return None 47 data = resp.json() 48 return data if isinstance(data, dict) else None 49 except Exception as exc: 50 logger.error("parlot: upload grant request failed — %s", exc) 51 logger.debug("parlot: upload grant request failed", exc_info=True) 52 return None
POST /v1/recordings/upload-grant; return JSON body or None on failure.
async def
complete_recording( runtime: ParlotRuntimeContext, *, session_id: str, format: str, conversation_id: Optional[str] = None, segment_count: Optional[int] = None, segments: Optional[list[Mapping[str, int]]] = None, sample_rate: Optional[int] = None, anchor_wall_ms: Optional[int] = None, sample_count: Optional[int] = None, wall_end_ms: Optional[int] = None, duration_ms: Optional[int] = None, audio_recording_uri: Optional[str] = None) -> Optional[dict[str, Any]]:
55async def complete_recording( 56 runtime: ParlotRuntimeContext, 57 *, 58 session_id: str, 59 format: str, 60 conversation_id: Optional[str] = None, 61 segment_count: Optional[int] = None, 62 segments: Optional[list[Mapping[str, int]]] = None, 63 sample_rate: Optional[int] = None, 64 anchor_wall_ms: Optional[int] = None, 65 sample_count: Optional[int] = None, 66 wall_end_ms: Optional[int] = None, 67 duration_ms: Optional[int] = None, 68 audio_recording_uri: Optional[str] = None, 69) -> Optional[dict[str, Any]]: 70 """POST ``/v1/recordings/complete``; return JSON body or ``None`` on failure.""" 71 url = f"{runtime.endpoint.rstrip('/')}/v1/recordings/complete" 72 headers = {"Authorization": f"Bearer {runtime.api_key}"} 73 payload: dict[str, Any] = { 74 "session_id": session_id, 75 "format": format, 76 } 77 if conversation_id: 78 payload["conversation_id"] = conversation_id 79 if segment_count is not None: 80 payload["segment_count"] = int(segment_count) 81 if segments: 82 payload["segments"] = [ 83 { 84 "segment_index": int(s["segment_index"]), 85 "sample_offset": int(s["sample_offset"]), 86 } 87 for s in segments 88 ] 89 if sample_rate is not None: 90 payload["sample_rate"] = int(sample_rate) 91 if anchor_wall_ms is not None: 92 payload["anchor_wall_ms"] = int(anchor_wall_ms) 93 if sample_count is not None: 94 payload["sample_count"] = int(sample_count) 95 if wall_end_ms is not None: 96 payload["wall_end_ms"] = int(wall_end_ms) 97 if duration_ms is not None: 98 payload["duration_ms"] = int(duration_ms) 99 if audio_recording_uri: 100 payload["audio_recording_uri"] = audio_recording_uri 101 try: 102 async with httpx.AsyncClient(timeout=120.0) as client: 103 resp = await client.post(url, json=payload, headers=headers) 104 if resp.status_code >= 400: 105 logger.error( 106 "parlot: recording complete failed status=%s body=%s", 107 resp.status_code, 108 resp.text[:500], 109 ) 110 return None 111 data = resp.json() 112 return data if isinstance(data, dict) else None 113 except Exception as exc: 114 logger.error("parlot: recording complete request failed — %s", exc) 115 logger.debug("parlot: recording complete request failed", exc_info=True) 116 return None
POST /v1/recordings/complete; return JSON body or None on failure.
async def
put_bytes_to_presigned_url( put_url: str, body: bytes, *, content_type: str = 'audio/wav', timeout: float = 120.0) -> bool:
119async def put_bytes_to_presigned_url( 120 put_url: str, 121 body: bytes, 122 *, 123 content_type: str = "audio/wav", 124 timeout: float = 120.0, 125) -> bool: 126 """PUT object bytes to a presigned URL. Returns True on 2xx.""" 127 try: 128 async with httpx.AsyncClient(timeout=timeout) as client: 129 resp = await client.put( 130 put_url, 131 content=body, 132 headers={"Content-Type": content_type}, 133 ) 134 if resp.status_code >= 400: 135 logger.error( 136 "parlot: presigned PUT failed status=%s body=%s", 137 resp.status_code, 138 resp.text[:300], 139 ) 140 return False 141 return True 142 except Exception as exc: 143 logger.error("parlot: presigned PUT failed — %s", exc) 144 logger.debug("parlot: presigned PUT failed", exc_info=True) 145 return False
PUT object bytes to a presigned URL. Returns True on 2xx.
def
should_capture_logs( agent_name: str, *, metadata_capture_logs: Optional[bool] = None, capture_logs_config: Union[bool, Sequence[str], NoneType] = None, bootstrap_globs: Optional[Sequence[str]] = None, bootstrap_agents: Optional[Mapping[str, bool]] = None, bootstrap_present: bool = False) -> bool:
47def should_capture_logs( 48 agent_name: str, 49 *, 50 metadata_capture_logs: Optional[bool] = None, 51 capture_logs_config: CaptureLogsConfig = None, 52 bootstrap_globs: Optional[Sequence[str]] = None, 53 bootstrap_agents: Optional[Mapping[str, bool]] = None, 54 bootstrap_present: bool = False, 55) -> bool: 56 """Return True when application logs should be captured for this job. 57 58 Precedence: job metadata > ``parlotize(capture_logs=…)`` > bootstrap (UI). 59 When bootstrap is absent (``bootstrap_present=False``) or globs are unset, 60 default is **on**. Explicit empty globs means off. 61 """ 62 if metadata_capture_logs is False: 63 return False 64 if metadata_capture_logs is True: 65 return True 66 67 if capture_logs_config is not None: 68 if isinstance(capture_logs_config, bool): 69 return capture_logs_config 70 return matches_allowlist(agent_name, list(capture_logs_config)) 71 72 agents = bootstrap_agents or {} 73 if agent_name in agents: 74 return bool(agents[agent_name]) 75 76 if not bootstrap_present: 77 return True 78 79 if bootstrap_globs is None: 80 return True 81 82 globs = list(bootstrap_globs) 83 if not globs: 84 return False 85 return matches_allowlist(agent_name, globs)
Return True when application logs should be captured for this job.
Precedence: job metadata > parlotize(capture_logs=…) > bootstrap (UI).
When bootstrap is absent (bootstrap_present=False) or globs are unset,
default is on. Explicit empty globs means off.
def
stamp_framework_refs(span: Any, framework: str, **kinds: str) -> None:
28def stamp_framework_refs(span: Any, framework: str, **kinds: str) -> None: 29 """Stamp non-empty ``platform.ref.*`` kinds for ``framework``. 30 31 Convenience wrapper over :func:`stamp_platform_refs` for adapters that 32 build refs from keyword ids (e.g. ``conversation_id=``, ``room_sid=``). 33 """ 34 fw = str(framework or "").strip() 35 if not fw: 36 return 37 triples = [ 38 (fw, str(kind).strip(), str(value).strip()) 39 for kind, value in kinds.items() 40 if str(kind or "").strip() and str(value or "").strip() 41 ] 42 if triples: 43 stamp_platform_refs(span, triples)
Stamp non-empty platform.ref.* kinds for framework.
Convenience wrapper over stamp_platform_refs() for adapters that
build refs from keyword ids (e.g. conversation_id=, room_sid=).
def
stamp_platform_refs(span: Any, refs: list[tuple[str, str, str]]) -> None:
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 aReadableSpanwith a mutable_attributesdict. - refs: Non-empty list of reference triples.
def
stamp_session_sdk_version(session_span: Any, state: typing.Any | None = None) -> None:
40def stamp_session_sdk_version(session_span: Any, state: Any | None = None) -> None: 41 """Stamp ``parlot.sdk.version`` and ``session.metadata.sdk_version`` on the session span and state.""" 42 version = resolve_parlot_sdk_version() 43 if not version: 44 return 45 46 if isinstance(session_span, dict): 47 session_span[ATTR_PARLOT_SDK_VERSION] = version 48 session_span[ATTR_SESSION_METADATA_SDK_VERSION] = version 49 elif session_span is not None and hasattr(session_span, "set_attribute"): 50 session_span.set_attribute(ATTR_PARLOT_SDK_VERSION, version) 51 session_span.set_attribute(ATTR_SESSION_METADATA_SDK_VERSION, version) 52 53 if state is None: 54 try: 55 from parlot.core.session import get_active_session 56 57 state = get_active_session() 58 except Exception: 59 state = None 60 61 if state is not None and hasattr(state, "custom_metadata") and isinstance(state.custom_metadata, dict): 62 state.custom_metadata[ATTR_SESSION_METADATA_SDK_VERSION] = version
Stamp parlot.sdk.version and session.metadata.sdk_version on the session span and state.
def
stamp_turn_utterance_text(span: Any, *, participant_role: str, utterance_text: str) -> None:
29def stamp_turn_utterance_text( 30 span: Any, 31 *, 32 participant_role: str, 33 utterance_text: str, 34) -> None: 35 """ 36 Stamp ``turn.user_text`` or ``turn.agent_text`` on a turn root span. 37 38 Adapters call this after committing an utterance so the OTLP contract stays 39 consistent regardless of source. 40 """ 41 text = (utterance_text or "").strip() 42 if not text: 43 return 44 role = (participant_role or "").strip().lower() 45 if role == "user": 46 span.set_attribute(ATTR_TURN_USER_TEXT, text) 47 elif role == "agent": 48 span.set_attribute(ATTR_TURN_AGENT_TEXT, text)
Stamp turn.user_text or turn.agent_text on a turn root span.
Adapters call this after committing an utterance so the OTLP contract stays consistent regardless of source.
def
start_parlot_session_span( tracer: Any, *, session_id: str | None = None, conversation_id: str, stamp_fn: Optional[Callable[[Any], NoneType]] = None, extra_attrs: Optional[Mapping[str, Any]] = None) -> tuple[typing.Any, str]:
207def start_parlot_session_span( 208 tracer: Any, 209 *, 210 session_id: str | None = None, 211 conversation_id: str, 212 stamp_fn: Callable[[Any], None] | None = None, 213 extra_attrs: Mapping[str, Any] | None = None, 214) -> tuple[Any, str]: 215 """Start a ``parlot.session`` span and stamp identity. 216 217 Returns ``(span, session_id)``. Generates a session id when omitted. 218 """ 219 sid = session_id or new_session_id() 220 span = tracer.start_span(SPAN_CONVERSATION_SESSION) 221 span.set_attribute(ATTR_SESSION_ID, sid) 222 span.set_attribute(ATTR_SESSION_CONVERSATION_ID, conversation_id) 223 span.set_attribute(ATTR_GEN_AI_CONVERSATION_ID, conversation_id) 224 if extra_attrs: 225 for key, value in extra_attrs.items(): 226 if value is None or value == "": 227 continue 228 span.set_attribute(key, value) 229 if stamp_fn is not None: 230 stamp_fn(span) 231 return span, sid
Start a parlot.session span and stamp identity.
Returns (span, session_id). Generates a session id when omitted.
def
try_set_global_tracer_provider(provider: Any) -> bool:
138def try_set_global_tracer_provider(provider: Any) -> bool: 139 """Set the global tracer provider, swallowing failures (already set, etc.).""" 140 from opentelemetry import trace 141 142 try: 143 trace.set_tracer_provider(provider) 144 return True 145 except Exception: 146 logger.debug("Could not set global tracer provider", exc_info=True) 147 return False
Set the global tracer provider, swallowing failures (already set, etc.).