XEngineers/CamTalk
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 将 Session Manager + Orchestrator 串入 WebSocket handler，实现端到端消息处理。 #38

Merged
huanghaosheng merged 3 commits from feature/phase5 into develop 2026-06-13 16:17:23 +08:00
Conversation 0 Commits 3 Files Changed 2 +114 -18
2 changed files with 114 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
backend/cmd/server/main.go
View File

@@ -10,8 +10,12 @@ import (
"github.com/gin-gonic/gin"
"github.com/hhs/camtalk/internal/ai/llm"
"github.com/hhs/camtalk/internal/ai/stt"
"github.com/hhs/camtalk/internal/ai/tts"
"github.com/hhs/camtalk/internal/config"
"github.com/hhs/camtalk/internal/logger"
"github.com/hhs/camtalk/internal/orchestrator"
"github.com/hhs/camtalk/internal/session"
"github.com/hhs/camtalk/internal/ws"
)
@@ -40,6 +44,14 @@ func main() {
sessionMgr = session.NewMemoryManager(30*time.Minute, 20)
defer sessionMgr.(*session.MemoryManager).Stop()
// 初始化 AI 服务
sttService := stt.NewDeepgramService(cfg.AI.STT.APIKey, cfg.AI.STT.Endpoint, logger.Log)
llmService := llm.NewOpenAIService(cfg.AI.LLM.APIKey, cfg.AI.LLM.Model, cfg.AI.LLM.Endpoint, cfg.AI.LLM.Timeout, logger.Log)
ttsService := tts.NewOpenAIService(cfg.AI.TTS.APIKey, cfg.AI.TTS.Voice, cfg.AI.TTS.Endpoint, cfg.AI.TTS.Speed, cfg.AI.TTS.Timeout, logger.Log)
// 初始化 Orchestrator
orch := orchestrator.New(sttService, llmService, ttsService, sessionMgr)
// Gin 模式
if cfg.App.Env == "prod" {
gin.SetMode(gin.ReleaseMode)
@@ -55,7 +67,7 @@ func main() {
}
// WebSocket
r.GET("/ws", ws.ServeWS(sessionMgr))
r.GET("/ws", ws.ServeWS(sessionMgr, orch))
// HTTP Server
srv := &http.Server{

112
backend/internal/ws/handler.go
View File

@@ -13,6 +13,7 @@ import (
"github.com/hhs/camtalk/internal/errors"
"github.com/hhs/camtalk/internal/logger"
"github.com/hhs/camtalk/internal/models"
"github.com/hhs/camtalk/internal/orchestrator"
"github.com/hhs/camtalk/internal/session"
)
@@ -24,6 +25,9 @@ var upgrader = websocket.Upgrader{
type Client struct {
conn *websocket.Conn
sessionID string
sessionMgr session.Manager
orchestrator orchestrator.Orchestrator
cancelFuncs map[string]context.CancelFunc // requestID → cancel func
mu sync.Mutex
}
@@ -34,14 +38,50 @@ func (c *Client) SendJSON(v any) error {
return c.conn.WriteJSON(v)
}
// WSClient 实现 orchestrator.Sender 接口，将消息推送到 WebSocket 连接。
type WSClient struct {
client *Client
requestID string
}
// SendSTTResult 发送语音识别结果。
func (w *WSClient) SendSTTResult(result models.WsSTTResult) error {
result.RequestID = w.requestID
return w.client.SendJSON(result)
}
// SendLLMChunk 发送 LLM 流式文本增量。
func (w *WSClient) SendLLMChunk(chunk models.WsLLMChunk) error {
chunk.RequestID = w.requestID
return w.client.SendJSON(chunk)
}
// SendLLMDone 发送 LLM 流结束信号。
func (w *WSClient) SendLLMDone(done models.WsLLMDone) error {
done.RequestID = w.requestID
return w.client.SendJSON(done)
}
// SendTTSAudio 发送 TTS 音频数据。
func (w *WSClient) SendTTSAudio(audio models.WsTTSAudio) error {
audio.RequestID = w.requestID
return w.client.SendJSON(audio)
}
// SendError 发送错误消息。
func (w *WSClient) SendError(err models.WsError) error {
err.RequestID = w.requestID
return w.client.SendJSON(err)
}
// ServeWS 处理 WebSocket 升级请求。
func ServeWS(sessionMgr session.Manager) gin.HandlerFunc {
func ServeWS(sessionMgr session.Manager, orch orchestrator.Orchestrator) gin.HandlerFunc {
return func(c *gin.Context) {
serveWS(c, sessionMgr)
serveWS(c, sessionMgr, orch)
}
}
func serveWS(c *gin.Context, sessionMgr session.Manager) {
func serveWS(c *gin.Context, sessionMgr session.Manager, orch orchestrator.Orchestrator) {
conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
if err != nil {
logger.Log.Errorw("websocket upgrade failed", "error", err)
@@ -56,7 +96,13 @@ func serveWS(c *gin.Context, sessionMgr session.Manager) {
return
}
client := &Client{conn: conn, sessionID: sessionID}
client := &Client{
conn: conn,
sessionID: sessionID,
sessionMgr: sessionMgr,
orchestrator: orch,
cancelFuncs: make(map[string]context.CancelFunc),
}
// 发送 connected 消息
_ = client.SendJSON(models.WsConnected{
@@ -124,19 +170,43 @@ func serveWS(c *gin.Context, sessionMgr session.Manager) {
logger.Log.Infow("query received", "session", sessionID, "request", msg.RequestID)
// 刷新会话 TTL
if err := sessionMgr.Touch(context.Background(), sessionID); err != nil {
if err := client.sessionMgr.Touch(context.Background(), sessionID); err != nil {
logger.Log.Warnw("touch session failed", "session", sessionID, "error", err)
}
// 标记活跃请求
if err := sessionMgr.SetActiveRequest(context.Background(), sessionID, msg.RequestID); err != nil {
if err := client.sessionMgr.SetActiveRequest(context.Background(), sessionID, msg.RequestID); err != nil {
logger.Log.Warnw("set active request failed", "session", sessionID, "error", err)
}
// 获取对话历史（供后续 Orchestrator 使用）
_, _ = sessionMgr.GetHistory(context.Background(), sessionID, 20)
// 获取对话历史
history, _ := client.sessionMgr.GetHistory(context.Background(), sessionID, 20)
// TODO: 解码 audio Base64 → 启动 orchestrator.ProcessQuery goroutine
// 创建可取消的 context
ctx, cancel := context.WithCancel(context.Background())
client.mu.Lock()
client.cancelFuncs[msg.RequestID] = cancel
client.mu.Unlock()
// 创建 sender
sender := &WSClient{client: client, requestID: msg.RequestID}
// 启动 orchestrator 处理 goroutine
go func() {
defer func() {
// 清理 cancel func
client.mu.Lock()
delete(client.cancelFuncs, msg.RequestID)
client.mu.Unlock()
cancel()
// 清除活跃请求
_ = client.sessionMgr.ClearActiveRequest(context.Background(), sessionID)
}()
if err := client.orchestrator.ProcessQuery(ctx, sessionID, msg, history, sender); err != nil {
logger.Log.Errorw("process query failed", "session", sessionID, "request", msg.RequestID, "error", err)
}
}()
case "config":
var msg models.WsConfig
@@ -150,7 +220,7 @@ func serveWS(c *gin.Context, sessionMgr session.Manager) {
DetailLevel: msg.Payload.DetailLevel,
Language: msg.Payload.Language,
}
if err := sessionMgr.UpdateConfig(context.Background(), sessionID, patch); err != nil {
if err := client.sessionMgr.UpdateConfig(context.Background(), sessionID, patch); err != nil {
errors.SendWSError(client, errors.CodeInternalError, "", err)
continue
}
@@ -159,11 +229,16 @@ func serveWS(c *gin.Context, sessionMgr session.Manager) {
case "interrupt":
logger.Log.Infow("interrupt received", "session", sessionID)
// 获取活跃请求 ID（实际 cancel 在 Phase 5 接入 orchestrator 后实现）
reqID, _ := sessionMgr.GetActiveRequestID(context.Background(), sessionID)
// 获取活跃请求 ID 并取消
reqID, _ := client.sessionMgr.GetActiveRequestID(context.Background(), sessionID)
if reqID != "" {
_ = sessionMgr.ClearActiveRequest(context.Background(), sessionID)
// TODO: 取消对应 context cancel func
client.mu.Lock()
if cancel, ok := client.cancelFuncs[reqID]; ok {
cancel()
delete(client.cancelFuncs, reqID)
}
client.mu.Unlock()
_ = client.sessionMgr.ClearActiveRequest(context.Background(), sessionID)
}
default:
@@ -177,6 +252,15 @@ func serveWS(c *gin.Context, sessionMgr session.Manager) {
close(done)
// 取消所有活跃请求
client.mu.Lock()
for reqID, cancel := range client.cancelFuncs {
logger.Log.Infow("canceling active request on disconnect", "session", sessionID, "request", reqID)
cancel()
}
client.cancelFuncs = make(map[string]context.CancelFunc)
client.mu.Unlock()
// 断开连接时不销毁会话，让其自然过期（支持重连恢复）
logger.Log.Infow("client disconnected", "session", sessionID)
}