From dd1bed253632c28b53a67a8094704df8855f6da7 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 27 Dec 2025 23:42:13 +0500 Subject: [PATCH] . --- .../service/OpenAIAnalyticsService.java | 73 ++++++++++++++----- src/main/resources/application.properties | 1 + 2 files changed, 54 insertions(+), 20 deletions(-) diff --git a/src/main/java/kz/konturai/parser/service/OpenAIAnalyticsService.java b/src/main/java/kz/konturai/parser/service/OpenAIAnalyticsService.java index 9d3a999..81ccc9f 100644 --- a/src/main/java/kz/konturai/parser/service/OpenAIAnalyticsService.java +++ b/src/main/java/kz/konturai/parser/service/OpenAIAnalyticsService.java @@ -44,6 +44,9 @@ public class OpenAIAnalyticsService { @Value("${openai.timeoutMs:90000}") private long timeoutMs; + @Value("${openai.timeoutMs.instruction:180000}") + private long timeoutMsInstruction; + @Value("${openai.retry.maxAttempts:5}") private int maxRetryAttempts; @@ -76,7 +79,8 @@ public class OpenAIAnalyticsService { private String openaiUrl; /** - * Helper for building OpenAI chat-completions request body (extracted for unit testing). + * Helper for building OpenAI chat-completions request body (extracted for unit + * testing). */ Map buildChatCompletionsRequestBody( String model, @@ -92,7 +96,8 @@ public class OpenAIAnalyticsService { } /** - * Helper for building chat messages list (system + user) (extracted for unit testing). + * Helper for building chat messages list (system + user) (extracted for unit + * testing). */ List> buildChatMessages(String userPrompt, String systemPrompt) { List> messages = new ArrayList<>(); @@ -237,6 +242,7 @@ public class OpenAIAnalyticsService { .retrieve() .bodyToMono(new ParameterizedTypeReference>() { }) + .timeout(Duration.ofMillis(timeoutMs)) .retryWhen(createRetrySpec("generate")) .onErrorResume(err -> { logger.error("OpenAI request failed after retries: {} - {}", err.getMessage(), @@ -251,10 +257,12 @@ public class OpenAIAnalyticsService { } else if (wcre.getStatusCode().value() >= 500) { logger.error("OpenAI API server error after all retry attempts."); } + } else if (err instanceof java.util.concurrent.TimeoutException) { + logger.error("OpenAI API request timed out after {}ms", timeoutMs); } return Mono.empty(); }) - .block(Duration.ofMillis(timeoutMs)); + .block(Duration.ofMillis(timeoutMs + 10000)); // Add buffer for block timeout if (response == null) { return null; @@ -282,7 +290,12 @@ public class OpenAIAnalyticsService { logger.warn("Interrupted while waiting for OpenAI request slot: {}", e.getMessage()); return null; } catch (Exception e) { - logger.warn("Error calling OpenAI generate: {}", e.getMessage()); + // .block() throws IllegalStateException on timeout, not TimeoutException + if (e.getMessage() != null && e.getMessage().contains("Timeout")) { + logger.warn("Timeout waiting for OpenAI response (block timeout): {}", e.getMessage()); + } else { + logger.warn("Error calling OpenAI generate: {} - {}", e.getMessage(), e.getClass().getSimpleName()); + } return null; } } @@ -312,12 +325,15 @@ public class OpenAIAnalyticsService { /** * Генерирует текст с поддержкой system prompt и override для max_tokens. * - * @param text Текст (контекст), который будет добавлен к instruction + * @param text Текст (контекст), который будет добавлен к + * instruction * @param instruction Основная инструкция для модели - * @param language Язык ответа (исторически передается, но не влияет на протокол) + * @param language Язык ответа (исторически передается, но не влияет на + * протокол) * @param modelName Название модели OpenAI * @param systemPrompt System prompt (если null/blank — не добавляется) - * @param maxTokensOverride Override для max_tokens (если null — используется дефолт) + * @param maxTokensOverride Override для max_tokens (если null — используется + * дефолт) * @return Сгенерированный текст */ public String generateWithInstructionWithModel( @@ -346,7 +362,8 @@ public class OpenAIAnalyticsService { } try { - logger.debug("Calling OpenAI API with model: {}", modelName); + logger.debug("Calling OpenAI API with model: {} (timeout: {}ms)", modelName, timeoutMsInstruction); + // Use timeout operator in the reactive chain so timeouts can be retried Map response = this.webClient.post() .contentType(MediaType.APPLICATION_JSON) .accept(MediaType.APPLICATION_JSON) @@ -354,6 +371,7 @@ public class OpenAIAnalyticsService { .retrieve() .bodyToMono(new ParameterizedTypeReference>() { }) + .timeout(Duration.ofMillis(timeoutMsInstruction)) .retryWhen(createRetrySpec("generateWithInstruction")) .onErrorResume(err -> { logger.error("OpenAI request failed after retries: {} - {}", err.getMessage(), @@ -368,10 +386,12 @@ public class OpenAIAnalyticsService { } else if (wcre.getStatusCode().value() >= 500) { logger.error("OpenAI API server error after all retry attempts."); } + } else if (err instanceof java.util.concurrent.TimeoutException) { + logger.error("OpenAI API request timed out after {}ms", timeoutMsInstruction); } return Mono.empty(); }) - .block(Duration.ofMillis(timeoutMs)); + .block(Duration.ofMillis(timeoutMsInstruction + 10000)); // Add buffer for block timeout if (response == null) { return null; @@ -399,7 +419,13 @@ public class OpenAIAnalyticsService { logger.warn("Interrupted while waiting for OpenAI request slot: {}", e.getMessage()); return null; } catch (Exception e) { - logger.warn("Error calling OpenAI generate (generic): {}", e.getMessage()); + // .block() throws IllegalStateException on timeout, not TimeoutException + if (e.getMessage() != null && e.getMessage().contains("Timeout")) { + logger.warn("Timeout waiting for OpenAI response (block timeout): {}", e.getMessage()); + } else { + logger.warn("Error calling OpenAI generate (generic): {} - {}", e.getMessage(), + e.getClass().getSimpleName()); + } return null; } } @@ -407,11 +433,11 @@ public class OpenAIAnalyticsService { private RetryBackoffSpec createRetrySpec(String operation) { // Store Retry-After duration for 429 errors AtomicReference retryAfterDuration = new AtomicReference<>(null); - + // Use the higher max attempts to handle rate limit errors better // This will apply to all retries, but rate limit errors need more attempts int effectiveMaxAttempts = Math.max(maxRetryAttempts, rateLimitMaxAttempts); - + return Retry.backoff(effectiveMaxAttempts, Duration.ofMillis(initialRetryDelayMs)) .maxBackoff(Duration.ofMillis(maxRetryDelayMs)) .multiplier(retryMultiplier) @@ -419,7 +445,7 @@ public class OpenAIAnalyticsService { if (throwable instanceof WebClientResponseException) { WebClientResponseException wcre = (WebClientResponseException) throwable; int statusCode = wcre.getStatusCode().value(); - + if (statusCode == 429) { // Extract Retry-After header for 429 errors String retryAfterHeader = wcre.getHeaders().getFirst("Retry-After"); @@ -442,22 +468,28 @@ public class OpenAIAnalyticsService { } else { retryAfterDuration.set(null); } - + // Only retry on 429 (rate limit) and 5xx (server errors) return statusCode == 429 || statusCode >= 500; } // Retry on network errors (timeouts, connection issues) retryAfterDuration.set(null); - return throwable instanceof java.util.concurrent.TimeoutException + boolean isTimeout = throwable instanceof java.util.concurrent.TimeoutException + || (throwable.getMessage() != null && throwable.getMessage().contains("Timeout")); + if (isTimeout) { + logger.warn("OpenAI API timeout detected for operation '{}'. Will retry.", operation); + } + return isTimeout || throwable instanceof java.net.ConnectException || throwable instanceof java.io.IOException; }) .doBeforeRetry(retrySignal -> { long attempt = retrySignal.totalRetries() + 1; Throwable failure = retrySignal.failure(); - + // If we have a Retry-After duration for 429 errors, sleep for that duration - // This ensures we respect the API's requested delay before Reactor applies its backoff + // This ensures we respect the API's requested delay before Reactor applies its + // backoff Duration customDelay = retryAfterDuration.get(); if (customDelay != null && failure instanceof WebClientResponseException) { WebClientResponseException wcre = (WebClientResponseException) failure; @@ -474,14 +506,15 @@ public class OpenAIAnalyticsService { } } } - + // Log retry attempt if (failure instanceof WebClientResponseException) { WebClientResponseException wcre = (WebClientResponseException) failure; int statusCode = wcre.getStatusCode().value(); int maxAttempts = statusCode == 429 ? rateLimitMaxAttempts : maxRetryAttempts; if (statusCode == 429 && customDelay == null) { - logger.warn("OpenAI API returned 429 for operation '{}'. Will retry with exponential backoff (attempt {}/{})", + logger.warn( + "OpenAI API returned 429 for operation '{}'. Will retry with exponential backoff (attempt {}/{})", operation, attempt, maxAttempts); } else if (statusCode != 429) { logger.warn("OpenAI API returned {} for operation '{}'. Will retry (attempt {}/{})", @@ -506,5 +539,5 @@ public class OpenAIAnalyticsService { return retrySignal.failure(); }); } - + } diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 0fa1d1b..4c8070b 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -64,6 +64,7 @@ openai.api.url=https://api.openai.com/v1/chat/completions openai.model.name=gpt-4o-mini openai.model.name.text=gpt-4o openai.timeoutMs=90000 +openai.timeoutMs.instruction=180000 openai.retry.maxAttempts=3 openai.retry.initialDelayMs=2000 openai.retry.maxDelayMs=60000