From fb611645201bcfd2f5bdbe075b189cac2177c663 Mon Sep 17 00:00:00 2001 From: JOY <5027251+JOY@users.noreply.github.com> Date: Wed, 4 Nov 2026 12:03:32 +0700 Subject: [PATCH 1/2] feat(importer): review ai points job calls models through the dos.ai gateway JOY decided on 2026-11-04 that the job uses dos.ai, not the Anthropic API (D2 of the AI review points design). The job now sends its batch through the gateway's OpenAI-compatible Files and Batch API with DOS_AI_API_KEY (DOS_AI_BASE_URL overrides the URL). That Batch API serves dos, qwen3.8 and deepseek models only, so the default model is dos (Qwen3.8 27B on DOS GPUs) and qwen3.8-max is the fallback, with the same monthly caps. Caps, kill switch, process-once, evidence checks and quality stops are unchanged. A cut-off or filtered answer (finish_reason length or content_filter) and an errored line are stored as failed. Cached tokens count at the full input price, an upper bound. @anthropic-ai/sdk is removed. The proposal notes the D2 change. Refs #1531 Co-Authored-By: Claude Opus 5.5 --- .../2026-11-03-ai-review-points.html | 14 +- apps/importer/.env.example | 7 +- apps/importer/package.json | 1 - .../shopee/__tests__/review-ai-points.test.ts | 165 +++++++++++-- .../src/shopee/jobs/review-ai-points.ts | 226 +++++++++++++++--- apps/importer/src/shopee/review-ai-prompt.ts | 33 ++- pnpm-lock.yaml | 52 ---- 7 files changed, 377 insertions(+), 121 deletions(-) diff --git a/Docs/Proposals/2026-11-03-ai-review-points.html b/Docs/Proposals/2026-11-03-ai-review-points.html index cca943450..99dcfc3c8 100644 --- a/Docs/Proposals/2026-11-03-ai-review-points.html +++ b/Docs/Proposals/2026-11-03-ai-review-points.html @@ -122,6 +122,15 @@

2.3 Dạng kết quả (structured output, JSON có schema)

criterion_key trong schema là một danh sách liệt kê đúng các khoá của ngành hàng, nên AI không thể trả một khoá lạ.

3. Chọn model

+
Cập nhật 04/11/2026: JOY đổi D2, gọi qua dos.ai. Job gọi model qua cổng dos.ai (Batch API kiểu OpenAI: POST /v1/files, POST /v1/batches), không gọi thẳng Anthropic. Batch API của dos.ai chỉ nhận model của chính DOS và Alibaba (dos, qwen3.8-*, deepseek-*), không nhận Claude, và tính nửa giá. Vì vậy: + + Bảng dưới đây là phương án Claude ban đầu, giữ lại để tham khảo.
@@ -187,7 +196,7 @@

