feat: AIOps智能助手流式输出(SSE代理,逐字渲染+思考中光标+断连安全持久化)

backend/internal/api/handlers/llm.go

@@ -1,6 +1,7 @@
 package handlers
 
 import (
+	"bufio"
 	"bytes"
 	"encoding/json"
 	"fmt"
@@ -258,6 +259,191 @@ func (h *LLMHandler) Chat(c *gin.Context) {
 	}})
 }
 
+// ChatStream AI 流式对话代理（SSE） // Logan.Li
+// POST /api/v1/aiops/chat/stream
+// 客户端用 EventSource/fetch-stream 消费，每条 SSE data 为 {"content": "..."}
+// 出错时发送 event: error，data 为 {"message": "..."}
+// 完成后服务端持久化 user + assistant 两条历史，data: [DONE] 结束
+func (h *LLMHandler) ChatStream(c *gin.Context) {
+	var cfg models.LLMConfig
+	if err := h.db.First(&cfg).Error; err != nil || !cfg.Enabled {
+		writeSSEError(c, 400, "LLM 服务未配置或未启用，请在「模型配置」中完成设置")
+		return
+	}
+
+	var req models.ChatRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		writeSSEError(c, 400, "请求参数错误: "+err.Error())
+		return
+	}
+	if len(req.Messages) == 0 {
+		writeSSEError(c, 400, "消息不能为空")
+		return
+	}
+
+	// 设置 SSE 响应头 // Logan.Li
+	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("X-Accel-Buffering", "no") // Nginx 关闭缓冲
+	c.Writer.WriteHeader(http.StatusOK)
+	// 注册写入心跳（立即 flush 一个注释行让客户端确认连接建立）
+	c.Writer.Flush()
+
+	// 组装系统提示词 + 上下文
+	systemPrompt := cfg.SystemPrompt
+	if strings.TrimSpace(systemPrompt) == "" {
+		systemPrompt = defaultSystemPrompt
+	}
+	messages := append([]models.ChatMessage{{Role: "system", Content: systemPrompt}}, req.Messages...)
+	// 构造 OpenAI 兼容流式请求
+	payload := struct {
+		Model       string                 `json:"model"`
+		Messages    []models.ChatMessage   `json:"messages"`
+		Temperature float64                `json:"temperature"`
+		MaxTokens   int                    `json:"max_tokens,omitempty"`
+		Stream      bool                   `json:"stream"`
+	}{
+		Model:       cfg.Model,
+		Messages:    messages,
+		Temperature: cfg.Temperature,
+		MaxTokens:   cfg.MaxTokens,
+		Stream:      true,
+	}
+	body, _ := json.Marshal(payload)
+
+	endpoint := resolveEndpoint(cfg.Provider, cfg.Endpoint)
+	url := endpoint + "/chat/completions"
+
+	httpReq, err := http.NewRequestWithContext(c.Request.Context(), http.MethodPost, url, bytes.NewReader(body))
+	if err != nil {
+		writeSSEError(c, 500, "构造请求失败: "+err.Error())
+		return
+	}
+	httpReq.Header.Set("Content-Type", "application/json")
+	httpReq.Header.Set("Authorization", "Bearer "+cfg.APIKey)
+	httpReq.Header.Set("Accept", "text/event-stream")
+
+	// 流式请求不能用全局 120s 超时客户端，构造无超时 client 让 ctx 控制 // Logan.Li
+	streamClient := &http.Client{Timeout: 0}
+	resp, err := streamClient.Do(httpReq)
+	if err != nil {
+		writeSSEError(c, 502, "LLM 服务连接失败，请检查网络或 Endpoint 配置: "+err.Error())
+		return
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		raw, _ := io.ReadAll(resp.Body)
+		msg := fmt.Sprintf("LLM 服务返回错误 (HTTP %d)", resp.StatusCode)
+		if e := parseOpenAIError(raw); e != "" {
+			msg += ": " + e
+		}
+		writeSSEError(c, resp.StatusCode, msg)
+		return
+	}
+
+	// 逐行解析上游 SSE，转发增量内容给前端 // Logan.Li
+	var contentBuilder strings.Builder
+	reader := bufio.NewReader(resp.Body)
+	clientGone := false
+	for {
+		// 客户端断开时停止读取（c.Request.Context().Done() 在客户端断开时触发） // Logan.Li
+		select {
+		case <-c.Request.Context().Done():
+			clientGone = true
+		default:
+		}
+		if clientGone {
+			break
+		}
+
+		line, readErr := reader.ReadString('\n')
+		line = strings.TrimSpace(line)
+		if line != "" {
+			// 仅处理 data: 开头的行
+			if strings.HasPrefix(line, "data:") {
+				data := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
+				if data == "[DONE]" {
+					// 上游结束，跳出循环准备持久化 + 发送 [DONE]
+					break
+				}
+				// 解析 {"choices":[{"delta":{"content":"..."}}]}
+				var chunk struct {
+					Choices []struct {
+						Delta struct {
+							Content string `json:"content"`
+						} `json:"delta"`
+					} `json:"choices"`
+				}
+				if json.Unmarshal([]byte(data), &chunk) == nil && len(chunk.Choices) > 0 {
+					delta := chunk.Choices[0].Delta.Content
+					if delta != "" {
+						contentBuilder.WriteString(delta)
+						// 转发给前端：data: {"content":"..."}
+						writeSSEChunk(c, delta)
+					}
+				}
+			}
+		}
+		if readErr != nil {
+			if readErr != io.EOF {
+				// 读取出错（非正常结束）
+				if contentBuilder.Len() == 0 {
+					writeSSEError(c, 502, "LLM 流式响应读取失败: "+readErr.Error())
+					return
+				}
+				// 已有部分内容则正常结束
+			}
+			break
+		}
+	}
+
+	fullContent := contentBuilder.String()
+
+	// 持久化对话历史（仅在成功且非空时入库；错误/中断不入库避免污染历史） // Logan.Li
+	if fullContent != "" && !clientGone {
+		uid := middleware.GetUserID(c)
+		if uid != 0 {
+			lastUser := req.Messages[len(req.Messages)-1]
+			now := time.Now()
+			h.db.Create(&[]models.ChatHistory{
+				{UserID: uid, Role: "user", Content: lastUser.Content, CreatedAt: now},
+				{UserID: uid, Role: "assistant", Content: fullContent, Model: cfg.Model, CreatedAt: now.Add(time.Millisecond)},
+			})
+		}
+		// 发送结束标志
+		c.Writer.WriteString("event: done\ndata: {\"model\":\"" + cfg.Model + "\"}\n\n")
+		c.Writer.Flush()
+	} else if !clientGone {
+		// 无内容时通知前端结束（避免前端无限等待）
+		c.Writer.WriteString("event: done\ndata: {}\n\n")
+		c.Writer.Flush()
+	}
+}
+
+// writeSSEChunk 写入一条内容增量 // Logan.Li
+func writeSSEChunk(c *gin.Context, content string) {
+	// 转义为 JSON 字符串避免换行符破坏 SSE 协议
+	b, _ := json.Marshal(map[string]string{"content": content})
+	c.Writer.WriteString("data: " + string(b) + "\n\n")
+	c.Writer.Flush()
+}
+
+// writeSSEError 发送一个 error 事件（写入后立即结束流） // Logan.Li
+func writeSSEError(c *gin.Context, code int, message string) {
+	// 若 header 尚未写入（连接前错误），需先设置 SSE 头
+	if !c.Writer.Written() {
+		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.WriteHeader(http.StatusOK)
+	}
+	b, _ := json.Marshal(map[string]any{"code": code, "message": message})
+	c.Writer.WriteString("event: error\ndata: " + string(b) + "\n\n")
+	c.Writer.Flush()
+}
+
 // parseOpenAIError 从上游错误响应中提取 message 字段 // Logan.Li
 func parseOpenAIError(raw []byte) string {
 	var m struct {

backend/internal/api/routes/routes.go

@@ -421,6 +421,7 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config) {
 		aiopsGroup.PUT("/llm-config", llmHandler.SaveLLMConfig)
 		aiopsGroup.POST("/llm-config/test", llmHandler.TestLLMConnection)
 		aiopsGroup.POST("/chat", llmHandler.Chat)
+		aiopsGroup.POST("/chat/stream", llmHandler.ChatStream)
 		aiopsGroup.GET("/chat-history", llmHandler.GetChatHistory)
 		aiopsGroup.DELETE("/chat-history", llmHandler.ClearChatHistory)
 	}

docs/dev-log.md

@@ -1704,3 +1704,53 @@ cd ~/devops/aiops
   - 快捷操作可扩展更多数据源: 容量预测(接入 agent history 时序数据)、K8s 集群健康(k8sApi)、云账单分析(billingApi)
   - 用户消息气泡中【平台实时数据】块较长，可考虑折叠展示（默认收起，点击展开）提升可读性
   - 前端组件拆分候选仍未变: TopologyView.tsx(576)、ExcelServerImport.tsx(549)、DomainManager.tsx(492)
+
+
+## 2026-06-18 05:30 自动推进（cron）— AIOps 智能助手流式输出（SSE）
+
+- **任务**: 为 AIOps 智能助手实现流式输出（SSE），长报告/长分析无需等待完整响应即可逐字显示，显著提升交互体验
+- **背景**: 上次 cron 完成快捷操作接入真实平台数据，dev-log 下一步建议第 1 项明确列出"Chat 流式输出（SSE/streaming）—— 长报告/长分析当前需等完整响应才显示，流式可显著提升体验"。此前 Chat handler 一次性 io.ReadAll 后 JSON 返回，长回复用户需盯 spinner 干等。本次实现完整 SSE 流式代理链路
+- **实现（后端 2 步）**:
+  - backend/internal/api/handlers/llm.go — 新增 ChatStream handler + writeSSEChunk + writeSSEError 辅助函数（~190 行新增）:
+    - 设置 SSE 响应头（Content-Type: text/event-stream + X-Accel-Buffering: no 关闭 Nginx 缓冲）+ 初始 flush
+    - 向 OpenAI 兼容端点发送 stream:true 请求；用无超时 http.Client（Timeout:0）让 ctx 控制生命周期
+    - bufio.Reader 逐行解析上游 SSE，解析 `{"choices":[{"delta":{"content":"..."}}]}` 提取增量内容
+    - 每个 delta 用 `data: {"content":"..."}` 格式转发给前端（JSON 序列化避免换行符破坏 SSE 协议）
+    - 收到 `[DONE]` 或 EOF 时跳出循环；累计完整内容成对写入 chat_histories（user + assistant），发送 `event: done` 通知前端结束
+    - 客户端断开检测：每轮迭代检查 `c.Request.Context().Done()`，断开时不再持久化（避免污染历史）
+    - 错误处理：连接前/连接中错误统一用 writeSSEError 发送 `event: error` + JSON `{code,message}`，避免 HTTP 错误码破坏已开始的 SSE 流
+  - backend/internal/api/routes/routes.go — 注册 POST /aiops/chat/stream（保留原 /chat 用于非流式回退）
+- **实现（前端 2 步）**:
+  - frontend/src/services/api.ts — 新增 aiApi.chatStream(messages, onChunk, signal)（~70 行）:
+    - 用裸 fetch（绕过 http.ts 的 await json() 一次性消费）+ ReadableStream.getReader() + TextDecoder
+    - 按 `\n\n` 分割 SSE 事件块，解析 `event:`/`data:` 字段
+    - message 事件解析 {"content":"..."} 调 onChunk(delta)；error 事件抛出 Error(message)；done 事件跳过
+    - 返回累计完整内容；支持 AbortSignal 中断
+  - frontend/src/components/aiops/AIOpsAssistant.tsx — sendMessage 重构为流式（~30 行改动）:
+    - 发送时立即插入 user 消息 + 空 assistant 占位气泡
+    - 调用 chatStream，每个 chunk 通过 setMessages 逐块追加到对应 assistant 气泡（React map + 不可变更新）
+    - 空内容 + isLoading 时显示"思考中..."spinner；有内容后末尾显示紫色闪烁光标（animate-pulse）提示流式进行中
+    - 错误处理：首个 chunk 到达前出错 → 整条气泡替换为红色错误；已部分输出时出错 → 末尾追加错误说明
+    - 移除原独立 spinner 块（流式占位气泡已包含思考提示）
+- **协议设计（SSE 事件类型）**:
+  - `data: {"content":"..."}` — 内容增量（默认 message 事件）
+  - `event: done` + `data: {"model":"..."}` — 流结束（成功）
+  - `event: error` + `data: {"code":N,"message":"..."}` — 错误（连接失败/上游非200/未配置）
+- **结果**:
+  - 后端 go build 通过；前端 npx tsc --noEmit 零错误；npm run build 通过(3.20s)
+  - ./run.sh restart 后 /health 正常 + 前端 200
+  - API 验证（curl + 模拟 OpenAI SSE 服务器 python3 /tmp/mock_openai_sse.py 监听 :18899）:
+    - 临时启用 config 指向 mock server → POST /aiops/chat/stream 返回 SSE 流：`data: {"content":"你好！"}\n\ndata: {"content":"我是 "}\n\n...` 逐块输出，最终 `event: done\ndata: {"model":"qwen2.5:7b"}`
+    - sqlite3 验证：成对写入 chat_histories（user:"你是谁" + assistant:"你好！我是 EZOps..."），刷新后历史可见
+    - 未配置时 → `event: error\ndata: {"code":400,"message":"LLM 服务未配置..."}`
+    - 原非流式 POST /chat 仍正常工作（向后兼容）
+  - 浏览器验证 /ai-assistant:
+    - 输入"你是谁"+发送 → 用户气泡立即出现 → 助手空气泡显示"思考中..." → 第一个 chunk 到达后切换为内容 + 末尾紫色光标闪烁 → 完整 66 字回复逐字渲染
+    - console 零 JS 错误
+  - 测试后清理：config endpoint 恢复原值(enabled=0) + 清空 chat_histories + 停掉 mock server
+  - git push 成功
+- **下一步建议**:
+  - 快捷操作可扩展更多数据源（容量预测接入 agent history 时序/K8s 集群健康/云账单分析）
+  - 用户消息气泡中【平台实时数据】块较长，可考虑折叠展示（默认收起，点击展开）
+  - 多会话支持（当前单用户单历史，可扩展多命名会话切换）
+  - 流式中支持"停止生成"按钮（AbortController 中断 fetch，后端 ctx.Done() 自动停止）

frontend/src/components/aiops/AIOpsAssistant.tsx

@@ -234,7 +234,7 @@ export function AIOpsAssistant() {
     }
   };
 
