diff --git a/.changeset/agent-turn-per-speech.md b/.changeset/agent-turn-per-speech.md new file mode 100644 index 000000000..629ea0f62 --- /dev/null +++ b/.changeset/agent-turn-per-speech.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +One `agent_turn` span per speech handle: the follow-up generation after a tool call continues the open span instead of opening a second turn, each generation is a `generation` event with `lk.generation_count` on the span, the turn ends with the speech, and a discarded preemptive generation hands its turn to the reply that answered. diff --git a/agents/etc/agents.api.md b/agents/etc/agents.api.md index c1242415c..656bfaed0 100644 --- a/agents/etc/agents.api.md +++ b/agents/etc/agents.api.md @@ -1133,7 +1133,7 @@ const ATTR_AGENT_PARENT_TURN_ID = "lk.parent_generation_id"; // Warning: (ae-missing-release-tag) "ATTR_AGENT_TURN_ID" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // -// @public (undocumented) +// @public const ATTR_AGENT_TURN_ID = "lk.generation_id"; // Warning: (ae-missing-release-tag) "ATTR_AMD_CATEGORY" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) @@ -1636,6 +1636,11 @@ const ATTR_GEN_AI_USAGE_TEXT_OUTPUT_TOKENS = "gen_ai.usage.text.output_tokens"; // @public (undocumented) const ATTR_GEN_AI_WORKFLOW_NAME = "gen_ai.workflow.name"; +// Warning: (ae-missing-release-tag) "ATTR_GENERATION_COUNT" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +const ATTR_GENERATION_COUNT = "lk.generation_count"; + // Warning: (ae-missing-release-tag) "ATTR_INSTRUCTIONS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -6135,6 +6140,11 @@ class MetadataLogProcessor implements LogRecordProcessor { shutdown(): Promise; } +// Warning: (ae-missing-release-tag) "METRIC_GEN_AI_INVOKE_AGENT_DURATION" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +const METRIC_GEN_AI_INVOKE_AGENT_DURATION = "gen_ai.invoke_agent.duration"; + declare namespace metrics { export { AgentMetrics, @@ -7963,7 +7973,13 @@ export class SpeechHandle { // @internal (undocumented) _addItemAddedCallback(callback: (item: ChatItem) => void): void; // @internal + _agentTurnAgentName?: string; + // @internal _agentTurnContext?: Context; + // @internal + _agentTurnSpan?: Span; + // @internal + _agentTurnStartedAt?: number; // (undocumented) get allowInterruptions(): boolean; set allowInterruptions(value: boolean); @@ -7977,6 +7993,10 @@ export class SpeechHandle { get chatItems(): ChatItem[]; // @internal (undocumented) _clearAuthorization(): void; + // Warning: (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "_takeAgentTurn" + // + // @internal + _continueAgentTurn(carry: AgentTurnCarry, discarded: SpeechHandle): void; // (undocumented) static create(options?: { allowInterruptions?: boolean; @@ -7987,6 +8007,8 @@ export class SpeechHandle { // (undocumented) done(): boolean; exception(): unknown; + // @internal + get _generationId(): string; // @internal (undocumented) get _hasGenerations(): boolean; // @internal (undocumented) @@ -8017,6 +8039,8 @@ export class SpeechHandle { // (undocumented) readonly parent?: SpeechHandle | undefined; // @internal + get _parentGenerationId(): string | undefined; + // @internal _queueWait(): number | undefined; // @internal (undocumented) _releaseInterruptions(): void; @@ -8033,6 +8057,10 @@ export class SpeechHandle { static SPEECH_PRIORITY_NORMAL: number; // @internal (undocumented) _stepIndex: number; + // Warning: (ae-forgotten-export) The symbol "AgentTurnCarry" needs to be exported by the entry point index.d.ts + // + // @internal + _takeAgentTurn(): AgentTurnCarry | undefined; // @internal (undocumented) _tasks: Task[]; then(onFulfilled?: ((value: ResolvedSpeechHandle) => R1 | PromiseLike) | null, onRejected?: ((reason: unknown) => R2 | PromiseLike) | null): Promise; @@ -9368,6 +9396,7 @@ declare namespace traceTypes { ATTR_CALLBACK_NAME, ATTR_AGENT_TURN_ID, ATTR_AGENT_PARENT_TURN_ID, + ATTR_GENERATION_COUNT, ATTR_USER_INPUT, ATTR_INSTRUCTIONS, ATTR_SPEECH_INTERRUPTED, @@ -9497,7 +9526,8 @@ declare namespace traceTypes { ATTR_EXCEPTION_TRACE, ATTR_EXCEPTION_TYPE, ATTR_EXCEPTION_MESSAGE, - ATTR_LANGFUSE_COMPLETION_START_TIME + ATTR_LANGFUSE_COMPLETION_START_TIME, + METRIC_GEN_AI_INVOKE_AGENT_DURATION } } diff --git a/agents/src/telemetry/otel_metrics.ts b/agents/src/telemetry/otel_metrics.ts index ecf06d9ea..5ce82cd32 100644 --- a/agents/src/telemetry/otel_metrics.ts +++ b/agents/src/telemetry/otel_metrics.ts @@ -3,28 +3,41 @@ // SPDX-License-Identifier: Apache-2.0 import { type Attributes, type Histogram, type MeterProvider, metrics } from '@opentelemetry/api'; import { getJobContext } from '../job.js'; +import * as traceTypes from './trace_types.js'; +// Instruments are looked up per call: the global meter provider may be installed after this +// module loads (the cloud pipeline is set up when the job registers), and a histogram created +// on the no-op provider would stay a no-op. let meterProvider: MeterProvider | undefined; -let blockedDuration: Histogram | undefined; +const histograms = new Map(); -function eventLoopBlockedHistogram(): Histogram { +function histogram(name: string, description: string): Histogram { const currentProvider = metrics.getMeterProvider(); - if (currentProvider !== meterProvider || !blockedDuration) { + if (currentProvider !== meterProvider) { meterProvider = currentProvider; - blockedDuration = metrics - .getMeter('livekit-agents') - .createHistogram('lk.agents.event_loop.blocked_duration', { - unit: 's', - description: 'Duration of synchronous blocks detected on an agent event loop', - }); + histograms.clear(); } - return blockedDuration; + let instrument = histograms.get(name); + if (!instrument) { + instrument = metrics.getMeter('livekit-agents').createHistogram(name, { + unit: 's', + description, + }); + histograms.set(name, instrument); + } + return instrument; } -/** Record an event-loop stall in seconds, with the severity and what caused it. */ -export function recordEventLoopBlocked(duration: number, severity: string, cause: string): void { +/** + * Per-measurement job attribution. + * + * The meter provider has process lifetime (the OTel metrics global is set-once), so per-job + * fields cannot live on its resource. Each measurement carries the same per-job attributes that + * are stamped on spans and logs instead. Returns a fresh object; callers may add to it. + */ +function jobAttrs(): Attributes { const ctx = getJobContext(false); - const attributes: Attributes = { severity, cause }; + const attributes: Attributes = {}; if (ctx) { Object.assign(attributes, ctx._otelMetadata()); const roomId = ctx.job.room?.sid; @@ -32,5 +45,27 @@ export function recordEventLoopBlocked(duration: number, severity: string, cause if (ctx.job.id) attributes.job_id = ctx.job.id; if (ctx.job.agentName) attributes['lk.agent_name'] = ctx.job.agentName; } - eventLoopBlockedHistogram().record(duration, attributes); + return attributes; +} + +/** Record an event-loop stall in seconds, with the severity and what caused it. */ +export function recordEventLoopBlocked(duration: number, severity: string, cause: string): void { + const attributes = jobAttrs(); + attributes.severity = severity; + attributes.cause = cause; + histogram( + 'lk.agents.event_loop.blocked_duration', + 'Duration of synchronous blocks detected on an agent event loop', + ).record(duration, attributes); +} + +/** `gen_ai.invoke_agent.duration` for one agent turn, in seconds. */ +export function recordInvokeAgentDuration(duration: number, agentName: string): void { + const attributes = jobAttrs(); + attributes[traceTypes.ATTR_GEN_AI_OPERATION_NAME] = traceTypes.GenAIOperationName.INVOKE_AGENT; + attributes[traceTypes.ATTR_GEN_AI_AGENT_NAME] = agentName; + histogram(traceTypes.METRIC_GEN_AI_INVOKE_AGENT_DURATION, 'Agent invocation duration').record( + duration, + attributes, + ); } diff --git a/agents/src/telemetry/trace_types.test.ts b/agents/src/telemetry/trace_types.test.ts index 922e0859c..e11514eb0 100644 --- a/agents/src/telemetry/trace_types.test.ts +++ b/agents/src/telemetry/trace_types.test.ts @@ -132,6 +132,7 @@ const SAFE_KEYS = new Set([ 'lk.deployment_id', 'lk.session_options', 'lk.generation_id', + 'lk.generation_count', 'lk.parent_generation_id', 'lk.interrupted', // LLM node metadata @@ -299,7 +300,8 @@ const SAFE_KEYS = new Set([ function declaredKeys(): Record { return Object.fromEntries( Object.entries(traceTypes).filter((entry): entry is [string, string] => { - return typeof entry[1] === 'string'; + // metric names are not attribute keys: they carry no values to classify + return typeof entry[1] === 'string' && !entry[0].startsWith('METRIC_'); }), ); } diff --git a/agents/src/telemetry/trace_types.ts b/agents/src/telemetry/trace_types.ts index 84f80eb25..e54e8bdf2 100644 --- a/agents/src/telemetry/trace_types.ts +++ b/agents/src/telemetry/trace_types.ts @@ -122,8 +122,17 @@ export const ATTR_SHUTDOWN_USER_INITIATED = 'lk.shutdown.user_initiated'; export const ATTR_CALLBACK_NAME = 'lk.callback.name'; // assistant turn +/** + * On `agent_turn`: the latest generation (LLM step) of the speech; each step is also a + * `generation` event carrying its own id. + */ export const ATTR_AGENT_TURN_ID = 'lk.generation_id'; export const ATTR_AGENT_PARENT_TURN_ID = 'lk.parent_generation_id'; +/** + * On `agent_turn`: how many generations (LLM steps) the speech took; more than one means tool + * calls were executed before the final reply. + */ +export const ATTR_GENERATION_COUNT = 'lk.generation_count'; export const ATTR_USER_INPUT = 'lk.pii.user_input'; export const ATTR_INSTRUCTIONS = 'lk.pii.instructions'; export const ATTR_SPEECH_INTERRUPTED = 'lk.interrupted'; @@ -488,3 +497,7 @@ export const ATTR_EXCEPTION_MESSAGE = 'exception.message'; // Platform-specific attributes export const ATTR_LANGFUSE_COMPLETION_START_TIME = 'langfuse.observation.completion_start_time'; + +// metric names (OpenTelemetry GenAI semantic conventions) +/** Histogram, seconds: one agent turn (`invoke_agent`), however many LLM steps it took. */ +export const METRIC_GEN_AI_INVOKE_AGENT_DURATION = 'gen_ai.invoke_agent.duration'; diff --git a/agents/src/voice/agent_activity.test.ts b/agents/src/voice/agent_activity.test.ts index 8f82706ee..95d53d614 100644 --- a/agents/src/voice/agent_activity.test.ts +++ b/agents/src/voice/agent_activity.test.ts @@ -905,7 +905,12 @@ function buildPreemptiveRunner(opts: Partial = {}) { }; const generateReply = vi.fn( - () => ({ id: 'speech_fake', _cancel: () => {} }) as unknown as SpeechHandle, + () => + ({ + id: 'speech_fake', + _cancel: () => {}, + _takeAgentTurn: () => undefined, + }) as unknown as SpeechHandle, ); const cancelPreemptiveGeneration = vi.fn(); diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index 215f7328d..214c5aa93 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -371,9 +371,10 @@ function recordQueueWait(speechHandle: SpeechHandle): void { if (queueWait === undefined || speechHandle._agentTurnContext === undefined) { return; // no agent_turn span yet: never fall back to whatever span is current } - trace - .getSpan(speechHandle._agentTurnContext) - ?.setAttribute(traceTypes.ATTR_SPEECH_QUEUE_WAIT, queueWait / 1000); + const span = trace.getSpan(speechHandle._agentTurnContext); + if (span?.isRecording()) { + span.setAttribute(traceTypes.ATTR_SPEECH_QUEUE_WAIT, queueWait / 1000); + } } /** @@ -384,12 +385,79 @@ function recordInterruption(speechHandle: SpeechHandle): void { if (!speechHandle.interrupted || speechHandle._agentTurnContext === undefined) { return; } - trace.getSpan(speechHandle._agentTurnContext)?.setAttributes({ + const span = trace.getSpan(speechHandle._agentTurnContext); + if (!span?.isRecording()) return; // the turn may have ended with the speech + span.setAttributes({ [traceTypes.ATTR_SPEECH_INTERRUPTED]: true, [traceTypes.ATTR_INTERRUPTION_SOURCE]: speechHandle._interruptSource ?? 'programmatic', }); } +/** + * Run `fn` under the speech's `agent_turn` span, made current for one generation. + * + * One speech handle is one agent turn, however many LLM steps it takes: the follow-up + * generation after a tool call runs in a new task but continues the open span instead of + * opening a second turn. Each generation is a `generation` event on the span, whose + * `lk.generation_id` names the latest one and `lk.generation_count` how many there were. The + * span ends with the speech (`SpeechHandle._markDone`), not with the step. + * + * Module-level for the same reason as `recordQueueWait`. + * @internal + */ +export async function withAgentTurn( + speechHandle: SpeechHandle, + options: { rootContext: Context | undefined; agentLabel: string }, + fn: (span: Span) => Promise, +): Promise { + let span = speechHandle._agentTurnSpan; + if (span === undefined) { + span = tracer.startSpan({ + name: 'agent_turn', + context: options.rootContext, + attributes: { [traceTypes.ATTR_SPEECH_ID]: speechHandle.id }, + }); + // an agent turn is the convention's `invoke_agent`: the framework running the agent + // in-process, with the inference and tool spans nested underneath + genAI.setAgentAttributes(span, { + operation: traceTypes.GenAIOperationName.INVOKE_AGENT, + agentName: options.agentLabel, + }); + speechHandle._agentTurnSpan = span; + speechHandle._agentTurnContext = trace.setSpan(options.rootContext ?? ROOT_CONTEXT, span); + speechHandle._agentTurnStartedAt = performance.now(); + speechHandle._agentTurnAgentName = options.agentLabel; + } + + const generationAttrs: Record = { + [traceTypes.ATTR_AGENT_TURN_ID]: speechHandle._generationId, + }; + const parentId = speechHandle._parentGenerationId; + if (parentId) generationAttrs[traceTypes.ATTR_AGENT_PARENT_TURN_ID] = parentId; + span.addEvent('generation', generationAttrs); + span.setAttributes({ + [traceTypes.ATTR_AGENT_TURN_ID]: speechHandle._generationId, + [traceTypes.ATTR_GENERATION_COUNT]: speechHandle._numSteps, + }); + const turnSpan = span; + return otelContext.with(speechHandle._agentTurnContext!, () => fn(turnSpan)); +} + +/** + * A preemptive generation discarded for `successor` (a newer attempt, or the real reply after + * the transcript changed) hands its open `agent_turn` over, so one turn shows the wasted + * generation and the one that answered. Module-level like `recordQueueWait`. + * @internal + */ +export function continueDiscardedTurn( + discarded: SpeechHandle | undefined, + successor: SpeechHandle, +): void { + if (discarded === undefined || discarded === successor) return; + const carry = discarded._takeAgentTurn(); + if (carry !== undefined) successor._continueAgentTurn(carry, discarded); +} + export class AgentActivity implements RecognitionHooks { agent: Agent; agentSession: AgentSession; @@ -2220,7 +2288,10 @@ export class AgentActivity implements RecognitionHooks { return; } - // more of the user's turn arrived: the attempt answered a transcript that is now stale + // a newer attempt supersedes the current one; if one is created below it continues the + // discarded attempt's agent_turn (the cancelled speech only ends once the loop runs). More + // of the user's turn arrived: the attempt answered a transcript that is now stale + const discarded = this._preemptiveGeneration?.speechHandle; this.cancelPreemptiveGeneration('user_turn'); if ( @@ -2255,6 +2326,8 @@ export class AgentActivity implements RecognitionHooks { chatCtx, scheduleSpeech: false, inputDetails: { modality: 'audio' }, + // a newer attempt supersedes the current one: it continues that attempt's agent_turn + continueTurnFrom: discarded, }); this._preemptiveGeneration = { @@ -2813,6 +2886,12 @@ export class AgentActivity implements RecognitionHooks { allowInterruptions?: boolean; scheduleSpeech?: boolean; inputDetails?: InputDetails; + /** + * A discarded preemptive attempt answering the same user turn: the new speech continues + * its open `agent_turn` (see {@link continueDiscardedTurn}). Handed over before the reply + * task starts, since the task opens the turn in its first synchronous statements. + */ + continueTurnFrom?: SpeechHandle; }): SpeechHandle { const { userMessage, @@ -2822,6 +2901,7 @@ export class AgentActivity implements RecognitionHooks { allowInterruptions: defaultAllowInterruptions, scheduleSpeech = true, inputDetails, + continueTurnFrom, } = options; let instructions: string | Instructions | undefined = defaultInstructions; @@ -2860,6 +2940,9 @@ export class AgentActivity implements RecognitionHooks { allowInterruptions: allowInterruptions ?? this.allowInterruptions, inputDetails, }); + // before the reply task below runs (Task starts its body synchronously): the task must find + // the adopted turn on the handle, or it opens a second one that nothing ever ends + continueDiscardedTurn(continueTurnFrom, handle); this.agentSession.emit( AgentSessionEventTypes.SpeechCreated, @@ -3197,6 +3280,7 @@ export class AgentActivity implements RecognitionHooks { } let speechHandle: SpeechHandle | undefined; + let discardedPreemptive: SpeechHandle | undefined; if (this._preemptiveGeneration !== undefined) { const preemptive = this._preemptiveGeneration; // make sure the onUserTurnCompleted didn't change some request parameters @@ -3227,6 +3311,7 @@ export class AgentActivity implements RecognitionHooks { this.logger.warn( 'preemptive generation invalidated after `onUserTurnCompleted` because the transcript, chat context, tools, or tool choice changed', ); + discardedPreemptive = preemptive.speechHandle; preemptive.speechHandle._cancel('user_turn'); } @@ -3240,6 +3325,8 @@ export class AgentActivity implements RecognitionHooks { userMessage, chatCtx, inputDetails: { modality: 'audio' }, + // the invalidated preemptive attempt answered this same turn: one agent_turn + continueTurnFrom: discardedPreemptive, }); } @@ -3266,9 +3353,31 @@ export class AgentActivity implements RecognitionHooks { modelSettings: ModelSettings, replyAbortController: AbortController, audio?: ReadableStream | null, + ): Promise { + return withAgentTurn( + stateLease.speechHandle, + { rootContext: this.agentSession.rootSpanContext, agentLabel: this.agent.id }, + () => + this.ttsTaskImpl( + stateLease, + text, + addToChatCtx, + modelSettings, + replyAbortController, + audio, + ), + ); + } + + private async ttsTaskImpl( + stateLease: AgentStateLease, + text: string | ReadableStream, + addToChatCtx: boolean, + modelSettings: ModelSettings, + replyAbortController: AbortController, + audio?: ReadableStream | null, ): Promise { const { speechHandle } = stateLease; - speechHandle._agentTurnContext = otelContext.active(); speechHandleStorage.enterWith(speechHandle); @@ -3494,7 +3603,6 @@ export class AgentActivity implements RecognitionHooks { _previousUserMetrics?: MetricsReport; }): Promise => { const { speechHandle } = stateLease; - speechHandle._agentTurnContext = otelContext.active(); span.setAttribute(traceTypes.ATTR_SPEECH_ID, speechHandle.id); if (instructions) { @@ -4142,17 +4250,6 @@ export class AgentActivity implements RecognitionHooks { } }; - /** - * An agent turn is the convention's `invoke_agent`: the framework running the agent - * in-process, with the inference (`chat`) and tool (`execute_tool`) spans nested underneath. - */ - private recordAgentTurn(span: Span): void { - genAI.setAgentAttributes(span, { - operation: traceTypes.GenAIOperationName.INVOKE_AGENT, - agentName: this.agent.id, - }); - } - private pipelineReplyTask = async ( stateLease: AgentStateLease, chatCtx: ChatContext, @@ -4163,9 +4260,10 @@ export class AgentActivity implements RecognitionHooks { newMessage?: ChatMessage, _previousUserMetrics?: MetricsReport, ): Promise => - tracer.startActiveSpan( - async (span) => ( - this.recordAgentTurn(span), + withAgentTurn( + stateLease.speechHandle, + { rootContext: this.agentSession.rootSpanContext, agentLabel: this.agent.id }, + (span) => this._pipelineReplyTaskImpl({ stateLease, chatCtx, @@ -4176,12 +4274,7 @@ export class AgentActivity implements RecognitionHooks { newMessage, span, _previousUserMetrics, - }) - ), - { - name: 'agent_turn', - context: this.agentSession.rootSpanContext, - }, + }), ); private async realtimeGenerationTask( @@ -4191,9 +4284,10 @@ export class AgentActivity implements RecognitionHooks { replyAbortController: AbortController, addToChatCtx: boolean = true, ): Promise { - return tracer.startActiveSpan( + return withAgentTurn( + stateLease.speechHandle, + { rootContext: this.agentSession.rootSpanContext, agentLabel: this.agent.id }, async (span) => { - this.recordAgentTurn(span); const inferenceSpan = tracer.startSpan({ name: 'realtime_inference' }); try { return await this._realtimeGenerationTaskImpl({ @@ -4209,10 +4303,6 @@ export class AgentActivity implements RecognitionHooks { inferenceSpan.end(); } }, - { - name: 'agent_turn', - context: this.agentSession.rootSpanContext, - }, ); } @@ -4234,7 +4324,6 @@ export class AgentActivity implements RecognitionHooks { inferenceSpan: Span; }): Promise { const { speechHandle } = stateLease; - speechHandle._agentTurnContext = otelContext.active(); span.setAttribute(traceTypes.ATTR_SPEECH_ID, speechHandle.id); diff --git a/agents/src/voice/agent_turn_span.test.ts b/agents/src/voice/agent_turn_span.test.ts new file mode 100644 index 000000000..7ba15595c --- /dev/null +++ b/agents/src/voice/agent_turn_span.test.ts @@ -0,0 +1,337 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 + +/** + * One `agent_turn` span per speech handle. + * + * A reply that calls a tool runs two generations (LLM steps) in two tasks; they used to be two + * `agent_turn` spans linked only by `lk.parent_generation_id`. The speech handle now owns a + * single span for its whole life: each generation is an event on it, tool and inference spans + * nest under it, and it ends with the speech. + */ +import { AudioFrame } from '@livekit/rtc-node'; +import { + INVALID_SPAN_CONTEXT, + ROOT_CONTEXT, + SpanStatusCode, + context as otelContext, + trace, +} from '@opentelemetry/api'; +import { + InMemorySpanExporter, + type ReadableSpan, + SimpleSpanProcessor, +} from '@opentelemetry/sdk-trace-base'; +import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node'; +import { ReadableStream } from 'node:stream/web'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { ChatMessage } from '../llm/chat_context.js'; +import { tool } from '../llm/tool_context.js'; +import { initializeLogger } from '../log.js'; +import { FakeSTT } from '../stt/testing/fake_stt.js'; +import { setTracerProvider, traceTypes, tracer } from '../telemetry/index.js'; +import * as otelMetrics from '../telemetry/otel_metrics.js'; +import { Agent } from './agent.js'; +import { continueDiscardedTurn, withAgentTurn } from './agent_activity.js'; +import { AgentSession } from './agent_session.js'; +import { AudioOutput } from './io.js'; +import { SpeechHandle } from './speech_handle.js'; +import { FakeLLM } from './testing/fake_llm.js'; + +initializeLogger({ pretty: false, level: 'silent' }); + +function spansNamed(exporter: InMemorySpanExporter, name: string): ReadableSpan[] { + return exporter.getFinishedSpans().filter((span) => span.name === name); +} + +function childrenOf(exporter: InMemorySpanExporter, parent: ReadableSpan, name: string) { + return spansNamed(exporter, name).filter( + (span) => span.parentSpanContext?.spanId === parent.spanContext().spanId, + ); +} + +function ms(time: [number, number]): number { + return time[0] * 1000 + time[1] / 1e6; +} + +/** Reports playout as soon as frames arrive, so a reply "plays" instantly. */ +class ImmediateOutput extends AudioOutput { + constructor() { + super(24_000); + } + + override async captureFrame(frame: AudioFrame): Promise { + const segmentCount = this.capturedPlayoutSegments; + await super.captureFrame(frame); + if (this.capturedPlayoutSegments > segmentCount) { + this.onPlaybackStarted(Date.now()); + } + } + + override flush(): void { + super.flush(); + if (this.pendingPlayoutSegments > 0) { + this.onPlaybackFinished({ playbackPosition: 0.02, interrupted: false }); + } + } + + override clearBuffer(): void { + if (this.pendingPlayoutSegments > 0) { + this.onPlaybackFinished({ playbackPosition: 0, interrupted: true }); + } + } +} + +class WeatherAgent extends Agent { + constructor() { + super({ + instructions: 'You are a helpful assistant.', + tools: { + get_weather: tool({ + description: 'Look up the weather', + execute: async () => 'sunny in Tokyo', + }), + }, + }); + } + + override async ttsNode(): Promise> { + return new ReadableStream({ + start(controller) { + controller.enqueue(new AudioFrame(new Int16Array(480), 24_000, 1, 480)); + controller.close(); + }, + }); + } +} + +describe.sequential('agent_turn span', () => { + let exporter: InMemorySpanExporter; + let provider: NodeTracerProvider; + let originalProvider: ReturnType; + + beforeEach(() => { + originalProvider = tracer.getProvider(); + exporter = new InMemorySpanExporter(); + provider = new NodeTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)] }); + provider.register(); + setTracerProvider(provider); + }); + + afterEach(async () => { + vi.restoreAllMocks(); + setTracerProvider(originalProvider); + await provider.shutdown(); + trace.disable(); + otelContext.disable(); + }); + + async function runReply(llm: FakeLLM, agent: Agent, userInput: string): Promise { + const session = new AgentSession({ llm, stt: new FakeSTT() }); + session.output.audio = new ImmediateOutput(); + await session.start({ agent }); + try { + const speech = session.generateReply({ userInput }); + await speech.waitForPlayout(); + } finally { + await session.close(); + } + } + + it('a tool call is one agent turn', async () => { + const llm = new FakeLLM([ + { + input: "What's the weather in Tokyo?", + content: '', + toolCalls: [{ name: 'get_weather', args: { location: 'Tokyo' } }], + }, + { input: '"sunny in Tokyo"', content: 'It is sunny in Tokyo.' }, + ]); + await runReply(llm, new WeatherAgent(), "What's the weather in Tokyo?"); + + const [root] = spansNamed(exporter, 'agent_session'); + const turns = spansNamed(exporter, 'agent_turn'); + expect( + turns.map((turn) => turn.attributes[traceTypes.ATTR_SPEECH_ID]), + 'one agent_turn per speech', + ).toHaveLength(1); + const [turn] = turns; + expect(turn!.parentSpanContext?.spanId).toBe(root!.spanContext().spanId); + + const attrs = turn!.attributes; + const speechId = attrs[traceTypes.ATTR_SPEECH_ID] as string; + expect(attrs[traceTypes.ATTR_GENERATION_COUNT]).toBe(2); + expect(attrs[traceTypes.ATTR_AGENT_TURN_ID]).toBe(`${speechId}_2`); + const generations = turn!.events.filter((event) => event.name === 'generation'); + expect(generations.map((event) => event.attributes?.[traceTypes.ATTR_AGENT_TURN_ID])).toEqual([ + `${speechId}_1`, + `${speechId}_2`, + ]); + expect(generations[0]!.attributes?.[traceTypes.ATTR_AGENT_PARENT_TURN_ID]).toBeUndefined(); + expect(generations[1]!.attributes?.[traceTypes.ATTR_AGENT_PARENT_TURN_ID]).toBe( + `${speechId}_1`, + ); + + // both generations' inference, the tool between them, and the speech all nest under it + expect(childrenOf(exporter, turn!, 'llm_node')).toHaveLength(2); + const [toolSpan] = childrenOf(exporter, turn!, 'function_tool'); + const [tts] = childrenOf(exporter, turn!, 'tts_node'); + const [speaking] = childrenOf(exporter, turn!, 'agent_speaking'); + expect(toolSpan).toBeDefined(); + expect(tts).toBeDefined(); + expect(speaking).toBeDefined(); + expect(ms(toolSpan!.startTime)).toBeLessThan(ms(tts!.startTime)); + // and the turn covers everything, ending with the speech rather than with the first step + // (2 ms of slack: the SDK anchors each span's clock at creation) + for (const child of [toolSpan!, tts!, speaking!]) { + expect(ms(turn!.startTime)).toBeLessThanOrEqual(ms(child.startTime) + 2); + expect(ms(child.endTime)).toBeLessThanOrEqual(ms(turn!.endTime) + 2); + } + }); + + it('a plain reply is one generation', async () => { + const llm = new FakeLLM([{ input: 'Hello', content: 'Hi there' }]); + await runReply(llm, new WeatherAgent(), 'Hello'); + + const [turn] = spansNamed(exporter, 'agent_turn'); + expect(spansNamed(exporter, 'agent_turn')).toHaveLength(1); + const attrs = turn!.attributes; + expect(attrs[traceTypes.ATTR_GENERATION_COUNT]).toBe(1); + expect(attrs[traceTypes.ATTR_AGENT_TURN_ID]).toBe(`${attrs[traceTypes.ATTR_SPEECH_ID]}_1`); + expect(turn!.events.filter((event) => event.name === 'generation')).toHaveLength(1); + expect(attrs[traceTypes.ATTR_AGENT_PARENT_TURN_ID]).toBeUndefined(); + }); + + it('a discarded preemptive generation hands its turn to the successor', async () => { + // a preemptive attempt discarded for the real reply (or a newer attempt) must not leave a + // second agent_turn behind: the successor continues the span, the discarded speech ends + // without touching it + const root = tracer.startSpan({ name: 'agent_session' }); + const rootCtx = trace.setSpan(ROOT_CONTEXT, root); + const attempt = SpeechHandle.create({ allowInterruptions: true }); + await withAgentTurn(attempt, { rootContext: rootCtx, agentLabel: 'a' }, async () => { + // the attempt's first generation ran here + }); + + const reply = SpeechHandle.create({ allowInterruptions: true }); + continueDiscardedTurn(attempt, reply); + attempt._markDone(); // the cancelled attempt finishes: the span must survive it + expect(spansNamed(exporter, 'agent_turn')).toEqual([]); + + await withAgentTurn(reply, { rootContext: rootCtx, agentLabel: 'a' }, async () => {}); + reply._markDone(); + root.end(); + + const [turn] = spansNamed(exporter, 'agent_turn'); + expect(spansNamed(exporter, 'agent_turn')).toHaveLength(1); + const attrs = turn!.attributes; + expect(attrs[traceTypes.ATTR_SPEECH_ID]).toBe(reply.id); + expect(attrs[traceTypes.ATTR_GENERATION_COUNT]).toBe(1); // the reply's own step count + expect(turn!.events.map((event) => event.name)).toEqual([ + 'generation', + 'preemptive_generation_discarded', + 'generation', + ]); + const discarded = turn!.events.find( + (event) => event.name === 'preemptive_generation_discarded', + ); + expect(discarded?.attributes?.[traceTypes.ATTR_SPEECH_ID]).toBe(attempt.id); + + // nothing to hand over: a plain successor is untouched + continueDiscardedTurn(undefined, reply); + continueDiscardedTurn(reply, reply); + }); + + it('hands the turn over before the successor task opens its own', async () => { + // Task runs its body synchronously, and the reply task opens agent_turn in its first + // statements: a handoff performed after generateReply() returned came too late, leaving the + // successor's own span unended and its llm_node / tts_node dangling from it. The handoff now + // happens inside generateReply, before the task starts. + const llm = new FakeLLM([{ input: 'Hello', content: 'Hi there' }]); + const session = new AgentSession({ llm, stt: new FakeSTT() }); + session.output.audio = new ImmediateOutput(); + const agent = new WeatherAgent(); + await session.start({ agent }); + try { + const activity = agent._agentActivity!; + const attempt = SpeechHandle.create({ allowInterruptions: true }); + await withAgentTurn( + attempt, + { rootContext: session.rootSpanContext, agentLabel: agent.id }, + async () => {}, + ); + const reply = activity.generateReply({ + userMessage: ChatMessage.create({ role: 'user', content: 'Hello' }), + inputDetails: { modality: 'audio' }, + continueTurnFrom: attempt, + }); + attempt._markDone(); + await reply.waitForPlayout(); + } finally { + await session.close(); + } + + const turns = spansNamed(exporter, 'agent_turn'); + expect(turns).toHaveLength(1); + const [turn] = turns; + expect(turn!.attributes[traceTypes.ATTR_GENERATION_COUNT]).toBe(1); + expect(turn!.events.map((event) => event.name)).toEqual([ + 'generation', + 'preemptive_generation_discarded', + 'generation', + ]); + // the successor's work nests under the adopted turn + expect(childrenOf(exporter, turn!, 'llm_node')).toHaveLength(1); + expect(childrenOf(exporter, turn!, 'tts_node')).toHaveLength(1); + expect(childrenOf(exporter, turn!, 'agent_speaking')).toHaveLength(1); + // and nothing dangles from a span that never ended + const exported = new Set(exporter.getFinishedSpans().map((span) => span.spanContext().spanId)); + const dangling = exporter + .getFinishedSpans() + .filter((span) => span.parentSpanContext && !exported.has(span.parentSpanContext.spanId)); + expect(dangling.map((span) => span.name)).toEqual([]); + }); + + it('an LLM failure stored on the handle fails the turn', async () => { + // the pipeline stores the failure on the handle when it marks it done; the turn must end as + // failed whichever step it was on + const handle = SpeechHandle.create({ allowInterruptions: true }); + await withAgentTurn(handle, { rootContext: undefined, agentLabel: 'a' }, async () => {}); + handle._markDone(new Error('llm down')); + + const [turn] = spansNamed(exporter, 'agent_turn'); + expect(turn!.status.code).toBe(SpanStatusCode.ERROR); + expect(turn!.events.filter((event) => event.name === 'exception')).toHaveLength(1); + }); + + it('records the turn duration metric even when the span is sampled out', () => { + const record = vi.spyOn(otelMetrics, 'recordInvokeAgentDuration').mockImplementation(() => {}); + const handle = SpeechHandle.create({ allowInterruptions: true }); + handle._agentTurnSpan = trace.wrapSpanContext(INVALID_SPAN_CONTEXT); + handle._agentTurnStartedAt = performance.now() - 1_000; + handle._agentTurnAgentName = 'a'; + handle._markDone(); + expect(record).toHaveBeenCalledTimes(1); + const [duration, agentName] = record.mock.calls[0]!; + expect(agentName).toBe('a'); + expect(duration).toBeGreaterThanOrEqual(1); + expect(duration).toBeLessThan(5); + }); + + it('hands a sampled-out turn to the successor too', () => { + // the successor adopts a non-recording turn as well, so the duration metric keeps the + // discarded attempt's start time + const attempt = SpeechHandle.create({ allowInterruptions: true }); + attempt._agentTurnSpan = trace.wrapSpanContext(INVALID_SPAN_CONTEXT); + attempt._agentTurnStartedAt = 1; + attempt._agentTurnAgentName = 'a'; + + const reply = SpeechHandle.create({ allowInterruptions: true }); + continueDiscardedTurn(attempt, reply); + expect(attempt._agentTurnSpan).toBeUndefined(); + expect(reply._agentTurnSpan).toBeDefined(); + expect(reply._agentTurnStartedAt).toBe(1); + expect(reply._agentTurnAgentName).toBe('a'); + }); +}); diff --git a/agents/src/voice/speech_handle.ts b/agents/src/voice/speech_handle.ts index 37160a766..3ba05a139 100644 --- a/agents/src/voice/speech_handle.ts +++ b/agents/src/voice/speech_handle.ts @@ -2,9 +2,12 @@ // // SPDX-License-Identifier: Apache-2.0 import { ThrowsPromise } from '@livekit/throws-transformer/throws'; -import type { Context } from '@opentelemetry/api'; +import { type Context, type Span, context as otelContext, trace } from '@opentelemetry/api'; import type { ChatItem } from '../llm/index.js'; import { log } from '../log.js'; +import { recordInvokeAgentDuration } from '../telemetry/otel_metrics.js'; +import * as traceTypes from '../telemetry/trace_types.js'; +import { recordException } from '../telemetry/utils.js'; import type { Task } from '../utils.js'; import { Event, Future, dedent, shortuuid } from '../utils.js'; import { functionCallStorage } from './agent.js'; @@ -96,6 +99,13 @@ export type ResolvedSpeechHandle = Omit; */ export type InterruptionSource = 'audio_activity' | 'user_turn' | 'programmatic'; +/** An open `agent_turn` handed from a discarded speech to its successor. @internal */ +export interface AgentTurnCarry { + span: Span; + startedAt: number | undefined; + agentName: string | undefined; +} + export class SpeechHandleCircularWaitError extends Error { constructor(functionCallName: string) { super(dedent` @@ -149,8 +159,17 @@ export class SpeechHandle { /** @internal */ _numSteps = 1; + /** + * @internal One `agent_turn` span for the whole speech, however many generations (LLM steps) + * it takes; opened by the first reply task, ended with the speech in `_markDone`. + */ + _agentTurnSpan?: Span; /** @internal - OpenTelemetry context for the agent turn span */ _agentTurnContext?: Context; + /** @internal - when the turn opened (performance.now), for the duration metric */ + _agentTurnStartedAt?: number; + /** @internal - the agent the turn was opened for, for the duration metric */ + _agentTurnAgentName?: string; /** @internal - when the speech was scheduled, for the queue-wait attribute */ _scheduledAt?: number; @@ -224,6 +243,17 @@ export class SpeechHandle { return this._id; } + /** @internal The id of the current generation (LLM step) of this speech. */ + get _generationId(): string { + return `${this._id}_${this._numSteps}`; + } + + /** @internal The id of the generation before the current one; undefined on the first. */ + get _parentGenerationId(): string | undefined { + if (this._numSteps <= 1) return undefined; + return `${this._id}_${this._numSteps - 1}`; + } + get scheduled(): boolean { return this.scheduledFut.done; } @@ -528,6 +558,8 @@ export class SpeechHandle { } this.doneFut.resolve(); } + // a pipeline LLM failure is stored on the handle before the tasks finish + this.endAgentTurn(error !== undefined ? error : this._error); // Keep this outside the doneFut guard: if the handle is already done but a // generation future is still active, _waitForGeneration() must be released. @@ -538,6 +570,69 @@ export class SpeechHandle { this.clearInterruptTimeout(); } + /** + * Detach this speech's open `agent_turn` so a successor can continue it. + * + * Used when a preemptive generation is discarded for another speech answering the same user + * turn: the wasted generation stays visible under the one turn instead of becoming a turn of + * its own. After this the speech ends without touching the span. + * @internal + */ + _takeAgentTurn(): AgentTurnCarry | undefined { + const span = this._agentTurnSpan; + if (span === undefined) return undefined; + const carry: AgentTurnCarry = { + span, + startedAt: this._agentTurnStartedAt, + agentName: this._agentTurnAgentName, + }; + this._agentTurnSpan = undefined; + this._agentTurnContext = undefined; + this._agentTurnStartedAt = undefined; + this._agentTurnAgentName = undefined; + return carry; + } + + /** @internal Adopt the `agent_turn` taken from `discarded` (see {@link _takeAgentTurn}). */ + _continueAgentTurn(carry: AgentTurnCarry, discarded: SpeechHandle): void { + // adopted even when sampled out: the duration metric still needs the start time + const { span, startedAt, agentName } = carry; + const own = this._agentTurnSpan; + if (own !== undefined && own !== span) { + // this speech already opened a turn of its own (the handoff came after its task started): + // close it rather than leak an unended span that its children would dangle from + own.addEvent('superseded_by_adopted_turn', { [traceTypes.ATTR_SPEECH_ID]: discarded.id }); + if (own.isRecording()) own.end(); + } + span.addEvent('preemptive_generation_discarded', { + [traceTypes.ATTR_SPEECH_ID]: discarded.id, + }); + span.setAttribute(traceTypes.ATTR_SPEECH_ID, this.id); + this._agentTurnSpan = span; + this._agentTurnContext = trace.setSpan(otelContext.active(), span); + this._agentTurnStartedAt = startedAt; + this._agentTurnAgentName = agentName; + } + + /** Close the speech's `agent_turn` span: the speech is done, whatever step it was on. */ + private endAgentTurn(error: unknown): void { + const span = this._agentTurnSpan; + this._agentTurnSpan = undefined; + if (span === undefined) return; + // the duration metric does not depend on the span being sampled in + if (this._agentTurnStartedAt !== undefined && this._agentTurnAgentName !== undefined) { + recordInvokeAgentDuration( + (performance.now() - this._agentTurnStartedAt) / 1000, + this._agentTurnAgentName, + ); + } + if (!span.isRecording()) return; + if (error instanceof Error) { + recordException(span, error); + } + span.end(); + } + /** @internal */ _markScheduled(): void { this._scheduledAt ??= performance.now();