feat: 实现 Eino Graph 构建与 Orchestrator 适配器，切换 main.go

- graph.go: 构建 Graph 拓扑 START→STT→History→ChatModel→Splitter→TTS→Done→END - 创建 eino-ext ChatModel 对接 DashScope OpenAI 兼容接口 - 统一使用值类型（PipelineInput/PipelineOutput） - Callback 在运行时通过 Stream option 传入 - adapter.go: EinoOrchestrator 实现 orchestrator.Orchestrator 接口 - 解码 base64 音频/图片，注入 context 值 - 调用 Graph.Stream() 触发惰性执行并消费输出 - 追加用户/助手消息到历史 - main.go: 移除旧 llmService + orchestrator.New() 替换为 eino.NewPipelineGraph() + eino.NewEinoOrchestrator() - 各节点统一使用值类型，State 传递请求元数据 Co-Authored-By: Claude <noreply@anthropic.com>
2026-06-19 21:58:17 +08:00
parent fd5c7712f8
commit 4b731b5ac0
9 changed files with 395 additions and 98 deletions
--- a/backend/internal/eino/adapter.go
+++ b/backend/internal/eino/adapter.go
@@ -0,0 +1,165 @@
+package eino
+
+import (
+	"context"
+	"encoding/base64"
+	"io"
+	"time"
+
+	"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"
+)
+
+// 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 {
+	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,
+	history []models.Message,
+	sender orchestrator.Sender,
+) error {
+	log := logger.Log
+	startTime := time.Now()
+
+	// 1. 设置活跃请求
+	if err := e.sessionMgr.SetActiveRequest(ctx, sessionID, req.RequestID); err != nil {
+		log.Errorw("设置活跃请求失败", "error", err)
+	}
+	defer e.sessionMgr.ClearActiveRequest(ctx, sessionID)
+
+	// 2. 获取会话配置
+	sess, err := e.sessionMgr.Get(ctx, sessionID)
+	if err != nil {
+		log.Errorw("获取会话失败", "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("音频解码失败", "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("图片解码失败", "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 = WithSessionID(ctx, sessionID)
+	ctx = WithStartTime(ctx, startTime)
+	ctx = WithPipelineState(ctx, genLocalState(ctx))
+
+	// 6. 追加用户消息到历史
+	if req.Text != "" {
+		_ = e.sessionMgr.AppendMessage(ctx, sessionID, models.Message{
+			Role:    "user",
+			Content: req.Text,
+		})
+	}
+
+	// 7. 调用 Graph（Stream 模式 + 运行时 Callback）
+	streamReader, err := e.graph.Runnable.Stream(ctx, input, e.callbacks)
+	if err != nil {
+		log.Errorw("Graph Stream 启动失败", "error", err)
+		sender.SendError(models.WsError{
+			Type:      "error",
+			RequestID: req.RequestID,
+			Code:      "INTERNAL_ERROR",
+			Message:   "编排器启动失败",
+		})
+		return err
+	}
+
+	// 8. 消费 StreamReader（触发整条链路执行，side effects 推送消息到客户端）
+	var output PipelineOutput
+	for {
+		o, err := streamReader.Recv()
+		if err != nil {
+			if err == io.EOF {
+				break
+			}
+			log.Errorw("Graph Stream 消费错误", "error", err)
+			break
+		}
+		output = o
+	}
+
+	// 9. 追加助手消息到历史
+	if output.FullResponse != "" {
+		_ = e.sessionMgr.AppendMessage(ctx, sessionID, models.Message{
+			Role:    "assistant",
+			Content: output.FullResponse,
+		})
+	}
+
+	latency := time.Since(startTime).Milliseconds()
+	log.Infow("Eino 编排完成",
+		"request_id", req.RequestID,
+		"latency_ms", latency,
+		"session_id", sessionID)
+
+	return nil
+}
--- a/backend/internal/eino/graph.go
+++ b/backend/internal/eino/graph.go
@@ -0,0 +1,113 @@
+package eino
+
+import (
+	"context"
+	"time"
+
+	openaiImpl "github.com/cloudwego/eino-ext/components/model/openai"
+	"github.com/cloudwego/eino/compose"
+
+	"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/models"
+	"github.com/hhs/camtalk/internal/session"
+)
+
+const (
+	nodeSTT      = "stt"
+	nodeHistory  = "history"
+	nodeLLM      = "llm"
+	nodeSplitter = "splitter"
+	nodeTTS      = "tts"
+	nodeDone     = "done"
+)
+
+// PipelineGraph 封装编译后的 Eino Graph。
+type PipelineGraph struct {
+	Runnable compose.Runnable[PipelineInput, PipelineOutput]
+}
+
+// NewPipelineGraph 构建 CamTalk AI 编排 Graph。
+//
+// 拓扑：START → STT → History → ChatModel → Splitter → TTS → Done → END
+//
+// Graph 使用 Stream 模式调用，ChatModel 实现真正的 token 级流式输出。
+// LLM token 通过 Callback 的 OnEndWithStreamOutput 实时推送到客户端。
+func NewPipelineGraph(
+	ctx context.Context,
+	cfg *config.Config,
+	sttService stt.Service,
+	ttsService tts.Service,
+	sessionMgr session.Manager,
+) (*PipelineGraph, error) {
+	log := logger.Log
+
+	// 1. 创建 eino-ext ChatModel（对接 DashScope OpenAI 兼容接口）
+	chatModel, err := openaiImpl.NewChatModel(ctx, &openaiImpl.ChatModelConfig{
+		APIKey:  cfg.AI.LLM.APIKey,
+		Model:   cfg.AI.LLM.Model,
+		BaseURL: cfg.AI.LLM.Endpoint,
+		Timeout: time.Duration(cfg.AI.LLM.Timeout) * time.Second,
+	})
+	if err != nil {
+		return nil, err
+	}
+	log.Infow("Eino ChatModel 初始化成功",
+		"model", cfg.AI.LLM.Model,
+		"endpoint", cfg.AI.LLM.Endpoint)
+
+	// 2. 构建 Graph（值类型，非指针）
+	g := compose.NewGraph[PipelineInput, PipelineOutput](
+		compose.WithGenLocalState(genLocalState),
+	)
+
+	// 3. 添加节点
+	maxHistory := cfg.Session.MaxHistory
+
+	_ = g.AddLambdaNode(nodeSTT, NewSTTLambda(sttService))
+	_ = g.AddLambdaNode(nodeHistory, NewHistoryLambda(sessionMgr.GetHistory, maxHistory))
+	_ = g.AddChatModelNode(nodeLLM, chatModel)
+	_ = g.AddLambdaNode(nodeSplitter, NewSplitterLambda())
+	_ = g.AddLambdaNode(nodeTTS, NewTTSLambda(
+		ttsService,
+		cfg.AI.TTS.Voice,
+		cfg.AI.TTS.Speed,
+		cfg.AI.TTS.OutputFormat,
+		cfg.AI.TTS.SampleRate,
+	))
+	_ = g.AddLambdaNode(nodeDone, NewDoneLambda(cfg.AI.LLM.Model))
+
+	// 4. 连接边
+	_ = g.AddEdge(compose.START, nodeSTT)
+	_ = g.AddEdge(nodeSTT, nodeHistory)
+	_ = g.AddEdge(nodeHistory, nodeLLM)
+	_ = g.AddEdge(nodeLLM, nodeSplitter)
+	_ = g.AddEdge(nodeSplitter, nodeTTS)
+	_ = g.AddEdge(nodeTTS, nodeDone)
+	_ = g.AddEdge(nodeDone, compose.END)
+
+	// 5. 编译（回调在运行时通过 Stream option 传入）
+	runnable, err := g.Compile(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	log.Infow("Eino Graph 编译成功", "nodes", 6)
+	return &PipelineGraph{Runnable: runnable}, nil
+}
+
+// buildPipelineInput 从 WebSocket 请求和会话配置构建 Graph 输入。
+func buildPipelineInput(req models.WsQuery, sessionID string, sess *models.Session, audioData, imageData []byte) PipelineInput {
+	return PipelineInput{
+		AudioData:  audioData,
+		ImageData:  imageData,
+		Text:       req.Text,
+		SessionID:  sessionID,
+		RequestID:  req.RequestID,
+		Language:   sess.Config.Language,
+		Scenario:   sess.Config.Scenario,
+		TTSEnabled: sess.Config.TTSEnabled,
+	}
+}
--- a/backend/internal/eino/nodes_done.go
+++ b/backend/internal/eino/nodes_done.go
@@ -8,49 +8,47 @@ import (

 	"github.com/hhs/camtalk/internal/logger"
 	"github.com/hhs/camtalk/internal/models"
-	"github.com/hhs/camtalk/internal/session"
 )

+// ctxKeyStartTime 请求开始时间的 context key。
+type ctxKeyStartTime struct{}
+
+// WithStartTime 将请求开始时间注入 context。
+func WithStartTime(ctx context.Context, t time.Time) context.Context {
+	return context.WithValue(ctx, ctxKeyStartTime{}, t)
+}
+
+// latencyFromCtx 从 context 获取开始时间并计算延迟（毫秒）。
+func latencyFromCtx(ctx context.Context) int64 {
+	if startTime, ok := ctx.Value(ctxKeyStartTime{}).(time.Time); ok {
+		return time.Since(startTime).Milliseconds()
+	}
+	return 0
+}
+
 // NewDoneLambda 创建 Done Lambda 节点。
-// 输入: struct{}（TTS 完成信号）→ 输出: PipelineOutput
+// 输入: struct{}（TTS 完成信号）→ 输出: *PipelineOutput
 //
-// 从 PipelineState 读取完整回复和 token 用量，发送 llm_done 到客户端，
-// 追加助手消息到会话历史，返回 PipelineOutput。
-func NewDoneLambda(sessionMgr session.Manager, model string) *compose.Lambda {
-	return compose.InvokableLambda(func(ctx context.Context, _ struct{}) (*PipelineOutput, error) {
+// 从 PipelineState 读取完整回复和 token 用量，发送 llm_done 到客户端。
+// 历史消息追加由适配器负责（避免重复写入）。
+func NewDoneLambda(defaultModel string) *compose.Lambda {
+	return compose.InvokableLambda(func(ctx context.Context, _ struct{}) (PipelineOutput, error) {
 		log := logger.Log
 		sender := senderFromCtx(ctx)
-		requestID := requestIDFromCtx(ctx)
 		state := stateFromCtx(ctx)

 		if state == nil {
-			return &PipelineOutput{}, nil
+			return PipelineOutput{}, nil
 		}

 		state.mu.Lock()
 		fullResponse := state.FullResponse.String()
 		transcribedText := state.TranscribedText
 		tokenUsage := state.TokenUsage
-		modelName := model
+		requestID := state.RequestID
+		modelName := defaultModel
 		state.mu.Unlock()

-		// 追加助手消息到会话历史
-		sessionID := ""
-		if state != nil {
-			// 从 context 获取 sessionID（由适配器注入）
-			if sid, ok := ctx.Value(ctxKeySessionID{}).(string); ok {
-				sessionID = sid
-			}
-		}
-		if sessionID != "" && sessionMgr != nil && fullResponse != "" {
-			if err := sessionMgr.AppendMessage(ctx, sessionID, models.Message{
-				Role:    "assistant",
-				Content: fullResponse,
-			}); err != nil {
-				log.Errorw("追加助手消息到历史失败", "error", err)
-			}
-		}
-
 		// 发送 llm_done
 		if sender != nil && requestID != "" {
 			done := models.WsLLMDone{
@@ -58,7 +56,7 @@ func NewDoneLambda(sessionMgr session.Manager, model string) *compose.Lambda {
 				RequestID: requestID,
 				FullText:  fullResponse,
 				Model:     modelName,
-				LatencyMs: 0, // 由适配器计算
+				LatencyMs: latencyFromCtx(ctx),
 			}
 			if tokenUsage != nil {
 				done.TokensUsed = struct {
@@ -78,9 +76,9 @@ func NewDoneLambda(sessionMgr session.Manager, model string) *compose.Lambda {

 		log.Infow("查询处理完成",
 			"request_id", requestID,
-			"text_length", len(fullResponse))
+			"response_length", len(fullResponse))

-		return &PipelineOutput{
+		return PipelineOutput{
 			TranscribedText: transcribedText,
 			FullResponse:    fullResponse,
 			Model:           modelName,
@@ -88,27 +86,3 @@ func NewDoneLambda(sessionMgr session.Manager, model string) *compose.Lambda {
 		}, nil
 	})
 }
-
-// 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)
-}
-
-// latencyFromCtx 从 context 获取开始时间并计算延迟。
-func latencyFromCtx(ctx context.Context) int64 {
-	if startTime, ok := ctx.Value(ctxKeyStartTime{}).(time.Time); ok {
-		return time.Since(startTime).Milliseconds()
-	}
-	return 0
-}
-
-// ctxKeyStartTime 请求开始时间的 context key。
-type ctxKeyStartTime struct{}
-
-// WithStartTime 将请求开始时间注入 context。
-func WithStartTime(ctx context.Context, t time.Time) context.Context {
-	return context.WithValue(ctx, ctxKeyStartTime{}, t)
-}
--- a/backend/internal/eino/nodes_history.go
+++ b/backend/internal/eino/nodes_history.go
@@ -3,6 +3,7 @@ package eino
 import (
 	"context"
 	"encoding/base64"
+
 	"github.com/cloudwego/eino/compose"
 	"github.com/cloudwego/eino/schema"

@@ -11,28 +12,33 @@ import (
 	"github.com/hhs/camtalk/internal/models"
 )

-// HistoryInput 历史组装节点的输入，包含 STT 输出和原始请求信息。
-type HistoryInput struct {
-	STTOutput  *STTOutput
-	SessionID  string
-	RequestID  string
-	ImageData  []byte
-	Scenario   string
-	DetailLevel string
-}
-
 // NewHistoryLambda 创建历史组装 Lambda 节点。
-// 输入: HistoryInput → 输出: []*schema.Message
+// 输入: *STTOutput → 输出: []*schema.Message
 //
+// 从 PipelineState 读取请求元数据（SessionID、Scenario、ImageData 等），
 // 构建系统提示词，组装历史消息和当前用户输入（含多模态图片）。
-func NewHistoryLambda(historyFetcher func(ctx context.Context, sessionID string, maxHistory int) ([]models.Message, error), maxHistory int) *compose.Lambda {
-	return compose.InvokableLambda(func(ctx context.Context, input *HistoryInput) ([]*schema.Message, error) {
+func NewHistoryLambda(historyFetcher func(ctx context.Context, sessionID string, limit int) ([]models.Message, error), maxHistory int) *compose.Lambda {
+	return compose.InvokableLambda(func(ctx context.Context, sttOut STTOutput) ([]*schema.Message, error) {
 		log := logger.Log
-		requestID := input.RequestID
+
+		// 从 State 读取请求元数据
+		state := stateFromCtx(ctx)
+		if state == nil {
+			return []*schema.Message{}, nil
+		}
+
+		state.mu.Lock()
+		sessionID := state.SessionID
+		requestID := state.RequestID
+		imageData := state.ImageData
+		scenario := state.Scenario
+		detailLevel := state.DetailLevel
+		language := sttOut.Language
+		state.mu.Unlock()

 		// 构建系统提示词
-		scenarioPrompt := llm.GetScenarioPrompt(input.Scenario, input.STTOutput.Language)
-		systemPrompt := llm.BuildSystemPrompt(input.STTOutput.Language, input.DetailLevel, scenarioPrompt)
+		scenarioPrompt := llm.GetScenarioPrompt(scenario, language)
+		systemPrompt := llm.BuildSystemPrompt(language, detailLevel, scenarioPrompt)

 		// 构建 system message（含图片）
 		systemMsg := &schema.Message{
@@ -41,9 +47,9 @@ func NewHistoryLambda(historyFetcher func(ctx context.Context, sessionID string,
 		}

 		// 如果有图片，添加到 system message 的多模态内容中
-		if len(input.ImageData) > 0 {
-			base64Str := base64.StdEncoding.EncodeToString(input.ImageData)
-			mimeType := detectImageMimeType(input.ImageData)
+		if len(imageData) > 0 {
+			base64Str := base64.StdEncoding.EncodeToString(imageData)
+			mimeType := detectImageMimeType(imageData)
 			systemMsg.UserInputMultiContent = []schema.MessageInputPart{
 				{
 					Type: schema.ChatMessagePartTypeImageURL,
@@ -61,8 +67,8 @@ func NewHistoryLambda(historyFetcher func(ctx context.Context, sessionID string,
 		messages := []*schema.Message{systemMsg}

 		// 获取并追加历史消息
-		if historyFetcher != nil && input.SessionID != "" {
-			history, err := historyFetcher(ctx, input.SessionID, maxHistory)
+		if historyFetcher != nil && sessionID != "" {
+			history, err := historyFetcher(ctx, sessionID, maxHistory)
 			if err != nil {
 				log.Warnw("获取历史消息失败，继续处理", "error", err, "request_id", requestID)
 			} else {
@@ -78,14 +84,14 @@ func NewHistoryLambda(historyFetcher func(ctx context.Context, sessionID string,
 		// 追加当前用户输入
 		messages = append(messages, &schema.Message{
 			Role:    schema.User,
-			Content: input.STTOutput.Text,
+			Content: sttOut.Text,
 		})

 		log.Infow("历史组装完成",
 			"request_id", requestID,
 			"message_count", len(messages),
-			"has_image", len(input.ImageData) > 0,
-			"scenario", input.Scenario)
+			"has_image", len(imageData) > 0,
+			"scenario", scenario)

 		return messages, nil
 	})
@@ -96,22 +102,17 @@ func detectImageMimeType(data []byte) string {
 	if len(data) < 4 {
 		return "image/jpeg"
 	}
-	// JPEG: FF D8 FF
 	if data[0] == 0xFF && data[1] == 0xD8 && data[2] == 0xFF {
 		return "image/jpeg"
 	}
-	// PNG: 89 50 4E 47
 	if data[0] == 0x89 && data[1] == 0x50 && data[2] == 0x4E && data[3] == 0x47 {
 		return "image/png"
 	}
-	// GIF: 47 49 46 38
 	if data[0] == 0x47 && data[1] == 0x49 && data[2] == 0x46 {
 		return "image/gif"
 	}
-	// WebP: 52 49 46 46
 	if data[0] == 0x52 && data[1] == 0x49 && data[2] == 0x46 && data[3] == 0x46 {
 		return "image/webp"
 	}
-	return "image/jpeg" // 默认
+	return "image/jpeg"
 }
-
--- a/backend/internal/eino/nodes_stt.go
+++ b/backend/internal/eino/nodes_stt.go
@@ -19,11 +19,24 @@ import (
 // 语音模式：调用 sttService.Recognize() 进行语音识别。
 // 识别结果通过 Sender 发送 stt_result 到客户端。
 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
 		sender := senderFromCtx(ctx)
 		requestID := requestIDFromCtx(ctx)

+		// 将输入元数据写入 State，供下游节点（History、Done）读取
+		if state := stateFromCtx(ctx); state != nil {
+			state.mu.Lock()
+			state.SessionID = input.SessionID
+			state.RequestID = input.RequestID
+			state.ImageData = input.ImageData
+			state.Scenario = input.Scenario
+			state.DetailLevel = "low"
+			state.Language = input.Language
+			state.TTSEnabled = input.TTSEnabled
+			state.mu.Unlock()
+		}
+
 		// 文本输入模式：跳过 STT
 		if input.Text != "" {
 			log.Infow("使用文本输入，跳过 STT",
@@ -48,7 +61,7 @@ func NewSTTLambda(sttService stt.Service) *compose.Lambda {
 				state.mu.Unlock()
 			}

-			return &STTOutput{
+			return STTOutput{
 				Text:      input.Text,
 				Language:  input.Language,
 				IsSkipped: true,
@@ -57,7 +70,7 @@ func NewSTTLambda(sttService stt.Service) *compose.Lambda {

 		// 语音模式：解码音频
 		if len(input.AudioData) == 0 {
-			return nil, fmt.Errorf("stt: no audio data provided")
+			return STTOutput{}, fmt.Errorf("stt: no audio data provided")
 		}

 		log.Infow("开始语音识别",
@@ -79,7 +92,7 @@ func NewSTTLambda(sttService stt.Service) *compose.Lambda {
 					Message:   "语音识别失败: " + err.Error(),
 				})
 			}
-			return nil, fmt.Errorf("stt: recognize: %w", err)
+			return STTOutput{}, fmt.Errorf("stt: recognize: %w", err)
 		}

 		// STT 返回空文本
@@ -109,7 +122,7 @@ func NewSTTLambda(sttService stt.Service) *compose.Lambda {
 			state.mu.Unlock()
 		}

-		return &STTOutput{
+		return STTOutput{
 			Text:      text,
 			Language:  input.Language,
 			IsSkipped: false,
--- a/backend/internal/eino/state.go
+++ b/backend/internal/eino/state.go
@@ -7,13 +7,22 @@ import (
 )

 // PipelineState Graph 全局状态，用于跨节点收集数据。
-// 通过 compose.WithGenLocalState 注册，各节点通过 StatePreHandler/StatePostHandler 读写。
+// 通过 compose.WithGenLocalState 注册，各节点通过 compose.ProcessState 读写。
 type PipelineState struct {
 	mu              sync.Mutex
 	FullResponse    strings.Builder // LLM 完整回复（由 Callback 累积）
 	TranscribedText string          // STT 识别文本
 	Model           string          // 实际使用的模型名
 	TokenUsage      *TokenUsage     // token 用量
+
+	// 从 PipelineInput 复制的元数据，供下游节点（History、Done）读取
+	SessionID   string
+	RequestID   string
+	ImageData   []byte
+	Scenario    string
+	DetailLevel string
+	Language    string
+	TTSEnabled  bool
 }

 // genLocalState 创建每请求的 PipelineState 实例。