在上一篇《结构化输出与自修复：别让 JSON 解析毁了整条链路》里，我们通过联动 mini vLLM 的 LogitsProcessor 掩码、本地确定性词法修复器和 JSONPath 递归校验器，彻底解决了“模型吐不出合法工具调用参数”的顽疾。
现在，当模型结束 Phase A（策略生成阶段）时，我们的 Runtime 已经能够稳稳当当地拿到一组结构化、类型合规的 tool_calls 列表了。
但当我们进入 Phase B（动作消化阶段），去真正执行这些工具调用时，之前在第 1 篇和第 3 篇里写下的那几行看似人畜无害的代码，在真实生产负载下会立刻成为整个系统的最大性能瓶颈与稳定性黑洞：
# 第三篇 Phase B 中的同步串行循环：简单，但脆弱且缓慢
for call_id in list(step_record.pending_call_ids):
tc = call_map[call_id]
obs = self.tool_registry.execute(tc)
session.commit_observation(step_record, obs)
为什么这几行简单的 for 循环在生产级 Agent Runtime 里是绝对不可接受的？我们随便举四个每天都会在线上发生的真实场景：
多工具串行等待导致端到端延迟爆炸（Serial I/O Bottleneck）。现代大模型（包括 DeepSeek、Qwen、Claude 和 OpenAI 系列）都具备极强的 Parallel Tool Calling（单轮并行工具调用） 能力。比如你让 Agent “对比一下项目里
engine.py、scheduler.py、block_manager.py三个文件的核心类接口，并顺便查一下 GitHub 上最新的兩個相关 Issue”，模型在一轮思考后会一次性发出 M=5 个互不依赖的ToolCall。如果每次外部 API 或磁盘读取耗时 1.5 s，串行跑完这一步要足足 7.5 s；而如果并发执行，理想情况下只需要 1.5 s！单个慢工具或死锁卡死整个进程（Missing Timeout & Cancellation）。如果模型调用了一个抓取外部网页的工具
fetch_url或者执行某段命令，而目标服务器正好处于“连得上但迟迟不返回数据（Read Hang）”的僵尸状态，没有严格超时熔断机制的同步调用会直接把当前 Worker 线程永久挂死。同步阻塞函数污染异步事件循环（Blocking the Event Loop）。在 Python 生态里，用户注册进来的工具函数五花八门：有的是用
httpx/aiohttp写的原生异步协程async def，有的则是用requests、subprocess.run或普通文件读写写的同步阻塞函数def。如果直接在asyncio循环里直接调同步阻塞函数，整个事件循环的心跳都会被卡停，其他并发协程全部陪葬。瞬态网络抖动与非幂等重试灾难（Transient Faults vs. Non-Idempotent Retries）。外部世界是充满噪声的，调搜索引擎 API 可能会偶发
502 Bad Gateway或429 Too Many Requests。如果我们不加重试，一次网络毛刺就会白白浪费大模型一整轮推理 Step；但反过来，如果我们无脑给所有工具都加上自动重试，当一个执行“发送邮件”或“向文件追加一行配置”的**非幂等（Non-Idempotent）**工具在超时瞬间其实已经执行成功了，盲目重试就会酿成“连发三封重复邮件”或“写脏数据”的线上事故！
回想一下我们在《从零实现一个 mini vLLM -- 流式输出和异步 Engine》里做过的架构升级：当时为了让推理引擎能够同时服务成百上千个并发客户端，我们将同步阻塞的 Engine 改造为了基于 asyncio 事件循环的异步调度内核。
这一篇，我们就来对 Agent Runtime 的动作执行层做一次同等量级的手术——从零手写一个工业级的 AsyncToolExecutor，一次性解决 异构函数（Sync/Async）统一并发调度、信号量背压限流（Semaphore Backpressure）、硬超时取消（Timeout Cancellation）、幂等性感知的带抖动指数退避重试（Exponential Backoff with Jitter），以及结果保序归约。
1. 算一笔延迟与可靠性账：并发加速比与尾部延迟（Tail Latency）分析
在写代码之前，我们先从数学上把“串行 vs. 并发”以及“超时与重试”对系统端到端指标的影响推导清楚。
回顾我们在第 1 篇推导过的 Agent 端到端耗时公式，在第 k 步中，环境执行耗时记为 T_env(Aₖ)。假设模型在第 k 步一次性发出了 M 个并行的工具调用：
Aₖ = {aₖ,₁, aₖ,₂, …, aₖ,ₘ}
设第 i 个工具调用的单次执行耗时为一个非负随机变量 Xᵢ（例如服从某种长尾分布，均值 E[Xᵢ] = μ）。
1.1 串行执行 vs. 有界并发执行的期望耗时
在之前的同步串行执行器下，总耗时是所有工具耗时的线性累加：
T_serial(Aₖ) = Σ(i = 1…M) Xᵢ ⟹ E[T_serial(Aₖ)] = M·μ
而在全并发异步执行器（并发度上限 C ≥ M）下，所有工具同时启动，但由于 Phase B 必须等待当前步的所有 Observation 收集齐后才能组装下一轮 Prompt 进入 Phase A（即存在一个 Barrier 同步屏障），该步的实际耗时由最慢的那个工具（Order Statistic X₍ₘ₎）决定：
T_parallel(Aₖ) = max(1 ≤ i ≤ M) Xᵢ
如果每个工具耗时近似独立同分布且服从参数为 λ = 1/μ 的指数分布，根据次序统计量（Order Statistics）的经典结论，M 个并发任务的最大值期望为：
E[T_parallel(Aₖ)] = E[max(1 ≤ i ≤ M) Xᵢ] = μ·Σ(i = 1…M) 1/i = μ·H_M ≈ μ·(ln M + γ)
其中 H_M 是第 M 个调和数，γ ≈ 0.5772 是欧拉常数。
我们对比一下两者的理论加速比 S(M)：
S(M) = E[T_serial] / E[T_parallel] = M / H_M ≈ M / (ln M + 0.5772)
当 M = 8（例如 Agent 同时读取 8 个源码文件）且平均耗时 μ = 0.5 s 时：
串行耗时为 8 × 0.5 = 4.0 s；
并发期望耗时仅为 0.5 × H₈ ≈ 0.5 × 2.718 = 1.36 s，速度直接提升了接近 3 倍！如果各工具耗时方差较小（接近常数 μ），加速比更是直接逼近理想的 M = 8 倍。
1.2 为什么并发必须配硬超时（Timeout τ_max）？
注意上面公式里隐藏的另一面：因为存在 max 同步屏障，并发度 M 越大，撞上单个长尾慢请求（Tail Latency）的概率就越高！
设单个工具调用因为网络卡死或外部服务拥塞而耗时超过阈值 τ 的概率为 p_tail = P(Xᵢ > τ)。在 M 个工具并发时，整步被至少一个慢工具拖慢超过 τ 的概率为：
P(max(1 ≤ i ≤ M) Xᵢ > τ) = 1 − (1 − p_tail)^M
哪怕单个工具卡死的概率只有 p_tail = 5%，当并发调用 M = 10 个工具时，这一步被卡住的概率就会飙升到 1 − 0.95^10 ≈ 40.1%！
因此，在异步并发执行器中，每个工具必须绑定独立的硬超时上限 τ_max,ᵢ。这相当于对随机变量 Xᵢ 施加了一个右截断算子（Right-Censoring Operator）：
X̃ᵢ = min(Xᵢ, τ_max,ᵢ)
一旦某个工具耗时触及 τ_max,ᵢ，Runtime 立刻取消该协程，并将其结果替换为一条耗时精确等于 τ_max,ᵢ 的超时错误观测 Observation(is_error=True, output="Timeout...")，绝不让一颗老鼠屎拖垮整轮迭代。
1.3 带全抖动的指数退避重试（Exponential Backoff with Full Jitter）
对于被标记为**幂等（Idempotent，如只读查询、网页搜索）**的工具，当遇到网络超时或 503 Service Unavailable 等瞬态错误（Transient Error）时，我们允许执行器在内部自动重试最多 R 次。
如果单次调用的瞬态失败率为 q ∈ (0, 1)，在经过最多 R 次独立重试（共 R+1 次尝试）后，最终依然失败的残余概率呈指数级下降：
P_fail(R) = q^(R+1)
例如单次网络抖动率为 q = 10%，只需开启 R = 2 次自动重试，最终暴露给大模型的工具失败率就会从 10% 暴降到 0.1^3 = 0.1%！
但在并发环境下，重试等待时间绝对不能写成固定的常数。如果 5 个并发请求在同一毫秒触发了外部 API 的 429 Rate Limit，并且都休眠固定的 1.0 s 后同时发起重试，它们会在第 1.0 s 再次制造一波脉冲流量（惊群效应 / Thundering Herd），继续全部撞上 429！
因此，我们在第 r 次重试（r = 1, 2, …, R）前，采用 AWS 架构团队经典的 Full Jitter（全抖动指数退避） 策略计算休眠时间 Δt_r：
Δt_r ~ Uniform( 0, min(Δt_max, Δt_base · 2^(r-1)) )
通过引入均匀分布随机采样，原本同一时刻撞墙的并发请求会被均匀地打散在时间轴上，极大提高了重试成功率。
2. 架构设计：异构工具（Sync + Async）的统一调度模型
理清了数学模型，我们来看工程实现上最棘手的问题：怎么在同一个 AsyncToolExecutor 里，既能丝滑调度原生 async def 协程，又能安全运行普通的 def 同步函数，还不阻塞事件循环？
我们来看看 AsyncToolExecutor 的内部流水线设计：
StepRecord.tool_calls = [tc_1, tc_2, ..., tc_M]
│
▼ asyncio.gather(*[_run_single_with_guard(tc) for tc in tool_calls])
┌───────────────────────────────────────────────────────────────────────────┐
│ グローバル并发限流信号量 (asyncio.Semaphore(C)) │
└───────┬───────────────────────────┬───────────────────────────┬───────────┘
│ │ │
▼ (Slot 1) ▼ (Slot 2) ▼ (Slot M)
┌───────────────────┐ ┌───────────────────┐ ┌───────────────────┐
│ Schema & 权限校验 │ │ Schema & 权限校验 │ │ Schema & 权限校验 │
└─────────┬─────────┘ └─────────┬─────────┘ └─────────┬─────────┘
│ │ │
▼ ▼ ▼
┌───────────────────┐ ┌───────────────────┐ ┌───────────────────┐
│ 幂等重试状态机 │ │ 幂等重试状态机 │ │ 幂等重试状态机 │
│ (Max Retries + │ │ (Max Retries + │ │ (Max Retries + │
│ Full Jitter) │ │ Full Jitter) │ │ Full Jitter) │
└─────────┬─────────┘ └─────────┬─────────┘ └─────────┬─────────┘
│ │ │
▼ ▼ ▼
┌───────────────────────────────────────────────────────────────────────────┐
│ 单次超时沙箱护栏 (asyncio.wait_for(timeout)) │
│ • 若 inspect.iscoroutinefunction(fn) -> 直接 await fn(**args) │
│ • 若为普通同步阻塞函数 def fn -> await asyncio.to_thread(fn, ...) │
└─────────────────────────────────────┬─────────────────────────────────────┘
│
▼
按原始 tool_calls 顺序保序组装 List[Observation]
这里有三个非常关键的工程细节：
自动识别协程与同步函数并卸载到线程池：在注册工具时，我们用
inspect.iscoroutinefunction(func)检测函数类型。如果是async def，直接在当前事件循环中await；如果是普通的同步def（比如里面调了阻塞的time.sleep、磁盘大文件读写或subprocess），我们自动使用await asyncio.to_thread(func, **args)将其扔进后台的ThreadPoolExecutor工作线程中执行！这样主事件循环永远不会被卡死，而且多个同步 I/O 工具也能在多线程下实现真正的并行。显式区分可重试异常与不可重试异常：
如果报错是 参数校验失败（
SchemaValidationError） 或者 代码逻辑错误（KeyError,ZeroDivisionError,TypeError），重试一万次结果也完全一样，必须在第 1 次失败时立即终止并返回给大模型修改参数；只有当报错属于 瞬态环境异常（
TimeoutError,ConnectionError,OSError或自定义的TransientToolError），且该工具在注册时声明了idempotent=True（幂等），状态机才会触发指数退避重试。
严格的保序归约（Order-Preserving Reduction）：虽然 M 个工具在后台是乱序完成的（比如第 3 个工具 50 ms 就跑完了，第 1 个工具跑了 800 ms），但在把
Observation写回AgentSession的消息列表时，必须严格按照tool_calls的原始顺序[obs_1, obs_2, ..., obs_M]排列。原因有两个：第一，OpenAI 等协议规范要求role: "tool"的响应顺序与assistant.tool_calls一一对应；第二，保持确定性的追加顺序是我们在第 1 篇和第 12 篇强调的 KV Cache 前缀稳定对齐 的前提！
3. 核心代码实现：编写 AsyncToolExecutor
我们来动手编写 async_executor.py。
为了让每个工具可以灵活配置自己的超时时间、幂等性标志和最大重试次数，我们在 @registry.register 装饰器上增加一组实用的执行期策略元数据 ToolExecutionPolicy。
from __future__ import annotations
import asyncio
import inspect
import json
import random
import time
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional, Tuple, Type
from approval import ApprovalRule
from json_repair import DeterministicJSONRepairer
from schema import Observation, ToolCall
from schema_validator import SchemaValidator
from tools import ToolDefinition, _python_type_to_json_schema
from typing import get_type_hints
class TransientToolError(RuntimeError):
"""
工具函数可主动抛出的瞬态异常（如 HTTP 429 / 503、数据库连接池暂满）。
当工具声明为 idempotent=True 时，捕获到此异常会自动触发退避重试。
"""
pass
@dataclass
class ToolExecutionPolicy:
"""单个工具的并发、超时与重试控制策略"""
timeout_sec: float = 15.0 # 单次执行硬超时上限 (秒)
idempotent: bool = True # 是否为幂等工具（非幂等工具严禁自动重试！）
max_retries: int = 0 # 最大自动重试次数（0 表示不重试）
retry_base_delay_sec: float = 0.2 # 指数退避基础延迟
retry_max_delay_sec: float = 5.0 # 指数退避最大延迟上限
retryable_exceptions: Tuple[Type[BaseException], ...] = (
TransientToolError,
TimeoutError,
asyncio.TimeoutError,
ConnectionError,
)
@dataclass
class AsyncToolDefinition(ToolDefinition):
"""扩展支持异步协程检测、审批规则与执行策略的工具定义"""
is_async: bool = False
policy: ToolExecutionPolicy = field(default_factory=ToolExecutionPolicy)
approval_rule: ApprovalRule = field(default_factory=ApprovalRule)
class AsyncToolExecutor:
"""
工业级异步工具调度与执行引擎：
- 统一支持 async def 协程与同步 def 函数（自动 to_thread 线程池卸载）
- 基于 asyncio.Semaphore 的全局并发度限流（Backpressure）
- 单工具独立硬超时熔断（Timeout Cancellation）
- 幂等性感知的全抖动指数退避重试（Full-Jitter Exponential Backoff）
- 集成第 4 篇的 DeterministicJSONRepairer 与 SchemaValidator
"""
def __init__(self, max_concurrency: int = 8) -> None:
self.max_concurrency = max_concurrency
self._semaphore: Optional[asyncio.Semaphore] = None
self._tools: Dict[str, AsyncToolDefinition] = {}
def _get_semaphore(self) -> asyncio.Semaphore:
# 延迟绑定到当前运行的事件循环
if self._semaphore is None:
self._semaphore = asyncio.Semaphore(self.max_concurrency)
return self._semaphore
def register(
self,
func: Optional[Callable[..., Any]] = None,
*,
name: Optional[str] = None,
description: Optional[str] = None,
timeout_sec: float = 15.0,
idempotent: bool = True,
max_retries: int = 0,
retry_base_delay_sec: float = 0.2,
requires_approval: bool = False,
approval_policy: Optional[Callable[[Dict[str, Any]], bool]] = None,
) -> Callable[..., Any]:
"""注册同步或异步工具函数，并绑定超时、幂等重试与审批策略"""
def _decorator(f: Callable[..., Any]) -> Callable[..., Any]:
tool_name = name or f.__name__
tool_desc = (description or inspect.getdoc(f) or f"Execute tool {tool_name}.").strip()
sig = inspect.signature(f)
try:
type_hints = get_type_hints(f)
except Exception:
type_hints = {}
properties: Dict[str, Any] = {}
required: List[str] = []
for param_name, param in sig.parameters.items():
if param_name in ("self", "cls"):
continue
if param.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
continue
hint = type_hints.get(param_name, param.annotation)
prop_schema = _python_type_to_json_schema(hint)
if param.default is not inspect.Parameter.empty:
prop_schema["default"] = param.default
else:
required.append(param_name)
properties[param_name] = prop_schema
parameters_schema = {
"type": "object",
"properties": properties,
"required": required,
}
exec_policy = ToolExecutionPolicy(
timeout_sec=timeout_sec,
idempotent=idempotent,
max_retries=max_retries if idempotent else 0, # 非幂等工具强制关闭自动重试！
retry_base_delay_sec=retry_base_delay_sec,
)
self._tools[tool_name] = AsyncToolDefinition(
name=tool_name,
description=tool_desc,
func=f,
parameters_schema=parameters_schema,
sig=sig,
type_hints=type_hints,
is_async=inspect.iscoroutinefunction(f),
policy=exec_policy,
approval_rule=ApprovalRule(
requires_approval=requires_approval,
policy_fn=approval_policy,
),
)
return f
if func is not None:
return _decorator(func)
return _decorator
def get_schemas(self) -> List[Dict[str, Any]]:
return [t.to_openai_schema() for t in self._tools.values()]
def needs_approval(self, tool_call: ToolCall) -> bool:
tool_def = self._tools.get(tool_call.name)
if tool_def is None:
return False
return tool_def.approval_rule.should_ask_human(tool_call.arguments)
async def _invoke_once_with_timeout(
self, tool_def: AsyncToolDefinition, validated_args: Dict[str, Any]
) -> Any:
"""在独立硬超时护栏内执行单次函数调用（自动适配 async 协程与同步函数）"""
if tool_def.is_async:
coro = tool_def.func(**validated_args)
else:
# 将同步阻塞函数卸载到工作线程池，避免阻塞主事件循环
coro = asyncio.to_thread(tool_def.func, **validated_args)
return await asyncio.wait_for(coro, timeout=tool_def.policy.timeout_sec)
async def execute_one(self, tool_call: ToolCall) -> Observation:
"""
执行单个 ToolCall，完整包含：
1. 工具存在性检查
2. 第 4 篇的 JSON 确定性修复与 Schema 深度校验
3. 信号量并发控制 + 幂等指数退避重试 + 硬超时熔断
"""
overall_start = time.perf_counter()
tool_name = tool_call.name
if tool_name not in self._tools:
elapsed = (time.perf_counter() - overall_start) * 1000.0
return Observation(
call_id=tool_call.id,
tool_name=tool_name,
output=f"Error: Tool '{tool_name}' is not registered. Available: {list(self._tools.keys())}",
is_error=True,
execution_time_ms=elapsed,
)
tool_def = self._tools[tool_name]
# 1. 如果上游解析失败（含 _parse_error），尝试调用第 4 篇的确定性修复器自动抢救
args_dict = tool_call.arguments
applied_fixes: List[str] = []
if "_parse_error" in args_dict and tool_call.raw_arguments:
repair_res = DeterministicJSONRepairer.parse_or_repair(tool_call.raw_arguments)
if repair_res.success and repair_res.parsed_data is not None:
args_dict = repair_res.parsed_data
applied_fixes = repair_res.applied_fixes
else:
elapsed = (time.perf_counter() - overall_start) * 1000.0
diag = SchemaValidator.format_repair_prompt(
tool_name, tool_call.raw_arguments, [], parse_error=repair_res.error_message
)
return Observation(
call_id=tool_call.id,
tool_name=tool_name,
output=diag,
is_error=True,
execution_time_ms=elapsed,
)
# 2. 调用第 4 篇的 SchemaValidator 做递归参数校验与类型矫正
coerced_args, issues = SchemaValidator.validate_and_coerce(args_dict, tool_def.parameters_schema)
if issues:
elapsed = (time.perf_counter() - overall_start) * 1000.0
diag = SchemaValidator.format_repair_prompt(
tool_name, tool_call.raw_arguments or json.dumps(args_dict), issues
)
return Observation(
call_id=tool_call.id,
tool_name=tool_name,
output=diag,
is_error=True,
execution_time_ms=elapsed,
)
# 3. 进入并发信号量护栏与重试状态机
policy = tool_def.policy
max_attempts = 1 + (policy.max_retries if policy.idempotent else 0)
attempt = 0
last_exception: Optional[BaseException] = None
sem = self._get_semaphore()
while attempt < max_attempts:
attempt += 1
try:
# 仅在实际执行 I/O 或计算时占用并发槽位，退避休眠期间释放槽位给其他协程！
async with sem:
result = await self._invoke_once_with_timeout(tool_def, coerced_args)
if isinstance(result, (dict, list)):
output_str = json.dumps(result, ensure_ascii=False, indent=2)
else:
output_str = str(result)
elapsed = (time.perf_counter() - overall_start) * 1000.0
return Observation(
call_id=tool_call.id,
tool_name=tool_name,
output=output_str,
is_error=False,
execution_time_ms=elapsed,
metadata={
"attempts": attempt,
"json_fixes": applied_fixes,
"is_async": tool_def.is_async,
},
)
except policy.retryable_exceptions as exc:
last_exception = exc
# 如果还有剩余重试次数，计算 Full Jitter 退避时间并休眠
if attempt < max_attempts:
cap = min(
policy.retry_max_delay_sec,
policy.retry_base_delay_sec * (2 ** (attempt - 1)),
)
sleep_sec = random.uniform(0.0, cap)
await asyncio.sleep(sleep_sec)
continue
break
except Exception as exc:
# 不可重试的业务逻辑异常（如 FileNotFoundError / ValueError）：立即退出重试循环
last_exception = exc
break
# 4. 走到这里说明所有尝试均已失败，生成带尝试次数与超时诊断的 Observation
elapsed = (time.perf_counter() - overall_start) * 1000.0
if isinstance(last_exception, (TimeoutError, asyncio.TimeoutError)):
err_msg = (
f"[TOOL TIMEOUT] Tool '{tool_name}' exceeded its execution timeout of "
f"{policy.timeout_sec}s after {attempt} attempt(s). "
f"Try narrowing your query or processing a smaller batch."
)
else:
err_msg = (
f"[TOOL EXECUTION ERROR] Tool '{tool_name}' failed after {attempt} attempt(s) "
f"with {type(last_exception).__name__}: {last_exception}"
)
return Observation(
call_id=tool_call.id,
tool_name=tool_name,
output=err_msg,
is_error=True,
execution_time_ms=elapsed,
metadata={"attempts": attempt, "failed": True},
)
async def execute_batch(self, tool_calls: List[ToolCall]) -> List[Observation]:
"""
并发执行一批 ToolCall，并严格按照输入列表的顺序返回 Observation 列表。
"""
if not tool_calls:
return []
if len(tool_calls) == 1:
return [await self.execute_one(tool_calls[0])]
tasks = [self.execute_one(tc) for tc in tool_calls]
# asyncio.gather 天然保证返回结果的顺序与传入 coroutines 的顺序严格一致
observations = await asyncio.gather(*tasks)
return list(observations)
请特别留意上面 execute_one 里的一个极其讲究的并发细节：
# 仅在实际执行 I/O 或计算时占用并发槽位，退避休眠期间释放槽位给其他协程！
async with sem:
result = await self._invoke_once_with_timeout(tool_def, coerced_args)
很多同学在写带重试的并发限流时，习惯把 async with sem: 写在 while attempt < max_attempts: 循环的最外层。那会导致什么后果？当一个协程因为遇到 429 Rate Limit 正在 await asyncio.sleep(2.0) 退避休眠时，它居然还死死霸占着宝贵的 Semaphore 并发槽位不放，导致其他正常等待执行的工具进不来！把 async with sem: 放在单次尝试内部，一旦进入退避休眠就立刻归还信号量，这才是高吞吐运行时该有的细腻度。
4. 将状态机升级为异步并发内核（AsyncStatefulRuntime）
有了 AsyncToolExecutor，我们把第 3 篇的 StatefulAgentRuntime 升級为异步并发版本就水到渠成了。
在 Phase B（动作消化阶段）中，我们不再一个一个串行调工具，而是把所有已经通过审批（或无需审批）的待执行调用一次性打包交给 await self.tool_executor.execute_batch(ready_calls) 并发推进：
from __future__ import annotations
import time
from typing import Dict, List, Optional
from approval import ApprovalDecisionType
from async_executor import AsyncToolExecutor
from loop_detector import ActionLoopDetector
from runtime import LLMClientProtocol
from schema import Observation, SessionStatus, ToolCall
from session_v2 import AgentSessionV2, FileSessionStore, StepRecordV2
class AsyncAgentRuntime:
"""
支持多工具并发执行、超时熔断与断点恢复的异步状态机 Runtime。
"""
def __init__(
self,
llm_client: LLMClientProtocol,
tool_executor: AsyncToolExecutor,
store: Optional[FileSessionStore] = None,
) -> None:
self.llm_client = llm_client
self.tool_executor = tool_executor
self.store = store
def _checkpoint(self, session: AgentSessionV2) -> None:
if self.store is not None:
self.store.save(session)
async def tick(self, session: AgentSessionV2) -> StepRecordV2:
if session.is_terminal():
raise RuntimeError(f"Session {session.session_id} already in terminal state {session.status.value}")
if session.is_suspended():
raise RuntimeError(f"Session {session.session_id} is WAITING_APPROVAL.")
tick_start = time.perf_counter()
has_pending_work = bool(session.steps and session.steps[-1].pending_call_ids)
if not has_pending_work:
# ================= Phase A: 策略生成阶段 =================
if session.current_step >= session.config.max_steps:
session.status = SessionStatus.MAX_STEPS_EXCEEDED
session.error_reason = f"Exceeded max_steps ({session.config.max_steps})."
self._checkpoint(session)
return session.steps[-1]
session.status = SessionStatus.THINKING
self._checkpoint(session)
history_fps = session.get_action_fingerprints()
llm_start = time.perf_counter()
llm_resp = self.llm_client.generate(
messages=session.messages,
tool_schemas=self.tool_executor.get_schemas(),
config=session.config,
)
llm_latency_ms = (time.perf_counter() - llm_start) * 1000.0
step_record = session.begin_step_from_llm(
thought=llm_resp.content,
tool_calls=llm_resp.tool_calls,
usage=llm_resp.usage,
llm_latency_ms=llm_latency_ms,
)
if not llm_resp.tool_calls:
session.status = SessionStatus.COMPLETED
session.final_answer = llm_resp.content or ""
step_record.step_total_ms = (time.perf_counter() - tick_start) * 1000.0
self._checkpoint(session)
return step_record
# 死循环检测
detector = ActionLoopDetector(max_consecutive_repeats=session.config.max_consecutive_loops)
loop_res = detector.check(history_fps, llm_resp.tool_calls)
if loop_res.is_looping:
session.loop_intervention_count += 1
if session.loop_intervention_count > session.config.max_loop_interventions:
session.status = SessionStatus.FAILED
session.error_reason = f"Aborted due to persistent action loop ({loop_res.loop_type})."
self._checkpoint(session)
return step_record
for tc in list(llm_resp.tool_calls):
session.commit_observation(
step_record,
Observation(call_id=tc.id, tool_name=tc.name, output=loop_res.warning_message, is_error=True),
)
step_record.step_total_ms = (time.perf_counter() - tick_start) * 1000.0
self._checkpoint(session)
return step_record
# 审批拦截检查
approval_needed_ids = [
tc.id for tc in llm_resp.tool_calls if self.tool_executor.needs_approval(tc)
]
if approval_needed_ids:
session.awaiting_approval_ids = approval_needed_ids
session.status = SessionStatus.WAITING_APPROVAL
step_record.step_total_ms = (time.perf_counter() - tick_start) * 1000.0
self._checkpoint(session)
return step_record
# ================= Phase B: 并发动作消化阶段 =================
step_record = session.steps[-1]
session.status = SessionStatus.EXECUTING_TOOLS
self._checkpoint(session)
call_map = {tc.id: tc for tc in step_record.tool_calls}
calls_to_execute: List[ToolCall] = []
rejected_observations: Dict[str, Observation] = {}
for call_id in list(step_record.pending_call_ids):
tc = call_map[call_id]
decision = session.approval_decisions.get(call_id)
if decision is not None:
if decision.decision == ApprovalDecisionType.REJECT:
rejected_observations[call_id] = Observation(
call_id=tc.id,
tool_name=tc.name,
output=f"[HUMAN REJECTED ACTION] Reason: {decision.feedback or 'Denied.'}",
is_error=True,
)
continue
elif decision.decision == ApprovalDecisionType.EDIT and decision.modified_arguments is not None:
tc = ToolCall(id=tc.id, name=tc.name, arguments=decision.modified_arguments)
calls_to_execute.append(tc)
# 并发执行所有获批的工具调用！
executed_obs_list = await self.tool_executor.execute_batch(calls_to_execute)
executed_obs_map = {obs.call_id: obs for obs in executed_obs_list}
# 严格按照原始 pending_call_ids 顺序将 Observation 写回轨迹，确保前缀缓存确定性
for call_id in list(step_record.pending_call_ids):
obs = rejected_observations.get(call_id) or executed_obs_map[call_id]
session.commit_observation(step_record, obs)
step_record.step_total_ms += (time.perf_counter() - tick_start) * 1000.0
self._checkpoint(session)
return step_record
5. 实战压测：并发加速比、硬超时熔断与幂等 vs. 非幂等重试对比
我们来写一个完整的压测脚本 demo_async_executor.py，在同一轮并行调用 5 个特性完全不同的工具，直观检验 AsyncToolExecutor 的全部工程特性：
工具 1（原生异步 I/O
fetch_doc_async）：耗时200ms，验证原生协程调度。工具 2（同步阻塞 I/O
compute_digest_sync）：内部使用阻塞的time.sleep(0.2)，验证asyncio.to_thread线程池卸载——它和工具 1 同时跑，总耗时依然只有~200ms而不是400ms！工具 3（偶发抖动的幂等查询
query_ unstable_api）：声明idempotent=True, max_retries=3，前两次调用故意抛出TransientToolError("HTTP 503 Service Unavailable")，第 3 次成功返回，验证 Full Jitter 指数退避自动自愈。工具 4（卡死的僵尸请求
hanging_database_query）：设置timeout_sec=0.25，内部故意休眠5.0s，验证在250ms时被 硬超时准时熔断 并转化为清晰的[TOOL TIMEOUT]观测，绝不拖慢整批任务。工具 5（非幂等写操作
send_webhook_notification）：声明idempotent=False, max_retries=3（即使误配了max_retries=3，执行器也会因为idempotent=False强制禁止重试），当第一次抛出TransientToolError时绝不自动重试，防止重复发送通知，直接将错误交还给大模型决策。
import asyncio
import time
from async_executor import AsyncToolExecutor, TransientToolError
from schema import ToolCall
executor = AsyncToolExecutor(max_concurrency=4)
# 1. 原生异步工具 (耗时 200ms)
@executor.register(timeout_sec=2.0, idempotent=True)
async def fetch_doc_async(doc_id: str) -> str:
"""异步拉取文档内容。"""
await asyncio.sleep(0.20)
return f"Document[{doc_id}]: PagedAttention reduces KV cache fragmentation."
# 2. 传统同步阻塞工具 (耗时 200ms，将被自动卸载到线程池并行执行)
@executor.register(timeout_sec=2.0, idempotent=True)
def compute_digest_sync(file_name: str) -> str:
"""同步计算文件摘要（内部包含阻塞调用）。"""
time.sleep(0.20)
return f"SHA256({file_name}) = 9f86d081884c7d65"
# 3. 偶发网络抖动的幂等工具（前 2 次抛 503 瞬态错误，第 3 次成功）
_unstable_counter = 0
@executor.register(timeout_sec=1.0, idempotent=True, max_retries=3, retry_base_delay_sec=0.05)
async def query_metrics_idempotent(metric_name: str) -> dict:
"""查询监控指标（幂等只读接口，支持自动退避重试）。"""
global _unstable_counter
_unstable_counter += 1
if _unstable_counter <= 2:
raise TransientToolError(f"503 Service Unavailable (attempt {_unstable_counter})")
return {"metric": metric_name, "p99_latency_ms": 42.5, "recovered_on_attempt": _unstable_counter}
# 4. 卡死的慢工具（耗时 5.0s，但硬超时设为 0.25s）
@executor.register(timeout_sec=0.25, idempotent=False)
async def slow_hanging_tool(target: str) -> str:
"""模拟远端死锁卡住的请求。"""
await asyncio.sleep(5.0)
return f"Unreachable result for {target}"
# 5. 非幂等写操作工具（即使发生瞬态错误也严禁自动重试！）
_webhook_call_count = 0
@executor.register(timeout_sec=1.0, idempotent=False, max_retries=3)
async def send_alert_non_idempotent(channel: str, message: str) -> str:
"""发送告警通知（非幂等操作，禁止自动重试以防重复轰炸）。"""
global _webhook_call_count
_webhook_call_count += 1
raise TransientToolError("Network reset while reading webhook response")
async def main() -> None:
batch_calls = [
ToolCall(id="call_1", name="fetch_doc_async", arguments={"doc_id": "vllm_01"}),
ToolCall(id="call_2", name="compute_digest_sync", arguments={"file_name": "engine.py"}),
ToolCall(id="call_3", name="query_metrics_idempotent", arguments={"metric_name": "gpu_cache_usage"}),
ToolCall(id="call_4", name="slow_hanging_tool", arguments={"target": "dead_server"}),
ToolCall(id="call_5", name="send_alert_non_idempotent", arguments={"channel": "#ops", "message": "high load"}),
]
print("=== 启动 5 个异构工具的并发 Batch 执行 ===")
start_wall = time.perf_counter()
observations = await executor.execute_batch(batch_calls)
total_wall_ms = (time.perf_counter() - start_wall) * 1000.0
sum_individual_ms = sum(obs.execution_time_ms for obs in observations)
for obs in observations:
print(f"\n[{obs.call_id}] Tool: {obs.tool_name}")
print(f" - Status : {'ERROR' if obs.is_error else 'SUCCESS'}")
print(f" - Time : {obs.execution_time_ms:.1f} ms | Metadata: {obs.metadata}")
print(f" - Output : {obs.output}")
print("\n=== 并发性能与可靠性汇总 ===")
print(f"5 个工具累计耗时总和 (若串行) : {sum_individual_ms:.1f} ms")
print(f"Batch 实际端到端总耗时 (并发) : {total_wall_ms:.1f} ms")
print(f"实际并发加速比 (Speedup) : {sum_individual_ms / total_wall_ms:.2f}x")
print(f"非幂等工具实际触发次数 : {_webhook_call_count} 次 (严格保证 At-Most-Once)")
if __name__ == "__main__":
asyncio.run(main())
我们运行一下这段压测代码，看看控制台输出的真实指标：
=== 启动 5 个异构工具的并发 Batch 执行 ===
[call_1] Tool: fetch_doc_async
- Status : SUCCESS
- Time : 200.6 ms | Metadata: {'attempts': 1, 'json_fixes': [], 'is_async': True}
- Output : Document[vllm_01]: PagedAttention reduces KV cache fragmentation.
[call_2] Tool: compute_digest_sync
- Status : SUCCESS
- Time : 201.2 ms | Metadata: {'attempts': 1, 'json_fixes': [], 'is_async': False}
- Output : SHA256(engine.py) = 9f86d081884c7d65
[call_3] Tool: query_metrics_idempotent
- Status : SUCCESS
- Time : 68.4 ms | Metadata: {'attempts': 3, 'json_fixes': [], 'is_async': True}
- Output : {
"metric": "gpu_cache_usage",
"p99_latency_ms": 42.5,
"recovered_on_attempt": 3
}
[call_4] Tool: slow_hanging_tool
- Status : ERROR
- Time : 250.8 ms | Metadata: {'attempts': 1, 'failed': True}
- Output : [TOOL TIMEOUT] Tool 'slow_hanging_tool' exceeded its execution timeout of 0.25s after 1 attempt(s). Try narrowing your query or processing a smaller batch.
[call_5] Tool: send_alert_non_idempotent
- Status : ERROR
- Time : 0.4 ms | Metadata: {'attempts': 1, 'failed': True}
- Output : [TOOL EXECUTION ERROR] Tool 'send_alert_non_idempotent' failed after 1 attempt(s) with TransientToolError: Network reset while reading webhook response
=== 并发性能与可靠性汇总 ===
5 个工具累计耗时总和 (若串行) : 721.4 ms
Batch 实际端到端总耗时 (并发) : 251.3 ms
实际并发加速比 (Speedup) : 2.87x
非幂等工具实际触发次数 : 1 次 (严格保证 At-Most-Once)
仔细看这组实验数据，每一个数字都精准印证了我们的系统设计：
总耗时严格收敛于最慢的受控工具：5 个工具里原本有一个要卡死
5000ms的slow_hanging_tool，被timeout_sec=0.25准时在250.8ms斩断；同时fetch_doc_async（协程200ms）和compute_digest_sync（同步阻塞200ms，通过to_thread卸载）完全并行，最终整个 Batch 的端到端耗时只有251.3ms！瞬态毛刺静默自愈：
query_metrics_idempotent在前两次遭遇503错误后，在执行器内部经过两次微秒/毫秒级的全抖动退避重试，在第 3 次尝试时成功拿到结果（attempts: 3）。对上层的大模型来说，它甚至完全感知不到网络曾经抖动过，直接省下了一整轮几千 Token 的重推理开销！非幂等红线绝不越界：
send_alert_non_idempotent虽然同样抛出了TransientToolError，但由于idempotent=False，执行器严格遵守At-Most-Once语义，只执行了1次就立刻将异常转化为 Observation 交还给模型，绝不在后台自作主张地重发告警。
小结
这一篇我们完成了第二阶段的核心执行引擎升级，把原本简陋的串行 for 循环改造成了生产级的 AsyncToolExecutor。
回顾一下我们在本篇建立的核心工程原则：
用次序统计量与右截断算子指导并发设计：并发可以将 M 个工具的期望耗时从 Mμ 压低到 μ·H_M，但同时也会放大遭遇长尾慢请求的概率 1 − (1 − p_tail)^M。因此，无超时不并发，每个工具必须绑定独立的
asyncio.wait_for硬超时护栏。异构函数统一调度：通过
inspect.iscoroutinefunction自动分流原生协程与普通同步阻塞函数，利用asyncio.to_thread避免阻塞事件循环心跳，并保证在退避休眠期间释放Semaphore信号量槽位。严格区分幂等与非幂等重试语义：只读幂等工具通过 Full Jitter 指数退避在执行器内部静默吸收网络抖动（将失败率从 q 压至 q^(R+1)）；非幂等修改类工具强制关闭自动重试，防止产生重复副作用。
到这里，我们的 Agent 已经可以在单步之内并发执行多个工具了。
但请思考一个更高维度的调度问题：如果一个复杂的工程任务包含 8 个子步骤（例如：1. 查文档 A → 2. 查文档 B → 3. 汇总 A 和 B 的接口差异 → 4. 扫描代码库 C → 5. 根据 3 和 4 生成重构补丁 → 6. 跑单元测试），单靠 ReAct 这种“走一步看一步”的贪心循环，模型怎么提前知道第 1、2、4 步其实根本没有相互依赖，完全可以在第一轮就并发启动？又怎么防止模型在做到第 5 步时早就忘了最初的任务全局结构？
下一篇（第 6 篇），也是第二阶段的收官篇，我们将实现 《Plan-and-Solve：从单步走一步看一步到 DAG 任务调度》——手写一个基于拓扑排序（Topological Sort）与动态重规划（Dynamic Re-planning）的 有向无环图（DAG）任务调度引擎！
