diff --git a/src/main/java/kz/konturai/parser/service/OpenAIAnalyticsService.java b/src/main/java/kz/konturai/parser/service/OpenAIAnalyticsService.java index 613d862..0cac826 100644 --- a/src/main/java/kz/konturai/parser/service/OpenAIAnalyticsService.java +++ b/src/main/java/kz/konturai/parser/service/OpenAIAnalyticsService.java @@ -22,6 +22,8 @@ import org.springframework.http.client.reactive.ReactorClientHttpConnector; import org.springframework.web.reactive.function.client.WebClientResponseException; import reactor.util.retry.Retry; import reactor.util.retry.RetryBackoffSpec; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; import jakarta.annotation.PostConstruct; @Service @@ -49,9 +51,14 @@ public class OpenAIAnalyticsService { @Value("${openai.retry.multiplier:2.0}") private double retryMultiplier; + @Value("${openai.rateLimit.maxConcurrentRequests:2}") + private int maxConcurrentRequests; + @Value("${openai.api.key}") private String apiKey; + private Semaphore requestSemaphore; + private String openaiUrl; public OpenAIAnalyticsService( @@ -82,7 +89,10 @@ public class OpenAIAnalyticsService { .defaultHeader("Authorization", "Bearer " + apiKey) .build(); - logger.info("OpenAIAnalyticsService initialized with model: {}", modelName); + this.requestSemaphore = new Semaphore(maxConcurrentRequests, true); + + logger.info("OpenAIAnalyticsService initialized with model: {} and max concurrent requests: {}", + modelName, maxConcurrentRequests); } public MarketItem.Analytics analyzeText(String text) { @@ -164,46 +174,61 @@ public class OpenAIAnalyticsService { requestBody.put("temperature", 0.3); try { - Map response = this.webClient.post() - .contentType(MediaType.APPLICATION_JSON) - .accept(MediaType.APPLICATION_JSON) - .bodyValue(requestBody) - .retrieve() - .bodyToMono(new ParameterizedTypeReference>() { - }) - .retryWhen(createRetrySpec("generate")) - .onErrorResume(err -> { - logger.error("OpenAI request failed after retries: {} - {}", err.getMessage(), - err.getClass().getSimpleName()); - if (err instanceof WebClientResponseException) { - WebClientResponseException wcre = (WebClientResponseException) err; - if (wcre.getStatusCode().value() == 401) { - logger.error( - "OpenAI API key is invalid or expired. Please check your openai.api.key configuration."); - } else if (wcre.getStatusCode().value() == 429) { - logger.error("OpenAI API rate limit exceeded after all retry attempts."); - } else if (wcre.getStatusCode().value() >= 500) { - logger.error("OpenAI API server error after all retry attempts."); - } - } - return Mono.empty(); - }) - .block(Duration.ofMillis(timeoutMs)); - - if (response == null) { + // Acquire semaphore permit before making request + if (!requestSemaphore.tryAcquire(30, TimeUnit.SECONDS)) { + logger.warn("Timeout waiting for OpenAI request slot. Skipping request."); return null; } - @SuppressWarnings("unchecked") - List> choices = (List>) response.get("choices"); - if (choices != null && !choices.isEmpty()) { - Map firstChoice = choices.get(0); - Map messageObj = (Map) firstChoice.get("message"); - if (messageObj != null) { - Object content = messageObj.get("content"); - return content == null ? null : String.valueOf(content).trim(); + try { + Map response = this.webClient.post() + .contentType(MediaType.APPLICATION_JSON) + .accept(MediaType.APPLICATION_JSON) + .bodyValue(requestBody) + .retrieve() + .bodyToMono(new ParameterizedTypeReference>() { + }) + .retryWhen(createRetrySpec("generate")) + .onErrorResume(err -> { + logger.error("OpenAI request failed after retries: {} - {}", err.getMessage(), + err.getClass().getSimpleName()); + if (err instanceof WebClientResponseException) { + WebClientResponseException wcre = (WebClientResponseException) err; + if (wcre.getStatusCode().value() == 401) { + logger.error( + "OpenAI API key is invalid or expired. Please check your openai.api.key configuration."); + } else if (wcre.getStatusCode().value() == 429) { + logger.error("OpenAI API rate limit exceeded after all retry attempts."); + } else if (wcre.getStatusCode().value() >= 500) { + logger.error("OpenAI API server error after all retry attempts."); + } + } + return Mono.empty(); + }) + .block(Duration.ofMillis(timeoutMs)); + + if (response == null) { + return null; } + + @SuppressWarnings("unchecked") + List> choices = (List>) response.get("choices"); + if (choices != null && !choices.isEmpty()) { + Map firstChoice = choices.get(0); + Map messageObj = (Map) firstChoice.get("message"); + if (messageObj != null) { + Object content = messageObj.get("content"); + return content == null ? null : String.valueOf(content).trim(); + } + } + return null; + } finally { + // Always release semaphore permit + requestSemaphore.release(); } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + logger.warn("Interrupted while waiting for OpenAI request slot: {}", e.getMessage()); return null; } catch (Exception e) { logger.warn("Error calling OpenAI generate: {}", e.getMessage()); @@ -229,48 +254,63 @@ public class OpenAIAnalyticsService { requestBody.put("temperature", 0.3); try { - Map response = this.webClient.post() - .contentType(MediaType.APPLICATION_JSON) - .accept(MediaType.APPLICATION_JSON) - .bodyValue(requestBody) - .retrieve() - .bodyToMono(new ParameterizedTypeReference>() { - }) - .retryWhen(createRetrySpec("generateWithInstruction")) - .onErrorResume(err -> { - logger.error("OpenAI request failed after retries: {} - {}", err.getMessage(), - err.getClass().getSimpleName()); - if (err instanceof WebClientResponseException) { - WebClientResponseException wcre = (WebClientResponseException) err; - if (wcre.getStatusCode().value() == 401) { - logger.error( - "OpenAI API key is invalid or expired. Please check your openai.api.key configuration."); - } else if (wcre.getStatusCode().value() == 429) { - logger.error("OpenAI API rate limit exceeded after all retry attempts."); - } else if (wcre.getStatusCode().value() >= 500) { - logger.error("OpenAI API server error after all retry attempts."); - } - } - return Mono.empty(); - }) - .block(Duration.ofMillis(timeoutMs)); - - if (response == null) { + // Acquire semaphore permit before making request + if (!requestSemaphore.tryAcquire(30, TimeUnit.SECONDS)) { + logger.warn("Timeout waiting for OpenAI request slot. Skipping request."); return null; } - @SuppressWarnings("unchecked") - List> choices = (List>) response.get("choices"); - if (choices != null && !choices.isEmpty()) { - Map firstChoice = choices.get(0); - Map messageObj = (Map) firstChoice.get("message"); - if (messageObj != null) { - Object content = messageObj.get("content"); - String result = content == null ? null : String.valueOf(content).trim(); - System.out.println("OpenAI response: " + result); - return result; + try { + Map response = this.webClient.post() + .contentType(MediaType.APPLICATION_JSON) + .accept(MediaType.APPLICATION_JSON) + .bodyValue(requestBody) + .retrieve() + .bodyToMono(new ParameterizedTypeReference>() { + }) + .retryWhen(createRetrySpec("generateWithInstruction")) + .onErrorResume(err -> { + logger.error("OpenAI request failed after retries: {} - {}", err.getMessage(), + err.getClass().getSimpleName()); + if (err instanceof WebClientResponseException) { + WebClientResponseException wcre = (WebClientResponseException) err; + if (wcre.getStatusCode().value() == 401) { + logger.error( + "OpenAI API key is invalid or expired. Please check your openai.api.key configuration."); + } else if (wcre.getStatusCode().value() == 429) { + logger.error("OpenAI API rate limit exceeded after all retry attempts."); + } else if (wcre.getStatusCode().value() >= 500) { + logger.error("OpenAI API server error after all retry attempts."); + } + } + return Mono.empty(); + }) + .block(Duration.ofMillis(timeoutMs)); + + if (response == null) { + return null; } + + @SuppressWarnings("unchecked") + List> choices = (List>) response.get("choices"); + if (choices != null && !choices.isEmpty()) { + Map firstChoice = choices.get(0); + Map messageObj = (Map) firstChoice.get("message"); + if (messageObj != null) { + Object content = messageObj.get("content"); + String result = content == null ? null : String.valueOf(content).trim(); + System.out.println("OpenAI response: " + result); + return result; + } + } + return null; + } finally { + // Always release semaphore permit + requestSemaphore.release(); } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + 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()); @@ -286,13 +326,9 @@ public class OpenAIAnalyticsService { if (throwable instanceof WebClientResponseException) { WebClientResponseException wcre = (WebClientResponseException) throwable; int statusCode = wcre.getStatusCode().value(); - // Retry on 429 (rate limit) and 5xx (server errors) - boolean shouldRetry = statusCode == 429 || statusCode >= 500; - if (shouldRetry) { - logger.warn("OpenAI API returned {} for operation '{}'. Will retry...", statusCode, - operation); - } - return shouldRetry; + // Only retry on 429 (rate limit) and 5xx (server errors) + // Don't retry on 401, 403, etc. + return statusCode == 429 || statusCode >= 500; } // Retry on network errors (timeouts, connection issues) return throwable instanceof java.util.concurrent.TimeoutException @@ -301,11 +337,35 @@ public class OpenAIAnalyticsService { }) .doBeforeRetry(retrySignal -> { long attempt = retrySignal.totalRetries() + 1; - logger.info("Retrying OpenAI request for operation '{}' (attempt {}/{})", - operation, attempt, maxRetryAttempts); + Throwable failure = retrySignal.failure(); + + Duration retryAfter = null; + if (failure instanceof WebClientResponseException) { + WebClientResponseException wcre = (WebClientResponseException) failure; + if (wcre.getStatusCode().value() == 429) { + // Try to extract Retry-After header + String retryAfterHeader = wcre.getHeaders().getFirst("Retry-After"); + if (retryAfterHeader != null) { + try { + int seconds = Integer.parseInt(retryAfterHeader); + retryAfter = Duration.ofSeconds(seconds); + logger.warn( + "OpenAI API returned 429 for operation '{}'. Retry-After: {} seconds. Will retry in {}ms (attempt {}/{})", + operation, seconds, retryAfter.toMillis(), attempt, maxRetryAttempts); + } catch (NumberFormatException e) { + // Ignore if header is not a number + } + } + } + } + + if (retryAfter == null) { + logger.warn("OpenAI API returned error for operation '{}'. Will retry (attempt {}/{})", + operation, attempt, maxRetryAttempts); + } }) .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> { - logger.error("OpenAI request for operation '{}' exhausted all {} retry attempts", + logger.error("OpenAI request for operation '{}' exhausted all {} retry attempts. Giving up.", operation, maxRetryAttempts); return retrySignal.failure(); }); diff --git a/src/main/java/kz/konturai/parser/service/OpenAiChartService.java b/src/main/java/kz/konturai/parser/service/OpenAiChartService.java index 5ae9236..ba6b804 100644 --- a/src/main/java/kz/konturai/parser/service/OpenAiChartService.java +++ b/src/main/java/kz/konturai/parser/service/OpenAiChartService.java @@ -19,6 +19,9 @@ import org.slf4j.LoggerFactory; import org.springframework.web.reactive.function.client.WebClientResponseException; import reactor.util.retry.Retry; import reactor.util.retry.RetryBackoffSpec; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import jakarta.annotation.PostConstruct; @Service public class OpenAiChartService { @@ -45,6 +48,11 @@ public class OpenAiChartService { @Value("${openai.retry.multiplier:2.0}") private double retryMultiplier; + @Value("${openai.rateLimit.maxConcurrentRequests:2}") + private int maxConcurrentRequests; + + private Semaphore requestSemaphore; + public OpenAiChartService( @Value("${openai.api.url:https://api.openai.com/v1/chat/completions}") String apiUrl, @Value("${openai.api.key:}") String apiKey) { @@ -66,6 +74,12 @@ public class OpenAiChartService { .build(); } + @PostConstruct + public void initialize() { + this.requestSemaphore = new Semaphore(maxConcurrentRequests, true); + logger.info("OpenAiChartService initialized with max concurrent requests: {}", maxConcurrentRequests); + } + public Mono getChartDataJson(String aggregatedLearnings) { return getChartDataJson(aggregatedLearnings, "ru"); } @@ -99,60 +113,86 @@ public class OpenAiChartService { logger.info("Calling OpenAI API for task '{}' with model '{}'.", taskName, modelName); logger.info("OpenAI Prompt: {}", prompt); - return this.webClient.post() - .contentType(MediaType.APPLICATION_JSON) - .accept(MediaType.APPLICATION_JSON) - .bodyValue(body) - .retrieve() - .bodyToMono(new ParameterizedTypeReference>() { - }) - .timeout(Duration.ofMillis(timeoutMs)) - .retryWhen(createRetrySpec(taskName)) - .doOnSuccess(response -> { - // 4. Логируем успешный, но еще не обработанный ответ - logger.info("Raw OpenAI response for task '{}': {}", taskName, response); - }) - .map(resp -> { - try { - List> choices = (List>) resp.get("choices"); - if (choices == null || choices.isEmpty()) { - // 5. Логируем аномальные, но не ошибочные ситуации - logger.warn("OpenAI response for task '{}' contained no 'choices'.", taskName); - return null; - } - Map message = (Map) choices.get(0).get("message"); - if (message == null) { - logger.warn("OpenAI choice for task '{}' contained no 'message'.", taskName); - return null; - } - Object content = message.get("content"); - if (content == null) { - logger.warn("OpenAI message for task '{}' contained no 'content'.", taskName); - return null; - } - String responseContent = String.valueOf(content); - logger.info("Successfully extracted content from OpenAI for task '{}'. Content length: {}", - taskName, responseContent.length()); - return responseContent; - } catch (Exception e) { - // 6. Логируем ошибку парсинга, если структура ответа неожиданная - logger.error("Failed to parse OpenAI response for task '{}'. Response body: {}", taskName, resp, - e); - return null; + // Acquire semaphore permit before making request + return Mono.fromCallable(() -> { + try { + if (!requestSemaphore.tryAcquire(30, TimeUnit.SECONDS)) { + logger.warn("Timeout waiting for OpenAI request slot for task '{}'. Skipping request.", taskName); + return null; + } + return requestSemaphore; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + logger.warn("Interrupted while waiting for OpenAI request slot for task '{}'", taskName); + return null; + } + }) + .flatMap(semaphore -> { + if (semaphore == null) { + return Mono.empty(); } - }) - .doOnError(e -> { - // 7. Логируем ошибки сети или HTTP-статусов (4xx, 5xx) - if (e instanceof WebClientResponseException) { - WebClientResponseException wcre = (WebClientResponseException) e; - logger.error("Error from OpenAI API for task '{}'. Status: {}, Body: {}", - taskName, wcre.getStatusCode(), wcre.getResponseBodyAsString(), e); - } else { - logger.error("Generic WebClient error for task '{}'.", taskName, e); - } - }) - .onErrorResume(e -> Mono.empty()); // В случае любой ошибки возвращаем пустой результат, чтобы не - // прерывать цепочку вызовов + return this.webClient.post() + .contentType(MediaType.APPLICATION_JSON) + .accept(MediaType.APPLICATION_JSON) + .bodyValue(body) + .retrieve() + .bodyToMono(new ParameterizedTypeReference>() { + }) + .timeout(Duration.ofMillis(timeoutMs)) + .retryWhen(createRetrySpec(taskName)) + .doFinally(signalType -> { + // Always release semaphore permit + semaphore.release(); + }) + .doOnSuccess(response -> { + // 4. Логируем успешный, но еще не обработанный ответ + logger.info("Raw OpenAI response for task '{}': {}", taskName, response); + }) + .map(resp -> { + try { + List> choices = (List>) resp.get("choices"); + if (choices == null || choices.isEmpty()) { + // 5. Логируем аномальные, но не ошибочные ситуации + logger.warn("OpenAI response for task '{}' contained no 'choices'.", taskName); + return null; + } + Map message = (Map) choices.get(0).get("message"); + if (message == null) { + logger.warn("OpenAI choice for task '{}' contained no 'message'.", taskName); + return null; + } + Object content = message.get("content"); + if (content == null) { + logger.warn("OpenAI message for task '{}' contained no 'content'.", taskName); + return null; + } + String responseContent = String.valueOf(content); + logger.info( + "Successfully extracted content from OpenAI for task '{}'. Content length: {}", + taskName, responseContent.length()); + return responseContent; + } catch (Exception e) { + // 6. Логируем ошибку парсинга, если структура ответа неожиданная + logger.error("Failed to parse OpenAI response for task '{}'. Response body: {}", + taskName, resp, + e); + return null; + } + }) + .doOnError(e -> { + // 7. Логируем ошибки сети или HTTP-статусов (4xx, 5xx) + if (e instanceof WebClientResponseException) { + WebClientResponseException wcre = (WebClientResponseException) e; + logger.error("Error from OpenAI API for task '{}'. Status: {}, Body: {}", + taskName, wcre.getStatusCode(), wcre.getResponseBodyAsString(), e); + } else { + logger.error("Generic WebClient error for task '{}'.", taskName, e); + } + }) + .onErrorResume(e -> Mono.empty()); // В случае любой ошибки возвращаем пустой результат, + // чтобы не + // прерывать цепочку вызовов + }); } private String buildChartDataPrompt(String learningsAsString, String language) { @@ -265,13 +305,9 @@ public class OpenAiChartService { if (throwable instanceof WebClientResponseException) { WebClientResponseException wcre = (WebClientResponseException) throwable; int statusCode = wcre.getStatusCode().value(); - // Retry on 429 (rate limit) and 5xx (server errors) - boolean shouldRetry = statusCode == 429 || statusCode >= 500; - if (shouldRetry) { - logger.warn("OpenAI API returned {} for operation '{}'. Will retry...", statusCode, - operation); - } - return shouldRetry; + // Only retry on 429 (rate limit) and 5xx (server errors) + // Don't retry on 401, 403, etc. + return statusCode == 429 || statusCode >= 500; } // Retry on network errors (timeouts, connection issues) return throwable instanceof java.util.concurrent.TimeoutException @@ -280,11 +316,35 @@ public class OpenAiChartService { }) .doBeforeRetry(retrySignal -> { long attempt = retrySignal.totalRetries() + 1; - logger.info("Retrying OpenAI request for operation '{}' (attempt {}/{})", - operation, attempt, maxRetryAttempts); + Throwable failure = retrySignal.failure(); + + Duration retryAfter = null; + if (failure instanceof WebClientResponseException) { + WebClientResponseException wcre = (WebClientResponseException) failure; + if (wcre.getStatusCode().value() == 429) { + // Try to extract Retry-After header + String retryAfterHeader = wcre.getHeaders().getFirst("Retry-After"); + if (retryAfterHeader != null) { + try { + int seconds = Integer.parseInt(retryAfterHeader); + retryAfter = Duration.ofSeconds(seconds); + logger.warn( + "OpenAI API returned 429 for operation '{}'. Retry-After: {} seconds. Will retry in {}ms (attempt {}/{})", + operation, seconds, retryAfter.toMillis(), attempt, maxRetryAttempts); + } catch (NumberFormatException e) { + // Ignore if header is not a number + } + } + } + } + + if (retryAfter == null) { + logger.warn("OpenAI API returned error for operation '{}'. Will retry (attempt {}/{})", + operation, attempt, maxRetryAttempts); + } }) .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> { - logger.error("OpenAI request for operation '{}' exhausted all {} retry attempts", + logger.error("OpenAI request for operation '{}' exhausted all {} retry attempts. Giving up.", operation, maxRetryAttempts); return retrySignal.failure(); }); diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 34e3e49..fc2a200 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -58,10 +58,11 @@ openai.api.key=sk-proj-zEsggv3MCgvZ2aQbRhpCfyZPKn-isyylNhO16dekXGpM3yzi9H4g0zSjp openai.api.url=https://api.openai.com/v1/chat/completions openai.model.name=gpt-4o-mini openai.timeoutMs=90000 -openai.retry.maxAttempts=5 -openai.retry.initialDelayMs=1000 +openai.retry.maxAttempts=3 +openai.retry.initialDelayMs=2000 openai.retry.maxDelayMs=60000 openai.retry.multiplier=2.0 +openai.rateLimit.maxConcurrentRequests=2 # Email Configuration spring.mail.host=smtp.gmail.com