Version: 0.4.5.dev.260307

feat: 📡 更新 SSE 消息流格式 * 将 SSE 消息流格式更新为 Apifox 可识别的 OpenAI 格式 * 便于后续与前端的对接与协作
2026-03-07 16:11:11 +08:00
parent 26c350f378
commit 3f95d23376
4 changed files with 257 additions and 86 deletions
--- a/backend/agent/graph.go
+++ b/backend/agent/graph.go
@@ -5,72 +5,99 @@ import (
 	"encoding/json"
 	"io"
 	"strings"
+	"time"

 	"github.com/cloudwego/eino-ext/components/model/ark"
 	"github.com/cloudwego/eino/schema"
+	"github.com/google/uuid"
 	arkModel "github.com/volcengine/volcengine-go-sdk/service/arkruntime/model"
 )

-// StreamResponse 专为 Apifox/前端 识别设计的极简结构
+// StreamResponse 为 OpenAI/DeepSeek 兼容的流式 chunk 结构
 type StreamResponse struct {
-	Choices []struct {
-		Delta struct {
-			Content string `json:"content"`
-		} `json:"delta"`
-	} `json:"choices"`
+	ID      string         `json:"id"`
+	Object  string         `json:"object"`
+	Created int64          `json:"created"`
+	Model   string         `json:"model"`
+	Choices []StreamChoice `json:"choices"`
 }

