Honor the concurrency option in evaluate()

package.json

@@ -51,5 +51,8 @@
     "trailingComma": "all",
     "arrowParens": "avoid",
     "semi": false
+  },
+  "dependencies": {
+    "p-map": "^7.0.5"
   }
 }

readme.md

@@ -95,7 +95,7 @@ evaluate<I, O, E>(name: string, config: EvalConfig<I, O, E>): Promise<EvalReport
 | `scorers`     | `Scorer<I, O, E>[]`                       | one or more scorers                |
 | `models`      | `string[]`                                | models to sweep; at least one      |
 | `task`        | `Task<I, O>`                              | the system under test              |
-| `concurrency` | `number`                                  | optional; accepted, not yet honored |
+| `concurrency` | `number`                                  | optional; max concurrent cases per model (default 1, serial) |
 | `baseline`    | `string`                                  | optional; accepted, not yet honored |
 
 ### `scorer(name, run, opts?)`

src/evaluate.ts

@@ -1,13 +1,15 @@
 import assert from 'node:assert'
-import type { CaseResult, EvalConfig, EvalReport, Usage } from './types.js'
+import pMap from 'p-map'
+import type { Case, CaseResult, EvalConfig, EvalReport, Usage } from './types.js'
 import { to } from './utils.js'
 import { aggregate } from './aggregate.js'
 import { report } from './usage.js'
 
 /**
  * Sweeps `config.models`, runs the task on each case, scores the output, and
  * aggregates per model. A case whose task throws is recorded with `output: null`
- * and does not abort the run.
+ * and does not abort the run. Within a model, up to `config.concurrency` cases
+ * run at once (default 1, i.e. serial); models are still swept one at a time.
  *
  * @param name   name for this eval, included in the report.
  * @param config the eval configuration.
@@ -30,87 +32,93 @@ export async function evaluate<I, O, E>(name: string, config: EvalConfig<I, O, E
   const task = config.task
   assert(task, 'task is required')
 
+  const concurrency = config.concurrency ?? 1
+  assert(Number.isInteger(concurrency) && concurrency > 0, 'concurrency must be a positive integer')
+
   const byModel: EvalReport<O>['byModel'] = {}
 
   for (const model of models) {
-    const results: CaseResult<O>[] = []
-
-    for (const c of cases) {
-      const tags = c.tags ?? []
-      const input = c.input
-      const expected = c.expected
-
-      const usage: { task: Usage; judge: Usage } = {
-        task: { inputTokens: 0, outputTokens: 0 },
-        judge: { inputTokens: 0, outputTokens: 0 },
-      }
-
-      const start = performance.now()
-      const [error, result] = await to(
-        task(input, {
-          model,
-          report: usg => report(usage.task, usg),
-        }),
-      )
-      const latencyMs = performance.now() - start
-
-      if (error) {
-        results.push({
-          tags,
-          usage,
-          latencyMs,
-          score: 0,
-          scores: [],
-          output: null,
-        })
-
-        continue
-      }
-
-      const output = result!
-      const scores: CaseResult<O>['scores'] = []
-
-      for (const scorer of scorers) {
-        try {
-          const value = await scorer.run({
-            input,
-            output,
-            expected,
-            tags,
-            report: usg => report(usage.judge, usg),
-          })
-
-          if (value === null) continue
-          const normalized = typeof value === 'number' ? { score: value, reason: '' } : value
-
-          scores.push({
-            name: scorer.name,
-            score: normalized.score,
-            weight: scorer.weight ?? 1,
-            reason: normalized.reason ?? '',
-          })
-        } catch (err) {
-          // A throwing scorer (e.g. a judge whose model call failed) scores 0
-          // with the error surfaced, rather than aborting the whole run.
-          const message = err instanceof Error ? err.message : String(err)
-          scores.push({ name: scorer.name, score: 0, weight: scorer.weight ?? 1, reason: `scorer threw: ${message}` })
-        }
-      }
-
-      let score = 0
-
-      if (scores.length > 0) {
-        const sum = scores.reduce((sum, s) => sum + s.score * s.weight, 0)
-        const weight = scores.reduce((sum, s) => sum + s.weight, 0)
-
-        score = sum / weight
-      }
-
-      results.push({ tags, output, score, scores, usage, latencyMs })
-    }
-
+    const results = await pMap(cases, c => runCase(c, model, task, scorers), { concurrency })
     byModel[model] = aggregate(results)
   }
 
   return { name, byModel }
 }
+
+/** Run the task on one case under one model and score the output. */
+async function runCase<I, O, E>(
+  c: Case<I, E>,
+  model: string,
+  task: EvalConfig<I, O, E>['task'],
+  scorers: EvalConfig<I, O, E>['scorers'],
+): Promise<CaseResult<O>> {
+  const tags = c.tags ?? []
+  const input = c.input
+  const expected = c.expected
+
+  const usage: { task: Usage; judge: Usage } = {
+    task: { inputTokens: 0, outputTokens: 0 },
+    judge: { inputTokens: 0, outputTokens: 0 },
+  }
+
+  const start = performance.now()
+  const [error, result] = await to(
+    task(input, {
+      model,
+      report: usg => report(usage.task, usg),
+    }),
+  )
+  const latencyMs = performance.now() - start
+
+  if (error) {
+    return {
+      tags,
+      usage,
+      latencyMs,
+      score: 0,
+      scores: [],
+      output: null,
+    }
+  }
+
+  const output = result!
+  const scores: CaseResult<O>['scores'] = []
+
+  for (const scorer of scorers) {
+    try {
+      const value = await scorer.run({
+        input,
+        output,
+        expected,
+        tags,
+        report: usg => report(usage.judge, usg),
+      })
+
+      if (value === null) continue
+      const normalized = typeof value === 'number' ? { score: value, reason: '' } : value
+
+      scores.push({
+        name: scorer.name,
+        score: normalized.score,
+        weight: scorer.weight ?? 1,
+        reason: normalized.reason ?? '',
+      })
+    } catch (err) {
+      // A throwing scorer (e.g. a judge whose model call failed) scores 0
+      // with the error surfaced, rather than aborting the whole run.
+      const message = err instanceof Error ? err.message : String(err)
+      scores.push({ name: scorer.name, score: 0, weight: scorer.weight ?? 1, reason: `scorer threw: ${message}` })
+    }
+  }
+
+  let score = 0
+
+  if (scores.length > 0) {
+    const sum = scores.reduce((sum, s) => sum + s.score * s.weight, 0)
+    const weight = scores.reduce((sum, s) => sum + s.weight, 0)
+
+    score = sum / weight
+  }
+
+  return { tags, output, score, scores, usage, latencyMs }
+}

