Files
metona-ai-desktop/electron/harness/agent-loop/engine.ts
T
thzxx 6f08759f63 v0.1.2: 全项目代码审查修复 + 子代理编排器 + 上下文压缩 + 并行工具执行
后端修复: Ollama adapter 移除死代码/修复超时硬编码/iteration硬编码/pullModel无超时/流异常断开补发DONE; SSE解析器支持非字符串arguments; Sandbox fail-closed安全加固; IPC移除未使用变量

新增功能: TaskOrchestrator子任务编排器; DelegateTaskTool委派工具; Agent Loop并行工具执行; 上下文自动压缩; 记忆检索注入System Prompt

前端修复: useAgentStream text_delta thought累积bug; 工具调用状态正确流转; 修复重复stateChange事件; TraceStep.thought正确填充
2026-07-06 22:48:33 +08:00

636 lines
20 KiB
TypeScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
/**
* 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<AgentLoopConfig> = {},
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<AgentLoopConfig>): 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<AgentLoopOutput> {
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<IterationStep> {
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<number, { name: string; argsBuffer: string }>();
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<string, unknown>;
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<MetonaToolResult> => {
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 使用率触发) ===
const estimatedTokens = this.estimateMessagesTokens(request.messages);
const compressionThreshold = this.config.compressionThreshold * this.config.contextWindow;
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<MetonaToolResult> {
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<MetonaStreamEvent> {
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<void> {
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<MetonaMessage[] | null> {
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);
// 构建摘要请求
const conversationText = toCompress.map((m) => {
const role = m.role.toUpperCase();
const content = m.content.slice(0, 500); // 每条消息最多 500 字符
return `[${role}] ${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 },
};
}
}