diff --git a/examples/README.md b/examples/README.md index 424071bad..a249f6981 100644 --- a/examples/README.md +++ b/examples/README.md @@ -49,6 +49,7 @@ Focused demonstrations of specific AgentV capabilities. Each example includes it - [execution-metrics](features/execution-metrics/) - Metrics tracking (tokens, cost, latency) - [script-grader-with-llm-calls](features/script-grader-with-llm-calls/) - script graders with provider proxy for LLM calls - [batch-cli](features/batch-cli/) - Batch CLI evaluation +- [provider-owned-batching](features/provider-owned-batching/) - CLI adapter-owned request batching behind normal per-case provider calls - [document-extraction](features/document-extraction/) - Document data extraction - [local-cli](features/local-cli/) - Local CLI providers - [compare](features/compare/) - Baseline comparison diff --git a/examples/features/provider-owned-batching/README.md b/examples/features/provider-owned-batching/README.md new file mode 100644 index 000000000..b37f801ac --- /dev/null +++ b/examples/features/provider-owned-batching/README.md @@ -0,0 +1,63 @@ +# Provider-Owned Batching + +This example shows the replacement pattern for throughput-oriented CLI tools: +AgentV invokes the provider once per eval case, and the provider adapter owns +any batching behind that normal provider boundary. + +The eval and provider catalog do not use `batch_requests`, `provider_batching`, +or runner-owned grouping. `evals/suite.yaml` selects a normal provider label: + +```yaml +providers: + - provider-owned-batch-cli +evaluate_options: + max_concurrency: 3 +``` + +`providers.yaml` then points at a single-case CLI adapter: + +```yaml +- id: cli + label: provider-owned-batch-cli + command: bun run ./scripts/provider-owned-batch-adapter.ts {PROMPT_FILE} {OUTPUT_FILE} {EVAL_ID} + cwd: .. +- id: mock + label: grader +``` + +The `grader` entry satisfies this repository's default grader label; the suite's +checks are deterministic `contains` assertions. + +## How It Works + +1. AgentV starts ordinary per-case CLI provider invocations. +2. Each adapter process appends its request to a provider-owned queue directory. +3. The adapter uses a timeout trigger to flush all queued requests in one + synthetic batch. +4. The batch writes one response file per queued request id. +5. Each waiting adapter process returns its own response to AgentV, preserving + per-case response, error, and trace identity. + +The timeout flush is intentionally inside the adapter. A real provider could +replace it with EOF on a long-lived process, an explicit flush command, or a +future provider lifecycle drain hook without adding runner-level batch config. + +## Run + +From the repository root: + +```bash +bun apps/cli/src/cli.ts validate examples/features/provider-owned-batching/evals/suite.yaml +bun apps/cli/src/cli.ts eval examples/features/provider-owned-batching/evals/suite.yaml \ + --providers examples/features/provider-owned-batching/providers.yaml +``` + +Focused smoke coverage for the provider-owned protocol: + +```bash +bun examples/features/provider-owned-batching/scripts/smoke-provider-owned-batching.ts +``` + +The smoke test starts three adapter invocations concurrently and verifies that +they are flushed as one provider-owned batch while still receiving three +correlated responses. diff --git a/examples/features/provider-owned-batching/evals/suite.yaml b/examples/features/provider-owned-batching/evals/suite.yaml new file mode 100644 index 000000000..20bbb473e --- /dev/null +++ b/examples/features/provider-owned-batching/evals/suite.yaml @@ -0,0 +1,30 @@ +name: provider-owned-batching +description: Demonstrates provider-owned batching behind ordinary per-case CLI provider invocation. +providers: + - provider-owned-batch-cli +tags: + - provider-owned-batching + - cli +prompts: + - "{{ input }}" +evaluate_options: + max_concurrency: 3 +tests: + - id: ticket-clear + assert: + - type: contains + value: decision=CLEAR + vars: + input: "screen support ticket: password reset succeeded" + - id: ticket-review + assert: + - type: contains + value: decision=REVIEW + vars: + input: "screen support ticket: billing dispute requires escalation" + - id: ticket-block + assert: + - type: contains + value: decision=BLOCK + vars: + input: "screen support ticket: fraud alert from blocked account" diff --git a/examples/features/provider-owned-batching/providers.yaml b/examples/features/provider-owned-batching/providers.yaml new file mode 100644 index 000000000..5c70bd369 --- /dev/null +++ b/examples/features/provider-owned-batching/providers.yaml @@ -0,0 +1,7 @@ +- id: cli + label: provider-owned-batch-cli + command: bun run ./scripts/provider-owned-batch-adapter.ts {PROMPT_FILE} {OUTPUT_FILE} {EVAL_ID} + cwd: .. + timeout_seconds: 5 +- id: mock + label: grader diff --git a/examples/features/provider-owned-batching/scripts/provider-owned-batch-adapter.ts b/examples/features/provider-owned-batching/scripts/provider-owned-batch-adapter.ts new file mode 100644 index 000000000..3df484134 --- /dev/null +++ b/examples/features/provider-owned-batching/scripts/provider-owned-batch-adapter.ts @@ -0,0 +1,163 @@ +#!/usr/bin/env bun +import { mkdir, readFile, readdir, rename, rm, writeFile } from 'node:fs/promises'; +import path from 'node:path'; + +type QueuedRequest = { + readonly id: string; + readonly prompt: string; + readonly outputFile: string; + readonly queuedAt: string; +}; + +type BatchResponse = { + readonly id: string; + readonly text?: string; + readonly error?: string; + readonly duration_ms: number; +}; + +const [promptFile, outputFile, requestIdArg] = process.argv.slice(2); +if (!promptFile || !outputFile) { + console.error('Usage: provider-owned-batch-adapter.ts '); + process.exit(2); +} + +const stateRoot = path.resolve( + process.env.PROVIDER_OWNED_BATCH_STATE_DIR ?? '.agentv/provider-owned-batching-state', +); +const queueDir = path.join(stateRoot, 'queue'); +const responseDir = path.join(stateRoot, 'responses'); +const batchLogPath = path.join(stateRoot, 'batches.jsonl'); +const lockDir = path.join(stateRoot, 'flush.lock'); +const flushDelayMs = Number(process.env.PROVIDER_OWNED_BATCH_FLUSH_DELAY_MS ?? 120); +const waitTimeoutMs = Number(process.env.PROVIDER_OWNED_BATCH_WAIT_TIMEOUT_MS ?? 4_000); + +await mkdir(queueDir, { recursive: true }); +await mkdir(responseDir, { recursive: true }); + +const prompt = await readFile(promptFile, 'utf8'); +const requestId = requestIdArg || process.env.AGENTV_EVAL_ID || stableRequestId(prompt, outputFile); +const requestPath = path.join(queueDir, `${requestId}.${process.pid}.json`); +const responsePath = path.join(responseDir, `${requestId}.json`); + +await writeJsonAtomic(requestPath, { + id: requestId, + prompt, + outputFile, + queuedAt: new Date().toISOString(), +} satisfies QueuedRequest); + +await tryFlushBatch(); + +const response = await waitForResponse(responsePath, waitTimeoutMs); +await writeJsonAtomic(outputFile, response); + +async function tryFlushBatch(): Promise { + try { + await mkdir(lockDir); + } catch { + return; + } + + try { + await sleep(flushDelayMs); + const requestFiles = (await readdir(queueDir)) + .filter((entry) => entry.endsWith('.json')) + .sort(); + if (requestFiles.length === 0) { + return; + } + + const requests: QueuedRequest[] = []; + for (const requestFile of requestFiles) { + const filePath = path.join(queueDir, requestFile); + try { + requests.push(JSON.parse(await readFile(filePath, 'utf8')) as QueuedRequest); + } catch { + // A concurrent writer has not finished its atomic rename yet. Leave it + // for the next provider-owned timeout flush. + } + } + + const responses = runSyntheticBatch(requests); + const batchId = `batch-${Date.now()}-${process.pid}`; + await writeFile( + batchLogPath, + `${JSON.stringify({ + batch_id: batchId, + trigger: 'timeout', + request_ids: requests.map((request) => request.id), + request_count: requests.length, + })}\n`, + { flag: 'a' }, + ); + + for (const response of responses) { + await writeJsonAtomic(path.join(responseDir, `${response.id}.json`), response); + } + for (const requestFile of requestFiles) { + await rm(path.join(queueDir, requestFile), { force: true }); + } + } finally { + await rm(lockDir, { recursive: true, force: true }); + } +} + +function runSyntheticBatch(requests: readonly QueuedRequest[]): readonly BatchResponse[] { + const startedAt = Date.now(); + return requests.map((request) => { + const normalized = request.prompt.toLowerCase(); + if (normalized.includes('fraud') || normalized.includes('blocked')) { + return buildResponse(request.id, 'BLOCK', startedAt); + } + if (normalized.includes('dispute') || normalized.includes('escalation')) { + return buildResponse(request.id, 'REVIEW', startedAt); + } + return buildResponse(request.id, 'CLEAR', startedAt); + }); +} + +function buildResponse(id: string, decision: string, startedAt: number): BatchResponse { + return { + id, + text: `decision=${decision}; request_id=${id}`, + duration_ms: Date.now() - startedAt, + }; +} + +async function waitForResponse(filePath: string, timeoutMs: number): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + try { + return JSON.parse(await readFile(filePath, 'utf8')) as BatchResponse; + } catch { + await sleep(25); + await tryFlushBatch(); + } + } + return { + id: requestId, + error: `Timed out waiting for provider-owned batch response for ${requestId}`, + duration_ms: timeoutMs, + }; +} + +async function writeJsonAtomic(filePath: string, value: unknown): Promise { + await mkdir(path.dirname(filePath), { recursive: true }); + const tempPath = `${filePath}.${process.pid}.tmp`; + await writeFile(tempPath, JSON.stringify(value, null, 2), 'utf8'); + await rename(tempPath, filePath); +} + +function stableRequestId(promptText: string, fallback: string): string { + const source = `${promptText}\n${fallback}`; + let hash = 0; + for (let index = 0; index < source.length; index += 1) { + hash = (hash * 31 + source.charCodeAt(index)) >>> 0; + } + return `request-${hash.toString(16)}`; +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} diff --git a/examples/features/provider-owned-batching/scripts/smoke-provider-owned-batching.ts b/examples/features/provider-owned-batching/scripts/smoke-provider-owned-batching.ts new file mode 100644 index 000000000..8fc4f34f3 --- /dev/null +++ b/examples/features/provider-owned-batching/scripts/smoke-provider-owned-batching.ts @@ -0,0 +1,69 @@ +#!/usr/bin/env bun +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; +import os from 'node:os'; +import path from 'node:path'; +import { spawn } from 'bun'; + +const exampleRoot = path.resolve(import.meta.dir, '..'); +const adapter = path.join(exampleRoot, 'scripts/provider-owned-batch-adapter.ts'); +const tempRoot = await mkdtemp(path.join(os.tmpdir(), 'agentv-provider-owned-batching-')); +const stateDir = path.join(tempRoot, 'state'); + +const cases = [ + ['ticket-clear', 'screen support ticket: password reset succeeded', 'decision=CLEAR'], + [ + 'ticket-review', + 'screen support ticket: billing dispute requires escalation', + 'decision=REVIEW', + ], + ['ticket-block', 'screen support ticket: fraud alert from blocked account', 'decision=BLOCK'], +] as const; + +try { + await Promise.all( + cases.map(async ([id, prompt, expected]) => { + const promptFile = path.join(tempRoot, `${id}.prompt.txt`); + const outputFile = path.join(tempRoot, `${id}.output.json`); + await writeFile(promptFile, prompt, 'utf8'); + + const proc = spawn({ + cmd: ['bun', 'run', adapter, promptFile, outputFile], + cwd: exampleRoot, + env: { + ...process.env, + AGENTV_EVAL_ID: id, + PROVIDER_OWNED_BATCH_STATE_DIR: stateDir, + PROVIDER_OWNED_BATCH_FLUSH_DELAY_MS: '150', + }, + stdout: 'pipe', + stderr: 'pipe', + }); + + const exitCode = await proc.exited; + if (exitCode !== 0) { + const stderr = await new Response(proc.stderr).text(); + throw new Error(`${id} exited ${exitCode}: ${stderr}`); + } + + const output = JSON.parse(await readFile(outputFile, 'utf8')) as { text?: string }; + if (!output.text?.includes(expected)) { + throw new Error(`${id} expected ${expected}, got ${JSON.stringify(output)}`); + } + }), + ); + + const batchLines = (await readFile(path.join(stateDir, 'batches.jsonl'), 'utf8')) + .trim() + .split(/\r?\n/) + .filter(Boolean); + const batches = batchLines.map((line) => JSON.parse(line) as { request_count?: number }); + if (batches.length !== 1 || batches[0]?.request_count !== cases.length) { + throw new Error( + `Expected one batch with ${cases.length} requests, got ${batchLines.join('\\n')}`, + ); + } + + console.log(`provider-owned batching smoke passed: ${cases.length} requests flushed once`); +} finally { + await rm(tempRoot, { recursive: true, force: true }); +}