diff --git a/.changeset/tall-spans-delta.md b/.changeset/tall-spans-delta.md new file mode 100644 index 0000000000..fd59f8cc1a --- /dev/null +++ b/.changeset/tall-spans-delta.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +Record LLM trace inputs as deltas of the previous committed generation when `record.inputDelta` is enabled. `gen_ai.system_instructions` now holds only the agent's instructions message; other system messages stay in `gen_ai.input.messages`. diff --git a/agents/etc/agents.api.md b/agents/etc/agents.api.md index db9bdd6ee5..cd3cfb7f7d 100644 --- a/agents/etc/agents.api.md +++ b/agents/etc/agents.api.md @@ -50,6 +50,7 @@ import { SimulationRun_Job } from '@livekit/protocol'; import type { SIPOutboundConfig } from '@livekit/protocol'; import { Span } from '@opentelemetry/api'; import type { Span as Span_2 } from '@opentelemetry/sdk-trace-base'; +import type { SpanContext } from '@opentelemetry/api'; import type { SpanExporter } from '@opentelemetry/sdk-trace-base'; import { SpanKind } from '@opentelemetry/api'; import type { SpanProcessor } from '@opentelemetry/sdk-trace-base'; @@ -78,6 +79,9 @@ export type Aborted = { isAborted: true; }; +// @public (undocumented) +function active(): boolean; + // @public export class AdaptiveNoiseGate implements AudioGate { constructor(options?: AdaptiveNoiseGateOptions); @@ -1335,6 +1339,15 @@ const ATTR_GEN_AI_WORKFLOW_NAME = "gen_ai.workflow.name"; // @public const ATTR_GENERATION_COUNT = "lk.generation_count"; +// @public (undocumented) +const ATTR_INPUT_BASE_SPAN_ID = "lk.input.base_span_id"; + +// @public +const ATTR_INPUT_DELTA = "lk.input.delta"; + +// @public +const ATTR_INPUT_DROPPED_FROM_BASE = "lk.input.dropped_from_base"; + // @public (undocumented) const ATTR_INSTRUCTIONS = "lk.pii.instructions"; @@ -2027,6 +2040,9 @@ export function calculateAudioDurationSeconds(frame: AudioBuffer_2): number; // @public (undocumented) export function cancelAndWait(tasks: Task[], timeout?: number): Promise; +// @public (undocumented) +function captureContentEnabled(): boolean; + // @public (undocumented) type CartesiaModels = 'cartesia/ink-whisper' | 'cartesia/ink-2'; @@ -2514,6 +2530,9 @@ export interface CompletionUsage { totalTokens: number; } +// @public +function compute(site: InputDeltaSite, chatCtx: ChatContext, span: Span): InputDelta; + // Warning: (ae-forgotten-export) The symbol "DiffOps" needs to be exported by the entry point index.d.ts // // @public @@ -2560,6 +2579,9 @@ export type ConversationItemAddedEvent = { createdAt: number; }; +// @internal (undocumented) +function conversationMessages(items: readonly ChatItem[]): ChatMessagePayload[]; + // @public function convertMarkup(provider: string, text: string): string; @@ -3857,10 +3879,15 @@ const GEN_AI_PROVIDER_NAMES: ReadonlySet; declare namespace genAI { export { setCaptureContent, + captureContentEnabled, withInferenceTracking, markInferenceSpanRecorded, + splitInstructions, toSystemInstructions, toInputMessages, + instructionParts, + conversationMessages, + messageLayout, toOutputMessages, toToolDefinitions, finishReasonFor, @@ -3998,7 +4025,6 @@ export class IdleTimeoutError extends Error { export interface ImageContent { // (undocumented) _cache: Record; - // (undocumented) id: string; image: string | VideoFrame_2; // (undocumented) @@ -4187,6 +4213,77 @@ export const initializeLogger: (input: LoggerOptions) => void; // @public (undocumented) function initPinoCloudExporter(config: PinoCloudExporterConfig | PinoCloudExporterUrlConfig): void; +// @internal (undocumented) +interface InputBaseline { + // (undocumented) + instructions: string; + keys: Array<[string, string]>; + layout: Array<'new' | 'merged' | 'skipped'>; + // (undocumented) + spanContext: SpanContext; +} + +// @public +class InputDelta { + constructor(chatCtx: ChatContext, instructions: ChatItem[], conversation: ChatItem[], base?: SpanContext | undefined, droppedFromBase?: number | undefined); + // (undocumented) + readonly base?: SpanContext | undefined; + // (undocumented) + readonly chatCtx: ChatContext; + // (undocumented) + readonly conversation: ChatItem[]; + // (undocumented) + readonly droppedFromBase?: number | undefined; + // (undocumented) + static full(chatCtx: ChatContext): InputDelta; + // (undocumented) + inputMessages(): ChatMessagePayload[]; + // (undocumented) + readonly instructions: ChatItem[]; + // (undocumented) + systemInstructions(): MessagePart[]; +} + +declare namespace inputDelta { + export { + runWithScope, + active, + compute, + setAttributes, + InputDeltaSite, + LLM_NODE, + LLM_REQUEST, + InputBaseline, + InputDelta, + InputDeltaTracker, + InputDeltaScope + } +} + +// @public +class InputDeltaScope { + constructor(tracker: InputDeltaTracker); + // (undocumented) + commit(): void; + // (undocumented) + delta(site: InputDeltaSite, chatCtx: ChatContext, span: Span): InputDelta; +} + +// @public +interface InputDeltaSite { + chatCtx: boolean; + // (undocumented) + name: 'llm_node' | 'llm_request'; +} + +// @public +class InputDeltaTracker { + // @internal (undocumented) + readonly _baselines: Map; + // (undocumented) + begin(): InputDeltaScope; +} + // @public export interface InputDetails { // (undocumented) @@ -4223,6 +4320,9 @@ interface InstructionParts { persona?: Instructions | string; } +// @internal (undocumented) +function instructionParts(items: readonly ChatItem[]): MessagePart[]; + // Warning: (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "asModality" // Warning: (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "asModality" // @@ -4754,6 +4854,12 @@ declare namespace llm_3 { } } +// @public (undocumented) +const LLM_NODE: InputDeltaSite; + +// @public (undocumented) +const LLM_REQUEST: InputDeltaSite; + // @public (undocumented) export type LLMCallbacks = { ['metrics_collected']: (metrics: LLMMetrics) => void; @@ -5052,6 +5158,9 @@ export interface MessageGeneration { textStream: ReadableStream_2; } +// @internal (undocumented) +function messageLayout(items: readonly ChatItem[]): Array<'new' | 'merged' | 'skipped'>; + // @public (undocumented) interface MessagePart { // (undocumented) @@ -6101,6 +6210,9 @@ export function runWithJobContext(context: JobContext, fn: () => T): T; // @internal export function runWithJobContextAsync(context: JobContext, fn: () => Promise): Promise; +// @public +function runWithScope(scope: InputDeltaScope, fn: () => T): T; + export { Scenario } export { ScenarioGroup } @@ -6367,6 +6479,9 @@ function setAgentAttributes(span: Span, params: { provider?: string; }): void; +// @public +function setAttributes(span: Span, delta: InputDelta): void; + // @public function setCaptureContent(enabled: boolean): void; @@ -6863,6 +6978,9 @@ function splitAllMarkup(text: string, options?: { // @public function splitExprMarkup(text: string): [string, ExpressiveTag[]]; +// @internal (undocumented) +function splitInstructions(items: readonly ChatItem[]): [ChatItem[], ChatItem[]]; + // @public const splitWords: (text: string, ignorePunctuation?: boolean) => [string, number, number][]; @@ -7441,6 +7559,7 @@ declare namespace telemetry { PinoCloudExporterUrlConfig, PinoLogObject, genAI, + inputDelta, REDACTED_EXCEPTION_MESSAGE, loopMonitor, rpc, @@ -7939,6 +8058,9 @@ declare namespace traceTypes { ATTR_RESPONSE_TEXT, ATTR_RESPONSE_FUNCTION_CALLS, ATTR_RESPONSE_TTFT, + ATTR_INPUT_DELTA, + ATTR_INPUT_BASE_SPAN_ID, + ATTR_INPUT_DROPPED_FROM_BASE, ATTR_FUNCTION_TOOL_ID, ATTR_FUNCTION_TOOL_NAME, ATTR_FUNCTION_TOOL_ARGS, @@ -9094,18 +9216,18 @@ export const zipFunctionCallsAndOutputs: (event: FunctionToolsExecutedEvent) => // src/_exceptions.ts:90:5 - (ae-forgotten-export) The symbol "APIStatusErrorOptions" needs to be exported by the entry point index.d.ts // src/_exceptions.ts:128:5 - (ae-forgotten-export) The symbol "APIErrorOptions" needs to be exported by the entry point index.d.ts // src/inference/tts.ts:282:5 - (ae-forgotten-export) The symbol "TTSEncoding" needs to be exported by the entry point index.d.ts -// src/llm/chat_context.ts:76:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "audio" +// src/llm/chat_context.ts:80:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "audio" // src/llm/tool_context.ts:702:3 - (ae-unresolved-link) The @link reference could not be resolved: The reference is ambiguous because "ToolFlag" has more than one declaration; you need to add a TSDoc member reference selector // src/llm/tool_context.ts:746:3 - (ae-unresolved-link) The @link reference could not be resolved: The reference is ambiguous because "ToolFlag" has more than one declaration; you need to add a TSDoc member reference selector // src/metrics/base.ts:213:3 - (ae-forgotten-export) The symbol "RealtimeModelMetricsInputTokenDetails" needs to be exported by the entry point index.d.ts // src/metrics/base.ts:217:3 - (ae-forgotten-export) The symbol "RealtimeModelMetricsOutputTokenDetails" needs to be exported by the entry point index.d.ts // src/stt/stt.ts:378:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "STT" // src/utils.ts:468:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "cancelled" -// src/voice/agent_session.ts:394:3 - (ae-unresolved-link) The @link reference could not be resolved: This type of declaration is not supported yet by the resolver -// src/voice/agent_session.ts:1060:5 - (ae-forgotten-export) The symbol "RecordingOptions" needs to be exported by the entry point index.d.ts -// src/voice/agent_session.ts:1776:5 - (ae-forgotten-export) The symbol "STTError" needs to be exported by the entry point index.d.ts -// src/voice/agent_session.ts:1776:5 - (ae-forgotten-export) The symbol "TTSError" needs to be exported by the entry point index.d.ts -// src/voice/agent_session.ts:1776:5 - (ae-forgotten-export) The symbol "LLMError" needs to be exported by the entry point index.d.ts +// src/voice/agent_session.ts:404:3 - (ae-unresolved-link) The @link reference could not be resolved: This type of declaration is not supported yet by the resolver +// src/voice/agent_session.ts:1070:5 - (ae-forgotten-export) The symbol "RecordingOptions" needs to be exported by the entry point index.d.ts +// src/voice/agent_session.ts:1786:5 - (ae-forgotten-export) The symbol "STTError" needs to be exported by the entry point index.d.ts +// src/voice/agent_session.ts:1786:5 - (ae-forgotten-export) The symbol "TTSError" needs to be exported by the entry point index.d.ts +// src/voice/agent_session.ts:1786:5 - (ae-forgotten-export) The symbol "LLMError" needs to be exported by the entry point index.d.ts // src/voice/amd.ts:315:3 - (ae-unresolved-link) The @link reference could not be resolved: The reference is ambiguous because "waitForTrackPublication" has more than one declaration; you need to add a TSDoc member reference selector // src/voice/amd.ts:315:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "gateListening" // src/voice/amd.ts:323:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "aclose" diff --git a/agents/src/llm/chat_context.test.ts b/agents/src/llm/chat_context.test.ts index dda99dc8f2..b3a34b9827 100644 --- a/agents/src/llm/chat_context.test.ts +++ b/agents/src/llm/chat_context.test.ts @@ -8,6 +8,7 @@ import { INSTRUCTIONS_MESSAGE_ID, applyInstructionsModality } from '../voice/gen import { FakeLLM } from '../voice/testing/fake_llm.js'; import { AgentConfigUpdate, + AgentHandoffItem, type AudioContent, ChatContext, type ChatItem, @@ -17,6 +18,7 @@ import { type ImageContent, Instructions, ReadonlyChatContext, + chatItemFingerprint, concatInstructions, isInstructions, renderInstructions, @@ -1660,3 +1662,135 @@ describe('AgentConfigUpdate.toJSON', () => { expect(withoutInstructions).not.toHaveProperty('instructions'); }); }); + +function equivalenceItems(): ChatItem[] { + return [ + new ChatMessage({ id: 'm', role: 'user', content: ['hi'] }), + new FunctionCall({ id: 'c', callId: '1', name: 'f', args: '{}' }), + new FunctionCallOutput({ id: 'o', callId: '1', name: 'f', output: 'ok', isError: false }), + new AgentHandoffItem({ id: 'h', newAgentId: 'b' }), + ]; +} + +describe('ChatItem essential fields and fingerprints', () => { + it.each([ + [0, { role: 'assistant' }], + [0, { content: ['hello'] }], + [0, { interrupted: true }], + [1, { args: '{"x": 1}' }], + [1, { name: 'g' }], + [1, { thoughtSignature: 'signature' }], + [1, { groupId: 'g' }], + [1, { extra: { k: 1 } }], + [2, { output: 'nope' }], + [2, { isError: true }], + ] as const)( + 'essential field change at item %i breaks equivalence and fingerprint', + (index, update) => { + const items = equivalenceItems(); + const changed = [...items]; + changed[index] = Object.assign( + Object.create(Object.getPrototypeOf(items[index]!)), + items[index], + update, + ); + + expect(new ChatContext(items).isEquivalent(new ChatContext(changed))).toBe(false); + expect(chatItemFingerprint(items[index]!)).not.toBe(chatItemFingerprint(changed[index]!)); + }, + ); + + it('metadata does not affect equivalence or fingerprints', () => { + const items = equivalenceItems(); + const changed = items.map((item) => + Object.assign(Object.create(Object.getPrototypeOf(item)), item), + ); + Object.assign(changed[0]!, { createdAt: 0, transcriptConfidence: 0.5, extra: { k: 1 } }); + Object.assign(changed[1]!, { createdAt: 0 }); + Object.assign(changed[2]!, { createdAt: 0 }); + Object.assign(changed[3]!, { newAgentId: 'c' }); + + expect(new ChatContext(items).isEquivalent(new ChatContext(changed))).toBe(true); + expect(items.map(chatItemFingerprint)).toEqual(changed.map(chatItemFingerprint)); + }); + + it('detects changed inline image data without keeping it', () => { + const image = (id: string, data: string): ImageContent => ({ + id, + type: 'image_content', + image: data, + inferenceDetail: 'auto', + _cache: {}, + }); + const message = (content: ImageContent) => + new ChatMessage({ role: 'user', content: [content] }); + const original = image('img', 'data:image/png;base64,AA'); + const fingerprint = chatItemFingerprint(message(original)); + + // the same id and length with different data + expect(chatItemFingerprint(message(image('img', 'data:image/png;base64,BB')))).not.toBe( + fingerprint, + ); + // the same data in another object + expect(chatItemFingerprint(message(image('img', 'data:image/png;base64,AA')))).toBe( + fingerprint, + ); + // the data edited in place + original.image = 'data:image/png;base64,CC'; + expect(chatItemFingerprint(message(original))).not.toBe(fingerprint); + expect(fingerprint).not.toContain('base64'); + }); + + it('compares audio frames by identity', () => { + const audio = (frame: object): AudioContent => + ({ type: 'audio_content', frame: [frame] }) as unknown as AudioContent; + const frame = {}; + const message = (content: AudioContent) => + new ChatMessage({ id: 'm', role: 'user', content: [content] }); + + expect( + new ChatContext([message(audio(frame))]).isEquivalent( + new ChatContext([message(audio(frame))]), + ), + ).toBe(true); + expect( + new ChatContext([message(audio(frame))]).isEquivalent(new ChatContext([message(audio({}))])), + ).toBe(false); + }); + + it('fingerprints function-call extra as plain data', () => { + // an object that looks like chat content, but is not in a message's content + const call = new FunctionCall({ + callId: '1', + name: 'f', + args: '{}', + extra: { metadata: { type: 'audio_content' } }, + }); + expect(() => chatItemFingerprint(call)).not.toThrow(); + }); + + it('tracks image URLs and inference settings', () => { + const image = ( + url: string, + inferenceDetail: ImageContent['inferenceDetail'], + ): ImageContent => ({ + id: 'img', + type: 'image_content', + image: url, + inferenceDetail, + _cache: {}, + }); + const a = new ChatMessage({ role: 'user', content: [image('https://a.example/img', 'auto')] }); + const changedUrl = new ChatMessage({ + role: 'user', + content: [image('https://b.example/img', 'auto')], + }); + const changedDetail = new ChatMessage({ + role: 'user', + content: [image('https://a.example/img', 'high')], + }); + + expect(chatItemFingerprint(a)).not.toBe(chatItemFingerprint(changedUrl)); + expect(chatItemFingerprint(a)).not.toBe(chatItemFingerprint(changedDetail)); + }); +}); diff --git a/agents/src/llm/chat_context.ts b/agents/src/llm/chat_context.ts index e31a26dd4a..d8370f6a29 100644 --- a/agents/src/llm/chat_context.ts +++ b/agents/src/llm/chat_context.ts @@ -2,6 +2,7 @@ // // SPDX-License-Identifier: Apache-2.0 import type { AudioFrame, VideoFrame } from '@livekit/rtc-node'; +import { createHash } from 'node:crypto'; import { stripExprMarkup } from '../tts/provider_format.js'; import { createImmutableArray, shortuuid } from '../utils.js'; import type { LLM } from './llm.js'; @@ -10,6 +11,9 @@ import type { JSONObject, JSONValue, ToolContext } from './tool_context.js'; export type ChatRole = 'developer' | 'system' | 'user' | 'assistant'; export interface ImageContent { + /** + * Unique identifier for the image. + */ id: string; type: 'image_content'; @@ -740,6 +744,121 @@ export type ChatItem = | AgentHandoffItem | AgentConfigUpdate; +/** + * What identifies an item's content beyond its id and type: what {@link ChatContext.isEquivalent} + * compares and {@link chatItemFingerprint} hashes, so the two cannot drift apart. Timestamps, + * metrics and similar metadata are left out. `content` maps a message's content entries: as they + * are for equivalence, as media-safe stand-ins for the fingerprint. + */ +function essentialFields( + item: ChatItem, + content: (entry: ChatContent) => unknown = (entry) => entry, +): unknown[] { + switch (item.type) { + case 'message': + return [item.role, item.interrupted, item.content.map(content)]; + case 'function_call': + return [item.name, item.callId, item.args, item.thoughtSignature, item.groupId, item.extra]; + case 'function_call_output': + return [item.name, item.callId, item.output, item.isError]; + case 'agent_handoff': + case 'agent_config_update': + return []; + } +} + +function contentEqual(a: ChatContent, b: ChatContent): boolean { + if (typeof a === 'string' || typeof b === 'string') return a === b; + if (isInstructions(a) || isInstructions(b)) { + return ( + isInstructions(a) && + isInstructions(b) && + a.audio === b.audio && + a.text === b.text && + a.value === b.value + ); + } + if (a.type === 'image_content' && b.type === 'image_content') { + return ( + a.id === b.id && + a.image === b.image && + a.inferenceDetail === b.inferenceDetail && + a.inferenceWidth === b.inferenceWidth && + a.inferenceHeight === b.inferenceHeight && + a.mimeType === b.mimeType + ); + } + if (a.type === 'audio_content' && b.type === 'audio_content') { + return ( + a.frame.length === b.frame.length && + a.frame.every((frame, index) => frame === b.frame[index]) && + a.transcript === b.transcript + ); + } + return false; +} + +function essentialFieldEqual(a: unknown, b: unknown): boolean { + if (Array.isArray(a) && Array.isArray(b)) { + return ( + a.length === b.length && + a.every((value, index) => contentEqual(value as ChatContent, b[index] as ChatContent)) + ); + } + if (typeof a === 'object' && a !== null && typeof b === 'object' && b !== null) { + return JSON.stringify(a) === JSON.stringify(b); + } + return a === b; +} + +function digest(text: string): string { + return createHash('blake2b512').update(text).digest('hex').slice(0, 16); +} + +// media payloads are never kept: a URL counts as itself, inline data by a digest computed once per +// image object (and again when its data changes), and a video frame by its identity +const imageSources = new WeakMap(); +const frameTokens = new WeakMap(); +let nextFrameToken = 0; + +function imageSource(content: ImageContent): string | number { + const image = content.image; + if (typeof image === 'string' && !image.startsWith('data:')) return image; + const cached = imageSources.get(content); + if (cached?.image === image) return cached.source; + let source: string | number; + if (typeof image === 'string') { + source = digest(image); + } else { + source = frameTokens.get(image) ?? nextFrameToken++; + frameTokens.set(image, source); + } + imageSources.set(content, { image, source }); + return source; +} + +function contentStandIn(entry: ChatContent): unknown { + if (typeof entry === 'string') return entry; + if (isInstructions(entry)) return ['instructions', entry.audio, entry.text, entry.value]; + if (entry.type === 'image_content') { + return [ + 'image', + entry.id, + imageSource(entry), + entry.inferenceWidth, + entry.inferenceHeight, + entry.inferenceDetail, + entry.mimeType, + ]; + } + return ['audio', entry.frame.length, entry.transcript]; +} + +/** @internal A short digest of an item's essential fields, to tell whether it changed. */ +export function chatItemFingerprint(item: ChatItem): string { + return digest(JSON.stringify([item.type, ...essentialFields(item, contentStandIn)])); +} + export class ChatContext { protected _items: ChatItem[]; @@ -1047,10 +1166,11 @@ export class ChatContext { * * Comparison rules: * - Messages: compares the full `content` list, `role` and `interrupted`. - * - Function calls: compares `name`, `callId`, and `args`. + * - Function calls: compares `name`, `callId`, `args`, `thoughtSignature`, `groupId` and + * `extra`. * - Function call outputs: compares `name`, `callId`, `output`, and `isError`. * - * Does not consider timestamps or other metadata. + * Does not consider timestamps or other metadata. Item fingerprints hash the same fields. */ isEquivalent(other: ChatContext): boolean { if (this === other) { @@ -1068,102 +1188,11 @@ export class ChatContext { if (a.id !== b.id || a.type !== b.type) { return false; } - - if (a.type === 'message' && b.type === 'message') { - if ( - a.role !== b.role || - a.interrupted !== b.interrupted || - !this.compareContent(a.content, b.content) - ) { - return false; - } - } else if (a.type === 'function_call' && b.type === 'function_call') { - if ( - a.name !== b.name || - a.callId !== b.callId || - a.args !== b.args || - a.thoughtSignature !== b.thoughtSignature || - a.groupId !== b.groupId || - JSON.stringify(a.extra) !== JSON.stringify(b.extra) - ) { - return false; - } - } else if (a.type === 'function_call_output' && b.type === 'function_call_output') { - if ( - a.name !== b.name || - a.callId !== b.callId || - a.output !== b.output || - a.isError !== b.isError - ) { - return false; - } - } - } - - return true; - } - - /** - * Compare two content arrays for equality. - */ - private compareContent(a: ChatContent[], b: ChatContent[]): boolean { - if (a.length !== b.length) { - return false; - } - - for (let i = 0; i < a.length; i++) { - const contentA = a[i]!; - const contentB = b[i]!; - - if (typeof contentA === 'string' && typeof contentB === 'string') { - if (contentA !== contentB) { - return false; - } - continue; - } - - if (typeof contentA !== typeof contentB) { - return false; - } - - if (isInstructions(contentA) && isInstructions(contentB)) { - if ( - contentA.audio !== contentB.audio || - contentA.text !== contentB.text || - contentA.value !== contentB.value - ) { - return false; - } - continue; - } - - if (isInstructions(contentA) || isInstructions(contentB)) { + const left = essentialFields(a); + const right = essentialFields(b); + if (!left.every((value, index) => essentialFieldEqual(value, right[index]))) { return false; } - - if (typeof contentA === 'object' && typeof contentB === 'object') { - if (contentA.type === 'image_content' && contentB.type === 'image_content') { - if ( - contentA.id !== contentB.id || - contentA.image !== contentB.image || - contentA.inferenceDetail !== contentB.inferenceDetail || - contentA.inferenceWidth !== contentB.inferenceWidth || - contentA.inferenceHeight !== contentB.inferenceHeight || - contentA.mimeType !== contentB.mimeType - ) { - return false; - } - } else if (contentA.type === 'audio_content' && contentB.type === 'audio_content') { - if (contentA.frame.length !== contentB.frame.length) { - return false; - } - if (contentA.transcript !== contentB.transcript) { - return false; - } - } else { - return false; - } - } } return true; diff --git a/agents/src/llm/llm.ts b/agents/src/llm/llm.ts index 9d8f91a134..ff3ea16c40 100644 --- a/agents/src/llm/llm.ts +++ b/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 { genAI, recordException, traceTypes, tracer } from '../telemetry/index.js'; +import { genAI, inputDelta, 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'; @@ -283,11 +283,21 @@ export abstract class LLMStream implements AsyncIterableIterator { stream: true, outputType: traceTypes.GenAIOutputType.TEXT, }); + // no input is recorded, so this span must not become the parent of a later delta either + if (!genAI.captureContentEnabled() || !span.isRecording()) return; + if (this.genAIOperationName === undefined && inputDelta.active()) { + genAI.setContentAttributes(span, { + toolDefinitions: this.#toolCtx ? genAI.toToolDefinitions(this.#toolCtx.functionTools) : [], + }); + return; + } + const delta = inputDelta.compute(inputDelta.LLM_REQUEST, this.#chatCtx, span); genAI.setContentAttributes(span, { - systemInstructions: genAI.toSystemInstructions(this.#chatCtx), - inputMessages: genAI.toInputMessages(this.#chatCtx), + systemInstructions: delta.systemInstructions(), + inputMessages: delta.inputMessages(), toolDefinitions: this.#toolCtx ? genAI.toToolDefinitions(this.#toolCtx.functionTools) : [], }); + inputDelta.setAttributes(span, delta); } private _mainTaskImpl = async (span: Span) => { diff --git a/agents/src/telemetry/gen_ai.test.ts b/agents/src/telemetry/gen_ai.test.ts index 40d8840cc7..c7b0cd1076 100644 --- a/agents/src/telemetry/gen_ai.test.ts +++ b/agents/src/telemetry/gen_ai.test.ts @@ -9,8 +9,15 @@ import { } from '@opentelemetry/sdk-trace-node'; import { afterEach, describe, expect, it } from 'vitest'; import { z } from 'zod'; -import { ChatContext, FunctionCall, FunctionCallOutput } from '../llm/chat_context.js'; +import { + AgentConfigUpdate, + ChatContext, + ChatMessage, + FunctionCall, + FunctionCallOutput, +} from '../llm/chat_context.js'; import { tool } from '../llm/tool_context.js'; +import { INSTRUCTIONS_MESSAGE_ID } from '../voice/generation.js'; import * as genAI from './gen_ai.js'; import * as traceTypes from './trace_types.js'; @@ -97,6 +104,52 @@ describe('gen_ai builders', () => { // a tool call emitted before the generation failed is not a successful handoff expect(genAI.finishReasonFor({ functionCalls: [{}], interrupted: true })).toBe('error'); }); + + it('uses only the canonical instructions message and keeps later instructions in place', () => { + const ctx = new ChatContext([ + // the id voice/generation.ts gives it, which gen_ai.ts repeats + new ChatMessage({ id: INSTRUCTIONS_MESSAGE_ID, role: 'developer', content: 'be brief' }), + new ChatMessage({ id: 'u1', role: 'user', content: 'hi' }), + new ChatMessage({ id: 'x', role: 'developer', content: 'greet' }), + new ChatMessage({ id: 'G', role: 'system', content: 'guide' }), + ]); + + expect(genAI.toSystemInstructions(ctx)).toEqual([{ type: 'text', content: 'be brief' }]); + expect(genAI.toInputMessages(ctx).map((message) => message.role)).toEqual([ + 'user', + 'system', + 'system', + ]); + + const standalone = new ChatContext([ + new ChatMessage({ id: 's', role: 'developer', content: 'you are a bot' }), + new ChatMessage({ id: 'G', role: 'system', content: 'guide' }), + new ChatMessage({ id: 'u1', role: 'user', content: 'hi' }), + ]); + expect(genAI.toSystemInstructions(standalone)).toEqual([ + { type: 'text', content: 'you are a bot' }, + ]); + expect(genAI.toInputMessages(standalone).map((message) => message.role)).toEqual([ + 'system', + 'user', + ]); + }); + + it('keeps tool calls merged across skipped items without resetting the previous role', () => { + const items = [ + new ChatMessage({ id: 'a1', role: 'assistant', content: 'checking' }), + new AgentConfigUpdate({ id: 'cfg', toolsAdded: ['f'] }), + new FunctionCall({ id: 'fc1', callId: 'c1', name: 'f', args: '{}' }), + new AgentConfigUpdate({ id: 'cfg2', toolsRemoved: ['g'] }), + new FunctionCall({ id: 'fc2', callId: 'c2', name: 'f', args: '{}' }), + ]; + + expect(genAI.messageLayout(items)).toEqual(['new', 'skipped', 'merged', 'skipped', 'merged']); + const messages = genAI.conversationMessages(items); + expect(messages).toHaveLength(1); + expect(messages[0]!.role).toBe('assistant'); + expect(messages[0]!.parts.map((part) => part.type)).toEqual(['text', 'tool_call', 'tool_call']); + }); }); describe('gen_ai span attributes', () => { diff --git a/agents/src/telemetry/gen_ai.ts b/agents/src/telemetry/gen_ai.ts index 0a78f40b76..7c9cd7c42b 100644 --- a/agents/src/telemetry/gen_ai.ts +++ b/agents/src/telemetry/gen_ai.ts @@ -33,6 +33,9 @@ import * as traceTypes from './trace_types.js'; // matched on their own `type` discriminants instead. const FALSY = new Set(['0', 'false', 'no', 'off']); +// the id `voice/generation.ts` gives the agent's instructions message; telemetry cannot import +// from voice (voice imports telemetry), and a test keeps the two equal +const INSTRUCTIONS_MESSAGE_ID = 'lk.agent_task.instructions'; // the env var name the GenAI conventions standardise for this opt-in let captureContent = !FALSY.has( @@ -51,6 +54,10 @@ export function setCaptureContent(enabled: boolean): void { captureContent = enabled; } +export function captureContentEnabled(): boolean { + return captureContent; +} + // --------------------------------------------------------------------------- // message models // --------------------------------------------------------------------------- @@ -175,6 +182,22 @@ function itemsOf(chatCtx: ChatContext | undefined): readonly ChatItem[] { return Array.isArray(items) ? items : []; } +function isSystemMessage(item: ChatItem): boolean { + return item.type === 'message' && (item.role === 'system' || item.role === 'developer'); +} + +/** @internal */ +export function splitInstructions(items: readonly ChatItem[]): [ChatItem[], ChatItem[]] { + const index = items.findIndex( + (item) => item.id === INSTRUCTIONS_MESSAGE_ID && isSystemMessage(item), + ); + if (index !== -1) { + return [[items[index]!], [...items.slice(0, index), ...items.slice(index + 1)]]; + } + if (items[0] && isSystemMessage(items[0])) return [[items[0]], [...items.slice(1)]]; + return [[], [...items]]; +} + /** * `gen_ai.system_instructions` — the agent instructions, as text parts. * @@ -183,57 +206,74 @@ function itemsOf(chatCtx: ChatContext | undefined): readonly ChatItem[] { * so they are reported as instructions rather than history. */ export function toSystemInstructions(chatCtx: ChatContext): MessagePart[] { - const parts: MessagePart[] = []; - for (const item of itemsOf(chatCtx)) { - if (item.type === 'message' && (item.role === 'system' || item.role === 'developer')) { - const text = item.rawTextContent; - if (text !== undefined) parts.push(textPart(text)); - } - } - return parts; + return instructionParts(splitInstructions(itemsOf(chatCtx))[0]); } /** * `gen_ai.input.messages` — the conversation history, in the order it was sent. * - * `system`/`developer` messages are reported in `gen_ai.system_instructions` instead, and - * non-conversational items (agent handoffs, config updates) are skipped. + * The canonical instructions message is reported in `gen_ai.system_instructions`; later + * `system`/`developer` messages remain in place. Non-conversational items are skipped. */ export function toInputMessages(chatCtx: ChatContext): ChatMessagePayload[] { - const messages: ChatMessagePayload[] = []; - for (const item of itemsOf(chatCtx)) { - let role: string; + return conversationMessages(splitInstructions(itemsOf(chatCtx))[1]); +} + +/** @internal */ +export function instructionParts(items: readonly ChatItem[]): MessagePart[] { + const parts: MessagePart[] = []; + for (const item of items) { if (item.type === 'message') { - if (item.role === 'system' || item.role === 'developer') continue; - role = item.role; - } else if (item.type === 'function_call') { - role = 'assistant'; - } else if (item.type === 'function_call_output') { - role = 'tool'; - } else { - continue; + const text = item.rawTextContent; + if (text !== undefined) parts.push(textPart(text)); } + } + return parts; +} + +function messageRole(item: ChatItem): string | undefined { + if (item.type === 'message') return item.role === 'developer' ? 'system' : item.role; + if (item.type === 'function_call') return 'assistant'; + if (item.type === 'function_call_output') return 'tool'; + return undefined; +} +/** @internal */ +export function conversationMessages(items: readonly ChatItem[]): ChatMessagePayload[] { + const messages: ChatMessagePayload[] = []; + const layout = messageLayout(items); + for (let index = 0; index < items.length; index++) { + const item = items[index]!; + const placement = layout[index]!; + if (placement === 'skipped') continue; const parts = messageParts(item); - if (!parts.length) continue; - - // consecutive tool calls from one assistant turn belong to a single message - const last = messages[messages.length - 1]; - if ( - last && - last.role === 'assistant' && - role === 'assistant' && - item.type === 'function_call' - ) { - last.parts.push(...parts); - continue; + if (placement === 'merged') { + messages[messages.length - 1]!.parts.push(...parts); + } else { + messages.push({ role: messageRole(item)!, parts }); } - - messages.push({ role, parts }); } return messages; } +/** @internal */ +export function messageLayout(items: readonly ChatItem[]): Array<'new' | 'merged' | 'skipped'> { + const layout: Array<'new' | 'merged' | 'skipped'> = []; + let lastRole: string | undefined; + for (const item of items) { + const role = messageRole(item); + if (role === undefined || messageParts(item).length === 0) { + layout.push('skipped'); + } else if (lastRole === 'assistant' && role === 'assistant' && item.type === 'function_call') { + layout.push('merged'); + } else { + layout.push('new'); + lastRole = role; + } + } + return layout; +} + export interface ToolCallLike { callId: string; name: string; diff --git a/agents/src/telemetry/index.ts b/agents/src/telemetry/index.ts index b1bca84425..3302d55373 100644 --- a/agents/src/telemetry/index.ts +++ b/agents/src/telemetry/index.ts @@ -24,6 +24,7 @@ export { type PinoLogObject, } from './pino_otel_transport.js'; export * as genAI from './gen_ai.js'; +export * as inputDelta from './input_delta.js'; export { REDACTED_EXCEPTION_MESSAGE } from './redaction.js'; export * as loopMonitor from './loop_monitor.js'; export * as rpc from './rpc.js'; diff --git a/agents/src/telemetry/input_delta.test.ts b/agents/src/telemetry/input_delta.test.ts new file mode 100644 index 0000000000..ab5dfdb4af --- /dev/null +++ b/agents/src/telemetry/input_delta.test.ts @@ -0,0 +1,680 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import type { Span } from '@opentelemetry/api'; +import { + BasicTracerProvider, + InMemorySpanExporter, + type ReadableSpan, + SimpleSpanProcessor, +} from '@opentelemetry/sdk-trace-base'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { + AgentConfigUpdate, + ChatContext, + type ChatItem, + ChatMessage, + type ChatRole, + FunctionCall, + FunctionCallOutput, + chatItemFingerprint, +} from '../llm/chat_context.js'; +import { FallbackAdapter } from '../llm/fallback_adapter.js'; +import type { LLM } from '../llm/llm.js'; +import { RUNNING_TOOL_PLACEHOLDER_KEY, _injectRunningToolCalls } from '../voice/generation.js'; +import { FakeLLM } from '../voice/testing/fake_llm.js'; +import * as genAI from './gen_ai.js'; +import { + type InputDelta, + type InputDeltaScope, + type InputDeltaSite, + InputDeltaTracker, + LLM_NODE, + LLM_REQUEST, + runWithScope, + setAttributes, +} from './input_delta.js'; +import * as traceTypes from './trace_types.js'; +import { tracer as frameworkTracer, setTracerProvider } from './traces.js'; + +type MessageSpec = readonly [id: string, role: ChatRole, text: string]; +type ContextItem = MessageSpec | ChatItem; + +const INSTRUCTIONS_ID = 'lk.agent_task.instructions'; +const INSTRUCTIONS = [INSTRUCTIONS_ID, 'system', 'be brief'] as const satisfies MessageSpec; +const CHANGED_INSTRUCTIONS = [ + INSTRUCTIONS_ID, + 'system', + 'be detailed', +] as const satisfies MessageSpec; + +function call(id: string, callId: string, createdAt?: number): FunctionCall { + return new FunctionCall({ id, callId, name: 'f', args: '{}', createdAt }); +} + +function output(id: string, callId: string, createdAt?: number): FunctionCallOutput { + return new FunctionCallOutput({ + id, + callId, + name: 'f', + output: 'ok', + isError: false, + createdAt, + }); +} + +function context(...items: ContextItem[]): ChatContext { + return new ChatContext( + items.map((item) => + Array.isArray(item) + ? new ChatMessage({ id: item[0], role: item[1], content: item[2] }) + : (item as ChatItem), + ), + ); +} + +function ids(items: readonly ChatItem[]): string[] { + return items.map((item) => item.id); +} + +describe('input delta tracking', () => { + let exporter: InMemorySpanExporter; + let provider: BasicTracerProvider; + let tracer: ReturnType; + let originalProvider: ReturnType; + + beforeEach(() => { + originalProvider = frameworkTracer.getProvider(); + exporter = new InMemorySpanExporter(); + provider = new BasicTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)] }); + tracer = provider.getTracer('input-delta-test'); + setTracerProvider(provider); + }); + + afterEach(async () => { + setTracerProvider(originalProvider); + await provider.shutdown(); + }); + + function delta( + scope: InputDeltaScope, + chatCtx: ChatContext, + site: InputDeltaSite = LLM_REQUEST, + ): [InputDelta, Span] { + const span = tracer.startSpan(site.name); + return [scope.delta(site, chatCtx, span), span]; + } + + function committed(tracker: InputDeltaTracker, chatCtx: ChatContext): Map { + const scope = tracker.begin(); + const spans = new Map(); + for (const site of [LLM_NODE, LLM_REQUEST]) { + const [, span] = delta(scope, chatCtx, site); + spans.set(site.name, span); + span.end(); + } + scope.commit(); + return spans; + } + + async function request(llm: LLM, chatCtx: ChatContext, tracker?: InputDeltaTracker) { + if (!tracker) { + await llm.chat({ chatCtx }).collect(); + return; + } + + const scope = tracker.begin(); + await runWithScope(scope, () => llm.chat({ chatCtx }).collect()); + scope.commit(); + } + + function requestSpans(): ReadableSpan[] { + return exporter.getFinishedSpans().filter((span) => span.name === LLM_REQUEST.name); + } + + function inputRoles(span: ReadableSpan): string[] { + const messages = JSON.parse( + String(span.attributes[traceTypes.ATTR_GEN_AI_INPUT_MESSAGES]), + ) as Array<{ role: string }>; + return messages.map((message) => message.role); + } + + it('records the first generation in full at both sites', () => { + const tracker = new InputDeltaTracker(); + const chatCtx = context(INSTRUCTIONS, ['u1', 'user', 'hi']); + + for (const site of [LLM_NODE, LLM_REQUEST]) { + const [record, span] = delta(tracker.begin(), chatCtx, site); + expect(record.base).toBeUndefined(); + expect(ids(record.chatCtx.items)).toEqual([INSTRUCTIONS_ID, 'u1']); + expect(ids(record.instructions)).toEqual([INSTRUCTIONS_ID]); + span.end(); + } + }); + + it('continues an appended turn from each site parent and records span linkage', () => { + const tracker = new InputDeltaTracker(); + const parent = committed(tracker, context(INSTRUCTIONS, ['u1', 'user', 'hi'])); + const chatCtx = context( + INSTRUCTIONS, + ['u1', 'user', 'hi'], + ['a1', 'assistant', 'hello'], + ['u2', 'user', 'bye'], + ); + const scope = tracker.begin(); + + const [node, nodeSpan] = delta(scope, chatCtx, LLM_NODE); + expect(node.base).toEqual(parent.get(LLM_NODE.name)!.spanContext()); + expect(node.droppedFromBase).toBe(0); + expect(ids(node.chatCtx.items)).toEqual(['a1', 'u2']); + nodeSpan.end(); + + const [request, requestSpan] = delta(scope, chatCtx, LLM_REQUEST); + expect(request.base).toEqual(parent.get(LLM_REQUEST.name)!.spanContext()); + expect(request.droppedFromBase).toBe(0); + expect(request.instructions).toEqual([]); + expect(ids(request.conversation)).toEqual(['a1', 'u2']); + setAttributes(requestSpan, request); + requestSpan.end(); + + const exported = exporter + .getFinishedSpans() + .find((span) => span.spanContext().spanId === requestSpan.spanContext().spanId)!; + expect(exported.attributes).toMatchObject({ + [traceTypes.ATTR_INPUT_DELTA]: true, + [traceTypes.ATTR_INPUT_BASE_SPAN_ID]: parent.get(LLM_REQUEST.name)!.spanContext().spanId, + [traceTypes.ATTR_INPUT_DROPPED_FROM_BASE]: 0, + }); + expect(exported.links.map((link) => link.context.spanId)).toEqual([ + parent.get(LLM_REQUEST.name)!.spanContext().spanId, + ]); + }); + + it('records changed instructions separately for requests but restarts node input', () => { + const tracker = new InputDeltaTracker(); + committed(tracker, context(INSTRUCTIONS, ['u1', 'user', 'hi'])); + const chatCtx = context(CHANGED_INSTRUCTIONS, ['u1', 'user', 'hi'], ['u2', 'user', 'bye']); + const scope = tracker.begin(); + + const [request, requestSpan] = delta(scope, chatCtx, LLM_REQUEST); + expect(request.base).toBeDefined(); + expect(request.droppedFromBase).toBe(0); + expect(ids(request.instructions)).toEqual([INSTRUCTIONS_ID]); + expect(ids(request.conversation)).toEqual(['u2']); + requestSpan.end(); + + const [node, nodeSpan] = delta(scope, chatCtx, LLM_NODE); + expect(node.base).toBeUndefined(); + expect(ids(node.chatCtx.items)).toEqual([INSTRUCTIONS_ID, 'u1', 'u2']); + nodeSpan.end(); + }); + + it('records from an edited item and drops the old suffix', () => { + const tracker = new InputDeltaTracker(); + committed( + tracker, + context(INSTRUCTIONS, ['u1', 'user', 'hi'], ['a1', 'assistant', 'hey'], ['u2', 'user', 'x']), + ); + const chatCtx = context( + INSTRUCTIONS, + ['u1', 'user', 'hi'], + ['a1', 'assistant', 'HEY'], + ['u2', 'user', 'x'], + ); + + const [request, span] = delta(tracker.begin(), chatCtx); + expect(request.droppedFromBase).toBe(2); + expect(ids(request.conversation)).toEqual(['a1', 'u2']); + span.end(); + }); + + it('never splits a tool call from its assistant message', () => { + const tracker = new InputDeltaTracker(); + committed( + tracker, + context(INSTRUCTIONS, ['u1', 'user', 'hi'], ['a1', 'assistant', 'checking']), + ); + const chatCtx = context( + INSTRUCTIONS, + ['u1', 'user', 'hi'], + ['a1', 'assistant', 'checking'], + call('fc', 'c1'), + output('fo', 'c1'), + ); + + const [request, span] = delta(tracker.begin(), chatCtx); + expect(request.droppedFromBase).toBe(1); + expect(ids(request.conversation)).toEqual(['a1', 'fc', 'fo']); + expect(request.inputMessages().map((message) => message.role)).toEqual(['assistant', 'tool']); + span.end(); + }); + + it('keeps a tool call with its message across a skipped AgentConfigUpdate', () => { + const tracker = new InputDeltaTracker(); + const config = new AgentConfigUpdate({ id: 'cfg', toolsAdded: ['f'] }); + const parentCtx = context( + INSTRUCTIONS, + ['u1', 'user', 'hi'], + ['a1', 'assistant', 'checking'], + config, + ); + const parent = committed(tracker, parentCtx); + const chatCtx = context(...parentCtx.items, call('fc', 'c1'), output('fo', 'c1')); + + const [request, span] = delta(tracker.begin(), chatCtx); + expect(request.base).toEqual(parent.get(LLM_REQUEST.name)!.spanContext()); + expect(request.droppedFromBase).toBe(1); + expect(ids(request.conversation)).toEqual(['a1', 'cfg', 'fc', 'fo']); + + const parentMessages = genAI.toInputMessages(parentCtx); + const rebuilt = [ + ...parentMessages.slice(0, parentMessages.length - 1), + ...request.inputMessages(), + ]; + expect(rebuilt).toEqual(genAI.toInputMessages(chatCtx)); + span.end(); + }); + + it('shares stable running placeholders and replaces them when the tool ends', () => { + const message = (id: string, role: ChatRole, createdAt: number) => + new ChatMessage({ id, role, content: id, createdAt }); + const running = call('rc', 'c1', 2.5); + const history: ChatItem[] = [ + message(INSTRUCTIONS_ID, 'system', 0), + message('u1', 'user', 2), + message('a1', 'assistant', 3), + ]; + const turn = (items: ChatItem[], stillRunning: boolean): ChatContext => { + const chatCtx = new ChatContext([...history, ...items]); + if (stillRunning) _injectRunningToolCalls(chatCtx, [running]); + return chatCtx; + }; + const tracker = new InputDeltaTracker(); + committed(tracker, turn([message('u2', 'user', 4)], true)); + + const chatCtx = turn( + [message('u2', 'user', 4), message('a2', 'assistant', 5), message('u3', 'user', 6)], + true, + ); + const scope = tracker.begin(); + const [stable, stableSpan] = delta(scope, chatCtx, LLM_NODE); + scope.commit(); + expect(stable.droppedFromBase).toBe(0); + expect(ids(stable.chatCtx.items)).toEqual(['a2', 'u3']); + stableSpan.end(); + + const finished = new FunctionCallOutput({ + id: 'ro', + callId: 'c1', + name: 'f', + output: 'done', + isError: false, + createdAt: 2.6, + }); + const finishedCtx = turn( + [ + running, + finished, + message('u2', 'user', 4), + message('a2', 'assistant', 5), + message('u3', 'user', 6), + message('u4', 'user', 7), + ], + false, + ); + finishedCtx.items.sort((left, right) => left.createdAt - right.createdAt); + const [replaced, replacedSpan] = delta(tracker.begin(), finishedCtx, LLM_NODE); + expect(replaced.chatCtx.items[0]!.id).toBe('rc'); + expect( + replaced.chatCtx.items.some( + (item) => + item.type === 'function_call' && item.extra[RUNNING_TOOL_PLACEHOLDER_KEY] === true, + ), + ).toBe(false); + replacedSpan.end(); + }); + + it('does not move the parent for an uncommitted generation', () => { + const tracker = new InputDeltaTracker(); + const parent = committed(tracker, context(INSTRUCTIONS, ['u1', 'user', 'hi'])); + const [, discardedSpan] = delta( + tracker.begin(), + context(INSTRUCTIONS, ['u1', 'user', 'hi'], ['p2', 'user', 'by']), + ); + discardedSpan.end(); + + const [request, span] = delta( + tracker.begin(), + context(INSTRUCTIONS, ['u1', 'user', 'hi'], ['u2', 'user', 'bye']), + ); + expect(request.base).toEqual(parent.get(LLM_REQUEST.name)!.spanContext()); + expect(ids(request.conversation)).toEqual(['u2']); + span.end(); + }); + + it('promotes a span created after its scope was committed', () => { + const tracker = new InputDeltaTracker(); + const scope = tracker.begin(); + scope.commit(); + const [, firstSpan] = delta(scope, context(INSTRUCTIONS, ['u1', 'user', 'hi'])); + firstSpan.end(); + + const [request, span] = delta( + tracker.begin(), + context(INSTRUCTIONS, ['u1', 'user', 'hi'], ['u2', 'user', 'x']), + ); + expect(request.base).toEqual(firstSpan.spanContext()); + span.end(); + }); + + it('keeps tracker baselines independent', () => { + const first = new InputDeltaTracker(); + const second = new InputDeltaTracker(); + committed(first, context(INSTRUCTIONS, ['u1', 'user', 'hi'])); + + const [request, span] = delta( + second.begin(), + context(INSTRUCTIONS, ['u1', 'user', 'hi'], ['u2', 'user', 'x']), + ); + expect(request.base).toBeUndefined(); + span.end(); + }); + + it('executes the expressive-mode incremental example', () => { + const u1 = ['u1', 'user', 'hi'] as const satisfies MessageSpec; + const a1 = ['a1', 'assistant', 'hello'] as const satisfies MessageSpec; + const u2 = ['u2', 'user', 'x'] as const satisfies MessageSpec; + const a2 = ['a2', 'assistant', 'sure'] as const satisfies MessageSpec; + const u3 = ['u3', 'user', 'weather?'] as const satisfies MessageSpec; + const guide = ['G', 'system', 'markup guide'] as const satisfies MessageSpec; + const tool: ChatItem[] = [call('fc', 'c1'), output('fo', 'c1')]; + const turns = [ + context(INSTRUCTIONS, u1, guide), + context(INSTRUCTIONS, u1, a1, u2, guide), + context(INSTRUCTIONS, u1, a1, u2, a2, u3, guide), + context(INSTRUCTIONS, u1, a1, u2, a2, u3, guide, ...tool), + ]; + const expected: Array<[string[], number | undefined]> = [ + [[INSTRUCTIONS_ID, 'u1', 'G'], undefined], + [['a1', 'u2', 'G'], 1], + [['a2', 'u3', 'G'], 1], + [['fc', 'fo'], 0], + ]; + const tracker = new InputDeltaTracker(); + + turns.forEach((chatCtx, index) => { + const scope = tracker.begin(); + const [record, span] = delta(scope, chatCtx, LLM_NODE); + scope.commit(); + expect(ids(record.chatCtx.items)).toEqual(expected[index]![0]); + expect(record.droppedFromBase).toBe(expected[index]![1]); + span.end(); + }); + }); + + it('recursively reconstructs node chat items, GenAI messages, and inherited instructions', () => { + const turn = (...items: ContextItem[]): ChatContext => { + const chatCtx = context(...items, ['G', 'system', 'markup guide']); + const configIndex = chatCtx.items[0]?.id === INSTRUCTIONS_ID ? 1 : 0; + chatCtx.items.splice( + configIndex, + 0, + new AgentConfigUpdate({ id: 'cfg', instructions: 'initial' }), + ); + return chatCtx; + }; + const a1 = ['a1', 'assistant', 'hello'] as const satisfies MessageSpec; + const u1 = ['u1', 'user', 'hi'] as const satisfies MessageSpec; + const a2 = ['a2', 'assistant', 'sure'] as const satisfies MessageSpec; + const u2 = ['u2', 'user', 'weather?'] as const satisfies MessageSpec; + const a3 = ['a3', 'assistant', 'sunny'] as const satisfies MessageSpec; + const u3 = ['u3', 'user', 'tomorrow?'] as const satisfies MessageSpec; + const a4 = ['a4', 'assistant', 'rain'] as const satisfies MessageSpec; + const u4 = ['u4', 'user', 'thanks'] as const satisfies MessageSpec; + const persistent = ['S', 'system', 'doctor has slots at 3pm'] as const satisfies MessageSpec; + const tool: ChatItem[] = [call('fc', 'c1'), output('fo', 'c1')]; + const toolTurn = turn(INSTRUCTIONS, a1, u1, a2, u2); + const turns = [ + turn(INSTRUCTIONS, ['x', 'system', 'greet the user']), + turn(INSTRUCTIONS, a1, u1), + turn(INSTRUCTIONS, a1, u1, a2, u2), + context(...toolTurn.items, ...tool), + turn(INSTRUCTIONS, a1, u1, a2, u2, ...tool, persistent, a3, u3, [ + 'R', + 'assistant', + 'rag: docs', + ]), + turn(INSTRUCTIONS, a1, u1, a2, u2, ...tool, persistent, a3, u3, a4, u4, [ + 'R2', + 'assistant', + 'rag: more', + ]), + turn(CHANGED_INSTRUCTIONS, a1, u1, a2, u2, ...tool, persistent, a3, u3, a4, u4, [ + 'a5', + 'assistant', + 'ok', + ]), + turn( + CHANGED_INSTRUCTIONS, + a1, + ['u1', 'user', 'hey'], + a2, + u2, + ...tool, + persistent, + a3, + u3, + a4, + u4, + ), + context(['sp', 'system', 'you are a bot'], ['G', 'system', 'guide'], u1), + ]; + const tracker = new InputDeltaTracker(); + const recorded = new Map(); + const shape: Record> = { + [LLM_NODE.name]: [], + [LLM_REQUEST.name]: [], + }; + + const parent = (record: InputDelta): InputDelta => { + expect(record.base).toBeDefined(); + return recorded.get(record.base!.spanId)!; + }; + const kept = (entries: T[], dropped: number): T[] => + entries.slice(0, entries.length - dropped); + const chatItems = (record: InputDelta): ChatItem[] => + record.droppedFromBase === undefined + ? [...record.chatCtx.items] + : [...kept(chatItems(parent(record)), record.droppedFromBase), ...record.chatCtx.items]; + const messages = (record: InputDelta): genAI.ChatMessagePayload[] => + record.droppedFromBase === undefined + ? record.inputMessages() + : [...kept(messages(parent(record)), record.droppedFromBase), ...record.inputMessages()]; + const instructions = (record: InputDelta): genAI.MessagePart[] => + record.instructions.length > 0 || !record.base + ? record.systemInstructions() + : instructions(parent(record)); + + for (const chatCtx of turns) { + const scope = tracker.begin(); + const latest = new Map(); + for (const site of [LLM_NODE, LLM_REQUEST]) { + const [record, span] = delta(scope, chatCtx, site); + recorded.set(span.spanContext().spanId, record); + latest.set(site.name, record); + shape[site.name]!.push([record.base !== undefined, record.instructions.length > 0]); + span.end(); + } + scope.commit(); + + expect( + chatItems(latest.get(LLM_NODE.name)!).map((item) => [item.id, chatItemFingerprint(item)]), + ).toEqual(chatCtx.items.map((item) => [item.id, chatItemFingerprint(item)])); + expect(messages(latest.get(LLM_REQUEST.name)!)).toEqual(genAI.toInputMessages(chatCtx)); + expect(instructions(latest.get(LLM_REQUEST.name)!)).toEqual( + genAI.toSystemInstructions(chatCtx), + ); + expect(instructions(latest.get(LLM_REQUEST.name)!)).toHaveLength(1); + } + + expect(shape[LLM_REQUEST.name]).toEqual([ + [false, true], + [true, false], + [true, false], + [true, false], + [true, false], + [true, false], + [true, true], + [true, false], + [false, true], + ]); + expect(shape[LLM_NODE.name]!.map(([hasBase]) => hasBase)).toEqual([ + false, + true, + true, + true, + true, + true, + false, + true, + false, + ]); + }); + + it('records successive LLMStream requests as a full input followed by deltas', async () => { + const tracker = new InputDeltaTracker(); + const llm = new FakeLLM(); + await request(llm, context(INSTRUCTIONS, ['u1', 'user', 'Hello']), tracker); + await request( + llm, + context( + INSTRUCTIONS, + ['u1', 'user', 'Hello'], + ['a1', 'assistant', 'Hi there'], + ['u2', 'user', "What's the weather in Tokyo?"], + ), + tracker, + ); + await request( + llm, + context( + INSTRUCTIONS, + ['u1', 'user', 'Hello'], + ['a1', 'assistant', 'Hi there'], + ['u2', 'user', "What's the weather in Tokyo?"], + call('fc', 'c1'), + output('fo', 'c1'), + ), + tracker, + ); + + const [first, second, toolStep] = requestSpans(); + expect(requestSpans()).toHaveLength(3); + expect(first!.attributes[traceTypes.ATTR_INPUT_DELTA]).toBeUndefined(); + expect(first!.attributes[traceTypes.ATTR_GEN_AI_SYSTEM_INSTRUCTIONS]).toBeDefined(); + expect(inputRoles(first!)).toEqual(['user']); + + expect(second!.attributes).toMatchObject({ + [traceTypes.ATTR_INPUT_DELTA]: true, + [traceTypes.ATTR_INPUT_BASE_SPAN_ID]: first!.spanContext().spanId, + [traceTypes.ATTR_INPUT_DROPPED_FROM_BASE]: 0, + }); + expect(second!.attributes[traceTypes.ATTR_GEN_AI_SYSTEM_INSTRUCTIONS]).toBeUndefined(); + expect(inputRoles(second!)).toEqual(['assistant', 'user']); + expect(second!.links.map((link) => link.context.spanId)).toEqual([first!.spanContext().spanId]); + + expect(toolStep!.attributes).toMatchObject({ + [traceTypes.ATTR_INPUT_DELTA]: true, + [traceTypes.ATTR_INPUT_BASE_SPAN_ID]: second!.spanContext().spanId, + [traceTypes.ATTR_INPUT_DROPPED_FROM_BASE]: 0, + }); + expect(inputRoles(toolStep!)).toEqual(['assistant', 'tool']); + }); + + it('does not continue a request whose input was not captured', async () => { + const tracker = new InputDeltaTracker(); + const llm = new FakeLLM(); + genAI.setCaptureContent(false); + try { + await request(llm, context(INSTRUCTIONS, ['u1', 'user', 'Hello']), tracker); + } finally { + genAI.setCaptureContent(true); + } + await request( + llm, + context(INSTRUCTIONS, ['u1', 'user', 'Hello'], ['u2', 'user', 'Weather?']), + tracker, + ); + + // the first span recorded no input, so the second records all of it + const [, second] = requestSpans(); + expect(second!.attributes[traceTypes.ATTR_INPUT_DELTA]).toBeUndefined(); + expect(inputRoles(second!)).toEqual(['user', 'user']); + }); + + it('records full LLMStream request inputs when input delta is off', async () => { + const llm = new FakeLLM(); + await request(llm, context(INSTRUCTIONS, ['u1', 'user', 'Hello'])); + await request( + llm, + context( + INSTRUCTIONS, + ['u1', 'user', 'Hello'], + ['a1', 'assistant', 'Hi there'], + ['u2', 'user', 'Weather?'], + ), + ); + + const spans = requestSpans(); + expect(spans).toHaveLength(2); + for (const span of spans) { + expect(Object.keys(span.attributes).some((key) => key.startsWith('lk.input.'))).toBe(false); + expect(span.attributes[traceTypes.ATTR_GEN_AI_SYSTEM_INSTRUCTIONS]).toBeDefined(); + } + expect(inputRoles(spans[1]!)).toEqual(['user', 'assistant', 'user']); + }); + + it('records fallback input only on provider spans and chains provider deltas', async () => { + const tracker = new InputDeltaTracker(); + const llm = new FallbackAdapter({ llms: [new FakeLLM()] }); + await request(llm, context(INSTRUCTIONS, ['u1', 'user', 'Hello']), tracker); + await request( + llm, + context( + INSTRUCTIONS, + ['u1', 'user', 'Hello'], + ['a1', 'assistant', 'Hi there'], + ['u2', 'user', 'Weather?'], + ), + tracker, + ); + + const spans = requestSpans(); + const providerSpans = spans.filter( + (span) => traceTypes.ATTR_GEN_AI_INPUT_MESSAGES in span.attributes, + ); + const wrapperSpans = spans.filter( + (span) => !(traceTypes.ATTR_GEN_AI_INPUT_MESSAGES in span.attributes), + ); + expect(spans).toHaveLength(4); + expect(providerSpans).toHaveLength(2); + expect(wrapperSpans).toHaveLength(2); + expect( + providerSpans.map((span) => span.attributes[traceTypes.ATTR_GEN_AI_OPERATION_NAME]), + ).toEqual(['chat', 'chat']); + expect( + wrapperSpans.every((span) => + Object.keys(span.attributes).every((key) => !key.startsWith('lk.input.')), + ), + ).toBe(true); + expect(providerSpans[1]!.attributes[traceTypes.ATTR_INPUT_BASE_SPAN_ID]).toBe( + providerSpans[0]!.spanContext().spanId, + ); + expect(providerSpans[1]!.links.map((link) => link.context.spanId)).toEqual([ + providerSpans[0]!.spanContext().spanId, + ]); + expect(wrapperSpans.map((span) => span.spanContext().spanId)).not.toContain( + providerSpans[1]!.attributes[traceTypes.ATTR_INPUT_BASE_SPAN_ID], + ); + }); +}); diff --git a/agents/src/telemetry/input_delta.ts b/agents/src/telemetry/input_delta.ts new file mode 100644 index 0000000000..4e1f3266d5 --- /dev/null +++ b/agents/src/telemetry/input_delta.ts @@ -0,0 +1,282 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 + +/** + * Input deltas for LLM spans (`record: { inputDelta: true }`). + * + * **Purpose** + * + * Each LLM span records the input of the model. In a long session, the input of each turn + * contains all of the previous turns. Thus, the spans become larger at each turn. When the session + * records with `inputDelta`, a span records only the part of the input that changed. The model + * always receives all of the input. This module changes only the telemetry. + * + * **Terms** + * + * - Record: the list of entries that a span records. On an `llm_node` span, the record is + * `lk.pii.chat_ctx`, with one entry for each chat item. On an `llm_request` span, the record is + * `gen_ai.input.messages`, with one entry for each GenAI message. + * - Instructions: the instructions message of the agent. On an `llm_request` span, the + * instructions are in `gen_ai.system_instructions`, not in the record. All other system messages + * are entries of the record, at their position. + * - Generation: one LLM inference of the agent. A turn with tool calls has more than one + * generation. + * - Commit: a generation is committed when its speech is scheduled. A discarded preemptive + * generation is not committed. + * - Parent: the span of the last committed generation that has the same span name. + * + * **Procedure** + * + * For each span, the tracker does these steps: + * + * 1. It calculates a key for each item of the record. The key is the item ID and a fingerprint of + * the content of the item (`chatItemFingerprint`). + * 2. It compares the keys with the keys of the parent, in order. It finds the longest prefix that + * the two records share. + * 3. The span records only the entries after this prefix. + * 4. `lk.input.dropped_from_base` gives the number of parent entries after this prefix. + * 5. On an `llm_request` span, the tracker compares the instructions with the instructions of the + * parent. If the text is the same, the span does not record `gen_ai.system_instructions`. + * 6. If the span shares no entries and no instructions with the parent, the span records all of + * the input, without `lk.input.*` attributes. + * 7. The tracker keeps the keys of the span. When the generation is committed, these keys replace + * the keys of the parent. + * + * **Rebuild** + * + * To make the full record of a span again: + * + * 1. Make the full record of the parent (`lk.input.base_span_id`). Use this procedure again if the + * parent also has a parent. + * 2. Remove the last `lk.input.dropped_from_base` entries. + * 3. Add the entries of this span. + * + * If an `llm_request` span with a parent has no `gen_ai.system_instructions`, use the + * instructions of the parent. + * + * **Rules** + * + * - The comparison does not use the role of an item. A system message that stays at its position + * is part of the shared prefix. The framework adds the expressive guide again at the end of each + * reply. Thus, the old guide is in the dropped entries, and the new guide is in the recorded + * entries. + * - An edited, removed or moved item stops the shared prefix at that item. The span records the + * entries from that item. It does not record all of the input. + * - In `gen_ai.input.messages`, consecutive tool calls of an assistant turn are one message, also + * when a skipped item (for example, a config update) is between them. The prefix never stops + * inside such a message. + * - On an `llm_node` span, the instructions are the first entry of `lk.pii.chat_ctx`. Thus, a + * change of the instructions causes a full record of `lk.pii.chat_ctx`. + * - The fingerprint does not keep media data. Inline image data counts by a digest, a video frame + * by its identity, and audio by its frame count and transcript. + * - A span that does not record its input (content capture is off) does not become a parent. + * - Each `AgentActivity` has its own tracker. After a handoff, the first span records all of the + * input. + * + * **Example** + * + * In expressive mode, `G` is the expressive guide, and `I` is the instructions message. + * + * | Generation | Input sent to the model | Record of `llm_node` | Dropped | + * | ------------------ | -------------------------- | -------------------- | ------- | + * | Turn 1 | `[I, u1, G]` | `[I, u1, G]` (full) | none | + * | Turn 2 | `[I, u1, a1, u2, G]` | `[a1, u2, G]` | 1 | + * | Turn 3, tool call | `[I, …, u2, a2, u3, G]` | `[a2, u3, G]` | 1 | + * | Turn 3, tool reply | `[I, …, u3, G, fc, fo]` | `[fc, fo]` | 0 | + */ +import type { Span, SpanContext } from '@opentelemetry/api'; +import { AsyncLocalStorage } from 'node:async_hooks'; +import { ChatContext, type ChatItem, chatItemFingerprint } from '../llm/chat_context.js'; +import { + type ChatMessagePayload, + type MessagePart, + conversationMessages, + instructionParts, + messageLayout, + splitInstructions, +} from './gen_ai.js'; +import * as traceTypes from './trace_types.js'; + +/** Where an LLM input is recorded, which decides what the record counts. */ +export interface InputDeltaSite { + name: 'llm_node' | 'llm_request'; + /** + * The span records `lk.pii.chat_ctx` (every chat item, in order) rather than + * `gen_ai.input.messages` and `gen_ai.system_instructions`. + */ + chatCtx: boolean; +} + +export const LLM_NODE: InputDeltaSite = { name: 'llm_node', chatCtx: true }; +export const LLM_REQUEST: InputDeltaSite = { name: 'llm_request', chatCtx: false }; + +/** @internal */ +export interface InputBaseline { + /** id and content fingerprint of each item the span records, in order */ + keys: Array<[string, string]>; + /** per item, how it lands in gen_ai.input.messages */ + layout: Array<'new' | 'merged' | 'skipped'>; + instructions: string; + spanContext: SpanContext; +} + +/** + * How much of an LLM input a span records; the model always receives all of it. + * + * With `base` set, the record continues the one on that span, its parent: remove the parent's + * last `droppedFromBase` entries, then append this span's (`chatCtx` on `llm_node`, + * `conversation` on `llm_request`). Empty `instructions` then mean the parent's apply. + */ +export class InputDelta { + constructor( + readonly chatCtx: ChatContext, + readonly instructions: ChatItem[], + readonly conversation: ChatItem[], + readonly base?: SpanContext, + readonly droppedFromBase?: number, + ) {} + + static full(chatCtx: ChatContext): InputDelta { + const [instructions, conversation] = splitInstructions(chatCtx.items); + return new InputDelta(chatCtx, instructions, conversation); + } + + systemInstructions(): MessagePart[] { + return instructionParts(this.instructions); + } + + inputMessages(): ChatMessagePayload[] { + return conversationMessages(this.conversation); + } +} + +/** + * Per-agent memory of the LLM input recorded for the last committed generation, which the next + * generation's spans continue (see `compute`). + */ +export class InputDeltaTracker { + /** @internal */ + readonly _baselines = new Map(); + + begin(): InputDeltaScope { + return new InputDeltaScope(this); + } +} + +/** + * One generation's view of an `InputDeltaTracker`: the inputs its spans recorded, which + * become the tracker's baseline once the generation is committed. + * + * A generation that is never committed (a discarded preemptive generation, an aborted reply) is + * compared against the baseline but never replaces it. + */ +export class InputDeltaScope { + #pending = new Map(); + #committed = false; + + constructor(private readonly tracker: InputDeltaTracker) {} + + commit(): void { + this.#committed = true; + for (const [site, baseline] of this.#pending) this.tracker._baselines.set(site, baseline); + } + + delta(site: InputDeltaSite, chatCtx: ChatContext, span: Span): InputDelta { + const full = InputDelta.full(chatCtx); + const records = site.chatCtx ? [...chatCtx.items] : full.conversation; + const current: InputBaseline = { + keys: records.map((item) => [item.id, chatItemFingerprint(item)]), + layout: messageLayout(records), + instructions: JSON.stringify(full.systemInstructions()), + spanContext: span.spanContext(), + }; + const parent = this.tracker._baselines.get(site); + + if (span.isRecording()) { + // the content recorded on the span: a preemptive generation's message may still change + // before it is committed, and the next span must then record that edit + this.#pending.set(site, current); + if (this.#committed) this.tracker._baselines.set(site, current); + } + + if (!parent) return full; + const shared = sharedPrefix(current, parent, !site.chatCtx); + if (site.chatCtx) { + if (shared === 0) return full; + return new InputDelta( + new ChatContext(records.slice(shared)), + full.instructions, + full.conversation, + parent.spanContext, + parent.keys.length - shared, + ); + } + + const sameInstructions = current.instructions === parent.instructions; + if (shared === 0 && !sameInstructions) return full; + return new InputDelta( + chatCtx, + sameInstructions ? [] : full.instructions, + records.slice(shared), + parent.spanContext, + parent.layout.slice(shared).filter((placement) => placement === 'new').length, + ); + } +} + +function sharedPrefix(current: InputBaseline, parent: InputBaseline, genAIMessages: boolean) { + let n = 0; + while ( + n < current.keys.length && + n < parent.keys.length && + current.keys[n]![0] === parent.keys[n]![0] && + current.keys[n]![1] === parent.keys[n]![1] + ) { + n++; + } + if (genAIMessages) { + // never cut inside a message, on either side + while (n > 0 && (insideMessage(current.layout, n) || insideMessage(parent.layout, n))) n--; + } + return n; +} + +/** + * Whether a cut before item `n` splits a message: the next item that lands in a message is a tool + * call merged into the one before the cut (also across skipped items, such as a config update). + */ +function insideMessage(layout: InputBaseline['layout'], n: number): boolean { + while (n < layout.length && layout[n] === 'skipped') n++; + return n < layout.length && layout[n] === 'merged'; +} + +const scopeStorage = new AsyncLocalStorage(); + +/** Run `fn` with `scope` as the generation that the spans it creates record against. */ +export function runWithScope(scope: InputDeltaScope, fn: () => T): T { + return scopeStorage.run(scope, fn); +} + +export function active(): boolean { + return scopeStorage.getStore() !== undefined; +} + +/** + * Decide how much of `chatCtx` the span records (see the module documentation). The model always + * receives all of it; this only affects telemetry. Without `inputDelta`, return all of `chatCtx`. + */ +export function compute(site: InputDeltaSite, chatCtx: ChatContext, span: Span): InputDelta { + return scopeStorage.getStore()?.delta(site, chatCtx, span) ?? InputDelta.full(chatCtx); +} + +/** Point a span whose record continues another span's at that parent. */ +export function setAttributes(span: Span, delta: InputDelta): void { + if (!delta.base || !span.isRecording()) return; + span.setAttributes({ + [traceTypes.ATTR_INPUT_DELTA]: true, + [traceTypes.ATTR_INPUT_BASE_SPAN_ID]: delta.base.spanId, + [traceTypes.ATTR_INPUT_DROPPED_FROM_BASE]: delta.droppedFromBase ?? 0, + }); + span.addLink({ context: delta.base }); +} diff --git a/agents/src/telemetry/trace_types.test.ts b/agents/src/telemetry/trace_types.test.ts index cef2187f99..e8615d658e 100644 --- a/agents/src/telemetry/trace_types.test.ts +++ b/agents/src/telemetry/trace_types.test.ts @@ -110,6 +110,10 @@ const SAFE_KEYS = new Set([ 'lk.agent_name', 'lk.cloud_agent_id', 'lk.deployment_id', + // Input delta markers: a flag, a count, and a span id. + 'lk.input.delta', + 'lk.input.base_span_id', + 'lk.input.dropped_from_base', 'lk.session_options', 'lk.generation_id', 'lk.generation_count', diff --git a/agents/src/telemetry/trace_types.ts b/agents/src/telemetry/trace_types.ts index 5c9ce6cd45..7a3cd8eaeb 100644 --- a/agents/src/telemetry/trace_types.ts +++ b/agents/src/telemetry/trace_types.ts @@ -146,6 +146,19 @@ export const ATTR_RESPONSE_FUNCTION_CALLS = 'lk.pii.response.function_calls'; /** Time to first token in seconds. */ export const ATTR_RESPONSE_TTFT = 'lk.response.ttft'; +// input deltas (`record: { inputDelta: true }`); see telemetry/input_delta.ts +/** + * The span's record (`lk.pii.chat_ctx` on `llm_node`, `gen_ai.input.messages` on `llm_request`) + * continues the one on its parent, `lk.input.base_span_id`. + */ +export const ATTR_INPUT_DELTA = 'lk.input.delta'; +export const ATTR_INPUT_BASE_SPAN_ID = 'lk.input.base_span_id'; +/** + * Rebuild: the parent's full record without its last N entries, followed by this span's. Without + * `gen_ai.system_instructions` on an `llm_request` span, the parent's apply. + */ +export const ATTR_INPUT_DROPPED_FROM_BASE = 'lk.input.dropped_from_base'; + // function tool export const ATTR_FUNCTION_TOOL_ID = 'lk.function_tool.id'; export const ATTR_FUNCTION_TOOL_NAME = 'lk.function_tool.name'; diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index b59f6867d0..a480aae7fd 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -72,6 +72,7 @@ import { MultiInputStream } from '../stream/multi_input_stream.js'; import { STT, type STTError, type SpeechEvent } from '../stt/stt.js'; import { genAI, + inputDelta, recordException, recordRealtimeMetrics, redactionEnabled, @@ -570,6 +571,7 @@ export class AgentActivity implements RecognitionHooks { private inlineTaskLock = new Mutex(); private audioStream = new MultiInputStream(); private audioStreamId?: string; + private readonly inputDeltaTracker = new inputDelta.InputDeltaTracker(); // default to null as None, which maps to the default provider tool choice value private toolChoice: ToolChoice | null = null; @@ -3835,16 +3837,23 @@ export class AgentActivity implements RecognitionHooks { const runningCalls = getRunningTasks(this.agentSession); _injectRunningToolCalls(chatCtx, runningCalls); const tasks: Array> = []; - const [llmTask, llmGenData] = performLLMInference( - // preserve `this` context in llmNode - (...args) => this.agent.llmNode(...args), - chatCtx, - toolCtx, - modelSettings, - replyAbortController, - this.llm?.model, - this.llm?.provider, - ); + const deltaScope = this.agentSession.sessionOptions.recordingOptions.inputDelta + ? this.inputDeltaTracker.begin() + : undefined; + const startInference = () => + performLLMInference( + // preserve `this` context in llmNode + (...args) => this.agent.llmNode(...args), + chatCtx, + toolCtx, + modelSettings, + replyAbortController, + this.llm?.model, + this.llm?.provider, + ); + const [llmTask, llmGenData] = deltaScope + ? inputDelta.runWithScope(deltaScope, startInference) + : startInference(); tasks.push(llmTask); // as python's _on_llm_task_done: a genuine LLM failure (not a cancellation) fails the // speech, through exception() and the agent_turn span. Nothing else awaits this task's @@ -3948,6 +3957,8 @@ export class AgentActivity implements RecognitionHooks { await speechHandle.waitIfNotInterrupted([speechHandle._waitForScheduled()]); + if (deltaScope && speechHandle.scheduled) deltaScope.commit(); + let userMetrics: MetricsReport | undefined = _previousUserMetrics; // Add new message to actual chat context if the speech is scheduled if (newMessage && speechHandle.scheduled) { diff --git a/agents/src/voice/agent_session.test.ts b/agents/src/voice/agent_session.test.ts index 23f36741d1..fa30af251c 100644 --- a/agents/src/voice/agent_session.test.ts +++ b/agents/src/voice/agent_session.test.ts @@ -188,6 +188,7 @@ describe('resolveRecordingOptions', () => { logs: true, transcript: true, redaction: false, + inputDelta: false, }); expect(resolveRecordingOptions(false)).toEqual({ audio: false, @@ -195,6 +196,7 @@ describe('resolveRecordingOptions', () => { logs: false, transcript: false, redaction: false, + inputDelta: false, }); }); @@ -205,6 +207,7 @@ describe('resolveRecordingOptions', () => { logs: true, transcript: true, redaction: false, + inputDelta: false, }); expect(resolveRecordingOptions({ redaction: true })).toEqual({ @@ -213,6 +216,7 @@ describe('resolveRecordingOptions', () => { logs: true, transcript: true, redaction: true, + inputDelta: false, }); // The granular form from the docs: keep audio, drop everything else. @@ -230,7 +234,10 @@ describe('resolveRecordingOptions', () => { logs: false, transcript: false, redaction: true, + inputDelta: false, }); + + expect(resolveRecordingOptions({ inputDelta: true }).inputDelta).toBe(true); }); it('returns a fresh object so callers cannot corrupt the shared defaults', () => { diff --git a/agents/src/voice/agent_session.ts b/agents/src/voice/agent_session.ts index 0f45c2c20a..195893e28c 100644 --- a/agents/src/voice/agent_session.ts +++ b/agents/src/voice/agent_session.ts @@ -158,9 +158,10 @@ export interface AgentSessionUsage { /** * Granular control over which recording features are active. * - * Recording keys default to `true` when omitted, so `{ logs: false }` means "record - * everything except logs". Redaction defaults to the project setting; `false` is ignored when - * redaction is enabled globally for the project. Pass to {@link AgentSession.start} as `record`: + * Content recording keys default to `true` when omitted, so `{ logs: false }` means "record + * everything except logs". Input deltas are opt-in. Redaction defaults to the project setting; + * `false` is ignored when redaction is enabled globally for the project. Pass to + * {@link AgentSession.start} as `record`: * * - `record: true` — all on (backward compatible) * - `record: false` — all off (backward compatible) @@ -178,6 +179,13 @@ export interface RecordingOptions { transcript?: boolean; /** Enable redaction. `false` does not disable project redaction. */ redaction?: boolean; + /** + * Record only input changes on successive LLM trace spans. The model still receives the full + * input. Delta spans link to `lk.input.base_span_id`; reconstruct the record by removing its + * last `lk.input.dropped_from_base` entries and appending the delta. Unchanged system + * instructions are inherited from the base span. Defaults to `false`. + */ + inputDelta?: boolean; } /** @internal Recording options with every category resolved to a boolean. */ @@ -189,6 +197,7 @@ const RECORDING_ALL_ON: ResolvedRecordingOptions = { logs: true, transcript: true, redaction: false, + inputDelta: false, }; const RECORDING_ALL_OFF: ResolvedRecordingOptions = { @@ -197,14 +206,15 @@ const RECORDING_ALL_OFF: ResolvedRecordingOptions = { logs: false, transcript: false, redaction: false, + inputDelta: false, }; const idleHoldStorage = new AsyncLocalStorage(); /** * Resolve a `record` argument into explicit per-category flags. A boolean turns - * every category on or off; a partial object is merged onto all-on so omitted - * keys default to `true`. + * every content category on or off; a partial object is merged onto those defaults. + * Input deltas remain off unless explicitly enabled. * * @internal */ diff --git a/agents/src/voice/agent_session_text_only.test.ts b/agents/src/voice/agent_session_text_only.test.ts index 8efe7f12ba..cc66e3571d 100644 --- a/agents/src/voice/agent_session_text_only.test.ts +++ b/agents/src/voice/agent_session_text_only.test.ts @@ -141,6 +141,7 @@ describe('AgentSession text-only gating', () => { logs: true, transcript: true, redaction: false, + inputDelta: false, }); expect(initRecording).toHaveBeenCalledWith(session.sessionOptions.recordingOptions); await session.close(); diff --git a/agents/src/voice/eou_wait_span.test.ts b/agents/src/voice/eou_wait_span.test.ts index e50059fea8..3d7f4718c4 100644 --- a/agents/src/voice/eou_wait_span.test.ts +++ b/agents/src/voice/eou_wait_span.test.ts @@ -802,6 +802,7 @@ describe.sequential('eou_wait span', () => { logs: false, transcript: false, redaction: true, + inputDelta: false, }; }); expect(session._redactionEnabled).toBe(true); diff --git a/agents/src/voice/generation.ts b/agents/src/voice/generation.ts index 3b6d7760ad..1641e9efaa 100644 --- a/agents/src/voice/generation.ts +++ b/agents/src/voice/generation.ts @@ -31,7 +31,7 @@ import { parseFunctionArguments } from '../llm/utils.js'; import { isZodSchema, parseZodSchema } from '../llm/zod-utils.js'; import { log } from '../log.js'; import { IdentityTransform } from '../stream/identity_transform.js'; -import { genAI, traceTypes, tracer } from '../telemetry/index.js'; +import { genAI, inputDelta, traceTypes, tracer } from '../telemetry/index.js'; import { stripAllMarkup } from '../tts/provider_format.js'; import { type FlushSentinel, @@ -101,6 +101,9 @@ export function _injectRunningToolCalls( chatCtx.insert( FunctionCall.create({ ...runningCall, + // an id of its own, stable across turns: input-delta telemetry tells items apart by + // id, so the copy matches itself while the tool runs and never the real call + id: `${runningCall.id}_running`, extra: { ...runningCall.extra, [RUNNING_TOOL_PLACEHOLDER_KEY]: true }, }), ); @@ -648,12 +651,14 @@ export function performLLMInference( span: Span, inference: genAI.InferenceMarker, ) => { + const delta = inputDelta.compute(inputDelta.LLM_NODE, chatCtx, span); span.setAttribute( traceTypes.ATTR_CHAT_CTX, // snake_case wire shape, matching Python's `chat_ctx.to_dict()` for this span attribute // (toJSON() emits camelCase). Defaults exclude image/audio/timestamps like the Python side. - JSON.stringify(toSnakeCaseDeep(chatCtx.toJSON())), + JSON.stringify(toSnakeCaseDeep(delta.chatCtx.toJSON())), ); + inputDelta.setAttributes(span, delta); span.setAttribute(traceTypes.ATTR_FUNCTION_TOOLS, JSON.stringify(sortedToolNames(toolCtx))); let nodeError: Error | string | undefined; @@ -794,8 +799,8 @@ export function performLLMInference( timeToFirstChunk: data.ttft, }); genAI.setContentAttributes(span, { - systemInstructions: genAI.toSystemInstructions(chatCtx), - inputMessages: genAI.toInputMessages(chatCtx), + systemInstructions: delta.systemInstructions(), + inputMessages: delta.inputMessages(), toolDefinitions: genAI.toToolDefinitions(toolCtx.functionTools), outputMessages: genAI.toOutputMessages({ text: data.generatedText, diff --git a/agents/src/voice/input_delta_session.test.ts b/agents/src/voice/input_delta_session.test.ts new file mode 100644 index 0000000000..19a8fb2e24 --- /dev/null +++ b/agents/src/voice/input_delta_session.test.ts @@ -0,0 +1,147 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import type { Room } from '@livekit/rtc-node'; +import { 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 { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { InferenceExecutor } from '../ipc/inference_executor.js'; +import { + JobContext, + type JobProcess, + type RunningJobInfo, + runWithJobContextAsync, +} from '../job.js'; +import { initializeLogger } from '../log.js'; +import { setTracerProvider, traceTypes, tracer } from '../telemetry/index.js'; +import { Agent } from './agent.js'; +import { AgentSession, type RecordingOptions } from './agent_session.js'; +import { FakeLLM } from './testing/fake_llm.js'; + +initializeLogger({ pretty: false, level: 'silent' }); + +function jobContext(): JobContext { + const room = { + name: 'room', + on: () => room, + off: () => room, + isConnected: false, + remoteParticipants: new Map(), + }; + return new JobContext( + {} as JobProcess, + { + acceptArguments: { name: 'agent', identity: 'agent', metadata: '' }, + job: { id: 'job-id', room: { name: 'room' } }, + url: 'wss://example.livekit.cloud', + token: 'token', + workerId: 'worker-id', + } as RunningJobInfo, + room as unknown as Room, + () => {}, + () => {}, + {} as InferenceExecutor, + ); +} + +function inputRoles(span: ReadableSpan): string[] { + const messages = JSON.parse( + String(span.attributes[traceTypes.ATTR_GEN_AI_INPUT_MESSAGES]), + ) as Array<{ role: string }>; + return messages.map((message) => message.role); +} + +describe.sequential('AgentSession input delta recording', () => { + 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 runTwoTurns(record: RecordingOptions): Promise { + const llm = new FakeLLM([ + { input: 'Hello', content: 'Hi there' }, + { input: 'Weather?', content: 'Sunny' }, + ]); + const session = new AgentSession({ llm, turnHandling: { turnDetection: 'manual' } }); + session.output.setAudioEnabled(false); + session.output.setTranscriptionEnabled(false); + + const ctx = jobContext(); + vi.spyOn(ctx, 'initRecording').mockResolvedValue(); + await runWithJobContextAsync(ctx, () => + session.start({ + agent: new Agent({ instructions: 'You are a helpful assistant.' }), + record, + }), + ); + try { + await session.generateReply({ userInput: 'Hello' }).waitForPlayout(); + await session.generateReply({ userInput: 'Weather?' }).waitForPlayout(); + } finally { + await session.close(); + } + + return exporter + .getFinishedSpans() + .filter((span) => span.name === 'llm_request') + .sort((a, b) => a.startTime[0] - b.startTime[0] || a.startTime[1] - b.startTime[1]); + } + + it('records the second scheduled turn as a delta of the first', async () => { + const spans = await runTwoTurns({ traces: true, inputDelta: true }); + const [first, second] = spans; + + expect(spans).toHaveLength(2); + expect(first).toBeDefined(); + expect(second).toBeDefined(); + expect(first!.attributes[traceTypes.ATTR_INPUT_DELTA]).toBeUndefined(); + expect(first!.attributes[traceTypes.ATTR_GEN_AI_SYSTEM_INSTRUCTIONS]).toBeDefined(); + expect(inputRoles(first!)).toEqual(['user']); + + expect(second!.attributes).toMatchObject({ + [traceTypes.ATTR_INPUT_DELTA]: true, + [traceTypes.ATTR_INPUT_BASE_SPAN_ID]: first!.spanContext().spanId, + [traceTypes.ATTR_INPUT_DROPPED_FROM_BASE]: 0, + }); + expect(second!.attributes[traceTypes.ATTR_GEN_AI_SYSTEM_INSTRUCTIONS]).toBeUndefined(); + expect(inputRoles(second!)).toEqual(['assistant', 'user']); + expect(second!.links.map((link) => link.context.spanId)).toEqual([first!.spanContext().spanId]); + }); + + it.each([ + ['false', { traces: true, inputDelta: false }], + ['omitted', { traces: true }], + ] satisfies Array<[string, RecordingOptions]>)( + 'records full input when inputDelta is %s', + async (_name, record) => { + const spans = await runTwoTurns(record); + + expect(spans).toHaveLength(2); + for (const span of spans) { + expect(Object.keys(span.attributes).some((key) => key.startsWith('lk.input.'))).toBe(false); + expect(span.attributes[traceTypes.ATTR_GEN_AI_SYSTEM_INSTRUCTIONS]).toBeDefined(); + } + expect(inputRoles(spans[1]!)).toEqual(['user', 'assistant', 'user']); + }, + ); +}); diff --git a/agents/src/voice/report.test.ts b/agents/src/voice/report.test.ts index c498721c49..82146fa526 100644 --- a/agents/src/voice/report.test.ts +++ b/agents/src/voice/report.test.ts @@ -42,6 +42,7 @@ function baseOptions(): ReportOptions { logs: false, transcript: false, redaction: false, + inputDelta: false, }, }; } @@ -450,6 +451,7 @@ describe('createSessionReport recordingOptions', () => { logs: true, transcript: false, redaction: true, + inputDelta: false, }; const report = makeReport(recordingOptions); expect(report.options.recordingOptions).toEqual(recordingOptions); diff --git a/agents/src/voice/running_tool_placeholders.test.ts b/agents/src/voice/running_tool_placeholders.test.ts index c4919ef668..3ec1cb2139 100644 --- a/agents/src/voice/running_tool_placeholders.test.ts +++ b/agents/src/voice/running_tool_placeholders.test.ts @@ -26,6 +26,11 @@ describe('running tool placeholders', () => { (item) => item.type === 'function_call_output' && item.callId === running.callId, ), ).toHaveLength(1); + // ids of their own, the same on every turn the call is still running + expect(chatCtx.items.map((item) => item.id)).toEqual([ + `${running.id}_running`, + `lk_running_placeholder/${running.callId}`, + ]); }); it('adds only the missing output when the running call is already in context', () => { diff --git a/agents/src/voice/turn_config/utils.ts b/agents/src/voice/turn_config/utils.ts index 87c415e1ce..91d0a83757 100644 --- a/agents/src/voice/turn_config/utils.ts +++ b/agents/src/voice/turn_config/utils.ts @@ -36,6 +36,7 @@ const defaultSessionOptions = { logs: false, transcript: false, redaction: false, + inputDelta: false, }, expressive: false, } as const satisfies AgentSessionOptions & { recordingOptions: ResolvedRecordingOptions };