Agent 工具调用的并发控制限制并行度与去重调用的性能保障机制一、一个文件搜索 Agent 的资源雪崩一个代码库搜索 Agent当用户请求找到所有使用 deprecated API 的文件并列出替代方案时Agent 同时触发了 47 个文件读取工具调用——对应 47 个匹配的文件。这 47 个并发调用占满了 LLM API 的连接池20 个连接上限27 个排队等待耗尽了服务器的文件描述符ulimit 4096临时达到 378010 个调用因为并发过高超时失败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, current2, avgResponseTime*1000) } }四、边界与权衡过度去重的风险将读取 /src/utils.ts和读取 /src/utils.test.ts合并为读取 /src/ 目录——虽然减少了一次调用但读取了不需要的文件内容增加了上下文 Token 消耗。需要权衡减少 API 调用和增加上下文噪音。信号量饥饿如果长任务一直占用信号量短任务排队时间过长。引入优先级机制短任务预估耗时 1s获得更高的执行优先级使用分离的信号量通道。去重缓存的失效文件可能在两次读取之间被修改。对于实时性要求高的场景如监控禁用缓存对于分析类场景如代码扫描启用 30 秒的短时缓存。五、总结Agent 工具调用的并发控制核心是限制并行度和去重合并。信号量控制并发上限建议 CPU 核心数 × 4避免调用风暴打垮基础设施。去重合并减少不必要的重复调用同目录文件合并读取、相同参数结果缓存。实施顺序先加信号量控制10 行代码效果立竿见影→ 再实现精确匹配去重缓存工具调用结果→ 最后做语义合并目录级文件读取。先跑最少血的方案验证有效后再叠加复杂优化。