Function Calling 并发调用编排多个工具并行执行时的依赖管理一、Agent 串行调用 5 个工具花了 12 秒其中 4 个工具互不依赖用户问对比三家酒店的评分、距离和价格。Agent 依次调用搜索酒店 A → 搜索酒店 B → 搜索酒店 C → 查询评分 → 查询价格。每次调用耗时约 2 秒总计 10 秒。但实际上三个酒店的搜索是独立的。评分和价格查询也互不依赖。如果并行执行总时间降到 4 秒。用户体验从慢变成流畅。问题在于Agent 默认不知道哪些调用可以并行。需要编排层分析工具调用的依赖关系。互不依赖的调用分到同一批并行执行。有依赖的调用保持串行顺序。这是 Agent 性能优化中最直接见效的手段之一。很多人花精力优化 Prompt、切换更快的模型却忽略了编排层面的延迟优化。在 Function Calling 场景中Agent 的输出本身就是一个 DAG有向无环图——LLM 返回的工具调用之间存在天然的并行机会。从端到端延迟的角度看串行执行的公式是T Σ(call_i)而并行执行的公式是T Σ(max(batch_j))。当工具调用之间 60% 没有依赖时并行化的理论加速比可以达到 2-3 倍。但在实践中依赖关系不是 LLM 自动输出的——它需要编排层来推断或由 Prompt 显式引导 LLM 标注。我在一个酒店比价 Agent 中做过实测串行执行 6 个工具调用3 个搜索 3 个详情查询平均耗时 8.7 秒加入依赖分析和并行编排后降至 3.1 秒用户感知的延迟减少了 65%。而且这个优化对 Prompt 和模型无关——只是编排层面做了改造。二、工具调用的依赖图与并行编排将 LLM 返回的多个 tool_call 构建为 DAG。分析每个调用的输入输出依赖。独立节点并行执行依赖节点等待上游。flowchart TB A[LLM 返回 3 个 tool_call] -- B[依赖分析] B -- C{依赖关系} C -- D[批次1: 搜索酒店A、B、Cbr/三者互不依赖并行执行] C -- E[批次2: 查询评分Abr/依赖搜索A的结果] C -- F[批次2: 查询评分Bbr/依赖搜索B的结果] C -- G[批次2: 查询评分Cbr/依赖搜索C的结果] D -- E D -- F D -- G E -- H[批次3: 对比汇总br/依赖所有评分结果] F -- H G -- H依赖关系的来源有两种方式。方式一让 LLM 在返回 tool_call 时同时标注依赖关系——在 Prompt 中要求标注depends_on字段。方式二编排层自动推断——分析每个调用的输入参数是否引用了其他调用的输出。方式二的准确性更高但实现复杂方式一更简单但依赖 LLM 的理解能力。推荐的做法是Prompt 标注为主 代码校验为辅。让 LLM 标注依赖然后编排层做基本合理性检查是否存在循环依赖、依赖的 ID 是否存在。这样既利用了 LLM 的语义理解能力又用代码兜底防止明显的错误标注。三、并行编排的 Go 实现下面的实现通过拓扑排序将工具调用分组为多个批次同批次内并行执行批次间串行等待。package orchestration import ( context fmt sync time ) // ToolCall 工具调用定义 type ToolCall struct { ID string json:id Name string json:name Params map[string]interface{} json:params Result interface{} json:result,omitempty DependsOn []string json:depends_on // 依赖的调用 ID Status string json:status // pending/running/done/failed } // Executor 工具执行器接口 type Executor interface { Execute(ctx context.Context, call ToolCall) (interface{}, error) } // BatchOrchestrator 批量编排器 type BatchOrchestrator struct { executor Executor maxParallel int timeout time.Duration } // NewBatchOrchestrator 创建编排器 func NewBatchOrchestrator( executor Executor, maxParallel int, timeout time.Duration, ) *BatchOrchestrator { if maxParallel 0 { maxParallel 5 } return BatchOrchestrator{ executor: executor, maxParallel: maxParallel, timeout: timeout, } } // AnalyzeDependencies 分析工具调用的依赖关系 func (o *BatchOrchestrator) AnalyzeDependencies( calls []ToolCall, ) ([][]ToolCall, error) { idMap : make(map[string]*ToolCall) for i : range calls { idMap[calls[i].ID] calls[i] } // 验证依赖关系 for _, call : range calls { for _, depID : range call.DependsOn { if _, ok : idMap[depID]; !ok { return nil, fmt.Errorf( 调用 %s 依赖了不存在的调用 %s, call.ID, depID, ) } } } // 拓扑排序分批次 remaining : make(map[string]bool) for _, call : range calls { remaining[call.ID] true } var batches [][]ToolCall for len(remaining) 0 { batch : make([]ToolCall, 0) // 找出所有依赖已满足的调用 for id : range remaining { call : idMap[id] depsSatisfied : true for _, depID : range call.DependsOn { if remaining[depID] { depsSatisfied false break } } if depsSatisfied { batch append(batch, *call) } } if len(batch) 0 { return nil, fmt.Errorf(检测到循环依赖) } batches append(batches, batch) for _, call : range batch { delete(remaining, call.ID) } } return batches, nil } // ExecuteParallel 并行执行编排好的批次 func (o *BatchOrchestrator) ExecuteParallel( ctx context.Context, batches [][]ToolCall, results map[string]interface{}, ) error { for batchIdx, batch : range batches { // 限制并行度 sem : make(chan struct{}, o.maxParallel) var wg sync.WaitGroup var mu sync.Mutex batchErr : make(chan error, len(batch)) for _, call : range batch { wg.Add(1) go func(c ToolCall) { defer wg.Done() // 获取信号量 select { case sem - struct{}{}: defer func() { -sem }() case -ctx.Done(): batchErr - ctx.Err() return } // 注入上游结果到参数 enrichedCall : c if c.DependsOn ! nil { mu.Lock() enrichedCall.Params o.injectDeps(c, results) mu.Unlock() } // 超时执行 execCtx, cancel : context.WithTimeout( ctx, o.timeout, ) defer cancel() result, err : o.executor.Execute(execCtx, enrichedCall) mu.Lock() if err ! nil { c.Status failed results[c.ID_error] err.Error() } else { c.Status done results[c.ID] result } mu.Unlock() if err ! nil { batchErr - fmt.Errorf( 批次 %d 调用 %s(%s) 失败: %w, batchIdx, c.Name, c.ID, err, ) } }(call) } // 等待当前批次完成 wg.Wait() close(batchErr) // 检查批次执行结果 for err : range batchErr { if err ! nil { return err } } } return nil } // injectDeps 将上游结果注入到当前调用的参数中 func (o *BatchOrchestrator) injectDeps( call ToolCall, results map[string]interface{}, ) map[string]interface{} { params : make(map[string]interface{}) for k, v : range call.Params { params[k] v } for _, depID : range call.DependsOn { if result, ok : results[depID]; ok { // 使用约定的参数名注入依赖结果 params[_dep_depID] result } } return params } // ---- 使用示例 ---- func demo() { orchestrator : NewBatchOrchestrator( MockExecutor{}, 3, 5*time.Second, ) calls : []ToolCall{ {ID: search_a, Name: search_hotel, Params: map[string]interface{}{name: A酒店}}, {ID: search_b, Name: search_hotel, Params: map[string]interface{}{name: B酒店}}, {ID: search_c, Name: search_hotel, Params: map[string]interface{}{name: C酒店}}, { ID: rating_a, Name: get_rating, Params: map[string]interface{}{hotel_id: $search_a.id}, DependsOn: []string{search_a}, }, { ID: rating_b, Name: get_rating, Params: map[string]interface{}{hotel_id: $search_b.id}, DependsOn: []string{search_b}, }, { ID: compare, Name: compare_hotels, Params: map[string]interface{}{}, DependsOn: []string{rating_a, rating_b}, }, } batches, err : orchestrator.AnalyzeDependencies(calls) if err ! nil { fmt.Printf(依赖分析失败: %v\n, err) return } fmt.Println(编排计划:) for i, batch : range batches { fmt.Printf(批次 %d:, i1) for _, c : range batch { fmt.Printf( %s(%s), c.Name, c.ID) } fmt.Println() } results : make(map[string]interface{}) if err : orchestrator.ExecuteParallel( context.Background(), batches, results, ); err ! nil { fmt.Printf(执行失败: %v\n, err) } }AnalyzeDependencies的循环依赖检测是一个重要的安全防线。如果 LLM 错误标注了循环依赖A 依赖 BB 依赖 A拓扑排序会陷入死循环。代码层面的检测保证了即使 LLM 犯错编排层也不会卡死。四、并行编排的风险并行调用加倍了 Token 速率消耗。如果 API 有速率限制并行可能触发限流。需要配合令牌桶限流器。依赖注入的命名约定需要统一。_dep_xxx这样的参数名侵入业务代码。更好的方式是定义接口适配层。不适用场景工具调用之间有隐性依赖同一数据库表锁调用成本极高需要精确控制的场景调试阶段的快速原型验证。除了速率限制并发编排还有一个更隐蔽的风险下游服务的承载能力。当 5 个工具调用并行发往同一个服务时瞬时 QPS 是 5 倍。如果下游服务没有做好相应的容量规划并行编排可能导致雪崩。建议在引入并行编排前先评估每个工具的下游调用频率和限流配置。另一个工程建议并行编排应该有降级开关。在生产环境部署时保留一个配置项允许将并行执行退化为串行执行。当并行导致的问题如下游限流、资源竞争超过带来的延迟收益时运维可以一键关闭并行模式而无需重新发版。五、总结工具调用并行编排的核心是依赖分析 分批执行。通过拓扑排序将调用分入不同批次。同批次无依赖的调用可安全并行执行。信号量控制并行度防止资源过载。并行编排可将多工具调用的总延迟降低 50-70%。落地建议先不做自动依赖推断从 Prompt 侧入手。在 System Prompt 中明确要求 LLM 标注每个 tool_call 的depends_on字段然后在代码层做拓扑排序。等这个模式跑稳了再考虑自动依赖推断作为补充。一步到位做全自动依赖分析往往因边界情况太多而难以维护。
