feat(telemetry): full OTel GenAI semantic conventions + in-process PII stripping

.changeset/genai-semconv-and-pii-stripping.md

@@ -0,0 +1,35 @@
+---
+'@livekit/agents': minor
+---
+
+Emit the full OpenTelemetry GenAI semantic conventions on agent spans.
+
+Spans now carry the standard `gen_ai.*` attributes — operation, provider, request/response
+model, per-modality token usage, cached tokens, finish reasons, time-to-first-chunk, tool
+name/type/call id, and the `gen_ai.input.messages` / `gen_ai.output.messages` /
+`gen_ai.system_instructions` / `gen_ai.tool.definitions` content payloads — so Datadog Agent
+Observability, Langfuse and any other backend that reads the conventions (OTel v1.37+)
+renders a LiveKit trace without a custom mapping. The session maps to `invoke_workflow`, an
+agent turn to `invoke_agent`, inference to `chat` (realtime turns to `generate_content` on
+their own nested span), and tool execution to `execute_tool`.
+
+`gen_ai.provider.name` now reports the registry spelling the convention requires. Plugins
+expose `provider` either as a display name (`MistralAI`, `AWS Bedrock`, `Vertex AI`) or as the
+client's base-URL host (`api.openai.com`, `api.anthropic.com`); both are normalized, so
+`openai`, `anthropic`, `mistral_ai`, `aws.bedrock`, `gcp.vertex_ai`, `gcp.gemini`, `x_ai`,
+`groq` and `perplexity` are recognized by GenAI backends. A provider outside the registry
+keeps its own id, which the convention allows.
+
+Conversational content reaches every configured exporter, as before. To withhold it from a
+third-party pipeline while LiveKit Cloud keeps receiving it, pass `allowPii: false`:
+
+```ts
+telemetry.setTracerProvider(provider, { registerSpanProcessor, allowPii: false });
+```
+
+or set `LIVEKIT_TELEMETRY_ALLOW_PII=0` when the framework adopts the ambient OpenTelemetry
+provider and there is no call site. What LiveKit Cloud receives stays governed by the
+project's PII setting in the dashboard; when that mandates redaction, PII is withheld from
+every destination and `allowPii` does not weaken it. Content can be dropped entirely with
+`telemetry.genAI.setCaptureContent(false)` or
+`OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=false`.

agents/src/llm/llm.ts

@@ -7,7 +7,7 @@ import { EventEmitter } from 'node:events';
 import { APIConnectionError, APIError } from '../_exceptions.js';
 import { log } from '../log.js';
 import type { LLMMetrics } from '../metrics/base.js';
-import { recordException, traceTypes, tracer } from '../telemetry/index.js';
+import { genAI, recordException, traceTypes, tracer } from '../telemetry/index.js';
 import { type APIConnectOptions, intervalForRetry } from '../types.js';
 import { AsyncIterableQueue, Task, delay, startSoon, toError } from '../utils.js';
 import { type ChatContext, type ChatRole, type FunctionCall } from './chat_context.js';
@@ -217,6 +217,11 @@ export abstract class LLMStream implements AsyncIterableIterator<ChatChunk> {
       this.closed = true;
     });
 
+    // tells an enclosing `llm_node` span that this call is instrumented, so it does not
+    // record the convention's attributes a second time. Read here rather than in mainTask,
+    // which startSoon defers out of the node's context.
+    genAI.markInferenceSpanRecorded();
+
     // this is a hack to immitate asyncio.create_task so that mainTask
     // is run **after** the constructor has finished. Otherwise we get
     // runtime error when trying to access class variables in the
@@ -232,9 +237,26 @@ export abstract class LLMStream implements AsyncIterableIterator<ChatChunk> {
     });
   }
 
