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

feat: Eino nodes 迁移到 trace 包（Phase 6.2）

Browse Source 
- nodes_stt.go 使用 trace.FromContext 替换 logger.Log
- nodes_history.go 使用 trace.FromContext
- nodes_tts.go 使用 trace.FromContext
- nodes_done.go 使用 trace.FromContext
- 移除所有 nodes 中的 request_id 手动字段（自动附加）
- 所有日志消息改为英文
This commit is contained in:
hhs
2026-06-21 22:36:15 +08:00
parent ad700743ef
commit 76d331c885
4 changed files with 24 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
backend/internal/eino/nodes_done.go
View File

@@ -6,8 +6,8 @@ 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/trace"
) )
// ctxKeyStartTime 请求开始时间的 context key。 // ctxKeyStartTime 请求开始时间的 context key。
@@ -33,7 +33,7 @@ func latencyFromCtx(ctx context.Context) int64 {
// 历史消息追加由适配器负责（避免重复写入）。 // 历史消息追加由适配器负责（避免重复写入）。
func NewDoneLambda(defaultModel string) *compose.Lambda { func NewDoneLambda(defaultModel string) *compose.Lambda {
return compose.InvokableLambda(func(ctx context.Context, _ struct{}) (PipelineOutput, error) { return compose.InvokableLambda(func(ctx context.Context, _ struct{}) (PipelineOutput, error) {
log := logger.Log log := trace.FromContext(ctx)
sender := senderFromCtx(ctx) sender := senderFromCtx(ctx)
state := stateFromCtx(ctx) state := stateFromCtx(ctx)
@@ -70,13 +70,11 @@ func NewDoneLambda(defaultModel string) *compose.Lambda {
} }
} }
if err := sender.SendLLMDone(done); err != nil { if err := sender.SendLLMDone(done); err != nil {
log.Errorw("发送 llm_done 失败", "error", err) log.Errorw("send llm_done failed", "error", err)
} }
} }
log.Infow("查询处理完成", log.Infow("query processing completed", "response_length", len(fullResponse))
"request_id", requestID,
"response_length", len(fullResponse))
return PipelineOutput{ return PipelineOutput{
TranscribedText: transcribedText, TranscribedText: transcribedText,

13
backend/internal/eino/nodes_history.go
View File

@@ -8,9 +8,9 @@ import (
"github.com/cloudwego/eino/schema" "github.com/cloudwego/eino/schema"
"github.com/hhs/camtalk/internal/ai/llm" "github.com/hhs/camtalk/internal/ai/llm"
"github.com/hhs/camtalk/internal/logger"
"github.com/hhs/camtalk/internal/models" "github.com/hhs/camtalk/internal/models"
"github.com/hhs/camtalk/internal/store" "github.com/hhs/camtalk/internal/store"
"github.com/hhs/camtalk/internal/trace"
) )
// NewHistoryLambda 创建历史组装 Lambda 节点。 // NewHistoryLambda 创建历史组装 Lambda 节点。
@@ -24,7 +24,7 @@ func NewHistoryLambda(
maxHistory int, maxHistory int,
) *compose.Lambda { ) *compose.Lambda {
return compose.InvokableLambda(func(ctx context.Context, sttOut STTOutput) ([]*schema.Message, error) { return compose.InvokableLambda(func(ctx context.Context, sttOut STTOutput) ([]*schema.Message, error) {
log := logger.Log log := trace.FromContext(ctx)
// 从 State 读取请求元数据 // 从 State 读取请求元数据
state := stateFromCtx(ctx) state := stateFromCtx(ctx)
@@ -48,7 +48,7 @@ func NewHistoryLambda(
if userID != "" && scenarioRepo != nil { if userID != "" && scenarioRepo != nil {
scenarios, err := scenarioRepo.FindByUserID(ctx, userID) scenarios, err := scenarioRepo.FindByUserID(ctx, userID)
if err != nil { if err != nil {
log.Warnw("加载用户自建情景失败", "user_id", userID, "error", err) log.Warnw("load user scenarios failed", "user_id", userID, "error", err)
} else if len(scenarios) > 0 { } else if len(scenarios) > 0 {
customScenarios = make(map[string]string, len(scenarios)) customScenarios = make(map[string]string, len(scenarios))
customGreetings = make(map[string]string, len(scenarios)) customGreetings = make(map[string]string, len(scenarios))
@@ -58,7 +58,7 @@ func NewHistoryLambda(
customGreetings[s.ID] = s.Greeting customGreetings[s.ID] = s.Greeting
} }
} }
log.Debugw("加载用户自建情景", "user_id", userID, "count", len(scenarios)) log.Debugw("loaded user scenarios", "user_id", userID, "count", len(scenarios))
} }
} }
@@ -78,7 +78,7 @@ func NewHistoryLambda(
if historyFetcher != nil && sessionID != "" { if historyFetcher != nil && sessionID != "" {
history, err := historyFetcher(ctx, sessionID, maxHistory) history, err := historyFetcher(ctx, sessionID, maxHistory)
if err != nil { if err != nil {
log.Warnw("获取历史消息失败，继续处理", "error", err, "request_id", requestID) log.Warnw("fetch history failed, continuing", "error", err, "request_id", requestID)
} else { } else {
for _, msg := range history { for _, msg := range history {
messages = append(messages, &schema.Message{ messages = append(messages, &schema.Message{
@@ -121,8 +121,7 @@ func NewHistoryLambda(
}) })
} }
log.Infow("历史组装完成", log.Infow("history assembled",
"request_id", requestID,
"message_count", len(messages), "message_count", len(messages),
"has_image", len(imageData) > 0, "has_image", len(imageData) > 0,
"scenario", scenario) "scenario", scenario)

17
backend/internal/eino/nodes_stt.go
View File

@@ -8,8 +8,8 @@ import (
"github.com/cloudwego/eino/compose" "github.com/cloudwego/eino/compose"
"github.com/hhs/camtalk/internal/ai/stt" "github.com/hhs/camtalk/internal/ai/stt"
"github.com/hhs/camtalk/internal/logger"
"github.com/hhs/camtalk/internal/models" "github.com/hhs/camtalk/internal/models"
"github.com/hhs/camtalk/internal/trace"
"github.com/hhs/camtalk/internal/util" "github.com/hhs/camtalk/internal/util"
) )
@@ -21,7 +21,7 @@ import (
// 识别结果通过 Sender 发送 stt_result 到客户端。 // 识别结果通过 Sender 发送 stt_result 到客户端。
func NewSTTLambda(sttService stt.Service) *compose.Lambda { func NewSTTLambda(sttService stt.Service) *compose.Lambda {
return compose.InvokableLambda(func(ctx context.Context, input PipelineInput) (STTOutput, error) { return compose.InvokableLambda(func(ctx context.Context, input PipelineInput) (STTOutput, error) {
log := logger.Log log := trace.FromContext(ctx)
sender := senderFromCtx(ctx) sender := senderFromCtx(ctx)
requestID := requestIDFromCtx(ctx) requestID := requestIDFromCtx(ctx)
@@ -41,7 +41,6 @@ func NewSTTLambda(sttService stt.Service) *compose.Lambda {
// 文本输入模式：跳过 STT // 文本输入模式：跳过 STT
if input.Text != "" { if input.Text != "" {
log.Debugw("text input mode, skipping stt", log.Debugw("text input mode, skipping stt",
"request_id", requestID,
"text_len", len(input.Text), "text_len", len(input.Text),
"text_preview", util.Truncate(input.Text, 50)) "text_preview", util.Truncate(input.Text, 50))
@@ -53,7 +52,7 @@ func NewSTTLambda(sttService stt.Service) *compose.Lambda {
Text: input.Text, Text: input.Text,
IsFinal: true, IsFinal: true,
}); err != nil { }); err != nil {
log.Errorw("发送 stt_result 失败", "error", err) log.Errorw("send stt_result failed", "error", err)
} }
} }
@@ -76,8 +75,7 @@ func NewSTTLambda(sttService stt.Service) *compose.Lambda {
return STTOutput{}, fmt.Errorf("stt: no audio data provided") return STTOutput{}, fmt.Errorf("stt: no audio data provided")
} }
log.Infow("开始语音识别", log.Infow("stt recognition started", "audio_bytes", len(input.AudioData))
"request_id", requestID, "audio_bytes", len(input.AudioData))
// 调用 STT 服务 // 调用 STT 服务
text, err := sttService.Recognize(ctx, input.AudioData, stt.Options{ text, err := sttService.Recognize(ctx, input.AudioData, stt.Options{
@@ -86,7 +84,7 @@ func NewSTTLambda(sttService stt.Service) *compose.Lambda {
Language: input.Language, Language: input.Language,
}) })
if err != nil { if err != nil {
log.Errorw("语音识别失败", "error", err, "request_id", requestID) log.Errorw("stt recognition failed", "error", err)
if sender != nil { if sender != nil {
sender.SendError(models.WsError{ sender.SendError(models.WsError{
Type: "error", Type: "error",
@@ -100,12 +98,11 @@ func NewSTTLambda(sttService stt.Service) *compose.Lambda {
// STT 返回空文本 // STT 返回空文本
if strings.TrimSpace(text) == "" { if strings.TrimSpace(text) == "" {
log.Infow("语音识别结果为空", "request_id", requestID) log.Infow("stt returned empty text")
text = "（未识别到语音）" text = "（未识别到语音）"
} }
log.Debugw("stt recognition completed", log.Debugw("stt recognition completed",
"request_id", requestID,
"text_len", len(text), "text_len", len(text),
"text_preview", util.Truncate(text, 50)) "text_preview", util.Truncate(text, 50))
@@ -117,7 +114,7 @@ func NewSTTLambda(sttService stt.Service) *compose.Lambda {
Text: text, Text: text,
IsFinal: true, IsFinal: true,
}); err != nil { }); err != nil {
log.Errorw("发送 stt_result 失败", "error", err) log.Errorw("send stt_result failed", "error", err)
} }
} }

14
backend/internal/eino/nodes_tts.go
View File

@@ -9,8 +9,8 @@ import (
"github.com/cloudwego/eino/schema" "github.com/cloudwego/eino/schema"
"github.com/hhs/camtalk/internal/ai/tts" "github.com/hhs/camtalk/internal/ai/tts"
"github.com/hhs/camtalk/internal/logger"
"github.com/hhs/camtalk/internal/models" "github.com/hhs/camtalk/internal/models"
"github.com/hhs/camtalk/internal/trace"
) )
// NewTTSLambda 创建 TTS Transform Lambda 节点。 // NewTTSLambda 创建 TTS Transform Lambda 节点。
@@ -26,7 +26,7 @@ func NewTTSLambda(ttsService tts.Service, ttsVoice string, ttsSpeed float64, tts
defer sw.Close() defer sw.Close()
defer input.Close() defer input.Close()
log := logger.Log log := trace.FromContext(ctx)
sender := senderFromCtx(ctx) sender := senderFromCtx(ctx)
requestID := requestIDFromCtx(ctx) requestID := requestIDFromCtx(ctx)
@@ -48,7 +48,7 @@ func NewTTSLambda(ttsService tts.Service, ttsVoice string, ttsSpeed float64, tts
if err == io.EOF { if err == io.EOF {
break break
} }
log.Errorw("TTS: stream recv error", "error", err, "request_id", requestID) log.Errorw("TTS: stream recv error", "error", err)
break break
} }
if sentence != "" { if sentence != "" {
@@ -61,7 +61,7 @@ func NewTTSLambda(ttsService tts.Service, ttsVoice string, ttsSpeed float64, tts
return return
} }
log.Infow("开始 TTS 合成", "request_id", requestID, "sentence_count", len(sentences)) log.Infow("开始 TTS 合成", "sentence_count", len(sentences))
// 将句子数组转为 channel // 将句子数组转为 channel
sentenceCh := make(chan string, len(sentences)) sentenceCh := make(chan string, len(sentences))
@@ -78,7 +78,7 @@ func NewTTSLambda(ttsService tts.Service, ttsVoice string, ttsSpeed float64, tts
SampleRate: ttsSampleRate, SampleRate: ttsSampleRate,
}) })
if err != nil { if err != nil {
log.Errorw("TTS 合成启动失败（已跳过）", "error", err, "request_id", requestID) log.Errorw("TTS 合成启动失败（已跳过）", "error", err)
sw.Send(struct{}{}, nil) sw.Send(struct{}{}, nil)
return return
} }
@@ -87,7 +87,7 @@ func NewTTSLambda(ttsService tts.Service, ttsVoice string, ttsSpeed float64, tts
for chunk := range ttsStream { for chunk := range ttsStream {
select { select {
case <-ctx.Done(): case <-ctx.Done():
log.Infow("TTS 流被中断", "request_id", requestID) log.Infow("TTS 流被中断")
sw.Send(struct{}{}, ctx.Err()) sw.Send(struct{}{}, ctx.Err())
return return
default: default:
@@ -107,7 +107,7 @@ func NewTTSLambda(ttsService tts.Service, ttsVoice string, ttsSpeed float64, tts
} }
} }
log.Infow("TTS 合成完成", "request_id", requestID) log.Infow("TTS 合成完成")
sw.Send(struct{}{}, nil) sw.Send(struct{}{}, nil)
}() }()