CamTalk/backend/internal/eino/adapter.go

package eino

import (
	"context"
	"encoding/base64"
	"io"
	"time"

	"github.com/cloudwego/eino/compose"

	"github.com/hhs/camtalk/internal/models"
	"github.com/hhs/camtalk/internal/orchestrator"
	"github.com/hhs/camtalk/internal/session"
	"github.com/hhs/camtalk/internal/trace"
)

// EinoOrchestrator 实现 orchestrator.Orchestrator 接口。
// 将 Eino Graph 包装为现有接口，WS Handler 几乎不用改。
type EinoOrchestrator struct {
	graph      *PipelineGraph
	sessionMgr session.Manager
	model      string
	callbacks  compose.Option // 运行时 Callback option
}

// NewEinoOrchestrator 创建 Eino 编排器适配器。
func NewEinoOrchestrator(graph *PipelineGraph, sessionMgr session.Manager, model string) *EinoOrchestrator {
	return &EinoOrchestrator{
		graph:      graph,
		sessionMgr: sessionMgr,
		model:      model,
		callbacks:  compose.WithCallbacks(BuildCallbackHandler()),
	}
}

// ProcessQuery 实现 orchestrator.Orchestrator 接口。
func (e *EinoOrchestrator) ProcessQuery(
	ctx context.Context,
	sessionID string,
	req models.WsQuery,
	sender orchestrator.Sender,
) error {
	log := trace.FromContext(ctx)
	startTime := time.Now()

	// 1. 设置活跃请求
	if err := e.sessionMgr.SetActiveRequest(ctx, sessionID, req.RequestID); err != nil {
		return err
	}
	defer e.sessionMgr.ClearActiveRequest(ctx, sessionID)

	// 2. 获取会话配置
	sess, err := e.sessionMgr.Get(ctx, sessionID)
	if err != nil {
		log.Errorw("get session failed", "error", err)
		sender.SendError(models.WsError{
			Type:      "error",
			RequestID: req.RequestID,
			Code:      "SESSION_NOT_FOUND",
			Message:   "会话不存在",
		})
		return err
	}

	// 3. 解码音频和图片
	var audioData []byte
	if req.Text == "" && req.Audio != "" {
		audioData, err = base64.StdEncoding.DecodeString(req.Audio)
		if err != nil {
			log.Errorw("audio decode failed", "error", err)
			sender.SendError(models.WsError{
				Type:      "error",
				RequestID: req.RequestID,
				Code:      "INVALID_MESSAGE",
				Message:   "音频数据解码失败",
			})
			return err
		}
	}

	var imageData []byte
	if req.Image != "" {
		imageData, err = base64.StdEncoding.DecodeString(req.Image)
		if err != nil {
			log.Errorw("image decode failed", "error", err)
			sender.SendError(models.WsError{
				Type:      "error",
				RequestID: req.RequestID,
				Code:      "INVALID_MESSAGE",
				Message:   "图片数据解码失败",
			})
			return err
		}
	}

	// 4. 构建 Graph 输入
	input := buildPipelineInput(req, sessionID, sess, audioData, imageData)

	// 5. 注入 context 值（供 Callback 和 Lambda 节点使用）
	ctx = WithSender(ctx, sender)
	ctx = WithRequestID(ctx, req.RequestID)
	ctx = trace.WithSessionID(ctx, sessionID)
	ctx = WithStartTime(ctx, startTime)

	// 创建 State 并从 input 复制元数据
	state := genLocalState(ctx)
	state.SessionID = input.SessionID
	state.RequestID = input.RequestID
	state.ImageData = input.ImageData
	state.Scenario = input.Scenario
	state.Language = input.Language
	state.DetailLevel = sess.Config.DetailLevel
	state.TTSEnabled = input.TTSEnabled
	state.UserID = input.UserID
	ctx = WithPipelineState(ctx, state)

	// 6. 调用 Graph（Stream 模式 + 运行时 Callback）
	streamReader, err := e.graph.Runnable.Stream(ctx, input, e.callbacks)
	if err != nil {
		log.Errorw("graph stream start failed", "error", err)
		sender.SendError(models.WsError{
			Type:      "error",
			RequestID: req.RequestID,
			Code:      "INTERNAL_ERROR",
			Message:   "编排器启动失败",
		})
		return err
	}

	// 7. 消费 StreamReader（触发整条链路执行，side effects 推送消息到客户端）
	var output PipelineOutput
	for {
		o, err := streamReader.Recv()
		if err != nil {
			if err == io.EOF {
				break
			}
			log.Errorw("graph stream consume error", "error", err)
			break
		}
		output = o
	}

	// 8. 追加用户消息到历史（使用 STT 结果，兼容文本输入和语音输入）
	userText := output.TranscribedText
	if userText == "" {
		userText = req.Text // fallback 到原始文本输入
	}
	if userText != "" {
		if err := e.sessionMgr.AppendMessage(ctx, sessionID, models.Message{
			Role:    "user",
			Content: userText,
		}); err != nil {
			log.Errorw("append user message failed", "error", err)
		}
	}

	// 9. 追加助手消息到历史
	if output.FullResponse != "" {
		if err := e.sessionMgr.AppendMessage(ctx, sessionID, models.Message{
			Role:    "assistant",
			Content: output.FullResponse,
		}); err != nil {
			log.Errorw("append assistant message failed", "error", err)
		}
	}

	latency := time.Since(startTime).Milliseconds()
	log.Infow("eino pipeline completed", "latency_ms", latency)

	return nil
}