Agent 工具调用的并发控制：限制并行度与去重调用的性能保障机制-平芜编程栈

Agent 工具调用的并发控制：限制并行度与去重调用的性能保障机制

一、一个文件搜索 Agent 的资源雪崩

一个代码库搜索 Agent，当用户请求"找到所有使用 deprecated API 的文件并列出替代方案"时，Agent 同时触发了 47 个文件读取工具调用——对应 47 个匹配的文件。这 47 个并发调用：

占满了 LLM API 的连接池（20 个连接上限，27 个排队等待）
耗尽了服务器的文件描述符（ulimit 4096，临时达到 3780）
10 个调用因为并发过高超时失败

Agent 在"调用失败 → 重试 → 更多并发"的恶性循环中耗尽了一轮对话的上下文窗口。最终返回的结果是部分文件的分析，缺失了 13 个文件。

问题不是 47 个调用本身，而是没有对并发度做控制和去重。如果 47 个调用中有 12 个访问了同一目录下的文件（如 node_modules），缓存和去重可以避免重复的文件读取。

二、工具调用的并发控制模型

2.1 带信号量的并发控制器

// 工具调用并发管理器：限制并行度 + 去重 type ToolCallManager struct { semaphore chan struct{} // 信号量控制并发度 cache *lru.Cache // 结果缓存 toolRegistry map[string]ToolHandler maxRetries int timeout time.Duration } func NewToolCallManager(maxConcurrency int, cacheSize int) *ToolCallManager { return &ToolCallManager{ semaphore: make(chan struct{}, maxConcurrency), cache: lru.New(cacheSize), toolRegistry: make(map[string]ToolHandler), maxRetries: 3, timeout: 30 * time.Second, } } func (m *ToolCallManager) ExecuteBatch( ctx context.Context, calls []ToolCall, ) ([]ToolResult, error) { // 第一步：去重——相同工具 + 相同参数的调用只执行一次 unique := m.deduplicate(calls) // 第二步：并行执行（受信号量控制） results := make([]ToolResult, len(calls)) errCh := make(chan error, len(calls)) var wg sync.WaitGroup for i, call := range calls { wg.Add(1) go func(idx int, c ToolCall) { defer wg.Done() // 检查是否与之前某个唯一调用相同，如果是则复用其结果 if uniqueResult, found := unique[c.cacheKey()]; found { results[idx] = *uniqueResult return } // 获取信号量 select { case m.semaphore <- struct{}{}: defer func() { <-m.semaphore }() case <-ctx.Done(): errCh <- ctx.Err() return } result, err := m.executeWithRetry(ctx, c) if err != nil { results[idx] = ToolResult{Error: err.Error()} return } results[idx] = *result }(i, call) } wg.Wait() close(errCh) // 检查是否有致命错误 if err, ok := <-errCh; ok { return nil, err } return results, nil } // 去重逻辑：相同工具的相同参数只执行一次 func (m *ToolCallManager) deduplicate(calls []ToolCall) map[string]*ToolResult { executed := make(map[string]*ToolResult) for i := range calls { key := calls[i].cacheKey() if _, exists := executed[key]; exists { continue // 已存在，跳过 } // 检查缓存 if cached, found := m.cache.Get(key); found { result := cached.(ToolResult) executed[key] = &result } } return executed }

2.2 智能去重策略

// 工具调用的去重策略 type DedupStrategy struct { // 精确匹配：tool_name + params 完全一致 exactDedup bool // 语义匹配：读取同一目录下的文件 → 批量读取 mergeReads bool } func (s *DedupStrategy) Optimize(calls []ToolCall) []ToolCall { if s.mergeReads { return s.mergeFileReads(calls) } return calls } // 合并同一目录下的文件读取 func (s *DedupStrategy) mergeFileReads(calls []ToolCall) []ToolCall { var dirReads = make(map[string][]string) // dirPath → [files] var otherCalls []ToolCall for _, call := range calls { if call.Name == "file_read" { dir := filepath.Dir(call.Params["path"]) // 如果目录相同且文件数 < 5，合并为一次批量读取 dirReads[dir] = append(dirReads[dir], call.Params["path"]) } else { otherCalls = append(otherCalls, call) } } var optimized []ToolCall for dir, files := range dirReads { if len(files) <= 5 { optimized = append(optimized, ToolCall{ Name: "file_read_batch", Params: map[string]string{"paths": strings.Join(files, ",")}, }) } else { // 文件太多，拆分或使用 shell 命令替代 optimized = append(optimized, ToolCall{ Name: "shell_exec", Params: map[string]string{ "command": fmt.Sprintf("cat %s/*", dir), }, }) } } return append(optimized, otherCalls...) }

去重的效果：47 个文件读取调用 → 合并为 8 个目录级批量读取。API 调用从 47 次降到 8 次，工具调用总延迟从 15 秒降到 3 秒。

三、自适应并发度的调整

// 根据历史响应时间自适应调整并发度 type AdaptiveConcurrency struct { current atomic.Int32 min, max int32 windowSize int32 // 滑动窗口大小 responseTimes []float64 mu sync.Mutex } func (a *AdaptiveConcurrency) Adjust(avgResponseTime float64, errorRate float64) { a.mu.Lock() defer a.mu.Unlock() current := a.current.Load() // 高错误率 → 降低并发度 if errorRate > 0.1 && current > a.min { a.current.Store(current / 2) log.Printf("并发度降低: %d → %d (错误率 %.2f%%)", current, current/2, errorRate*100) return } // 低错误率 + 低响应时间 → 提高并发度 if errorRate < 0.02 && avgResponseTime < 100*time.Millisecond && current < a.max { a.current.Store(current + 2) log.Printf("并发度提升: %d → %d (平均响应 %.0fms)", current, current+2, avgResponseTime*1000) } }