Agent Harness 实战：从零实现并行工具调用引擎
原创
Z
Z
威胁情报Z分析
2026年10月3日 08:11
广东
在小说阅读器读本章
去阅读
在公众号小说中沉浸阅读
| 本文目标：讲清楚一个能在生产用的 Agent 运行时（Harness）里，\"并行工具调用引擎\"到底由哪几块组成、为什么必须按 DAG 调度、关键代码怎么写。读完你可以直接照抄一个 200 行以内的 Python asyncio 最小引擎，并理解为什么 OpenAI / Anthropic 把 parallel tool calls 做成一等公民。 |
一、为什么必须做并行工具调用
绝大多数人第一次写 Agent，用的都是最朴素的 ReAct 循环：LLM 决定调一个工具 → 等它返回 → 把结果塞回上下文 → 再问 LLM。这个模型在工具少、时延低的时候没问题，但一旦工具数上来，端到端时延就变成所有工具延迟之和。
真实 Agent 一次用户请求往往要调多个工具：查天气、搜新闻、读文件、跑计算器、写库——其中大部分彼此没有数据依赖。串行调用等于让用户干等 N 次网络往返，而这些往返本可以在一个事件循环里同时飞出去。
并行工具调用引擎要解决的不是\"能不能并发\"这种操作系统问题，而是这三件更难的事：
1依赖识别：LLM 一次吐出 N 个工具调用，引擎怎么知道哪些能并行、哪些必须等前面的结果？
1调度执行：怎么在不打爆下游 API、不打爆自己事件循环的前提下，把无依赖的工具同时发出去？
1失败隔离：一个工具超时或炸了，不能拖垮整轮；要让 LLM 看到 error 而不是整个会话崩溃。
二、整体架构：五层各司其职
整个引擎在概念上可以拆成五层。规划层负责让 LLM 说清楚要调什么；调度层把这堆调用编译成可执行的依赖图；执行层真正并发跑工具；收敛层把结果整理回上下文，再决定是继续问 LLM 还是结束。
几个容易踩坑的点：
1Planner 不直接负责执行。LLM 输出的是一个\"计划\"（一组 tool\_call，每个带 depends\_on），Harness 负责把它编译成 DAG 而不是逐条 await。
1调度层必须有环检测。LLM 偶尔会输出\"A 依赖 B、B 又依赖 A\"的计划，这一步必须在执行前拒绝并回喂 LLM，不能带着环跑。
1执行层用信号量而非无脑 gather。一次吐 20 个工具、每个都打 HTTP，不加并发上限会直接把对方限流掉。
三、一次 Turn 的状态机
把视角收敛到\"一次用户请求\"内部，引擎其实是一个状态机：从用户输入出发，要么走到\"输出最终回答\"结束，要么进入\"工具执行 → 结果回灌 → 再问 LLM\"的循环，直到 LLM 不再要求调工具为止。
关键设计点是循环边界在 LLM 这一侧，不在工具层。引擎不知道\"什么时候算够了\"，它只知道\"LLM 这一轮还要不要继续调工具\"。这种边界划分让引擎本身保持很薄，所有策略（什么时候停、要不要重试）都能通过 prompt 或外层 system message 调，不用改调度代码。
四、DAG 调度核心：把工具列表变成可并行执行的图
调度层最核心的数据结构是一张有向无环图（DAG）。节点是工具调用，边是\"我必须等你出结果才能跑\"。拓扑排序之后，同一层的节点之间没有依赖，可以一次性并发飞出去；下一层节点必须等上一层全部 settle。
这套调度带来两个直接结论：
1端到端时延由关键路径决定，而不是工具总数。本例中工具总数 4 个、串行要 4s；DAG 调度后关键路径 3 层共 3s。工具再多，只要每层内可并行，时延就是\"层数 × 单层最大耗时\"。
1依赖关系是 LLM 写出来的，不是引擎猜的。这要求工具 schema 里要么让 LLM 显式输出 depends\_on，要么由 Harness 静态分析参数里是否引用了上游 tool\_call 的结果。OpenAI 风格是后者：模型在 args 里写 ${tool\_call\_3.result}，Harness 做字符串替换时顺带建边。
五、关键设计决策
| 决策点 | 做法 | 为什么 |
| 并发上限 | asyncio.Semaphore(N)，N 默认 5~10 | 防打爆下游、防自己 fd 耗尽；N 按下游限流配置调 |
| 单工具超时 | asyncio.wait\_for(tool(...), timeout=10) | 一个工具 hang 住不能拖死整轮；超时结果作为 error 喂回 LLM |
| 失败策略 | 默认不重试；可重试错误（429/5xx）指数退避 2 次；其他错误直接落 error\_result | LLM 看到错误信息后可以自己决定换个参数再调一次 |
| 错误传播 | 依赖该工具的下游节点自动收到 error\_dependency\_failed，不强行执行 | 避免拿半成品算半天，结果上游就错了 |
| 结果回灌格式 | 每个 tool\_call 对应一条 tool 角色消息，带 tool\_call\_id 关联 | 对齐 OpenAI / Anthropic 的协议，LLM 才能把结果对上当初的调用 |
| 最大轮次 | 硬性 cap（如 10 轮），超过直接截断并告诉 LLM | 防止 LLM 陷入\"调工具-看结果-再调同一个\"的死循环烧 token |
六、代码实现：一个 200 行的最小引擎
下面是完整可运行的 Python 实现，只用标准库 asyncio。它包含工具注册、DAG 构建、拓扑并行调度、超时与错误隔离，最后用一个多工具查询场景跑通。
6.1 工具注册与调用计划
| python tools.py —— 工具注册表与装饰器 import asyncio import time from dataclasses import dataclass, field from typing import Any, Callable @dataclass class ToolCall: id: str # 例如 \"call\_1\" name: str # 工具名 args: dict # 参数（已替换好上游引用） depends\_on: list[str] = field(default\_factory=list) @dataclass class ToolResult: call\_id: str name: str ok: bool value: Any = None # 成功时的返回值 error: str |
6.2 DAG 构建与分层调度
| python scheduler.py —— 拓扑分层 + 并发执行 async def run\_plan(registry: ToolRegistry, plan: list[ToolCall], max\_concurrency: int = 5) -> dict[str, ToolResult]: # 1) 建图：统计每个节点的入度 & 后继 by\_id = {c.id: c for c in plan} indeg = {c.id: len(c.depends\_on) for c in plan} successors: dict[str, list[str]] = {c.id: [] for c in plan} for c in plan: for dep in c.depends\_on: if dep not in by\_id: raise ValueError(f\"{c.id} 依赖不存在的 {dep}\") successors[dep].append(c.id) # 2) 环检测：拓扑排序后若仍有剩余节点，就是有环 order, ready = [], [cid for cid, d in indeg.items() if d == 0] while ready: cid = ready.pop() order.append(cid) for nxt in successors[cid]: indeg[nxt] -= 1 if indeg[nxt] == 0: ready.append(nxt) if len(order) != len(plan): raise ValueError(\"工具调用计划存在循环依赖\") # 3) 按层执行：每一层内部并发 sem = asyncio.Semaphore(max\_concurrency) results: dict[str, ToolResult] = {} done = set() while len(done) < len(plan): layer = [c for c in plan if c.id not in done and all(d in done for d in c.depends\_on)] if not layer: break # 兜底，正常已被环检测拦住 async def run\_one(call: ToolCall) -> ToolResult: async with sem: # 上游若失败，本节点不执行，直接标记依赖失败 failed\_upstream = next((results[d] for d in call.depends\_on if not results[d].ok), None) if failed\_upstream: return ToolResult(call.id, call.name, ok=False, error=f\"依赖 {failed\_upstream.call\_id} 失败\") t0 = time.perf\_counter() try: value = await registry.invoke(call.name, call.args) return ToolResult(call.id, call.name, ok=True, value=value) except Exception as e: return ToolResult(call.id, call.name, ok=False, error=str(e)) finally: print(f\" [{time.perf\_counter()-t0:5.2f}s] {call.name} done\") for r in await asyncio.gather(\*(run\_one(c) for c in layer)): results[r.call\_id] = r done.add(r.call\_id) return results |
6.3 端到端例子
| python demo.py —— 一次真实的多工具并发调用 import asyncio registry = ToolRegistry() @registry.register(\"search\web\", \"联网搜索\", {\"type\": \"object\", \"properties\": {\"query\": {\"type\": \"string\"}}}) async def search\_web(query: str) -> list[str]: await asyncio.sleep(1.0) # 模拟网络往返 return [f\"《{query}》的搜索结果 1\", f\"《{query}》的搜索结果 2\"] @registry.register(\"get\_weather\", \"查天气\", {\"type\": \"object\", \"properties\": {\"city\": {\"type\": \"string\"}}}) async def get\_weather(city: str) -> dict: await asyncio.sleep(1.0) return {\"city\": city, \"temp\_c\": 28, \"condition\": \"晴\"} @registry.register(\"calculator\", \"算数\", {\"type\": \"object\", \"properties\": {\"expr\": {\"type\": \"string\"}}}) async def calculator(expr: str) -> float: await asyncio.sleep(1.0) return eval(expr, {\"\\builtins\\_\": {}}) # 示例：生产用 ast.literal\_eval @registry.register(\"gen\_report\", \"写汇总\", {\"type\": \"object\", \"properties\": {\"text\": {\"type\": \"string\"}}}) async def gen\_report(text: str) -> str: await asyncio.sleep(1.0) return f\"[报告] {text}\" async def main(): # 这是 LLM 这一轮吐出来的计划：call\_1/call\_2 并行；call\_3 等 call\_1；call\_4 等 call\_2 & call\_3 plan = [ ToolCall(\"call\_1\", \"search\_web\", {\"query\": \"深圳天气\"}, []), ToolCall(\"call\_2\", \"get\_weather\", {\"city\": \"深圳\"}, []), ToolCall(\"call\_3\", \"calculator\", {\"expr\": \"28 \* 1.8 + 32\"}, [\"call\_1\"]), ToolCall(\"call\_4\", \"gen\_report\", {\"text\": \"深圳 28°C / 华氏度 82.4 / 搜索 2 条\"}, [\"call\_2\", \"call\_3\"]), ] t0 = time.perf\_counter() results = await run\_plan(registry, plan, max\_concurrency=5) print(f\"\ |
| 总耗时 {time.perf\_counter()-t0:.2f}s（串行预期 4.00s）\") for cid, r in results.items(): print(f\" {cid} {r.name}: ok={r.ok} value={r.value} err={r.error}\") asyncio.run(main()) |
预期输出：
| text 运行结果 [ 1.00s] search\_web done [ 1.00s] get\_weather done [ 1.00s] calculator done [ 1.00s] gen\_report done 总耗时 3.01s（串行预期 4.00s） call\_1 search\_web: ok=True value=['《深圳天气》的搜索结果 1', ...] call\_2 get\_weather: ok=True value={'city': '深圳', 'temp\_c': 28, ...} call\_3 calculator: ok=True value=82.4 call\_4 gen\_report: ok=True value='[报告] 深圳 28°C / ...' |
七、时延收益到底有多大
把上面这个例子画成对比图：4 个工具各 1s，串行要 4s，DAG 并行后 3s。收益不是固定比例，它取决于两件事——一层内能并行多少个工具、关键路径有多长。
在真实业务里，一次 Agent turn 往往要打 5~10 个工具，其中 70% 互相无依赖。这种场景下从串行切到 DAG 并行，端到端 P50 时延砍半是常见结果；瓶颈会从\"工具数量\"转移到\"最深那条依赖链\"。
八、上生产还要补的几件事
上面 200 行是骨架，真要上线还得补这几块，每一块都对应一个线上踩过的坑：
1观测与 trace：每个 tool\_call 记录开始时间、结束时间、入参摘要、返回字节数，挂到 OpenTelemetry 上。没有 trace 的并行调度就是黑盒，出了性能问题根本不知道慢在哪一层。
1token 预算：并行工具返回的内容会一次性灌回上下文，几个大响应能直接把 context window 撑爆。引擎层要做截断与摘要，不能无脑全量塞回去。
1幂等与副作用：并行意味着同一个工具可能在同一轮被并发触发两次（LLM 抽风）。写操作（发邮件、扣款、写库）必须由 Harness 加互斥锁或要求 LLM 显式去重。
1流式输出：工具没全跑完之前，最终答案还出不来；但中间层的工具结果可以先在前端展示\"正在查天气…正在搜新闻…\"，体感时延会进一步下降。
1结构化输出校验：LLM 吐出来的 plan 可能缺字段、依赖不存在、参数类型不对。调度器入口要做 schema 校验，失败就把校验错误作为 message 回喂 LLM 让它自己改，而不是抛异常终止会话。
九、小结
并行工具调用引擎的本质，是把 LLM 吐出的\"一组工具调用\"从\"一个 list 顺序 await\"升级成\"一张 DAG 拓扑执行\"。核心代码量并不大：DAG 构建 + 分层 + asyncio.gather + 信号量 + 超时隔离，200 行能跑通。真正拉开差距的是工程细节——trace、token 预算、幂等、错误回喂，以及把\"什么时候停、什么时候重试\"这些策略留在 LLM 这一侧，而不是写死在调度器里。
把这套引擎跑通之后，你再看 OpenAI 的 parallel\_tool\_calls、Anthropic 的 tool use 多轮循环，就会发现它们做的事和上面这套骨架是同构的，只是加了更厚的协议、重试与观测层。
版权声明
本站仅做备份收录，仅供研究与教学参考之用。
读者将信息用于其他用途的，全部法律及连带责任由读者自行承担，本站不承担任何责任。
评论