@@ -143,11 +143,53 @@ interface TTSAudioMessage {
type : "tts_audio" ;
request_id : string ;
audio : string ; // Base64 编码的音频片段
mime_type : string ; // "audio/mp3" 或 "audio/pcm "
mime_type : string ; // "audio/mpeg "
is_last : boolean ; // 是否为最后一片
}
```
**音频格式规范 ** （前端播放依赖此约定）：
| 属性 | 值 | 说明 |
|------|------|------|
| 编码 | `audio/mpeg` （ ） `<audio>` 原生支持，
| 采样率 | 24kHz | OpenAI TTS 默认 |
| 声道 | 单声道 | 语音不需要立体声 |
| 传输 | Base64 编码的 MP3 片段 | 每个 `tts_audio` 消息携带一个句子的音频 |
| 切片粒度 | 按句子切分 | LLM 输出中按 `。！？\n` 等标点切分，每个句子独立合成 |
**流式播放时序 ** ： `tts_audio` 消息按句子顺序到达，前端应按序排队播放，不要等全部到齐再播。
**前端播放实现要点 ** ：
1. **排队播放 ** ：收到 `tts_audio` 时，将 Base64 解码为 Blob URL 并加入播放队列。第一片到达即开始播放，后续片段在 `onended` 回调中自动衔接。
2. **错误容错 ** ：单个片段播放失败时跳过，继续播放队列中下一个，不中断整个回复。
3. **打断清理 ** ：收到 `interrupt` 消息或用户触发打断时，清空播放队列并释放所有 Blob URL。
4. **类型锁定 ** ： `mime_type` 字段固定为 `"audio/mpeg"` ，前端解码时直接使用，无需运行时判断。
``` typescript
// 前端播放器伪代码
class AudioPlayer {
private queue : string [ ] = [ ] ; // Blob URL 队列
enqueue ( base64 : string ) {
const url = decodeBase64Audio ( base64 , "audio/mpeg" ) ;
this . queue . push ( url ) ;
if ( this . queue . length === 1 ) this . playNext ( ) ; // 第一片到了就开始播
}
private playNext() {
if ( this . queue . length === 0 ) return ;
const audio = new Audio ( this . queue [ 0 ] ) ;
audio . onended = ( ) = > { URL . revokeObjectURL ( this . queue . shift ( ) ! ) ; this . playNext ( ) ; } ;
audio . onerror = ( ) = > { URL . revokeObjectURL ( this . queue . shift ( ) ! ) ; this . playNext ( ) ; } ;
audio . play ( ) ;
}
clear() { this . queue . forEach ( url = > URL . revokeObjectURL ( url ) ) ; this . queue = [ ] ; }
}
```
#### `error` — 错误通知
``` typescript
@@ -250,7 +292,619 @@ DELETE /api/sessions/{session_id}
---
## 三、数据模型
## 三、AI 服务层接口
Go 网关内部与外部 AI 服务（ ） ，
### STT 服务接口
语音识别：接收前端采集的音频，返回识别文本。
``` go
// STTService 语音识别服务契约。
type STTService interface {
// Recognize 识别一段完整音频，返回最终文本。
Recognize ( ctx context . Context , audio [ ] byte , opts STTOptions ) ( string , error )
// RecognizeStream 流式识别（边说边识别，可选实现）。
// audioStream 持续接收音频片段，返回的 channel 持续输出中间结果。
RecognizeStream ( ctx context . Context , audioStream <- chan [ ] byte , opts STTOptions ) ( <- chan STTPartial , error )
}
// STTOptions 语音识别参数。
type STTOptions struct {
Encoding string // "pcm_s16le" — 前端 VAD 输出格式
SampleRate int // 16000 — 前端麦克风采样率
Language string // "zh-CN"
}
// STTPartial 流式识别的中间/最终结果。
type STTPartial struct {
Text string
IsFinal bool
}
```
**Deepgram 接入约定 ** ：
- 连接方式： `wss://api.deepgram.com/v1/listen`
- 音频格式： ， `MicManager` 输出一致）
- 返回格式：`channel.alternatives[0].transcript` ， `is_final` 字段标识最终结果
- 超时：单次识别 5 秒超时
### LLM 服务接口
多模态推理：接收图像 + 文本 + 对话历史，流式返回回复。
``` go
// LLMService 多模态大模型服务契约。
type LLMService interface {
// ChatStream 流式推理，返回增量文本的 channel。
// 调用方必须消费 channel 直到 Done=true，
ChatStream ( ctx context . Context , req LLMRequest ) ( <- chan LLMChunk , error )
}
// LLMRequest 推理请求。
type LLMRequest struct {
Image [ ] byte // JPEG 图片（已从 Base64 解码）
Text string // 用户语音识别后的文本
History [ ] Message // 最近 N 轮对话历史
Language string // "zh-CN"
}
// LLMChunk 流式推理的一个增量片段。
type LLMChunk struct {
Delta string // 增量文本
Done bool // 是否结束
TokensUsed * TokenUsage // 仅 Done=true 时有值
Model string // 实际使用的模型名
}
// TokenUsage 用量统计。
type TokenUsage struct {
Prompt int
Completion int
Total int
}
```
**OpenAI API 接入约定 ** ：
- 端点：`POST https://api.openai.com/v1/chat/completions`
- 图片传入：`image_url` 字段使用 `data:image/jpeg;base64,...` 格式
- 流式响应：`stream: true` ，通过 SSE 逐 chunk 返回
- Prompt 结构：
```
请用简洁自然的中文回答。如果涉及视觉描述，先说'我看到...'。"
user: [图片 + 用户语音文本]
（重复 History 中的历史消息）
```
- 超时： `LLM_TIMEOUT` 错误
- 模型选择：默认 `gpt-4o` ，由 Model Router 按需切换
### TTS 服务接口
语音合成：接收文本流，输出音频 chunk 流。
``` go
// TTSService 语音合成服务契约。
type TTSService interface {
// SynthesizeStream 流式合成。
// textStream 接收句子级文本（由 Orchestrator 的句子切分器产出），
// 返回的 channel 持续输出 MP3 音频 chunk。
SynthesizeStream ( ctx context . Context , textStream <- chan string , opts TTSOptions ) ( <- chan TTSChunk , error )
}
// TTSOptions 合成参数。
type TTSOptions struct {
Voice string // "alloy" | "nova" | "shimmer" | ...
Speed float64 // 1.0 为正常语速
OutputFmt string // "mp3" — 固定使用 MP3，
SampleRate int // 24000
}
// TTSChunk 一个音频片段。
type TTSChunk struct {
Audio [ ] byte // MP3 音频数据（未 Base64 编码，由发送层编码）
IsLast bool // 是否为最后一片
}
```
**OpenAI TTS 接入约定 ** ：
- 端点：`POST https://api.openai.com/v1/audio/speech`
- 模型：`tts-1` （低延迟优先）或 `tts-1-hd` （高音质）
- 输出格式：`mp3` ，
- 流式：使用 `response_format: "mp3"` 并读取 response body 流
- 超时：单个句子 5 秒超时
---
## 四、AI 编排器（ ）
### 编排策略：句子级流式并行
核心矛盾： ，
```
↓
┌── 句子检测器（按 。！？\n 切分）──┐
↓ ↓
句子1: "这是一朵红色的花。" 句子2: "它看起来很美丽。"
↓ ↓
TTS 合成 TTS 合成
↓ ↓
音频 chunk → 推送前端 音频 chunk → 推送前端
```
**时序保证 ** ：
- `llm_chunk` 消息一定先于对应句子的 `tts_audio` 到达客户端
- 用户先看到文字，紧接着听到语音（感知延迟 < 0.5 秒）
- 不必等 LLM 全部输出完才开始 TTS
### Orchestrator 接口
``` go
// Orchestrator AI 编排器，协调 STT → LLM → TTS 全链路。
type Orchestrator struct {
stt STTService
llm LLMService
tts TTSService
}
// ProcessQuery 处理一次完整的视觉对话请求。
// 通过 client 向前端实时推送 stt_result、llm_chunk、llm_done、tts_audio 消息。
func ( o * Orchestrator ) ProcessQuery ( ctx context . Context , client MessageSender , req * QueryRequest ) {
ctx , cancel := context . WithTimeout ( ctx , 10 * time . Second )
defer cancel ( )
// Step 1: STT — 识别用户语音
text , err := o . stt . Recognize ( ctx , req . Audio , STTOptions {
Encoding : "pcm_s16le" , SampleRate : 16000 , Language : "zh-CN" ,
} )
if err != nil {
client . SendError ( req . RequestID , "STT_ERROR" , err . Error ( ) )
return
}
client . SendSTTResult ( req . RequestID , text , true )
// Step 2: LLM 流式输出 + 句子切分
llmStream , _ := o . llm . ChatStream ( ctx , LLMRequest {
Image : req . Image , Text : text , Language : "zh-CN" ,
} )
sentenceCh := make ( chan string , 4 )
go func ( ) {
defer close ( sentenceCh )
var buf strings . Builder
var fullText strings . Builder
for chunk := range llmStream {
// 即时推送文字给客户端（逐 token 显示）
client . SendLLMChunk ( req . RequestID , chunk . Delta )
fullText . WriteString ( chunk . Delta )
buf . WriteString ( chunk . Delta )
// 遇到句子边界就吐出
if isSentenceEnd ( chunk . Delta ) {
sentenceCh <- buf . String ( )
buf . Reset ( )
}
}
// 最后一段不足一句的也吐出
if buf . Len ( ) > 0 {
sentenceCh <- buf . String ( )
}
// 推送 llm_done
client . SendLLMDone ( req . RequestID , fullText . String ( ) , chunk . TokensUsed , chunk . Model )
} ( )
// Step 3: TTS 并行消费句子流
ttsStream , _ := o . tts . SynthesizeStream ( ctx , sentenceCh , TTSOptions {
Voice : "alloy" , OutputFmt : "mp3" , SampleRate : 24000 ,
} )
for chunk := range ttsStream {
client . SendTTSAudio ( req . RequestID , chunk . Audio , chunk . IsLast )
}
}
// isSentenceEnd 判断 delta 中是否包含句子结束标志。
func isSentenceEnd ( delta string ) bool {
return strings . ContainsAny ( delta , "。！？\n.!?\n" )
}
```
### 并发控制
- 每个 `ProcessQuery` 调用在独立 goroutine 中运行
- `context.WithTimeout` 确保 10 秒总超时
- `interrupt` 消息触发 `cancel()` ，
- 同一 session 内同时只允许一个活跃请求，新请求自动取消上一个
### 错误处理与降级
| 故障点 | 处理策略 | 客户端表现 |
|--------|---------|-----------|
| STT 失败 | 发送 `STT_ERROR` ，终止本次请求 | 回退到纯文本模式 |
| LLM 超时（>10s） `LLM_TIMEOUT` ，取消 TTS | 提示用户重试 |
| LLM 部分输出后失败 | 已推送的 `llm_chunk` 保留，发送 `error` 通知中断 | 显示已收到的部分文字 |
| TTS 失败 | 静默跳过，`llm_done` 正常发送 | 只有文字回复，无语音 |
| TTS 部分失败 | 已推送的音频保留，后续句子跳过 | 部分句子有语音 |
| interrupt 打断 | cancel context，
---
## 五、Session Manager
WebSocket Handler 和 AI Orchestrator 之间的会话管理层。负责维护会话生命周期、对话上下文和配置状态。
### Redis 数据结构
每个会话在 Redis 中占 2 个 key：
```
session:{id}:history → List （对话历史）
```
**Hash — `session:{id}:meta` **
| field | 类型 | 示例 | 说明 |
|-------|------|------|------|
| `session_id` | string | `"550e8400-..."` | 主键冗余 |
| `config.tts_enabled` | string | `"true"` | Redis Hash 值均为 string |
| `config.detail_level` | string | `"low"` | |
| `config.language` | string | `"zh-CN"` | |
| `created_at` | string | `"2026-06-13T10:00:00Z"` | RFC3339 |
| `last_active` | string | `"2026-06-13T10:05:30Z"` | 每次消息刷新 |
| `active_request_id` | string | `"uuid"` 或 `""` | 当前处理中的请求 ID，
**List — `session:{id}:history` **
每个元素是一条 JSON 序列化的 Message：
``` json
{ "role" : "user" , "content" : "这是什么花？" }
{ "role" : "assistant" , "content" : "这是一朵红色的玫瑰。" }
```
- `LPUSH` 新消息到左头（最新在前）
- `LRANGE 0 {limit-1}` 取最近 N 轮
- `LTRIM 0 {max-1}` 限制总条数（默认保留最近 20 条 = 10 轮对话）
### TTL 策略
| 场景 | TTL | 说明 |
|------|-----|------|
| 创建时 | 30 分钟 | `EXPIRE` 设置 |
| 每次收到消息 | 重置 30 分钟 | `EXPIRE` 刷新 |
| WebSocket 断开 | 不主动删 | 等自然过期，支持重连恢复 |
| 超过 30 分钟无活动 | 自动过期 | Redis 自动清理 meta + history |
| 显式销毁（ ） `DEL` | 两个 key 一起删 |
### 接口定义
``` go
// SessionManager 会话管理器。
// WebSocket Handler 通过此接口操作会话，不直接接触 Redis。
type SessionManager interface {
// Create 创建新会话，返回 session ID。
Create ( ctx context . Context , config models . SessionConfig ) ( string , error )
// Get 获取会话（含 config）
Get ( ctx context . Context , sessionID string ) ( * models . Session , error )
// UpdateConfig 更新会话配置（
UpdateConfig ( ctx context . Context , sessionID string , patch models . SessionConfigPatch ) error
// GetHistory 获取最近 N 轮对话历史（供 Orchestrator 构建 LLM 上下文）。
GetHistory ( ctx context . Context , sessionID string , limit int ) ( [ ] models . Message , error )
// AppendMessage 追加一条对话消息，同时刷新 TTL。
AppendMessage ( ctx context . Context , sessionID string , msg models . Message ) error
// SetActiveRequest 标记当前正在处理的请求 ID（
SetActiveRequest ( ctx context . Context , sessionID string , requestID string ) error
// ClearActiveRequest 清除活跃请求标记（请求完成或中断后）。
ClearActiveRequest ( ctx context . Context , sessionID string ) error
// Touch 刷新 TTL（ ）
Touch ( ctx context . Context , sessionID string ) error
// Destroy 显式销毁会话（
Destroy ( ctx context . Context , sessionID string ) error
}
```
### WebSocket Handler 集成
``` go
// query 分支
case "query" :
var msg models . WsQuery
json . Unmarshal ( message , & msg )
sessionMgr . Touch ( ctx , sessionID ) // 刷新 TTL
sessionMgr . SetActiveRequest ( ctx , sessionID , msg . RequestID ) // 标记活跃请求
history , _ := sessionMgr . GetHistory ( ctx , sessionID , 20 ) // 获取对话上下文
go orchestrator . ProcessQuery ( ctx , client , & msg , history ) // 异步编排
// interrupt 分支
case "interrupt" :
reqID , _ := sessionMgr . GetActiveRequestID ( ctx , sessionID )
if reqID != "" {
cancelFunc ( reqID ) // 取消对应 context
sessionMgr . ClearActiveRequest ( ctx , sessionID )
}
// 连接断开
// 不调用 Destroy，
```
### MVP 内存实现
联调阶段无 Redis 时，用同一接口的内存实现：
``` go
type InMemorySessionManager struct {
mu sync . RWMutex
sessions map [ string ] * sessionEntry
}
type sessionEntry struct {
session models . Session
history [ ] models . Message
activeReqID string
}
```
注入时根据配置切换：
``` go
var sessionMgr SessionManager
if cfg . Redis . Addr != "" {
sessionMgr = NewRedisSessionManager ( redisClient , 30 * time . Minute , 20 )
} else {
sessionMgr = NewInMemorySessionManager ( )
}
```
---
## 六、配置管理
使用 Viper 加载配置，支持 YAML 文件 + 环境变量覆盖。**环境变量优先级高于配置文件**。
### 配置文件位置
```
backend/config.dev.yaml # 开发环境（
backend/config.prod.yaml # 生产环境
```
Viper 加载顺序：先读 `config.yaml` ，再根据 `APP_ENV` 环境变量尝试读 `config.{env}.yaml` 覆盖，最后所有环境变量自动覆盖对应字段。
### Go 配置结构体
``` go
// Config 应用配置。
type Config struct {
App AppConfig ` mapstructure:"app" `
Server ServerConfig ` mapstructure:"server" `
Redis RedisConfig ` mapstructure:"redis" `
AI AIConfig ` mapstructure:"ai" `
Storage StorageConfig ` mapstructure:"storage" `
Log LogConfig ` mapstructure:"log" `
}
type AppConfig struct {
Env string ` mapstructure:"env" ` // "dev" | "prod"，默认 "dev"
Version string ` mapstructure:"version" ` // 由编译时注入
}
type ServerConfig struct {
Host string ` mapstructure:"host" ` // 默认 "0.0.0.0"
Port int ` mapstructure:"port" ` // 默认 8080
ReadTimeout int ` mapstructure:"read_timeout" ` // 秒，默认 30
WriteTimeout int ` mapstructure:"write_timeout" ` // 秒，默认 30
}
type RedisConfig struct {
Addr string ` mapstructure:"addr" ` // "localhost:6379"
Password string ` mapstructure:"password" ` // 无密码留空
DB int ` mapstructure:"db" ` // 默认 0
}
type AIConfig struct {
STT STTConfig ` mapstructure:"stt" `
LLM LLMConfig ` mapstructure:"llm" `
TTS TTSConfig ` mapstructure:"tts" `
}
type STTConfig struct {
Provider string ` mapstructure:"provider" ` // "deepgram"
APIKey string ` mapstructure:"api_key" `
Endpoint string ` mapstructure:"endpoint" ` // 默认 "wss://api.deepgram.com/v1/listen"
}
type LLMConfig struct {
Provider string ` mapstructure:"provider" ` // "openai"
APIKey string ` mapstructure:"api_key" `
Model string ` mapstructure:"model" ` // 默认 "gpt-4o"
Endpoint string ` mapstructure:"endpoint" ` // 默认 "https://api.openai.com/v1"
Timeout int ` mapstructure:"timeout" ` // 秒，默认 10
}
type TTSConfig struct {
Provider string ` mapstructure:"provider" ` // "openai"
APIKey string ` mapstructure:"api_key" `
Voice string ` mapstructure:"voice" ` // 默认 "alloy"
Speed float64 ` mapstructure:"speed" ` // 默认 1.0
Endpoint string ` mapstructure:"endpoint" ` // 默认 "https://api.openai.com/v1"
Timeout int ` mapstructure:"timeout" ` // 秒，默认 5
}
type StorageConfig struct {
Driver string ` mapstructure:"driver" ` // "memory" | "postgres"
DSN string ` mapstructure:"dsn" ` // PostgreSQL 连接串，
}
type LogConfig struct {
Level string ` mapstructure:"level" ` // "debug" | "info" | "warn" | "error"，默认 "info"
Format string ` mapstructure:"format" ` // "json" | "console"，生产用 json
}
```
### 配置文件示例
``` yaml
# config.yaml — 所有环境共享的默认值
app :
env : dev
server :
host : "0.0.0.0"
port : 8080
read_timeout : 30
write_timeout : 30
redis :
addr : "localhost:6379"
password : ""
db : 0
ai :
stt :
provider : deepgram
endpoint : "wss://api.deepgram.com/v1/listen"
llm :
provider : openai
model : gpt-4o
endpoint : "https://api.openai.com/v1"
timeout : 10
tts :
provider : openai
voice : alloy
speed : 1.0
endpoint : "https://api.openai.com/v1"
timeout : 5
storage :
driver : memory
log :
level : info
format : console
```
### 环境变量覆盖规则
Viper 自动将配置项映射为环境变量，规则：**前缀 `CAMTALK_` + 路径大写用 `_` 连接**。
| 配置项 | 环境变量 | 示例 |
|--------|---------|------|
| `server.port` | `CAMTALK_SERVER_PORT` | `8080` |
| `redis.addr` | `CAMTALK_REDIS_ADDR` | `redis:6379` |
| `redis.password` | `CAMTALK_REDIS_PASSWORD` | — |
| `ai.stt.api_key` | `CAMTALK_AI_STT_API_KEY` | — |
| `ai.llm.api_key` | `CAMTALK_AI_LLM_API_KEY` | — |
| `ai.tts.api_key` | `CAMTALK_AI_TTS_API_KEY` | — |
| `ai.llm.model` | `CAMTALK_AI_LLM_MODEL` | `gpt-4o` |
| `storage.driver` | `CAMTALK_STORAGE_DRIVER` | `postgres` |
| `storage.dsn` | `CAMTALK_STORAGE_DSN` | — |
| `app.env` | `CAMTALK_APP_ENV` | `prod` |
| `log.level` | `CAMTALK_LOG_LEVEL` | `warn` |
| `log.format` | `CAMTALK_LOG_FORMAT` | `json` |
> API Key 和密码**只通过环境变量注入**，不写入配置文件，避免泄露到版本控制。
### 配置加载代码
``` go
// internal/config/config.go
func Load ( ) ( * Config , error ) {
v := viper . New ( )
// 1. 读默认配置文件
v . SetConfigName ( "config" )
v . SetConfigType ( "yaml" )
v . AddConfigPath ( "./config" ) // go run 时
v . AddConfigPath ( "." ) // 二进制运行时
if err := v . ReadInConfig ( ) ; err != nil {
if _ , ok := err . ( viper . ConfigFileNotFoundError ) ; ! ok {
return nil , fmt . Errorf ( "read config: %w" , err )
}
}
// 2. 按环境覆盖
env := os . Getenv ( "CAMTALK_APP_ENV" )
if env == "" {
env = "dev"
}
v . SetConfigName ( "config." + env )
v . MergeInConfig ( ) // 忽略文件不存在
// 3. 环境变量覆盖
v . SetEnvPrefix ( "CAMTALK" )
v . SetEnvKeyReplacer ( strings . NewReplacer ( "." , "_" ) )
v . AutomaticEnv ( )
// 4. 解析
var cfg Config
if err := v . Unmarshal ( & cfg ) ; err != nil {
return nil , fmt . Errorf ( "unmarshal config: %w" , err )
}
return & cfg , nil
}
```
### main.go 集成
``` go
func main ( ) {
cfg , err := config . Load ( )
if err != nil {
log . Fatalf ( "load config: %v" , err )
}
r := gin . Default ( )
// 使用 cfg.Server.Port 替代硬编码 :8080
addr := fmt . Sprintf ( "%s:%d" , cfg . Server . Host , cfg . Server . Port )
log . Printf ( "CamTalk gateway starting on %s (env=%s)" , addr , cfg . App . Env )
r . Run ( addr )
}
```
### 启动方式
``` bash
# 开发环境（默认 config.yaml，
CAMTALK_AI_LLM_API_KEY = sk-xxx \
CAMTALK_AI_STT_API_KEY = xxx \
# 生产环境
CAMTALK_APP_ENV = prod \
CAMTALK_REDIS_ADDR = redis:6379 \
CAMTALK_AI_LLM_API_KEY = sk-xxx \
CAMTALK_AI_STT_API_KEY = xxx \
CAMTALK_AI_TTS_API_KEY = xxx \
CAMTALK_STORAGE_DRIVER = postgres \
CAMTALK_STORAGE_DSN = "postgres://user:pass@db:5432/camtalk?sslmode=disable" \
CAMTALK_LOG_LEVEL = warn \
CAMTALK_LOG_FORMAT = json \
```
---
## 七、数据模型
> 注：
### Go 后端模型
@@ -326,7 +980,7 @@ type ClientMessage =
---
## 四 、扩展接口设计
## 八 、扩展接口设计
通过 Repository 接口隔离存储层，
@@ -402,7 +1056,7 @@ func NewApp(cfg *Config) *App {
---
## 五 、错误码
## 九 、错误码
| 错误码 | 含义 | 客户端处理建议 |
|--------|------|--------------|
@@ -417,7 +1071,7 @@ func NewApp(cfg *Config) *App {
| `TTS_ERROR` | 语音合成失败 | 静默回退到纯文本回复 |
| `INTERNAL_ERROR` | 服务端内部错误 | 提示用户重试 |
## 六 、连接管理
## 十 、连接管理
**心跳机制 ** ：客户端每 30 秒发送 `ping` ，服务端回复 `pong` 。超过 60 秒无 `ping` ，服务端判定连接断开并清理会话资源。
