// Package core 实现 Agent CLI 的 ReAct 核心引擎。 // Logan.Li // 设计参考 opsxcli internal/agent/core,剥离 TUI(Bubble Tea),改用 EventCallback 推送事件流 // (由 controller 序列化为 WebSocket 消息推给前端 + 持久化到 DB)。 // 核心算法保留:16 轮 ReAct 主循环、自适应检查点、循环检测(同参>3 次拒绝)、连续失败追踪、Evolver 集成。 package core import ( "context" "encoding/json" "fmt" "strings" "sync" "time" "ezops/pkg/agentcli/evolver" "ezops/pkg/agentcli/llm" "ezops/pkg/agentcli/prompt" "ezops/pkg/agentcli/safety" "ezops/pkg/agentcli/tools" ) // EventType 事件类型(驱动前端渲染与 DB 持久化) // Logan.Li type EventType string const ( EventThought EventType = "thought" // AI 思考/推理文本(流式结束后的完整内容) EventThoughtDelta EventType = "thought_delta" // AI 思考流式增量 token EventToolCall EventType = "tool_call" // 即将执行的工具(名/参数/风险) EventToolResult EventType = "tool_result" // 工具执行结果(输出/成败/耗时) EventProgress EventType = "progress" // 轮次进度 + token EventDone EventType = "done" // 任务完成(最终答案) EventError EventType = "error" // 错误(含达到最大轮次) ) // Event 引擎产生的事件(controller 序列化后通过 WebSocket 推送) // Logan.Li type Event struct { Type EventType `json:"type"` Round int `json:"round,omitempty"` Thought string `json:"thought,omitempty"` ToolName string `json:"tool_name,omitempty"` ToolArgs map[string]interface{} `json:"tool_args,omitempty"` Risk string `json:"risk,omitempty"` Output string `json:"output,omitempty"` Success bool `json:"success,omitempty"` Duration string `json:"duration,omitempty"` Tokens int `json:"tokens,omitempty"` Answer string `json:"answer,omitempty"` Message string `json:"message,omitempty"` } // EventCallback 事件回调(controller 注入) // Logan.Li type EventCallback func(Event) // RunResult ReAct 执行结果(含对话历史供多轮追问) // Logan.Li type RunResult struct { Success bool Output string History []llm.Message // system 之后的对话历史 TotalTokens int } // Config Agent 运行配置 // Logan.Li type Config struct { Mode string // host | kubernetes MaxIterations int Temperature float64 ToolTimeout time.Duration MaxTokens int SafetyMode safety.Mode OutputMaxLength int MaxContextTokens int MemoryDir string // Evolver 记忆目录 RoundOffset int // 轮次偏移(多轮追问时接续上一轮编号) // Logan.Li } // DefaultConfig 默认配置 // Logan.Li func DefaultConfig() *Config { return &Config{ MaxIterations: 16, Temperature: 0.3, ToolTimeout: 120 * time.Second, MaxTokens: 4096, SafetyMode: safety.ModeBalanced, OutputMaxLength: 10000, MaxContextTokens: 8000, MemoryDir: "data/agentcli", } } // Agent ReAct 核心引擎 // Logan.Li type Agent struct { llm llm.Client registry *tools.Registry config *Config safetyCtl *safety.Controller evolver *evolver.EvolverEngine callback EventCallback totalTokens int mu sync.Mutex } // NewAgent 创建引擎。evolver 为 nil 时降级(无自学习)。 // Logan.Li func NewAgent(client llm.Client, registry *tools.Registry, cfg *Config, sc *safety.Controller, ev *evolver.EvolverEngine) *Agent { if cfg == nil { cfg = DefaultConfig() } if sc == nil { sc = safety.NewController(cfg.SafetyMode) } if cfg.MaxContextTokens <= 0 { cfg.MaxContextTokens = 8000 } return &Agent{llm: client, registry: registry, config: cfg, safetyCtl: sc, evolver: ev} } // SetCallback 注入事件回调 // Logan.Li func (a *Agent) SetCallback(cb EventCallback) { a.callback = cb } // emit 安全触发回调 // Logan.Li func (a *Agent) emit(e Event) { if a.callback != nil { a.callback(e) } } // Run 执行单次任务(ReAct 循环 + Evolver 集成);prior 为不含 system 的历史对话 // Logan.Li func (a *Agent) Run(ctx context.Context, query, serverInfo string, prior []llm.Message, priorTokens int) (*RunResult, error) { a.totalTokens = priorTokens start := time.Now() // Evolver PREDICT:获取历史经验注入 system prompt evolveCtx := "" if a.evolver != nil { evolveCtx = a.evolver.GetContextForPrompt(query) } // 工具描述 toolDescs := a.buildToolDescriptions() systemPrompt := prompt.BuildSystemPrompt(a.config.Mode, serverInfo, toolDescs, evolveCtx) messages := []llm.Message{{Role: "system", Content: systemPrompt}} if len(prior) > 0 { messages = append(messages, prior...) } messages = append(messages, llm.Message{Role: "user", Content: query}) llmTools := a.registry.ToLLMTools() toolCallHistory := make(map[string]int) // 循环检测 var toolRecords []evolver.ToolCallRecord var consecutiveFailures int var totalToolCalls int for iter := 0; iter < a.config.MaxIterations; iter++ { // 检查中断 if err := ctx.Err(); err != nil { return nil, err } a.emit(Event{Type: EventProgress, Round: a.roundNum(iter)}) // 自适应检查点 if hint := a.checkAdaptiveCheckpoint(iter, toolCallHistory, consecutiveFailures, totalToolCalls); hint != "" { messages = append(messages, llm.Message{Role: "user", Content: hint}) } // 裁剪历史 trimmed := a.trimMessages(messages) // ReAct 须用 Complete:Stream 无法组装 tool_calls/tool_use(GLM/Claude 等会只回文字不执行) // Logan.Li round := a.roundNum(iter) resp, err := a.llm.Complete(ctx, &llm.CompletionRequest{ Messages: trimmed, Tools: llmTools, Temperature: a.config.Temperature, MaxTokens: a.config.MaxTokens, }) if err != nil { a.emit(Event{Type: EventError, Message: "LLM 调用失败: " + err.Error()}) return nil, fmt.Errorf("LLM调用失败: %w", err) } a.totalTokens += resp.Usage.TotalTokens a.emit(Event{Type: EventProgress, Tokens: a.totalTokens}) messages = append(messages, resp.Message) // 无工具调用:首轮若仅描述计划未动手,注入提示重试一次 // Logan.Li if len(resp.Message.ToolCalls) == 0 && totalToolCalls == 0 && iter < 2 { // 计划阶段可展示简短思考,便于用户感知进度 // Logan.Li if thought := strings.TrimSpace(resp.Message.Content); thought != "" { a.emit(Event{Type: EventThought, Round: round, Thought: thought}) } messages = append(messages, llm.Message{ Role: "user", Content: fmt.Sprintf("【系统提示】请立即调用 %s 工具获取实际结果,不要仅用文字描述计划。", a.primaryExecTool()), }) continue } // 无工具调用 → 最终答案 if len(resp.Message.ToolCalls) == 0 { answer := strings.TrimSpace(resp.Message.Content) // 部分模型最终轮 content 为空,回退到最近工具输出或默认文案 // Logan.Li if answer == "" { if fallback := lastToolOutput(messages); fallback != "" { answer = "任务已执行完成,命令输出如下:\n" + fallback } else { answer = "任务已完成。" } } a.emit(Event{Type: EventDone, Answer: answer, Tokens: a.totalTokens}) a.triggerEvolve(ctx, query, serverInfo, answer, toolRecords, true, start) return &RunResult{ Success: true, Output: answer, History: a.extractHistory(messages), TotalTokens: a.totalTokens, }, nil } // 中间轮思考(后续还有 tool_calls,与终局 done 分开,避免重复输出) // Logan.Li if thought := strings.TrimSpace(resp.Message.Content); thought != "" { a.emit(Event{Type: EventThought, Round: round, Thought: thought}) } // 处理工具调用 observations, records, failInc := a.processToolCalls(ctx, resp.Message.ToolCalls, toolCallHistory, round) toolRecords = append(toolRecords, records...) messages = append(messages, observations...) totalToolCalls += len(resp.Message.ToolCalls) consecutiveFailures = a.updateConsecutiveFailures(observations, consecutiveFailures) + failInc } // 达到最大轮次 msg := fmt.Sprintf("达到最大迭代次数(%d),任务未能完成。可能任务过于复杂或需要更多步骤。", a.config.MaxIterations) a.emit(Event{Type: EventError, Message: msg, Tokens: a.totalTokens}) a.triggerEvolve(ctx, query, serverInfo, "", toolRecords, false, start) return &RunResult{ Success: false, History: a.extractHistory(messages), TotalTokens: a.totalTokens, }, fmt.Errorf("%s", msg) } // processToolCalls 处理工具调用,返回 observation 消息 + 记录 // Logan.Li func (a *Agent) processToolCalls(ctx context.Context, calls []llm.ToolCall, history map[string]int, round int) ([]llm.Message, []evolver.ToolCallRecord, int) { obs := make([]llm.Message, 0, len(calls)) records := make([]evolver.ToolCallRecord, 0, len(calls)) var failInc int for _, tc := range calls { key := tc.Function.Name + ":" + tc.Function.Arguments history[key]++ // 循环检测 if history[key] > 3 { obs = append(obs, llm.Message{Role: "tool", ToolCallID: tc.ID, Name: tc.Function.Name, Content: "错误: 检测到循环调用,相同参数已调用超过3次,请换一种方式"}) continue } tool, err := a.registry.Get(tc.Function.Name) if err != nil { obs = append(obs, llm.Message{Role: "tool", ToolCallID: tc.ID, Name: tc.Function.Name, Content: fmt.Sprintf("错误: 工具 '%s' 未找到", tc.Function.Name)}) continue } var args map[string]interface{} if err := json.Unmarshal([]byte(tc.Function.Arguments), &args); err != nil { obs = append(obs, llm.Message{Role: "tool", ToolCallID: tc.ID, Name: tc.Function.Name, Content: "错误: 参数解析失败: " + err.Error()}) continue } risk := a.toolRisk(tool, args) a.emit(Event{Type: EventToolCall, Round: round, ToolName: tc.Function.Name, ToolArgs: args, Risk: risk.Display()}) // 安全检查(高危触发 confirmFn,由 controller 异步确认) approved, err := a.safetyCtl.Check(tool, args) if err != nil { obs = append(obs, llm.Message{Role: "tool", ToolCallID: tc.ID, Name: tc.Function.Name, Content: "安全检查失败: " + err.Error()}) continue } if !approved { obs = append(obs, llm.Message{Role: "tool", ToolCallID: tc.ID, Name: tc.Function.Name, Content: "操作已被用户拒绝执行"}) a.emit(Event{Type: EventToolResult, Round: round, ToolName: tc.Function.Name, Output: "(用户拒绝执行)", Success: false}) continue } // 执行 toolStart := time.Now() toolCtx, cancel := context.WithTimeout(ctx, a.config.ToolTimeout) result, execErr := tool.Execute(toolCtx, args) cancel() dur := time.Since(toolStart) var output, errMsg string success := false if result != nil { output, errMsg, success = result.Output, result.Error, result.Success } if execErr != nil { errMsg = execErr.Error() success = false } // 超长命令输出自动分析,避免 token 爆炸 // Logan.Li if (tc.Function.Name == "ssh_execute" || tc.Function.Name == "pod_execute") && success && len(output) > a.autoAnalyzeThreshold() { if analyzed := a.autoAnalyzeOutput(ctx, output); analyzed != "" { output = analyzed } } // 输出截断 if len(output) > a.config.OutputMaxLength { output = output[:a.config.OutputMaxLength] + fmt.Sprintf("\n[输出已截断,原始 %d 字符]", len(output)) } a.emit(Event{Type: EventToolResult, Round: round, ToolName: tc.Function.Name, Output: output, Success: success, Duration: dur.Truncate(time.Millisecond).String()}) records = append(records, evolver.ToolCallRecord{ ToolName: tc.Function.Name, Args: args, Output: output, Duration: dur, Success: success, RiskLevel: int(risk), }) // 构建 observation var content string switch { case execErr != nil: content = "执行错误: " + execErr.Error() failInc++ case !success: content = fmt.Sprintf("执行失败: %s\n输出: %s", errMsg, output) failInc++ default: content = output } obs = append(obs, llm.Message{Role: "tool", ToolCallID: tc.ID, Name: tc.Function.Name, Content: content}) } return obs, records, failInc } // triggerEvolve 异步触发进化(带超时 + panic 恢复,绝不阻塞主循环) // Logan.Li func (a *Agent) triggerEvolve(ctx context.Context, query, serverHost, answer string, records []evolver.ToolCallRecord, success bool, start time.Time) { if a.evolver == nil { return } exec := &evolver.TaskExecution{ Query: query, ToolCalls: records, TotalSteps: len(records), TotalTokens: a.totalTokens, Duration: time.Since(start), Success: success, FinalAnswer: answer, Timestamp: time.Now(), ServerHost: serverHost, } go func() { defer func() { _ = recover() }() eCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() _ = a.evolver.Evolve(eCtx, exec) }() } // roundNum 计算对外展示的 ReAct 轮次(含多轮追问偏移) // Logan.Li func (a *Agent) roundNum(iter int) int { return a.config.RoundOffset + iter + 1 } // primaryExecTool 返回当前模式下的主执行工具名 // Logan.Li func (a *Agent) primaryExecTool() string { if a.config.Mode == "kubernetes" || a.config.Mode == "k8s" { return "kubernetes_get" } return "ssh_execute" } // lastToolOutput 取最近一条工具 observation,供最终答案为空时回退展示 // Logan.Li func lastToolOutput(messages []llm.Message) string { for i := len(messages) - 1; i >= 0; i-- { if messages[i].Role == "tool" { if out := strings.TrimSpace(messages[i].Content); out != "" { return out } } } return "" } // extractHistory 提取 system 之后的对话历史(供多轮追问) // Logan.Li func (a *Agent) extractHistory(messages []llm.Message) []llm.Message { if len(messages) <= 1 { return nil } out := make([]llm.Message, len(messages)-1) copy(out, messages[1:]) return out } // autoAnalyzeThreshold 触发自动 analyze_output 的输出长度阈值 // Logan.Li func (a *Agent) autoAnalyzeThreshold() int { t := a.config.OutputMaxLength / 2 if t < 4000 { return 4000 } return t } // autoAnalyzeOutput 对超长命令输出自动调用 analyze_output // Logan.Li func (a *Agent) autoAnalyzeOutput(ctx context.Context, output string) string { tool, err := a.registry.Get("analyze_output") if err != nil { return "" } r, err := tool.Execute(ctx, map[string]interface{}{"output": output}) if err != nil || r == nil || !r.Success || strings.TrimSpace(r.Output) == "" { return "" } return r.Output + "\n[系统自动分析超长输出]" } // buildToolDescriptions 工具描述文本 // Logan.Li func (a *Agent) buildToolDescriptions() string { var sb strings.Builder for _, t := range a.registry.List() { sb.WriteString(fmt.Sprintf("- %s: %s\n", t.Name(), t.Description())) } return sb.String() } // toolRisk 取工具风险(动态优先) // Logan.Li func (a *Agent) toolRisk(tool tools.Tool, args map[string]interface{}) tools.RiskLevel { if dt, ok := tool.(tools.DynamicRiskTool); ok { return dt.RiskLevelForArgs(args) } return tool.RiskLevel() } // checkAdaptiveCheckpoint 自适应检查点(iter 3/7/11 注入引导) // Logan.Li func (a *Agent) checkAdaptiveCheckpoint(iter int, history map[string]int, consecFail, totalCalls int) string { switch iter { case 3: repeat := 0 for _, c := range history { if c > 1 { repeat += c - 1 } } if totalCalls > 0 && (float64(repeat)/float64(totalCalls) > 0.5 || consecFail >= 3) { return "【系统提示】检测到重复使用相同工具或连续失败,请尝试换一种思路或不同命令。" } case 7: return "【系统提示】已执行到中段,请总结进展。信息足够就直接给出最终答案,否则说明剩余步骤。" case 11: return fmt.Sprintf("【系统提示】仅剩 %d 步,请立即整理已有信息给出最终答案,不要再调用新工具。", a.config.MaxIterations-iter-1) } return "" } // updateConsecutiveFailures 连续失败计数 // Logan.Li func (a *Agent) updateConsecutiveFailures(obs []llm.Message, current int) int { errs := 0 for _, o := range obs { c := o.Content if strings.HasPrefix(c, "错误") || strings.HasPrefix(c, "失败") || strings.HasPrefix(c, "执行错误") || strings.HasPrefix(c, "执行失败") { errs++ } } if errs == 0 { return 0 } return current + errs } // trimMessages 裁剪历史(保留 system + 最近消息,token 不超 90%) // Logan.Li func (a *Agent) trimMessages(msgs []llm.Message) []llm.Message { if len(msgs) <= 1 { return msgs } maxT := a.config.MaxContextTokens * 9 / 10 sysT := estimateTokens(msgs[0].Content) kept := make([]llm.Message, 0, len(msgs)) keptT := 0 for i := len(msgs) - 1; i >= 1; i-- { t := estimateTokens(msgs[i].Content) if sysT+keptT+t > maxT { break } keptT += t kept = append([]llm.Message{msgs[i]}, kept...) } // 最少保留最近 4 条,避免上下文过短 if len(kept) < 4 && len(msgs) > 4 { kept = msgs[len(msgs)-4:] } result := append([]llm.Message{msgs[0]}, kept...) return ensureToolMessageIntegrity(result) } // ensureToolMessageIntegrity 确保 tool 消息前面有对应 assistant(tool_calls) // Logan.Li func ensureToolMessageIntegrity(msgs []llm.Message) []llm.Message { validIDs := make(map[string]bool) for _, m := range msgs { if m.Role == "assistant" { for _, tc := range m.ToolCalls { validIDs[tc.ID] = true } } } out := make([]llm.Message, 0, len(msgs)) for i, m := range msgs { if m.Role == "tool" { if !validIDs[m.ToolCallID] || i == 0 || msgs[i-1].Role != "assistant" { continue } matched := false for _, tc := range msgs[i-1].ToolCalls { if tc.ID == m.ToolCallID { matched = true break } } if !matched { continue } } out = append(out, m) } return out } // estimateTokens 粗略 token 估算(中文约 1 字/token,英文约 1/4,折中 rune/3) // Logan.Li func estimateTokens(s string) int { return len([]rune(s))/3 + 1 } // Close 释放资源(SSH 长连接由 Engine.Close 统一释放) // Logan.Li func (a *Agent) Close() {}