From c759f0a1a8a07224465c78e5f5e726025b38bc6d Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 5 Jul 2026 15:49:09 +0000 Subject: [PATCH 1/2] feat: honor the concurrency option in evaluate The concurrency option was accepted in EvalConfig but silently ignored; cases always ran serially. Now up to config.concurrency cases run at once within each model (default 1, preserving the old serial behavior); models are still swept one at a time so per-model rate limits stay predictable. - add mapConcurrent to utils: an order-preserving worker-pool map - extract the per-case body of evaluate into runCase so it can be scheduled through the pool - validate concurrency as a positive integer - update the types comment and README row that said 'not yet honored' Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_015XMPpi4PX2bWwAoMLaZZ4k --- readme.md | 2 +- src/evaluate.ts | 165 ++++++++++++++++++++++-------------------- src/types.ts | 2 +- src/utils.ts | 20 +++++ test/evaluate.test.ts | 50 +++++++++++++ test/utils.test.ts | 42 ++++++++++- 6 files changed, 199 insertions(+), 82 deletions(-) diff --git a/readme.md b/readme.md index 2949d4d..6a67255 100644 --- a/readme.md +++ b/readme.md @@ -95,7 +95,7 @@ evaluate(name: string, config: EvalConfig): Promise[]` | one or more scorers | | `models` | `string[]` | models to sweep; at least one | | `task` | `Task` | 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?)` diff --git a/src/evaluate.ts b/src/evaluate.ts index 93a77fc..20966f1 100644 --- a/src/evaluate.ts +++ b/src/evaluate.ts @@ -1,13 +1,14 @@ import assert from 'node:assert' -import type { CaseResult, EvalConfig, EvalReport, Usage } from './types.js' -import { to } from './utils.js' +import type { Case, CaseResult, EvalConfig, EvalReport, Usage } from './types.js' +import { mapConcurrent, 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 +31,93 @@ export async function evaluate(name: string, config: EvalConfig 0, 'concurrency must be a positive integer') + const byModel: EvalReport['byModel'] = {} for (const model of models) { - const results: CaseResult[] = [] - - 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['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 mapConcurrent(cases, concurrency, c => runCase(c, model, task, scorers)) byModel[model] = aggregate(results) } return { name, byModel } } + +/** Run the task on one case under one model and score the output. */ +async function runCase( + c: Case, + model: string, + task: EvalConfig['task'], + scorers: EvalConfig['scorers'], +): Promise> { + 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['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 } +} diff --git a/src/types.ts b/src/types.ts index 47e15c0..865d467 100644 --- a/src/types.ts +++ b/src/types.ts @@ -64,7 +64,7 @@ export type EvalConfig = { models: string[]; /** The system under test. */ task: Task; - /** 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; diff --git a/src/utils.ts b/src/utils.ts index 929bd2c..ea7cfea 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -21,6 +21,26 @@ export function percentile(values: number[], p: number): number { return sorted[idx] ?? 0 } +/** + * Map `items` through an async `fn`, running at most `limit` calls at a time. + * Results keep the order of `items`. A rejection from `fn` propagates and stops + * scheduling new items (in-flight items still settle). + */ +export async function mapConcurrent(items: T[], limit: number, fn: (item: T, index: number) => Promise): Promise { + const results = new Array(items.length) + let next = 0 + + const worker = async () => { + while (next < items.length) { + const i = next++ + results[i] = await fn(items[i]!, i) + } + } + + await Promise.all(Array.from({ length: Math.max(1, Math.min(limit, items.length)) }, worker)) + return results +} + /** Escape the five characters that are unsafe in HTML text and attributes. */ export function escapeHtml(value: string): string { return value diff --git a/test/evaluate.test.ts b/test/evaluate.test.ts index 90c2af7..042c0bd 100644 --- a/test/evaluate.test.ts +++ b/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 = 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[] = [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 = 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 = { name: 'zero', weight: 0, run: () => 1 } await expect(evaluate('w0', config({ scorers: [zero], data: [cases[0]!] }))).rejects.toThrow('weight must be > 0') diff --git a/test/utils.test.ts b/test/utils.test.ts index 5bf0bee..65b000f 100644 --- a/test/utils.test.ts +++ b/test/utils.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from 'vitest' -import { escapeHtml, ms, pct, percentile, to, usd } from '../src/utils.js' +import { escapeHtml, mapConcurrent, ms, pct, percentile, to, usd } from '../src/utils.js' describe('to', () => { it('returns [null, result] on success', async () => { @@ -23,6 +23,46 @@ describe('to', () => { }) }) +describe('mapConcurrent', () => { + it('maps every item and preserves input order', async () => { + const result = await mapConcurrent([3, 1, 2], 2, async n => { + await new Promise(resolve => setTimeout(resolve, n)) + return n * 10 + }) + expect(result).toEqual([30, 10, 20]) + }) + + it('never exceeds the concurrency limit', async () => { + let inFlight = 0 + let max = 0 + await mapConcurrent([1, 2, 3, 4, 5], 2, async () => { + inFlight++ + max = Math.max(max, inFlight) + await new Promise(resolve => setTimeout(resolve, 5)) + inFlight-- + }) + expect(max).toBe(2) + }) + + it('passes the item index to the callback', async () => { + const result = await mapConcurrent(['a', 'b'], 1, async (item, i) => `${item}${i}`) + expect(result).toEqual(['a0', 'b1']) + }) + + it('resolves to [] for an empty list', async () => { + await expect(mapConcurrent([], 4, async () => 1)).resolves.toEqual([]) + }) + + it('propagates a rejection from the callback', async () => { + await expect( + mapConcurrent([1, 2], 2, async n => { + if (n === 2) throw new Error('boom') + return n + }), + ).rejects.toThrow('boom') + }) +}) + describe('percentile', () => { it('returns 0 for an empty list', () => { expect(percentile([], 50)).toBe(0) From b2444551d9cee9d831bed271c0651e0cb26d33d8 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 7 Jul 2026 00:45:26 +0000 Subject: [PATCH 2/2] refactor: use p-map for case concurrency Swap the hand-rolled mapConcurrent pool for p-map, the battle-tested equivalent. Behavior is unchanged: order-preserving, at most config.concurrency cases in flight, rejections propagate. The evaluate-level concurrency tests stay as the behavior contract; the pool's own unit tests go with it. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_015XMPpi4PX2bWwAoMLaZZ4k --- package.json | 3 +++ pnpm-lock.yaml | 20 ++++++++++---------- src/evaluate.ts | 5 +++-- src/utils.ts | 20 -------------------- test/utils.test.ts | 42 +----------------------------------------- 5 files changed, 17 insertions(+), 73 deletions(-) diff --git a/package.json b/package.json index 46b8c3c..9ca36f1 100644 --- a/package.json +++ b/package.json @@ -51,5 +51,8 @@ "trailingComma": "all", "arrowParens": "avoid", "semi": false + }, + "dependencies": { + "p-map": "^7.0.5" } } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index c63cfa7..18c15d2 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -7,6 +7,10 @@ settings: importers: .: + dependencies: + p-map: + specifier: ^7.0.5 + version: 7.0.5 devDependencies: '@types/node': specifier: ^22.0.0 @@ -235,42 +239,36 @@ packages: engines: {node: ^20.19.0 || >=22.12.0} cpu: [arm64] os: [linux] - libc: [glibc] '@rolldown/binding-linux-arm64-musl@1.1.3': resolution: {integrity: sha512-BO9+oPL8K9poZJBfYPsXNtYjPE5uM3qeehT3aFcW4LITOl+iSqhp0abzjR2nWBUNjIZeKXjAEWBZ64WjNoHd6w==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [arm64] os: [linux] - libc: [musl] '@rolldown/binding-linux-ppc64-gnu@1.1.3': resolution: {integrity: sha512-f3VpLB1vQ0Eo6ecr/6cekLnvYMFF4YBFoVGkfkvPLq1bAkbAwHYQPZKoAmG6OJyTcxxoC+AvezGx/S1obNC0Mw==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [ppc64] os: [linux] - libc: [glibc] '@rolldown/binding-linux-s390x-gnu@1.1.3': resolution: {integrity: sha512-AmurZ26Pqx/RI9N1gzEOCklkKXl927yjfXWUUS0O7Puh8ARM/Ob8qfrD3qnWksScdw6cSrW5PSHE9DyLu7+PtA==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [s390x] os: [linux] - libc: [glibc] '@rolldown/binding-linux-x64-gnu@1.1.3': resolution: {integrity: sha512-JJpqs8bRGITDOdbkNKnlojzBabbOHrqjSvDr0IVsZObE1lBcPjxItUEY9eWIDbxaJ3cGrXPWGfGkIxFijg/URg==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [x64] os: [linux] - libc: [glibc] '@rolldown/binding-linux-x64-musl@1.1.3': resolution: {integrity: sha512-rSJcdjPxzA/by/6/rYs+v+bXU7UjvnbUWz8MJb6kh6+knqB1dCrtHg0uu7C/4haqJvqdkYHQ5IGn+tCH9GLW/g==} engines: {node: ^20.19.0 || >=22.12.0} cpu: [x64] os: [linux] - libc: [musl] '@rolldown/binding-openharmony-arm64@1.1.3': resolution: {integrity: sha512-hQ3/PYkDJICgevvyNcVrihVeqq7k1Pp3VZ9lY+dauAYUJKO+auqApvANhvR1An9BhmqYKvW2Mu1F9u4DXSMLxQ==} @@ -424,28 +422,24 @@ packages: engines: {node: '>= 12.0.0'} cpu: [arm64] os: [linux] - libc: [glibc] lightningcss-linux-arm64-musl@1.32.0: resolution: {integrity: sha512-UpQkoenr4UJEzgVIYpI80lDFvRmPVg6oqboNHfoH4CQIfNA+HOrZ7Mo7KZP02dC6LjghPQJeBsvXhJod/wnIBg==} engines: {node: '>= 12.0.0'} cpu: [arm64] os: [linux] - libc: [musl] lightningcss-linux-x64-gnu@1.32.0: resolution: {integrity: sha512-V7Qr52IhZmdKPVr+Vtw8o+WLsQJYCTd8loIfpDaMRWGUZfBOYEJeyJIkqGIDMZPwPx24pUMfwSxxI8phr/MbOA==} engines: {node: '>= 12.0.0'} cpu: [x64] os: [linux] - libc: [glibc] lightningcss-linux-x64-musl@1.32.0: resolution: {integrity: sha512-bYcLp+Vb0awsiXg/80uCRezCYHNg1/l3mt0gzHnWV9XP1W5sKa5/TCdGWaR/zBM2PeF/HbsQv/j2URNOiVuxWg==} engines: {node: '>= 12.0.0'} cpu: [x64] os: [linux] - libc: [musl] lightningcss-win32-arm64-msvc@1.32.0: resolution: {integrity: sha512-8SbC8BR40pS6baCM8sbtYDSwEVQd4JlFTOlaD3gWGHfThTcABnNDBda6eTZeqbofalIJhFx0qKzgHJmcPTnGdw==} @@ -475,6 +469,10 @@ packages: resolution: {integrity: sha512-9miFgM2OFba7hB+pRgvtV84pYTBaoTHohvmIgiRt6dRIzbwEOIaNaP+dIlGs2fNFoB0SeISs0Jz5WFVRid6Xyg==} engines: {node: '>=12.20.0'} + p-map@7.0.5: + resolution: {integrity: sha512-e8vJF4XdVkzqqSHguEMz41mQO1wKwxKm5ENrUJQUu9kLDCtn83cxbyHZcszr4QC5zEA7WffRRC4gsTecC7J9oA==} + engines: {node: '>=18'} + pathe@2.0.3: resolution: {integrity: sha512-WUjGcAqP1gQacoQe+OBJsFA7Ld4DyXuUIjZ5cc75cLHvJ7dtNsTugphxIADwspS+AraAUePCKrSVtPLFj/F88w==} @@ -955,6 +953,8 @@ snapshots: obug@2.1.3: {} + p-map@7.0.5: {} + pathe@2.0.3: {} picocolors@1.1.1: {} diff --git a/src/evaluate.ts b/src/evaluate.ts index 20966f1..fdf0d41 100644 --- a/src/evaluate.ts +++ b/src/evaluate.ts @@ -1,6 +1,7 @@ import assert from 'node:assert' +import pMap from 'p-map' import type { Case, CaseResult, EvalConfig, EvalReport, Usage } from './types.js' -import { mapConcurrent, to } from './utils.js' +import { to } from './utils.js' import { aggregate } from './aggregate.js' import { report } from './usage.js' @@ -37,7 +38,7 @@ export async function evaluate(name: string, config: EvalConfig['byModel'] = {} for (const model of models) { - const results = await mapConcurrent(cases, concurrency, c => runCase(c, model, task, scorers)) + const results = await pMap(cases, c => runCase(c, model, task, scorers), { concurrency }) byModel[model] = aggregate(results) } diff --git a/src/utils.ts b/src/utils.ts index ea7cfea..929bd2c 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -21,26 +21,6 @@ export function percentile(values: number[], p: number): number { return sorted[idx] ?? 0 } -/** - * Map `items` through an async `fn`, running at most `limit` calls at a time. - * Results keep the order of `items`. A rejection from `fn` propagates and stops - * scheduling new items (in-flight items still settle). - */ -export async function mapConcurrent(items: T[], limit: number, fn: (item: T, index: number) => Promise): Promise { - const results = new Array(items.length) - let next = 0 - - const worker = async () => { - while (next < items.length) { - const i = next++ - results[i] = await fn(items[i]!, i) - } - } - - await Promise.all(Array.from({ length: Math.max(1, Math.min(limit, items.length)) }, worker)) - return results -} - /** Escape the five characters that are unsafe in HTML text and attributes. */ export function escapeHtml(value: string): string { return value diff --git a/test/utils.test.ts b/test/utils.test.ts index 65b000f..5bf0bee 100644 --- a/test/utils.test.ts +++ b/test/utils.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from 'vitest' -import { escapeHtml, mapConcurrent, ms, pct, percentile, to, usd } from '../src/utils.js' +import { escapeHtml, ms, pct, percentile, to, usd } from '../src/utils.js' describe('to', () => { it('returns [null, result] on success', async () => { @@ -23,46 +23,6 @@ describe('to', () => { }) }) -describe('mapConcurrent', () => { - it('maps every item and preserves input order', async () => { - const result = await mapConcurrent([3, 1, 2], 2, async n => { - await new Promise(resolve => setTimeout(resolve, n)) - return n * 10 - }) - expect(result).toEqual([30, 10, 20]) - }) - - it('never exceeds the concurrency limit', async () => { - let inFlight = 0 - let max = 0 - await mapConcurrent([1, 2, 3, 4, 5], 2, async () => { - inFlight++ - max = Math.max(max, inFlight) - await new Promise(resolve => setTimeout(resolve, 5)) - inFlight-- - }) - expect(max).toBe(2) - }) - - it('passes the item index to the callback', async () => { - const result = await mapConcurrent(['a', 'b'], 1, async (item, i) => `${item}${i}`) - expect(result).toEqual(['a0', 'b1']) - }) - - it('resolves to [] for an empty list', async () => { - await expect(mapConcurrent([], 4, async () => 1)).resolves.toEqual([]) - }) - - it('propagates a rejection from the callback', async () => { - await expect( - mapConcurrent([1, 2], 2, async n => { - if (n === 2) throw new Error('boom') - return n - }), - ).rejects.toThrow('boom') - }) -}) - describe('percentile', () => { it('returns 0 for an empty list', () => { expect(percentile([], 50)).toBe(0)