From 17aae1222a8999b4c4fdbe5e89ae97c2a9f20a6d Mon Sep 17 00:00:00 2001 From: Neko Ayaka Date: Thu, 16 Jan 2025 16:25:30 +0800 Subject: [PATCH] chore(services/discord-bot): remote stt, fix server bugs --- cspell.config.yaml | 2 + packages/server-runtime/src/index.ts | 21 +- packages/server-sdk/src/client.ts | 41 +- pnpm-lock.yaml | 15 + services/discord-bot/.env | 5 +- services/discord-bot/package.json | 3 +- .../src/bots/discord/commands/summon.ts | 809 +++++++++++++++--- services/discord-bot/src/index.ts | 14 +- services/discord-bot/src/pipelines/tts.ts | 37 +- services/discord-bot/src/utils/audio.ts | 28 +- services/telegram-bot/.env | 3 +- services/telegram-bot/package.json | 2 +- 12 files changed, 826 insertions(+), 154 deletions(-) diff --git a/cspell.config.yaml b/cspell.config.yaml index 3d6300ca0..37de4c07a 100644 --- a/cspell.config.yaml +++ b/cspell.config.yaml @@ -77,6 +77,7 @@ words: - opusscript - pgvector - picklist + - picovoice - pinia - pixi - pixiv @@ -104,6 +105,7 @@ words: - wavefile - webgpu - worklet + - Xenova - xsai ignoreWords: [] import: [] diff --git a/packages/server-runtime/src/index.ts b/packages/server-runtime/src/index.ts index 80e6d9254..c186e7861 100644 --- a/packages/server-runtime/src/index.ts +++ b/packages/server-runtime/src/index.ts @@ -1,4 +1,5 @@ import type { WebSocketEvent } from '@proj-airi/server-shared/types' +import type { Peer } from 'crossws' import { Format, LogLevel, setGlobalFormat, setGlobalLogLevel, useLogg } from '@guiiai/logg' import { createApp, createRouter, defineWebSocketHandler } from 'h3' @@ -15,24 +16,34 @@ export const app = createApp({ const router = createRouter() app.use(router) +const peers = new Set() + router.get('/ws', defineWebSocketHandler({ open: (peer) => { - websocketLogger.withFields({ peer: peer.id }).log('connected') + peers.add(peer) + websocketLogger.withFields({ peer: peer.id, activePeers: peers.size }).log('connected') }, message: (peer, message) => { const event = message.json() as WebSocketEvent - websocketLogger.withFields({ peer: peer.id, message: event }).log('received message') switch (event.type) { - case 'input:text:voice': - websocketLogger.withFields({ message: event }).log('transcribed') + case 'input:text': break + case 'input:text:voice': + break + } + + for (const p of peers) { + if (p.id !== peer.id) { + p.send(JSON.stringify(event)) + } } }, error: (peer, error) => { websocketLogger.withFields({ peer: peer.id }).withError(error).error('an error occurred') }, close: (peer, details) => { - websocketLogger.withFields({ peer: peer.id, details }).log('closed') + websocketLogger.withFields({ peer: peer.id, details, activePeers: peers.size }).log('closed') + peers.delete(peer) }, })) diff --git a/packages/server-sdk/src/client.ts b/packages/server-sdk/src/client.ts index a121dae57..9c215aae2 100644 --- a/packages/server-sdk/src/client.ts +++ b/packages/server-sdk/src/client.ts @@ -1,4 +1,4 @@ -import type { WebSocketEvent, WebSocketEvents } from '@proj-airi/server-shared/types' +import type { WebSocketBaseEvent, WebSocketEvent, WebSocketEvents } from '@proj-airi/server-shared/types' import type { Blob } from 'node:buffer' import WebSocket from 'crossws/websocket' import { defu } from 'defu' @@ -11,18 +11,43 @@ export interface ClientOptions { export class Client { private websocket: WebSocket + private eventListeners: Map) => void | Promise>> = new Map() constructor(options: ClientOptions) { const opts = defu, Required>[]>(options, { url: 'ws://localhost:6121/ws', possibleEvents: [] }) this.websocket = new WebSocket(opts.url) - this.send({ - type: 'module:announce', - data: { - name: opts.name, - possibleEvents: opts.possibleEvents, - }, - }) + this.websocket.onmessage = this.handleMessage.bind(this) + this.websocket.onopen = () => { + this.send({ + type: 'module:announce', + data: { + name: opts.name, + possibleEvents: opts.possibleEvents, + }, + }) + } + } + + private async handleMessage(event: any) { + const data = JSON.parse(event.data) as WebSocketEvent + const listeners = this.eventListeners.get(data.type) + if (!listeners) + return + + for (const listener of listeners) { + await listener(data) + } + } + + onEvent(event: E, callback: (data: WebSocketBaseEvent) => void | Promise): void { + if (!this.eventListeners.get(event)) { + this.eventListeners.set(event, []) + } + + const listeners = this.eventListeners.get(event) + listeners.push(callback) + this.eventListeners.set(event, listeners) } send(data: WebSocketEvent): void { diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index beaf1a73d..8f20fa7ef 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -824,6 +824,9 @@ importers: '@xsai/generate-text': specifier: 'catalog:' version: 0.0.27 + '@xsai/generate-transcription': + specifier: ^0.0.28 + version: 0.0.28 '@xsai/providers': specifier: 'catalog:' version: 0.0.27 @@ -4319,6 +4322,9 @@ packages: '@xsai/generate-text@0.0.27': resolution: {integrity: sha512-pkmBvNwL2AxWh0CUKnurbKzXtZpZ+x4VAXqiX8MrbnKxUPfYqUaz8k/BZ2yKvda+Y+9r3KNNFVfTWrNK5ofzwQ==} + '@xsai/generate-transcription@0.0.28': + resolution: {integrity: sha512-ChfJfh269XhFxWp2ARIEBXfOjYTQbq7E6ifva0Hv/0oPnModrDvxX6ZXyfhbzvIQl0ZBenKo4M8cTu6NYhADiw==} + '@xsai/model@0.0.27': resolution: {integrity: sha512-2ckL7bLscS+viib7zi/gFS4M55Q7hi19cfDmSWNOXJ7ELCoFo6n2A2BWTuNXye/K5kfU6/StNYo9kvxGL2CQyQ==} @@ -4331,6 +4337,9 @@ packages: '@xsai/shared@0.0.27': resolution: {integrity: sha512-U+lvjD6HdX3xJq57ASoDXk7z12zKTBwuDb30oRuaCq2+jBxUcMbXTMtWbI1EJL/TDNFtrXBjpLj35/7Q6LXUDQ==} + '@xsai/shared@0.0.28': + resolution: {integrity: sha512-IEVh6NI5dEl+Loxn2I8EoZBjtAQWpwBGGk9niNVCNSsbATJnfIbLOUGL/UNqX7ZNbh3wQid1mTHMyJGMkn8sZQ==} + '@xsai/stream-text@0.0.27': resolution: {integrity: sha512-q23LkBFAyDb5gPZ8LxDSXayt98QtHDgwXwz8t2RyViz/58UD5Ffd+OLTtAJQ5kj0RQJTHRV7mDokcwzt4nTxyA==} @@ -13226,6 +13235,10 @@ snapshots: dependencies: '@xsai/shared-chat': 0.0.27 + '@xsai/generate-transcription@0.0.28': + dependencies: + '@xsai/shared': 0.0.28 + '@xsai/model@0.0.27': dependencies: '@xsai/shared': 0.0.27 @@ -13240,6 +13253,8 @@ snapshots: '@xsai/shared@0.0.27': {} + '@xsai/shared@0.0.28': {} + '@xsai/stream-text@0.0.27': dependencies: '@xsai/shared-chat': 0.0.27 diff --git a/services/discord-bot/.env b/services/discord-bot/.env index b859022d0..4bd35b543 100644 --- a/services/discord-bot/.env +++ b/services/discord-bot/.env @@ -2,8 +2,11 @@ DISCORD_TOKEN='' DISCORD_BOT_CLIENT_ID='' OPENAI_MODEL='' -OPENAI_API_KEY='' OPENAI_API_BASE_URL='' +OPENAI_API_KEY='' + +OPENAI_STT_API_BASE_URL='' +OPENAI_STT_API_KEY='' ELEVENLABS_API_KEY='' ELEVENLABS_API_BASE_URL='' diff --git a/services/discord-bot/package.json b/services/discord-bot/package.json index b25e3727a..123e329bc 100644 --- a/services/discord-bot/package.json +++ b/services/discord-bot/package.json @@ -15,7 +15,7 @@ "directory": "services/discord-bot" }, "scripts": { - "start": "dotenvx run -f .env.local -f .env --ignore=MISSING_ENV_FILE -- tsx src/index.ts" + "start": "dotenvx run -f .env -f .env.local --overload --ignore=MISSING_ENV_FILE -- tsx src/index.ts" }, "dependencies": { "@discordjs/voice": "^0.18.0", @@ -26,6 +26,7 @@ "@proj-airi/server-shared": "workspace:^", "@xsai/generate-speech": "catalog:", "@xsai/generate-text": "catalog:", + "@xsai/generate-transcription": "^0.0.28", "@xsai/providers": "catalog:", "@xsai/shared-chat": "catalog:", "discord.js": "^14.17.3", diff --git a/services/discord-bot/src/bots/discord/commands/summon.ts b/services/discord-bot/src/bots/discord/commands/summon.ts index c2d0f581b..5f5c65afc 100644 --- a/services/discord-bot/src/bots/discord/commands/summon.ts +++ b/services/discord-bot/src/bots/discord/commands/summon.ts @@ -1,151 +1,714 @@ -import type { AudioReceiveStream } from '@discordjs/voice' -import type { useLogg } from '@guiiai/logg' -import type { Client } from '@proj-airi/server-sdk' -import type { CacheType, ChatInputCommandInteraction, GuildMember } from 'discord.js' +import type { AudioPlayer, VoiceConnection } from '@discordjs/voice' +import type { Client as AiriClient } from '@proj-airi/server-sdk' +import type { Discord } from '@proj-airi/server-shared/types' +import type { BaseGuildVoiceChannel, CacheType, ChatInputCommandInteraction, Client as DiscordClient, GuildMember } from 'discord.js' +import type { Readable } from 'node:stream' import { Buffer } from 'node:buffer' -import { Writable } from 'node:stream' -import { createAudioPlayer, EndBehaviorType, entersState, joinVoiceChannel, NoSubscriberBehavior, VoiceConnectionStatus } from '@discordjs/voice' +import { EventEmitter } from 'node:events' +import { pipeline, Transform } from 'node:stream' +import { createAudioPlayer, createAudioResource, entersState, getVoiceConnections, joinVoiceChannel, NoSubscriberBehavior, StreamType, VoiceConnectionStatus } from '@discordjs/voice' +import { useLogg } from '@guiiai/logg' + import OpusScript from 'opusscript' +import { openaiTranscribe } from '../../../pipelines/tts' +import { getWavHeader } from '../../../utils/audio' -import { transcribe } from '../../../pipelines/tts' +class OpusDecoderStream extends Transform { + private decoder: OpusScript -const decoder = new OpusScript(48000, 2) + /** + * @param sampleRate - The audio sample rate (e.g., 16000 Hz) + * @param channels - Number of audio channels (e.g., 1 for mono) + */ + constructor(sampleRate: 8000 | 12000 | 16000 | 24000 | 48000, channels: number) { + super() + this.decoder = new OpusScript(sampleRate, channels) + } -async function transcribeTextFromAudioReceiveStream(stream: AudioReceiveStream) { - return new Promise((resolve, reject) => { + _transform(chunk: Buffer, encoding: BufferEncoding, callback: (...args: any[]) => void) { try { - let pcmBuffer = Buffer.alloc(0) - const pcmStream = new Writable({ - write(chunk, _encoding, callback) { - pcmBuffer = Buffer.concat([pcmBuffer, chunk]) - callback() + // Decode Opus chunk to PCM + const pcm = this.decoder.decode(chunk) + if (pcm) { + this.push(Buffer.from(pcm)) + } + callback() + } + catch (error) { + this.emit('error', error) + callback(error) + } + } + + _flush(callback: (...args: any[]) => void) { + callback() + } +} + +// These values are chosen for compatibility with picovoice components +// const DECODE_FRAME_SIZE = 1024 +const DECODE_SAMPLE_RATE = 16000 + +// eliza/packages/client-discord/src/voice.ts at develop · elizaOS/eliza +// https://github.com/elizaOS/eliza/blob/develop/packages/client-discord/src/voice.ts +export class AudioMonitor { + private readable: Readable + private buffers: Buffer[] = [] + private maxSize: number + private lastFlagged: number = -1 + private ended: boolean = false + private logger = useLogg('AudioMonitor').useGlobalConfig() + + constructor( + readable: Readable, + maxSize: number, + onStart: () => void, + callback: (buffer: Buffer) => void, + ) { + this.readable = readable + this.maxSize = maxSize + this.readable.on('data', (chunk: Buffer) => { + // this.logger.log('AudioMonitor got data'); + if (this.lastFlagged < 0) { + this.lastFlagged = this.buffers.length + } + this.buffers.push(chunk) + const currentSize = this.buffers.reduce( + (acc, cur) => acc + cur.length, + 0, + ) + while (currentSize > this.maxSize) { + this.buffers.shift() + this.lastFlagged-- + } + }) + this.readable.on('end', () => { + this.logger.log('AudioMonitor ended') + this.ended = true + if (this.lastFlagged < 0) + return + callback(this.getBufferFromStart()) + this.lastFlagged = -1 + }) + this.readable.on('speakingStopped', () => { + if (this.ended) + return + this.logger.log('Speaking stopped') + if (this.lastFlagged < 0) + return + callback(this.getBufferFromStart()) + }) + this.readable.on('speakingStarted', () => { + if (this.ended) + return + onStart() + this.logger.log('Speaking started') + this.reset() + }) + } + + stop() { + this.readable.removeAllListeners('data') + this.readable.removeAllListeners('end') + this.readable.removeAllListeners('speakingStopped') + this.readable.removeAllListeners('speakingStarted') + } + + isFlagged() { + return this.lastFlagged >= 0 + } + + getBufferFromFlag() { + if (this.lastFlagged < 0) { + return null + } + const buffer = Buffer.concat(this.buffers.slice(this.lastFlagged)) + return buffer + } + + getBufferFromStart() { + const buffer = Buffer.concat(this.buffers) + return buffer + } + + reset() { + this.buffers = [] + this.lastFlagged = -1 + } + + isEnded() { + return this.ended + } +} + +function isValidTranscription(text: string): boolean { + if (!text || text.includes('[BLANK_AUDIO]')) + return false + return true +} + +// eliza/packages/client-discord/src/voice.ts at develop · elizaOS/eliza +// https://github.com/elizaOS/eliza/blob/develop/packages/client-discord/src/voice.ts + +export class VoiceManager extends EventEmitter { + private logger = useLogg('VoiceManager').useGlobalConfig() + private processingVoice: boolean = false + private transcriptionTimeout: NodeJS.Timeout | null = null + private userStates: Map< + string, + { + buffers: Buffer[] + totalLength: number + lastActive: number + transcriptionText: string + } + > = new Map() + + private activeAudioPlayer: AudioPlayer | null = null + private client: DiscordClient + private airiClient: AiriClient + private streams: Map = new Map() + private connections: Map = new Map() + private activeMonitors: Map< + string, + { channel: BaseGuildVoiceChannel, monitor: AudioMonitor } + > = new Map() + + constructor(client: DiscordClient, airiClient: AiriClient) { + super() + this.client = client + this.airiClient = airiClient + } + + async joinChannel(interaction: ChatInputCommandInteraction, channel: BaseGuildVoiceChannel) { + const oldConnection = this.getVoiceConnection( + channel.guildId as string, + ) + if (oldConnection) { + try { + oldConnection.destroy() + // Remove all associated streams and monitors + this.streams.clear() + this.activeMonitors.clear() + } + catch (error) { + console.error('Error leaving voice channel:', error) + } + } + + const connection = joinVoiceChannel({ + channelId: channel.id, + guildId: channel.guild.id, + adapterCreator: channel.guild.voiceAdapterCreator as any, + selfDeaf: false, + selfMute: false, + group: this.client.user.id, + }) + + try { + // Wait for either Ready or Signalling state + await Promise.race([ + entersState(connection, VoiceConnectionStatus.Ready, 20_000), + entersState(connection, VoiceConnectionStatus.Signalling, 20_000), + ]) + + // Log connection success + this.logger.log( + `Voice connection established in state: ${connection.state.status}`, + ) + + await interaction.reply(`Joined: ${channel.name}.`) + + // Set up ongoing state change monitoring + connection.on('stateChange', async (oldState, newState) => { + this.logger.log( + `Voice connection state changed from ${oldState.status} to ${newState.status}`, + ) + + if (newState.status === VoiceConnectionStatus.Disconnected) { + this.logger.log('Handling disconnection...') + + try { + // Try to reconnect if disconnected + await Promise.race([ + entersState(connection, VoiceConnectionStatus.Signalling, 5_000), + entersState(connection, VoiceConnectionStatus.Connecting, 5_000), + ]) + // Seems to be reconnecting to a new channel + this.logger.log('Reconnecting to channel...') + } + catch (e) { + // Seems to be a real disconnect, destroy and cleanup + this.logger.log(`Disconnection confirmed - cleaning up...${e}`) + connection.destroy() + this.connections.delete(channel.id) + } + } + else if ( + newState.status === VoiceConnectionStatus.Destroyed + ) { + this.connections.delete(channel.id) + } + else if ( + !this.connections.has(channel.id) + && (newState.status === VoiceConnectionStatus.Ready + || newState.status === VoiceConnectionStatus.Signalling) + ) { + this.connections.set(channel.id, connection) + } + }) + + connection.on('error', (error) => { + this.logger.log('Voice connection error:', error) + // Don't immediately destroy - let the state change handler deal with it + this.logger.log('Connection error - will attempt to recover...') + }) + + // Store the connection + this.connections.set(channel.id, connection) + + // Continue with voice state modifications + const me = channel.guild.members.me + if (me?.voice && me.permissions.has('DeafenMembers')) { + try { + await me.voice.setDeaf(false) + await me.voice.setMute(false) + } + catch (error) { + this.logger.log('Failed to modify voice state:', error) + // Continue even if this fails + } + } + + connection.receiver.speaking.on('start', async (userId: string) => { + let user = channel.members.get(userId) + if (!user) { + try { + user = await channel.guild.members.fetch(userId) + } + catch (error) { + console.error('Failed to fetch user:', error) + } + } + if (user && !user?.user.bot) { + this.logger.log(`User speaking: ${user.displayName}`) + this.monitorMember(user as GuildMember, channel.id) + this.streams.get(userId)?.emit('speakingStarted') + } + }) + + connection.receiver.speaking.on('end', async (userId: string) => { + const user = channel.members.get(userId) + if (!user?.user.bot) { + this.logger.log(`User stopped speaking: ${user.displayName}`) + this.streams.get(userId)?.emit('speakingStopped') + } + }) + } + catch (error) { + this.logger.log('Failed to establish voice connection:', error) + connection.destroy() + this.connections.delete(channel.id) + throw error + } + } + + private getVoiceConnection(guildId: string) { + const connections = getVoiceConnections(this.client.user.id) + if (!connections) { + this.logger.warn('No voice connections found') + return + } + const connection = [...connections.values()].find( + connection => connection.joinConfig.guildId === guildId, + ) + if (!connection) { + this.logger.warn('No voice connection found for guild') + } + + return connection + } + + private async monitorMember( + member: GuildMember, + channelId: string, + ) { + const userId = member?.id + const connection = this.getVoiceConnection(member?.guild?.id) + const receiveStream = connection?.receiver.subscribe(userId, { + autoDestroy: true, + emitClose: true, + }) + if (!receiveStream) { + this.logger.warn('No voice data received') + return + } + + const opusDecoder = new OpusDecoderStream(DECODE_SAMPLE_RATE, 1) + const volumeBuffer: number[] = [] + const VOLUME_WINDOW_SIZE = 30 + const SPEAKING_THRESHOLD = 0.05 + opusDecoder.on('data', (pcmData: Buffer) => { + // Monitor the audio volume while the agent is speaking. + // If the average volume of the user's audio exceeds the defined threshold, it indicates active speaking. + // When active speaking is detected, stop the agent's current audio playback to avoid overlap. + + if (this.activeAudioPlayer) { + const samples = new Int16Array( + pcmData.buffer, + pcmData.byteOffset, + pcmData.length / 2, + ) + const maxAmplitude = Math.max(...samples.map(Math.abs)) / 32768 + volumeBuffer.push(maxAmplitude) + + if (volumeBuffer.length > VOLUME_WINDOW_SIZE) { + volumeBuffer.shift() + } + const avgVolume + = volumeBuffer.reduce((sum, v) => sum + v, 0) + / VOLUME_WINDOW_SIZE + + if (avgVolume > SPEAKING_THRESHOLD) { + volumeBuffer.length = 0 + this.cleanupAudioPlayer(this.activeAudioPlayer) + this.processingVoice = false + } + } + }) + + pipeline(receiveStream, opusDecoder, (err) => { + this.logger.withError(err).error('Opus decoding pipeline error') + }) + + this.streams.set(userId, opusDecoder) + this.connections.set(userId, connection as VoiceConnection) + opusDecoder.on('error', (err: any) => { + this.logger.log(`Opus decoding error: ${err}`) + }) + const errorHandler = (err: any) => { + this.logger.log(`Opus decoding error: ${err}`) + } + const streamCloseHandler = () => { + this.logger.log(`voice stream from ${member?.displayName} closed`) + this.streams.delete(userId) + this.connections.delete(userId) + } + const closeHandler = () => { + this.logger.log(`Opus decoder for ${member?.displayName} closed`) + opusDecoder.removeListener('error', errorHandler) + opusDecoder.removeListener('close', closeHandler) + receiveStream?.removeListener('close', streamCloseHandler) + } + opusDecoder.on('error', errorHandler) + opusDecoder.on('close', closeHandler) + receiveStream?.on('close', streamCloseHandler) + + this.logger.log(`Monitoring user: ${member.displayName}`) + await this.handleUserStream(userId, member.displayName, member.nickname, member.guild.id, channelId, opusDecoder) + } + + leaveChannel(channel: BaseGuildVoiceChannel) { + const connection = this.connections.get(channel.id) + if (connection) { + connection.destroy() + this.connections.delete(channel.id) + } + + // Stop monitoring all members in this channel + for (const [memberId, monitorInfo] of this.activeMonitors) { + if ( + monitorInfo.channel.id === channel.id + && memberId !== this.client.user?.id + ) { + this.stopMonitoringMember(memberId) + } + } + + this.logger.log(`Left voice channel: ${channel.name} (${channel.id})`) + } + + stopMonitoringMember(memberId: string) { + const monitorInfo = this.activeMonitors.get(memberId) + if (monitorInfo) { + monitorInfo.monitor.stop() + this.activeMonitors.delete(memberId) + this.streams.delete(memberId) + this.logger.log(`Stopped monitoring user ${memberId}`) + } + } + + async debouncedProcessTranscription( + userId: string, + displayName: string, + nickname: string, + guildId: string, + channelId: string, + ) { + const DEBOUNCE_TRANSCRIPTION_THRESHOLD = 1500 // wait for 1.5 seconds of silence + + if (this.activeAudioPlayer?.state?.status === 'idle') { + this.logger.log('Cleaning up idle audio player.') + this.cleanupAudioPlayer(this.activeAudioPlayer) + } + + if (this.activeAudioPlayer || this.processingVoice) { + const state = this.userStates.get(userId) + state.buffers.length = 0 + state.totalLength = 0 + return + } + + if (this.transcriptionTimeout) { + clearTimeout(this.transcriptionTimeout) + } + + this.transcriptionTimeout = setTimeout(async () => { + this.processingVoice = true + try { + await this.processTranscription( + userId, + displayName, + nickname, + guildId, + channelId, + ) + + // Clean all users' previous buffers + this.userStates.forEach((state, _) => { + state.buffers.length = 0 + state.totalLength = 0 + }) + } + finally { + this.processingVoice = false + } + }, DEBOUNCE_TRANSCRIPTION_THRESHOLD) + } + + private async handleUserStream( + userId: string, + displayName: string, + nickname: string, + guildId: string, + channelId: string, + audioStream: Readable, + ) { + this.logger.log(`Starting audio monitor for user: ${userId}`) + if (!this.userStates.has(userId)) { + this.userStates.set(userId, { + buffers: [], + totalLength: 0, + lastActive: Date.now(), + transcriptionText: '', + }) + } + + const state = this.userStates.get(userId) + + const processBuffer = async (buffer: Buffer) => { + try { + state!.buffers.push(buffer) + state!.totalLength += buffer.length + state!.lastActive = Date.now() + this.debouncedProcessTranscription( + userId, + displayName, + nickname, + guildId, + channelId, + ) + } + catch (error) { + console.error( + `Error processing buffer for user ${userId}:`, + error, + ) + } + } + + const _ = new AudioMonitor( + audioStream, + 10000000, + () => { + if (this.transcriptionTimeout) { + clearTimeout(this.transcriptionTimeout) + } + }, + async (buffer) => { + if (!buffer) { + console.error('Received empty buffer') + return + } + await processBuffer(buffer) + }, + ) + } + + private async processTranscription( + userId: string, + displayName: string, + nickname: string, + guildId: string, + channelId: string, + ) { + const state = this.userStates.get(userId) + if (!state || state.buffers.length === 0) + return + try { + const inputBuffer = Buffer.concat(state.buffers, state.totalLength) + + state.buffers.length = 0 // Clear the buffers + state.totalLength = 0 + // Convert Opus to WAV + const wavBuffer = await this.convertOpusToWav(inputBuffer) + + const result = await openaiTranscribe(wavBuffer) + const transcriptionText = result + + const discordContext = { + channelId, + guildId, + guildMember: { + id: userId, + nickname, + displayName, + }, + } satisfies Discord + + this.airiClient.send({ + type: 'input:text:voice', + data: { + transcription: transcriptionText, + discord: discordContext, }, }) - stream.on('error', (err) => { - reject(err) + this.airiClient.send({ + type: 'input:text', + data: { + text: transcriptionText, + discord: discordContext, + }, }) - // Create the pipeline - stream.on('data', async (chunk) => { - try { - const pcm = decoder.decode(chunk) - pcmStream.write(pcm) - } - catch (err) { - reject(err) - } - }) + if (transcriptionText && isValidTranscription(transcriptionText)) { + state.transcriptionText += transcriptionText + } - // When user stops talking, stop the stream and generate an mp3 file. - stream.on('end', async () => { - try { - pcmStream.end() - - const result = await transcribe(pcmBuffer) - resolve(result) - } - catch (err) { - reject(err) - } - }) + if (state.transcriptionText.length) { + this.cleanupAudioPlayer(this.activeAudioPlayer) + const finalText = state.transcriptionText + state.transcriptionText = '' + this.logger.withField('transcription', finalText).log('Transcription complete') + } } - catch (err) { - reject(err) + catch (error) { + console.error( + `Error transcribing audio for user ${userId}:`, + error, + ) } - }) -} - -export async function handleSummon(log: ReturnType, interaction: ChatInputCommandInteraction, airiClient: Client) { - const currVoiceChannel = (interaction.member as GuildMember).voice.channel - if (!currVoiceChannel) { - return await interaction.reply('Please join a voice channel first.') } - try { - const connection = joinVoiceChannel({ - channelId: currVoiceChannel.id, - guildId: interaction.guild.id, - adapterCreator: interaction.guild.voiceAdapterCreator, - }) + private async convertOpusToWav(pcmBuffer: Buffer): Promise { + try { + // Generate the WAV header + const wavHeader = getWavHeader( + pcmBuffer.length, + DECODE_SAMPLE_RATE, + ) - const player = createAudioPlayer({ + // Concatenate the WAV header and PCM data + const wavBuffer = Buffer.concat([wavHeader, pcmBuffer]) + + return wavBuffer + } + catch (error) { + console.error('Error converting PCM to WAV:', error) + throw error + } + } + + async playAudioStream(userId: string, audioStream: Readable) { + const connection = this.connections.get(userId) + if (connection == null) { + this.logger.log(`No connection for user ${userId}`) + return + } + this.cleanupAudioPlayer(this.activeAudioPlayer) + const audioPlayer = createAudioPlayer({ behaviors: { noSubscriber: NoSubscriberBehavior.Pause, }, }) + this.activeAudioPlayer = audioPlayer + connection.subscribe(audioPlayer) - connection.subscribe(player) + const audioStartTime = Date.now() - connection.on(VoiceConnectionStatus.Signalling, async () => { - log.log('Connection is signalling') + const resource = createAudioResource(audioStream, { + inputType: StreamType.Arbitrary, + }) + audioPlayer.play(resource) + + audioPlayer.on('error', (err: any) => { + this.logger.log(`Audio player error: ${err}`) }) - connection.on(VoiceConnectionStatus.Connecting, async () => { - log.log('Connection is connecting') - }) - - connection.on(VoiceConnectionStatus.Ready, async () => { - await interaction.reply(`Joined: ${currVoiceChannel.name}.`) - }) - - connection.on(VoiceConnectionStatus.Disconnected, async (_oldState, _newState) => { - try { - await Promise.race([ - entersState(connection, VoiceConnectionStatus.Signalling, 5_000), - entersState(connection, VoiceConnectionStatus.Connecting, 5_000), - ]) - // Seems to be reconnecting to a new channel - ignore disconnect - } - catch (error) { - log.withError(error).log('Failed to reconnect to channel') - // Seems to be a real disconnect which SHOULDN'T be recovered from - connection.destroy() - } - }) - - connection.on(VoiceConnectionStatus.Destroyed, async () => { - log.log('Destroyed connection') - }) - - connection.receiver.speaking.on('start', async (userId) => { - log.log(`User ${userId} started speaking`) - - try { - const listenStream = connection.receiver.subscribe(userId, { - end: { - behavior: EndBehaviorType.AfterSilence, - duration: 2000, // Max 2s of silence before ending the stream. - }, - }) - - const speakingUser = await interaction.guild.members.fetch(userId) - const result = await transcribeTextFromAudioReceiveStream(listenStream) - - airiClient.send({ type: 'input:text:voice', data: { - transcription: result, - discord: { - guildId: interaction.guild.id, - channelId: currVoiceChannel.id, - guildMember: { - id: userId, - nickname: speakingUser.nickname, - displayName: speakingUser.displayName, - }, - }, - } }) - } - catch (err) { - log.withError(err).log('Error handling user speaking') - } - }) - - connection.receiver.speaking.on('end', (userId) => { - log.log(`User ${userId} stopped speaking`) - }) + audioPlayer.on( + 'stateChange', + (_oldState: any, newState: { status: string }) => { + if (newState.status === 'idle') { + const idleTime = Date.now() + this.logger.log( + `Audio playback took: ${idleTime - audioStartTime}ms`, + ) + } + }, + ) } - catch (error) { - log.error(error) - await interaction.reply('Could not join voice channel.') + + cleanupAudioPlayer(audioPlayer: AudioPlayer) { + if (!audioPlayer) + return + + audioPlayer.stop() + audioPlayer.removeAllListeners() + if (audioPlayer === this.activeAudioPlayer) { + this.activeAudioPlayer = null + } + } + + async handleJoinChannelCommand(interaction: ChatInputCommandInteraction) { + try { + const currVoiceChannel = (interaction.member as GuildMember).voice.channel + if (!currVoiceChannel) { + return await interaction.reply('Please join a voice channel first.') + } + + await this.joinChannel(interaction, currVoiceChannel) + } + catch (error) { + this.logger.withError(error).log('Error joining voice channel') + } + } + + async handleLeaveChannelCommand(interaction: any) { + const connection = this.getVoiceConnection(interaction.guildId as any) + + if (!connection) { + await interaction.reply('Not currently in a voice channel.') + return + } + + try { + connection.destroy() + await interaction.reply('Left the voice channel.') + } + catch (error) { + this.logger.withError(error).log('Error leaving voice channel') + await interaction.reply('Failed to leave the voice channel.') + } } } diff --git a/services/discord-bot/src/index.ts b/services/discord-bot/src/index.ts index 64c8dfc64..8d12fee17 100644 --- a/services/discord-bot/src/index.ts +++ b/services/discord-bot/src/index.ts @@ -3,10 +3,7 @@ import { Format, LogLevel, setGlobalFormat, setGlobalLogLevel, useLogg } from '@ import { Client as AiriClient } from '@proj-airi/server-sdk' import { Client, Events, GatewayIntentBits } from 'discord.js' -import { handlePing, handleSummon, registerCommands } from './bots/discord/commands' -import { WhisperLargeV3Pipeline } from './pipelines/tts' - -import 'dotenv/config' +import { handlePing, registerCommands, VoiceManager } from './bots/discord/commands' setGlobalFormat(Format.Pretty) setGlobalLogLevel(LogLevel.Log) @@ -14,10 +11,11 @@ const log = useLogg('Bot').useGlobalConfig() // Create a new client instance async function main() { - await WhisperLargeV3Pipeline.getInstance() + // await WhisperLargeV3Pipeline.getInstance() const client = new Client({ intents: [GatewayIntentBits.Guilds, GatewayIntentBits.GuildVoiceStates] }) - const airiClient = new AiriClient({ name: 'discord-voice-bot', possibleEvents: ['input:text', 'input:text:voice', 'input:voice'] }) + const airiClient = new AiriClient({ name: 'discord-bot', possibleEvents: ['input:text', 'input:text:voice', 'input:voice'] }) + const voiceManager = new VoiceManager(client, airiClient) // When the client is ready, run this code (only once). // The distinction between `client: Client` and `readyClient: Client` is important for TypeScript developers. @@ -37,7 +35,7 @@ async function main() { await handlePing(interaction) break case 'summon': - await handleSummon(log, interaction, airiClient) + await voiceManager.handleJoinChannelCommand(interaction) break } }) @@ -47,4 +45,4 @@ async function main() { await client.login(env.DISCORD_TOKEN) } -main().catch(log.error) +main().catch(err => log.withError(err).error('An error occurred')) diff --git a/services/discord-bot/src/pipelines/tts.ts b/services/discord-bot/src/pipelines/tts.ts index b545f8f68..14a63e761 100644 --- a/services/discord-bot/src/pipelines/tts.ts +++ b/services/discord-bot/src/pipelines/tts.ts @@ -1,20 +1,20 @@ import type { Buffer } from 'node:buffer' +import { env } from 'node:process' import { useLogg } from '@guiiai/logg' import { pipeline, type PipelineType } from '@huggingface/transformers' +import { generateTranscription } from '@xsai/generate-transcription' +import { createOpenAI } from '@xsai/providers' import wavefile from 'wavefile' import { pcmToWav } from '../utils/audio' export class WhisperLargeV3Pipeline { static task: PipelineType = 'automatic-speech-recognition' - static model = 'Xenova/whisper-tiny.en' + static model = 'Xenova/whisper-medium.en' static instance = null static async getInstance(progress_callback = null) { if (this.instance === null) { - // NOTE: Uncomment this to change the cache directory - // env.cacheDir = './.cache'; - this.instance = await pipeline(this.task, this.model, { progress_callback }) } @@ -42,7 +42,7 @@ export function textFromResult(result: Array<{ text: string }> | { text: string } export async function transcribe(pcmBuffer: Buffer) { - const log = useLogg('Transcribe').useGlobalConfig() + const log = useLogg('Memory:Transcribe').useGlobalConfig() const pcmConvertedWav = pcmToWav(pcmBuffer, 48000, 2) log.withFields({ from: pcmBuffer.byteLength, to: pcmConvertedWav.byteLength }).log('Audio data received') @@ -65,3 +65,30 @@ export async function transcribe(pcmBuffer: Buffer) { log.withField('result', text).log('Transcription result') return text } + +export async function openaiTranscribe(wavBuffer: Buffer) { + const log = useLogg('Remote:Transcribe').useGlobalConfig() + + log.log('Transcribing audio...') + + const wavFile = new Blob([wavBuffer], { type: 'audio/wav' }) + const openai = createOpenAI({ + baseURL: env.OPENAI_STT_API_BASE_URL, + apiKey: env.OPENAI_STT_API_KEY, + }) + + try { + const result = await generateTranscription({ + ...openai.transcription('whisper-1'), + file: wavFile, + }) + + log.withField('result', result.text).log('Transcription result') + return result.text + } + catch (err) { + log.withError(err).error('Failed to transcribe audio') + } + + return '' +} diff --git a/services/discord-bot/src/utils/audio.ts b/services/discord-bot/src/utils/audio.ts index 5303e2eb0..4e62611f5 100644 --- a/services/discord-bot/src/utils/audio.ts +++ b/services/discord-bot/src/utils/audio.ts @@ -1,4 +1,4 @@ -import type { Buffer } from 'node:buffer' +import { Buffer } from 'node:buffer' export function pcmToWav(pcmBuffer: Buffer, sampleRate: number, numChannels: number): Uint8Array { const byteRate = sampleRate * numChannels * 2 // Assuming 16-bit PCM (2 bytes per sample) @@ -49,3 +49,29 @@ function writeString(view, offset, string) { view.setUint8(offset + i, string.charCodeAt(i)) } } + +export function getWavHeader( + audioLength: number, + sampleRate: number, + channelCount: number = 1, + bitsPerSample: number = 16, +): Buffer { + const wavHeader = Buffer.alloc(44) + wavHeader.write('RIFF', 0) + wavHeader.writeUInt32LE(36 + audioLength, 4) // Length of entire file in bytes minus 8 + wavHeader.write('WAVE', 8) + wavHeader.write('fmt ', 12) + wavHeader.writeUInt32LE(16, 16) // Length of format data + wavHeader.writeUInt16LE(1, 20) // Type of format (1 is PCM) + wavHeader.writeUInt16LE(channelCount, 22) // Number of channels + wavHeader.writeUInt32LE(sampleRate, 24) // Sample rate + wavHeader.writeUInt32LE( + (sampleRate * bitsPerSample * channelCount) / 8, + 28, + ) // Byte rate + wavHeader.writeUInt16LE((bitsPerSample * channelCount) / 8, 32) // Block align ((BitsPerSample * Channels) / 8) + wavHeader.writeUInt16LE(bitsPerSample, 34) // Bits per sample + wavHeader.write('data', 36) // Data chunk header + wavHeader.writeUInt32LE(audioLength, 40) // Data chunk size + return wavHeader +} diff --git a/services/telegram-bot/.env b/services/telegram-bot/.env index f06b0c0e0..f0ddde766 100644 --- a/services/telegram-bot/.env +++ b/services/telegram-bot/.env @@ -1,4 +1,5 @@ DATABASE_URL=postgres://postgres:123456@localhost:5432/postgres TELEGRAM_BOT_TOKEN='' -OPENAI_API_KEY='' + OPENAI_API_BASE_URL='' +OPENAI_API_KEY='' diff --git a/services/telegram-bot/package.json b/services/telegram-bot/package.json index 8f4d90ad7..266588c26 100644 --- a/services/telegram-bot/package.json +++ b/services/telegram-bot/package.json @@ -15,7 +15,7 @@ "directory": "services/telegram-bot" }, "scripts": { - "start": "dotenvx run -f .env.local -f .env --ignore=MISSING_ENV_FILE -- tsx src/index.ts" + "start": "dotenvx run -f .env -f .env.local --overload --ignore=MISSING_ENV_FILE -- tsx src/index.ts" }, "dependencies": { "@dotenvx/dotenvx": "^1.33.0",