feat: Eino adapter 和 callback 迁移到 trace 包(Phase 6.1)

- 移除 adapter.go 中的 ctxKeySessionID 定义
- 移除 callback.go 中的 ctxKeyRequestID 定义
- 统一使用 trace.WithSessionID/WithRequestID
- adapter.go 使用 trace.FromContext 替换 logger.Log
- callback.go 使用 trace.FromContext
- 移除双重日志,SetActiveRequest 失败直接返回错误
- 更新测试文件导入 trace 包
This commit is contained in:
hhs
2026-06-21 22:33:49 +08:00
parent 6ab4776e08
commit ad700743ef
3 changed files with 20 additions and 32 deletions

View File

@@ -8,20 +8,12 @@ import (
"github.com/cloudwego/eino/compose" "github.com/cloudwego/eino/compose"
"github.com/hhs/camtalk/internal/logger"
"github.com/hhs/camtalk/internal/models" "github.com/hhs/camtalk/internal/models"
"github.com/hhs/camtalk/internal/orchestrator" "github.com/hhs/camtalk/internal/orchestrator"
"github.com/hhs/camtalk/internal/session" "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 接口。 // EinoOrchestrator 实现 orchestrator.Orchestrator 接口。
// 将 Eino Graph 包装为现有接口WS Handler 几乎不用改。 // 将 Eino Graph 包装为现有接口WS Handler 几乎不用改。
type EinoOrchestrator struct { type EinoOrchestrator struct {
@@ -48,19 +40,19 @@ func (e *EinoOrchestrator) ProcessQuery(
req models.WsQuery, req models.WsQuery,
sender orchestrator.Sender, sender orchestrator.Sender,
) error { ) error {
log := logger.Log log := trace.FromContext(ctx)
startTime := time.Now() startTime := time.Now()
// 1. 设置活跃请求 // 1. 设置活跃请求
if err := e.sessionMgr.SetActiveRequest(ctx, sessionID, req.RequestID); err != nil { if err := e.sessionMgr.SetActiveRequest(ctx, sessionID, req.RequestID); err != nil {
log.Errorw("设置活跃请求失败", "error", err) return err
} }
defer e.sessionMgr.ClearActiveRequest(ctx, sessionID) defer e.sessionMgr.ClearActiveRequest(ctx, sessionID)
// 2. 获取会话配置 // 2. 获取会话配置
sess, err := e.sessionMgr.Get(ctx, sessionID) sess, err := e.sessionMgr.Get(ctx, sessionID)
if err != nil { if err != nil {
log.Errorw("获取会话失败", "error", err) log.Errorw("get session failed", "error", err)
sender.SendError(models.WsError{ sender.SendError(models.WsError{
Type: "error", Type: "error",
RequestID: req.RequestID, RequestID: req.RequestID,
@@ -75,7 +67,7 @@ func (e *EinoOrchestrator) ProcessQuery(
if req.Text == "" && req.Audio != "" { if req.Text == "" && req.Audio != "" {
audioData, err = base64.StdEncoding.DecodeString(req.Audio) audioData, err = base64.StdEncoding.DecodeString(req.Audio)
if err != nil { if err != nil {
log.Errorw("音频解码失败", "error", err) log.Errorw("audio decode failed", "error", err)
sender.SendError(models.WsError{ sender.SendError(models.WsError{
Type: "error", Type: "error",
RequestID: req.RequestID, RequestID: req.RequestID,
@@ -90,7 +82,7 @@ func (e *EinoOrchestrator) ProcessQuery(
if req.Image != "" { if req.Image != "" {
imageData, err = base64.StdEncoding.DecodeString(req.Image) imageData, err = base64.StdEncoding.DecodeString(req.Image)
if err != nil { if err != nil {
log.Errorw("图片解码失败", "error", err) log.Errorw("image decode failed", "error", err)
sender.SendError(models.WsError{ sender.SendError(models.WsError{
Type: "error", Type: "error",
RequestID: req.RequestID, RequestID: req.RequestID,
@@ -107,7 +99,7 @@ func (e *EinoOrchestrator) ProcessQuery(
// 5. 注入 context 值(供 Callback 和 Lambda 节点使用) // 5. 注入 context 值(供 Callback 和 Lambda 节点使用)
ctx = WithSender(ctx, sender) ctx = WithSender(ctx, sender)
ctx = WithRequestID(ctx, req.RequestID) ctx = WithRequestID(ctx, req.RequestID)
ctx = WithSessionID(ctx, sessionID) ctx = trace.WithSessionID(ctx, sessionID)
ctx = WithStartTime(ctx, startTime) ctx = WithStartTime(ctx, startTime)
// 创建 State 并从 input 复制元数据 // 创建 State 并从 input 复制元数据
@@ -125,7 +117,7 @@ func (e *EinoOrchestrator) ProcessQuery(
// 6. 调用 GraphStream 模式 + 运行时 Callback // 6. 调用 GraphStream 模式 + 运行时 Callback
streamReader, err := e.graph.Runnable.Stream(ctx, input, e.callbacks) streamReader, err := e.graph.Runnable.Stream(ctx, input, e.callbacks)
if err != nil { if err != nil {
log.Errorw("Graph Stream 启动失败", "error", err) log.Errorw("graph stream start failed", "error", err)
sender.SendError(models.WsError{ sender.SendError(models.WsError{
Type: "error", Type: "error",
RequestID: req.RequestID, RequestID: req.RequestID,
@@ -143,7 +135,7 @@ func (e *EinoOrchestrator) ProcessQuery(
if err == io.EOF { if err == io.EOF {
break break
} }
log.Errorw("Graph Stream 消费错误", "error", err) log.Errorw("graph stream consume error", "error", err)
break break
} }
output = o output = o
@@ -159,7 +151,7 @@ func (e *EinoOrchestrator) ProcessQuery(
Role: "user", Role: "user",
Content: userText, Content: userText,
}); err != nil { }); 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", Role: "assistant",
Content: output.FullResponse, Content: output.FullResponse,
}); err != nil { }); err != nil {
log.Errorw("追加助手消息到历史失败", "session", sessionID, "error", err) log.Errorw("append assistant message failed", "error", err)
} }
} }
latency := time.Since(startTime).Milliseconds() latency := time.Since(startTime).Milliseconds()
log.Infow("Eino 编排完成", log.Infow("eino pipeline completed", "latency_ms", latency)
"request_id", req.RequestID,
"latency_ms", latency,
"session_id", sessionID)
return nil return nil
} }

View File

@@ -9,14 +9,13 @@ import (
"github.com/cloudwego/eino/schema" "github.com/cloudwego/eino/schema"
callbacksHelper "github.com/cloudwego/eino/utils/callbacks" callbacksHelper "github.com/cloudwego/eino/utils/callbacks"
"github.com/hhs/camtalk/internal/logger"
"github.com/hhs/camtalk/internal/models" "github.com/hhs/camtalk/internal/models"
"github.com/hhs/camtalk/internal/orchestrator" "github.com/hhs/camtalk/internal/orchestrator"
"github.com/hhs/camtalk/internal/trace"
) )
// context key 类型,避免与其他包冲突。 // context key 类型,避免与其他包冲突。
type ctxKeySender struct{} type ctxKeySender struct{}
type ctxKeyRequestID struct{}
type ctxKeyState struct{} type ctxKeyState struct{}
// WithSender 将 Sender 注入 context。 // WithSender 将 Sender 注入 context。
@@ -24,9 +23,9 @@ func WithSender(ctx context.Context, sender orchestrator.Sender) context.Context
return context.WithValue(ctx, ctxKeySender{}, sender) return context.WithValue(ctx, ctxKeySender{}, sender)
} }
// WithRequestID 将 requestID 注入 context。 // WithRequestID 将 requestID 注入 context(使用 trace 包)
func WithRequestID(ctx context.Context, requestID string) context.Context { func WithRequestID(ctx context.Context, requestID string) context.Context {
return context.WithValue(ctx, ctxKeyRequestID{}, requestID) return trace.WithRequestID(ctx, requestID)
} }
// WithPipelineState 将 PipelineState 注入 context。 // WithPipelineState 将 PipelineState 注入 context。
@@ -40,10 +39,9 @@ func senderFromCtx(ctx context.Context) orchestrator.Sender {
return s return s
} }
// requestIDFromCtx 从 context 获取 requestID。 // requestIDFromCtx 从 context 获取 requestID(使用 trace 包)
func requestIDFromCtx(ctx context.Context) string { func requestIDFromCtx(ctx context.Context) string {
s, _ := ctx.Value(ctxKeyRequestID{}).(string) return trace.GetRequestID(ctx)
return s
} }
// stateFromCtx 从 context 获取 PipelineState。 // stateFromCtx 从 context 获取 PipelineState。
@@ -62,7 +60,7 @@ func BuildCallbackHandler() callbacks.Handler {
return callbacksHelper.NewHandlerHelper(). return callbacksHelper.NewHandlerHelper().
ChatModel(&callbacksHelper.ModelCallbackHandler{ ChatModel(&callbacksHelper.ModelCallbackHandler{
OnEndWithStreamOutput: func(ctx context.Context, info *callbacks.RunInfo, output *schema.StreamReader[*model.CallbackOutput]) context.Context { 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) sender := senderFromCtx(ctx)
requestID := requestIDFromCtx(ctx) requestID := requestIDFromCtx(ctx)
state := stateFromCtx(ctx) state := stateFromCtx(ctx)

View File

@@ -13,6 +13,7 @@ import (
"github.com/hhs/camtalk/internal/ai/tts" "github.com/hhs/camtalk/internal/ai/tts"
"github.com/hhs/camtalk/internal/models" "github.com/hhs/camtalk/internal/models"
"github.com/hhs/camtalk/internal/orchestrator" "github.com/hhs/camtalk/internal/orchestrator"
"github.com/hhs/camtalk/internal/trace"
) )
// --- Mock STT Service --- // --- Mock STT Service ---
@@ -176,7 +177,7 @@ func TestContextInjection(t *testing.T) {
sender := &mockSender{} sender := &mockSender{}
ctx = WithSender(ctx, sender) ctx = WithSender(ctx, sender)
ctx = WithRequestID(ctx, "req-123") ctx = WithRequestID(ctx, "req-123")
ctx = WithSessionID(ctx, "sess-456") ctx = trace.WithSessionID(ctx, "sess-456")
ctx = WithStartTime(ctx, time.Now()) ctx = WithStartTime(ctx, time.Now())
ctx = WithPipelineState(ctx, genLocalState(ctx)) ctx = WithPipelineState(ctx, genLocalState(ctx))