examples: switch homepage agent to Fish Audio with expressive mode

.changeset/lucky-moons-sing.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': patch
+---
+
+Add expressive mode: `AgentSession({ expressive: true })` injects the TTS provider's markup guide into the LLM prompt so the model emits inline `<expr/>` delivery markers (emotion, pacing, non-verbal sounds). The markers are lowered to each provider's native syntax before synthesis (Cartesia, Inworld TTS 2, xAI, Fish Audio) and stripped from transcripts, with the segment's leading expression surfaced as the `lk.expression` transcription attribute. Steer delivery with `ExpressiveOptions.speechSteering` or override the injected prompt entirely.

agents/src/constants.ts

@@ -5,6 +5,16 @@ export const ATTRIBUTE_TRANSCRIPTION_TRACK_ID = 'lk.transcribed_track_id';
 export const ATTRIBUTE_TRANSCRIPTION_FINAL = 'lk.transcription_final';
 export const TOPIC_TRANSCRIPTION = 'lk.transcription';
 export const ATTRIBUTE_TRANSCRIPTION_SEGMENT_ID = 'lk.segment_id';
+/**
+ * The expression (delivery/emotion) the agent used for a transcription segment, surfaced
+ * so the frontend can react to it, when expressive markup is stripped from the transcript.
+ * The value is a JSON object `{"expression": ..., "mood": ...}` carrying the segment's
+ * leading expression — the `<expression>` tag for Inworld/xAI or the `<emotion>` tag for
+ * Cartesia — plus the mood it normalizes to, e.g.
+ * `{"expression":"speak happy","mood":"happy"}`. A JSON object (rather than a bare string)
+ * so the shape can gain fields later without breaking parsers.
+ */
+export const ATTRIBUTE_TRANSCRIPTION_EXPRESSION = 'lk.expression';
 export const ATTRIBUTE_PUBLISH_ON_BEHALF = 'lk.publish_on_behalf';
 export const TOPIC_CHAT = 'lk.chat';
 

agents/src/index.ts

@@ -14,6 +14,7 @@ export * from './audio.js';
 export * as beta from './beta/index.js';
 export * as cli from './cli.js';
 export * from './connection_pool.js';
+export { ATTRIBUTE_TRANSCRIPTION_EXPRESSION } from './constants.js';
 export { defineAgent, isAgent, type AgentDefinition } from './generator.js';
 export * as inference from './inference/index.js';
 export * from './inference_runner.js';

agents/src/inference/tts.ts

@@ -10,9 +10,9 @@ import { ConnectionPool } from '../connection_pool.js';
 import { type LanguageCode, normalizeLanguage } from '../language.js';
 import { log } from '../log.js';
 import { createStreamChannel } from '../stream/stream_channel.js';
-import { basic as tokenizeBasic } from '../tokenize/index.js';
 import type { ChunkedStream } from '../tts/index.js';
 import { SynthesizeStream as BaseSynthesizeStream, TTS as BaseTTS } from '../tts/index.js';
