Skip to content

Commit 1070190

Browse files
tinalenguyentinalenguyen
authored
(realtime): refactor session updates (#1224)
1 parent dfe5f1b commit 1070190

4 files changed

Lines changed: 68 additions & 54 deletions

File tree

‎.changeset/tiny-tips-clean.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
---
2+
'@livekit/agents-plugin-phonic': patch
3+
'@livekit/agents': patch
4+
---
5+
6+
refactor \_updateSession in phonic and base realtimesession class

‎agents/src/llm/realtime.ts‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
import type { AudioFrame } from '@livekit/rtc-node';
55
import { EventEmitter } from 'events';
66
import type { ReadableStream } from 'node:stream/web';
7+
import { log } from '../log.js';
78
import { MultiInputStream } from '../stream/multi_input_stream.js';
89
import { Task } from '../utils.js';
910
import type { TimedString } from '../voice/io.js';
@@ -88,6 +89,7 @@ export abstract class RealtimeModel {
8889

8990
export abstract class RealtimeSession extends EventEmitter {
9091
protected _realtimeModel: RealtimeModel;
92+
protected logger = log();
9193
private inputAudioStream = new MultiInputStream<AudioFrame>();
9294
private inputAudioStreamId?: string;
9395
private _mainTask: Task<void>;
@@ -149,6 +151,34 @@ export abstract class RealtimeSession extends EventEmitter {
149151
audioTranscript?: string;
150152
}): Promise<void>;
151153

154+
async _updateSession(
155+
instructions?: string,
156+
chatCtx?: ChatContext,
157+
tools?: ToolContext,
158+
): Promise<void> {
159+
if (instructions !== undefined) {
160+
try {
161+
await this.updateInstructions(instructions);
162+
} catch (error) {
163+
this.logger.error(error, 'failed to update the instructions');
164+
}
165+
}
166+
if (chatCtx !== undefined) {
167+
try {
168+
await this.updateChatCtx(chatCtx);
169+
} catch (error) {
170+
this.logger.error(error, 'failed to update the chat context');
171+
}
172+
}
173+
if (tools !== undefined) {
174+
try {
175+
await this.updateTools(tools);
176+
} catch (error) {
177+
this.logger.error(error, 'failed to update the tools');
178+
}
179+
}
180+
}
181+
152182
async close(): Promise<void> {
153183
this._mainTask.cancel();
154184
await this.inputAudioStream.close();

‎agents/src/voice/agent_activity.ts‎

Lines changed: 10 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -409,42 +409,16 @@ export class AgentActivity implements RecognitionHooks {
409409
// skip the update if the session is reused and no mid-session update is supported
410410
// this means the content is the same as the previous session
411411
const capabilities = this.llm.capabilities;
412-
if (rtReused && this.realtimeSession?.realtimeModel.provider === 'phonic') {
413-
// if the session is being reused, then call phonic's _updateSession to send a full mid-session config update.
414-
// otherwise, call the separate update_* functions to build the initial config.
415-
try {
416-
await (this.realtimeSession as any)._updateSession(
417-
this.agent.instructions,
418-
this.agent.chatCtx,
419-
this.tools,
420-
);
421-
} catch (error) {
422-
this.logger.error(error, 'failed to update phonic session');
423-
}
424-
} else {
425-
if (!rtReused || capabilities.midSessionInstructionsUpdate) {
426-
try {
427-
await this.realtimeSession!.updateInstructions(this.agent.instructions);
428-
} catch (error) {
429-
this.logger.error(error, 'failed to update the instructions');
430-
}
431-
}
432-
433-
if (!rtReused || capabilities.midSessionChatCtxUpdate) {
434-
try {
435-
await this.realtimeSession!.updateChatCtx(this.agent.chatCtx);
436-
} catch (error) {
437-
this.logger.error(error, 'failed to update the chat context');
438-
}
439-
}
440-
441-
if (!rtReused || capabilities.midSessionToolsUpdate) {
442-
try {
443-
await this.realtimeSession!.updateTools(this.tools);
444-
} catch (error) {
445-
this.logger.error(error, 'failed to update the tools');
446-
}
447-
}
412+
try {
413+
await this.realtimeSession!._updateSession(
414+
!rtReused || capabilities.midSessionInstructionsUpdate
415+
? this.agent.instructions
416+
: undefined,
417+
!rtReused || capabilities.midSessionChatCtxUpdate ? this.agent.chatCtx : undefined,
418+
!rtReused || capabilities.midSessionToolsUpdate ? this.tools : undefined,
419+
);
420+
} catch (error) {
421+
this.logger.error(error, 'failed to update realtime session');
448422
}
449423

450424
if (!capabilities.audioOutput && !this.tts && this.agentSession.output.audio) {

‎plugins/phonic/src/realtime/realtime_model.ts‎

Lines changed: 22 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -245,7 +245,7 @@ export class RealtimeSession extends llm.RealtimeSession {
245245

246246
private client: PhonicClient;
247247
private socket?: Awaited<ReturnType<PhonicClient['conversations']['connect']>>;
248-
private logger = log();
248+
#logger = log();
249249
private closed = false;
250250
private configSent = false;
251251
private instructionsReady = new Future<void>();
@@ -288,7 +288,7 @@ export class RealtimeSession extends llm.RealtimeSession {
288288

289289
async updateInstructions(instructions: string): Promise<void> {
290290
if (this.configSent) {
291-
this.logger.warn(
291+
this.#logger.warn(
292292
'updateInstructions called after config was already sent. Phonic does not support updating instructions mid-session.',
293293
);
294294
return;
@@ -311,7 +311,7 @@ export class RealtimeSession extends llm.RealtimeSession {
311311
.map((item) => `${item.role}: ${item.textContent}`)
312312
.join('\n');
313313
if (turnHistory.trim() !== '') {
314-
this.logger.debug(
314+
this.#logger.debug(
315315
'updateChatCtx called with messages prior to config being sent to Phonic. Including conversation state in system instructions.',
316316
);
317317
this.systemPromptPostfix = CONVERSATION_HISTORY_PREFIX + turnHistory;
@@ -329,7 +329,7 @@ export class RealtimeSession extends llm.RealtimeSession {
329329
const item = chatCtx.getById(itemId);
330330
if (item?.type === 'function_call_output' && this.pendingToolCallIds.has(item.callId)) {
331331
this.pendingToolCallIds.delete(item.callId);
332-
this.logger.info(`Sending tool call output for ${item.name} (call_id: ${item.callId})`);
332+
this.#logger.info(`Sending tool call output for ${item.name} (call_id: ${item.callId})`);
333333
this.socket?.sendToolCallOutput({
334334
type: 'tool_call_output',
335335
tool_call_id: item.callId,
@@ -339,7 +339,7 @@ export class RealtimeSession extends llm.RealtimeSession {
339339
}
340340
if (item?.type === 'message') {
341341
if ((item.role === 'system' || item.role === 'developer') && item.textContent) {
342-
this.logger.debug(`Sending add system message: ${item.textContent}`);
342+
this.#logger.debug(`Sending add system message: ${item.textContent}`);
343343
this.socket?.sendAddSystemMessage({
344344
type: 'add_system_message',
345345
system_message: item.textContent,
@@ -352,7 +352,7 @@ export class RealtimeSession extends llm.RealtimeSession {
352352
this._chatCtx = chatCtx.copy();
353353

354354
if (!sentToolCallOutput && !sentAddSystemMessage) {
355-
this.logger.warn(
355+
this.#logger.warn(
356356
'updateChatCtx called but no new tool call outputs to send. Phonic does not support general chat context updates.',
357357
);
358358
}
@@ -363,7 +363,7 @@ export class RealtimeSession extends llm.RealtimeSession {
363363

364364
async updateTools(tools: llm.ToolContext): Promise<void> {
365365
if (this.configSent) {
366-
this.logger.warn(
366+
this.#logger.warn(
367367
'updateTools called after config was already sent. Phonic does not support updating tools mid-session.',
368368
);
369369
return;
@@ -393,11 +393,15 @@ export class RealtimeSession extends llm.RealtimeSession {
393393
this.toolsReady.resolve();
394394
}
395395

396-
async _updateSession(
396+
override async _updateSession(
397397
instructions?: string,
398398
chatCtx?: llm.ChatContext,
399399
tools?: llm.ToolContext,
400400
): Promise<void> {
401+
if (!this.configSent) {
402+
await super._updateSession(instructions, chatCtx, tools);
403+
return;
404+
}
401405
await this.readyToStart.await;
402406
if (instructions !== undefined) {
403407
this.options.instructions = instructions;
@@ -442,7 +446,7 @@ export class RealtimeSession extends llm.RealtimeSession {
442446
];
443447

444448
if (this.socket) {
445-
this.logger.info('Sending mid-session reset to Phonic');
449+
this.#logger.info('Sending mid-session reset to Phonic');
446450
this.socket.sendReset({
447451
type: 'reset',
448452
config: this.buildConfigOptions({ systemPrompt, toolsPayload }),
@@ -451,7 +455,7 @@ export class RealtimeSession extends llm.RealtimeSession {
451455
}
452456

453457
updateOptions(_options: { toolChoice?: llm.ToolChoice | null }): void {
454-
this.logger.warn('updateOptions is not supported by the Phonic realtime model.');
458+
this.#logger.warn('updateOptions is not supported by the Phonic realtime model.');
455459
}
456460

457461
pushAudio(frame: AudioFrame): void {
@@ -512,20 +516,20 @@ export class RealtimeSession extends llm.RealtimeSession {
512516
}
513517

514518
async commitAudio(): Promise<void> {
515-
this.logger.warn('commitAudio is not supported by the Phonic realtime model.');
519+
this.#logger.warn('commitAudio is not supported by the Phonic realtime model.');
516520
}
517521
async clearAudio(): Promise<void> {
518-
this.logger.warn('clearAudio is not supported by the Phonic realtime model.');
522+
this.#logger.warn('clearAudio is not supported by the Phonic realtime model.');
519523
}
520524

521525
async interrupt(): Promise<void> {
522-
this.logger.warn(
526+
this.#logger.warn(
523527
'interrupt() is not supported by Phonic realtime model. User interruptions are automatically handled by Phonic.',
524528
);
525529
}
526530

527531
async truncate(_options: { messageId: string; audioEndMs: number; audioTranscript?: string }) {
528-
this.logger.warn('truncate is not supported by the Phonic realtime model.');
532+
this.#logger.warn('truncate is not supported by the Phonic realtime model.');
529533
}
530534

531535
async close(): Promise<void> {
@@ -652,7 +656,7 @@ export class RealtimeSession extends llm.RealtimeSession {
652656
break;
653657
case 'conversation_created':
654658
this.conversationId = message.conversation_id;
655-
this.logger.info(`Phonic Conversation began with ID: ${this.conversationId}`);
659+
this.#logger.info(`Phonic Conversation began with ID: ${this.conversationId}`);
656660
break;
657661
case 'tool_call_interrupted':
658662
this.handleToolCallInterrupted(message);
@@ -675,7 +679,7 @@ export class RealtimeSession extends llm.RealtimeSession {
675679
* we only process the chunks when the assistant is speaking to align with the generations model, whereby new streams are created for each turn.
676680
*/
677681
if (this.currentGeneration === undefined && message.text) {
678-
this.logger.debug('Starting new generation due to text in audio chunk');
682+
this.#logger.debug('Starting new generation due to text in audio chunk');
679683
this.startNewAssistantTurn({ userInitiated: false });
680684
}
681685

@@ -727,7 +731,7 @@ export class RealtimeSession extends llm.RealtimeSession {
727731
this.pendingToolCallIds.add(message.tool_call_id);
728732

729733
if (this.currentGeneration === undefined) {
730-
this.logger.warn('Encountered tool call but no active generation. Starting new turn.');
734+
this.#logger.warn('Encountered tool call but no active generation. Starting new turn.');
731735
this.startNewAssistantTurn({ userInitiated: false });
732736
}
733737

@@ -744,7 +748,7 @@ export class RealtimeSession extends llm.RealtimeSession {
744748

745749
private handleToolCallInterrupted(message: Phonic.ToolCallInterruptedPayload): void {
746750
this.pendingToolCallIds.delete(message.tool_call_id);
747-
this.logger.warn(
751+
this.#logger.warn(
748752
`Tool call for ${message.tool_name} (call_id: ${message.tool_call_id}) was cancelled due to user interruption.`,
749753
);
750754
}

0 commit comments

Comments
 (0)