From ad700743efb464a4442c524d3ff1d08e59aebabd Mon Sep 17 00:00:00 2001 From: hhs <386998068@qq.com> Date: Sun, 21 Jun 2026 22:33:49 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20Eino=20adapter=20=E5=92=8C=20callback?= =?UTF-8?q?=20=E8=BF=81=E7=A7=BB=E5=88=B0=20trace=20=E5=8C=85=EF=BC=88Phas?= =?UTF-8?q?e=206.1=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 移除 adapter.go 中的 ctxKeySessionID 定义 - 移除 callback.go 中的 ctxKeyRequestID 定义 - 统一使用 trace.WithSessionID/WithRequestID - adapter.go 使用 trace.FromContext 替换 logger.Log - callback.go 使用 trace.FromContext - 移除双重日志，SetActiveRequest 失败直接返回错误 - 更新测试文件导入 trace 包 --- backend/internal/eino/adapter.go | 35 ++++++++++------------------- backend/internal/eino/callback.go | 14 +++++------- backend/internal/eino/graph_test.go | 3 ++- 3 files changed, 20 insertions(+), 32 deletions(-) diff --git a/backend/internal/eino/adapter.go b/backend/internal/eino/adapter.go index 36e8583..665b391 100644 --- a/backend/internal/eino/adapter.go +++ b/backend/internal/eino/adapter.go @@ -8,20 +8,12 @@ import ( "github.com/cloudwego/eino/compose" - "github.com/hhs/camtalk/internal/logger" "github.com/hhs/camtalk/internal/models" "github.com/hhs/camtalk/internal/orchestrator" "github.com/hhs/camtalk/internal/session" + "github.com/hhs/camtalk/internal/trace" ) -// ctxKeySessionID sessionID 的 context key。 -type ctxKeySessionID struct{} - -// WithSessionID 将 sessionID 注入 context。 -func WithSessionID(ctx context.Context, sessionID string) context.Context { - return context.WithValue(ctx, ctxKeySessionID{}, sessionID) -} - // EinoOrchestrator 实现 orchestrator.Orchestrator 接口。 // 将 Eino Graph 包装为现有接口，WS Handler 几乎不用改。 type EinoOrchestrator struct { @@ -48,19 +40,19 @@ func (e *EinoOrchestrator) ProcessQuery( req models.WsQuery, sender orchestrator.Sender, ) error { - log := logger.Log + log := trace.FromContext(ctx) startTime := time.Now() // 1. 设置活跃请求 if err := e.sessionMgr.SetActiveRequest(ctx, sessionID, req.RequestID); err != nil { - log.Errorw("设置活跃请求失败", "error", err) + return err } defer e.sessionMgr.ClearActiveRequest(ctx, sessionID) // 2. 获取会话配置 sess, err := e.sessionMgr.Get(ctx, sessionID) if err != nil { - log.Errorw("获取会话失败", "error", err) + log.Errorw("get session failed", "error", err) sender.SendError(models.WsError{ Type: "error", RequestID: req.RequestID, @@ -75,7 +67,7 @@ func (e *EinoOrchestrator) ProcessQuery( if req.Text == "" && req.Audio != "" { audioData, err = base64.StdEncoding.DecodeString(req.Audio) if err != nil { - log.Errorw("音频解码失败", "error", err) + log.Errorw("audio decode failed", "error", err) sender.SendError(models.WsError{ Type: "error", RequestID: req.RequestID, @@ -90,7 +82,7 @@ func (e *EinoOrchestrator) ProcessQuery( if req.Image != "" { imageData, err = base64.StdEncoding.DecodeString(req.Image) if err != nil { - log.Errorw("图片解码失败", "error", err) + log.Errorw("image decode failed", "error", err) sender.SendError(models.WsError{ Type: "error", RequestID: req.RequestID, @@ -107,7 +99,7 @@ func (e *EinoOrchestrator) ProcessQuery( // 5. 注入 context 值（供 Callback 和 Lambda 节点使用） ctx = WithSender(ctx, sender) ctx = WithRequestID(ctx, req.RequestID) - ctx = WithSessionID(ctx, sessionID) + ctx = trace.WithSessionID(ctx, sessionID) ctx = WithStartTime(ctx, startTime) // 创建 State 并从 input 复制元数据 @@ -125,7 +117,7 @@ func (e *EinoOrchestrator) ProcessQuery( // 6. 调用 Graph（Stream 模式 + 运行时 Callback） streamReader, err := e.graph.Runnable.Stream(ctx, input, e.callbacks) if err != nil { - log.Errorw("Graph Stream 启动失败", "error", err) + log.Errorw("graph stream start failed", "error", err) sender.SendError(models.WsError{ Type: "error", RequestID: req.RequestID, @@ -143,7 +135,7 @@ func (e *EinoOrchestrator) ProcessQuery( if err == io.EOF { break } - log.Errorw("Graph Stream 消费错误", "error", err) + log.Errorw("graph stream consume error", "error", err) break } output = o @@ -159,7 +151,7 @@ func (e *EinoOrchestrator) ProcessQuery( Role: "user", Content: userText, }); err != nil { - log.Errorw("追加用户消息到历史失败", "session", sessionID, "error", err) + log.Errorw("append user message failed", "error", err) } } @@ -169,15 +161,12 @@ func (e *EinoOrchestrator) ProcessQuery( Role: "assistant", Content: output.FullResponse, }); err != nil { - log.Errorw("追加助手消息到历史失败", "session", sessionID, "error", err) + log.Errorw("append assistant message failed", "error", err) } } latency := time.Since(startTime).Milliseconds() - log.Infow("Eino 编排完成", - "request_id", req.RequestID, - "latency_ms", latency, - "session_id", sessionID) + log.Infow("eino pipeline completed", "latency_ms", latency) return nil } diff --git a/backend/internal/eino/callback.go b/backend/internal/eino/callback.go index 880c003..d65799e 100644 --- a/backend/internal/eino/callback.go +++ b/backend/internal/eino/callback.go @@ -9,14 +9,13 @@ import ( "github.com/cloudwego/eino/schema" callbacksHelper "github.com/cloudwego/eino/utils/callbacks" - "github.com/hhs/camtalk/internal/logger" "github.com/hhs/camtalk/internal/models" "github.com/hhs/camtalk/internal/orchestrator" + "github.com/hhs/camtalk/internal/trace" ) // context key 类型，避免与其他包冲突。 type ctxKeySender struct{} -type ctxKeyRequestID struct{} type ctxKeyState struct{} // WithSender 将 Sender 注入 context。 @@ -24,9 +23,9 @@ func WithSender(ctx context.Context, sender orchestrator.Sender) context.Context return context.WithValue(ctx, ctxKeySender{}, sender) } -// WithRequestID 将 requestID 注入 context。 +// WithRequestID 将 requestID 注入 context（使用 trace 包）。 func WithRequestID(ctx context.Context, requestID string) context.Context { - return context.WithValue(ctx, ctxKeyRequestID{}, requestID) + return trace.WithRequestID(ctx, requestID) } // WithPipelineState 将 PipelineState 注入 context。 @@ -40,10 +39,9 @@ func senderFromCtx(ctx context.Context) orchestrator.Sender { return s } -// requestIDFromCtx 从 context 获取 requestID。 +// requestIDFromCtx 从 context 获取 requestID（使用 trace 包）。 func requestIDFromCtx(ctx context.Context) string { - s, _ := ctx.Value(ctxKeyRequestID{}).(string) - return s + return trace.GetRequestID(ctx) } // stateFromCtx 从 context 获取 PipelineState。 @@ -62,7 +60,7 @@ func BuildCallbackHandler() callbacks.Handler { return callbacksHelper.NewHandlerHelper(). ChatModel(&callbacksHelper.ModelCallbackHandler{ OnEndWithStreamOutput: func(ctx context.Context, info *callbacks.RunInfo, output *schema.StreamReader[*model.CallbackOutput]) context.Context { - log := logger.Log + log := trace.FromContext(ctx) sender := senderFromCtx(ctx) requestID := requestIDFromCtx(ctx) state := stateFromCtx(ctx) diff --git a/backend/internal/eino/graph_test.go b/backend/internal/eino/graph_test.go index 72252ce..a5ed074 100644 --- a/backend/internal/eino/graph_test.go +++ b/backend/internal/eino/graph_test.go @@ -13,6 +13,7 @@ import ( "github.com/hhs/camtalk/internal/ai/tts" "github.com/hhs/camtalk/internal/models" "github.com/hhs/camtalk/internal/orchestrator" + "github.com/hhs/camtalk/internal/trace" ) // --- Mock STT Service --- @@ -176,7 +177,7 @@ func TestContextInjection(t *testing.T) { sender := &mockSender{} ctx = WithSender(ctx, sender) ctx = WithRequestID(ctx, "req-123") - ctx = WithSessionID(ctx, "sess-456") + ctx = trace.WithSessionID(ctx, "sess-456") ctx = WithStartTime(ctx, time.Now()) ctx = WithPipelineState(ctx, genLocalState(ctx))