WuYongJie/furniture-3d-system-Wu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(bom-costing): bound concurrent sku synthesis

Browse Source
This commit is contained in:
WuYongJie 2026-08-04 19:14:41 +08:00
parent f0687f2d5d
commit 143aa7fa3f
3 changed files with 157 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-analyzer.test.ts
View File

@ -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<void>((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,
})
})

156
furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-analyzer.ts
View File

@ -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<typeof competitorSkuSchema>[],
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<typeof competitorSkuSchema>[] = []
let product: z.infer<typeof aiAnalysisPayloadSchema>['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<ReturnType<typeof callBomCostingMultimodalModel>>
parsedPayload: z.infer<typeof aiAnalysisPayloadSchema>
batchSkus: z.infer<typeof competitorSkuSchema>[]
} | 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 <T>(operation: () => Promise<T>) => {
const previousCommit = commitQueue
let releaseCommit: () => void = () => undefined
commitQueue = new Promise<void>((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,

2
furniture-3d-system/src/server/services/bom-costing/bom-costing-processing.service.ts
View File

@ -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,