// 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 思考/推理文本 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) // Config Agent 运行配置 // Logan.Li type Config struct { MaxIterations int Temperature float64 ToolTimeout time.Duration MaxTokens int SafetyMode safety.Mode OutputMaxLength int MaxContextTokens int MemoryDir string // Evolver 记忆目录 } // 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 集成) // Logan.Li func (a *Agent) Run(ctx context.Context, query, serverInfo string) (*tools.Result, error) { a.totalTokens = 0 start := time.Now() // Evolver PREDICT:获取历史经验注入 system prompt evolveCtx := "" if a.evolver != nil { evolveCtx = a.evolver.GetContextForPrompt(query) } // 工具描述 toolDescs := a.buildToolDescriptions() systemPrompt := prompt.BuildSystemPrompt(serverInfo, toolDescs, evolveCtx) messages := []llm.Message{ {Role: "system", Content: systemPrompt}, {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: iter + 1}) // 自适应检查点 if hint := a.checkAdaptiveCheckpoint(iter, toolCallHistory, consecutiveFailures, totalToolCalls); hint != "" { messages = append(messages, llm.Message{Role: "user", Content: hint}) } // 裁剪历史 trimmed := a.trimMessages(messages) // 调用 LLM 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}) // 推送思考文本 if strings.TrimSpace(resp.Message.Content) != "" { a.emit(Event{Type: EventThought, Round: iter + 1, Thought: resp.Message.Content}) } messages = append(messages, resp.Message) // 无工具调用 → 最终答案 if len(resp.Message.ToolCalls) == 0 { answer := resp.Message.Content a.emit(Event{Type: EventDone, Answer: answer, Tokens: a.totalTokens}) a.triggerEvolve(ctx, query, serverInfo, answer, toolRecords, true, start) return &tools.Result{Success: true, Output: answer}, nil } // 处理工具调用 observations, records, failInc := a.processToolCalls(ctx, resp.Message.ToolCalls, toolCallHistory, iter+1) 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 nil, 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 } // 输出截断 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) }() } // 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 释放资源 // Logan.Li func (a *Agent) Close() { if a.registry != nil { // Registry 无 Closer 实现(Web 工具无连接池),保留接口位 } }