concurrency

并发控制确保系统在高负载下稳定运行，避免资源耗尽。

并发限制

工具执行并发

const MAX_TOOL_CONCURRENCY = 5;

class ConcurrencyLimiter {
  private running = 0;
  private queue: Array<() => void> = [];
  
  async acquire(): Promise<void> {
    if (this.running < MAX_TOOL_CONCURRENCY) {
      this.running++;
      return;
    }
    
    // 等待空闲
    return new Promise(resolve => {
      this.queue.push(resolve);
    });
  }
  
  release(): void {
    this.running--;
    
    if (this.queue.length > 0) {
      const next = this.queue.shift();
      this.running++;
      next();
    }
  }
}

Agent 并发

const MAX_AGENT_CONCURRENCY = 3;

async function executeAgents(
  tasks: AgentTask[]
): Promise<AgentResult[]> {
  const results = [];
  
  for (let i = 0; i < tasks.length; i += MAX_AGENT_CONCURRENCY) {
    const batch = tasks.slice(i, i + MAX_AGENT_CONCURRENCY);
    const batchResults = await Promise.all(
      batch.map(task => executeAgent(task))
    );
    results.push(...batchResults);
  }
  
  return results;
}

资源管理

内存限制

const MAX_MEMORY_MB = 512;

function checkMemoryUsage(): void {
  const usage = process.memoryUsage();
  const heapMB = usage.heapUsed / 1024 / 1024;
  
  if (heapMB > MAX_MEMORY_MB) {
    // 触发垃圾回收
    if (global.gc) {
      global.gc();
    }
    
    // 清理缓存
    clearCaches();
  }
}

文件句柄限制

const MAX_OPEN_FILES = 100;
const openFiles = new Set<string>();

async function readFileWithLimit(path: string): Promise<string> {
  if (openFiles.size >= MAX_OPEN_FILES) {
    // 等待其他文件关闭
    await waitForFileSlot();
  }
  
  openFiles.add(path);
  
  try {
    return await fs.readFile(path, 'utf-8');
  } finally {
    openFiles.delete(path);
  }
}

优先级调度

工具优先级

enum ToolPriority {
  HIGH = 3,    // 用户直接请求的工具
  NORMAL = 2,  // 常规工具
  LOW = 1,     // 后台任务
}

class PriorityQueue {
  private queues = {
    [ToolPriority.HIGH]: [],
    [ToolPriority.NORMAL]: [],
    [ToolPriority.LOW]: [],
  };
  
  enqueue(call: ToolCall, priority: ToolPriority): void {
    this.queues[priority].push(call);
  }
  
  dequeue(): ToolCall | null {
    // 优先执行高优先级任务
    for (const priority of [ToolPriority.HIGH, ToolPriority.NORMAL, ToolPriority.LOW]) {
      if (this.queues[priority].length > 0) {
        return this.queues[priority].shift();
      }
    }
    return null;
  }
}

背压处理

队列限制

const MAX_QUEUE_SIZE = 20;

async function enqueueToolCall(call: ToolCall): Promise<void> {
  if (queue.size >= MAX_QUEUE_SIZE) {
    // 队列已满，等待
    await waitForQueueSpace();
  }
  
  queue.push(call);
}

流量控制

class RateLimiter {
  private tokens: number;
  private lastRefill: number;
  
  constructor(
    private maxTokens: number,
    private refillRate: number // tokens per second
  ) {
    this.tokens = maxTokens;
    this.lastRefill = Date.now();
  }
  
  async acquire(cost: number = 1): Promise<void> {
    // 补充 tokens
    this.refill();
    
    // 等待足够的 tokens
    while (this.tokens < cost) {
      await sleep(100);
      this.refill();
    }
    
    this.tokens -= cost;
  }
  
  private refill(): void {
    const now = Date.now();
    const elapsed = (now - this.lastRefill) / 1000;
    const newTokens = elapsed * this.refillRate;
    
    this.tokens = Math.min(this.maxTokens, this.tokens + newTokens);
    this.lastRefill = now;
  }
}

超时控制

工具超时

async function executeWithTimeout(
  call: ToolCall,
  timeout: number
): Promise<ToolResult> {
  return Promise.race([
    executeTool(call),
    new Promise<ToolResult>((_, reject) =>
      setTimeout(() => reject(new Error('Timeout')), timeout)
    ),
  ]);
}

全局超时

const GLOBAL_TIMEOUT = 60000; // 60 秒

async function executeAllTools(
  calls: ToolCall[]
): Promise<ToolResult[]> {
  const timeoutPromise = new Promise<never>((_, reject) =>
    setTimeout(() => reject(new Error('Global timeout')), GLOBAL_TIMEOUT)
  );
  
  return Promise.race([
    Promise.all(calls.map(call => executeTool(call))),
    timeoutPromise,
  ]);
}

下一步

• 查看 缓存策略 的实现

• 了解 启动优化 的技术

• 探索 上下文管理 的压缩策略