From 143aa7fa3f397996d74d03db51f4cae3855a495d Mon Sep 17 00:00:00 2001 From: WuYongJie Date: Tue, 4 Aug 2026 19:14:41 +0800 Subject: [PATCH] fix(bom-costing): bound concurrent sku synthesis --- .../bom-costing-multimodal-analyzer.test.ts | 47 ++++-- .../bom-costing-multimodal-analyzer.ts | 156 ++++++++++++++---- .../bom-costing-processing.service.ts | 2 +- 3 files changed, 157 insertions(+), 48 deletions(-) diff --git a/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-analyzer.test.ts b/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-analyzer.test.ts index 8b17f4df..d079954c 100644 --- a/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-analyzer.test.ts +++ b/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-analyzer.test.ts @@ -220,6 +220,8 @@ describe('BOM 多模态分析器', () => { const originalAnalysisTimeout = process.env.BOM_COSTING_AI_TIMEOUT_MS const originalSynthesisSkuBatchSize = process.env.BOM_COSTING_SYNTHESIS_SKU_BATCH_SIZE + const originalSynthesisBatchConcurrency = + process.env.BOM_COSTING_SYNTHESIS_BATCH_CONCURRENCY beforeEach(() => { process.env.BOM_COSTING_AI_FACT_BATCH_SIZE = '6' @@ -253,6 +255,12 @@ describe('BOM 多模态分析器', () => { process.env.BOM_COSTING_SYNTHESIS_SKU_BATCH_SIZE = originalSynthesisSkuBatchSize } + if (originalSynthesisBatchConcurrency === undefined) { + delete process.env.BOM_COSTING_SYNTHESIS_BATCH_CONCURRENCY + } else { + process.env.BOM_COSTING_SYNTHESIS_BATCH_CONCURRENCY = + originalSynthesisBatchConcurrency + } jest.restoreAllMocks() }) @@ -731,7 +739,8 @@ describe('BOM 多模态分析器', () => { }) it('将高规格商品拆成多个汇总批次，并覆盖每一个平台 SKU', async () => { - process.env.BOM_COSTING_SYNTHESIS_SKU_BATCH_SIZE = '100' + process.env.BOM_COSTING_SYNTHESIS_SKU_BATCH_SIZE = '20' + process.env.BOM_COSTING_SYNTHESIS_BATCH_CONCURRENCY = '3' const highCardinalitySnapshot: MarketDetailSnapshot = { ...snapshot, skuTotalRows: 552, @@ -744,8 +753,15 @@ describe('BOM 多模态分析器', () => { })), } const batchSizes: number[] = [] + let activeRequestCount = 0 + let peakConcurrentRequestCount = 0 const fetchSpy = jest.spyOn(global, 'fetch').mockImplementation( async (_input, init) => { + activeRequestCount += 1 + peakConcurrentRequestCount = Math.max( + peakConcurrentRequestCount, + activeRequestCount, + ) const requestBody = JSON.parse(String((init as RequestInit).body)) as { messages?: Array<{ content: string }> } @@ -756,11 +772,16 @@ describe('BOM 多模态分析器', () => { } const skuTruth = prompt.skuTruth ?? [] batchSizes.push(skuTruth.length) - return createStreamingResponse(JSON.stringify({ - product: { category: '书桌' }, - structureGroups: [analyzedStructureGroup()], - skus: skuTruth.map((sku) => analyzedSku(sku.externalSkuId)), - })) + try { + await new Promise((resolve) => setTimeout(resolve, 5)) + return createStreamingResponse(JSON.stringify({ + product: { category: '书桌' }, + structureGroups: [analyzedStructureGroup()], + skus: skuTruth.map((sku) => analyzedSku(sku.externalSkuId)), + })) + } finally { + activeRequestCount -= 1 + } }, ) @@ -770,16 +791,20 @@ describe('BOM 多模态分析器', () => { existingFactSnapshot: createCompleteFactSnapshot(highCardinalitySnapshot), }) - expect(fetchSpy).toHaveBeenCalledTimes(6) - expect(batchSizes).toEqual([100, 100, 100, 100, 100, 52]) + expect(fetchSpy).toHaveBeenCalledTimes(28) + expect(batchSizes.sort((left, right) => left - right)).toEqual([ + 12, + ...Array.from({ length: 27 }, () => 20), + ]) + expect(peakConcurrentRequestCount).toBe(3) expect(output.result.skus).toHaveLength(552) expect(new Set(output.result.skus.map((sku) => sku.externalSkuId)).size).toBe(552) expect(output.result.skus[0]?.structureGroupKey).toBe('batch-1-desk-120') - expect(output.result.skus[100]?.structureGroupKey).toBe('batch-2-desk-120') - expect(output.result.skus[551]?.structureGroupKey).toBe('batch-6-desk-120') + expect(output.result.skus[20]?.structureGroupKey).toBe('batch-2-desk-120') + expect(output.result.skus[551]?.structureGroupKey).toBe('batch-28-desk-120') expect(output.trace).toMatchObject({ inputSkuCount: 552, - synthesisBatchCount: 6, + synthesisBatchCount: 28, }) }) diff --git a/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-analyzer.ts b/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-analyzer.ts index 8511f46a..74290734 100644 --- a/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-analyzer.ts +++ b/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-analyzer.ts @@ -221,7 +221,11 @@ const aiAnalysisPayloadSchema = z.object({ }) const MAX_MULTIMODAL_ANALYSIS_SKUS = 600 -const DEFAULT_SYNTHESIS_SKU_BATCH_SIZE = 80 +// Large SKU catalogs need to fit below the gateway's per-request response +// window. Keep each structured response small, then use a bounded worker pool +// so the overall task does not turn into dozens of serial requests. +const DEFAULT_SYNTHESIS_SKU_BATCH_SIZE = 20 +const DEFAULT_SYNTHESIS_BATCH_CONCURRENCY = 3 const MAX_EXPLICIT_FACT_SKU_IDS = 20 function describeHighCardinalitySkuCatalog(skus: MarketSkuSnapshot[]) { @@ -392,6 +396,15 @@ function splitSkuBatches(skus: MarketSkuSnapshot[]) { ) } +function getSynthesisBatchConcurrency() { + return readBomCostingBoundedInteger( + process.env.BOM_COSTING_SYNTHESIS_BATCH_CONCURRENCY, + DEFAULT_SYNTHESIS_BATCH_CONCURRENCY, + 1, + 3, + ) +} + function scopeBatchStructureGroupKeys( skus: z.infer[], batchIndex: number, @@ -608,6 +621,7 @@ export class BomCostingMultimodalAnalyzer { ? Math.min(configuredMaxTokens, 12_000) : configuredMaxTokens const skuBatches = splitSkuBatches(input.snapshot.skus) + const synthesisBatchConcurrency = getSynthesisBatchConcurrency() const analyzedSkus: z.infer[] = [] let product: z.infer['product'] | null = null @@ -617,13 +631,41 @@ export class BomCostingMultimodalAnalyzer { let synthesisDurationMs = 0 let synthesisPromptTokens: number | null = null let synthesisCompletionTokens: number | null = null + const completedBatchResults: Array<{ + responsePayload: Awaited> + parsedPayload: z.infer + batchSkus: z.infer[] + } | undefined> = new Array(skuBatches.length) + let completedSkuCount = 0 + let nextBatchOffset = 0 + let firstError: unknown = null + let commitQueue = Promise.resolve() - for (const [batchIndex, skuBatch] of skuBatches.entries()) { - await input.onSynthesisBatchStarted?.({ - batchIndex: batchIndex + 1, - totalBatches: skuBatches.length, - batchSkuCount: skuBatch.length, - completedSkuCount: analyzedSkus.length, + const withCommitLock = async (operation: () => Promise) => { + const previousCommit = commitQueue + let releaseCommit: () => void = () => undefined + commitQueue = new Promise((resolve) => { + releaseCommit = resolve + }) + await previousCommit + try { + return await operation() + } finally { + releaseCommit() + } + } + + const analyzeSynthesisBatch = async ( + batchIndex: number, + skuBatch: MarketSkuSnapshot[], + ) => { + await withCommitLock(async () => { + await input.onSynthesisBatchStarted?.({ + batchIndex: batchIndex + 1, + totalBatches: skuBatches.length, + batchSkuCount: skuBatch.length, + completedSkuCount, + }) }) const responsePayload = await callBomCostingMultimodalModel({ config, @@ -645,36 +687,24 @@ export class BomCostingMultimodalAnalyzer { ) const batchSkus = expandStructureGroupBom(parsedPayload) assertExactSkuCoverage(skuBatch, batchSkus) - analyzedSkus.push( - ...scopeBatchStructureGroupKeys( + await withCommitLock(async () => { + completedBatchResults[batchIndex] = { + responsePayload, + parsedPayload, batchSkus, - batchIndex, - skuBatches.length, - ), - ) - product ??= parsedPayload.product - model = responsePayload.model || model - callId ??= responsePayload.id - if (responsePayload.id) synthesisCallIds.push(responsePayload.id) - synthesisDurationMs += responsePayload.durationMs - synthesisPromptTokens = addNullableNumber( - synthesisPromptTokens, - responsePayload.promptTokens, - ) - synthesisCompletionTokens = addNullableNumber( - synthesisCompletionTokens, - responsePayload.completionTokens, - ) - await input.onSynthesisBatchCompleted?.({ - batchIndex: batchIndex + 1, - totalBatches: skuBatches.length, - batchSkuCount: skuBatch.length, - completedSkuCount: analyzedSkus.length, - model: responsePayload.model, - callId: responsePayload.id, - durationMs: responsePayload.durationMs, - promptTokens: responsePayload.promptTokens, - completionTokens: responsePayload.completionTokens, + } + completedSkuCount += skuBatch.length + await input.onSynthesisBatchCompleted?.({ + batchIndex: batchIndex + 1, + totalBatches: skuBatches.length, + batchSkuCount: skuBatch.length, + completedSkuCount, + model: responsePayload.model, + callId: responsePayload.id, + durationMs: responsePayload.durationMs, + promptTokens: responsePayload.promptTokens, + completionTokens: responsePayload.completionTokens, + }) }) } catch (error) { console.warn('[BOM costing] multimodal synthesis contract rejected', { @@ -694,6 +724,60 @@ export class BomCostingMultimodalAnalyzer { } } + const runSynthesisWorker = async () => { + while (!firstError) { + const batchOffset = nextBatchOffset + nextBatchOffset += 1 + const skuBatch = skuBatches[batchOffset] + if (!skuBatch) return + try { + await analyzeSynthesisBatch(batchOffset, skuBatch) + } catch (error) { + if (!firstError) firstError = error + } + } + } + + await Promise.all( + Array.from( + { length: Math.min(synthesisBatchConcurrency, skuBatches.length) }, + () => runSynthesisWorker(), + ), + ) + if (firstError) throw firstError + + for (const [batchIndex] of skuBatches.entries()) { + const batchResult = completedBatchResults[batchIndex] + if (!batchResult) { + throw new AppError( + ErrorCode.BOM_COSTING_INVALID_PROVIDER, + `第 ${batchIndex + 1} 批规格结构未返回，无法生成完整 BOM`, + ) + } + analyzedSkus.push( + ...scopeBatchStructureGroupKeys( + batchResult.batchSkus, + batchIndex, + skuBatches.length, + ), + ) + product ??= batchResult.parsedPayload.product + model = batchResult.responsePayload.model || model + callId ??= batchResult.responsePayload.id + if (batchResult.responsePayload.id) { + synthesisCallIds.push(batchResult.responsePayload.id) + } + synthesisDurationMs += batchResult.responsePayload.durationMs + synthesisPromptTokens = addNullableNumber( + synthesisPromptTokens, + batchResult.responsePayload.promptTokens, + ) + synthesisCompletionTokens = addNullableNumber( + synthesisCompletionTokens, + batchResult.responsePayload.completionTokens, + ) + } + if (!product) { throw new AppError( ErrorCode.BOM_COSTING_INVALID_PROVIDER, diff --git a/furniture-3d-system/src/server/services/bom-costing/bom-costing-processing.service.ts b/furniture-3d-system/src/server/services/bom-costing/bom-costing-processing.service.ts index f6ebc55d..11d135ee 100644 --- a/furniture-3d-system/src/server/services/bom-costing/bom-costing-processing.service.ts +++ b/furniture-3d-system/src/server/services/bom-costing/bom-costing-processing.service.ts @@ -891,7 +891,7 @@ export class BomCostingProcessingService { status: BomCostingCaseStatus.ANALYZING, progress: caseProgress, message: - `正在汇总第 ${progress.batchIndex}/${progress.totalBatches} 批规格结构` + + `正在并行汇总第 ${progress.batchIndex}/${progress.totalBatches} 批规格结构` + `（${progress.batchSkuCount} 个 SKU）`, payload: { batchIndex: progress.batchIndex,