+import { dropBracketCues, sentenceTokenizer } from '../tts/provider_format.js';
 import { type APIConnectOptions, DEFAULT_API_CONNECT_OPTIONS } from '../types.js';
 import {
   Event,
@@ -412,6 +412,23 @@ export class TTS<TModel extends TTSModels> extends BaseTTS {
     return 'inference.TTS';
   }
 
+  /**
+   * Key the shared markup tables off the gateway model's provider.
+   *
+   * `llmInstructions` / `normalize` / `convert` are inherited from the base markup helper,
+   * keyed on this.
+   */
+  protected override markupProviderKey(): string {
+    const model = this.opts?.model ?? '';
+    const provider = model.split('/')[0] ?? '';
+    if (provider === 'inworld' && model.includes('tts-2')) {
+      return 'inworld';
+    } else if (provider === 'inworld') {
+      return ''; // older inworld models don't support markup
+    }
+    return provider;
+  }
+
   get model(): string {
     return this.opts.model ?? 'unknown';
   }
@@ -547,13 +564,25 @@ export class TTS<TModel extends TTSModels> extends BaseTTS {
 export class SynthesizeStream<TModel extends TTSModels> extends BaseSynthesizeStream {
   private opts: InferenceTTSOptions<TModel>;
   private tts: TTS<TModel>;
+  /**
+   * Snapshot whether expressive is active now, while the framework holds it fixed for this
+   * synthesis (set synchronously before `stream()`). Reading it lazily in `run` would race
+   * with the next turn/session mutating the shared TTS instance.
+   */
+  private expressive: boolean;
+  /**
+   * Alignment arrives finer-grained than a cue, so `dropBracketCues` parks the tail of an
+   * unclosed span here between messages.
+   */
+  private heldTokens: TimedString[] = [];
 
   #logger = log();
 
   constructor(tts: TTS<TModel>, opts: InferenceTTSOptions<TModel>, connOptions: APIConnectOptions) {
     super(tts, connOptions);
     this.opts = opts;
     this.tts = tts;
+    this.expressive = tts.expressive;
   }
 
   get label() {
@@ -587,7 +616,11 @@ export class SynthesizeStream<TModel extends TTSModels> extends BaseSynthesizeSt
     // Python side.
     let pendingTimedTranscripts: TimedString[] = [];
 
-    const sendTokenizerStream = new tokenizeBasic.SentenceTokenizer().stream();
+    // chunking defaults (cap + expressive batch size) live in provider_format
+    const provider = (this.opts.model ?? '').split('/')[0] ?? '';
+    const sendTokenizerStream = sentenceTokenizer(provider, {
+      expressive: this.expressive,
+    }).stream();
     const eventChannel = createStreamChannel<TtsServerEvent>();
     const requestId = shortuuid('tts_request_');
     const inputSentEvent = new Event();
@@ -637,7 +670,9 @@ export class SynthesizeStream<TModel extends TTSModels> extends BaseSynthesizeSt
           sendTokenizerStream.flush();
           continue;
         }
-        sendTokenizerStream.pushText(data);
+        // only expressive turns can carry markup; without it the text is a plain
+        // utterance and must reach the provider byte-for-byte
+        sendTokenizerStream.pushText(this.expressive ? this.tts.markup.normalize(data) : data);
       }
       // Only call endInput if the stream hasn't been closed by cleanup
       if (!closing) {
@@ -657,11 +692,17 @@ export class SynthesizeStream<TModel extends TTSModels> extends BaseSynthesizeSt
         if (this.opts.model) generationConfig.model = this.opts.model;
         if (this.opts.language) generationConfig.language = this.opts.language;
 
+        // re-normalize at sentence level: tags split across input chunks aren't caught by
+        // the per-chunk normalize in the input task
+        const converted = this.expressive
+          ? this.tts.markup.convert(this.tts.markup.normalize(ev.token))
+          : ev.token;
+
         this.markStarted();
         await sendClientEvent(
           {
             type: 'input_transcript',
-            transcript: ev.token + ' ',
+            transcript: converted + ' ',
             generation_config: generationConfig,
             extra: (this.opts.modelOptions as Record<string, unknown>) ?? {},
           },
@@ -801,29 +842,31 @@ export class SynthesizeStream<TModel extends TTSModels> extends BaseSynthesizeSt
               }
               break;
             case 'output_alignment':
+              let aligned: TimedString[] = [];
               if (serverEvent.words && serverEvent.words.length > 0) {
-                for (const w of serverEvent.words) {
-                  pendingTimedTranscripts.push(
-                    createTimedString({
-                      text: w.word,
-                      startTime: w.start,
-                      endTime: w.end,
-                    }),
-                  );
-                }
+                aligned = serverEvent.words.map((w) =>
+                  createTimedString({ text: w.word, startTime: w.start, endTime: w.end }),
+                );
               } else if (serverEvent.chars && serverEvent.chars.length > 0) {
-                for (const c of serverEvent.chars) {
-                  pendingTimedTranscripts.push(
-                    createTimedString({
-                      text: c.char,
-                      startTime: c.start,
-                      endTime: c.end,
-                    }),
-                  );
-                }
+                aligned = serverEvent.chars.map((c) =>
+                  createTimedString({ text: c.char, startTime: c.start, endTime: c.end }),
+                );
+              }
+              if (aligned.length > 0) {
+                // the provider aligned the *converted* text, so under expressive it carries
+                // native cues that were never spoken
+                pendingTimedTranscripts.push(
+                  ...(this.expressive ? dropBracketCues(aligned, this.heldTokens) : aligned),
+                );
               }
               break;
             case 'done':
+              if (this.heldTokens.length > 0) {
+                // release an unclosed span, cue unresolved
+                pendingTimedTranscripts.push(
+                  ...dropBracketCues([], this.heldTokens, { final: true }),
+                );
+              }
               for (const frame of bstream.flush()) {
                 sendLastFrame(currentSessionId!, false);
                 lastFrame = frame;

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 { stripExprMarkup } from '../tts/provider_format.js';
 import { createImmutableArray, shortuuid } from '../utils.js';
 import type { LLM } from './llm.js';
 import { type ProviderFormat, toChatCtx } from './provider_format/index.js';
@@ -228,10 +229,6 @@ export function concatInstructions(...parts: Array<string | Instructions>): stri
 
 export type ChatContent = ImageContent | AudioContent | Instructions | string;
 
-function stripExprMarkup(text: string): string {
-  return text.replace(/<expr\b[^>]*>/g, '').replace(/<\/expr\s*>/g, '');
-}
-
 export function createImageContent(params: {
   image: string | VideoFrame;
   id?: string;

agents/src/tokenize/basic/basic.ts

@@ -1,7 +1,7 @@
 // SPDX-FileCopyrightText: 2024 LiveKit, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
-import { BufferedSentenceStream, BufferedWordStream } from '../token_stream.js';
+import { BufferedSentenceStream, BufferedWordStream, xmlWrapTokenizer } from '../token_stream.js';
 import * as tokenizer from '../tokenizer.js';
 import { hyphenator } from './hyphenator.js';
 import { splitParagraphs } from './paragraph.js';
@@ -10,9 +10,40 @@ import { splitWords } from './word.js';
 
 interface TokenizerOptions {
   language: string;
+  /**
+   * Minimum length for a span to be treated as its own sentence; shorter spans are merged
+   * forward into the next one.
+   */
   minSentenceLength: number;
+  /** Minimum buffered text before the stream emits. */
   streamContextLength: number;
+  /** Keep original whitespace/formatting in emitted tokens. */
   retainFormat: boolean;
+  /**
+   * Hard cap on emitted token length; a token is flushed before appending a sentence that
+   * would exceed it. Unlimited when omitted.
+   */
+  maxTokenLength?: number;
+  /**
+   * Minimum length a token must reach before it is emitted. Sentences are batched together
+   * until the running token reaches this length, so raising it (e.g. toward
+   * `maxTokenLength`) yields larger, fewer chunks. Defaults to `minSentenceLength`
+   * (per-sentence emission).
+   */
+  minTokenLength?: number;
+  /**
+   * Minimum length for the *first* token of each segment, when it should differ from
+   * `minTokenLength`. Lets a batching consumer still emit its opening chunk as soon as one
+   * sentence is ready, so batching costs nothing at the head of a segment.
+   */
+  firstTokenLength?: number;
+  /**
+   * Treat XML markup as atomic — never split a tag across tokens and keep tags attached to
+   * the following sentence. Only enable when the input actually carries markup (e.g.
+   * expressive TTS): a stray "<" in plain text can otherwise hold back streaming until
+   * flush.
+   */
+  xmlAware?: boolean;
 }
 
 const defaultTokenizerOptions: TokenizerOptions = {
@@ -35,16 +66,24 @@ export class SentenceTokenizer extends tokenizer.SentenceTokenizer {
 
   // eslint-disable-next-line @typescript-eslint/no-unused-vars
   tokenize(text: string, language?: string): string[] {
-    return splitSentences(text, this.#config.minSentenceLength).map((tok) => tok[0]);
+    const split = (input: string) =>
+      splitSentences(input, this.#config.minSentenceLength, this.#config.retainFormat);
+    const tokenizeFnc = this.#config.xmlAware ? xmlWrapTokenizer(split) : split;
+    return tokenizeFnc(text).map((tok) => (Array.isArray(tok) ? tok[0] : tok));
   }
 
   // eslint-disable-next-line @typescript-eslint/no-unused-vars
   stream(language?: string): tokenizer.SentenceStream {
     return new BufferedSentenceStream(
       (text: string) =>
         splitSentences(text, this.#config.minSentenceLength, this.#config.retainFormat),
-      this.#config.minSentenceLength,
+      this.#config.minTokenLength ?? this.#config.minSentenceLength,
       this.#config.streamContextLength,
+      {
+        maxTokenLength: this.#config.maxTokenLength,
+        firstTokenLength: this.#config.firstTokenLength,
+        xmlAware: this.#config.xmlAware,
+      },
     );
   }
 }

agents/src/tokenize/basic/sentence.ts

@@ -82,7 +82,10 @@ export const splitSentences = (
   }
 
   if (buf) {
-    sentences.push([buf.slice(prePad.length), start, text.length - 1]);
+    // the trailing buffer runs to the end of the text — an end offset of `length - 1`
+    // leaves the last character stranded for any caller that slices by these offsets
+    // (see `xmlWrapTokenizer`, which rebuilds sentences from them)
+    sentences.push([buf.slice(prePad.length), start, text.length]);
   }
 
   return sentences;