6. Nơi chạy và ranh giới open-core

  • test chạy bằng dữ liệu giả và một client AI giả. Không commit chữ review thật từ sàn vào repo.
    • -
  • DOS/Rate-Services (riêng tư): chỉ thêm một lịch chạy, giữ khoá ANTHROPIC_API_KEY và tài khoản rate_worker. Thứ tự mỗi đêm: crawl xong, rồi review-ai-points, rồi product-points (để gom luôn các dòng mới), rồi review-values. Không chạy trùng giờ crawl. Không có gì của Rate-Services (proxy, scraper, API nội bộ của sàn) đi vào DOS/Rate.
    • +
  • DOS/Rate-Services (riêng tư): chỉ thêm một lịch chạy, giữ khoá DOS_AI_API_KEY của cổng dos.ai (trước là ANTHROPIC_API_KEY, đổi theo D2 ngày 04/11) và tài khoản rate_worker. Thứ tự mỗi đêm: crawl xong, rồi review-ai-points, rồi product-points (để gom luôn các dòng mới), rồi review-values. Không chạy trùng giờ crawl. Không có gì của Rate-Services (proxy, scraper, API nội bộ của sàn) đi vào DOS/Rate.
  • Database:
    • Job đọc bằng rate_worker.
      • @@ -258,7 +267,8 @@

      10. Các quyết định cần JOY

      D1. Model. Haiku 5.5 làm mặc định, chuyển lên Sonnet 5.5 nếu không đạt ngưỡng ở 7.1.

      Đề xuất: duyệt. Chi phí cả hai đều nhỏ; chất lượng quyết định bằng bộ mẫu, không bằng đoán.

      D2. Gọi AI qua đâu. (a) Thẳng Claude API bằng SDK Anthropic, khoá trong Rate-Services; (b) qua cổng dos-ai kiểu OpenAI như importer đang làm. -

      Đề xuất: (a). Cần structured outputs, Batch API và prompt caching. Qua cổng kiểu OpenAI thì mất hoặc phải tự làm lại các tính năng này.

      +

      Đề xuất: (a). Cần structured outputs, Batch API và prompt caching. Qua cổng kiểu OpenAI thì mất hoặc phải tự làm lại các tính năng này.

      +

      Quyết định 04/11/2026: JOY chọn (b), gọi qua dos.ai ("Tao có nguyên 1 cái dos.ai cơ mà?"). Cổng có Batch API và nhận json_schema, nên job giữ được gửi theo lô, chỉ xử lý một lần và ép schema. Model đổi thành dos / qwen3.8-max vì Batch API của cổng không nhận Claude (xem mục 3).

      D3. Điều kiện chọn review. Các ngưỡng ở mục 1: từ 40 ký tự và 8 từ, tiếng Việt hoặc tiếng Anh, qua cổng chặn, chưa có dòng của người viết.

      Đề xuất: duyệt. Chỉnh lại sau lần chạy thử đầu nếu bỏ sót quá nhiều.

      D4. Trọng số dòng AI trong D9 (mục 9). diff --git a/apps/importer/.env.example b/apps/importer/.env.example index 1c33ceca6..01927701e 100644 --- a/apps/importer/.env.example +++ b/apps/importer/.env.example @@ -19,8 +19,13 @@ AI_BASE_URL= AI_MODEL= AI_TIMEOUT_MS=25000 -# Provider-specific keys +# Provider-specific keys. The review AI points job (jobs/review-ai-points.ts) uses only +# DOS_AI_API_KEY (a dos_sk_ key) and DOS_AI_BASE_URL (default https://api.dos.me/v1), +# with REVIEW_AI_POINTS_ENABLED=true and REVIEW_AI_MODEL (dos or qwen3.8-max). DOS_AI_API_KEY= +DOS_AI_BASE_URL= +REVIEW_AI_POINTS_ENABLED= +REVIEW_AI_MODEL= OPENAI_API_KEY= GEMINI_API_KEY= DEEPSEEK_API_KEY= diff --git a/apps/importer/package.json b/apps/importer/package.json index fe8fa907a..143cd899f 100644 --- a/apps/importer/package.json +++ b/apps/importer/package.json @@ -58,7 +58,6 @@ "node": ">=22.0.0" }, "dependencies": { - "@anthropic-ai/sdk": "1.3.0", "@repo/shared-data": "workspace:*", "@supabase/supabase-js": "^2.117.2", "dotenv": "^18.0.8" diff --git a/apps/importer/src/shopee/__tests__/review-ai-points.test.ts b/apps/importer/src/shopee/__tests__/review-ai-points.test.ts index 04c896aa7..bba102692 100644 --- a/apps/importer/src/shopee/__tests__/review-ai-points.test.ts +++ b/apps/importer/src/shopee/__tests__/review-ai-points.test.ts @@ -17,6 +17,8 @@ import { aiCandidates, buildRequests, collectBatch, + dosAiReviewAiClient, + fromDosAiOutputLine, qualityStop, runReviewAiPointsJob, type AiMarkerRow, @@ -169,24 +171,25 @@ function testPrompt() { too_many: 1, }); - // Prices: Haiku 5.5 at batch prices, cache reads and writes included. + // Prices: dos.ai list prices at the Batch API's half, cached tokens at the full input + // price (an upper bound). assert.equal( - batchCostUsd('claude-haiku-5-5', { + batchCostUsd('dos', { input_tokens: 1_000_000, output_tokens: 1_000_000, cache_read_input_tokens: 1_000_000, cache_creation_input_tokens: 1_000_000, }).toFixed(4), - ((0.1 + 0.5 + 0.01 + 0.125) * 0.5).toFixed(4) + ((0.25 + 1.5 + 0.25 + 0.25) * 0.5).toFixed(4) ); assert.equal( - batchCostUsd('claude-sonnet-5-5', { output_tokens: 1_000_000 }), - 20 * batchCostUsd('claude-haiku-5-5', { output_tokens: 1_000_000 }) + batchCostUsd('qwen3.8-max', { output_tokens: 1_000_000 }), + 4 * batchCostUsd('dos', { output_tokens: 1_000_000 }) ); - const ceiling = reserveUsd('claude-haiku-5-5', [ + const ceiling = reserveUsd('dos', [ { instructions: 'x'.repeat(3000), message: 'y'.repeat(8000) }, ]); - assert.ok(ceiling > 0.0015 && ceiling < 0.003, `one request's ceiling ${ceiling}`); + assert.ok(ceiling > 0.004 && ceiling < 0.007, `one request's ceiling ${ceiling}`); console.log(' ✅ hash, qualifying text, schema, evidence, line checks and prices verified'); } @@ -252,16 +255,23 @@ function testCandidates() { textHash: hash, category: i < 41 ? 'smartphones' : 'laptops', })); - const built = buildRequests(many, 'claude-haiku-5-5'); + const built = buildRequests(many, 'dos'); assert.deepEqual( built.manifest.map(r => `${r.custom_id}:${r.reviews.length}`), ['laptops-0:4', 'smartphones-1:20', 'smartphones-2:20', 'smartphones-3:1'] ); + // One line of an OpenAI-format batch file, as dos.ai's Batch API takes it. const request = built.requests[1]!; - assert.equal(request.params.model, 'claude-haiku-5-5'); - assert.equal(request.params.output_config.effort, 'low'); - assert.equal(request.params.system[0]!.cache_control.type, 'ephemeral'); - const sent = JSON.parse(request.params.messages[0]!.content.split('\n').slice(1).join('\n')) as { + assert.deepEqual([request.method, request.url], ['POST', '/v1/chat/completions']); + assert.equal(request.body.model, 'dos'); + assert.equal(request.body.stream, false); + assert.equal(request.body.messages[0].role, 'system'); + assert.equal(request.body.messages[0].content, reviewAiInstructions('smartphones')); + assert.deepEqual( + [request.body.response_format.type, request.body.response_format.json_schema.strict], + ['json_schema', true] + ); + const sent = JSON.parse(request.body.messages[1].content.split('\n').slice(1).join('\n')) as { ref: number; text: string; }[]; @@ -269,7 +279,7 @@ function testCandidates() { sent.map(r => r.ref), Array.from({ length: 20 }, (_, i) => i + 1) ); - assert.ok(!request.params.messages[0]!.content.includes(id(100)), 'no review id is sent'); + assert.ok(!JSON.stringify(request).includes(id(100)), 'no review id is sent'); console.log(' ✅ the gate, the process-once rule and the requests verified'); } @@ -351,7 +361,7 @@ function testCollect() { ]; const texts = new Map([1, 2, 3, 4, 5, 6, 7].map(n => [id(n), LONG])); const collected = collectBatch( - { id: 'msgbatch_test0001', model: 'claude-haiku-5-5', prompt_version: 'v1', manifest }, + { id: 'msgbatch_test0001', model: 'dos', prompt_version: 'v1', manifest }, results, texts ); @@ -370,12 +380,12 @@ function testCollect() { assert.equal(collected.usage.input_tokens, 16000, 'cut-off and refused requests are paid for'); assert.equal( collected.costUsd, - batchCostUsd('claude-haiku-5-5', { input_tokens: 16000, output_tokens: 3600 }) + batchCostUsd('dos', { input_tokens: 16000, output_tokens: 3600 }) ); const errored = collectBatch( { id: 'msgbatch_test0002', - model: 'claude-haiku-5-5', + model: 'dos', prompt_version: 'v1', manifest: manifest.slice(1, 2), }, @@ -476,7 +486,7 @@ async function testJob() { assert.equal(batch.status, 'stored'); assert.equal( Number(batch.cost_usd), - Number(batchCostUsd('claude-haiku-5-5', { input_tokens: 4000, output_tokens: 900 }).toFixed(4)) + Number(batchCostUsd('dos', { input_tokens: 4000, output_tokens: 900 }).toFixed(4)) ); assert.equal(db.stores.flat().length, 3); assert.ok( @@ -500,7 +510,7 @@ async function testJob() { const crashed = fakeSupabase(seed()); crashed.tables.review_ai_batches!.push({ id: 'msgbatch_crashed01', - model: 'claude-haiku-5-5', + model: 'dos', prompt_version: REVIEW_AI_PROMPT_VERSION, status: 'submitted', manifest: [ @@ -606,7 +616,7 @@ async function testJob() { const capped = fakeSupabase(seed()); capped.tables.review_ai_batches!.push({ id: 'msgbatch_spent0001', - model: 'claude-haiku-5-5', + model: 'dos', prompt_version: 'v0', status: 'stored', manifest: [], @@ -617,7 +627,7 @@ async function testJob() { }); capped.tables.review_ai_batches!.push({ id: 'msgbatch_lastmonth', - model: 'claude-haiku-5-5', + model: 'dos', prompt_version: 'v0', status: 'stored', manifest: [], @@ -638,7 +648,7 @@ async function testJob() { assert.equal(ai.calls.length, 0); const sonnet = await runReviewAiPointsJob(capped.client, { enabled: true, - model: 'claude-sonnet-5-5', + model: 'qwen3.8-max', now, }); assert.equal(sonnet.capped, false, 'each model has its own cap'); @@ -697,8 +707,118 @@ async function testJob() { ); } +async function testDosAiClient() { + // Output lines: a normal answer, a cut-off one, a filtered one, an error, a 400. + const ok = (finish: string) => ({ + status_code: 200, + body: { + choices: [{ message: { content: '{"reviews":[]}' }, finish_reason: finish }], + usage: { + prompt_tokens: 1000, + completion_tokens: 50, + prompt_tokens_details: { cached_tokens: 600 }, + }, + }, + }); + const read = (line: object) => fromDosAiOutputLine(line); + const stopped = read({ custom_id: 'a', response: ok('stop') }); + assert.equal(stopped?.result.type, 'succeeded'); + if (stopped?.result.type === 'succeeded') { + assert.equal(stopped.result.message.stop_reason, 'end_turn'); + assert.deepEqual(stopped.result.message.usage, { + input_tokens: 400, + output_tokens: 50, + cache_read_input_tokens: 600, + cache_creation_input_tokens: 0, + }); + } + const reason = (line: object) => { + const r = read(line); + return r?.result.type === 'succeeded' ? r.result.message.stop_reason : r?.result.type; + }; + assert.equal(reason({ custom_id: 'b', response: ok('length') }), 'max_tokens'); + assert.equal(reason({ custom_id: 'c', response: ok('content_filter') }), 'refusal'); + assert.equal( + reason({ custom_id: 'd', error: { code: 'model_sunset', message: 'x' } }), + 'errored' + ); + assert.equal(reason({ custom_id: 'e', response: { status_code: 400, body: {} } }), 'errored'); + assert.equal(read({ response: ok('stop') }), null, 'a line with no custom_id is skipped'); + + // The client against a fake gateway: upload, create, status, results, cancel, auth, retry, 402. + const calls: string[] = []; + let busy = 1; + const gateway = (async (url: string | URL | Request, init?: RequestInit) => { + const path = String(url).replace('https://beta-api.dos.ai/v1', ''); + const method = init?.method ?? 'GET'; + calls.push(`${method} ${path}`); + assert.equal( + (init?.headers as Record).Authorization, + 'Bearer dos_sk_test', + 'the key as a Bearer token' + ); + const reply = (status: number, body: unknown) => + new Response(typeof body === 'string' ? body : JSON.stringify(body), { status }); + if (path === '/files' && method === 'POST') { + const form = init!.body as FormData; + assert.equal(form.get('purpose'), 'batch'); + const lines = (await (form.get('file') as Blob).text()).trim().split('\n'); + assert.equal(JSON.parse(lines[0]!).url, '/v1/chat/completions'); + return reply(200, { id: 'file_in' }); + } + if (path === '/batches' && method === 'POST') { + const body = JSON.parse(String(init!.body)) as Record; + assert.deepEqual(body, { + input_file_id: 'file_in', + endpoint: '/v1/chat/completions', + completion_window: '24h', + }); + return reply(200, { id: 'batch_abc123' }); + } + if (path === '/batches/batch_abc123') { + if (busy-- > 0) return reply(503, 'busy'); + return reply(200, { status: 'completed', output_file_id: 'file_out' }); + } + if (path === '/files/file_out/content') { + return reply( + 200, + [ + JSON.stringify({ custom_id: 'smartphones-0', response: ok('stop') }), + 'not json', + JSON.stringify({ custom_id: 'smartphones-1', error: { message: 'x' } }), + ].join('\n') + ); + } + if (path === '/batches/batch_abc123/cancel') return reply(200, { status: 'cancelling' }); + if (path === '/batches/batch_broke') return reply(402, { error: 'insufficient credits' }); + return reply(404, 'no route'); + }) as typeof fetch; + const client = dosAiReviewAiClient('dos_sk_test', 'https://beta-api.dos.ai/v1/', gateway); + const built = buildRequests( + [{ id: id(1), text: LONG, textHash: reviewTextHash(LONG), category: 'smartphones' }], + 'dos' + ); + assert.deepEqual(await client.createBatch(built.requests), { id: 'batch_abc123' }); + assert.equal(await client.batchStatus('batch_abc123'), 'ended', 'a 503 is retried'); + const out: BatchResult[] = []; + for await (const r of client.results('batch_abc123')) out.push(r); + assert.deepEqual( + out.map(r => `${r.custom_id}:${r.result.type}`), + ['smartphones-0:succeeded', 'smartphones-1:errored'] + ); + await client.cancelBatch('batch_abc123'); + await assert.rejects(client.batchStatus('batch_broke'), /no credits left/); + await assert.rejects(client.batchStatus('batch_missing'), /404/); + assert.deepEqual(calls.slice(0, 3), [ + 'POST /files', + 'POST /batches', + 'GET /batches/batch_abc123', + ]); + console.log(' ✅ dos.ai batch client: upload, create, status, results, cancel, retry and 402'); +} + function refsOf(request: BatchRequest): number[] { - const content = request.params.messages[0]!.content; + const content = request.body.messages[1].content; return (JSON.parse(content.slice(content.indexOf('\n') + 1)) as { ref: number }[]).map( r => r.ref ); @@ -855,6 +975,7 @@ async function run() { testCandidates(); testCollect(); await testJob(); + await testDosAiClient(); } run().catch(err => { diff --git a/apps/importer/src/shopee/jobs/review-ai-points.ts b/apps/importer/src/shopee/jobs/review-ai-points.ts index a04677923..47abae6bd 100644 --- a/apps/importer/src/shopee/jobs/review-ai-points.ts +++ b/apps/importer/src/shopee/jobs/review-ai-points.ts @@ -6,15 +6,17 @@ * [--apply] [--purge ] [--max ] [--wait-minutes ] * * Reads crawled user reviews, picks those that qualify (design section 1, D3), - * sends them through the Message Batches API with a structured output schema - * per category (section 2), checks every line it gets back (section 7.2) and + * sends them through the dos.ai gateway's Batch API (OpenAI format, D2 as JOY + * decided on 2026-11-04) with a structured output schema per category + * (section 2), checks every line it gets back (section 7.2) and * stores what passes as rate.review_points method 'ai' through * rate.store_review_ai_points, which writes the lines and the process-once * marker (rate.review_ai_extractions) in one transaction. * * Dry run by default: it reads, plans and logs counts and the cost ceiling, - * and never calls a model (no API key needed). `--apply` sends one batch, - * waits for it (at most --wait-minutes, default 60) and stores the results. + * and never calls a model (no key needed). `--apply` sends one batch with the + * key in DOS_AI_API_KEY (DOS_AI_BASE_URL overrides the gateway URL), waits for + * it (at most --wait-minutes, default 60) and stores the results. * It signs in as the rate_worker user (RATE_WORKER_EMAIL and * RATE_WORKER_PASSWORD with the publishable key), which reads the review * tables, writes the batch record and writes lines only through the writer. @@ -22,9 +24,10 @@ * Switches and limits (section 8, D5): * - REVIEW_AI_POINTS_ENABLED must be 'true', or the job exits before any * model call; with --apply it also cancels a batch still running. - * - REVIEW_AI_MODEL: claude-haiku-5-5 (default) or claude-sonnet-5-5 (D1). + * - REVIEW_AI_MODEL: dos (default, Qwen3.8 27B on DOS's GPUs) or + * qwen3.8-max, the models the gateway's Batch API serves (D1's shape). * - At most 2000 reviews a run (--max lowers it). - * - A monthly cap per model ($5 Haiku, $15 Sonnet). The month's spend is the + * - A monthly cap per model ($5 dos, $15 qwen3.8-max). The month's spend is the * reported cost of its batches, or their reserved ceiling while running, * read from rate.review_ai_batches. A batch whose ceiling would pass the * cap is not sent, and the CLI exits with code 3. @@ -97,22 +100,26 @@ export const SHARE_JUMP_STOP = 0.3; /** ...when both batches have at least this many reviews. */ export const SHARE_JUMP_MIN_REVIEWS = 50; const DEADLOCK_ATTEMPTS = 4; -/** The Batch API keeps results 29 days; past 28 a batch is treated as gone. */ +/** A batch whose results cannot be read after 28 days is treated as gone. */ const RESULTS_KEPT_MS = 28 * 24 * 60 * 60 * 1000; const PROMPT_VERSION_RE = /^[a-z0-9][a-z0-9._-]{0,31}$/; // --- The model client: the Batch API as the job uses it, so tests pass a fake. ------------ +/** One line of a dos.ai (OpenAI-format) batch input file. */ export interface BatchRequest { custom_id: string; - params: { + method: 'POST'; + url: '/v1/chat/completions'; + body: { model: ReviewAiModel; max_tokens: number; - system: { type: 'text'; text: string; cache_control: { type: 'ephemeral' } }[]; - messages: { role: 'user'; content: string }[]; - output_config: { - effort: 'low'; - format: { type: 'json_schema'; schema: Record }; + temperature: number; + stream: false; + messages: [{ role: 'system'; content: string }, { role: 'user'; content: string }]; + response_format: { + type: 'json_schema'; + json_schema: { name: string; strict: true; schema: Record }; }; }; } @@ -145,29 +152,167 @@ export interface ReviewAiClient { cancelBatch(id: string): Promise; } -/** The real client: the official Anthropic SDK, the key from ANTHROPIC_API_KEY. */ -export async function anthropicReviewAiClient(apiKey: string): Promise { - const { default: Anthropic } = await import('@anthropic-ai/sdk'); - const client = new Anthropic({ apiKey }); +/** The dos.ai gateway's default base URL, as the importer's other AI clients use it. */ +export const DOS_AI_DEFAULT_BASE_URL = 'https://api.dos.me/v1'; +const DOS_AI_TIMEOUT_MS = 60_000; +const DOS_AI_ATTEMPTS = 3; + +/** One line of a dos.ai batch output file (OpenAI batch format). */ +interface DosAiOutputLine { + custom_id?: string; + response?: { + status_code?: number; + body?: { + choices?: { message?: { content?: string | null }; finish_reason?: string | null }[]; + usage?: { + prompt_tokens?: number; + completion_tokens?: number; + prompt_tokens_details?: { cached_tokens?: number; cache_write_tokens?: number } | null; + }; + }; + } | null; + error?: { code?: string; message?: string } | null; +} + +/** + * A dos.ai output line as the job reads results: a 200 response is a + * message whose stop reason is the finish reason ('stop' ends normally, + * 'length' is cut off, 'content_filter' is refused); anything else errored. + * Cached prompt tokens are split out of prompt_tokens. + */ +export function fromDosAiOutputLine(line: DosAiOutputLine): BatchResult | null { + if (typeof line.custom_id !== 'string') return null; + const body = line.response?.body; + if (line.error || line.response?.status_code !== 200 || !body) { + return { custom_id: line.custom_id, result: { type: 'errored' } }; + } + const choice = body.choices?.[0]; + const finish = choice?.finish_reason ?? null; + const n = (v: unknown) => (typeof v === 'number' && Number.isFinite(v) && v > 0 ? v : 0); + const cachedRead = n(body.usage?.prompt_tokens_details?.cached_tokens); + const cachedWrite = n(body.usage?.prompt_tokens_details?.cache_write_tokens); + return { + custom_id: line.custom_id, + result: { + type: 'succeeded', + message: { + stop_reason: + finish === 'stop' + ? 'end_turn' + : finish === 'length' + ? 'max_tokens' + : finish === 'content_filter' + ? 'refusal' + : finish, + content: [{ type: 'text', text: choice?.message?.content ?? '' }], + usage: { + input_tokens: Math.max(0, n(body.usage?.prompt_tokens) - cachedRead - cachedWrite), + output_tokens: n(body.usage?.completion_tokens), + cache_read_input_tokens: cachedRead, + cache_creation_input_tokens: cachedWrite, + }, + }, + }, + }; +} + +/** + * The real client: the dos.ai gateway's OpenAI-compatible Files and Batch API + * (POST /files, POST /batches, GET /batches/:id, POST /batches/:id/cancel, + * GET /files/:id/content), with the key in DOS_AI_API_KEY as a Bearer token. + * A 429 or 5xx is retried; a 402 (no credits left) stops the run. + */ +export function dosAiReviewAiClient( + apiKey: string, + baseUrl: string = DOS_AI_DEFAULT_BASE_URL, + fetchImpl: typeof fetch = fetch +): ReviewAiClient { + const base = baseUrl.replace(/\/+$/, ''); + const call = async (path: string, init: RequestInit = {}): Promise => { + for (let attempt = 1; ; attempt++) { + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), DOS_AI_TIMEOUT_MS); + let response: Response | null = null; + try { + response = await fetchImpl(`${base}${path}`, { + ...init, + headers: { Authorization: `Bearer ${apiKey}`, ...(init.headers ?? {}) }, + signal: controller.signal, + }); + } catch (err) { + if (attempt >= DOS_AI_ATTEMPTS) throw err; + } finally { + clearTimeout(timer); + } + if (response) { + if (response.ok) return response; + if (response.status === 402) { + throw new Error('dos.ai refused the request: no credits left on the key (402).'); + } + if ((response.status !== 429 && response.status < 500) || attempt >= DOS_AI_ATTEMPTS) { + // The body names the reason; it holds no review text of ours. + const detail = (await response.text().catch(() => '')).slice(0, 300); + throw new Error(`dos.ai ${init.method ?? 'GET'} ${path}: ${response.status} ${detail}`); + } + } + await new Promise(resolve => setTimeout(resolve, 1000 * 2 ** (attempt - 1))); + } + }; + const json = async (path: string, init?: RequestInit) => + (await (await call(path, init)).json()) as T; return { async createBatch(requests) { - const batch = await client.messages.batches.create({ - requests: requests as unknown as Parameters< - typeof client.messages.batches.create - >[0]['requests'], + const form = new FormData(); + form.append('purpose', 'batch'); + form.append( + 'file', + new Blob([requests.map(r => JSON.stringify(r)).join('\n') + '\n'], { + type: 'application/jsonl', + }), + 'review-ai-points.jsonl' + ); + const file = await json<{ id: string }>('/files', { method: 'POST', body: form }); + const batch = await json<{ id: string }>('/batches', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + input_file_id: file.id, + endpoint: '/v1/chat/completions', + completion_window: '24h', + }), }); return { id: batch.id }; }, async batchStatus(id) { - return (await client.messages.batches.retrieve(id)).processing_status; + const batch = await json<{ status: string }>(`/batches/${encodeURIComponent(id)}`); + if (batch.status === 'cancelling') return 'canceling'; + return ['completed', 'failed', 'cancelled', 'expired'].includes(batch.status) + ? 'ended' + : 'in_progress'; }, async *results(id) { - for await (const entry of await client.messages.batches.results(id)) { - yield entry as unknown as BatchResult; + const batch = await json<{ output_file_id?: string | null }>( + `/batches/${encodeURIComponent(id)}` + ); + // A failed or canceled batch may have no output: its lines come back as failed rows. + if (!batch.output_file_id) return; + const text = await ( + await call(`/files/${encodeURIComponent(batch.output_file_id)}/content`) + ).text(); + for (const raw of text.split('\n')) { + if (!raw.trim()) continue; + let line: DosAiOutputLine; + try { + line = JSON.parse(raw) as DosAiOutputLine; + } catch { + continue; + } + const result = fromDosAiOutputLine(line); + if (result) yield result; } }, async cancelBatch(id) { - await client.messages.batches.cancel(id); + await call(`/batches/${encodeURIComponent(id)}/cancel`, { method: 'POST' }); }, }; } @@ -384,17 +529,26 @@ export function buildRequests( const refs = part.map((c, j) => ({ ref: j + 1, candidate: c })); requests.push({ custom_id, - params: { + method: 'POST', + url: '/v1/chat/completions', + body: { model, max_tokens: REQUEST_MAX_TOKENS, - system: [{ type: 'text', text: system, cache_control: { type: 'ephemeral' } }], + temperature: 0, + stream: false, + // The instructions first and the same for a category: a shared prefix the + // gateway's backends can cache on their own. messages: [ + { role: 'system', content: system }, { role: 'user', content: reviewAiUserMessage(refs.map(r => ({ ref: r.ref, text: r.candidate.text }))), }, ], - output_config: { effort: 'low', format: { type: 'json_schema', schema } }, + response_format: { + type: 'json_schema', + json_schema: { name: 'review_points', strict: true, schema }, + }, }, }); instructions.push(system); @@ -673,7 +827,7 @@ export async function runReviewAiPointsJob( purged: null, }; const needAi = (): ReviewAiClient => { - if (!options.ai) throw new Error('ANTHROPIC_API_KEY is needed to send or collect a batch.'); + if (!options.ai) throw new Error('DOS_AI_API_KEY is needed to send or collect a batch.'); return options.ai; }; @@ -879,7 +1033,7 @@ export async function runReviewAiPointsJob( model, requests.map((r, i) => ({ instructions: instructions[i]!, - message: r.params.messages[0]!.content, + message: r.body.messages[1].content, })) ); const refreshed = await fetchByKey( @@ -1055,9 +1209,15 @@ if (isDirectExecution) { throw new Error(`The signed-in user has role '${role}', not rate_worker; refusing to run.`); } const supabase = scopeSupabaseToRate(client); - // The model key is read only for a run that may call the API. - const apiKey = (process.env.ANTHROPIC_API_KEY || '').trim(); - const ai = apply && apiKey ? await anthropicReviewAiClient(apiKey) : null; + // The gateway key is read only for a run that may call it. DOS_AI_BASE_URL alone + // overrides the gateway (beta-api.dos.ai, say); AI_BASE_URL is the other + // importer jobs' and may name another provider. + const apiKey = (process.env.DOS_AI_API_KEY || '').trim(); + const baseUrl = (process.env.DOS_AI_BASE_URL || DOS_AI_DEFAULT_BASE_URL).trim(); + if (!/^https:\/\/[^/\s]+(\/[^\s]*)?$/.test(baseUrl)) { + throw new Error('DOS_AI_BASE_URL must be an https URL.'); + } + const ai = apply && apiKey ? dosAiReviewAiClient(apiKey, baseUrl) : null; logger.info( 'review_ai.target', `Target: ${host} (${apply ? 'apply' : 'dry run'}, ${enabled ? 'enabled' : 'disabled'}, ${modelName})`, diff --git a/apps/importer/src/shopee/review-ai-prompt.ts b/apps/importer/src/shopee/review-ai-prompt.ts index 48ab48551..36991a646 100644 --- a/apps/importer/src/shopee/review-ai-prompt.ts +++ b/apps/importer/src/shopee/review-ai-prompt.ts @@ -29,11 +29,11 @@ import { export const REVIEW_AI_PROMPT_VERSION = 'v1'; export { RATING_CRITERIA_VERSION }; -/** Reviews per request: one category, one cached instruction block. */ +/** Reviews per request: one category, one instruction block (a shared prefix). */ export const REVIEWS_PER_REQUEST = 20; /** Lines per side per review (design 2.2). */ export const AI_POINTS_PER_KIND = 3; -/** Output tokens per request: 20 reviews x ~120 tokens, with room for low-effort thinking. */ +/** Output tokens per request: 20 reviews x ~120 tokens, with room for any reasoning. */ export const REQUEST_MAX_TOKENS = 6000; /** Evidence length, in characters as stored (the writer takes 2 to 300). */ export const EVIDENCE_MAX = 300; @@ -42,18 +42,31 @@ export const EVIDENCE_MAX = 300; export const REVIEW_AI_MIN_CHARS = 40; export const REVIEW_AI_MIN_WORDS = 8; -/** Models the job may use (D1) and their list prices in USD per million tokens. */ +/** + * Models the job may use through the dos.ai gateway's Batch API (D2 as JOY + * decided on 2026-11-04: the job calls dos.ai, not a model vendor), and their + * dos.ai list prices in USD per million tokens (dosai.model_pricing, cents + * per million: dos 25/150, qwen3.8-max 200/600). The gateway's Batch API + * serves only its own and Alibaba models (dos, qwen3.8-*, deepseek-*); dos is + * Qwen3.8 27B on DOS's own GPUs. D1's shape stays: the cheap model by + * default, the stronger one only if the default misses the gold set (7.1). + */ export const REVIEW_AI_MODELS = { - 'claude-haiku-5-5': { input: 0.1, output: 0.5, monthlyCapUsd: 5 }, - 'claude-sonnet-5-5': { input: 2, output: 10, monthlyCapUsd: 15 }, + dos: { input: 0.25, output: 1.5, monthlyCapUsd: 5 }, + 'qwen3.8-max': { input: 2, output: 6, monthlyCapUsd: 15 }, } as const; export type ReviewAiModel = keyof typeof REVIEW_AI_MODELS; -export const DEFAULT_REVIEW_AI_MODEL: ReviewAiModel = 'claude-haiku-5-5'; -/** The Batch API halves every token price. */ +export const DEFAULT_REVIEW_AI_MODEL: ReviewAiModel = 'dos'; +/** The gateway bills Batch API lines at half the synchronous price. */ export const BATCH_DISCOUNT = 0.5; -/** Cache reads cost 0.1x the input price; cache writes 1.25x. */ -export const CACHE_READ_FACTOR = 0.1; -export const CACHE_WRITE_FACTOR = 1.25; +/** + * Cached prompt tokens, read or written, are counted at the full input price: + * the gateway reports them (prompt_tokens_details) but its cached rates per + * model are not published, so the job's figure is an upper bound. The + * gateway's own ledger is the bill. + */ +export const CACHE_READ_FACTOR = 1; +export const CACHE_WRITE_FACTOR = 1; export function isReviewAiModel(value: unknown): value is ReviewAiModel { return typeof value === 'string' && Object.prototype.hasOwnProperty.call(REVIEW_AI_MODELS, value); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 98f965971..f91e3db04 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -130,9 +130,6 @@ importers: apps/importer: dependencies: - '@anthropic-ai/sdk': - specifier: 1.3.0 - version: 1.3.0(zod@4.6.7) '@repo/shared-data': specifier: workspace:* version: link:../../packages/shared-data @@ -350,15 +347,6 @@ packages: resolution: {integrity: sha512-U4+70Pc5ZS9osnCBCE5Jha/ciHM+Yp+CNMNC/7HvYbNRk1Ldd+f7qO65W5qfhu/TCv+/ozljlXXe9Nj8419DMA==} engines: {node: '>=10'} - '@anthropic-ai/sdk@1.3.0': - resolution: {integrity: sha512-k7z9BlLLPxfZH+ov7dS6EnaN/wiZeV8YNT1LZfrpAHIy7oY7pHr4UzirF9f2OlHtFEnRBOX+AZQ91Y5fK+TPTw==} - hasBin: true - peerDependencies: - zod: ^3.25.0 || ^4.0.0 - peerDependenciesMeta: - zod: - optional: true - '@axe-core/playwright@4.13.0': resolution: {integrity: sha512-6YLx+kxXu5GJceG4ozFg+33a2EMTdjYwWGloJ3sb9Kta5pp+ZNS53uxGVog5JetIY8s++P5UrtX+cri+u0VAVg==} peerDependencies: @@ -1651,9 +1639,6 @@ packages: resolution: {integrity: sha512-fCTuZK4QBa+39Oz9l4OGfJfz+GpwCp3AqO7Zch3to99xHPgstVsRFpeQ8LNd2o1Gv8raL2mCFwiaHh7bFSp5DQ==} engines: {node: '>=22'} - '@stablelib/base64@1.0.1': - resolution: {integrity: sha512-1bnPQqSxSuc3Ii6MhBysoWCg58j97aUjuCSZrGSmDxNqtytIi0k8utUenAwTZN4V5mXXYGsVUI9zeBqy+jBOSQ==} - '@supabase/auth-js@2.117.2': resolution: {integrity: sha512-Z3WnGwrphYZubrLbxp5Iv0tLA9A5GhvKzJ/ZXglxqwd2QEH2R4dXRltjVJ8sIn4xEW2BSFIGT0yALVJIgZeDYw==} engines: {node: '>=22.0.0'} @@ -2725,9 +2710,6 @@ packages: fast-levenshtein@2.0.6: resolution: {integrity: sha512-DCXu6Ifhqcks7TZKY3Hxp3y6qphY5SJZmrWMDrKcERSOXWQdMhU9Ig/PYrzyw/ul9jOIyh0N4M0tbC5hodg8dw==} - fast-sha256@1.3.0: - resolution: {integrity: sha512-n11RGP/lrWEFI/bWdygLxhI+pVeo1ZYIVwvvPkW7azl/rOy+F3HYRZ2K5zeE9mmkhQppyv9sQFx0JM9UabnpPQ==} - fast-uri@3.1.9: resolution: {integrity: sha512-hS/0Wt3Ya043oKXs0sIxrMpKQITXhMLIrVAKKXi4gozGC9v7YRWVnXIZzjNPTO02M9BdxLCWqK1yRXgNziVyJQ==} @@ -3157,10 +3139,6 @@ packages: json-parse-even-better-errors@2.3.1: resolution: {integrity: sha512-xyFwyhro/JEof6Ghe2iz2NcXoj2sloNsWr/XsERDK/oiPCfaNhl5ONfp+jQdAZRQQ0IJWNzH9zIZF7li91kh2w==} - json-schema-to-ts@3.1.1: - resolution: {integrity: sha512-+DWg8jCJG2TEnpy7kOm/7/AxaYoaRbjVB4LFZLySZlWn8exGs3A4OLJR966cVvU26N7X9TWxl+Jsw7dzAqKT6g==} - engines: {node: '>=16'} - json-schema-traverse@0.4.1: resolution: {integrity: sha512-xbbCH5dCYU5T8LcEhhuh7HJ88HXuW3qsI3Y0zOZFKfZEHcpWiHU/Jxzk629Brsab/mMiHQti9wMP+845RPe3Vg==} @@ -3860,9 +3838,6 @@ packages: resolution: {integrity: sha512-WjlahMgHmCJpqzU8bIBy4qtsZdU9lRlcZE3Lvyej6t4tuOuv1vk57OW3MBrj6hXBFx/nNoC9MPMTcr5YA7NQbg==} engines: {node: '>=6'} - standardwebhooks@1.1.1: - resolution: {integrity: sha512-bCbX9ZEyFkWPsRz7Bl3NuQUJohmwGSev/yhr7vhaGPlc4AfIrspIRa6cPTBuI1ItmrTDJ4d/S2hCsfe4+vQGnQ==} - stop-iteration-iterator@1.1.0: resolution: {integrity: sha512-eLoXW/DHyl62zxY4SCaIgnRhuMr6ri4juEYARS8E6sCEqzKpOiE521Ucofdx+KnDZl5xmvGYaaKCk5FEOxJCoQ==} engines: {node: '>= 0.4'} @@ -3981,9 +3956,6 @@ packages: resolution: {integrity: sha512-65P7iz6X5yEr1cwcgvQxbbIw7Uk3gOy5dIdtZ4rDveLqhrdJP+Li/Hx6tyK0NEb+2GCyneCMJiGqrADCSNk8sQ==} engines: {node: '>=8.0'} - ts-algebra@2.0.0: - resolution: {integrity: sha512-FPAhNPFMrkwz76P7cdjdmiShwMynZYN6SgOujD1urY4oNm80Ou9oMdmbR45LotcKOXoy7wSmHkRFE6Mxbrhefw==} - ts-api-utils@2.5.0: resolution: {integrity: sha512-OJ/ibxhPlqrMM0UiNHJ/0CKQkoKF243/AEmplt3qpRgkW8VG7IfOS41h7V8TjITqdByHzrjcS/2si+y4lIh8NA==} engines: {node: '>=18.12'} @@ -4186,14 +4158,6 @@ snapshots: '@alloc/quick-lru@5.3.0': {} - '@anthropic-ai/sdk@1.3.0(zod@4.6.7)': - dependencies: - '@opentelemetry/api': 1.9.1 - json-schema-to-ts: 3.1.1 - standardwebhooks: 1.1.1 - optionalDependencies: - zod: 4.6.7 - '@axe-core/playwright@4.13.0(playwright-core@1.64.1)': dependencies: axe-core: 4.13.0 @@ -5261,8 +5225,6 @@ snapshots: '@simple-libs/stream-utils@2.0.0': {} - '@stablelib/base64@1.0.1': {} - '@supabase/auth-js@2.117.2': dependencies: tslib: 2.8.1 @@ -6461,8 +6423,6 @@ snapshots: fast-levenshtein@2.0.6: {} - fast-sha256@1.3.0: {} - fast-uri@3.1.9: {} fastq@1.20.3: @@ -6900,11 +6860,6 @@ snapshots: json-parse-even-better-errors@2.3.1: {} - json-schema-to-ts@3.1.1: - dependencies: - '@babel/runtime': 7.29.7 - ts-algebra: 2.0.0 - json-schema-traverse@0.4.1: {} json-schema-traverse@1.0.0: {} @@ -7635,11 +7590,6 @@ snapshots: dependencies: type-fest: 0.7.1 - standardwebhooks@1.1.1: - dependencies: - '@stablelib/base64': 1.0.1 - fast-sha256: 1.3.0 - stop-iteration-iterator@1.1.0: dependencies: es-errors: 1.3.0 @@ -7777,8 +7727,6 @@ snapshots: dependencies: is-number: 7.0.0 - ts-algebra@2.0.0: {} - ts-api-utils@2.5.0: dependencies: typescript: 6.0.3 From bfe87d665f2d6ad54d1c8316907d4f576dbd0cd9 Mon Sep 17 00:00:00 2001 From: JOY <5027251+JOY@users.noreply.github.com> Date: Wed, 4 Nov 2026 12:24:18 +0700 Subject: [PATCH 2/2] fix(importer): dos.ai batch client never creates a batch twice Review findings on the dos.ai client (#1531): - POST /batches is retried only on 429; a lost answer throws and says a batch may exist, instead of creating and billing a second one. - A batch failed as a whole, one with no output, or one whose output file is gone (404) closes its row and stores nothing, so its reviews are due again without spending an attempt; a line not run for want of credits is unbilled, not failed. - Each call's timeout covers reading the body. - Tests: create-once on a network error and a 502, retry on 429, failed and purged batches, 402 lines, the request body; batch ids renamed. - The proposal notes the dos.ai batch path in sections 4 and 5. Refs #1531 Co-Authored-By: Claude Opus 5.5 --- .../2026-11-03-ai-review-points.html | 2 + .../shopee/__tests__/review-ai-points.test.ts | 141 ++++++++++++--- .../src/shopee/jobs/review-ai-points.ts | 166 +++++++++++++----- 3 files changed, 244 insertions(+), 65 deletions(-) diff --git a/Docs/Proposals/2026-11-03-ai-review-points.html b/Docs/Proposals/2026-11-03-ai-review-points.html index 99dcfc3c8..b5580e0db 100644 --- a/Docs/Proposals/2026-11-03-ai-review-points.html +++ b/Docs/Proposals/2026-11-03-ai-review-points.html @@ -141,6 +141,7 @@

      3. Chọn model

      Gọi qua đâu. Importer hiện gọi AI qua chuẩn OpenAI (biến AI_PROVIDER, cổng dos-ai). Đề xuất bước này gọi thẳng Claude API bằng SDK chính thức của Anthropic, vì cần structured outputs, Batch API và prompt caching. Khoá API chỉ nằm trong secret của Rate-Services. Câu hỏi này để JOY quyết ở D2.

      4. Gửi theo lô và chỉ xử lý một lần

      +

      Từ 04/11/2026 lô được gửi qua Batch API của dos.ai, theo chuẩn OpenAI (file JSONL rồi tạo batch). Dấu đã xử lý, mã lô lưu ngay khi tạo và các luật thử lại bên dưới giữ nguyên. Phần nói về cache và giá Claude chỉ còn để tham khảo.

      • Gom lô: mỗi yêu cầu chứa tối đa 20 review cùng ngành hàng. Phần hướng dẫn của ngành hàng (khoảng 1.500 token) dùng chung, nên được cache: các yêu cầu sau chỉ trả 10% cho phần này. Tối thiểu để cache trên Haiku 5.5 là 512 token, phần hướng dẫn vượt mức này.
      • Batch API: mỗi lần chạy gửi một batch, chờ kết quả, rồi mới ghi. Việc này không cần trả lời ngay.
        • @@ -165,6 +166,7 @@

        4. Gửi theo lô và chỉ xử lý một lần

      5. Ước tính chi phí

      +

      Từ 04/11/2026: với dos qua Batch API của dos.ai (0,125 / 0,75 USD mỗi 1 triệu token sau khi giảm 50%), 1.000 review tốn khoảng 0,12 USD (200.000 token đọc vào, 120.000 token viết ra); kho ~250 review dưới 0,05 USD. Bảng dưới là ước tính cũ cho Claude.

      Giả định:

      • Một review sàn qua điều kiện ở mục 1 dài trung bình khoảng 200 token. Tiếng Việt có dấu tốn token hơn tiếng Anh nên ước rộng tay.
        • diff --git a/apps/importer/src/shopee/__tests__/review-ai-points.test.ts b/apps/importer/src/shopee/__tests__/review-ai-points.test.ts index bba102692..8669eb436 100644 --- a/apps/importer/src/shopee/__tests__/review-ai-points.test.ts +++ b/apps/importer/src/shopee/__tests__/review-ai-points.test.ts @@ -16,6 +16,7 @@ import { import { aiCandidates, buildRequests, + BatchGoneError, collectBatch, dosAiReviewAiClient, fromDosAiOutputLine, @@ -361,7 +362,7 @@ function testCollect() { ]; const texts = new Map([1, 2, 3, 4, 5, 6, 7].map(n => [id(n), LONG])); const collected = collectBatch( - { id: 'msgbatch_test0001', model: 'dos', prompt_version: 'v1', manifest }, + { id: 'batch_test0001', model: 'dos', prompt_version: 'v1', manifest }, results, texts ); @@ -376,7 +377,7 @@ function testCollect() { for (const n of [3, 4, 5, 6, 7]) assert.equal(rows.get(id(n))!.failed, true, `review ${n} failed`); assert.equal(collected.failedReviews, 5); - assert.equal(rows.get(id(1))!.batch_id, 'msgbatch_test0001'); + assert.equal(rows.get(id(1))!.batch_id, 'batch_test0001'); assert.equal(collected.usage.input_tokens, 16000, 'cut-off and refused requests are paid for'); assert.equal( collected.costUsd, @@ -384,7 +385,7 @@ function testCollect() { ); const errored = collectBatch( { - id: 'msgbatch_test0002', + id: 'batch_test0002', model: 'dos', prompt_version: 'v1', manifest: manifest.slice(1, 2), @@ -509,7 +510,7 @@ async function testJob() { // A crash after the batch was created: the next run collects it without sending again. const crashed = fakeSupabase(seed()); crashed.tables.review_ai_batches!.push({ - id: 'msgbatch_crashed01', + id: 'batch_crashed01', model: 'dos', prompt_version: REVIEW_AI_PROMPT_VERSION, status: 'submitted', @@ -527,7 +528,7 @@ async function testJob() { stored_at: null, }); ai.calls.length = 0; - ai.batches.set('msgbatch_crashed01', [ + ai.batches.set('batch_crashed01', [ { custom_id: 'smartphones-0', ...answer([1, 2, 3].map(ref => ({ ref, points: GOOD_POINTS }))) }, ]); // A dry run while it is open does not count its reviews as due. @@ -542,7 +543,7 @@ async function testJob() { category_slug: 'smartphones', outcome: 'stored', attempts: 1, - batch_id: 'msgbatch_crashed01', + batch_id: 'batch_crashed01', }); const resumed = await runReviewAiPointsJob(crashed.client, { apply: true, @@ -550,7 +551,7 @@ async function testJob() { ai: ai.client, now, }); - assert.deepEqual(resumed.collected, ['msgbatch_crashed01']); + assert.deepEqual(resumed.collected, ['batch_crashed01']); assert.equal(resumed.sent, 0, 'nothing sent twice'); assert.ok(!ai.calls.includes('create')); assert.deepEqual( @@ -563,7 +564,7 @@ async function testJob() { const running = fakeSupabase(seed()); running.tables.review_ai_batches!.push({ ...crashed.tables.review_ai_batches![0]!, - id: 'msgbatch_running01', + id: 'batch_running01', status: 'submitted', cost_usd: null, }); @@ -577,7 +578,7 @@ async function testJob() { waitMs: 0, wait: async () => {}, }); - assert.deepEqual(waited.waiting, ['msgbatch_running01']); + assert.deepEqual(waited.waiting, ['batch_running01']); assert.equal(waited.candidates, 0); assert.ok(!ai.calls.includes('create')); // ...and the kill switch cancels it. @@ -587,7 +588,7 @@ async function testJob() { ai: ai.client, now, }); - assert.deepEqual(stopped.canceled, ['msgbatch_running01']); + assert.deepEqual(stopped.canceled, ['batch_running01']); assert.ok(ai.calls.includes('cancel')); assert.equal( running.tables.review_ai_batches![0]!.status, @@ -597,9 +598,7 @@ async function testJob() { // Switched on again: the canceled batch ends; its finished request is stored, the // canceled one comes back as failed rows, and nothing is sent twice. ai.status = 'ended'; - ai.batches.set('msgbatch_running01', [ - { custom_id: 'smartphones-0', result: { type: 'canceled' } }, - ]); + ai.batches.set('batch_running01', [{ custom_id: 'smartphones-0', result: { type: 'canceled' } }]); ai.calls.length = 0; const back = await runReviewAiPointsJob(running.client, { apply: true, @@ -607,7 +606,7 @@ async function testJob() { ai: ai.client, now, }); - assert.equal(back.collected[0], 'msgbatch_running01', 'the old batch first, then the new one'); + assert.equal(back.collected[0], 'batch_running01', 'the old batch first, then the new one'); assert.equal(back.stored.failed, 3); // The fake writer has no attempts; the failed reviews are due again, in a new batch. assert.equal(back.sent, 3); @@ -615,7 +614,7 @@ async function testJob() { // The monthly cap: this month's spend plus the ceiling over $5 sends nothing. const capped = fakeSupabase(seed()); capped.tables.review_ai_batches!.push({ - id: 'msgbatch_spent0001', + id: 'batch_spent0001', model: 'dos', prompt_version: 'v0', status: 'stored', @@ -626,7 +625,7 @@ async function testJob() { stored_at: '2026-11-02T01:00:00Z', }); capped.tables.review_ai_batches!.push({ - id: 'msgbatch_lastmonth', + id: 'batch_lastmonth', model: 'dos', prompt_version: 'v0', status: 'stored', @@ -646,12 +645,12 @@ async function testJob() { assert.equal(over.capped, true); assert.equal(over.monthSpendUsd, 4.999, 'last month does not count'); assert.equal(ai.calls.length, 0); - const sonnet = await runReviewAiPointsJob(capped.client, { + const qwenMax = await runReviewAiPointsJob(capped.client, { enabled: true, model: 'qwen3.8-max', now, }); - assert.equal(sonnet.capped, false, 'each model has its own cap'); + assert.equal(qwenMax.capped, false, 'each model has its own cap'); // A quality stop: most quotes made up; nothing stored, the batch left ended. const bad = fakeSupabase(seed()); @@ -689,6 +688,37 @@ async function testJob() { /--wait-minutes/ ); + // A batch the gateway failed as a whole: the row is closed, nothing is stored, and the + // reviews are due again with no attempt spent. + const goneDb = fakeSupabase(seed()); + goneDb.tables.review_ai_batches!.push({ + ...crashed.tables.review_ai_batches![0]!, + id: 'batch_failed01', + status: 'submitted', + cost_usd: null, + stored_at: null, + }); + const goneAi: ReviewAiClient = { + ...ai.client, + async createBatch() { + throw new Error('not in this test'); + }, + // eslint-disable-next-line require-yield + async *results() { + throw new BatchGoneError('dos.ai batch ended failed with no output', 'canceled'); + }, + }; + const afterGone = await runReviewAiPointsJob(goneDb.client, { + enabled: true, + ai: goneAi, + now, + apply: true, + maxReviews: 1, + }).catch((err: unknown) => err); + assert.match(String(afterGone), /not in this test/, 'its reviews were due again and sent'); + assert.equal(goneDb.tables.review_ai_batches![0]!.status, 'canceled'); + assert.equal(goneDb.stores.length, 0); + // The purge: a dry run says so; --apply calls the writer; a bad version is refused. const purge = fakeSupabase(seed()); const purgeDry = await runReviewAiPointsJob(purge.client, { purge: 'v1' }); @@ -791,9 +821,16 @@ async function testDosAiClient() { } if (path === '/batches/batch_abc123/cancel') return reply(200, { status: 'cancelling' }); if (path === '/batches/batch_broke') return reply(402, { error: 'insufficient credits' }); + if (path === '/batches/batch_failed') { + return reply(200, { status: 'failed', output_file_id: null, errmsg: 'zdr org' }); + } + if (path === '/batches/batch_purged') { + return reply(200, { status: 'completed', output_file_id: 'file_gone' }); + } return reply(404, 'no route'); }) as typeof fetch; - const client = dosAiReviewAiClient('dos_sk_test', 'https://beta-api.dos.ai/v1/', gateway); + const noWait = async () => {}; + const client = dosAiReviewAiClient('dos_sk_test', 'https://beta-api.dos.ai/v1/', gateway, noWait); const built = buildRequests( [{ id: id(1), text: LONG, textHash: reviewTextHash(LONG), category: 'smartphones' }], 'dos' @@ -814,7 +851,69 @@ async function testDosAiClient() { 'POST /batches', 'GET /batches/batch_abc123', ]); - console.log(' ✅ dos.ai batch client: upload, create, status, results, cancel, retry and 402'); + // A batch failed as a whole, or one whose output file is gone: no results, ever. + const drain = async (batchId: string) => { + for await (const r of client.results(batchId)) void r; + }; + await assert.rejects(drain('batch_failed'), (err: unknown) => { + assert.ok(err instanceof BatchGoneError); + assert.equal(err.status, 'canceled'); + assert.match(err.message, /ended failed with no output: zdr org/); + return true; + }); + await assert.rejects(drain('batch_purged'), (err: unknown) => { + assert.ok(err instanceof BatchGoneError); + assert.equal(err.status, 'expired'); + return true; + }); + // A line the gateway did not run for want of credits is unbilled, not failed. + assert.equal(reason({ custom_id: 'f', response: { status_code: 402, body: {} } }), 'unbilled'); + assert.equal( + reason({ custom_id: 'g', error: { code: 'insufficient_credits', message: 'x' } }), + 'unbilled' + ); + + // POST /batches is never retried on an answer that may hide a created batch. + for (const lost of ['network', '502'] as const) { + let creates = 0; + const flaky = (async (url: string | URL | Request, init?: RequestInit) => { + const path = String(url).replace('https://api.dos.me/v1', ''); + if (path === '/files') return new Response(JSON.stringify({ id: 'file_in' })); + creates += 1; + if (init?.method === 'POST' && lost === 'network') throw new Error('socket hang up'); + return new Response('bad gateway', { status: 502 }); + }) as typeof fetch; + const once = dosAiReviewAiClient('dos_sk_test', undefined, flaky, noWait); + await assert.rejects( + once.createBatch(built.requests), + lost === 'network' ? /A batch may have been created/ : /502/ + ); + assert.equal(creates, 1, `POST /batches sent once (${lost})`); + } + // ...but a 429, refused before anything exists, is retried. + let throttled = 0; + const busyGateway = (async (url: string | URL | Request) => { + const path = String(url).replace('https://api.dos.me/v1', ''); + if (path === '/files') return new Response(JSON.stringify({ id: 'file_in' })); + throttled += 1; + return throttled === 1 + ? new Response('slow down', { status: 429 }) + : new Response(JSON.stringify({ id: 'batch_after_429' })); + }) as typeof fetch; + assert.deepEqual( + await dosAiReviewAiClient('dos_sk_test', undefined, busyGateway, noWait).createBatch( + built.requests + ), + { id: 'batch_after_429' } + ); + // The request body: deterministic and bounded. + assert.deepEqual( + [built.requests[0]!.body.temperature, built.requests[0]!.body.max_tokens], + [0, 6000] + ); + console.log( + ' ✅ dos.ai batch client: upload, create once, status, results, gone batches, cancel, retry and 402' + ); } function refsOf(request: BatchRequest): number[] { @@ -837,7 +936,7 @@ function fakeAi(answerFor: (requests: BatchRequest[]) => BatchResult[]) { async createBatch(requests) { state.calls.push('create'); state.onCreate?.(); - const batchId = `msgbatch_fake${String((n += 1)).padStart(4, '0')}`; + const batchId = `batch_fake${String((n += 1)).padStart(4, '0')}`; state.batches.set(batchId, answerFor(requests)); return { id: batchId }; }, diff --git a/apps/importer/src/shopee/jobs/review-ai-points.ts b/apps/importer/src/shopee/jobs/review-ai-points.ts index 47abae6bd..a7eaf57dc 100644 --- a/apps/importer/src/shopee/jobs/review-ai-points.ts +++ b/apps/importer/src/shopee/jobs/review-ai-points.ts @@ -143,7 +143,7 @@ export type BatchResult = }; }; } - | { custom_id: string; result: { type: 'errored' | 'expired' | 'canceled' } }; + | { custom_id: string; result: { type: 'errored' | 'expired' | 'canceled' | 'unbilled' } }; export interface ReviewAiClient { createBatch(requests: BatchRequest[]): Promise<{ id: string }>; @@ -177,12 +177,20 @@ interface DosAiOutputLine { /** * A dos.ai output line as the job reads results: a 200 response is a * message whose stop reason is the finish reason ('stop' ends normally, - * 'length' is cut off, 'content_filter' is refused); anything else errored. + * 'length' is cut off, 'content_filter' is refused); a line the gateway did + * not run for want of credits (402, insufficient_credits) is 'unbilled', so + * the review stays due without spending an attempt; anything else errored. * Cached prompt tokens are split out of prompt_tokens. */ export function fromDosAiOutputLine(line: DosAiOutputLine): BatchResult | null { if (typeof line.custom_id !== 'string') return null; const body = line.response?.body; + if ( + line.response?.status_code === 402 || + /insufficient|credit|balance/i.test(line.error?.code ?? '') + ) { + return { custom_id: line.custom_id, result: { type: 'unbilled' } }; + } if (line.error || line.response?.status_code !== 200 || !body) { return { custom_id: line.custom_id, result: { type: 'errored' } }; } @@ -216,50 +224,84 @@ export function fromDosAiOutputLine(line: DosAiOutputLine): BatchResult | null { }; } +/** + * A batch whose results will never come: the gateway failed it as a whole + * (a refused file, a Zero Data Retention organization, no credits at start), + * it ended with no output, or its output file is gone. The job closes its row + * and stores nothing, so its reviews are due again without spending an attempt. + */ +export class BatchGoneError extends Error { + constructor( + message: string, + readonly status: 'canceled' | 'expired' + ) { + super(message); + this.name = 'BatchGoneError'; + } +} + /** * The real client: the dos.ai gateway's OpenAI-compatible Files and Batch API * (POST /files, POST /batches, GET /batches/:id, POST /batches/:id/cancel, * GET /files/:id/content), with the key in DOS_AI_API_KEY as a Bearer token. - * A 429 or 5xx is retried; a 402 (no credits left) stops the run. + * Each call, body included, has a timeout. A 429 or 5xx, or a network error, + * is retried, except on POST /batches: a batch may have been created behind a + * lost answer, and a retry would run and bill a second one, so that call is + * retried only on 429 (refused before anything was created). A 402 (no + * credits left) stops the run. */ export function dosAiReviewAiClient( apiKey: string, baseUrl: string = DOS_AI_DEFAULT_BASE_URL, - fetchImpl: typeof fetch = fetch + fetchImpl: typeof fetch = fetch, + wait: (ms: number) => Promise = ms => new Promise(r => setTimeout(r, ms)) ): ReviewAiClient { const base = baseUrl.replace(/\/+$/, ''); - const call = async (path: string, init: RequestInit = {}): Promise => { + const call = async ( + path: string, + init: RequestInit = {}, + retry: 'all' | 'throttled' = 'all' + ): Promise<{ status: number; text: string }> => { for (let attempt = 1; ; attempt++) { const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), DOS_AI_TIMEOUT_MS); - let response: Response | null = null; + let status = 0; + let text = ''; + let failure: unknown = null; try { - response = await fetchImpl(`${base}${path}`, { + const response = await fetchImpl(`${base}${path}`, { ...init, headers: { Authorization: `Bearer ${apiKey}`, ...(init.headers ?? {}) }, signal: controller.signal, }); + status = response.status; + text = await response.text(); } catch (err) { - if (attempt >= DOS_AI_ATTEMPTS) throw err; + failure = err; } finally { clearTimeout(timer); } - if (response) { - if (response.ok) return response; - if (response.status === 402) { - throw new Error('dos.ai refused the request: no credits left on the key (402).'); - } - if ((response.status !== 429 && response.status < 500) || attempt >= DOS_AI_ATTEMPTS) { - // The body names the reason; it holds no review text of ours. - const detail = (await response.text().catch(() => '')).slice(0, 300); - throw new Error(`dos.ai ${init.method ?? 'GET'} ${path}: ${response.status} ${detail}`); + if (failure === null && status >= 200 && status < 300) return { status, text }; + if (status === 402) { + throw new Error('dos.ai refused the request: no credits left on the key (402).'); + } + const retryable = + retry === 'all' ? failure !== null || status === 429 || status >= 500 : status === 429; + if (!retryable || attempt >= DOS_AI_ATTEMPTS) { + if (failure !== null && retry === 'throttled') { + throw new Error( + `dos.ai ${init.method ?? 'GET'} ${path}: no answer (${String(failure)}). A batch may have been created; check the key's batches on dos.ai before the next run.` + ); } + if (failure !== null) throw failure; + // The body names the reason; it holds no review text of ours. + throw new Error(`dos.ai ${init.method ?? 'GET'} ${path}: ${status} ${text.slice(0, 300)}`); } - await new Promise(resolve => setTimeout(resolve, 1000 * 2 ** (attempt - 1))); + await wait(1000 * 2 ** (attempt - 1)); } }; - const json = async (path: string, init?: RequestInit) => - (await (await call(path, init)).json()) as T; + const json = async (path: string, init?: RequestInit, retry?: 'all' | 'throttled') => + JSON.parse((await call(path, init, retry)).text) as T; return { async createBatch(requests) { const form = new FormData(); @@ -272,15 +314,19 @@ export function dosAiReviewAiClient( 'review-ai-points.jsonl' ); const file = await json<{ id: string }>('/files', { method: 'POST', body: form }); - const batch = await json<{ id: string }>('/batches', { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ - input_file_id: file.id, - endpoint: '/v1/chat/completions', - completion_window: '24h', - }), - }); + const batch = await json<{ id: string }>( + '/batches', + { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + input_file_id: file.id, + endpoint: '/v1/chat/completions', + completion_window: '24h', + }), + }, + 'throttled' + ); return { id: batch.id }; }, async batchStatus(id) { @@ -291,14 +337,27 @@ export function dosAiReviewAiClient( : 'in_progress'; }, async *results(id) { - const batch = await json<{ output_file_id?: string | null }>( - `/batches/${encodeURIComponent(id)}` - ); - // A failed or canceled batch may have no output: its lines come back as failed rows. - if (!batch.output_file_id) return; - const text = await ( - await call(`/files/${encodeURIComponent(batch.output_file_id)}/content`) - ).text(); + const batch = await json<{ + status?: string; + output_file_id?: string | null; + errmsg?: string | null; + }>(`/batches/${encodeURIComponent(id)}`); + if (!batch.output_file_id) { + // Failed as a whole, or ended with no line run: nothing to store. + throw new BatchGoneError( + `dos.ai batch ${id} ended ${batch.status ?? 'without a status'} with no output${batch.errmsg ? `: ${batch.errmsg.slice(0, 200)}` : ''}`, + batch.status === 'expired' ? 'expired' : 'canceled' + ); + } + let text: string; + try { + text = (await call(`/files/${encodeURIComponent(batch.output_file_id)}/content`)).text; + } catch (err) { + if (/ 404 /.test(String(err))) { + throw new BatchGoneError(`dos.ai batch ${id}: its output file is gone (404)`, 'expired'); + } + throw err; + } for (const raw of text.split('\n')) { if (!raw.trim()) continue; let line: DosAiOutputLine; @@ -577,6 +636,8 @@ export interface CollectedBatch { rejects: Record; unaccented: number; failedReviews: number; + /** Reviews in requests the gateway did not run for want of credits: still due. */ + unbilledReviews: number; reviewsWithLines: number; reviewsAnswered: number; } @@ -629,6 +690,7 @@ export function collectBatch( }, unaccented: 0, failedReviews: 0, + unbilledReviews: 0, reviewsWithLines: 0, reviewsAnswered: 0, }; @@ -653,6 +715,11 @@ export function collectBatch( } } } + // A request the gateway never ran for want of credits: its reviews stay due, no attempt spent. + if (result?.result.type === 'unbilled') { + out.unbilledReviews += request.reviews.length; + continue; + } for (const review of request.reviews) { const base = { id: review.id, @@ -892,16 +959,26 @@ export async function runReviewAiPointsJob( try { for await (const entry of ai.results(batch.id)) results.push(entry); } catch (err) { - // The API keeps results 29 days: an older batch is gone, so mark it and move on. - if (now.getTime() - new Date(batch.submitted_at).getTime() < RESULTS_KEPT_MS) throw err; + // Results that will never come (the client says so, or the batch is past 28 days): + // close the row and store nothing, so the reviews are due again with no attempt spent. + const gone = err instanceof BatchGoneError; + if (!gone && now.getTime() - new Date(batch.submitted_at).getTime() < RESULTS_KEPT_MS) { + throw err; + } + const status = gone ? err.status : 'expired'; const { error } = await supabase .from('review_ai_batches') - .update({ status: 'expired', ended_at: now.toISOString() }) + .update({ status, ended_at: now.toISOString() }) .eq('id', batch.id); - if (error) throw new Error(`review_ai_batches expire: ${error.message}`); - logger.warn('review_ai.expired', `Batch ${batch.id} results are gone; marked expired.`, { - batch: batch.id, - }); + if (error) throw new Error(`review_ai_batches close: ${error.message}`); + logger.warn( + 'review_ai.gone', + `Batch ${batch.id} has no results to store; marked ${status}.`, + { + batch: batch.id, + reason: err instanceof Error ? err.message : String(err), + } + ); return false; } const ids = batch.manifest.flatMap(r => r.reviews.map(v => v.id)); @@ -962,6 +1039,7 @@ export async function runReviewAiPointsJob( rejects: collected.rejects, unaccented: collected.unaccented, failed_reviews: collected.failedReviews, + unbilled_reviews: collected.unbilledReviews, cost_usd: Number(collected.costUsd.toFixed(4)), ...counts, });
    • ModelGiá đọc vào / viết ra (USD mỗi 1 triệu token)Nhận xét
    Claude Haiku 5.5 (claude-haiku-5-5)0,10 / 0,50Rẻ nhất. Việc này đơn giản: đọc một đoạn ngắn, rút ý có trích dẫn. Đề xuất dùng mặc định.