Fix managed ASR settlement and empty polish recovery
Accept the final sequence format returned by Volcengine without weakening ordering checks, and retry one safe buffered DeepSeek empty response under the same credit reservation.
This commit is contained in:
+111
-9
@@ -34,6 +34,7 @@ import kotlinx.serialization.json.JsonNull
|
||||
import kotlinx.serialization.json.jsonArray
|
||||
import kotlinx.serialization.json.jsonObject
|
||||
import kotlinx.serialization.json.jsonPrimitive
|
||||
import org.slf4j.LoggerFactory
|
||||
|
||||
data class DeepSeekConfig(
|
||||
val endpoint: String,
|
||||
@@ -91,7 +92,25 @@ class DeepSeekProvider(
|
||||
): ProviderUsage {
|
||||
require(request is TextProviderRequest) { "DeepSeek only accepts text requests" }
|
||||
validate(request)
|
||||
return upstream.complete(request, output)
|
||||
var attempt = 1
|
||||
while (true) {
|
||||
try {
|
||||
return upstream.complete(request, output)
|
||||
} catch (failure: DeepSeekEmptyResultException) {
|
||||
LOG.warn(
|
||||
"DeepSeek returned empty content requestId={} capability={} attempt={} " +
|
||||
"finishReason={} reasoningContentPresent={} usagePresent={}",
|
||||
request.requestId,
|
||||
request.capability.name,
|
||||
attempt,
|
||||
failure.finishReason,
|
||||
failure.reasoningContentPresent,
|
||||
failure.usagePresent,
|
||||
)
|
||||
if (request.stream || attempt >= MAX_BUFFERED_ATTEMPTS) throw failure
|
||||
attempt += 1
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun validate(request: TextProviderRequest) {
|
||||
@@ -110,6 +129,11 @@ class DeepSeekProvider(
|
||||
"agent requests must be non-streaming"
|
||||
}
|
||||
}
|
||||
|
||||
private companion object {
|
||||
const val MAX_BUFFERED_ATTEMPTS = 2
|
||||
val LOG = LoggerFactory.getLogger(DeepSeekProvider::class.java)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -182,8 +206,14 @@ class KtorDeepSeekClient(
|
||||
): ProviderUsage {
|
||||
val payload = bytes.decodeToString()
|
||||
val content = extractAssistantContent(payload)
|
||||
?: throw DeepSeekEmptyResultException()
|
||||
if (content.isBlank()) throw DeepSeekEmptyResultException()
|
||||
if (content.isNullOrBlank()) {
|
||||
val metadata = emptyResultMetadata(payload)
|
||||
throw DeepSeekEmptyResultException(
|
||||
finishReason = metadata.finishReason,
|
||||
reasoningContentPresent = metadata.reasoningContentPresent,
|
||||
usagePresent = metadata.usagePresent,
|
||||
)
|
||||
}
|
||||
if (request.capability == GatewayCapability.AGENT) validateAgentContent(content)
|
||||
val usage = extractUsage(payload)?.toProviderUsageOrNull()
|
||||
?: throw DeepSeekProviderException("DeepSeek response omitted token usage")
|
||||
@@ -203,6 +233,8 @@ class KtorDeepSeekClient(
|
||||
var contentBytes = 0L
|
||||
var terminalChoiceSeen = false
|
||||
var doneSeen = false
|
||||
var finishReason = "missing"
|
||||
var reasoningContentPresent = false
|
||||
val assistantContent = StringBuilder()
|
||||
while (true) {
|
||||
val line = channel.readLineStrict(
|
||||
@@ -237,6 +269,7 @@ class KtorDeepSeekClient(
|
||||
}
|
||||
val eventUsage = extractUsage(data)
|
||||
eventUsage?.let { usage = it }
|
||||
if (hasStreamReasoningContent(data)) reasoningContentPresent = true
|
||||
val choices = runCatching {
|
||||
event["choices"]?.jsonArray
|
||||
?: throw IllegalArgumentException("choices is missing")
|
||||
@@ -250,13 +283,14 @@ class KtorDeepSeekClient(
|
||||
if (choice == null && eventUsage == null) {
|
||||
throw DeepSeekProviderException("DeepSeek returned an empty non-usage event")
|
||||
}
|
||||
val finishReason = choice?.get("finish_reason")
|
||||
if (finishReason != null && finishReason !is JsonNull) {
|
||||
if (!finishReason.jsonPrimitive.isString ||
|
||||
finishReason.jsonPrimitive.content.isBlank()
|
||||
val finishReasonElement = choice?.get("finish_reason")
|
||||
if (finishReasonElement != null && finishReasonElement !is JsonNull) {
|
||||
if (!finishReasonElement.jsonPrimitive.isString ||
|
||||
finishReasonElement.jsonPrimitive.content.isBlank()
|
||||
) {
|
||||
throw DeepSeekProviderException("DeepSeek returned an invalid finish reason")
|
||||
}
|
||||
finishReason = normalizeFinishReason(finishReasonElement.jsonPrimitive.content)
|
||||
terminalChoiceSeen = true
|
||||
}
|
||||
extractStreamContent(data)?.let { chunk ->
|
||||
@@ -270,7 +304,13 @@ class KtorDeepSeekClient(
|
||||
output.emit(encoded)
|
||||
}
|
||||
if (!doneSeen) throw DeepSeekProviderException("DeepSeek stream closed before [DONE]")
|
||||
if (assistantContent.isBlank()) throw DeepSeekEmptyResultException()
|
||||
if (assistantContent.isBlank()) {
|
||||
throw DeepSeekEmptyResultException(
|
||||
finishReason = finishReason,
|
||||
reasoningContentPresent = reasoningContentPresent,
|
||||
usagePresent = usage != null,
|
||||
)
|
||||
}
|
||||
if (request.capability == GatewayCapability.AGENT) {
|
||||
validateAgentContent(assistantContent.toString())
|
||||
}
|
||||
@@ -354,6 +394,58 @@ class KtorDeepSeekClient(
|
||||
?.content
|
||||
}.getOrNull()
|
||||
|
||||
private fun hasStreamReasoningContent(payload: String): Boolean = runCatching {
|
||||
json.parseToJsonElement(payload)
|
||||
.jsonObject["choices"]
|
||||
?.let { choices -> choices.jsonArray.firstOrNull() }
|
||||
?.jsonObject
|
||||
?.get("delta")
|
||||
?.jsonObject
|
||||
?.get("reasoning_content")
|
||||
?.jsonPrimitive
|
||||
?.takeIf { it.isString }
|
||||
?.content
|
||||
?.isNotBlank() == true
|
||||
}.getOrDefault(false)
|
||||
|
||||
private fun emptyResultMetadata(payload: String): DeepSeekEmptyResultMetadata {
|
||||
val root = runCatching { json.parseToJsonElement(payload).jsonObject }.getOrNull()
|
||||
?: return DeepSeekEmptyResultMetadata()
|
||||
val choice = runCatching {
|
||||
root["choices"]?.jsonArray?.firstOrNull()?.jsonObject
|
||||
}.getOrNull()
|
||||
val message = runCatching { choice?.get("message")?.jsonObject }.getOrNull()
|
||||
val rawFinishReason = runCatching {
|
||||
choice?.get("finish_reason")
|
||||
?.takeUnless { it is JsonNull }
|
||||
?.jsonPrimitive
|
||||
?.takeIf { it.isString }
|
||||
?.content
|
||||
}.getOrNull()
|
||||
val reasoningContentPresent = runCatching {
|
||||
message?.get("reasoning_content")
|
||||
?.takeUnless { it is JsonNull }
|
||||
?.jsonPrimitive
|
||||
?.takeIf { it.isString }
|
||||
?.content
|
||||
?.isNotBlank() == true
|
||||
}.getOrDefault(false)
|
||||
return DeepSeekEmptyResultMetadata(
|
||||
finishReason = normalizeFinishReason(rawFinishReason),
|
||||
reasoningContentPresent = reasoningContentPresent,
|
||||
usagePresent = root["usage"] != null && root["usage"] !is JsonNull,
|
||||
)
|
||||
}
|
||||
|
||||
private fun normalizeFinishReason(value: String?): String =
|
||||
when (value?.trim()?.lowercase()) {
|
||||
"stop", "length", "content_filter", "tool_calls", "insufficient_system_resource" ->
|
||||
value.trim().lowercase()
|
||||
|
||||
null, "" -> "missing"
|
||||
else -> "other"
|
||||
}
|
||||
|
||||
private suspend fun ByteReadChannel.readBounded(): ByteArray {
|
||||
val bytes = readRemaining(GatewayLimits.MAX_UPSTREAM_RESPONSE_BYTES.toLong() + 1L)
|
||||
.readByteArray()
|
||||
@@ -439,10 +531,20 @@ class DeepSeekConfigurationException(message: String) : IllegalStateException(me
|
||||
|
||||
class DeepSeekProviderException(message: String) : RuntimeException(message)
|
||||
class DeepSeekUsageException(message: String) : ProviderCompletionException(message)
|
||||
class DeepSeekEmptyResultException : RuntimeException("DeepSeek returned an empty result")
|
||||
class DeepSeekEmptyResultException(
|
||||
val finishReason: String = "missing",
|
||||
val reasoningContentPresent: Boolean = false,
|
||||
val usagePresent: Boolean = false,
|
||||
) : RuntimeException("DeepSeek returned an empty result")
|
||||
|
||||
private data class DeepSeekUsage(
|
||||
val total: Long,
|
||||
val input: Long?,
|
||||
val output: Long?,
|
||||
)
|
||||
|
||||
private data class DeepSeekEmptyResultMetadata(
|
||||
val finishReason: String = "missing",
|
||||
val reasoningContentPresent: Boolean = false,
|
||||
val usagePresent: Boolean = false,
|
||||
)
|
||||
|
||||
+5
-1
@@ -231,7 +231,11 @@ class SaucSequenceValidator {
|
||||
?: throw SaucProtocolException("SAUC server frame omitted its sequence")
|
||||
val expected = Math.addExact(lastSequence, 1)
|
||||
if (frame.isLast) {
|
||||
if (sequence >= 0 || Math.abs(sequence.toLong()) != expected.toLong()) {
|
||||
// Volcengine's protocol table documents a negative final sequence,
|
||||
// while its examples and production service return the same
|
||||
// strictly increasing sequence as a positive value. Accept both
|
||||
// representations without weakening continuity validation.
|
||||
if (Math.abs(sequence.toLong()) != expected.toLong()) {
|
||||
throw SaucProtocolException("SAUC final sequence is invalid")
|
||||
}
|
||||
finalSeen = true
|
||||
|
||||
Reference in New Issue
Block a user