merge: BOM non-stream synthesis v6

This commit is contained in:
WuYongJie 2026-08-11 02:50:17 +08:00
commit 3c8d7f0801
3 changed files with 114 additions and 15 deletions

View File

@ -11,6 +11,49 @@ describe('BOM 多模态网关', () => {
jest.restoreAllMocks()
})
it('结构汇总默认请求一次性 JSON,并读取标准非流式响应', async () => {
const fetchSpy = jest.spyOn(global, 'fetch').mockResolvedValue(
new Response(JSON.stringify({
id: 'call-json-response',
model: 'gpt-5.6-terra',
choices: [{
message: { content: '{"ok":true}' },
finish_reason: 'stop',
}],
usage: { prompt_tokens: 12, completion_tokens: 5 },
}), {
status: 200,
headers: { 'Content-Type': 'application/json' },
}),
)
const result = await callBomCostingMultimodalModel({
config: {
baseUrl: 'https://ai.example.com/v1',
apiKey: 'test-key',
model: 'gpt-5.6-terra',
},
operation: 'SYNTHESIS',
systemPrompt: 'test',
userContent: 'test',
maxTokens: 100,
timeoutMs: 5_000,
})
const requestBody = JSON.parse(String(fetchSpy.mock.calls[0]?.[1]?.body)) as {
stream?: boolean
}
expect(requestBody.stream).toBe(false)
expect(result).toMatchObject({
id: 'call-json-response',
model: 'gpt-5.6-terra',
content: '{"ok":true}',
finishReason: 'stop',
promptTokens: 12,
completionTokens: 5,
})
})
it('流式响应空转超过时限时取消读取并返回可恢复的超时错误', async () => {
let cancelReason: unknown
const stalledStream = new ReadableStream({

View File

@ -30,6 +30,16 @@ export interface BomCostingMultimodalCallResult {
completedFromJsonPayload?: boolean
}
interface BomCostingChatCompletionResponse {
id: string | null
model: string | null
content: string
promptTokens: number | null
completionTokens: number | null
finishReason: string | null
completedFromJsonPayload: boolean
}
export type BomCostingMultimodalOperation = 'IMAGE_FACTS' | 'SYNTHESIS'
const BOM_COSTING_REASONING_EFFORTS = new Set([
@ -285,7 +295,7 @@ function isCompleteJsonObjectPayload(content: string) {
async function readStreamingChatCompletion(
response: Response,
options: { signal?: AbortSignal } = {},
) {
): Promise<BomCostingChatCompletionResponse> {
if (!response.body) {
throw new AppError(
ErrorCode.BOM_COSTING_INVALID_PROVIDER,
@ -466,6 +476,41 @@ async function readStreamingChatCompletion(
}
}
async function readNonStreamingChatCompletion(
response: Response,
): Promise<BomCostingChatCompletionResponse> {
const payload = await response.json() as {
id?: string
model?: string
choices?: Array<{
message?: { content?: string }
finish_reason?: string | null
}>
usage?: {
prompt_tokens?: number
completion_tokens?: number
}
}
const choice = payload.choices?.[0]
return {
id: payload.id?.trim() || null,
model: payload.model?.trim() || null,
content: typeof choice?.message?.content === 'string'
? choice.message.content.trim()
: '',
promptTokens: Number.isFinite(payload.usage?.prompt_tokens)
? payload.usage!.prompt_tokens!
: null,
completionTokens: Number.isFinite(payload.usage?.completion_tokens)
? payload.usage!.completion_tokens!
: null,
finishReason: typeof choice?.finish_reason === 'string' && choice.finish_reason.trim()
? choice.finish_reason.trim()
: null,
completedFromJsonPayload: false,
}
}
function readTransportErrorCode(error: unknown): string | null {
let current = error
for (let depth = 0; depth < 4; depth += 1) {
@ -577,6 +622,12 @@ export async function callBomCostingMultimodalModel(input: {
for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
let response: Response
// The UI streams durable stage events, not raw model tokens. Synthesis only
// needs one strict JSON object, so asking the provider for a regular JSON
// response avoids OpenAI-compatible gateways that emit a partial SSE body
// and then keep the socket open for minutes. Image analysis keeps streaming
// because those responses can be substantially larger.
const requestStreamingResponse = input.operation === 'IMAGE_FACTS'
const timeoutSignal = AbortSignal.timeout(input.timeoutMs)
const requestSignal = input.signal
? AbortSignal.any([timeoutSignal, input.signal])
@ -597,7 +648,7 @@ export async function callBomCostingMultimodalModel(input: {
response_format: { type: 'json_object' },
temperature: 0.1,
max_tokens: input.maxTokens,
stream: true,
stream: requestStreamingResponse,
...(reasoningEffort ? { reasoning_effort: reasoningEffort } : {}),
}),
cache: 'no-store',
@ -673,11 +724,15 @@ export async function callBomCostingMultimodalModel(input: {
)
}
let result: Awaited<ReturnType<typeof readStreamingChatCompletion>>
let result: BomCostingChatCompletionResponse
try {
result = await readStreamingChatCompletion(response, {
signal: requestSignal,
})
const responseContentType = response.headers.get('content-type')?.toLowerCase() ?? ''
const responseUsesEventStream = responseContentType.includes('text/event-stream') || (
requestStreamingResponse && !responseContentType.includes('application/json')
)
result = responseUsesEventStream
? await readStreamingChatCompletion(response, { signal: requestSignal })
: await readNonStreamingChatCompletion(response)
} catch (error) {
const taskDeadlineError = readTaskDeadlineError(requestSignal)
if (taskDeadlineError) throw taskDeadlineError

View File

@ -360,13 +360,14 @@ function sanitizeBlueprintFactIds(
}
}
function getTimeoutMs() {
return readBomCostingBoundedInteger(
function getTimeoutMs(maximumMs = 60_000) {
const configured = readBomCostingBoundedInteger(
process.env.BOM_COSTING_STRUCTURE_TIMEOUT_MS,
100_000,
60_000,
30_000,
180_000,
)
return Math.min(configured, maximumMs)
}
function recordResponse(trace: BomCostingStructureTrace, response: BomCostingMultimodalCallResult) {
@ -463,8 +464,8 @@ async function generateBlueprint(input: {
reviewInstruction: input.analysisInstruction?.trim() || undefined,
}),
maxTokens: 2_400,
timeoutMs: getTimeoutMs(),
retrySynthesisTimeout: true,
timeoutMs: getTimeoutMs(60_000),
retrySynthesisTimeout: false,
signal: input.signal,
})
recordResponse(input.trace, response)
@ -557,8 +558,8 @@ async function verifyBlueprintGroup(input: {
systemPrompt: VERIFICATION_SYSTEM_PROMPT,
userContent: content,
maxTokens: 4_000,
timeoutMs: getTimeoutMs(),
retrySynthesisTimeout: true,
timeoutMs: getTimeoutMs(60_000),
retrySynthesisTimeout: false,
signal: input.signal,
})
recordResponse(input.trace, response)
@ -933,8 +934,8 @@ async function instantiateBlueprint(input: {
reviewInstruction: input.analysisInstruction?.trim() || undefined,
}),
maxTokens: 2_200,
timeoutMs: getTimeoutMs(),
retrySynthesisTimeout: true,
timeoutMs: getTimeoutMs(45_000),
retrySynthesisTimeout: false,
signal: input.signal,
})
recordResponse(input.trace, response)