Node.js环境集成Fish-Speech-1.5语音服务API开发最近在做一个智能客服项目需要把文字实时转成语音播报给用户。试了几个方案要么声音太机械要么延迟太高要么就是成本扛不住。后来发现了Fish-Speech-1.5这个开源模型试了一下效果确实不错声音自然支持的语言也多。但问题来了我们后端用的是Node.js而Fish-Speech是基于Python的。总不能每次生成语音都去调Python脚本吧那延迟和并发都成问题。所以我就琢磨着怎么在Node.js环境里把它集成起来做成一个高可用的语音服务API。今天就跟大家分享一下我的实践过程从环境搭建到API设计再到性能优化一步步带你构建一个能扛住高并发的语音服务。1. 项目环境准备与模型部署在开始集成之前我们需要先把基础环境搭好。这里假设你已经有了Node.js的开发环境我们重点说说Fish-Speech的部署。1.1 Fish-Speech模型部署Fish-Speech提供了几种部署方式对于生产环境我推荐使用Docker容器化部署这样隔离性好也方便扩展。首先你需要准备一台有GPU的服务器至少4GB显存然后拉取官方镜像# 拉取Fish-Speech官方镜像 docker pull fishaudio/fish-speech:1.5 # 运行容器 docker run -d \ --gpus all \ -p 7860:7860 \ -p 8000:8000 \ --name fish-speech \ fishaudio/fish-speech:1.5这里开了两个端口7860是WebUI界面8000是API服务端口。启动后你可以通过http://你的服务器IP:7860访问Web界面测试模型是否正常工作。如果不想用Docker也可以直接本地安装。不过对于Node.js集成来说我更建议用API服务的方式这样Node.js应用和模型服务可以分开部署互不影响。1.2 Node.js项目初始化创建一个新的Node.js项目安装必要的依赖mkdir fish-speech-api cd fish-speech-api npm init -y # 安装核心依赖 npm install express axios multer npm install -D typescript types/node types/express ts-node-dev # 初始化TypeScript配置 npx tsc --init我们的项目结构大概长这样fish-speech-api/ ├── src/ │ ├── app.ts # Express应用入口 │ ├── routes/ # 路由定义 │ ├── controllers/ # 控制器 │ ├── services/ # 业务逻辑 │ └── utils/ # 工具函数 ├── uploads/ # 上传文件临时目录 ├── package.json └── tsconfig.json2. 基础API接口设计与实现有了基础环境我们先来实现最核心的文本转语音功能。Fish-Speech提供了HTTP API我们可以直接调用。2.1 封装Fish-Speech客户端首先创建一个服务类专门负责和Fish-Speech模型服务通信// src/services/fishSpeechService.ts import axios, { AxiosInstance } from axios; import FormData from form-data; import fs from fs; import path from path; export interface TTSRequest { text: string; language?: string; speaker?: string; referenceAudio?: Buffer; referenceText?: string; emotion?: string; speed?: number; } export interface TTSResponse { success: boolean; audioUrl?: string; audioData?: Buffer; duration?: number; error?: string; } export class FishSpeechService { private client: AxiosInstance; private baseUrl: string; constructor(baseUrl: string http://localhost:8000) { this.baseUrl baseUrl; this.client axios.create({ baseURL: baseUrl, timeout: 30000, // 30秒超时 }); } /** * 基础文本转语音 */ async textToSpeech(request: TTSRequest): PromiseTTSResponse { try { const formData new FormData(); formData.append(text, request.text); if (request.language) { formData.append(language, request.language); } if (request.speaker) { formData.append(speaker, request.speaker); } if (request.emotion) { formData.append(emotion, request.emotion); } if (request.speed) { formData.append(speed, request.speed.toString()); } // 如果有参考音频 if (request.referenceAudio request.referenceText) { formData.append(reference_audio, request.referenceAudio, { filename: reference.wav, contentType: audio/wav }); formData.append(reference_text, request.referenceText); } const response await this.client.post(/api/tts, formData, { headers: { ...formData.getHeaders(), }, responseType: arraybuffer, // 接收二进制音频数据 }); return { success: true, audioData: Buffer.from(response.data), duration: this.calculateAudioDuration(response.data), }; } catch (error: any) { console.error(TTS请求失败:, error.message); return { success: false, error: error.response?.data?.message || error.message, }; } } /** * 批量文本转语音 */ async batchTextToSpeech(requests: TTSRequest[]): PromiseTTSResponse[] { const promises requests.map(request this.textToSpeech(request)); return Promise.all(promises); } /** * 估算音频时长简化版 */ private calculateAudioDuration(audioBuffer: ArrayBuffer): number { // 简单估算假设16kHz采样率16位单声道 // 实际应该解析WAV头获取准确信息 const bytesPerSecond 16000 * 2; // 16kHz * 2 bytes return audioBuffer.byteLength / bytesPerSecond; } /** * 检查服务健康状态 */ async healthCheck(): Promiseboolean { try { const response await this.client.get(/health); return response.status 200; } catch { return false; } } }2.2 实现Express API接口有了服务类我们来创建API接口// src/controllers/ttsController.ts import { Request, Response } from express; import { FishSpeechService, TTSRequest } from ../services/fishSpeechService; const fishSpeechService new FishSpeechService(process.env.FISH_SPEECH_URL || http://localhost:8000); export class TTSController { /** * 单次文本转语音 */ static async generateSpeech(req: Request, res: Response) { try { const { text, language, speaker, emotion, speed } req.body; if (!text) { return res.status(400).json({ error: text字段不能为空 }); } const request: TTSRequest { text, language: language || zh, speaker, emotion, speed: speed ? parseFloat(speed) : undefined, }; // 处理参考音频如果有上传 if (req.file req.body.referenceText) { request.referenceAudio req.file.buffer; request.referenceText req.body.referenceText; } const result await fishSpeechService.textToSpeech(request); if (!result.success) { return res.status(500).json({ error: result.error }); } // 返回音频文件 res.set({ Content-Type: audio/wav, Content-Length: result.audioData!.length, X-Audio-Duration: result.duration, }); res.send(result.audioData); } catch (error: any) { console.error(生成语音失败:, error); res.status(500).json({ error: 内部服务器错误 }); } } /** * 批量生成语音 */ static async batchGenerateSpeech(req: Request, res: Response) { try { const requests req.body.requests; if (!Array.isArray(requests) || requests.length 0) { return res.status(400).json({ error: requests必须是非空数组 }); } // 限制批量请求数量 if (requests.length 10) { return res.status(400).json({ error: 单次批量请求不能超过10个 }); } const results await fishSpeechService.batchTextToSpeech(requests); // 返回ZIP包包含所有音频 // 这里简化处理实际应该打包成ZIP res.json({ success: true, count: results.length, results: results.map((result, index) ({ index, success: result.success, duration: result.duration, error: result.error, })), }); } catch (error: any) { console.error(批量生成失败:, error); res.status(500).json({ error: 内部服务器错误 }); } } /** * 服务健康检查 */ static async healthCheck(req: Request, res: Response) { const isHealthy await fishSpeechService.healthCheck(); res.json({ service: fish-speech-api, status: isHealthy ? healthy : unhealthy, timestamp: new Date().toISOString(), }); } }2.3 配置路由和中间件// src/routes/ttsRoutes.ts import { Router } from express; import multer from multer; import { TTSController } from ../controllers/ttsController; const router Router(); const upload multer({ storage: multer.memoryStorage(), limits: { fileSize: 10 * 1024 * 1024, // 10MB限制 }, }); // 单次TTS生成 router.post(/tts, upload.single(referenceAudio), TTSController.generateSpeech); // 批量TTS生成 router.post(/tts/batch, TTSController.batchGenerateSpeech); // 健康检查 router.get(/health, TTSController.healthCheck); export default router;// src/app.ts import express from express; import cors from cors; import helmet from helmet; import rateLimit from express-rate-limit; import ttsRoutes from ./routes/ttsRoutes; const app express(); const PORT process.env.PORT || 3000; // 安全中间件 app.use(helmet()); app.use(cors()); // 请求体解析 app.use(express.json({ limit: 10mb })); app.use(express.urlencoded({ extended: true })); // 速率限制 const limiter rateLimit({ windowMs: 15 * 60 * 1000, // 15分钟 max: 100, // 每个IP限制100次请求 message: 请求过于频繁请稍后再试, }); app.use(/api/, limiter); // 路由 app.use(/api/tts, ttsRoutes); // 错误处理中间件 app.use((err: any, req: express.Request, res: express.Response, next: express.NextFunction) { console.error(err.stack); res.status(500).json({ error: 服务器内部错误 }); }); // 启动服务器 app.listen(PORT, () { console.log(语音服务API运行在 http://localhost:${PORT}); console.log(API文档: http://localhost:${PORT}/api-docs); });现在基础API就完成了。启动服务后你可以用curl测试一下# 启动服务 npm run dev # 测试TTS接口 curl -X POST http://localhost:3000/api/tts \ -H Content-Type: application/json \ -d { text: 你好欢迎使用Fish-Speech语音服务, language: zh, emotion: 高兴 } \ --output output.wav3. 高并发与性能优化基础功能跑通后接下来要解决性能问题。在实际项目中语音服务可能会面临高并发请求我们需要做一些优化。3.1 异步任务队列对于TTS这种比较耗时的操作直接同步处理会阻塞请求。我们可以引入任务队列把生成任务放到后台处理。// src/services/queueService.ts import Queue from bull; import { TTSRequest, TTSResponse } from ./fishSpeechService; import { FishSpeechService } from ./fishSpeechService; export class QueueService { private ttsQueue: Queue.Queue; private fishSpeechService: FishSpeechService; constructor() { // 使用Redis作为队列存储 this.ttsQueue new Queue(tts-queue, { redis: { host: process.env.REDIS_HOST || localhost, port: parseInt(process.env.REDIS_PORT || 6379), }, defaultJobOptions: { attempts: 3, // 重试3次 backoff: { type: exponential, delay: 1000, // 1秒后重试 }, timeout: 30000, // 30秒超时 }, }); this.fishSpeechService new FishSpeechService(); // 设置队列处理器 this.setupProcessors(); } private setupProcessors() { // TTS任务处理器 this.ttsQueue.process(generate, async (job) { const { request, jobId } job.data; console.log(开始处理TTS任务: ${jobId}); try { const result await this.fishSpeechService.textToSpeech(request); if (!result.success) { throw new Error(result.error || TTS生成失败); } return { success: true, audioData: result.audioData?.toString(base64), // 转base64便于存储 duration: result.duration, jobId, }; } catch (error: any) { console.error(TTS任务失败: ${jobId}, error); throw error; } }); // 批量任务处理器 this.ttsQueue.process(batch, 5, async (job) { // 5个并发 const { requests, jobId } job.data; console.log(开始处理批量TTS任务: ${jobId}, 数量: ${requests.length}); const results []; for (let i 0; i requests.length; i) { try { const result await this.fishSpeechService.textToSpeech(requests[i]); results.push({ index: i, success: result.success, audioData: result.audioData?.toString(base64), duration: result.duration, error: result.error, }); } catch (error: any) { results.push({ index: i, success: false, error: error.message, }); } } return { success: true, results, jobId }; }); } /** * 添加TTS任务到队列 */ async addTTSTask(request: TTSRequest): Promisestring { const jobId tts_${Date.now()}_${Math.random().toString(36).substr(2, 9)}; const job await this.ttsQueue.add(generate, { request, jobId, }, { jobId, // 使用自定义jobId便于追踪 }); return job.id as string; } /** * 添加批量TTS任务 */ async addBatchTTSTask(requests: TTSRequest[]): Promisestring { const jobId batch_${Date.now()}_${Math.random().toString(36).substr(2, 9)}; const job await this.ttsQueue.add(batch, { requests, jobId, }, { jobId, }); return job.id as string; } /** * 获取任务状态 */ async getJobStatus(jobId: string): Promiseany { const job await this.ttsQueue.getJob(jobId); if (!job) { return { status: not_found }; } const state await job.getState(); const result await job.finished().catch(() null); return { jobId, status: state, progress: job.progress(), result: result || null, failedReason: job.failedReason, }; } /** * 清理已完成的任务 */ async cleanupCompletedJobs(days: number 7): Promisevoid { const completed await this.ttsQueue.getCompleted(); const failed await this.ttsQueue.getFailed(); const cutoff Date.now() - days * 24 * 60 * 60 * 1000; for (const job of [...completed, ...failed]) { if (job.timestamp cutoff) { await job.remove(); } } } }3.2 流式响应支持对于长文本我们可以支持流式响应让客户端边生成边接收// src/controllers/streamingController.ts import { Request, Response } from express; import { Transform } from stream; import { FishSpeechService } from ../services/fishSpeechService; export class StreamingController { private fishSpeechService: FishSpeechService; constructor() { this.fishSpeechService new FishSpeechService(); } /** * 流式TTS生成 */ static async streamTTS(req: Request, res: Response) { const { text, language, chunkSize 100 } req.body; if (!text) { return res.status(400).json({ error: text字段不能为空 }); } // 设置流式响应头 res.set({ Content-Type: audio/wav, Transfer-Encoding: chunked, Cache-Control: no-cache, }); // 将文本分割成块 const chunks this.splitTextIntoChunks(text, chunkSize); // 创建转换流将文本块转换为音频 const audioStream new Transform({ transform: async (chunk, encoding, callback) { try { const textChunk chunk.toString(); const result await this.fishSpeechService.textToSpeech({ text: textChunk, language: language || zh, }); if (result.success result.audioData) { this.push(result.audioData); } } catch (error) { console.error(流式生成失败:, error); } callback(); }, }); // 管道传输 audioStream.pipe(res); // 发送文本块 for (const chunk of chunks) { audioStream.write(chunk); } audioStream.end(); } /** * 按标点分割文本 */ private static splitTextIntoChunks(text: string, chunkSize: number): string[] { const sentences text.split(/[。.!?]/).filter(s s.trim()); const chunks: string[] []; let currentChunk ; for (const sentence of sentences) { if ((currentChunk sentence).length chunkSize currentChunk) { chunks.push(currentChunk); currentChunk sentence; } else { currentChunk sentence; } } if (currentChunk) { chunks.push(currentChunk); } return chunks; } }3.3 负载均衡与水平扩展当单个Fish-Speech实例无法满足并发需求时我们需要部署多个实例并通过负载均衡分发请求// src/services/loadBalancer.ts import { FishSpeechService, TTSRequest, TTSResponse } from ./fishSpeechService; interface ServiceInstance { id: string; url: string; healthy: boolean; lastHealthCheck: Date; concurrentRequests: number; maxConcurrent: number; } export class LoadBalancer { private instances: ServiceInstance[] []; private healthCheckInterval: NodeJS.Timeout | null null; constructor(instanceUrls: string[]) { this.instances instanceUrls.map((url, index) ({ id: instance_${index}, url, healthy: true, lastHealthCheck: new Date(), concurrentRequests: 0, maxConcurrent: 5, // 每个实例最大并发数 })); // 启动健康检查 this.startHealthCheck(); } /** * 选择最合适的实例 */ private selectInstance(): ServiceInstance | null { // 过滤健康且未满负荷的实例 const availableInstances this.instances.filter( instance instance.healthy instance.concurrentRequests instance.maxConcurrent ); if (availableInstances.length 0) { return null; } // 使用最少连接算法 return availableInstances.reduce((prev, current) prev.concurrentRequests current.concurrentRequests ? prev : current ); } /** * 执行TTS请求 */ async textToSpeech(request: TTSRequest): PromiseTTSResponse { const instance this.selectInstance(); if (!instance) { return { success: false, error: 所有服务实例都不可用或已达到最大并发限制, }; } // 增加并发计数 instance.concurrentRequests; try { const service new FishSpeechService(instance.url); const result await service.textToSpeech(request); return result; } catch (error: any) { // 标记实例为不健康 instance.healthy false; console.error(实例 ${instance.id} 请求失败:, error.message); return { success: false, error: error.message, }; } finally { // 减少并发计数 instance.concurrentRequests Math.max(0, instance.concurrentRequests - 1); } } /** * 健康检查 */ private async checkInstanceHealth(instance: ServiceInstance): Promisevoid { try { const service new FishSpeechService(instance.url); const isHealthy await service.healthCheck(); instance.healthy isHealthy; instance.lastHealthCheck new Date(); if (!isHealthy) { console.warn(实例 ${instance.id} 健康检查失败); } } catch (error) { instance.healthy false; console.error(实例 ${instance.id} 健康检查异常:, error); } } /** * 启动定期健康检查 */ private startHealthCheck(intervalMs: number 30000): void { this.healthCheckInterval setInterval(async () { for (const instance of this.instances) { await this.checkInstanceHealth(instance); } }, intervalMs); } /** * 停止健康检查 */ stopHealthCheck(): void { if (this.healthCheckInterval) { clearInterval(this.healthCheckInterval); this.healthCheckInterval null; } } /** * 获取实例状态 */ getInstanceStatus(): any[] { return this.instances.map(instance ({ id: instance.id, url: instance.url, healthy: instance.healthy, concurrentRequests: instance.concurrentRequests, maxConcurrent: instance.maxConcurrent, lastHealthCheck: instance.lastHealthCheck, })); } }4. 实际应用场景与优化建议在实际项目中应用这套方案时我总结了一些经验和建议。4.1 缓存策略优化语音生成比较耗时对于相同的文本内容我们可以使用缓存避免重复生成// src/services/cacheService.ts import Redis from ioredis; import crypto from crypto; export class CacheService { private redis: Redis; private prefix tts:; constructor() { this.redis new Redis({ host: process.env.REDIS_HOST || localhost, port: parseInt(process.env.REDIS_PORT || 6379), }); } /** * 生成缓存键 */ private generateCacheKey(request: any): string { const hash crypto .createHash(md5) .update(JSON.stringify(request)) .digest(hex); return ${this.prefix}${hash}; } /** * 获取缓存 */ async getCachedAudio(request: any): PromiseBuffer | null { const key this.generateCacheKey(request); try { const cached await this.redis.get(key); return cached ? Buffer.from(cached, base64) : null; } catch (error) { console.error(缓存读取失败:, error); return null; } } /** * 设置缓存 */ async setCache(request: any, audioData: Buffer, ttl: number 3600): Promisevoid { const key this.generateCacheKey(request); try { await this.redis.setex(key, ttl, audioData.toString(base64)); } catch (error) { console.error(缓存设置失败:, error); } } /** * 批量获取缓存 */ async batchGetCache(requests: any[]): Promise(Buffer | null)[] { const keys requests.map(req this.generateCacheKey(req)); const pipeline this.redis.pipeline(); keys.forEach(key pipeline.get(key)); try { const results await pipeline.exec(); return results!.map(([err, data]) err || !data ? null : Buffer.from(data as string, base64) ); } catch (error) { console.error(批量缓存读取失败:, error); return requests.map(() null); } } }4.2 监控与日志在生产环境中完善的监控和日志是必不可少的// src/utils/logger.ts import winston from winston; import DailyRotateFile from winston-daily-rotate-file; export const logger winston.createLogger({ level: info, format: winston.format.combine( winston.format.timestamp(), winston.format.json() ), transports: [ // 控制台输出 new winston.transports.Console({ format: winston.format.combine( winston.format.colorize(), winston.format.simple() ), }), // 按天轮转的文件日志 new DailyRotateFile({ filename: logs/tts-%DATE%.log, datePattern: YYYY-MM-DD, maxSize: 20m, maxFiles: 30d, }), // 错误日志单独文件 new DailyRotateFile({ filename: logs/error-%DATE%.log, datePattern: YYYY-MM-DD, level: error, maxSize: 20m, maxFiles: 30d, }), ], }); // 性能监控中间件 export const performanceMonitor (req: any, res: any, next: any) { const start Date.now(); const originalEnd res.end; res.end function(...args: any[]) { const duration Date.now() - start; logger.info({ type: request, method: req.method, path: req.path, status: res.statusCode, duration: ${duration}ms, userAgent: req.get(User-Agent), ip: req.ip, }); originalEnd.apply(this, args); }; next(); };4.3 配置管理使用环境变量和配置文件管理不同环境的配置// src/config/index.ts export const config { // 服务器配置 server: { port: parseInt(process.env.PORT || 3000), env: process.env.NODE_ENV || development, }, // Fish-Speech服务配置 fishSpeech: { urls: (process.env.FISH_SPEECH_URLS || http://localhost:8000).split(,), timeout: parseInt(process.env.FISH_SPEECH_TIMEOUT || 30000), maxConcurrent: parseInt(process.env.FISH_SPEECH_MAX_CONCURRENT || 5), }, // Redis配置 redis: { host: process.env.REDIS_HOST || localhost, port: parseInt(process.env.REDIS_PORT || 6379), password: process.env.REDIS_PASSWORD, db: parseInt(process.env.REDIS_DB || 0), }, // 队列配置 queue: { tts: { concurrency: parseInt(process.env.QUEUE_TTS_CONCURRENCY || 3), attempts: parseInt(process.env.QUEUE_TTS_ATTEMPTS || 3), }, }, // 缓存配置 cache: { ttl: parseInt(process.env.CACHE_TTL || 3600), enabled: process.env.CACHE_ENABLED ! false, }, // 限流配置 rateLimit: { windowMs: parseInt(process.env.RATE_LIMIT_WINDOW_MS || 900000), max: parseInt(process.env.RATE_LIMIT_MAX || 100), }, }; // 验证配置 export function validateConfig() { const required [FISH_SPEECH_URLS]; for (const key of required) { if (!process.env[key]) { throw new Error(缺少必要环境变量: ${key}); } } }5. 部署与运维建议最后说说部署和运维方面的一些建议。5.1 Docker容器化部署建议使用Docker Compose编排所有服务# docker-compose.yml version: 3.8 services: # Redis服务 redis: image: redis:alpine ports: - 6379:6379 volumes: - redis_data:/data command: redis-server --appendonly yes # Fish-Speech服务多个实例 fish-speech-1: image: fishaudio/fish-speech:1.5 deploy: resources: reservations: devices: - driver: nvidia count: 1 capabilities: [gpu] ports: - 8001:8000 environment: - CUDA_VISIBLE_DEVICES0 fish-speech-2: image: fishaudio/fish-speech:1.5 deploy: resources: reservations: devices: - driver: nvidia count: 1 capabilities: [gpu] ports: - 8002:8000 environment: - CUDA_VISIBLE_DEVICES0 # Node.js API服务 api: build: . ports: - 3000:3000 environment: - NODE_ENVproduction - REDIS_HOSTredis - FISH_SPEECH_URLShttp://fish-speech-1:8000,http://fish-speech-2:8000 depends_on: - redis - fish-speech-1 - fish-speech-2 restart: unless-stopped volumes: redis_data:5.2 健康检查与自动恢复在Docker Compose中配置健康检查services: api: # ... 其他配置 healthcheck: test: [CMD, curl, -f, http://localhost:3000/api/tts/health] interval: 30s timeout: 10s retries: 3 start_period: 40s fish-speech-1: # ... 其他配置 healthcheck: test: [CMD, curl, -f, http://localhost:8000/health] interval: 30s timeout: 10s retries: 35.3 性能监控指标建议监控以下关键指标API响应时间P95、P99延迟并发请求数当前活跃请求数队列长度等待处理的任务数错误率API错误比例GPU利用率Fish-Speech实例的GPU使用情况缓存命中率缓存效果评估可以使用Prometheus Grafana搭建监控面板或者使用云服务商提供的监控服务。6. 总结这套Node.js集成Fish-Speech的方案我们在实际项目中跑了几个月整体效果还不错。从最初的单实例部署到现在支持多实例负载均衡、异步队列、缓存优化系统越来越稳定。最大的感受是语音服务这种对延迟和并发要求比较高的场景一定要做好架构设计。单纯的同步调用肯定不行必须考虑异步处理、队列、缓存这些机制。Fish-Speech-1.5本身的质量确实可以声音自然度比很多商业方案都好。开源的优势就是可以自己掌控根据业务需求做定制优化。如果你也在考虑做语音服务建议先从简单的单实例开始跑通基本流程。然后根据实际压力逐步引入队列、缓存、负载均衡这些组件。监控一定要从一开始就做好这样才能及时发现问题优化性能。实际用下来这套方案能支撑的并发量还不错成本也比商业方案低很多。当然也有些小问题需要持续优化比如内存管理、错误重试策略等。后面我们还会继续探索更高效的音频编码、更智能的缓存策略有机会再跟大家分享。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。