-  // 发送一条消息的核心逻辑（供输入框发送和快捷操作复用） // Logan.Li
+  // 发送一条消息的核心逻辑（流式输出，逐步填充助手气泡） // Logan.Li
   const sendMessage = async (text: string) => {
     const userMsg: Message = {
       id: Date.now().toString(),
@@ -242,23 +242,44 @@ export function AIOpsAssistant() {
       content: text,
       timestamp: new Date().toISOString(),
     };
-    const newMessages = [...messages, userMsg];
+    const assistantId = (Date.now() + 1).toString();
+    const newMessages: Message[] = [
+      ...messages,
+      userMsg,
+      // 预先插入空助手消息，随流式内容逐步填充
+      { id: assistantId, role: 'assistant', content: '', timestamp: new Date().toISOString() },
+    ];
     setMessages(newMessages);
     setIsLoading(true);
+    // 用于在收到第一个 chunk 前显示"思考中"占位
+    let firstChunkArrived = false;
     try {
-      // 仅保留最近 10 条上下文避免 token 超限 // Logan.Li
+      // 仅保留最近 10 条上下文避免 token 超限（不含刚插入的空助手消息） // Logan.Li
-      const history = newMessages.slice(-10).map((m) => ({ role: m.role, content: m.content }));
+      const history = [...messages, userMsg]
-      const result = await aiApi.chat(history);
+        .slice(-10)
-      setMessages((prev) => [
+        .map((m) => ({ role: m.role, content: m.content }));
-        ...prev,
+      await aiApi.chatStream(history, (delta) => {
-        { id: (Date.now() + 1).toString(), role: 'assistant', content: result.content, timestamp: new Date().toISOString() },
+        firstChunkArrived = true;
-      ]);
+        // 逐块追加到助手消息内容
+        setMessages((prev) =>
+          prev.map((m) =>
+            m.id === assistantId ? { ...m, content: m.content + delta } : m,
+          ),
+        );
+      });
     } catch (e) {
       const errMsg = e instanceof Error ? e.message : '对话失败';
-      setMessages((prev) => [
+      setMessages((prev) =>
-        ...prev,
+        prev.map((m) =>
-        { id: (Date.now() + 1).toString(), role: 'assistant', content: `⚠️ ${errMsg}`, timestamp: new Date().toISOString(), isError: true },
+          m.id === assistantId
-      ]);
+            ? {
+                ...m,
+                content: firstChunkArrived ? m.content + `\n\n⚠️ ${errMsg}` : `⚠️ ${errMsg}`,
+                isError: !firstChunkArrived,
+              }
+            : m,
+        ),
+      );
     } finally {
       setIsLoading(false);
     }
@@ -431,7 +452,10 @@ export function AIOpsAssistant() {
                   )}
                 </div>
               )}
