镜像站点 · 本页由第三方 GitHub 只读镜像提供,非 GitHub 官方站点,不接受任何登录或凭据输入。前往 github.com
Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 20 additions & 1 deletion livekit-agents/livekit/agents/job.py
Original file line number Diff line number Diff line change
Expand Up @@ -907,10 +907,29 @@ def init_recording(self, options: RecordingOptions) -> None:
or options.get("transcript", True)
)
obs_url = _observability_url(self._info.url)
if not (needs_cloud and obs_url):
if not obs_url:
self._stop_log_buffering()
return

if not needs_cloud:
# Recording is fully disabled, but _on_session_end still uploads the
# session's evaluations, tag metadata, and outcome (its has_evals
# branch). Those records travel through the log pipeline, so create
# it and register the job; traces, Python logs, and metrics stay off.
self._stop_log_buffering()
self._telemetry_state = _setup_cloud_tracer(
room_id=self.job.room.sid,
job_id=self.job.id,
agent_name=self.job.agent_name,
observability_url=obs_url,
enable_traces=False,
enable_logs=False,
enable_session_report=True,
enable_metrics=False,
Comment thread
devin-ai-integration[bot] marked this conversation as resolved.
Resolved
metadata=self._otel_metadata(options, redaction_enabled=redaction_enabled),
Comment thread
devin-ai-integration[bot] marked this conversation as resolved.
Resolved
Comment thread
devin-ai-integration[bot] marked this conversation as resolved.
Resolved
)
return

