`interlens.runner.pool`¶

interlens.runner.pool ¶

The execution engine behind Conversation.rollout and interlens.run.

A job is a (job_id, Conversation) pair — an unrun, fully-resolved conversation (lazy participants, no dataset). run_jobs runs a list of them across devices with checkpointing, resume, per-job failure isolation, and (by default) batched co-stepping within each device. Each finished conversation is returned on its RunResult so it can be sampled/inspected afterwards. The analyzer travels ON each conversation (conv._analyzer) — a callable in-process, or a registered name across a spawn boundary.

RunReport `dataclass` ¶

RunReport(results: dict = dict(), skipped: list = list())

Aggregate outcome of a run. Failures are isolated, not fatal: results holds every attempted job, failed lists the ones that errored, skipped lists resume-skipped (already-checkpointed) ids.

RunResult `dataclass` ¶

RunResult(
    job_id: str,
    conversation: object = None,
    transcript: object = None,
    analysis: object = None,
    error: str | None = None,
    device: str | None = None,
)

Outcome of one job: the finished conversation (weightless participants + completed transcript), its transcript and (serializable) analysis, or an error string if it failed. conversation is the object to sample()/inspect after a rollout (the source recipe is never mutated).

tokens_generated `property` ¶

tokens_generated: int

Total generated tokens in this conversation (summed from each turn's metadata['n_tokens']) — the realized compute, for verifying matched-compute comparisons. 0 if the job failed.

run ¶

run(
    conversations,
    devices=None,
    out_dir=None,
    resume=False,
    batched=True,
    max_batch_size=None,
) -> RunReport

Run several conversation lineups in ONE pool — the multi-lineup entry point (e.g. a ladder of model pairs × conditions in one overnight job).

Each conversation is expanded to its jobs (one per data() row if it has data, else a single conversation), with job ids namespaced by that conversation's name (default conv{i}) so ids stay unique and resumable. All jobs go into one pool, so GPUs stay packed across lineups (no idle tail between sequential rollouts) under a single out_dir/resume namespace, returning one merged RunReport. Mixing lineups is safe: batched co-stepping groups by schedule signature, so each distinct lineup forms its own batch group.

Conversation.rollout is the single-lineup sugar for this (run([conv], ...)-equivalent via run_jobs).

Source code in src/interlens/runner/pool.py

def run(conversations, devices=None, out_dir=None, resume=False, batched=True, max_batch_size=None) -> RunReport:
	"""Run several conversation lineups in ONE pool — the multi-lineup entry point (e.g. a ladder of model pairs ×
	conditions in one overnight job).

	Each conversation is expanded to its jobs (one per ``data()`` row if it has data, else a single conversation),
	with job ids namespaced by that conversation's ``name`` (default ``conv{i}``) so ids stay unique and resumable.
	All jobs go into one pool, so GPUs stay packed across lineups (no idle tail between sequential rollouts) under a
	single ``out_dir``/resume namespace, returning one merged ``RunReport``. Mixing lineups is safe: batched
	co-stepping groups by schedule signature, so each distinct lineup forms its own batch group.

	``Conversation.rollout`` is the single-lineup sugar for this (``run([conv], ...)``-equivalent via ``run_jobs``).
	"""
	jobs = []
	for i, conv in enumerate(conversations):
		jobs.extend(conv._jobs_for_run(i))
	return run_jobs(jobs, devices=devices, out_dir=out_dir, resume=resume, batched=batched,
	                max_batch_size=max_batch_size)

run_jobs ¶

run_jobs(
    jobs,
    devices=None,
    out_dir=None,
    resume=False,
    batched=True,
    max_batch_size=None,
) -> RunReport

Run (job_id, Conversation) jobs across devices, with checkpointing, resume, and per-job failure isolation.

Parallel by default on two axes: one worker process per device (jobs round-robined; multi-GPU spawns via torch.multiprocessing since fork+CUDA is broken — a single device runs in-process), and within each device batched co-stepping (batched=True). Jobs are grouped by co-step schedule signature so same-schedule jobs batch into one model.generate — correct for ANY mix. batched=False gives the DETERMINISTIC path.

Source code in src/interlens/runner/pool.py

def run_jobs(jobs, devices=None, out_dir=None, resume=False, batched=True, max_batch_size=None) -> RunReport:
	"""Run ``(job_id, Conversation)`` jobs across devices, with checkpointing, resume, and per-job failure isolation.

	Parallel by default on **two** axes: one worker **process per device** (jobs round-robined; multi-GPU spawns via
	``torch.multiprocessing`` since fork+CUDA is broken — a single device runs in-process), and within each device
	**batched co-stepping** (``batched=True``). Jobs are grouped by co-step schedule signature so same-schedule jobs
	batch into one ``model.generate`` — correct for ANY mix. ``batched=False`` gives the DETERMINISTIC path.
	"""
	devices = devices or available_devices()
	pending = [(jid, conv) for jid, conv in jobs if not (resume and _already_done(out_dir, jid))]
	skipped = [jid for jid, conv in jobs if resume and _already_done(out_dir, jid)]

	if len(devices) > 1:
		results = _run_spawn(pending, devices, out_dir, batched, max_batch_size)
	elif batched:
		results = _run_group(pending, devices[0], out_dir, max_batch_size)
	else:
		results = {}
		for i, (jid, conv) in enumerate(pending):
			results[jid] = _run_one(jid, conv, devices[i % len(devices)], out_dir)
	return RunReport(results=results, skipped=skipped)

interlens.runner.pool¶