-// ToStreamResponseDTO 将 Eino 的内部 Chunk 转换为 StreamResponse DTO
-func ToStreamResponseDTO(chunk *schema.Message) StreamResponse {
-	var dto StreamResponse
-	dto.Choices = append(dto.Choices, struct {
-		Delta struct {
-			Content string `json:"content"`
-		} `json:"delta"`
-	}{})
-	dto.Choices[0].Delta.Content = chunk.Content
-	return dto
+type StreamChoice struct {
+	Index        int         `json:"index"`
+	Delta        StreamDelta `json:"delta"`
+	FinishReason *string     `json:"finish_reason"`
 }

-func ToStreamReasoningResponseDTO(chunk *schema.Message) StreamResponse {
-	var dto StreamResponse
-	dto.Choices = append(dto.Choices, struct {
-		Delta struct {
-			Content string `json:"content"`
-		} `json:"delta"`
-	}{})
-	dto.Choices[0].Delta.Content = chunk.ReasoningContent
-	return dto
+type StreamDelta struct {
+	Role             string `json:"role,omitempty"`
+	Content          string `json:"content,omitempty"`
+	ReasoningContent string `json:"reasoning_content,omitempty"`
 }

-// ToOpenAIStream 负责将 Eino 的内部 Chunk 转换为 OpenAI 兼容的 data: {JSON} 字符串
-func ToOpenAIStream(chunk *schema.Message) (string, error) {
-	var dto StreamResponse
-	if chunk.ReasoningContent != "" {
-		dto = ToStreamReasoningResponseDTO(chunk)
-	} else {
-		dto = ToStreamResponseDTO(chunk)
+// ToOpenAIStream 将单个 Eino chunk 转为 OpenAI 兼容 JSON
+func ToOpenAIStream(chunk *schema.Message, requestID, modelName string, created int64, includeRole bool) (string, error) {
+	delta := StreamDelta{}
+	if includeRole {
+		delta.Role = "assistant"
+	}
+	if chunk != nil {
+		delta.Content = chunk.Content
+		delta.ReasoningContent = chunk.ReasoningContent
+	}
+
+	if delta.Role == "" && delta.Content == "" && delta.ReasoningContent == "" {
+		return "", nil
+	}
+
+	dto := StreamResponse{
+		ID:      requestID,
+		Object:  "chat.completion.chunk",
+		Created: created,
+		Model:   modelName,
+		Choices: []StreamChoice{{
+			Index:        0,
+			Delta:        delta,
+			FinishReason: nil,
+		}},
 	}
 	jsonBytes, err := json.Marshal(dto)
 	if err != nil {
 		return "", err
 	}
-	// 严格遵循 SSE 协议格式
 	return string(jsonBytes), nil
 }

-func StreamChat(ctx context.Context, llm *ark.ChatModel, userInput string, ifThinking bool, chatHistory []*schema.Message, outChan chan<- string) (string, error) {
-	// 1. 组装消息
+// ToOpenAIFinishStream 生成结束 chunk（finish_reason=stop）
+func ToOpenAIFinishStream(requestID, modelName string, created int64) (string, error) {
+	stop := "stop"
+	dto := StreamResponse{
+		ID:      requestID,
+		Object:  "chat.completion.chunk",
+		Created: created,
+		Model:   modelName,
+		Choices: []StreamChoice{{
+			Index:        0,
+			Delta:        StreamDelta{},
+			FinishReason: &stop,
+		}},
+	}
+	jsonBytes, err := json.Marshal(dto)
+	if err != nil {
+		return "", err
+	}
+	return string(jsonBytes), nil
+}
+
+func StreamChat(ctx context.Context, llm *ark.ChatModel, modelName string, userInput string, ifThinking bool, chatHistory []*schema.Message, outChan chan<- string) (string, error) {
+	// 1) 组装提示消息
 	messages := make([]*schema.Message, 0)
-	// A. 塞入 System Message (人设)
 	messages = append(messages, schema.SystemMessage(SystemPrompt))
-	// B. 塞入历史记录 (上下文)
 	if len(chatHistory) > 0 {
 		messages = append(messages, chatHistory...)
 	}
-	// C. 塞入用户当前的消息 (当前需求)
 	messages = append(messages, schema.UserMessage(userInput))
-	// 2. 调用流式接口
+
+	// 2) 发起流式请求
 	var thinking *ark.Thinking
 	if ifThinking {
 		thinking = &arkModel.Thinking{Type: arkModel.ThinkingTypeEnabled}
@@ -81,28 +108,45 @@ func StreamChat(ctx context.Context, llm *ark.ChatModel, userInput string, ifThi
 	if err != nil {
 		return "", err
 	}
-	defer reader.Close() // 记得关闭 Reader
+	defer reader.Close()

-	// 3. 循环读取直到结束
+	if strings.TrimSpace(modelName) == "" {
+		modelName = "smartflow-worker"
+	}
+	requestID := "chatcmpl-" + uuid.NewString()
+	created := time.Now().Unix()
+	firstChunk := true
+
+	// 3) 持续转发 chunk
 	var fullText strings.Builder
 	for {
 		chunk, err := reader.Recv()
 		if err == io.EOF {
-			break // 读取完成
+			break
 		}
 		if err != nil {
 			return "", err
 		}
-		/*if chunk.Content == "" { // 过滤掉空内容，避免发送无效消息
-			continue
-		}*/
+
 		fullText.WriteString(chunk.Content)
-		// 将内容发送到通道中供前端消费
-		retChuck, err := ToOpenAIStream(chunk)
+
+		payload, err := ToOpenAIStream(chunk, requestID, modelName, created, firstChunk)
 		if err != nil {
 			return "", err
 		}
-		outChan <- retChuck
+		if payload != "" {
+			outChan <- payload
+			firstChunk = false
+		}
 	}
+
+	// 4) 发送结束 chunk 和 [DONE]
+	finishChunk, err := ToOpenAIFinishStream(requestID, modelName, created)
+	if err != nil {
+		return "", err
+	}
+	outChan <- finishChunk
+	outChan <- "[DONE]"
+
 	return fullText.String(), nil
 }
--- a/backend/api/agent.go
+++ b/backend/api/agent.go
@@ -1,6 +1,7 @@
 package api

 import (
+	"encoding/json"
 	"io"
 	"net/http"
 	"strings"
@@ -16,50 +17,65 @@ type AgentHandler struct {
 	svc *service.AgentService
 }

-// NewAgentHandler 组装 Handler 的“工厂”
+// NewAgentHandler 组装 AgentHandler
 func NewAgentHandler(svc *service.AgentService) *AgentHandler {
 	return &AgentHandler{
-		svc: svc, // 把传进来的 Service 揣进口袋里
+		svc: svc,
 	}
 }

+func writeSSEData(w io.Writer, payload string) error {
+	_, err := io.WriteString(w, "data: "+payload+"\n\n")
+	return err
+}
+
 func (api *AgentHandler) ChatAgent(c *gin.Context) {
-	// 1. 设置请求头
+	// 1) 设置 SSE 响应头
 	c.Writer.Header().Set("Content-Type", "text/event-stream")
 	c.Writer.Header().Set("Cache-Control", "no-cache")
 	c.Writer.Header().Set("Connection", "keep-alive")
 	c.Writer.Header().Set("Transfer-Encoding", "chunked")
-	// 2. 从请求中获取用户输入
+	c.Writer.Header().Set("X-Accel-Buffering", "no")
+
+	// 2) 解析请求体
 	var req model.UserSendMessageRequest
 	if err := c.ShouldBindJSON(&req); err != nil {
 		c.JSON(http.StatusBadRequest, respond.WrongParamType)
 		return
 	}

-	// 兼容：如果前端没传会话 ID，后端兜底创建一个
+	// 3) 规范化会话 ID
 	conversationID := strings.TrimSpace(req.ConversationID)
 	if conversationID == "" {
 		conversationID = uuid.NewString()
 	}
-	// 把最终生效的会话 ID 回传给前端，方便后续继续同一会话
 	c.Writer.Header().Set("X-Conversation-ID", conversationID)

-	userID := c.GetInt("user_id") // 从上下文中获取用户 ID
-	// 3. 调用 Service 层的聊天方法，获取输出通道和错误通道
-	outChan, errChan := api.svc.AgentChat(c.Request.Context(), req.Message, req.Thinking, userID, conversationID)
-	// 4. 循环转发消息/错误
+	userID := c.GetInt("user_id")
+	outChan, errChan := api.svc.AgentChat(c.Request.Context(), req.Message, req.Thinking, req.Model, userID, conversationID)
+
+	// 4) 转发 SSE 流
 	c.Stream(func(w io.Writer) bool {
 		select {
 		case err, ok := <-errChan:
 			if ok && err != nil {
-				respond.DealWithError(c, err)
+				errPayload, _ := json.Marshal(map[string]any{
+					"error": map[string]any{
+						"message": err.Error(),
+						"type":    "server_error",
+					},
+				})
+				_ = writeSSEData(w, string(errPayload))
+				_ = writeSSEData(w, "[DONE]")
 			}
 			return false
 		case msg, ok := <-outChan:
 			if !ok {
 				return false
 			}
-			c.SSEvent("message", msg) // 发送 SSE 格式消息
+			if err := writeSSEData(w, msg); err != nil {
+				return false
+			}
 			return true
 		case <-c.Request.Context().Done():
 			return false
--- a/backend/service/agent.go
+++ b/backend/service/agent.go
@@ -9,6 +9,7 @@ import (
 	"github.com/LoveLosita/smartflow/backend/conv"
 	"github.com/LoveLosita/smartflow/backend/dao"
 	"github.com/LoveLosita/smartflow/backend/inits"
+	"github.com/cloudwego/eino-ext/components/model/ark"
 	"github.com/cloudwego/eino/schema"
 	"github.com/google/uuid"
 )
@@ -35,15 +36,24 @@ func normalizeConversationID(chatID string) string {
 	return trimmed
 }

-func (s *AgentService) AgentChat(ctx context.Context, userMessage string, ifThinking bool, userID int, chatID string) (<-chan string, <-chan error) {
-	//1. 创建一个输出通道
+func (s *AgentService) pickChatModel(requestModel string) (*ark.ChatModel, string) {
+	model := strings.TrimSpace(requestModel)
+	if strings.EqualFold(model, "strategist") {
+		return s.AIHub.Strategist, "strategist"
+	}
+	return s.AIHub.Worker, "worker"
+}
+
+func (s *AgentService) AgentChat(ctx context.Context, userMessage string, ifThinking bool, modelName string, userID int, chatID string) (<-chan string, <-chan error) {
+	// 1) 准备输出通道
 	outChan := make(chan string, 5)
 	errChan := make(chan error, 1)
-	//补充：会话 ID 兜底，避免上层漏传
-	chatID = normalizeConversationID(chatID)

-	//2. 先确保这个会话存在（如果不存在就创建一个新的）
-	//先看看缓存里面有没有这个会话
+	// 2) 规范化会话并选择模型
+	chatID = normalizeConversationID(chatID)
+	selectedModel, resolvedModelName := s.pickChatModel(modelName)
+
+	// 3) 确保会话存在
 	result, err := s.agentCache.GetConversationStatus(ctx, chatID)
 	if err != nil {
 		errChan <- err
@@ -51,7 +61,6 @@ func (s *AgentService) AgentChat(ctx context.Context, userMessage string, ifThin
 		close(errChan)
 		return outChan, errChan
 	}
-	//如果缓存里面没有，就去查库
 	if !result {
 		innerResult, err := s.repo.IfChatExists(ctx, userID, chatID)
 		if err != nil {
@@ -61,7 +70,6 @@ func (s *AgentService) AgentChat(ctx context.Context, userMessage string, ifThin
 			return outChan, errChan
 		}
 		if !innerResult {
-			//如果会话不存在，先创建一个新的会话
 			if _, err = s.repo.CreateNewChat(userID, chatID); err != nil {
 				errChan <- err
 				close(outChan)
@@ -69,15 +77,12 @@ func (s *AgentService) AgentChat(ctx context.Context, userMessage string, ifThin
 				return outChan, errChan
 			}
 		}
-		//补充：把“会话存在”状态回写缓存，后续请求可直接命中
 		if err = s.agentCache.SetConversationStatus(ctx, chatID); err != nil {
-			//缓存回写失败不影响主流程
 			log.Printf("failed to set conversation status cache for %s: %v", chatID, err)
 		}
 	}
-	//能走到这里，要么缓存里有这个会话，要么数据库里有这个会话了
-	//4. 提取出历史消息，构建上下文
-	//先尝试从缓存里拿历史消息
+
+	// 4) 构建历史上下文
 	chatHistory, err := s.agentCache.GetHistory(ctx, chatID)
 	if err != nil {
 		errChan <- err
@@ -85,9 +90,7 @@ func (s *AgentService) AgentChat(ctx context.Context, userMessage string, ifThin
 		close(errChan)
 		return outChan, errChan
 	}
-	//如果缓存里没有历史消息，就从数据库里拿
 	if chatHistory == nil {
-		//先从数据库拿到历史消息
 		histories, err := s.repo.GetUserChatHistories(ctx, userID, 20, chatID)
 		if err != nil {
 			errChan <- err
@@ -95,9 +98,7 @@ func (s *AgentService) AgentChat(ctx context.Context, userMessage string, ifThin
 			close(errChan)
 			return outChan, errChan
 		}
-		//再转换成 Eino 的消息格式
 		chatHistory = conv.ToEinoMessages(histories)
-		//把历史消息放到缓存里，方便下次直接拿
 		if err = s.agentCache.BackfillHistory(ctx, chatID, chatHistory); err != nil {
 			errChan <- err
 			close(outChan)
@@ -105,9 +106,9 @@ func (s *AgentService) AgentChat(ctx context.Context, userMessage string, ifThin
 			return outChan, errChan
 		}
 	}
-	//3. 将用户消息异步落缓存和库
+
+	// 5) 异步落用户消息
 	go func() {
-		//这里先不管落库成功与否了，毕竟不想因为落库失败而影响用户的聊天体验
 		bg := context.Background()
 		_ = s.agentCache.PushMessage(bg, chatID, &schema.Message{
 			Role:    schema.User,
@@ -116,17 +117,17 @@ func (s *AgentService) AgentChat(ctx context.Context, userMessage string, ifThin
 		_ = s.repo.SaveChatHistory(bg, userID, chatID, "user", userMessage)
 	}()

-	//5. 启动一个 goroutine 来处理聊天逻辑
+	// 6) 流式输出模型回复
 	go func() {
-		defer close(outChan) // 确保在函数结束时关闭通道
-		defer close(errChan)
-		//3. 调用 StreamChat 函数进行流式聊天
-		fullText, err := agent.StreamChat(ctx, s.AIHub.Worker, userMessage, ifThinking, chatHistory, outChan)
+		defer close(outChan)
+
+		fullText, err := agent.StreamChat(ctx, selectedModel, resolvedModelName, userMessage, ifThinking, chatHistory, outChan)
 		if err != nil {
 			errChan <- err
 			return
 		}
-		//4. 将 AI 的回复异步落缓存和库
+
+		// 7) 异步落助手消息
 		go func() {
 			bg := context.Background()
 			_ = s.agentCache.PushMessage(bg, chatID, &schema.Message{
@@ -135,9 +136,9 @@ func (s *AgentService) AgentChat(ctx context.Context, userMessage string, ifThin
 			})
 			if saveErr := s.repo.SaveChatHistory(bg, userID, chatID, "assistant", fullText); saveErr != nil {
 				log.Printf("failed to save chat history to database: %v", saveErr)
-				return
 			}
 		}()
 	}()
+
 	return outChan, errChan
 }