Run instances concurrently, for agents that wait on a network

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

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

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,33 +114,75 @@ 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
         )
 
     return Report(
         task=task.name,
-        agent=agent.name,
+        agent=name,
         instances=instances,
         episode_seeds=episode_seeds,
         agent_return=float(arms.agent.mean()),

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"):