Files
aiops/backend/internal/api/handlers/llm.go
T

527 lines
17 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
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"
"gorm.io/gorm"
)
// LLMHandler 大语言模型配置 + 对话代理 // Logan.Li
type LLMHandler struct {
db *gorm.DB
client *http.Client
}
// NewLLMHandler 创建 LLM 处理器 // Logan.Li
func NewLLMHandler(db *gorm.DB) *LLMHandler {
return &LLMHandler{
db: db,
// 超时控制:LLM 推理可能较慢,给 120s
client: &http.Client{Timeout: 120 * time.Second},
}
}
// maskAPIKey 脱敏 API Key(仅保留后 4 位) // Logan.Li
func maskAPIKey(key string) string {
if len(key) <= 4 {
return strings.Repeat("*", len(key))
}
return strings.Repeat("*", len(key)-4) + key[len(key)-4:]
}
// getActiveConfig 读取启用的配置(脱敏后) // Logan.Li
func (h *LLMHandler) getActiveConfig(maskKey bool) (models.LLMConfig, error) {
var cfg models.LLMConfig
if err := h.db.First(&cfg).Error; err != nil {
return cfg, err
}
if maskKey {
cfg.APIKey = maskAPIKey(cfg.APIKey)
}
return cfg, nil
}
// GetLLMConfig 获取 LLM 配置(API Key 脱敏) // Logan.Li
// GET /api/v1/aiops/llm-config
func (h *LLMHandler) GetLLMConfig(c *gin.Context) {
cfg, err := h.getActiveConfig(true)
if err != nil {
// 未配置返回空对象(前端渲染空表单)
c.JSON(http.StatusOK, models.Response{Code: 200, Data: gin.H{
"provider": "", "model": "", "api_key": "",
"endpoint": "", "temperature": 0.7, "max_tokens": 2000,
"system_prompt": "", "enabled": false, "configured": false,
}})
return
}
c.JSON(http.StatusOK, models.Response{Code: 200, Data: gin.H{
"id": cfg.ID,
"provider": cfg.Provider,
"model": cfg.Model,
"api_key": cfg.APIKey,
"endpoint": cfg.Endpoint,
"temperature": cfg.Temperature,
"max_tokens": cfg.MaxTokens,
"system_prompt": cfg.SystemPrompt,
"enabled": cfg.Enabled,
"configured": true,
}})
}
// SaveLLMConfig 保存 LLM 配置(upsert 单行) // Logan.Li
// PUT /api/v1/aiops/llm-config
func (h *LLMHandler) SaveLLMConfig(c *gin.Context) {
var req models.SaveLLMConfigRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "请求参数错误: " + err.Error()})
return
}
var cfg models.LLMConfig
h.db.First(&cfg)
// 若前端回传脱敏值(含 ****),保留原 Key
if strings.Contains(req.APIKey, "****") && cfg.ID != 0 {
req.APIKey = cfg.APIKey
}
cfg.Provider = req.Provider
cfg.Model = req.Model
cfg.APIKey = req.APIKey
cfg.Endpoint = req.Endpoint
cfg.Temperature = req.Temperature
cfg.MaxTokens = req.MaxTokens
cfg.SystemPrompt = req.SystemPrompt
cfg.Enabled = req.Enabled
if cfg.ID == 0 {
if err := h.db.Create(&cfg).Error; err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "保存失败: " + err.Error()})
return
}
} else {
if err := h.db.Save(&cfg).Error; err != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "保存失败: " + err.Error()})
return
}
}
// 返回脱敏后的配置
cfg.APIKey = maskAPIKey(cfg.APIKey)
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "保存成功", Data: gin.H{
"id": cfg.ID,
"provider": cfg.Provider,
"model": cfg.Model,
"api_key": cfg.APIKey,
"endpoint": cfg.Endpoint,
"temperature": cfg.Temperature,
"max_tokens": cfg.MaxTokens,
"system_prompt": cfg.SystemPrompt,
"enabled": cfg.Enabled,
"configured": true,
}})
}
// OpenAI 兼容请求/响应结构(仅取必要字段) // Logan.Li
type openaiChatRequest struct {
Model string `json:"model"`
Messages []models.ChatMessage `json:"messages"`
Temperature float64 `json:"temperature"`
MaxTokens int `json:"max_tokens,omitempty"`
}
type openaiChoice struct {
Message models.ChatMessage `json:"message"`
}
type openaiChatResponse struct {
Choices []openaiChoice `json:"choices"`
Usage struct {
TotalTokens int `json:"total_tokens"`
} `json:"usage"`
}
// resolveEndpoint 根据提供商补全默认 endpoint // Logan.Li
func resolveEndpoint(provider, endpoint string) string {
if endpoint != "" {
return strings.TrimRight(endpoint, "/")
}
switch provider {
case "deepseek":
return "https://api.deepseek.com/v1"
case "qwen":
return "https://dashscope.aliyuncs.com/compatible-mode/v1"
case "moonshot":
return "https://api.moonshot.cn/v1"
case "ollama":
return "http://localhost:11434/v1"
default: // openai / custom
return "https://api.openai.com/v1"
}
}
// defaultSystemPrompt AIOps 运维助手默认系统提示词 // Logan.Li
const defaultSystemPrompt = `你是 EZOps AIOps 智能运维助手,专注于企业级运维场景。你可以:
1. 分析服务器、容器、K8s 集群的性能瓶颈与异常
2. 解读告警事件并给出根因分析与处置建议
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: "对话历史已清空"})
}