refactor: 拆分 llm.go(526行) 为三文件(llm.go配置/llm_chat.go对话流式/llm_history.go历史),符合400行handler规范
This commit is contained in:
attacker committed
1 parent
94e40ebd3f
commit
d4fe62f053
3 files changed
+383
-362
No files matched your search
@@ -1,16 +1,10 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"ezops/internal/api/middleware"
|
||||
"ezops/internal/models"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
@@ -130,10 +124,10 @@ func (h *LLMHandler) SaveLLMConfig(c *gin.Context) {
|
||||
|
||||
// OpenAI 兼容请求/响应结构（仅取必要字段） // Logan.Li
|
||||
type openaiChatRequest struct {
|
||||
Model string `json:"model"`
|
||||
Model string `json:"model"`
|
||||
Messages []models.ChatMessage `json:"messages"`
|
||||
Temperature float64 `json:"temperature"`
|
||||
MaxTokens int `json:"max_tokens,omitempty"`
|
||||
Temperature float64 `json:"temperature"`
|
||||
MaxTokens int `json:"max_tokens,omitempty"`
|
||||
}
|
||||
type openaiChoice struct {
|
||||
Message models.ChatMessage `json:"message"`
|
||||
@@ -171,356 +165,3 @@ const defaultSystemPrompt = `你是 EZOps AIOps 智能运维助手，专注于
|
||||
3. 生成 Shell / Ansible / Python 运维脚本
|
||||
4. 提供容量规划、成本优化、高可用架构建议
|
||||
回答请简洁、专业、可执行，优先给出具体命令或步骤。使用中文回答。`
|
||||
|
||||
// Chat AI 对话代理（OpenAI 兼容接口） // Logan.Li
|
||||
// POST /api/v1/aiops/chat
|
||||
func (h *LLMHandler) Chat(c *gin.Context) {
|
||||
var cfg models.LLMConfig
|
||||
if err := h.db.First(&cfg).Error; err != nil || !cfg.Enabled {
|
||||
c.JSON(http.StatusOK, models.Response{Code: 400, Message: "LLM 服务未配置或未启用，请在「模型配置」中完成设置"})
|
||||
return
|
||||
}
|
||||
|
||||
var req models.ChatRequest
|
||||
if err := c.ShouldBindJSON(&req); err != nil {
|
||||
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "请求参数错误: " + err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
// 组装系统提示词
|
||||
systemPrompt := cfg.SystemPrompt
|
||||
if strings.TrimSpace(systemPrompt) == "" {
|
||||
systemPrompt = defaultSystemPrompt
|
||||
}
|
||||
messages := append([]models.ChatMessage{{Role: "system", Content: systemPrompt}}, req.Messages...)
|
||||
|
||||
// 构造 OpenAI 兼容请求
|
||||
payload := openaiChatRequest{
|
||||
Model: cfg.Model,
|
||||
Messages: messages,
|
||||
Temperature: cfg.Temperature,
|
||||
MaxTokens: cfg.MaxTokens,
|
||||
}
|
||||
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 {
|
||||
c.JSON(http.StatusOK, models.Response{Code: 500, Message: "构造请求失败: " + err.Error()})
|
||||
return
|
||||
}
|
||||
httpReq.Header.Set("Content-Type", "application/json")
|
||||
httpReq.Header.Set("Authorization", "Bearer "+cfg.APIKey)
|
||||
|
||||
resp, err := h.client.Do(httpReq)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusOK, models.Response{Code: 502, Message: "LLM 服务连接失败，请检查网络或 Endpoint 配置: " + err.Error()})
|
||||
return
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
raw, _ := io.ReadAll(resp.Body)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
// 提取上游错误信息
|
||||
msg := fmt.Sprintf("LLM 服务返回错误 (HTTP %d)", resp.StatusCode)
|
||||
if upErr := parseOpenAIError(raw); upErr != "" {
|
||||
msg += ": " + upErr
|
||||
}
|
||||
c.JSON(http.StatusOK, models.Response{Code: resp.StatusCode, Message: msg})
|
||||
return
|
||||
}
|
||||
|
||||
var chatResp openaiChatResponse
|
||||
if err := json.Unmarshal(raw, &chatResp); err != nil || len(chatResp.Choices) == 0 {
|
||||
c.JSON(http.StatusOK, models.Response{Code: 500, Message: "解析 LLM 响应失败"})
|
||||
return
|
||||
}
|
||||
replyContent := chatResp.Choices[0].Message.Content
|
||||
|
||||
// 持久化对话历史（按用户隔离，成对写入 user + assistant） // Logan.Li
|
||||
// 仅在成功时入库；错误气泡为瞬时 UI 不持久化
|
||||
uid := middleware.GetUserID(c)
|
||||
if uid != 0 && len(req.Messages) > 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: replyContent, Model: cfg.Model, CreatedAt: now.Add(time.Millisecond)},
|
||||
})
|
||||
}
|
||||
|
||||
c.JSON(http.StatusOK, models.Response{Code: 200, Data: models.ChatResponse{
|
||||
Content: replyContent,
|
||||
Model: cfg.Model,
|
||||
TokensUsed: chatResp.Usage.TotalTokens,
|
||||
}})
|
||||
}
|
||||
|
||||
// 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 {
|
||||
Error struct {
|
||||
Message string `json:"message"`
|
||||
} `json:"error"`
|
||||
}
|
||||
if json.Unmarshal(raw, &m) == nil && m.Error.Message != "" {
|
||||
return m.Error.Message
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// TestLLMConnection 测试 LLM 连接（发送一条 ping 消息） // Logan.Li
|
||||
// POST /api/v1/aiops/llm-config/test
|
||||
func (h *LLMHandler) TestLLMConnection(c *gin.Context) {
|
||||
var cfg models.LLMConfig
|
||||
if err := h.db.First(&cfg).Error; err != nil {
|
||||
c.JSON(http.StatusOK, models.Response{Code: 400, Message: "请先保存配置"})
|
||||
return
|
||||
}
|
||||
payload := openaiChatRequest{
|
||||
Model: cfg.Model,
|
||||
Messages: []models.ChatMessage{{Role: "user", Content: "ping"}},
|
||||
Temperature: 0,
|
||||
MaxTokens: 16,
|
||||
}
|
||||
body, _ := json.Marshal(payload)
|
||||
endpoint := resolveEndpoint(cfg.Provider, cfg.Endpoint)
|
||||
httpReq, _ := http.NewRequestWithContext(c.Request.Context(), http.MethodPost, endpoint+"/chat/completions", bytes.NewReader(body))
|
||||
httpReq.Header.Set("Content-Type", "application/json")
|
||||
httpReq.Header.Set("Authorization", "Bearer "+cfg.APIKey)
|
||||
resp, err := h.client.Do(httpReq)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusOK, models.Response{Code: 502, Message: "连接失败: " + err.Error()})
|
||||
return
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
raw, _ := io.ReadAll(resp.Body)
|
||||
msg := fmt.Sprintf("HTTP %d", resp.StatusCode)
|
||||
if e := parseOpenAIError(raw); e != "" {
|
||||
msg += ": " + e
|
||||
}
|
||||
c.JSON(http.StatusOK, models.Response{Code: resp.StatusCode, Message: msg})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "连接成功"})
|
||||
}
|
||||
|
||||
// GetChatHistory 获取当前用户的对话历史（按时间正序，仅返回最近 100 条避免过大） // Logan.Li
|
||||
// GET /api/v1/aiops/chat-history
|
||||
func (h *LLMHandler) GetChatHistory(c *gin.Context) {
|
||||
uid := middleware.GetUserID(c)
|
||||
var rows []models.ChatHistory
|
||||
// 子查询取最近 100 条后再正序返回，保证前端渲染顺序正确 // Logan.Li
|
||||
h.db.Where("user_id = ?", uid).
|
||||
Order("id DESC").
|
||||
Limit(100).
|
||||
Find(&rows)
|
||||
// 反转为时间正序（最早在前）
|
||||
for i, j := 0, len(rows)-1; i < j; i, j = i+1, j-1 {
|
||||
rows[i], rows[j] = rows[j], rows[i]
|
||||
}
|
||||
c.JSON(http.StatusOK, models.Response{Code: 200, Data: gin.H{
|
||||
"list": rows,
|
||||
"total": len(rows),
|
||||
}})
|
||||
}
|
||||
|
||||
// ClearChatHistory 清空当前用户的对话历史 // Logan.Li
|
||||
// DELETE /api/v1/aiops/chat-history
|
||||
func (h *LLMHandler) ClearChatHistory(c *gin.Context) {
|
||||
uid := middleware.GetUserID(c)
|
||||
if err := h.db.Where("user_id = ?", uid).Delete(&models.ChatHistory{}).Error; err != nil {
|
||||
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "清除失败: " + err.Error()})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "对话历史已清空"})
|
||||
}
|
||||
@@ -0,0 +1,339 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"ezops/internal/api/middleware"
|
||||
"ezops/internal/models"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// Chat AI 对话代理（OpenAI 兼容接口） // Logan.Li
|
||||
// POST /api/v1/aiops/chat
|
||||
func (h *LLMHandler) Chat(c *gin.Context) {
|
||||
var cfg models.LLMConfig
|
||||
if err := h.db.First(&cfg).Error; err != nil || !cfg.Enabled {
|
||||
c.JSON(http.StatusOK, models.Response{Code: 400, Message: "LLM 服务未配置或未启用，请在「模型配置」中完成设置"})
|
||||
return
|
||||
}
|
||||
|
||||
var req models.ChatRequest
|
||||
if err := c.ShouldBindJSON(&req); err != nil {
|
||||
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "请求参数错误: " + err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
// 组装系统提示词
|
||||
systemPrompt := cfg.SystemPrompt
|
||||
if strings.TrimSpace(systemPrompt) == "" {
|
||||
systemPrompt = defaultSystemPrompt
|
||||
}
|
||||
messages := append([]models.ChatMessage{{Role: "system", Content: systemPrompt}}, req.Messages...)
|
||||
|
||||
// 构造 OpenAI 兼容请求
|
||||
payload := openaiChatRequest{
|
||||
Model: cfg.Model,
|
||||
Messages: messages,
|
||||
Temperature: cfg.Temperature,
|
||||
MaxTokens: cfg.MaxTokens,
|
||||
}
|
||||
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 {
|
||||
c.JSON(http.StatusOK, models.Response{Code: 500, Message: "构造请求失败: " + err.Error()})
|
||||
return
|
||||
}
|
||||
httpReq.Header.Set("Content-Type", "application/json")
|
||||
httpReq.Header.Set("Authorization", "Bearer "+cfg.APIKey)
|
||||
|
||||
resp, err := h.client.Do(httpReq)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusOK, models.Response{Code: 502, Message: "LLM 服务连接失败，请检查网络或 Endpoint 配置: " + err.Error()})
|
||||
return
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
raw, _ := io.ReadAll(resp.Body)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
// 提取上游错误信息
|
||||
msg := fmt.Sprintf("LLM 服务返回错误 (HTTP %d)", resp.StatusCode)
|
||||
if upErr := parseOpenAIError(raw); upErr != "" {
|
||||
msg += ": " + upErr
|
||||
}
|
||||
c.JSON(http.StatusOK, models.Response{Code: resp.StatusCode, Message: msg})
|
||||
return
|
||||
}
|
||||
|
||||
var chatResp openaiChatResponse
|
||||
if err := json.Unmarshal(raw, &chatResp); err != nil || len(chatResp.Choices) == 0 {
|
||||
c.JSON(http.StatusOK, models.Response{Code: 500, Message: "解析 LLM 响应失败"})
|
||||
return
|
||||
}
|
||||
replyContent := chatResp.Choices[0].Message.Content
|
||||
|
||||
// 持久化对话历史（按用户隔离，成对写入 user + assistant） // Logan.Li
|
||||
// 仅在成功时入库；错误气泡为瞬时 UI 不持久化
|
||||
uid := middleware.GetUserID(c)
|
||||
if uid != 0 && len(req.Messages) > 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: replyContent, Model: cfg.Model, CreatedAt: now.Add(time.Millisecond)},
|
||||
})
|
||||
}
|
||||
|
||||
c.JSON(http.StatusOK, models.Response{Code: 200, Data: models.ChatResponse{
|
||||
Content: replyContent,
|
||||
Model: cfg.Model,
|
||||
TokensUsed: chatResp.Usage.TotalTokens,
|
||||
}})
|
||||
}
|
||||
|
||||
// ChatStream AI 流式对话代理（SSE） // Logan.Li
|
||||
// POST /api/v1/aiops/chat/stream
|
||||
// 客户端用 EventSource/fetch-stream 消费，每条 SSE data 为 {"content": "..."}
|
||||
// 出错时发送 event: error，data 为 {"message": "..."}
|
||||
// 完成后服务端持久化 user + assistant 两条历史，event: 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 {
|
||||
Error struct {
|
||||
Message string `json:"message"`
|
||||
} `json:"error"`
|
||||
}
|
||||
if json.Unmarshal(raw, &m) == nil && m.Error.Message != "" {
|
||||
return m.Error.Message
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// TestLLMConnection 测试 LLM 连接（发送一条 ping 消息） // Logan.Li
|
||||
// POST /api/v1/aiops/llm-config/test
|
||||
func (h *LLMHandler) TestLLMConnection(c *gin.Context) {
|
||||
var cfg models.LLMConfig
|
||||
if err := h.db.First(&cfg).Error; err != nil {
|
||||
c.JSON(http.StatusOK, models.Response{Code: 400, Message: "请先保存配置"})
|
||||
return
|
||||
}
|
||||
payload := openaiChatRequest{
|
||||
Model: cfg.Model,
|
||||
Messages: []models.ChatMessage{{Role: "user", Content: "ping"}},
|
||||
Temperature: 0,
|
||||
MaxTokens: 16,
|
||||
}
|
||||
body, _ := json.Marshal(payload)
|
||||
endpoint := resolveEndpoint(cfg.Provider, cfg.Endpoint)
|
||||
httpReq, _ := http.NewRequestWithContext(c.Request.Context(), http.MethodPost, endpoint+"/chat/completions", bytes.NewReader(body))
|
||||
httpReq.Header.Set("Content-Type", "application/json")
|
||||
httpReq.Header.Set("Authorization", "Bearer "+cfg.APIKey)
|
||||
resp, err := h.client.Do(httpReq)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusOK, models.Response{Code: 502, Message: "连接失败: " + err.Error()})
|
||||
return
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
raw, _ := io.ReadAll(resp.Body)
|
||||
msg := fmt.Sprintf("HTTP %d", resp.StatusCode)
|
||||
if e := parseOpenAIError(raw); e != "" {
|
||||
msg += ": " + e
|
||||
}
|
||||
c.JSON(http.StatusOK, models.Response{Code: resp.StatusCode, Message: msg})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "连接成功"})
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"ezops/internal/api/middleware"
|
||||
"ezops/internal/models"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// GetChatHistory 获取当前用户的对话历史（按时间正序，仅返回最近 100 条避免过大） // Logan.Li
|
||||
// GET /api/v1/aiops/chat-history
|
||||
func (h *LLMHandler) GetChatHistory(c *gin.Context) {
|
||||
uid := middleware.GetUserID(c)
|
||||
var rows []models.ChatHistory
|
||||
// 子查询取最近 100 条后再正序返回，保证前端渲染顺序正确 // Logan.Li
|
||||
h.db.Where("user_id = ?", uid).
|
||||
Order("id DESC").
|
||||
Limit(100).
|
||||
Find(&rows)
|
||||
// 反转为时间正序（最早在前）
|
||||
for i, j := 0, len(rows)-1; i < j; i, j = i+1, j-1 {
|
||||
rows[i], rows[j] = rows[j], rows[i]
|
||||
}
|
||||
c.JSON(http.StatusOK, models.Response{Code: 200, Data: gin.H{
|
||||
"list": rows,
|
||||
"total": len(rows),
|
||||
}})
|
||||
}
|
||||
|
||||
// ClearChatHistory 清空当前用户的对话历史 // Logan.Li
|
||||
// DELETE /api/v1/aiops/chat-history
|
||||
func (h *LLMHandler) ClearChatHistory(c *gin.Context) {
|
||||
uid := middleware.GetUserID(c)
|
||||
if err := h.db.Where("user_id = ?", uid).Delete(&models.ChatHistory{}).Error; err != nil {
|
||||
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "清除失败: " + err.Error()})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "对话历史已清空"})
|
||||
}
|
||||
Reference in new issue
Block a user