feat(api-server): 接入 StepFun 流式 TTS

Author: luoling8192

server/apps/api/src/routes/admin/config/router/index.ts

@@ -5,8 +5,10 @@ import { Hono } from 'hono'
 import {
   array,
   boolean,
+  check,
   literal,
   maxLength,
+  minLength,
   nonEmpty,
   object,
   optional,
@@ -104,6 +106,31 @@ const StepfunSliceSchema = object({
   existingKeyEntryId: optional(pipe(string(), nonEmpty(), maxLength(200), NO_PIPE)),
 })
 
+const STEPFUN_STREAMING_MODEL_ID = /^stepfun\/(?:stepaudio-2\.5-tts|step-tts-2|step-tts-mini)$/
+
+const StepfunStreamingSliceSchema = pipe(object({
+  kind: literal('stepfun-streaming'),
+  enabled: boolean(),
+  upstreamURL: pipe(string(), regex(/^wss?:\/\/\S+$/, 'upstreamURL must start with ws:// or wss://'), maxLength(500)),
+  models: pipe(array(object({
+    id: pipe(string(), regex(STEPFUN_STREAMING_MODEL_ID, 'models[].id must be a supported StepFun streaming model'), maxLength(200)),
+    name: optional(pipe(string(), nonEmpty(), maxLength(200))),
+    description: optional(pipe(string(), nonEmpty(), maxLength(500))),
+  })), minLength(1, 'models must not be empty')),
+  defaultModel: pipe(string(), regex(STEPFUN_STREAMING_MODEL_ID, 'defaultModel must be a supported StepFun streaming model'), maxLength(200)),
+  voices: pipe(array(object({
+    id: pipe(string(), nonEmpty('voices[].id is required'), maxLength(200)),
+    name: optional(pipe(string(), nonEmpty(), maxLength(200))),
+    description: optional(pipe(string(), nonEmpty(), maxLength(500))),
+    labels: optional(record(string(), string())),
+    languages: optional(array(object({ code: pipe(string(), nonEmpty()), title: pipe(string(), nonEmpty()) }))),
+  })), minLength(1, 'voices must not be empty')),
+  instruction: optional(pipe(string(), nonEmpty(), maxLength(200))),
+  plaintextKey: optional(pipe(string(), nonEmpty('plaintextKey must not be empty when provided'), maxLength(MAX_KEY_LENGTH))),
+  keyEntryId: optional(pipe(string(), nonEmpty(), maxLength(200), NO_PIPE)),
+  existingKeyEntryId: optional(pipe(string(), nonEmpty(), maxLength(200), NO_PIPE)),
+}), check(config => new Set(config.models.map(model => model.id)).size === config.models.length, 'models[].id must be unique'), check(config => config.models.some(model => model.id === config.defaultModel), 'defaultModel must be present in models'))
+
 const AliyunNlsAsrSliceSchema = object({
   kind: literal('aliyun-nls-asr'),
   modelName: pipe(string(), nonEmpty('modelName is required'), maxLength(200), NO_PIPE),
@@ -162,6 +189,7 @@ const SliceSchema = variant('kind', [
   AzureSliceSchema,
   DashscopeSliceSchema,
   StepfunSliceSchema,
+  StepfunStreamingSliceSchema,
   AliyunNlsAsrSliceSchema,
   UnspeechSliceSchema,
 ])

server/apps/api/src/routes/audio-speech-ws/route.test.ts

@@ -28,6 +28,7 @@ interface MockUpstream {
 async function startMockUpstream(
   scriptedResponses: MockUpstream['scriptedResponses'],
   voices: Array<{ id: string, name?: string }> = [{ id: 'mock', name: 'Mock Voice' }],
+  protocol: 'unspeech' | 'stepfun' = 'unspeech',
 ): Promise<MockUpstream> {
   const receivedFrames: MockUpstream['receivedFrames'] = []
   let observedAuth: string | undefined
@@ -45,6 +46,12 @@ async function startMockUpstream(
 
   wss.on('connection', (ws, req) => {
     observedAuth = req.headers.authorization
+    if (protocol === 'stepfun') {
+      ws.send(JSON.stringify({
+        type: 'tts.connection.done',
+        data: { session_id: 'stepfun-session' },
+      }))
+    }
     let replayed = false
     ws.on('message', async (data, isBinary) => {
       const buf = Buffer.isBuffer(data) ? data : Buffer.from(data as ArrayBuffer)
@@ -63,11 +70,23 @@ async function startMockUpstream(
         return
 
       let triggerReplay = false
+      if (protocol === 'stepfun') {
+        if (!isBinary) {
+          const event = JSON.parse(decoded as string) as { type?: string }
+          if (event.type === 'tts.create') {
+            ws.send(JSON.stringify({ type: 'tts.response.created', data: { session_id: 'stepfun-session' } }))
+            return
+          }
+          if (event.type === 'tts.text.done')
+            triggerReplay = true
+        }
+      }
+
       if (isBinary) {
         // Streaming protocol's only legal client→server binary frames
         // would be raw audio (we never send any in tests).
       }
-      else {
+      else if (protocol === 'unspeech') {
         try {
           const ev = JSON.parse(decoded as string) as { event?: string }
           if (ev.event === 'finish' || ev.event === 'cancel')
@@ -165,6 +184,13 @@ function makeFakeDeps(overrides: {
   fluxBalance: number
   decryptedKey?: string
   streamingModels?: Array<{ id: string, name?: string, description?: string }>
+  stepfunStreaming?: {
+    enabled: boolean
+    baseURL: string
+    models: Array<{ id: string, name?: string, description?: string }>
+    defaultModel: string
+    voices: Array<{ id: string, name?: string }>
+  }
 }) {
   const ttsMeter = {
     assertCanAfford: vi.fn(async (_userId: string, _newUnits: number, currentBalance: number) => {
@@ -205,6 +231,16 @@ function makeFakeDeps(overrides: {
           },
         }
       }
+      if (key === 'STEPFUN_STREAMING_TTS_UPSTREAM') {
+        const config = overrides.stepfunStreaming
+        return config
+          ? {
+              ...config,
+              keys: [{ id: 'test-key-1', ciphertext: 'ENCRYPTED_PLACEHOLDER' }],
+              voices: config.voices.map(voice => ({ ...voice, labels: {}, languages: [] })),
+            }
+          : null
+      }
       return null
     }),
   }
@@ -312,6 +348,85 @@ describe('audio-speech-ws route', () => {
     }))
   })
 
+  it('translates the AIRI stream protocol to native StepFun websocket events', async () => {
+    const audioPayload = Buffer.from('STEPFUN_AUDIO', 'utf8').toString('base64')
+    const text = 'x'.repeat(2001)
+    upstream = await startMockUpstream([
+      { kind: 'json', payload: { type: 'tts.response.sentence.start', data: { session_id: 'stepfun-session', text: 'hello' } } },
+      { kind: 'json', payload: { type: 'tts.response.audio.delta', data: { session_id: 'stepfun-session', audio: audioPayload } } },
+      { kind: 'json', payload: { type: 'tts.response.sentence.end', data: { session_id: 'stepfun-session', text: 'hello' } } },
+      { kind: 'json', payload: { type: 'tts.response.audio.done', data: { session_id: 'stepfun-session', audio: '' } } },
+    ], [{ id: 'lively-girl', name: 'Lively Girl' }], 'stepfun')
+
+    const deps = makeFakeDeps({
+      upstreamURL: upstream.url,
+      restBaseURL: upstream.restBaseURL,
+      fluxBalance: 100,
+      stepfunStreaming: {
+        enabled: true,
+        baseURL: upstream.url,
+        models: [{ id: 'stepfun/step-tts-2', name: 'Step TTS 2' }],
+        defaultModel: 'stepfun/step-tts-2',
+        voices: [{ id: 'lively-girl', name: 'Lively Girl' }],
+      },
+    })
+    const handlers = createAudioSpeechWsHandlers(deps as any)
+    const events = handlers('user-stepfun')
+    const client = makeMockClientWs()
+
+    await driveClientSession(events, client, [
+      JSON.stringify({ event: 'start', model: 'stepfun/step-tts-2', voice: 'lively-girl' }),
+      JSON.stringify({ event: 'text', text }),
+      JSON.stringify({ event: 'finish' }),
+    ])
+    await new Promise(r => setTimeout(r, 200))
+
+    const upstreamFrames = upstream.receivedFrames.map(frame => JSON.parse(frame.data as string))
+    expect(upstreamFrames.map(frame => frame.type)).toEqual(['tts.create', 'tts.text.delta', 'tts.text.delta', 'tts.text.delta', 'tts.text.done'])
+    expect(upstreamFrames[0].data).toMatchObject({
+      session_id: 'stepfun-session',
+      voice_id: 'lively-girl',
+      mode: 'default',
+      response_format: 'mp3_stream',
+    })
+    expect(upstreamFrames.slice(1, 4).map(frame => frame.data.text.length)).toEqual([1000, 1000, 1])
+
+    const clientTextFrames = client.sent.filter(s => s.kind === 'text').map(s => JSON.parse(s.data as string))
+    expect(clientTextFrames.map(frame => frame.event)).toEqual(['session.started', 'sentence.start', 'sentence.end', 'session.finished'])
+    expect(client.sent.filter(s => s.kind === 'binary')).toHaveLength(1)
+    expect(deps.ttsMeter.accumulate).toHaveBeenCalledWith(expect.objectContaining({ userId: 'user-stepfun', units: 2001 }))
+  })
+
+  it('settles StepFun text usage when the client cancels before audio completion', async () => {
+    upstream = await startMockUpstream([], [{ id: 'lively-girl', name: 'Lively Girl' }], 'stepfun')
+    const deps = makeFakeDeps({
+      upstreamURL: upstream.url,
+      restBaseURL: upstream.restBaseURL,
+      fluxBalance: 100,
+      stepfunStreaming: {
+        enabled: true,
+        baseURL: upstream.url,
+        models: [{ id: 'stepfun/step-tts-2', name: 'Step TTS 2' }],
+        defaultModel: 'stepfun/step-tts-2',
+        voices: [{ id: 'lively-girl', name: 'Lively Girl' }],
+      },
+    })
+    const events = createAudioSpeechWsHandlers(deps as any)('user-stepfun-cancel')
+    const client = makeMockClientWs()
+
+    await driveClientSession(events, client, [
+      JSON.stringify({ event: 'start', model: 'stepfun/step-tts-2', voice: 'lively-girl' }),
+      JSON.stringify({ event: 'text', text: 'paid text' }),
+    ])
+    events.onClose?.(new Event('close') as any, client.ctx)
+    await new Promise(r => setTimeout(r, 100))
+
+    expect(deps.ttsMeter.accumulate).toHaveBeenCalledWith(expect.objectContaining({
+      userId: 'user-stepfun-cancel',
+      units: 9,
+    }))
+  })
+
   it('refuses the session with insufficient_flux when the user is broke', async () => {
     upstream = await startMockUpstream([])
     const deps = makeFakeDeps({ upstreamURL: upstream.url, restBaseURL: upstream.restBaseURL, fluxBalance: 0 })