-              {messages.map((message) => (
+              {messages.map((message) => {
+                // 流式中：助手消息内容为空且正在加载时显示"思考中"占位 // Logan.Li
+                const isStreaming = isLoading && message.role === 'assistant' && message.content === '';
+                return (
                 <div key={message.id} className={`flex gap-3 ${message.role === 'user' ? 'flex-row-reverse' : ''}`}>
                   <Avatar className={message.role === 'assistant' ? 'bg-purple-500' : 'bg-blue-500'}>
                     <AvatarFallback>
@@ -448,27 +472,25 @@ export function AIOpsAssistant() {
                             : 'bg-blue-600 text-white'
                       }`}
                     >
-                      {message.content}
+                      {isStreaming ? (
+                        <div className="flex items-center gap-2 text-slate-400">
+                          <Loader2 className="h-4 w-4 animate-spin" />
+                          <span className="text-sm">思考中...</span>
+                        </div>
+                      ) : (
+                        message.content
+                      )}
+                      {/* 流式接收中且已有内容时，显示末尾闪烁光标提示 */}
+                      {isLoading && message.role === 'assistant' && !isStreaming && message.content && (
+                        <span className="inline-block w-2 h-4 ml-0.5 bg-purple-400 animate-pulse align-middle" />
+                      )}
                     </div>
                     <div className="text-xs text-slate-500 mt-1">{fmtTime(message.timestamp)}</div>
                   </div>
                 </div>
-              ))}
+                );
-              {isLoading && (
+              })}
-                <div className="flex gap-3">
+              {/* 流式输出时占位气泡已包含"思考中"提示，无需独立 spinner */}
-                  <Avatar className="bg-purple-500">
-                    <AvatarFallback>
-                      <Bot className="h-4 w-4" />
-                    </AvatarFallback>
-                  </Avatar>
-                  <div className="bg-slate-700 rounded-lg p-3">
-                    <div className="flex items-center gap-2 text-slate-400">
-                      <Loader2 className="h-4 w-4 animate-spin" />
-                      <span className="text-sm">思考中...</span>
-                    </div>
-                  </div>
-                </div>
-              )}
               <div ref={messagesEndRef} />
             </div>
           </ScrollArea>

frontend/src/services/api.ts

@@ -673,6 +673,78 @@ export const aiApi = {
   // AI 对话（messages 含历史 + 当前用户消息）
   chat: (messages: { role: 'user' | 'assistant'; content: string }[]) =>
     http.post<ChatResult>('/aiops/chat', { messages }),
+  // AI 流式对话（SSE）。逐块回调 onChunk，返回完整内容。错误抛出 Error。
+  // 使用裸 fetch 绕过 http.ts（后者会 await json() 一次性消费） // Logan.Li
+  chatStream: async (
+    messages: { role: 'user' | 'assistant'; content: string }[],
+    onChunk: (delta: string) => void,
+    signal?: AbortSignal,
+  ): Promise<string> => {
+    const token = useUserStore.getState().token;
+    const res = await fetch('/api/v1/aiops/chat/stream', {
+      method: 'POST',
+      headers: {
+        'Content-Type': 'application/json',
+        ...(token ? { Authorization: `Bearer ${token}` } : {}),
+      },
+      body: JSON.stringify({ messages }),
+      signal,
+    });
+    if (res.status === 401) {
+      useUserStore.getState().logout();
+      window.location.href = '/login';
+      throw new Error('登录已过期，请重新登录');
+    }
+    if (!res.body) throw new Error('流式响应不可用');
+    const reader = res.body.getReader();
+    const decoder = new TextDecoder('utf-8');
+    let buffer = '';
+    let fullContent = '';
+    for (;;) {
+      const { done, value } = await reader.read();
+      if (done) break;
+      buffer += decoder.decode(value, { stream: true });
+      // SSE 事件之间以 \n\n 分隔
+      const events = buffer.split('\n\n');
+      buffer = events.pop() ?? '';
+      for (const evt of events) {
+        const lines = evt.split('\n');
+        let eventType = 'message';
+        let dataLines: string[] = [];
+        for (const ln of lines) {
+          if (ln.startsWith('event:')) eventType = ln.slice(6).trim();
+          else if (ln.startsWith('data:')) dataLines.push(ln.slice(5).trim());
+        }
+        const data = dataLines.join('');
+        if (!data) continue;
+        if (eventType === 'error') {
+          // {code, message}
+          try {
+            const obj = JSON.parse(data);
+            throw new Error(obj.message || 'LLM 流式响应错误');
+          } catch (parseErr) {
+            if (parseErr instanceof Error && parseErr.message) throw parseErr;
+            throw new Error('LLM 流式响应错误');
+          }
+        }
+        if (eventType === 'done') {
+          // 服务端结束（可能含 model 字段）
+          continue;
+        }
+        // message：data 是 {"content":"..."}
+        try {
+          const obj = JSON.parse(data);
+          if (typeof obj.content === 'string') {
+            fullContent += obj.content;
+            onChunk(obj.content);
+          }
+        } catch {
+          // 非 JSON 数据忽略（如 [DONE] 或注释）
+        }
+      }
+    }
+    return fullContent;
+  },
   // 获取当前用户对话历史（按时间正序，最近 100 条）
   getHistory: () => http.get<{ list: ChatHistoryItem[]; total: number }>('/aiops/chat-history'),
   // 清空当前用户对话历史