|
6 | 6 | reference. Tool calls are always executed and delivered via the |
7 | 7 | loop's ``_execute_tool_call`` / ``_deliver_tool_result`` methods so |
8 | 8 | subclass or test overrides of those methods keep working. |
| 9 | +
|
| 10 | +When every call in a round is readonly (``Tool.is_readonly = True``), |
| 11 | +the round is dispatched concurrently via a thread pool: readonly tools |
| 12 | +only read state, so none can depend on another's side effects, and |
| 13 | +running them in parallel reduces latency for read-heavy rounds (e.g. |
| 14 | +the model reading several files at once). Mixed rounds (any |
| 15 | +non-readonly tool) fall back to the original sequential dispatch. |
9 | 16 | """ |
10 | 17 |
|
11 | 18 | from __future__ import annotations |
12 | 19 |
|
13 | 20 | import json |
14 | 21 | import time |
| 22 | +from concurrent.futures import ThreadPoolExecutor |
15 | 23 | from typing import Any |
16 | 24 |
|
17 | 25 | from . import config |
|
22 | 30 | "Error: tool produced no result (it may have been interrupted or failed to return)." |
23 | 31 | ) |
24 | 32 |
|
| 33 | +# Upper bound on threads spawned for a parallel readonly round. The |
| 34 | +# round size is driven by model output (a single response can emit |
| 35 | +# dozens of Read/Grep calls), so cap peak concurrency to avoid a |
| 36 | +# thread explosion; the pool still drains every call, just fewer at a |
| 37 | +# time. |
| 38 | +MAX_PARALLEL_READONLY = 8 |
| 39 | + |
25 | 40 |
|
26 | 41 | def sanitize_tool_result(result: object) -> str: |
27 | 42 | """Sanitize a tool result for the model. |
@@ -106,12 +121,22 @@ def run_tools(self, calls: list[ToolCall], results: dict[str, str]) -> None: |
106 | 121 | afterwards, again in original call order. Delivery happens |
107 | 122 | later, in original tool-call order, by the caller. |
108 | 123 |
|
| 124 | + When every call in the round is readonly (``Tool.is_readonly``), |
| 125 | + the round is dispatched concurrently via a thread pool: |
| 126 | + readonly tools only read state, so none can depend on |
| 127 | + another's side effects, and running them in parallel reduces |
| 128 | + latency for read-heavy rounds (e.g. the model reading several |
| 129 | + files at once). Mixed rounds fall back to sequential dispatch. |
| 130 | +
|
109 | 131 | A cancel landing before a call starts skips it (tools have side |
110 | 132 | effects); a call already running — or an async tool already |
111 | 133 | dispatched — cannot be stopped, but its result stays local to |
112 | 134 | the (dead) run. |
113 | 135 | """ |
114 | 136 | loop = self.loop |
| 137 | + if calls and self._all_readonly(calls): |
| 138 | + self._run_parallel(calls, results) |
| 139 | + return |
115 | 140 | async_calls: list[tuple[ToolCall, PendingToolResult]] = [] |
116 | 141 | for p in calls: |
117 | 142 | # A cancel landing while a call is still QUEUED must skip |
@@ -151,6 +176,63 @@ def run_tools(self, calls: list[ToolCall], results: dict[str, str]) -> None: |
151 | 176 | p.elapsed = time.monotonic() - start |
152 | 177 | results[p.id] = sanitize_tool_result(result) |
153 | 178 |
|
| 179 | + def _all_readonly(self, calls: list[ToolCall]) -> bool: |
| 180 | + """True when every call's tool is marked ``is_readonly``.""" |
| 181 | + loop = self.loop |
| 182 | + for p in calls: |
| 183 | + tool = loop.session.registry.get(p.name) |
| 184 | + if tool is None or not tool.is_readonly: |
| 185 | + return False |
| 186 | + return True |
| 187 | + |
| 188 | + def _run_parallel(self, calls: list[ToolCall], results: dict[str, str]) -> None: |
| 189 | + """Dispatch all calls concurrently via a thread pool. |
| 190 | +
|
| 191 | + Used only when every call is readonly. Each tool runs in its |
| 192 | + own thread; results are collected in original call order. |
| 193 | + Cancel is checked before dispatching (a call that has not |
| 194 | + started yet is skipped); a call already running cannot be |
| 195 | + stopped, but its result stays local to the (dead) run. |
| 196 | + """ |
| 197 | + loop = self.loop |
| 198 | + futures: dict[str, Any] = {} |
| 199 | + with ThreadPoolExecutor(max_workers=min(len(calls), MAX_PARALLEL_READONLY)) as pool: |
| 200 | + for p in calls: |
| 201 | + if loop._is_cancelled(): |
| 202 | + results[p.id] = "Error: tool call cancelled (user aborted the run)." |
| 203 | + continue |
| 204 | + if loop.top_level: |
| 205 | + loop.session.notify("tool_running", p.name) |
| 206 | + futures[p.id] = pool.submit(self._exec_one, p) |
| 207 | + for p in calls: |
| 208 | + fut = futures.get(p.id) |
| 209 | + if fut is None: |
| 210 | + continue |
| 211 | + try: |
| 212 | + result = fut.result() |
| 213 | + except Exception as e: # noqa: BLE001 - containment boundary |
| 214 | + results[p.id] = f"Error: tool {p.name!r} crashed during execution — {e}" |
| 215 | + else: |
| 216 | + if isinstance(result, PendingToolResult): |
| 217 | + result = result.wait() |
| 218 | + results[p.id] = sanitize_tool_result(result) |
| 219 | + |
| 220 | + def _exec_one(self, p: ToolCall) -> str | PendingToolResult: |
| 221 | + """Execute one tool call and return its raw result. |
| 222 | +
|
| 223 | + Thin wrapper around ``loop._execute_tool_call`` that records |
| 224 | + elapsed time on the call object. Used by ``_run_parallel``. |
| 225 | + """ |
| 226 | + loop = self.loop |
| 227 | + start = time.monotonic() |
| 228 | + try: |
| 229 | + result = loop._execute_tool_call(p) |
| 230 | + except Exception as e: # noqa: BLE001 - containment boundary |
| 231 | + p.elapsed = time.monotonic() - start |
| 232 | + return f"Error: tool {p.name!r} crashed during execution — {e}" |
| 233 | + p.elapsed = time.monotonic() - start |
| 234 | + return result |
| 235 | + |
154 | 236 | def execute_pending(self) -> None: |
155 | 237 | """TOOL state: run the round's pending tool calls. |
156 | 238 |
|
|
0 commit comments