From ae994a4b80f9fdc7caff96037e295f50e390e947 Mon Sep 17 00:00:00 2001 From: Denis_Drobyshev Date: Mon, 10 Aug 2026 22:27:39 +0300 Subject: [PATCH] run instances concurrently, for agents that wait on a network MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Measuring what the benchmark exists for turned up the obstacle to doing it. An episode is a chain of round-trips, because a decision cannot start until the previous one's outcome is known: 100 of them on queueing, 60 on inventory, energy and supply-chain, 40 on joint-pricing. One episode of every task is about 325 model calls, and a modest run across all six is 5,200 — an hour and a half of a language model waiting, single file. Instances are independent, so they overlap. Eight workers against an agent with 8 ms of simulated latency: 0.77s to 0.10s, with the returns bit-identical, because every seed is fixed before the pool starts and rows are collected by index. A test pins that equality — the moment concurrency changes a score, the scores stop being reproducible and this stops being a benchmark. Above one worker the agent argument is a factory rather than an instance, and passing an instance is refused with the reason. An agent driving a language model keeps the brief and the recent decisions for the episode it is in; shared across threads that memory interleaves, and the failure would look like a bad policy rather than a bug. --- README.md | 26 +++++++++++++++ README.ru.md | 27 +++++++++++++++ src/stadion/core/runner.py | 67 +++++++++++++++++++++++++++++++------- tests/test_runner.py | 29 +++++++++++++++++ 4 files changed, 137 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index 4762b17..299507e 100644 --- a/README.md +++ b/README.md @@ -83,6 +83,32 @@ same task without either being translated through the other's interface. For a language model, `stadion.llm.LLMAgent` takes any `prompt -> reply` callable — no client library, no provider. +### Running a language model + +An episode is a chain of round-trips: a decision cannot start until the previous +one's outcome is known, so a model spends the evaluation waiting rather than +computing. Instances do not depend on each other, so run them at once. + +```python +report = stadion.evaluate(task, lambda: LLMAgent(complete), instances=20, workers=8) +``` + +Above one worker the first argument is a factory, not an agent — an agent that +remembers anything within an episode cannot be shared across threads. The +numbers are identical either way; every seed is fixed before the pool starts. + +Budget the run in model calls first. One episode costs one call per decision: + +| Task | Calls per episode | Menu | +|---|---:|---:| +| `queueing` | 100 | 2 | +| `inventory`, `energy`, `supply-chain` | 60 | 9, 9, 25 | +| `joint-pricing` | 40 | 48 | +| `pricing` | ~5 | 8 | + +One episode of every task is about 325 calls, so a run of eight instances by two +episodes across all six is roughly 5,200. + From the shell: ```bash diff --git a/README.ru.md b/README.ru.md index 535b007..0b69918 100644 --- a/README.ru.md +++ b/README.ru.md @@ -83,6 +83,33 @@ class MyAgent(stadion.Agent): `stadion.llm.LLMAgent` принимает любую функцию `prompt -> reply`: ни клиентской библиотеки, ни провайдера. +### Прогон языковой модели + +Эпизод — цепочка обращений: следующее решение нельзя начать, пока не известен +исход предыдущего, поэтому модель проводит оценку в ожидании, а не в счёте. +Инстансы друг от друга не зависят, значит их можно гнать разом. + +```python +report = stadion.evaluate(task, lambda: LLMAgent(complete), instances=20, workers=8) +``` + +Больше одного воркера — и первым аргументом идёт фабрика, а не агент: агент, +который помнит что-либо внутри эпизода, между потоками не делится. Числа при +этом одни и те же, все сиды зафиксированы до старта пула. + +Бюджет прогона считается в вызовах модели заранее. Один эпизод стоит одного +вызова на решение: + +| Задача | Вызовов на эпизод | Меню | +|---|---:|---:| +| `queueing` | 100 | 2 | +| `inventory`, `energy`, `supply-chain` | 60 | 9, 9, 25 | +| `joint-pricing` | 40 | 48 | +| `pricing` | ~5 | 8 | + +Один эпизод каждой задачи — около 325 вызовов, то есть прогон на восьми +инстансах по два эпизода по всем шести задачам обойдётся примерно в 5 200. + Из консоли: ```bash diff --git a/src/stadion/core/runner.py b/src/stadion/core/runner.py index 601b6cd..34dec9b 100644 --- a/src/stadion/core/runner.py +++ b/src/stadion/core/runner.py @@ -18,6 +18,7 @@ from __future__ import annotations from collections.abc import Callable, Iterable, Sequence +from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from typing import TypeVar @@ -104,7 +105,7 @@ class Arms: def evaluate( task: Task, - agent: Agent, + agent: Agent | Callable[[], Agent], *, instances: int = 30, episodes: int = 20, @@ -113,25 +114,67 @@ def evaluate( resamples: int = 10_000, level: float = 0.95, bootstrap_seed: int = 0, + workers: int = 1, ) -> Report: - """Score ``agent`` on ``task`` against the classical method and the optimum.""" + """Score ``agent`` on ``task`` against the classical method and the optimum. + + ``workers`` runs instances concurrently. It exists for agents that wait on a + network: a decision cannot start until the previous one's outcome is known, + so an episode is a chain of round-trips and a language model spends the + evaluation waiting rather than computing. Instances do not depend on each + other, so they overlap freely, and the result is identical either way — + every seed is fixed in advance and the rows are collected by index. + + Above one worker, ``agent`` must be a factory rather than an instance: an + agent that remembers anything within an episode — as one driving a language + model must — cannot be shared across threads without its memory interleaving. + """ if instances < 2: raise ValueError(f"need at least 2 instances for an interval, got {instances}") + if workers < 1: + raise ValueError(f"workers must be at least 1, got {workers}") episode_seeds = tuple(episode_seed + j for j in range(episodes)) - rows_agent, rows_base, rows_opt = [], [], [] - for i in range(instances): - inst = task.instance(instance_seed + i) - rows_agent.append(play_instance(task, agent, inst, episode_seeds)) - rows_base.append(play_instance(task, task.baseline(inst), inst, episode_seeds)) - rows_opt.append(play_instance(task, task.optimal(inst), inst, episode_seeds)) + if isinstance(agent, Agent): + if workers > 1: + raise ValueError( + "workers > 1 needs a factory, not an agent instance.\n" + " An agent that keeps state within an episode would have that state " + "interleaved across threads.\n" + " Pass a zero-argument callable that returns a fresh agent, e.g. " + "evaluate(task, lambda: MyAgent(), workers=8)." + ) + shared = agent + + def build() -> Agent: + return shared + + name = agent.name + else: + build = agent + name = build().name + + def one(index: int) -> tuple[float, float, float]: + inst = task.instance(instance_seed + index) + return ( + play_instance(task, build(), inst, episode_seeds), + play_instance(task, task.baseline(inst), inst, episode_seeds), + play_instance(task, task.optimal(inst), inst, episode_seeds), + ) + + if workers == 1: + rows = [one(i) for i in range(instances)] + else: + with ThreadPoolExecutor(max_workers=workers) as pool: + rows = list(pool.map(one, range(instances))) arms = Arms( seeds=tuple(instance_seed + i for i in range(instances)), - agent=np.asarray(rows_agent), - baseline=np.asarray(rows_base), - optimal=np.asarray(rows_opt), + agent=np.asarray([r[0] for r in rows]), + baseline=np.asarray([r[1] for r in rows]), + optimal=np.asarray([r[2] for r in rows]), ) + def compare(x: np.ndarray, y: np.ndarray, label: str) -> Comparison: return paired_bootstrap( x, y, label=label, resamples=resamples, level=level, seed=bootstrap_seed @@ -139,7 +182,7 @@ def compare(x: np.ndarray, y: np.ndarray, label: str) -> Comparison: return Report( task=task.name, - agent=agent.name, + agent=name, instances=instances, episode_seeds=episode_seeds, agent_return=float(arms.agent.mean()), diff --git a/tests/test_runner.py b/tests/test_runner.py index 64a8a13..3398a9d 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -87,6 +87,35 @@ def test_a_deterministic_agent_reproduces_its_own_returns(task_name: str) -> Non assert stadion.reproducible(task, stadion.RandomAgent()) +def test_running_instances_concurrently_changes_nothing_but_the_wall_clock() -> None: + """Concurrency is for agents that wait on a network, not a different measurement. + + Every seed is fixed before the pool starts and rows are collected by index, + so the numbers have to come out bit-identical. If they ever do not, the + scores stop being reproducible and the benchmark stops being one. + """ + task = stadion.get("pricing") + serial = stadion.evaluate(task, _First(), instances=6, episodes=4) + parallel = stadion.evaluate(task, _First, instances=6, episodes=4, workers=4) + assert serial.agent_return == parallel.agent_return + assert serial.baseline_return == parallel.baseline_return + assert serial.optimal_return == parallel.optimal_return + assert serial.vs_baseline.ci == parallel.vs_baseline.ci + + +def test_sharing_one_agent_across_workers_is_refused() -> None: + """An agent that remembers the episode cannot be shared; say so before the run.""" + task = stadion.get("pricing") + with pytest.raises(ValueError, match="factory"): + stadion.evaluate(task, _First(), instances=4, episodes=2, workers=4) + + +def test_a_factory_is_accepted_on_a_single_worker_too() -> None: + task = stadion.get("pricing") + report = stadion.evaluate(task, _First, instances=4, episodes=2) + assert report.agent == "first" + + def test_an_evaluation_needs_enough_instances_for_an_interval() -> None: task = stadion.get("queueing") with pytest.raises(ValueError, match="at least 2 instances"):