package handler import ( "context" "encoding/json" "fmt" "net/http" "strings" "time" "github.com/gin-gonic/gin" "flowstack/internal/database" "flowstack/internal/llm" "flowstack/internal/model" ) // buildPlatformContext 平台实时数据注入(降低 AI 幻觉):流水线概况 + 最近运行 func buildPlatformContext() string { var pipelines []model.Pipeline database.DB.Find(&pipelines) var runs []model.PipelineRun database.DB.Order("started_at DESC").Limit(5).Find(&runs) running := 0 for _, r := range runs { if r.Status == "running" || r.Status == "pending" { running++ } } var b strings.Builder fmt.Fprintf(&b, "【平台实时数据】\n流水线总数: %d(运行中 %d)\n", len(pipelines), running) if len(runs) > 0 { b.WriteString("最近运行:\n") for _, r := range runs { fmt.Fprintf(&b, "- %s %s %s %s %s\n", r.PipelineName, r.RunID, r.Branch, r.Status, fmtDurMs(r.DurationMs)) } } return b.String() } func fmtDurMs(ms int64) string { if ms <= 0 { return "-" } return fmt.Sprintf("%ds", ms/1000) } // AiChatStream POST /api/ai/chat/stream —— 流式对话(SSE) func AiChatStream(c *gin.Context) { var body struct { Messages []ChatMsg `json:"messages" binding:"required"` } if err := c.ShouldBindJSON(&body); err != nil || len(body.Messages) == 0 { c.JSON(http.StatusBadRequest, gin.H{"error": "messages 不能为空"}) return } var lastUser string for i := len(body.Messages) - 1; i >= 0; i-- { if body.Messages[i].Role == "user" { lastUser = body.Messages[i].Content break } } if lastUser == "" { c.JSON(http.StatusBadRequest, gin.H{"error": "没有用户消息"}) return } candidates := allCandidates() names := make([]string, 0, len(candidates)) if len(candidates) > 200 { candidates = candidates[:200] } for _, cd := range candidates { names = append(names, cd.Name) } ruleRes := parseByRules(lastUser, candidates) client := llmClientFromSettings() if client == nil { // 未配置 LLM:直接返回规则结果(不流式) reply, action := ruleReply(ruleRes, candidates) c.JSON(http.StatusOK, gin.H{"reply": reply, "viaAI": false, "action": action}) return } c.Header("Content-Type", "text/event-stream") c.Header("Cache-Control", "no-cache") c.Header("Connection", "keep-alive") c.Header("X-Accel-Buffering", "no") flush := func() { if f, ok := c.Writer.(http.Flusher); ok { f.Flush() } } emit := func(event string, data any) { b, _ := json.Marshal(data) fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", event, b) flush() } system := fmt.Sprintf(`你是 FlowStack 发布系统的智能发布助手。用户会用自然语言描述要发布的服务。 %s 可发布的服务清单(流水线与 Jenkins Job 名称)如下: %s 规则: - 若用户意图是发布/构建服务:只输出 JSON(不要 markdown 代码块、不要多余文字): {"reply":"一句中文确认语","jobs":["服务名1"],"branch":"分支或留空","env":"环境或留空"} jobs 必须来自上面的清单按原文匹配;用户提到的服务不在清单时 jobs 留空并在 reply 中说明。 - 其他问题(如询问状态、概念解释):用简洁中文自然语言回答,禁止编造清单中不存在的服务或字段。`, buildPlatformContext(), strings.Join(names, "\n")) emit("start", gin.H{"viaAI": true}) ctx := c.Request.Context() full, err := client.ChatStream(ctx, system, lastUser, func(d llm.StreamDelta) { if d.Content != "" || d.Reasoning != "" { emit("delta", gin.H{"content": d.Content, "reasoning": d.Reasoning}) } }) if err != nil { // LLM 失败:降级规则结果 reply, action := ruleReply(ruleRes, candidates) emit("done", gin.H{"reply": reply + "\n(AI 调用失败已降级本地规则: " + err.Error() + ")", "viaAI": false, "action": action}) return } reply, action := parseLLMJson(full, candidates) if action == nil && len(ruleRes.Jobs) > 0 { // LLM 没识别出发布指令但规则识别到了,合并规则结果 _, action = ruleReply(ruleRes, candidates) } emit("done", gin.H{"reply": reply, "viaAI": true, "action": action}) } func ruleReply(ruleRes ruleParseResult, candidates []Candidate) (string, *aiReleaseAction) { if len(ruleRes.Jobs) == 0 { return "未识别到要发布的服务。请直接告诉我服务名,例如:发布 cex-agent-server、cex-agent-api 到 pre 分支。", nil } jobs := make([]aiActionJob, 0, len(ruleRes.Jobs)) for _, name := range ruleRes.Jobs { m := matchCandidate(name, candidates) jobs = append(jobs, aiActionJob{Name: name, Matched: m != nil, Source: sourceOf(m)}) } action := &aiReleaseAction{Type: "release", Jobs: jobs, Branch: ruleRes.Branch, Env: ruleRes.Env} return fmt.Sprintf("识别到 %d 个服务%s,确认后开始发布。", len(jobs), branchSuffix(ruleRes.Branch)), action } // analyzeFailure 发布失败时用 LLM 分析日志尾部(opsx-loki AnalyzeJenkins 同款 prompt) func analyzeFailure(run model.PipelineRun) { client := llmClientFromSettings() if client == nil { return } var logs []model.LogEntry database.DB.Where("run_id = ?", run.ID).Order("timestamp DESC").Limit(100).Find(&logs) if len(logs) == 0 { return } // 倒序取出后反转为正序 for i, j := 0, len(logs)-1; i < j; i, j = i+1, j-1 { logs[i], logs[j] = logs[j], logs[i] } var b strings.Builder for _, l := range logs { b.WriteString(l.Message) b.WriteString("\n") } system := `你是资深发布助手。根据流水线执行日志分析发布失败原因。用简洁中文输出,结构固定: 1. 结论:一句话说明失败原因 2. 失败阶段或关键步骤 3. 根因(如编译错误、依赖缺失、测试失败、镜像推送失败、部署超时) 4. 处理建议(1-3 条,可执行) 不要复述整段日志。` user := fmt.Sprintf("流水线: %s\n分支: %s\n状态: %s\n\n日志尾部:\n%s", run.PipelineName, run.Branch, run.Status, b.String()) ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) defer cancel() reply, err := client.Chat(ctx, system, user) if err != nil { return } database.DB.Model(&model.PipelineRun{}).Where("id = ?", run.ID).Update("analysis", reply) }