# tts_demo **Repository Path**: tian-chenfeng/tts_de ## Basic Information - **Project Name**: tts_demo - **Description**: No description available - **Primary Language**: Unknown - **License**: Not specified - **Default Branch**: master - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 0 - **Forks**: 0 - **Created**: 2026-05-03 - **Last Updated**: 2026-05-03 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README # AI 语音对话系统 - 从零开始教程 ## 项目概述 这是一个支持 **文字聊天** 和 **语音对话** 的 AI 系统，包含三个子项目： ``` tts_all/ ├── ai-stream-server/ # 后端 (NestJS) ├── ai-stream-client/ # 前端 (React + Vite) └── voice-agent/ # 语音服务 (Node.js) ``` ## 技术栈 | 组件 | 技术 | |------|------| | 前端 | React 19 + TypeScript + Vite | | 后端 | NestJS + OpenAI SDK | | 语音识别 (ASR) | 阿里百炼 qwen3-asr-flash-realtime | | 语音合成 (TTS) | 阿里百炼 qwen3-tts-flash-realtime | | 实时通信 | LiveKit Cloud | | AI 模型 | DeepSeek (OpenAI 兼容接口) | ## 完整数据流 ``` ┌─────────────────────────────────────────────────────────────┐ │ 语音对话流程 │ ├─────────────────────────────────────────────────────────────┤ │ │ │ 用户说话 │ │ ↓ │ │ 浏览器麦克风 → LiveKit Cloud → voice-agent │ │ ↓ │ │ 阿里 ASR (语音→文字) │ │ ↓ │ │ 后端 /chat/stream → DeepSeek AI (文字回复) │ │ ↓ │ │ 阿里 TTS (文字→语音) │ │ ↓ │ │ LiveKit Cloud → 浏览器扬声器 │ │ │ └─────────────────────────────────────────────────────────────┘ ┌─────────────────────────────────────────────────────────────┐ │ 文字聊天流程 │ ├─────────────────────────────────────────────────────────────┤ │ │ │ 用户输入文字 │ │ ↓ │ │ 前端 EventSource (SSE) → 后端 /chat/stream │ │ ↓ │ │ DeepSeek AI 流式返回 │ │ ↓ │ │ 前端实时渲染 Markdown │ │ │ └─────────────────────────────────────────────────────────────┘ ``` --- ## 第一章：后端开发 (ai-stream-server) ### 1.1 初始化项目 ```bash mkdir ai-stream-server cd ai-stream-server pnpm init ``` ### 1.2 安装依赖 ```bash # 核心依赖 pnpm add @nestjs/common @nestjs/core @nestjs/platform-express @nestjs/config pnpm add rxjs reflect-metadata openai livekit-server-sdk # 开发依赖 pnpm add -D @nestjs/cli @nestjs/schematics typescript @types/node @types/express ``` ### 1.3 项目结构 ``` ai-stream-server/ ├── src/ │ ├── main.ts # 入口文件 │ ├── app.module.ts # 根模块 │ ├── app.controller.ts # 根控制器 │ ├── app.service.ts # 根服务 │ ├── chat/ # 聊天模块 │ │ ├── chat.module.ts │ │ ├── chat.controller.ts │ │ └── chat.service.ts │ ├── stream/ # 流式处理模块 │ │ └── stream.service.ts │ ├── memory/ # 对话记忆模块 │ │ └── memory.service.ts │ ├── decision/ # 决策模块 │ │ └── decision.service.ts │ ├── tool/ # 工具模块 │ │ └── tool.service.ts │ └── livekit/ # LiveKit 模块 │ ├── livekit.module.ts │ ├── livekit.controller.ts │ └── livekit.service.ts ├── .env # 环境变量 ├── package.json ├── tsconfig.json └── nest-cli.json ``` ## 创建项目 nest new ai-stream-server ### 生成模块 nest g module chat nest g module livekit nest g module stream nest g module memory nest g module decision nest g module tool ### 生成控制器和服务 nest g controller chat nest g service chat nest g controller livekit nest g service livekit ### 1.4 入口文件 main.ts ```typescript import { NestFactory } from '@nestjs/core'; import { AppModule } from './app.module'; async function bootstrap() { const app = await NestFactory.create(AppModule); // 允许跨域，前端运行在 localhost:5173 app.enableCors({ origin: 'http://localhost:5173', credentials: true, methods: 'GET,HEAD,PUT,PATCH,POST,DELETE,OPTIONS', allowedHeaders: ['Content-Type', 'Authorization'], }); await app.listen(process.env.PORT ?? 3000); } bootstrap(); ``` **要点：** - NestJS 默认端口 3000 - 必须开启 CORS，否则前端无法访问 ### 1.5 根模块 app.module.ts ```typescript import { Module } from '@nestjs/common'; import { AppController } from './app.controller'; import { AppService } from './app.service'; import { ChatModule } from './chat/chat.module'; import { ConfigModule } from '@nestjs/config'; import { LivekitModule } from './livekit/livekit.module'; @Module({ imports: [ ChatModule, LivekitModule, ConfigModule.forRoot({ isGlobal: true, // 全局可用 ConfigService }), ], controllers: [AppController], providers: [AppService], }) export class AppModule {} ``` **要点：** - `ConfigModule.forRoot({ isGlobal: true })` 让环境变量全局可用 - 按功能拆分模块：ChatModule、LivekitModule ### 1.6 流式处理服务 stream.service.ts 这是核心服务，负责调用 AI 模型并流式返回结果。 ```typescript import { Injectable } from '@nestjs/common'; import { Observable } from 'rxjs'; import OpenAI from 'openai'; import { ChatMessage } from '../memory/memory.service'; @Injectable() export class StreamService { private client = new OpenAI({ apiKey: process.env.OPENAI_API_KEY, baseURL: process.env.OPENAI_BASE_URL || 'https://api.deepseek.com', }); // 调用 LLM 流式接口 createLLMStream( requestMessages: ChatMessage[], onFinish: (assistantText: string) => void, ): Observable { return new Observable((observer) => { let aborted = false; (async () => { try { const stream = await this.client.chat.completions.create({ model: process.env.OPENAI_MODEL || 'deepseek-chat', messages: requestMessages, stream: true, // 开启流式 }); let assistantText = ''; for await (const chunk of stream) { if (aborted) break; const content = chunk.choices?.[0]?.delta?.content; if (content) { assistantText += content; // 发送 SSE 事件 observer.next({ data: { type: 'chunk', content: content, }, } as MessageEvent); } } // 回调：保存完整回复 onFinish(assistantText); // 发送完成事件 observer.next({ data: { type: 'done', content: '[DONE]', }, } as MessageEvent); observer.complete(); } catch (error: any) { observer.next({ data: { type: 'error', content: error?.message || 'stream error', }, } as MessageEvent); observer.complete(); } })(); return () => { aborted = true; // 支持取消 }; }); } } ``` **要点：** - 使用 RxJS Observable 包装 SSE 流 - OpenAI SDK 兼容 DeepSeek 接口 - `stream: true` 开启流式返回 - 支持取消操作（aborted 标志） ### 1.7 对话记忆服务 memory.service.ts ```typescript import { Injectable } from '@nestjs/common'; export type ChatMessage = { role: 'user' | 'assistant' | 'system'; content: string; }; @Injectable() export class MemoryService { // 简单用 Map 存储对话历史 private conversations = new Map(); getMessages(conversationId: string): ChatMessage[] { return this.conversations.get(conversationId) || []; } saveMessages(conversationId: string, messages: ChatMessage[]): void { this.conversations.set(conversationId, messages); } } ``` **要点：** - 生产环境应使用 Redis 或数据库 - 每个 conversationId 对应一个对话历史 ### 1.8 聊天服务 chat.service.ts ```typescript import { Injectable } from '@nestjs/common'; import { Observable } from 'rxjs'; import { MemoryService, ChatMessage } from '../memory/memory.service'; import { StreamService } from '../stream/stream.service'; @Injectable() export class ChatService { constructor( private readonly memoryService: MemoryService, private readonly streamService: StreamService, ) {} streamReply(message: string, conversationId: string): Observable { // 获取历史消息 const oldMessages = this.memoryService.getMessages(conversationId); const currentUserMessage: ChatMessage = { role: 'user', content: message, }; const requestMessages = oldMessages.concat(currentUserMessage); // 调用 LLM 流式接口 return this.streamService.createLLMStream( requestMessages, (assistantText: string) => { // 保存对话历史 const assistantMessage: ChatMessage = { role: 'assistant', content: assistantText, }; const finalMessages = requestMessages.concat(assistantMessage); this.memoryService.saveMessages(conversationId, finalMessages); }, ); } } ``` ### 1.9 聊天控制器 chat.controller.ts ```typescript import { Controller, Get, Query, Sse } from '@nestjs/common'; import { Observable } from 'rxjs'; import { ChatService } from './chat.service'; @Controller('chat') export class ChatController { constructor(private readonly chatService: ChatService) {} @Get('stream') @Sse() // 声明 SSE 端点 stream( @Query('message') message: string, @Query('conversationId') conversationId: string, ): Observable { return this.chatService.streamReply(message, conversationId); } } ``` **要点：** - `@Sse()` 装饰器声明 SSE 端点 - 前端用 EventSource 连接此接口 ### 1.10 LiveKit 服务 livekit.service.ts ```typescript import { Injectable, InternalServerErrorException } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; import { AccessToken } from 'livekit-server-sdk'; @Injectable() export class LivekitService { constructor(private readonly configService: ConfigService) {} async createToken() { const apiKey = this.configService.get('LIVEKIT_API_KEY'); const apiSecret = this.configService.get('LIVEKIT_API_SECRET'); const livekitUrl = this.configService.get('LIVEKIT_URL'); if (!apiKey || !apiSecret || !livekitUrl) { throw new InternalServerErrorException('LiveKit env is missing'); } const roomName = 'voice-room-demo'; const participantName = `user-${Date.now()}`; // 生成 Token const at = new AccessToken(apiKey, apiSecret, { identity: participantName, name: participantName, }); at.addGrant({ room: roomName, roomJoin: true, canPublish: true, canSubscribe: true, canPublishData: true, }); const token = await at.toJwt(); return { livekitUrl, roomName, participantName, token, }; } } ``` **要点：** - AccessToken 用于生成用户进入房间的凭证 - `addGrant` 控制用户权限 ### 1.11 LiveKit 控制器 livekit.controller.ts ```typescript import { Controller, Get } from '@nestjs/common'; import { LivekitService } from './livekit.service'; @Controller('livekit') export class LivekitController { constructor(private readonly livekitService: LivekitService) {} @Get('token') createToken() { return this.livekitService.createToken(); } } ``` ### 1.12 环境变量 .env ```env # DeepSeek API OPENAI_API_KEY=sk-xxxxxxxx OPENAI_BASE_URL=https://api.deepseek.com OPENAI_MODEL=deepseek-chat # LiveKit LIVEKIT_API_KEY=xxxxxxxx LIVEKIT_API_SECRET=xxxxxxxx LIVEKIT_URL=wss://xxxxx.livekit.cloud ``` --- ## 第二章：前端开发 (ai-stream-client) ### 2.1 初始化项目 ```bash pnpm create vite ai-stream-client --template react-ts cd ai-stream-client pnpm install ``` ### 2.2 安装依赖 ```bash # LiveKit 语音组件 pnpm add @livekit/components-react @livekit/components-styles livekit-client # Markdown 渲染 pnpm add react-markdown remark-gfm rehype-highlight highlight.js # Socket.IO (备用) pnpm add socket.io-client ``` ### 2.3 项目结构 ``` ai-stream-client/ ├── src/ │ ├── main.tsx # 入口 │ ├── App.tsx # 主组件 │ ├── App.css # 样式 │ ├── index.css # 全局样式 │ ├── VoiceRoom.tsx # 语音房间组件 │ ├── Complete/ # 聊天组件 │ │ ├── types.ts # 类型定义 │ │ ├── stream.ts # SSE 流式连接 │ │ ├── ChatInput.tsx # 输入框 │ │ ├── MessageList.tsx # 消息列表 │ │ ├── MessageItem.tsx # 单条消息 │ │ ├── MarkdownRender.tsx # Markdown 渲染 │ │ └── utils/ │ │ ├── messageUtlis.ts # 消息工具 │ │ └── streamBuffer.ts # 流式缓冲 │ └── Hooks/ │ └── useChat.ts # 聊天 Hook ├── index.html ├── package.json ├── vite.config.ts └── tsconfig.json ``` ### 2.4 类型定义 types.ts ```typescript // 消息类型 export type Message = { role: 'user' | 'assistant'; content: string; }; // 流式数据类型 export type StreamData = { type: 'chunk' | 'done'; content?: string; }; ``` ### 2.5 SSE 流式连接 stream.ts ```typescript import type { StreamData } from './types'; type StreamHandlers = { onChunk: (text: string) => void; onDone: () => void; onError: () => void; }; export function createChatStream( question: string, conversationId: string, handlers: StreamHandlers, ) { const url = 'http://localhost:3000/chat/stream?message=' + encodeURIComponent(question) + '&conversationId=' + conversationId; // 创建 SSE 连接 const es = new EventSource(url); es.onmessage = function (event) { const data: StreamData = JSON.parse(event.data); if (data.type === 'chunk') { if (data.content) { handlers.onChunk(data.content); } } if (data.type === 'done') { handlers.onDone(); es.close(); } }; es.onerror = function () { handlers.onError(); es.close(); }; return es; } ``` **要点：** - EventSource 是浏览器原生 SSE API - 后端 `@Sse()` 装饰器配合使用 - 每次收到 chunk 触发 onChunk，收到 done 触发 onDone ### 2.6 流式缓冲 streamBuffer.ts ```typescript export function createStreamBuffer() { let buffer = ''; function push(text: string) { buffer = buffer + text; } function read() { return buffer; } function clear() { buffer = ''; } return { push, read, clear }; } ``` **要点：** - 缓冲区用于批量更新 UI，避免每次 chunk 都触发渲染 ### 2.7 消息工具 messageUtlis.ts ```typescript import type { Message } from '../types'; export function appendToLastMessage(list: Message[], text: string): Message[] { const newList = list.slice(); if (newList.length === 0) { return newList; } const lastIndex = newList.length - 1; const lastMessage = newList[lastIndex]; newList[lastIndex] = { role: lastMessage.role, content: lastMessage.content + text, }; return newList; } ``` **要点：** - 不可变更新：创建新数组而非修改原数组 - 只追加到最后一条消息 ### 2.8 聊天 Hook useChat.ts ```typescript import { useState, useRef } from 'react'; import type { Message } from '../Complete/types'; import { createChatStream } from '../Complete/stream'; import { appendToLastMessage } from '../Complete/utils/messageUtlis'; import { createStreamBuffer } from '../Complete/utils/streamBuffer'; export function useChat() { const [input, setInput] = useState(''); const [list, setList] = useState([]); const [loading, setLoading] = useState(false); const eventSourceRef = useRef(null); const stoppedRef = useRef(false); const bufferRef = useRef(createStreamBuffer()); const timerRef = useRef(null); // 定时刷新缓冲区 function startFlushLoop() { if (timerRef.current !== null) return; timerRef.current = window.setInterval(function () { flushBuffer(); }, 50); // 每 50ms 刷新一次 } function stopFlushLoop() { if (timerRef.current !== null) { clearInterval(timerRef.current); timerRef.current = null; } } // 把缓冲区内容写入消息列表 function flushBuffer() { const text = bufferRef.current.read(); if (text === '') return; setList(function (oldList) { return appendToLastMessage(oldList, text); }); bufferRef.current.clear(); } // 发送消息 function handleSend() { if (input === '') return; if (loading) return; stoppedRef.current = false; const question = input; const userMessage: Message = { role: 'user', content: question }; const aiMessage: Message = { role: 'assistant', content: '' }; setList(function (oldList) { return oldList.concat(userMessage).concat(aiMessage); }); setInput(''); setLoading(true); startFlushLoop(); const es = createChatStream(question, 'demo-1', { onChunk: function (text) { if (stoppedRef.current) return; bufferRef.current.push(text); }, onDone: function () { stopFlushLoop(); flushBuffer(); // 最后一次刷新 setLoading(false); eventSourceRef.current = null; }, onError: function () { setLoading(false); eventSourceRef.current = null; }, }); eventSourceRef.current = es; } // 停止生成 function handleStop() { stoppedRef.current = true; if (eventSourceRef.current) { eventSourceRef.current.close(); eventSourceRef.current = null; } setLoading(false); } return { input, setInput, list, handleSend, handleStop, loading, }; } ``` **要点：** - 缓冲区 + 定时器 = 批量更新，避免频繁渲染 - `stoppedRef` 防止停止后仍有数据写入 - 每次发送先追加 user + empty assistant 消息 ### 2.9 输入框组件 ChatInput.tsx ```typescript type ChatInputProps = { value: string; onChange: (value: string) => void; onSend: () => void; onStop: () => void; loading: boolean; }; function ChatInput(props: ChatInputProps) { return (
); } export default ChatInput; ``` ### 2.10 消息列表 MessageList.tsx ```typescript import MessageItem from './MessageItem'; import type { Message } from './types'; type MessageListProps = { list: Message[]; }; function MessageList(props: MessageListProps) { return (
{props.list.map(function (item, index) { return ( ); })}
); } export default MessageList; ``` ### 2.11 单条消息 MessageItem.tsx ```typescript import type { Message } from './types'; import MarkdownRenderer from './MarkdownRender'; type MessageItemProps = Message; function MessageItem(props: MessageItemProps) { return (
{props.role}：
); } export default MessageItem; ``` ### 2.12 Markdown 渲染 MarkdownRender.tsx ```typescript import ReactMarkdown from 'react-markdown'; import remarkGfm from 'remark-gfm'; import rehypeHighlight from 'rehype-highlight'; import 'highlight.js/styles/github.css'; type MarkdownRendererProps = { content: string; }; function MarkdownRenderer(props: MarkdownRendererProps) { return ( {props.content} ); } export default MarkdownRenderer; ``` ### 2.13 语音房间 VoiceRoom.tsx ```typescript import { useState } from 'react'; import { LiveKitRoom, RoomAudioRenderer, ControlBar, } from '@livekit/components-react'; import '@livekit/components-styles'; type LiveKitTokenResponse = { livekitUrl: string; roomName: string; participantName: string; token: string; }; export default function VoiceRoom() { const [config, setConfig] = useState(null); async function startRoom() { const res = await fetch('http://localhost:3000/livekit/token'); if (!res.ok) { throw new Error('获取 LiveKit token 失败'); } const data = (await res.json()) as LiveKitTokenResponse; setConfig(data); } function stopRoom() { setConfig(null); } if (!config) { return (
); } return (

LiveKit 语音房间

房间：{config.roomName}

用户：{config.participantName}

); } ``` **要点：** - `LiveKitRoom` 自动连接 LiveKit 服务器 - `RoomAudioRenderer` 自动播放房间内所有音频 - `ControlBar` 提供麦克风控制 ### 2.14 主组件 App.tsx ```typescript import ChatInput from './Complete/ChatInput'; import MessageList from './Complete/MessageList'; import { useChat } from './Hooks/useChat'; import VoiceRoom from './VoiceRoom'; function App() { const chat = useChat(); return (

AI Chat Demo

); } export default App; ``` --- ## 第三章：语音服务开发 (voice-agent) ### 3.1 初始化项目 ```bash mkdir voice-agent cd voice-agent pnpm init ``` ### 3.2 安装依赖 ```bash pnpm add @livekit/agents @livekit/rtc-node dotenv eventsource tsx typescript ws pnpm add -D @types/node @types/ws ``` ### 3.3 项目结构 ``` voice-agent/ ├── agent.ts # 主入口，LiveKit Agent ├── qwer-asr.ts # 阿里 ASR 语音识别 ├── qwer-tts.ts # 阿里 TTS 语音合成 ├── chat-client.ts # 调用后端 AI 接口 ├── .env # 环境变量 └── package.json ``` ### 3.4 TTS 语音合成 qwer-tts.ts ```typescript import WebSocket from 'ws'; import fs from 'node:fs'; let onAudio: ((audio: Buffer) => void) | null = null; let onDone: (() => void) | null = null; export function setTtsAudioHandler(handler: (audio: Buffer) => void) { onAudio = handler; } export function setTtsDoneHandler(handler: () => void) { onDone = handler; } function eventId() { return `event_${Date.now()}`; } export function createTTS() { const apiKey = process.env.DASHSCOPE_API_KEY; const model = process.env.QWEN_TTS_MODEL || 'qwen3-tts-flash-realtime'; let readyResolve: (() => void) | null = null; let readyReject: ((error: Error) => void) | null = null; const ready = new Promise((resolve, reject) => { readyResolve = resolve; readyReject = reject; }); // 连接阿里百炼 TTS WebSocket const ws = new WebSocket( `wss://dashscope.aliyuncs.com/api-ws/v1/realtime?model=${model}`, { headers: { Authorization: `Bearer ${apiKey}`, 'OpenAI-Beta': 'realtime=v1', }, }, ); ws.on('open', () => { console.log('[tts] connected'); // 发送 TTS 配置 ws.send( JSON.stringify({ event_id: eventId(), type: 'session.update', session: { voice: 'Cherry', // 音色 response_format: 'pcm', // 输出格式 mode: 'server_commit', // 服务端自动判断合成时机 sample_rate: 24000, // 采样率 }, }), ); }); ws.on('message', (data) => { const event = JSON.parse(data.toString()); if (event.type === 'session.updated') { readyResolve?.(); } if (event.type === 'response.audio.delta') { // 收到音频片段 const audio = Buffer.from(event.delta, 'base64'); fs.appendFileSync('tts-test.pcm', audio); // 调试用 onAudio?.(audio); // 交给 agent.ts 处理 } if (event.type === 'response.done') { onDone?.(); } if (event.type === 'session.finished') { ws.close(); } if (event.type === 'error') { readyReject?.(new Error(event.error?.message || 'tts error')); } }); return { async sendText(text: string) { await ready; if (ws.readyState !== WebSocket.OPEN) return; ws.send( JSON.stringify({ event_id: eventId(), type: 'input_text_buffer.append', text, }), ); ws.send( JSON.stringify({ event_id: eventId(), type: 'session.finish', }), ); }, close() { if (ws.readyState !== WebSocket.OPEN) return; ws.send( JSON.stringify({ event_id: eventId(), type: 'session.finish', }), ); ws.close(); }, }; } ``` **要点：** - WebSocket 连接阿里百炼实时 API - `session.update` 配置音色、采样率 - `response.audio.delta` 接收 base64 音频片段 - `ready` Promise 确保连接完成后再发送文本 ### 3.5 ASR 语音识别 qwer-asr.ts ```typescript import WebSocket from 'ws'; import { askChat } from './chat-client'; function eventId() { return `event_${Date.now()}`; } export function createASR() { const model = process.env.QWEN_ASR_MODEL || 'qwen3-asr-flash-realtime'; const apiKey = process.env.DASHSCOPE_API_KEY; // 连接阿里百炼 ASR WebSocket const ws = new WebSocket( `wss://dashscope.aliyuncs.com/api-ws/v1/realtime?model=${model}`, { headers: { Authorization: `Bearer ${apiKey}`, 'OpenAI-Beta': 'realtime=v1', }, }, ); ws.on('open', () => { console.log('[asr] connected'); // 发送 ASR 配置 ws.send( JSON.stringify({ event_id: eventId(), type: 'session.update', session: { input_audio_format: 'pcm', sample_rate: 16000, input_audio_transcription: { language: 'zh' }, turn_detection: { type: 'server_vad', silence_duration_ms: 800 }, }, }), ); }); ws.on('message', (data) => { const event = JSON.parse(data.toString()); // 实时识别结果 if (event.type === 'conversation.item.input_audio_transcription.text') { console.log('[asr.partial]', event.text || event.stash || ''); } // 最终识别结果 if (event.type === 'conversation.item.input_audio_transcription.completed') { console.log('[asr.final]', event.transcript); askChat(event.transcript); // 发送给 AI } }); return { sendAudio(buffer: Buffer) { if (ws.readyState !== WebSocket.OPEN) return; ws.send( JSON.stringify({ event_id: eventId(), type: 'input_audio_buffer.append', audio: buffer.toString('base64'), }), ); }, close() { if (ws.readyState !== WebSocket.OPEN) return; ws.send( JSON.stringify({ event_id: eventId(), type: 'session.finish' }), ); ws.close(); }, }; } ``` **要点：** - `server_vad` 服务端语音活动检测，自动断句 - `silence_duration_ms: 800` 静音 800ms 视为说完一句 - 识别完成后调用 `askChat` 发送给 AI ### 3.6 聊天客户端 chat-client.ts ```typescript import { EventSource } from 'eventsource'; import { createTTS } from './qwer-tts'; export function askChat(message: string) { const tts = createTTS(); let fullText = ''; const url = 'http://localhost:3000/chat/stream?message=' + encodeURIComponent(message) + '&conversationId=demo-1'; const es = new EventSource(url); es.onmessage = (event) => { const data = JSON.parse(event.data); if (data.type === 'chunk') { const text = data.content || ''; fullText += text; process.stdout.write(text); // 控制台流式打印 } if (data.type === 'done') { console.log('\n[chat.done]'); // 把完整 AI 回复送给 TTS void tts.sendText(fullText).catch((error) => { console.log('[tts.send.error]', error?.message || error); }); es.close(); } }; es.onerror = () => { console.log('[chat.error]'); es.close(); }; } ``` **要点：** - 等 AI 回复完整后才发给 TTS - `process.stdout.write` 实现流式打印 ### 3.7 主入口 agent.ts ```typescript import dotenv from 'dotenv'; import { fileURLToPath } from 'node:url'; import { AutoSubscribe, cli, defineAgent, ServerOptions, type JobContext, } from '@livekit/agents'; import { AudioFrame, AudioSource, AudioStream, LocalAudioTrack, RemoteTrack, RoomEvent, TrackPublishOptions, TrackSource, } from '@livekit/rtc-node'; import { createASR } from './qwer-asr'; import { setTtsAudioHandler, setTtsDoneHandler } from './qwer-tts'; dotenv.config(); let isAiSpeaking = false; let aiSpeakingTimer: NodeJS.Timeout | null = null; // 处理用户音频 async function handleAudioTrack(track: RemoteTrack) { console.log('[voice-agent] 收到用户音频轨道'); const asr = createASR(); const audioStream = new AudioStream(track, 16000, 1); for await (const frame of audioStream) { if (isAiSpeaking) continue; // AI 说话时不处理用户音频 const pcmBuffer = Buffer.from( frame.data.buffer, frame.data.byteOffset, frame.data.byteLength, ); asr.sendAudio(pcmBuffer); } asr.close(); } export default defineAgent({ entry: async (ctx: JobContext) => { const room = ctx.room; console.log('[voice-agent] 准备连接房间'); // 监听用户音频轨道 room.on(RoomEvent.TrackSubscribed, async (track: RemoteTrack) => { console.log('[track.subscribed]', track.kind); void handleAudioTrack(track); }); // 连接房间，只订阅音频 await ctx.connect(undefined, AutoSubscribe.AUDIO_ONLY); console.log('[voice-agent] 已连接房间：', room.name); // 创建 AI 语音源 const ttsSource = new AudioSource(24000, 1); const ttsTrack = LocalAudioTrack.createAudioTrack('ai-voice', ttsSource); const options = new TrackPublishOptions(); options.source = TrackSource.SOURCE_MICROPHONE; const localParticipant = room.localParticipant; if (!localParticipant) { throw new Error('localParticipant is undefined'); } // 发布 AI 语音轨道 await localParticipant.publishTrack(ttsTrack, options); console.log('[voice-agent] AI 音频轨道已发布'); let ttsRemain = Buffer.alloc(0); let ttsQueue = Promise.resolve(); const frameBytes = 960; // 24kHz * 20ms * 2 bytes function markAiSpeaking() { isAiSpeaking = true; if (aiSpeakingTimer) { clearTimeout(aiSpeakingTimer); aiSpeakingTimer = null; } } function scheduleAsrResume() { const resumeDelayMs = Math.max(ttsSource.queuedDuration, 300) + 300; aiSpeakingTimer = setTimeout(() => { isAiSpeaking = false; aiSpeakingTimer = null; }, resumeDelayMs); } async function capturePcmFrame(chunk: Buffer) { const frame = AudioFrame.create(24000, 1, 480); for (let i = 0; i < 480; i += 1) { frame.data[i] = chunk.readInt16LE(i * 2); } await ttsSource.captureFrame(frame); } // TTS 音频回调 setTtsAudioHandler((audio: Buffer) => { ttsQueue = ttsQueue.then(async () => { markAiSpeaking(); const buffer = Buffer.concat([ttsRemain, audio]); let offset = 0; while (offset + frameBytes <= buffer.length) { const chunk = buffer.subarray(offset, offset + frameBytes); await capturePcmFrame(chunk); offset += frameBytes; } ttsRemain = buffer.subarray(offset); scheduleAsrResume(); }); }); // TTS 完成回调 setTtsDoneHandler(() => { ttsQueue = ttsQueue.then(async () => { markAiSpeaking(); if (ttsRemain.length > 0) { const padded = Buffer.alloc(frameBytes); ttsRemain.copy(padded); ttsRemain = Buffer.alloc(0); await capturePcmFrame(padded); } scheduleAsrResume(); }); }); }, }); cli.runApp( new ServerOptions({ agent: fileURLToPath(import.meta.url), agentName: 'voice-agent', }), ); ``` **要点：** - `defineAgent` 定义 LiveKit Agent - `AutoSubscribe.AUDIO_ONLY` 只订阅音频 - `isAiSpeaking` 防止 AI 说话时把声音送入 ASR - `ttsQueue` 保证 TTS 音频按顺序播放 ### 3.8 环境变量 .env ```env LIVEKIT_API_KEY=xxxxxxxx LIVEKIT_API_SECRET=xxxxxxxx LIVEKIT_URL=wss://xxxxx.livekit.cloud DASHSCOPE_API_KEY=sk-xxxxxxxx QWEN_ASR_MODEL=qwen3-asr-flash-realtime QWEN_TTS_MODEL=qwen3-tts-flash-realtime ``` --- ## 第四章：运行项目 ### 4.1 获取 API Key | 服务 | 获取方式 | |------|----------| | DeepSeek | https://platform.deepseek.com | | 阿里百炼 | https://dashscope.console.aliyun.com | | LiveKit Cloud | https://cloud.livekit.io | ### 4.2 配置环境变量 **ai-stream-server/.env** ```env OPENAI_API_KEY=你的DeepSeek_Key OPENAI_BASE_URL=https://api.deepseek.com OPENAI_MODEL=deepseek-chat LIVEKIT_API_KEY=你的LiveKit_Key LIVEKIT_API_SECRET=你的LiveKit_Secret LIVEKIT_URL=wss://你的LiveKit地址 ``` **voice-agent/.env** ```env LIVEKIT_API_KEY=同上 LIVEKIT_API_SECRET=同上 LIVEKIT_URL=同上 DASHSCOPE_API_KEY=你的阿里百炼_Key QWEN_ASR_MODEL=qwen3-asr-flash-realtime QWEN_TTS_MODEL=qwen3-tts-flash-realtime ``` ### 4.3 启动服务 **终端 1 - 后端** ```bash cd ai-stream-server pnpm install pnpm run start:dev ``` **终端 2 - 前端** ```bash cd ai-stream-client pnpm install pnpm dev ``` **终端 3 - 语音服务** ```bash cd voice-agent pnpm install npx tsx agent.ts connect --room voice-room-demo ``` ### 4.4 访问 - 前端：http://localhost:5173 - 后端：http://localhost:3000 --- ## 第五章：常见问题 ### Q1: 环境变量加载为 0 A: 检查 `.env` 文件是否有空行，删除空行后重启。 ### Q2: LiveKit token 获取失败 A: 检查后端 `.env` 是否配置了 LiveKit 相关变量。 ### Q3: 语音没有回应 A: 确保 voice-agent 显示 `[asr] connected`，并且有 `[asr.partial]` 输出。 ### Q4: CORS 错误 A: 确保后端 `main.ts` 配置了正确的 CORS origin。 --- ## 总结 这个项目的核心架构： 1. **后端**：NestJS 提供 SSE 流式接口，调用 DeepSeek AI 2. **前端**：React + EventSource 实时渲染，LiveKit 语音房间 3. **语音服务**：阿里百炼 ASR/TTS + LiveKit 实时通信 三个服务通过 LiveKit Cloud 和 HTTP API 协作，实现完整的语音对话体验。