From 19535e6a3735fc79211728ecff37e82448c034a1 Mon Sep 17 00:00:00 2001 From: hhs <386998068@qq.com> Date: Sat, 13 Jun 2026 16:13:22 +0800 Subject: [PATCH 1/3] =?UTF-8?q?feat:=20Phase=205.1=20-=20Client=20?= =?UTF-8?q?=E6=89=A9=E5=B1=95=EF=BC=8C=E6=B7=BB=E5=8A=A0=20orchestrator=20?= =?UTF-8?q?=E4=BE=9D=E8=B5=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 扩展 Client 结构体，添加 sessionManager、orchestrator、cancelFuncs 字段 - 创建 WSClient 适配器实现 orchestrator.Sender 接口 - 更新 ServeWS 函数签名，接收 orchestrator 参数 - 在 main.go 中初始化 AI 服务和 Orchestrator --- backend/cmd/server/main.go | 14 +++++++- backend/internal/ws/handler.go | 60 ++++++++++++++++++++++++++++++---- 2 files changed, 66 insertions(+), 8 deletions(-) diff --git a/backend/cmd/server/main.go b/backend/cmd/server/main.go index a475fa4..61edbaf 100644 --- a/backend/cmd/server/main.go +++ b/backend/cmd/server/main.go @@ -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{ diff --git a/backend/internal/ws/handler.go b/backend/internal/ws/handler.go index 83316b4..4c77ff6 100644 --- a/backend/internal/ws/handler.go +++ b/backend/internal/ws/handler.go @@ -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" ) @@ -22,9 +23,12 @@ var upgrader = websocket.Upgrader{ // Client 代表一个 WebSocket 客户端连接。 type Client struct { - conn *websocket.Conn - sessionID string - mu sync.Mutex + conn *websocket.Conn + sessionID string + sessionMgr session.Manager + orchestrator orchestrator.Orchestrator + cancelFuncs map[string]context.CancelFunc // requestID → cancel func + mu sync.Mutex } // SendJSON 向客户端发送 JSON 消息（公开以便 errors 包调用）。 @@ -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{ -- 2.49.1 From 314554ded7821f84fb4563b73828d3eebc121d1e Mon Sep 17 00:00:00 2001 From: hhs <386998068@qq.com> Date: Sat, 13 Jun 2026 16:14:20 +0800 Subject: [PATCH 2/3] =?UTF-8?q?feat:=20Phase=205.2=20-=20=E5=AE=9E?= =?UTF-8?q?=E7=8E=B0=20query=20=E5=A4=84=E7=90=86=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 创建可取消的 context 并存储 cancel func - 获取对话历史并传递给 orchestrator - 启动 orchestrator.ProcessQuery goroutine 处理查询 - 请求完成后自动清理 cancel func 和活跃请求标记 --- backend/internal/ws/handler.go | 34 +++++++++++++++++++++++++++++----- 1 file changed, 29 insertions(+), 5 deletions(-) diff --git a/backend/internal/ws/handler.go b/backend/internal/ws/handler.go index 4c77ff6..52f90e5 100644 --- a/backend/internal/ws/handler.go +++ b/backend/internal/ws/handler.go @@ -170,19 +170,43 @@ func serveWS(c *gin.Context, sessionMgr session.Manager, orch orchestrator.Orche 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 -- 2.49.1 From 085e19038b388bdd6cec0f6cc745346abc0f7a78 Mon Sep 17 00:00:00 2001 From: hhs <386998068@qq.com> Date: Sat, 13 Jun 2026 16:15:17 +0800 Subject: [PATCH 3/3] =?UTF-8?q?fix:=20Phase=205.3=20-=20=E7=BB=9F=E4=B8=80?= =?UTF-8?q?=E4=BD=BF=E7=94=A8=20client.sessionMgr=20=E8=B0=83=E7=94=A8=20U?= =?UTF-8?q?pdateConfig?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/ws/handler.go | 24 +++++++++++++++++++----- 1 file changed, 19 insertions(+), 5 deletions(-) diff --git a/backend/internal/ws/handler.go b/backend/internal/ws/handler.go index 52f90e5..eb84af9 100644 --- a/backend/internal/ws/handler.go +++ b/backend/internal/ws/handler.go @@ -220,7 +220,7 @@ func serveWS(c *gin.Context, sessionMgr session.Manager, orch orchestrator.Orche 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 } @@ -229,11 +229,16 @@ func serveWS(c *gin.Context, sessionMgr session.Manager, orch orchestrator.Orche 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: @@ -247,6 +252,15 @@ func serveWS(c *gin.Context, sessionMgr session.Manager, orch orchestrator.Orche 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) } -- 2.49.1