/** * Agent Loop — ReAct 状态机引擎 * * 生产级 ReAct Agent Loop,负责: * 1. 状态机管理循环生命周期 * 2. 调用 Provider Adapter 获取 LLM 响应(支持流式) * 3. 解析输出、执行工具、注入观察 * 4. 流式事件推送 UI * 5. 超时、重试、上下文压缩 * * @see docs/生产级通用 AI Agent 智能体桌面应用:完整设计与构建指南.html — 第四章 */ import { EventEmitter } from 'events'; import { nanoid } from 'nanoid'; import { AgentLoopState, TerminationReason, type IterationStep, type Thought, type AgentLoopConfig, type AgentLoopOutput, type TokenUsage, } from './types'; import type { MetonaRequest, MetonaResponse, MetonaMessage, MetonaSystemPrompt, MetonaToolCall, MetonaToolResult, MetonaStreamEvent, IMetonaProviderAdapter, MetonaToolDef, } from '../types'; import { MetonaStreamEventType, MetonaFinishReason } from '../types'; import log from 'electron-log'; const DEFAULT_CONFIG: AgentLoopConfig = { maxIterations: 20, timeoutMs: 120_000, totalTimeoutMs: 600_000, enableReflection: false, compressionThreshold: 0.8, contextWindow: 128_000, retryCount: 3, temperature: 0.0, maxTokens: 63488, thinkingEnabled: true, thinkingEffort: 'high', }; /** * ReAct Agent Loop 引擎 */ export class AgentLoopEngine extends EventEmitter { private currentState: AgentLoopState = AgentLoopState.INIT; private iterations: IterationStep[] = []; private currentIteration = 0; private startTime = 0; private totalTokens: TokenUsage = { promptTokens: 0, completionTokens: 0, totalTokens: 0 }; private aborted = false; private config: AgentLoopConfig; private tools: MetonaToolDef[] = []; private currentSessionId: string = ''; private currentRequestId: string = ''; private workspacePath: string = ''; constructor( config: Partial = {}, private adapter: IMetonaProviderAdapter, private toolRegistry?: import('../tools/registry').ToolRegistry, private preToolHooks: import('../hooks/pre-tool').PreToolHook[] = [], private postToolHooks: import('../hooks/post-tool').PostToolHook[] = [], ) { super(); this.config = { ...DEFAULT_CONFIG, ...config }; } /** * 设置工作空间路径(工具执行时传入 context) */ setWorkspacePath(path: string): void { this.workspacePath = path; } /** * 设置可用工具列表 */ setTools(tools: MetonaToolDef[]): void { this.tools = tools; } /** * 获取当前工具列表(用于子任务恢复) */ getTools(): MetonaToolDef[] { return [...this.tools]; } /** * 热切换 Provider Adapter(设置变更时调用) */ setAdapter(adapter: IMetonaProviderAdapter): void { this.adapter = adapter; } /** * 热更新 Engine 配置(设置变更时调用) * * 支持 maxIterations, totalTimeoutMs, thinkingEnabled, thinkingEffort, contextLength */ updateConfig(partial: Partial): void { this.config = { ...this.config, ...partial }; } /** * 执行完整的 ReAct 循环(流式模式) * * @param userInput 用户输入文本 * @param sessionId 会话 ID * @param history 历史消息 * @param systemPrompt System Prompt */ async runStream( userMessage: MetonaMessage, sessionId: string, history: MetonaMessage[], systemPrompt: MetonaSystemPrompt, ): Promise { this.startTime = Date.now(); this.aborted = false; this.iterations = []; this.currentIteration = 0; this.currentSessionId = sessionId; this.totalTokens = { promptTokens: 0, completionTokens: 0, totalTokens: 0 }; try { await this.transitionTo(AgentLoopState.INIT); // 构建消息列表(保留 images 字段) const messages: MetonaMessage[] = [ ...history, { role: 'user', content: userMessage.content, images: userMessage.images, timestamp: Date.now(), }, ]; // === 主循环 === while ( this.currentIteration < this.config.maxIterations && !this.aborted && Date.now() - this.startTime < this.config.totalTimeoutMs ) { this.currentIteration++; // 构建请求 const request: MetonaRequest = { meta: { sessionId, iteration: this.currentIteration, requestId: `r_${nanoid(12)}`, timestamp: Date.now(), agentVersion: '1.0.0', }, systemPrompt, messages, tools: this.tools.length > 0 ? this.tools : undefined, params: { maxTokens: this.config.maxTokens, temperature: this.config.temperature, stream: true, thinkingEnabled: this.config.thinkingEnabled, thinkingEffort: this.config.thinkingEffort, contextLength: this.config.contextLength, }, }; const step = await this.executeOneIterationStream(request, sessionId); this.iterations.push(step); // 将 assistant 回复加入消息历史 if (step.thought) { const assistantMsg: MetonaMessage = { role: 'assistant', content: step.thought.content, reasoningContent: step.thought.reasoningContent, toolCalls: step.toolCalls, timestamp: Date.now(), iteration: this.currentIteration, }; messages.push(assistantMsg); } // 如果没有工具调用,视为最终输出 if (!step.toolCalls || step.toolCalls.length === 0) { return this.finish(TerminationReason.COMPLETED, step.thought?.content); } // 执行工具并将结果加入消息 if (step.toolResults) { for (const result of step.toolResults) { messages.push({ role: 'tool', content: typeof result.result === 'string' ? result.result : JSON.stringify(result.result), toolResult: result, timestamp: Date.now(), iteration: this.currentIteration, }); } } } // 循环退出判断 if (this.aborted) return this.finish(TerminationReason.USER_INTERRUPT); if (this.currentIteration >= this.config.maxIterations) return this.finish(TerminationReason.MAX_ITERATIONS); return this.finish(TerminationReason.TIMEOUT); } catch (error) { this.emit('error', { error: (error as Error).message }); return this.finish(TerminationReason.ERROR, undefined, error as Error); } } /** 获取当前 Provider Adapter(供 SubAgent 创建独立引擎实例) */ getAdapter(): IMetonaProviderAdapter { return this.adapter; } /** 获取工作空间路径(供 SubAgent 继承) */ getWorkspacePath(): string { return this.workspacePath; } /** 中断循环 */ abort(): void { this.aborted = true; this.emit('aborted'); } /** 获取当前状态 */ getState(): AgentLoopState { return this.currentState; } // ========== 私有方法 ========== /** * 执行单次迭代(流式模式) */ private async executeOneIterationStream( request: MetonaRequest, sessionId: string, ): Promise { this.currentRequestId = request.meta.requestId; const step: IterationStep = { iteration: this.currentIteration, state: AgentLoopState.THINKING, startedAt: Date.now(), }; try { // === THINKING: 流式调用 LLM === // transitionTo 已发射 stateChange 事件,无需重复 emit await this.transitionTo(AgentLoopState.THINKING); let fullContent = ''; let reasoningContent = ''; const toolCallsBuffer = new Map(); let tokenUsage: TokenUsage | undefined; // 流式接收响应 for await (const event of this.chatStreamWithRetry(request)) { if (this.aborted) break; // 过滤掉每轮的 DONE 事件 — 只在全部迭代完成后发送一个最终 DONE if (event.type === MetonaStreamEventType.DONE) continue; // 转发流式事件到渲染进程 this.emit('streamEvent', event); switch (event.type) { case MetonaStreamEventType.TEXT_DELTA: if (event.delta) fullContent += event.delta; break; case MetonaStreamEventType.REASONING_DELTA: if (event.delta) reasoningContent += event.delta; break; case MetonaStreamEventType.TOOL_CALL_DELTA: if (event.toolCallDelta) { const { index, name, argsDelta } = event.toolCallDelta; if (!toolCallsBuffer.has(index)) { toolCallsBuffer.set(index, { name: name ?? '', argsBuffer: '' }); } const buf = toolCallsBuffer.get(index)!; if (name) buf.name = name; if (argsDelta) buf.argsBuffer += argsDelta; } break; case MetonaStreamEventType.TOOL_CALL_COMPLETE: if (event.toolCall) { // 工具调用完成,记录到 step if (!step.toolCalls) step.toolCalls = []; step.toolCalls.push(event.toolCall); } break; case MetonaStreamEventType.USAGE: if (event.usage) { tokenUsage = { promptTokens: event.usage.inputTokens ?? 0, completionTokens: event.usage.outputTokens ?? 0, totalTokens: event.usage.totalTokens ?? 0, }; } break; case MetonaStreamEventType.ERROR: if (event.error) { throw new Error(event.error.message); } break; } } // 处理缓冲区中的工具调用(TOOL_CALL_DELTA 拼接) if (toolCallsBuffer.size > 0 && (!step.toolCalls || step.toolCalls.length === 0)) { step.toolCalls = []; for (const [, buf] of toolCallsBuffer) { let args: Record; try { args = buf.argsBuffer ? JSON.parse(buf.argsBuffer) : {}; } catch { // JSON 解析失败,使用空参数(LLM 仍请求了该工具调用) args = {}; } step.toolCalls.push({ id: `tc_${nanoid(8)}`, name: buf.name, args, iteration: this.currentIteration, timestamp: Date.now(), }); } } // 记录 Thought if (fullContent || reasoningContent) { step.thought = { id: `thought-${this.currentIteration}`, content: fullContent, reasoningContent, timestamp: Date.now(), iteration: this.currentIteration, }; } // 记录 Token 使用 if (tokenUsage) { step.tokenUsage = tokenUsage; this.accumulateTokens(tokenUsage); } // === EXECUTING: 执行工具调用 === if (step.toolCalls && step.toolCalls.length > 0) { // transitionTo 已发射 stateChange 事件,无需重复 emit await this.transitionTo(AgentLoopState.EXECUTING); // 并行执行所有工具调用(独立工具之间无依赖,可安全并发) const executeAndForward = async (tc: MetonaToolCall): Promise => { const result = await this.executeToolSafely(tc); // 立即转发工具结果到渲染进程(不等其他工具完成) this.emit('streamEvent', { type: MetonaStreamEventType.TOOL_RESULT, requestId: request.meta.requestId, sessionId, iteration: this.currentIteration, seq: 0, timestamp: Date.now(), toolResult: result, }); return result; }; // Promise.all 保持结果顺序与 toolCalls 一致 step.toolResults = await Promise.all( step.toolCalls.map((tc) => executeAndForward(tc)), ); } // === OBSERVING === await this.transitionTo(AgentLoopState.OBSERVING); // === 上下文压缩(基于 token 使用率触发) === // 有效上下文窗口:Ollama 使用 contextLength (numCtx),其他 Provider 使用 contextWindow const effectiveContextWindow = this.config.contextLength ?? this.config.contextWindow; const estimatedTokens = this.estimateMessagesTokens(request.messages); const compressionThreshold = this.config.compressionThreshold * effectiveContextWindow; if (estimatedTokens > compressionThreshold && request.messages.length > 10) { await this.transitionTo(AgentLoopState.COMPRESSING); const compressed = await this.compressMessages(request.messages); if (compressed) { // 原地替换数组内容,确保外层 messages 引用同步更新 request.messages.splice(0, request.messages.length, ...compressed); this.emit('compressed', { iteration: this.currentIteration, originalTokens: estimatedTokens, compressedTokens: this.estimateMessagesTokens(compressed), }); } } step.completedAt = Date.now(); step.state = AgentLoopState.OBSERVING; return step; } catch (error) { step.completedAt = Date.now(); step.state = AgentLoopState.TERMINATED; throw error; } } /** * 安全执行工具调用(经过 Hook 管道) */ private async executeToolSafely(toolCall: MetonaToolCall): Promise { const startTs = Date.now(); if (!this.toolRegistry) { return { toolCallId: toolCall.id, toolName: toolCall.name, result: null, success: false, error: `Tool '${toolCall.name}' not available: no ToolRegistry configured`, durationMs: Date.now() - startTs, timestamp: Date.now(), }; } // 前置 Hook 管道 for (const hook of this.preToolHooks) { const result = await hook.beforeExecute(toolCall, this.currentSessionId); if (result.blocked) { return { toolCallId: toolCall.id, toolName: toolCall.name, result: null, success: false, error: `Blocked: ${result.reason}`, durationMs: Date.now() - startTs, timestamp: Date.now(), }; } } // 执行工具 const toolResult = await this.toolRegistry.execute(toolCall, { sessionId: this.currentSessionId ?? '', workspacePath: this.workspacePath, iteration: this.currentIteration, requestId: this.currentRequestId, }); // 后置 Hook 管道 for (const hook of this.postToolHooks) { await hook.afterExecute(toolCall, toolResult, this.currentSessionId); } return toolResult; } /** * 带重试的流式调用 * * 如果 adapter 抛出错误,在 retryCount 次数内重试,每次等待 1 秒。 */ private async *chatStreamWithRetry(request: MetonaRequest): AsyncIterable { let lastError: unknown; for (let attempt = 0; attempt <= this.config.retryCount; attempt++) { try { yield* this.adapter.chatStream(request); return; } catch (error) { lastError = error; if (attempt < this.config.retryCount) { await new Promise((resolve) => setTimeout(resolve, 1000)); } } } throw lastError; } private async transitionTo(state: AgentLoopState): Promise { const previous = this.currentState; this.currentState = state; this.emit('stateChange', { previous, current: state, state, sessionId: this.currentSessionId, iteration: this.currentIteration, }); } /** * 估算消息列表的 token 数(粗略:1 字符 ≈ 0.5 token) */ private estimateMessagesTokens(messages: MetonaMessage[]): number { let total = 0; for (const msg of messages) { total += Math.ceil(msg.content.length / 2); if (msg.reasoningContent) total += Math.ceil(msg.reasoningContent.length / 2); if (msg.toolCalls) { for (const tc of msg.toolCalls) { total += Math.ceil(JSON.stringify(tc.args).length / 2); } } } return total; } /** * 上下文压缩 — 将旧消息摘要为一条 system 消息,保留最近 3 轮完整对话 * * 策略: * 1. 保留最后 keepRecent 条消息(约 3 轮对话) * 2. 将前面的所有消息交给 LLM 生成摘要 * 3. 用 [Context Summary] system 消息 + 近期消息替换原数组 * * @returns 压缩后的消息数组,压缩失败时返回 null(调用方保持原数组) */ private async compressMessages(messages: MetonaMessage[]): Promise { const keepRecent = 10; // 保留最近 10 条消息(约 3 轮 user+assistant+tool) if (messages.length <= keepRecent) return null; const toCompress = messages.slice(0, messages.length - keepRecent); const toKeep = messages.slice(messages.length - keepRecent); // 构建摘要请求 // 不截断单条消息——摘要请求是独立 API 调用,不共享主对话上下文窗口 const conversationText = toCompress.map((m) => { const role = m.role.toUpperCase(); return `[${role}] ${m.content}`; }).join('\n\n'); const summaryRequest: MetonaRequest = { meta: { sessionId: this.currentSessionId, iteration: this.currentIteration, requestId: `r_${nanoid(12)}`, timestamp: Date.now(), agentVersion: '1.0.0', }, systemPrompt: { roleDefinition: 'You are a conversation summarizer.', outputConstraints: 'Summarize the following conversation history concisely. Preserve key facts, decisions, tool results, and context needed for future reasoning. Output in the same language as the conversation. Maximum 300 words.', safetyGuidelines: 'Do not include sensitive data like passwords or API keys in the summary.', }, messages: [{ role: 'user', content: `Please summarize the following conversation history:\n\n${conversationText}`, timestamp: Date.now(), }], params: { maxTokens: 2048, temperature: 0.0, stream: false, thinkingEnabled: false, thinkingEffort: 'low', }, }; try { const response = await this.adapter.chat(summaryRequest); const summary = response.content.trim(); if (!summary) return null; const summaryMessage: MetonaMessage = { role: 'system', content: `[Context Summary] The following is a summary of earlier conversation:\n\n${summary}`, timestamp: Date.now(), }; log.info(`[AgentLoop] Context compressed: ${toCompress.length} messages → 1 summary, kept ${toKeep.length} recent`); return [summaryMessage, ...toKeep]; } catch (error) { log.warn('[AgentLoop] Context compression failed, keeping original messages:', (error as Error).message); return null; } } private accumulateTokens(usage: TokenUsage): void { this.totalTokens.promptTokens += usage.promptTokens; this.totalTokens.completionTokens += usage.completionTokens; this.totalTokens.totalTokens += usage.totalTokens; } private finish( reason: TerminationReason, answer?: string, error?: Error, ): AgentLoopOutput { // 发送唯一的最终 DONE 事件 — UI 只在此处结束流式状态 this.emit('streamEvent', { type: MetonaStreamEventType.DONE, requestId: this.currentRequestId, sessionId: this.currentSessionId, iteration: this.currentIteration, seq: 0, timestamp: Date.now(), }); this.currentState = AgentLoopState.TERMINATED; // 发射 complete 事件(供托盘通知等外部监听器使用) this.emit('complete', { sessionId: this.currentSessionId, durationMs: Date.now() - this.startTime, terminationReason: reason, iterations: this.iterations.length, totalTokens: this.totalTokens.totalTokens, }); return { finalAnswer: answer ?? (error ? error.message : 'No answer produced'), terminationReason: reason, iterations: this.iterations, totalTokenUsage: this.totalTokens, durationMs: Date.now() - this.startTime, metadata: { config: this.config, error: error?.message }, }; } }