/** * 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'; 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, }; /** * 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]; } /** * 执行完整的 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: 63488, temperature: this.config.temperature, stream: true, thinkingEnabled: true, thinkingEffort: 'high', 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); } } /** 中断循环 */ 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 === await this.transitionTo(AgentLoopState.THINKING); this.emit('stateChange', { sessionId, iteration: this.currentIteration, state: 'THINKING', }); let fullContent = ''; let reasoningContent = ''; const toolCallsBuffer = new Map(); let tokenUsage: TokenUsage | undefined; // 流式接收响应 for await (const event of this.adapter.chatStream(request)) { if (this.aborted) break; // 转发流式事件到渲染进程 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.DONE: 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 [index, buf] of toolCallsBuffer) { try { step.toolCalls.push({ id: `tc_${nanoid(8)}`, name: buf.name, args: buf.argsBuffer ? JSON.parse(buf.argsBuffer) : {}, iteration: this.currentIteration, timestamp: Date.now(), }); } catch { // JSON 解析失败,跳过 } } } // 记录 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) { await this.transitionTo(AgentLoopState.EXECUTING); this.emit('stateChange', { sessionId, iteration: this.currentIteration, state: 'EXECUTING', }); step.toolResults = []; for (const tc of step.toolCalls) { const result = await this.executeToolSafely(tc); step.toolResults.push(result); // 转发工具结果到渲染进程 this.emit('streamEvent', { type: 'tool_result', requestId: request.meta.requestId, sessionId, iteration: this.currentIteration, seq: 0, timestamp: Date.now(), toolResult: result, }); } } // === OBSERVING === await this.transitionTo(AgentLoopState.OBSERVING); // === 上下文压缩(每 5 轮检查一次) === if (this.currentIteration % 5 === 0) { await this.transitionTo(AgentLoopState.COMPRESSING); this.emit('stateChange', { sessionId, iteration: this.currentIteration, state: 'COMPRESSING', }); // 在 context 中标记压缩点 const keepRecent = 3; if (this.iterations.length > keepRecent) { this.emit('compressed', { iteration: this.currentIteration, keptRounds: keepRecent }); } } 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; } private async transitionTo(state: AgentLoopState): Promise { const previous = this.currentState; this.currentState = state; this.emit('stateChange', { previous, current: state }); } 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 { this.currentState = AgentLoopState.TERMINATED; 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 }, }; } }