+  /** The GenAI inference span's request side, per the OTel GenAI conventions. */
+  private recordGenAIRequest(span: Span) {
+    genAI.setRequestAttributes(span, {
+      operation: traceTypes.GenAIOperationName.CHAT,
+      provider: this.#llm.provider,
+      model: this.#llm.model,
+      stream: true,
+      outputType: traceTypes.GenAIOutputType.TEXT,
+    });
+    genAI.setContentAttributes(span, {
+      systemInstructions: genAI.toSystemInstructions(this.#chatCtx),
+      inputMessages: genAI.toInputMessages(this.#chatCtx),
+      toolDefinitions: this.#toolCtx ? genAI.toToolDefinitions(this.#toolCtx.functionTools) : [],
+    });
+  }
+
   private _mainTaskImpl = async (span: Span) => {
     this.#llmRequestSpan = span;
     span.setAttribute(traceTypes.ATTR_GEN_AI_REQUEST_MODEL, this.#llm.model);
+    this.recordGenAIRequest(span);
 
     for (let i = 0; i < this._connOptions.maxRetry + 1; i++) {
       try {
@@ -292,11 +314,12 @@ export abstract class LLMStream implements AsyncIterableIterator<ChatChunk> {
     }
   };
 
-  private mainTask = async () =>
-    tracer.startActiveSpan(async (span) => this._mainTaskImpl(span), {
+  private mainTask = async () => {
+    return tracer.startActiveSpan(async (span) => this._mainTaskImpl(span), {
       name: 'llm_request',
       endOnExit: false,
     });
+  };
 
   private emitError({ error, recoverable }: { error: Error; recoverable: boolean }) {
     this.#llm.emit('error', {
@@ -314,6 +337,9 @@ export abstract class LLMStream implements AsyncIterableIterator<ChatChunk> {
     let requestId = '';
     let usage: CompletionUsage | undefined;
     let completionStartTime: string | undefined;
+    // accumulated for `gen_ai.output.messages`
+    let responseContent = '';
+    const toolCalls: FunctionCall[] = [];
 
     for await (const ev of this.queue) {
       if (this.abortController.signal.aborted) {
@@ -330,6 +356,12 @@ export abstract class LLMStream implements AsyncIterableIterator<ChatChunk> {
         ttft = process.hrtime.bigint() - startTime;
         completionStartTime = new Date().toISOString();
       }
+      if (ev.delta?.content) {
+        responseContent += ev.delta.content;
+      }
+      if (ev.delta?.toolCalls?.length) {
+        toolCalls.push(...ev.delta.toolCalls);
+      }
       if (ev.usage) {
         usage = ev.usage;
       }
@@ -366,9 +398,25 @@ export abstract class LLMStream implements AsyncIterableIterator<ChatChunk> {
     if (this.#llmRequestSpan) {
       this.#llmRequestSpan.setAttribute(traceTypes.ATTR_LLM_METRICS, JSON.stringify(metrics));
 
-      this.#llmRequestSpan.setAttributes({
-        [traceTypes.ATTR_GEN_AI_USAGE_INPUT_TOKENS]: metrics.promptTokens,
-        [traceTypes.ATTR_GEN_AI_USAGE_OUTPUT_TOKENS]: metrics.completionTokens,
+      // the GenAI response side; the request side was recorded at span creation
+      genAI.setUsageAttributes(this.#llmRequestSpan, metrics);
+
+      const finishReason = genAI.finishReasonFor({
+        functionCalls: toolCalls,
+        interrupted: metrics.cancelled,
+      });
+      genAI.setResponseAttributes(this.#llmRequestSpan, {
+        responseId: requestId || undefined,
+        model: this.#llm.model,
+        finishReasons: [finishReason],
+        timeToFirstChunk: metrics.ttftMs >= 0 ? metrics.ttftMs / 1000 : undefined,
+      });
+      genAI.setContentAttributes(this.#llmRequestSpan, {
+        outputMessages: genAI.toOutputMessages({
+          text: responseContent,
+          functionCalls: toolCalls,
+          finishReason,
+        }),
       });
 
       if (completionStartTime) {