src/types.ts

@@ -64,7 +64,7 @@ export type EvalConfig<I, O, E> = {
   models: string[];
   /** The system under test. */
   task: Task<I, O>;
-  /** Max concurrent cases. Not yet honored; cases run serially. */
+  /** Max concurrent cases within a model. Default 1 (serial). */
   concurrency?: number;
   /** Path to a baseline report for gating. Not yet honored. */
   baseline?: string;

test/evaluate.test.ts

@@ -259,6 +259,56 @@ describe('evaluate', () => {
     expect(calls).toBe(2) // ...yet the task still ran twice — duplicate ids should arguably dedupe or throw
   })
 
+  describe('concurrency', () => {
+    /** A task that tracks how many invocations are in flight at once. */
+    function trackingTask() {
+      let inFlight = 0
+      let maxInFlight = 0
+      const task: Task<Input, Output> = async (input, ctx) => {
+        inFlight++
+        maxInFlight = Math.max(maxInFlight, inFlight)
+        await new Promise(resolve => setTimeout(resolve, 5))
+        inFlight--
+        return { y: input.x * 2, model: ctx.model }
+      }
+      return { task, max: () => maxInFlight }
+    }
+
+    const fourCases: Case<Input, Expected>[] = [1, 2, 3, 4].map(x => ({ input: { x }, expected: { y: x * 2 } }))
+
+    it('runs cases serially by default', async () => {
+      const { task, max } = trackingTask()
+      await evaluate('serial', config({ task, data: fourCases }))
+      expect(max()).toBe(1)
+    })
+
+    it('runs up to `concurrency` cases at once within a model', async () => {
+      const { task, max } = trackingTask()
+      await evaluate('pool', config({ task, data: fourCases, concurrency: 2 }))
+      expect(max()).toBe(2)
+    })
+
+    it('keeps results in case order regardless of completion order', async () => {
+      const slowFirst: Task<Input, Output> = async (input, ctx) => {
+        // the first case finishes last
+        await new Promise(resolve => setTimeout(resolve, input.x === 1 ? 20 : 1))
+        return { y: input.x * 2, model: ctx.model }
+      }
+      const report = await evaluate('order', config({ task: slowFirst, data: fourCases, concurrency: 4 }))
+      expect(report.byModel.m1?.cases.map(c => c.output?.y)).toEqual([2, 4, 6, 8])
+    })
+
+    it('scores concurrent cases exactly as serial runs do', async () => {
+      const report = await evaluate('same', config({ data: fourCases, concurrency: 3 }))
+      expect(report.byModel.m1?.overall).toBe(1)
+      expect(report.byModel.m1?.cases).toHaveLength(4)
+    })
+
+    it.each([0, -1, 1.5])('rejects a non-positive or fractional concurrency (%s)', async value => {
+      await expect(evaluate('bad', config({ concurrency: value }))).rejects.toThrow('concurrency must be a positive integer')
+    })
+  })
+
   it('rejects a scorer with a non-positive weight', async () => {
     const zero: Scorer<Input, Output, Expected> = { name: 'zero', weight: 0, run: () => 1 }
     await expect(evaluate('w0', config({ scorers: [zero], data: [cases[0]!] }))).rejects.toThrow('weight must be > 0')