logger.debug("configuring session recording")
self._telemetry_state = _setup_cloud_tracer(
room_id=self.job.room.sid,
Expand Down
29 changes: 29 additions & 0 deletions livekit-agents/livekit/agents/telemetry/otel_metrics.py
Original file line number Diff line number Diff line change
Expand Up @@ -148,6 +148,8 @@ def _record_genai_tokens(attrs: dict[str, Any], *, input_tokens: int, output_tok

def record_execute_tool_duration(duration: float, *, tool_name: str, error: bool = False) -> None:
"""``gen_ai.execute_tool.duration`` for one function-tool execution."""
if not _metrics_enabled():
return
attrs = _job_attrs()
attrs[trace_types.ATTR_GEN_AI_OPERATION_NAME] = trace_types.GenAIOperationName.EXECUTE_TOOL
attrs[trace_types.ATTR_GEN_AI_TOOL_NAME] = tool_name
Expand All @@ -158,12 +160,33 @@ def record_execute_tool_duration(duration: float, *, tool_name: str, error: bool

def record_invoke_agent_duration(duration: float, *, agent_name: str) -> None:
"""``gen_ai.invoke_agent.duration`` for one agent turn."""
if not _metrics_enabled():
return
attrs = _job_attrs()
attrs[trace_types.ATTR_GEN_AI_OPERATION_NAME] = trace_types.GenAIOperationName.INVOKE_AGENT
attrs[trace_types.ATTR_GEN_AI_AGENT_NAME] = agent_name
_genai_invoke_agent_duration.record(duration, attributes=attrs)


def _metrics_enabled() -> bool:
"""Whether the job on this context resolved metrics on.

Instruments bind to the process-wide meter, so not creating a meter provider for a
job that disabled metrics does not stop a provider installed by an earlier or
concurrent job from exporting the measurements it records. Gate recording on the
job's own resolved state instead. Outside a job context (process health) recording
continues, and before the job resolves its options there is no decision to apply.
"""
from ..job import get_job_context # local import: job.py imports this module

ctx = get_job_context(required=False)
if ctx is None:
return True
if (state := ctx._telemetry_state) is not None:
return state.metrics_enabled
return True


def _job_attrs() -> dict[str, Any]:
"""Per-measurement job attribution.

Expand Down Expand Up @@ -197,6 +220,8 @@ def _model_attrs(metadata: Metadata | None) -> dict[str, Any]:

def flush_turn_metrics(chat_ctx: ChatContext) -> None:
"""Emit per-turn latency histograms from the chat history. Called at session end."""
if not _metrics_enabled():
return
for msg in chat_ctx.messages():
_record_turn_metrics(msg.metrics)

Expand Down Expand Up @@ -234,6 +259,8 @@ def _record_turn_metrics(report: MetricsReport) -> None:

def collect_usage(ev: AgentMetrics) -> None:
"""Record usage counters directly from each metrics event."""
if not _metrics_enabled():
return
if isinstance(ev, LLMMetrics):
attrs = _model_attrs(ev.metadata)
if ev.prompt_tokens:
Expand Down Expand Up @@ -308,6 +335,8 @@ def collect_usage(ev: AgentMetrics) -> None:


def record_event_loop_blocked(duration: float, *, severity: str) -> None:
if not _metrics_enabled():
return
attrs = _job_attrs()
attrs["severity"] = severity
_event_loop_blocked.record(duration, attributes=attrs)
97 changes: 80 additions & 17 deletions livekit-agents/livekit/agents/telemetry/traces.py
Original file line number Diff line number Diff line change
Expand Up @@ -339,6 +339,7 @@ class _JobTelemetry:
attributes: dict[str, AttributeValue]
traces_enabled: bool
logs_enabled: bool
metrics_enabled: bool = True


def _job_stamp_attributes() -> dict[str, AttributeValue] | None:
Expand Down Expand Up @@ -536,6 +537,21 @@ def shutdown(self) -> None:
self._inner.shutdown()


# Attribute carried by the records _upload_session_report emits. They belong to the
# session recording, not to Python log export, so a job's `logs` option does not gate
# them: a record=False session with evals still uploads them. The origin is proved by
# this internal marker, never by the logger's (public) scope name, which any user
# logger can claim.
_SESSION_REPORT_ATTRIBUTE = "lk.internal.session_report"


def _is_session_report_record(record: ReadableLogRecord) -> bool:
attributes = record.log_record.attributes
if not attributes:
return False
return attributes.get(_SESSION_REPORT_ATTRIBUTE) is True


class _GatedLogExporter(LogRecordExporter):
"""Log counterpart of :class:`_GatedSpanExporter`."""

Expand All @@ -544,14 +560,24 @@ def __init__(self, inner: LogRecordExporter, export_jobs: Mapping[str, _JobTelem
self._export_jobs = export_jobs

def export(self, batch: Sequence[ReadableLogRecord]) -> LogRecordExportResult:
exportable = [
r
for r in batch
if (state := _job_export_state(self._export_jobs, r.log_record.attributes)) is not None
and state.logs_enabled
]
exportable: list[ReadableLogRecord] = []
for r in batch:
state = _job_export_state(self._export_jobs, r.log_record.attributes)
if state is None:
continue
if state.logs_enabled or _is_session_report_record(r):
exportable.append(r)
if not exportable:
return LogRecordExportResult.SUCCESS
for r in exportable:
attributes = r.log_record.attributes
if attributes and _SESSION_REPORT_ATTRIBUTE in attributes:
# the marker is for the gate, not for Cloud
r.log_record.attributes = {
key: value
for key, value in attributes.items()
if key != _SESSION_REPORT_ATTRIBUTE
}
return self._inner.export(exportable)

def force_flush(self, timeout_millis: int = 30000) -> bool:
Expand Down Expand Up @@ -923,11 +949,18 @@ def configure(
observability_url: str,
enable_traces: bool = True,
enable_logs: bool = True,
enable_session_report: bool = False,
enable_metrics: bool = True,
metadata: dict[str, AttributeValue] | None = None,
) -> _JobTelemetry:
"""Set up (or reuse) the pipelines for a job. Returns the job's
telemetry state, for the caller to keep on the JobContext — per-record
stamping and upload gating resolve it from there."""
stamping and upload gating resolve it from there.

``enable_session_report`` creates the log pipeline for the records
``_upload_session_report`` emits (evaluations, tags, outcome, chat
history) without turning on Python log export; it is what lets a job
that records nothing else still upload its evals."""
base_metadata: dict[str, AttributeValue] = {"room_id": room_id, "job_id": job_id}
if agent_name:
# identifies the agent for LiveKit Cloud agent insights (explicit dispatch
Expand Down Expand Up @@ -969,10 +1002,12 @@ def configure(
# evaluations, and chat history, not just Python log export.
self._ensure_logger_provider()

if enable_logs:
if enable_logs or enable_session_report:
self._ensure_log_pipeline(url)
if self._log_metadata_processor is not None:
self._log_metadata_processor.set_metadata(session_metadata)

if enable_logs:
if self._log_handler is not None:
root = logging.getLogger()
if self._log_handler not in root.handlers:
Expand All @@ -982,11 +1017,12 @@ def configure(
# set-once), so its resource carries only process-stable identity;
# per-job room_id/job_id ride on each measurement instead
# (otel_metrics._job_attrs)
process_metadata = {
k: v for k, v in base_metadata.items() if k not in ("room_id", "job_id")
}
meter_resource = _resource({SERVICE_NAME: "livekit-agents", **process_metadata})
self._ensure_meter_provider(meter_resource, url)
if enable_metrics:
process_metadata = {
k: v for k, v in base_metadata.items() if k not in ("room_id", "job_id")
}
meter_resource = _resource({SERVICE_NAME: "livekit-agents", **process_metadata})
self._ensure_meter_provider(meter_resource, url)

if not self._atexit_registered:
self._atexit_registered = True
Expand All @@ -996,6 +1032,7 @@ def configure(
attributes=session_metadata,
traces_enabled=enable_traces,
logs_enabled=enable_logs,
metrics_enabled=enable_metrics,
)
self._export_jobs[job_id] = state
if self._span_gate is not None:
Expand Down Expand Up @@ -1231,6 +1268,8 @@ def _setup_cloud_tracer(
observability_url: str,
enable_traces: bool = True,
enable_logs: bool = True,
enable_session_report: bool = False,
enable_metrics: bool = True,
metadata: dict[str, AttributeValue] | None = None,
) -> _JobTelemetry:
_upload_gate.reset()
Expand All @@ -1241,6 +1280,8 @@ def _setup_cloud_tracer(
observability_url=observability_url,
enable_traces=enable_traces,
enable_logs=enable_logs,
enable_session_report=enable_session_report,
enable_metrics=enable_metrics,
metadata=metadata,
)

Expand Down Expand Up @@ -1313,13 +1354,22 @@ def _log(
otel_logger.emit(
body=body,
timestamp=timestamp,
attributes=attributes,
attributes={**attributes, _SESSION_REPORT_ATTRIBUTE: True},
severity_number=severity,
severity_text=severity_text,
)

chat_logger = _get_logger("chat_history")
recording_options = report.options.recording_options
# The report's attribute names belong to the session-report schema and carry no
# `pii` segment, so the log redaction processor cannot drop them. When the project
# mandates redaction, withhold the free-form parts that can quote the conversation
# here instead; the verdict, tag name, and outcome stay because they carry none.
redaction_enabled = (
bool((metadata or {}).get(ATTRIBUTE_REDACTION_ENABLED))
or bool(recording_options.get("redaction", False))
or telemetry_utils.redaction_enabled()
)

if recording_enabled(recording_options):
_log(
Expand Down Expand Up @@ -1370,28 +1420,41 @@ def _log(
severity = SeverityNumber.ERROR
severity_text = "error"

# judge reasoning and criteria quote the conversation
evaluation_payload = evaluation
if redaction_enabled:
evaluation_payload = {
key: value
for key, value in evaluation.items()
if key not in ("reasoning", "instructions")
}

_log(
eval_logger,
body="evaluation",
timestamp=int(report.timestamp * 1e9),
attributes={"evaluation": evaluation},
attributes={"evaluation": evaluation_payload},
severity=severity,
severity_text=severity_text,
)

for tag, entry in tagger._tags.items():
if entry.metadata:
# tag metadata is caller-supplied and may quote session content
tag_payload: dict[str, Any] = {"name": tag}
if not redaction_enabled:
tag_payload["metadata"] = entry.metadata
_log(
eval_logger,
body="tag",
timestamp=int(entry.timestamp * 1e9),
attributes={"tag": {"name": tag, "metadata": entry.metadata}},
attributes={"tag": tag_payload},
)

if tagger.outcome:
is_fail = tagger.outcome == "fail"
outcome_data: dict[str, Any] = {"outcome": tagger.outcome}
if tagger.outcome_reason:
if tagger.outcome_reason and not redaction_enabled:
outcome_data["reason"] = tagger.outcome_reason

_log(
Expand Down
Loading
Loading