diff --git a/src/main/java/kz/konturai/parser/service/MarketingAnalysisService.java b/src/main/java/kz/konturai/parser/service/MarketingAnalysisService.java index 3ed3dea..cfce573 100644 --- a/src/main/java/kz/konturai/parser/service/MarketingAnalysisService.java +++ b/src/main/java/kz/konturai/parser/service/MarketingAnalysisService.java @@ -20,21 +20,11 @@ import com.fasterxml.jackson.core.JsonParseException; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; -import jakarta.annotation.PreDestroy; - import java.time.LocalDateTime; import java.util.*; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Semaphore; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; @Service public class MarketingAnalysisService { - private final Semaphore openAiSemaphore = new Semaphore(3); // Разрешить только 3 активных запроса к OpenAI на всё - // приложение private static final String AI_MANDATORY_JSON_USAGE_WARNING = "ВНИМАНИЕ: Если в предоставленном JSON есть данные, ты ОБЯЗАН использовать их и вставить соответствующий тег [[CHART_...]]. " + "Никогда не пиши, что данных нет, если JSON не пустой.\n\n"; @@ -359,7 +349,6 @@ public class MarketingAnalysisService { private static final Logger logger = LoggerFactory.getLogger(MarketingAnalysisService.class); private static final int MAX_RETRY_ATTEMPTS = 3; private static final long RETRY_DELAY_MS = 1000; - private static final AtomicInteger SECTION_THREAD_COUNTER = new AtomicInteger(0); private final MarketingAnalysisRepository repository; private final OpenAIAnalyticsService openAIAnalyticsService; @@ -368,18 +357,6 @@ public class MarketingAnalysisService { private final SerperSearchService serperSearchService; private final ObjectMapper objectMapper = new ObjectMapper(); - /** - * Ограничиваем параллелизм генерации секций, чтобы не "забивать" семафор - * OpenAIAnalyticsService - * и не получать: "Timeout waiting for OpenAI request slot. Skipping request." - */ - private final ExecutorService sectionGenerationExecutor = Executors.newFixedThreadPool(2, r -> { - Thread t = new Thread(r); - t.setName("report-sections-" + SECTION_THREAD_COUNTER.incrementAndGet()); - t.setDaemon(true); - return t; - }); - @Value("${openai.model.name.text:gpt-4o}") private String textModelName; @@ -501,15 +478,6 @@ public class MarketingAnalysisService { promptBuilder.append("\n"); } - @PreDestroy - public void shutdownExecutors() { - try { - sectionGenerationExecutor.shutdown(); - } catch (Exception e) { - logger.warn("Failed to shutdown sectionGenerationExecutor: {}", e.getMessage()); - } - } - public List startAnalysis(MarketingAnalysisRequest request, String userId) { List analysisTypes = request.getAnalysisType(); if (analysisTypes == null || analysisTypes.isEmpty()) { @@ -685,121 +653,59 @@ public class MarketingAnalysisService { private Map generateJsonAnalysis(MarketingAnalysisRequest request) { Map result = new HashMap<>(); - logger.info("Starting parallel JSON analysis generation"); + logger.info("Starting sequential JSON analysis generation"); - // Создаем CompletableFuture для каждого типа анализа - CompletableFuture> audienceFuture = (CompletableFuture>) CompletableFuture - .>supplyAsync(() -> { - logger.info("Starting audience analysis generation"); - Map analysis = getAudienceAnalysis(request); - if (analysis != null && !analysis.isEmpty()) { - logger.info("Audience analysis generated successfully"); - } else { - logger.warn("Audience analysis returned empty result"); - } - return analysis != null ? analysis : new HashMap<>(); - }) - .exceptionally(ex -> { - logger.error("Error in audience analysis future: {}", ex.getMessage(), ex); - return new HashMap<>(); - }); - - CompletableFuture> competitorFuture = (CompletableFuture>) CompletableFuture - .>supplyAsync(() -> { - logger.info("Starting competitor analysis generation"); - Map analysis = getCompetitorAnalysis(request); - if (analysis != null && !analysis.isEmpty()) { - logger.info("Competitor analysis generated successfully"); - } else { - logger.warn("Competitor analysis returned empty result"); - } - return analysis != null ? analysis : new HashMap<>(); - }) - .exceptionally(ex -> { - logger.error("Error in competitor analysis future: {}", ex.getMessage(), ex); - return new HashMap<>(); - }); - - CompletableFuture> seasonalityFuture = (CompletableFuture>) CompletableFuture - .>supplyAsync(() -> { - logger.info("Starting seasonality and market analysis generation"); - Map analysis = getSeasonalityAndMarket(request); - if (analysis != null && !analysis.isEmpty()) { - logger.info("Seasonality and market analysis generated successfully"); - } else { - logger.warn("Seasonality and market analysis returned empty result"); - } - return analysis != null ? analysis : new HashMap<>(); - }) - .exceptionally(ex -> { - logger.error("Error in seasonality analysis future: {}", ex.getMessage(), ex); - return new HashMap<>(); - }); - - CompletableFuture> strategyFuture = (CompletableFuture>) CompletableFuture - .>supplyAsync(() -> { - logger.info("Starting strategy and funnel analysis generation"); - Map analysis = getStrategyAndFunnel(request); - if (analysis != null && !analysis.isEmpty()) { - logger.info("Strategy and funnel analysis generated successfully"); - } else { - logger.warn("Strategy and funnel analysis returned empty result"); - } - return analysis != null ? analysis : new HashMap<>(); - }) - .exceptionally(ex -> { - logger.error("Error in strategy analysis future: {}", ex.getMessage(), ex); - return new HashMap<>(); - }); - - // Ожидаем завершения всех futures + // Последовательное выполнение анализов try { - CompletableFuture.allOf(audienceFuture, competitorFuture, seasonalityFuture, strategyFuture) - .get(5, TimeUnit.MINUTES); // Таймаут 5 минут - - // Собираем результаты - Map audienceAnalysis = audienceFuture.join(); - Map competitorAnalysis = competitorFuture.join(); - Map seasonalityAndMarket = seasonalityFuture.join(); - Map strategyAndFunnel = strategyFuture.join(); - + logger.info("Starting audience analysis generation"); + Map audienceAnalysis = getAudienceAnalysis(request); if (audienceAnalysis != null && !audienceAnalysis.isEmpty()) { + logger.info("Audience analysis generated successfully"); result.putAll(audienceAnalysis); + } else { + logger.warn("Audience analysis returned empty result"); } - if (competitorAnalysis != null && !competitorAnalysis.isEmpty()) { - result.putAll(competitorAnalysis); - } - if (seasonalityAndMarket != null && !seasonalityAndMarket.isEmpty()) { - result.putAll(seasonalityAndMarket); - } - if (strategyAndFunnel != null && !strategyAndFunnel.isEmpty()) { - result.putAll(strategyAndFunnel); - } - } catch (Exception e) { - logger.error("Error waiting for analysis futures: {}", e.getMessage(), e); - // Пытаемся получить результаты даже при ошибке - try { - Map audienceAnalysis = audienceFuture.getNow(new HashMap<>()); - Map competitorAnalysis = competitorFuture.getNow(new HashMap<>()); - Map seasonalityAndMarket = seasonalityFuture.getNow(new HashMap<>()); - Map strategyAndFunnel = strategyFuture.getNow(new HashMap<>()); + logger.error("Error in audience analysis: {}", e.getMessage(), e); + } - if (audienceAnalysis != null && !audienceAnalysis.isEmpty()) { - result.putAll(audienceAnalysis); - } - if (competitorAnalysis != null && !competitorAnalysis.isEmpty()) { - result.putAll(competitorAnalysis); - } - if (seasonalityAndMarket != null && !seasonalityAndMarket.isEmpty()) { - result.putAll(seasonalityAndMarket); - } - if (strategyAndFunnel != null && !strategyAndFunnel.isEmpty()) { - result.putAll(strategyAndFunnel); - } - } catch (Exception ex) { - logger.error("Error getting partial results: {}", ex.getMessage(), ex); + try { + logger.info("Starting competitor analysis generation"); + Map competitorAnalysis = getCompetitorAnalysis(request); + if (competitorAnalysis != null && !competitorAnalysis.isEmpty()) { + logger.info("Competitor analysis generated successfully"); + result.putAll(competitorAnalysis); + } else { + logger.warn("Competitor analysis returned empty result"); } + } catch (Exception e) { + logger.error("Error in competitor analysis: {}", e.getMessage(), e); + } + + try { + logger.info("Starting seasonality and market analysis generation"); + Map seasonalityAndMarket = getSeasonalityAndMarket(request); + if (seasonalityAndMarket != null && !seasonalityAndMarket.isEmpty()) { + logger.info("Seasonality and market analysis generated successfully"); + result.putAll(seasonalityAndMarket); + } else { + logger.warn("Seasonality and market analysis returned empty result"); + } + } catch (Exception e) { + logger.error("Error in seasonality analysis: {}", e.getMessage(), e); + } + + try { + logger.info("Starting strategy and funnel analysis generation"); + Map strategyAndFunnel = getStrategyAndFunnel(request); + if (strategyAndFunnel != null && !strategyAndFunnel.isEmpty()) { + logger.info("Strategy and funnel analysis generated successfully"); + result.putAll(strategyAndFunnel); + } else { + logger.warn("Strategy and funnel analysis returned empty result"); + } + } catch (Exception e) { + logger.error("Error in strategy analysis: {}", e.getMessage(), e); } // Проверяем, что хотя бы одна секция была успешно сгенерирована @@ -1562,7 +1468,6 @@ public class MarketingAnalysisService { String sectionId) { try { - openAiSemaphore.acquire(); int maxAttempts = Math.max(1, attempts); String last = null; DetailLevel detailLevel = resolveDetailLevelFromContext(contextJson); @@ -1603,8 +1508,6 @@ public class MarketingAnalysisService { logger.error("generateWithRetryForSection: Error generating section: {}", e.getMessage(), e); logger.error("generateWithRetryForSection: Stack trace: ", e); return null; - } finally { - openAiSemaphore.release(); } } @@ -3777,53 +3680,14 @@ public class MarketingAnalysisService { report.put("market", analysisJson.get("market")); } - // Шаг 2: Генерация профессионального текстового отчёта (11 параллельных вызовов - // AI) + // Шаг 2: Генерация профессионального текстового отчёта (11 последовательных + // вызовов AI) logger.info("generateMarketingReport: Starting modular text report generation (11 sections)"); - // Создаем CompletableFuture для каждого раздела - CompletableFuture section1 = CompletableFuture - .supplyAsync(() -> genSectionSummary(prepareSectionContext(analysisJson, "summary")), - sectionGenerationExecutor); - CompletableFuture section2 = CompletableFuture - .supplyAsync(() -> genSectionProduct(prepareSectionContext(analysisJson, "product")), - sectionGenerationExecutor); - CompletableFuture section3 = CompletableFuture - .supplyAsync(() -> genSectionMarket(prepareSectionContext(analysisJson, "market")), - sectionGenerationExecutor); - CompletableFuture section4 = CompletableFuture - .supplyAsync(() -> genSectionAudience(prepareSectionContext(analysisJson, "audience")), - sectionGenerationExecutor); - CompletableFuture section5 = CompletableFuture - .supplyAsync(() -> genSectionCompetitors(prepareSectionContext(analysisJson, "competitors")), - sectionGenerationExecutor); - CompletableFuture section6 = CompletableFuture - .supplyAsync(() -> genSectionSWOT(prepareSectionContext(analysisJson, "swot")), - sectionGenerationExecutor); - CompletableFuture section7 = CompletableFuture - .supplyAsync(() -> genSectionChannels(prepareSectionContext(analysisJson, "channels")), - sectionGenerationExecutor); - CompletableFuture section8 = CompletableFuture - .supplyAsync(() -> genSectionFunnel(prepareSectionContext(analysisJson, "funnel")), - sectionGenerationExecutor); - CompletableFuture section9 = CompletableFuture - .supplyAsync(() -> genSectionPositioning(prepareSectionContext(analysisJson, "positioning")), - sectionGenerationExecutor); - CompletableFuture section10 = CompletableFuture - .supplyAsync(() -> genSectionContent(prepareSectionContext(analysisJson, "content")), - sectionGenerationExecutor); - CompletableFuture section11 = CompletableFuture - .supplyAsync(() -> genSectionStrategy(prepareSectionContext(analysisJson, "strategy")), - sectionGenerationExecutor); - - // Ждем завершения всех разделов - CompletableFuture.allOf(section1, section2, section3, section4, section5, section6, - section7, section8, section9, section10, section11).join(); - - // Собираем результаты в правильном порядке (I → XI) + // Последовательное выполнение всех разделов StringBuilder fullReportBuilder = new StringBuilder(); try { - String s1 = section1.get(); + String s1 = genSectionSummary(prepareSectionContext(analysisJson, "summary")); if (s1 != null && !s1.trim().isEmpty() && isValidAiResponse(s1)) { // Очищаем от ошибок перед добавлением String cleaned = removeErrorMessages(s1); @@ -3843,7 +3707,7 @@ public class MarketingAnalysisService { } try { - String s2 = section2.get(); + String s2 = genSectionProduct(prepareSectionContext(analysisJson, "product")); if (s2 != null && !s2.trim().isEmpty() && isValidAiResponse(s2)) { String cleaned = removeErrorMessages(s2); if (cleaned != null && !cleaned.trim().isEmpty()) { @@ -3862,7 +3726,7 @@ public class MarketingAnalysisService { } try { - String s3 = section3.get(); + String s3 = genSectionMarket(prepareSectionContext(analysisJson, "market")); if (s3 != null && !s3.trim().isEmpty() && isValidAiResponse(s3)) { String cleaned = removeErrorMessages(s3); if (cleaned != null && !cleaned.trim().isEmpty()) { @@ -3882,7 +3746,7 @@ public class MarketingAnalysisService { } try { - String s4 = section4.get(); + String s4 = genSectionAudience(prepareSectionContext(analysisJson, "audience")); if (s4 != null && !s4.trim().isEmpty() && isValidAiResponse(s4)) { String cleaned = removeErrorMessages(s4); if (cleaned != null && !cleaned.trim().isEmpty()) { @@ -3902,7 +3766,7 @@ public class MarketingAnalysisService { } try { - String s5 = section5.get(); + String s5 = genSectionCompetitors(prepareSectionContext(analysisJson, "competitors")); if (s5 != null && !s5.trim().isEmpty() && isValidAiResponse(s5)) { String cleaned = removeErrorMessages(s5); if (cleaned != null && !cleaned.trim().isEmpty()) { @@ -3921,7 +3785,7 @@ public class MarketingAnalysisService { } try { - String s6 = section6.get(); + String s6 = genSectionSWOT(prepareSectionContext(analysisJson, "swot")); if (s6 != null && !s6.trim().isEmpty() && isValidAiResponse(s6)) { String cleaned = removeErrorMessages(s6); if (cleaned != null && !cleaned.trim().isEmpty()) { @@ -3940,7 +3804,7 @@ public class MarketingAnalysisService { } try { - String s7 = section7.get(); + String s7 = genSectionChannels(prepareSectionContext(analysisJson, "channels")); if (s7 != null && !s7.trim().isEmpty() && isValidAiResponse(s7)) { String cleaned = removeErrorMessages(s7); if (cleaned != null && !cleaned.trim().isEmpty()) { @@ -3961,7 +3825,7 @@ public class MarketingAnalysisService { } try { - String s8 = section8.get(); + String s8 = genSectionFunnel(prepareSectionContext(analysisJson, "funnel")); if (s8 != null && !s8.trim().isEmpty() && isValidAiResponse(s8)) { String cleaned = removeErrorMessages(s8); if (cleaned != null && !cleaned.trim().isEmpty()) { @@ -3980,7 +3844,7 @@ public class MarketingAnalysisService { } try { - String s9 = section9.get(); + String s9 = genSectionPositioning(prepareSectionContext(analysisJson, "positioning")); if (s9 != null && !s9.trim().isEmpty() && isValidAiResponse(s9)) { String cleaned = removeErrorMessages(s9); if (cleaned != null && !cleaned.trim().isEmpty()) { @@ -4002,7 +3866,7 @@ public class MarketingAnalysisService { } try { - String s10 = section10.get(); + String s10 = genSectionContent(prepareSectionContext(analysisJson, "content")); if (s10 != null && !s10.trim().isEmpty() && isValidAiResponse(s10)) { String cleaned = removeErrorMessages(s10); if (cleaned != null && !cleaned.trim().isEmpty()) { @@ -4021,7 +3885,7 @@ public class MarketingAnalysisService { } try { - String s11 = section11.get(); + String s11 = genSectionStrategy(prepareSectionContext(analysisJson, "strategy")); if (s11 != null && !s11.trim().isEmpty() && isValidAiResponse(s11)) { String cleaned = removeErrorMessages(s11); if (cleaned != null && !cleaned.trim().isEmpty()) { diff --git a/src/main/java/kz/konturai/parser/service/OpenAIAnalyticsService.java b/src/main/java/kz/konturai/parser/service/OpenAIAnalyticsService.java index c1000ba..ff03af2 100644 --- a/src/main/java/kz/konturai/parser/service/OpenAIAnalyticsService.java +++ b/src/main/java/kz/konturai/parser/service/OpenAIAnalyticsService.java @@ -23,8 +23,6 @@ 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 java.util.concurrent.atomic.AtomicReference; import jakarta.annotation.PostConstruct; @@ -68,14 +66,9 @@ public class OpenAIAnalyticsService { @Value("${openai.retry.rateLimitMaxDelayMs:300000}") private long rateLimitMaxDelayMs; - @Value("${openai.rateLimit.maxConcurrentRequests:2}") - private int maxConcurrentRequests; - @Value("${openai.api.key}") private String apiKey; - private Semaphore requestSemaphore; - private String openaiUrl; /** @@ -143,10 +136,7 @@ public class OpenAIAnalyticsService { .defaultHeader("Authorization", "Bearer " + apiKey) .build(); - this.requestSemaphore = new Semaphore(maxConcurrentRequests, true); - - logger.info("OpenAIAnalyticsService initialized with model: {} and max concurrent requests: {}", - modelName, maxConcurrentRequests); + logger.info("OpenAIAnalyticsService initialized with model: {}", modelName); } public MarketItem.Analytics analyzeText(String text) { @@ -228,70 +218,55 @@ public class OpenAIAnalyticsService { DEFAULT_TEMPERATURE); try { - // Acquire semaphore permit before making request - if (!requestSemaphore.tryAcquire(30, TimeUnit.SECONDS)) { - logger.warn("Timeout waiting for OpenAI request slot. Skipping request."); - return null; - } - - try { - Mono> responseMono = this.webClient.post() - .contentType(MediaType.APPLICATION_JSON) - .accept(MediaType.APPLICATION_JSON) - .bodyValue(requestBody) - .retrieve() - .bodyToMono(new ParameterizedTypeReference>() { - }) - .timeout(Duration.ofMillis(timeoutMs)) - .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."); - } - } else if (err instanceof java.util.concurrent.TimeoutException) { - logger.error("OpenAI API request timed out after {}ms", timeoutMs); + Mono> responseMono = this.webClient.post() + .contentType(MediaType.APPLICATION_JSON) + .accept(MediaType.APPLICATION_JSON) + .bodyValue(requestBody) + .retrieve() + .bodyToMono(new ParameterizedTypeReference>() { + }) + .timeout(Duration.ofMillis(timeoutMs)) + .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(); - }); - // Calculate block timeout: request timeout + (max retries * max delay) + buffer - long blockTimeoutMs = timeoutMs - + (maxRetryAttempts * maxRetryDelayMs) - + 30000; // 30 second buffer - Map response = responseMono.block(Duration.ofMillis(blockTimeoutMs)); + } else if (err instanceof java.util.concurrent.TimeoutException) { + logger.error("OpenAI API request timed out after {}ms", timeoutMs); + } + return Mono.empty(); + }); + // Calculate block timeout: request timeout + (max retries * max delay) + buffer + long blockTimeoutMs = timeoutMs + + (maxRetryAttempts * maxRetryDelayMs) + + 30000; // 30 second buffer + Map response = responseMono.block(Duration.ofMillis(blockTimeoutMs)); - if (response == null) { - return null; - } - - @SuppressWarnings("unchecked") - List> choices = (List>) response.get("choices"); - if (choices != null && !choices.isEmpty()) { - Map firstChoice = choices.get(0); - Object msg = firstChoice.get("message"); - if (msg instanceof Map) { - @SuppressWarnings("unchecked") - Map messageObj = (Map) msg; - Object content = messageObj.get("content"); - return content == null ? null : String.valueOf(content).trim(); - } - } + if (response == null) { 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()); + + @SuppressWarnings("unchecked") + List> choices = (List>) response.get("choices"); + if (choices != null && !choices.isEmpty()) { + Map firstChoice = choices.get(0); + Object msg = firstChoice.get("message"); + if (msg instanceof Map) { + @SuppressWarnings("unchecked") + Map messageObj = (Map) msg; + Object content = messageObj.get("content"); + return content == null ? null : String.valueOf(content).trim(); + } + } return null; } catch (Exception e) { // .block() throws IllegalStateException on timeout, not TimeoutException @@ -359,75 +334,60 @@ public class OpenAIAnalyticsService { DEFAULT_TEMPERATURE); try { - // Acquire semaphore permit before making request - if (!requestSemaphore.tryAcquire(30, TimeUnit.SECONDS)) { - logger.warn("Timeout waiting for OpenAI request slot. Skipping request."); - return null; - } - - try { - logger.debug("Calling OpenAI API with model: {} (timeout: {}ms)", modelName, timeoutMsInstruction); - // Use timeout operator in the reactive chain so timeouts can be retried - Mono> responseMono = this.webClient.post() - .contentType(MediaType.APPLICATION_JSON) - .accept(MediaType.APPLICATION_JSON) - .bodyValue(requestBody) - .retrieve() - .bodyToMono(new ParameterizedTypeReference>() { - }) - .timeout(Duration.ofMillis(timeoutMsInstruction)) - .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."); - } - } else if (err instanceof java.util.concurrent.TimeoutException) { - logger.error("OpenAI API request timed out after {}ms", timeoutMsInstruction); + logger.debug("Calling OpenAI API with model: {} (timeout: {}ms)", modelName, timeoutMsInstruction); + // Use timeout operator in the reactive chain so timeouts can be retried + Mono> responseMono = this.webClient.post() + .contentType(MediaType.APPLICATION_JSON) + .accept(MediaType.APPLICATION_JSON) + .bodyValue(requestBody) + .retrieve() + .bodyToMono(new ParameterizedTypeReference>() { + }) + .timeout(Duration.ofMillis(timeoutMsInstruction)) + .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(); - }); - // Calculate block timeout: request timeout + (max retries * max delay) + buffer - // This ensures we can wait through all retry attempts with rate limit delays - long blockTimeoutMs = timeoutMsInstruction - + (rateLimitMaxAttempts * rateLimitMaxDelayMs) - + 30000; // 30 second buffer - logger.debug("Block timeout calculated: {}ms (request: {}ms, retries: {} * {}ms, buffer: 30000ms)", - blockTimeoutMs, timeoutMsInstruction, rateLimitMaxAttempts, rateLimitMaxDelayMs); - Map response = responseMono.block(Duration.ofMillis(blockTimeoutMs)); + } else if (err instanceof java.util.concurrent.TimeoutException) { + logger.error("OpenAI API request timed out after {}ms", timeoutMsInstruction); + } + return Mono.empty(); + }); + // Calculate block timeout: request timeout + (max retries * max delay) + buffer + // This ensures we can wait through all retry attempts with rate limit delays + long blockTimeoutMs = timeoutMsInstruction + + (rateLimitMaxAttempts * rateLimitMaxDelayMs) + + 30000; // 30 second buffer + logger.debug("Block timeout calculated: {}ms (request: {}ms, retries: {} * {}ms, buffer: 30000ms)", + blockTimeoutMs, timeoutMsInstruction, rateLimitMaxAttempts, rateLimitMaxDelayMs); + Map response = responseMono.block(Duration.ofMillis(blockTimeoutMs)); - if (response == null) { - return null; - } - - @SuppressWarnings("unchecked") - List> choices = (List>) response.get("choices"); - if (choices != null && !choices.isEmpty()) { - Map firstChoice = choices.get(0); - Object msg = firstChoice.get("message"); - if (msg instanceof Map) { - @SuppressWarnings("unchecked") - Map messageObj = (Map) msg; - Object content = messageObj.get("content"); - return content == null ? null : String.valueOf(content).trim(); - } - } + if (response == null) { 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()); + + @SuppressWarnings("unchecked") + List> choices = (List>) response.get("choices"); + if (choices != null && !choices.isEmpty()) { + Map firstChoice = choices.get(0); + Object msg = firstChoice.get("message"); + if (msg instanceof Map) { + @SuppressWarnings("unchecked") + Map messageObj = (Map) msg; + Object content = messageObj.get("content"); + return content == null ? null : String.valueOf(content).trim(); + } + } return null; } catch (Exception e) { // .block() throws IllegalStateException on timeout, not TimeoutException