diff --git a/backend/internal/api/handlers/llm.go b/backend/internal/api/handlers/llm.go index 8b3287a..e7cc8a3 100644 --- a/backend/internal/api/handlers/llm.go +++ b/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 { diff --git a/backend/internal/api/routes/routes.go b/backend/internal/api/routes/routes.go index 52ab525..3acb1a9 100644 --- a/backend/internal/api/routes/routes.go +++ b/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) } diff --git a/docs/dev-log.md b/docs/dev-log.md index 30c7ffd..482e605 100644 --- a/docs/dev-log.md +++ b/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() 自动停止） diff --git a/frontend/src/components/aiops/AIOpsAssistant.tsx b/frontend/src/components/aiops/AIOpsAssistant.tsx index f21cc72..3be4f7d 100644 --- a/frontend/src/components/aiops/AIOpsAssistant.tsx +++ b/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 - const history = newMessages.slice(-10).map((m) => ({ role: m.role, content: m.content })); - const result = await aiApi.chat(history); - setMessages((prev) => [ - ...prev, - { id: (Date.now() + 1).toString(), role: 'assistant', content: result.content, timestamp: new Date().toISOString() }, - ]); + // 仅保留最近 10 条上下文避免 token 超限（不含刚插入的空助手消息） // Logan.Li + const history = [...messages, userMsg] + .slice(-10) + .map((m) => ({ role: m.role, content: m.content })); + await aiApi.chatStream(history, (delta) => { + 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) => [ - ...prev, - { id: (Date.now() + 1).toString(), role: 'assistant', content: `⚠️ ${errMsg}`, timestamp: new Date().toISOString(), isError: true }, - ]); + setMessages((prev) => + prev.map((m) => + 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() { )} )} - {messages.map((message) => ( + {messages.map((message) => { + // 流式中：助手消息内容为空且正在加载时显示"思考中"占位 // Logan.Li + const isStreaming = isLoading && message.role === 'assistant' && message.content === ''; + return (
@@ -448,27 +472,25 @@ export function AIOpsAssistant() { : 'bg-blue-600 text-white' }`} > - {message.content} + {isStreaming ? ( +
+ + 思考中... +
+ ) : ( + message.content + )} + {/* 流式接收中且已有内容时，显示末尾闪烁光标提示 */} + {isLoading && message.role === 'assistant' && !isStreaming && message.content && ( + + )}
{fmtTime(message.timestamp)}
- ))} - {isLoading && ( -
- - - - - -
-
- - 思考中... -
-
-
- )} + ); + })} + {/* 流式输出时占位气泡已包含"思考中"提示，无需独立 spinner */}
diff --git a/frontend/src/services/api.ts b/frontend/src/services/api.ts index d442f2d..902da16 100644 --- a/frontend/src/services/api.ts +++ b/frontend/src/services/api.ts @@ -673,6 +673,78 @@ export const aiApi = { // AI 对话（messages 含历史 + 当前用户消息） chat: (messages: { role: 'user' | 'assistant'; content: string }[]) => http.post('/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 => { + 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'), // 清空当前用户对话历史