From 4dc63d1bf84120f5c5c8ac3913e22ea3d1bc1003 Mon Sep 17 00:00:00 2001 From: root Date: Sun, 28 Dec 2025 00:21:38 +0500 Subject: [PATCH] . --- .../service/MarketingAnalysisService.java | 288 ++++++++++++++++-- .../service/OpenAIAnalyticsService.java | 39 ++- src/main/resources/application.properties | 2 +- 3 files changed, 295 insertions(+), 34 deletions(-) diff --git a/src/main/java/kz/konturai/parser/service/MarketingAnalysisService.java b/src/main/java/kz/konturai/parser/service/MarketingAnalysisService.java index cfce573..b14db3c 100644 --- a/src/main/java/kz/konturai/parser/service/MarketingAnalysisService.java +++ b/src/main/java/kz/konturai/parser/service/MarketingAnalysisService.java @@ -22,6 +22,9 @@ import com.fasterxml.jackson.databind.ObjectMapper; import java.time.LocalDateTime; import java.util.*; +import java.util.concurrent.Semaphore; +import jakarta.annotation.PostConstruct; +import org.springframework.web.reactive.function.client.WebClientResponseException; @Service public class MarketingAnalysisService { @@ -363,6 +366,11 @@ public class MarketingAnalysisService { @Value("${openai.model.name:gpt-4o-mini}") private String miniModelName; + @Value("${openai.rateLimit.maxConcurrentRequests:3}") + private int maxConcurrentRequests; + + private Semaphore requestSemaphore; + public MarketingAnalysisService( MarketingAnalysisRepository repository, OpenAIAnalyticsService openAIAnalyticsService, @@ -376,6 +384,13 @@ public class MarketingAnalysisService { this.serperSearchService = serperSearchService; } + @PostConstruct + public void initializeSemaphore() { + this.requestSemaphore = new Semaphore(maxConcurrentRequests, true); + logger.info("MarketingAnalysisService initialized with max concurrent OpenAI requests: {}", + maxConcurrentRequests); + } + private boolean shouldIncludeResearchPackInSection(String section) { if (section == null) { return false; @@ -408,21 +423,96 @@ public class MarketingAnalysisService { String preferredSites = "(site:stat.gov.kz OR site:kapital.kz OR site:kursiv.media OR site:forbes.kz OR site:pwc.kz OR site:kpmg.kz)"; List general = new ArrayList<>(); - general.add(serperSearchService - .search(topic + " рынок " + regionStr + " 2024 2025 статистика объем рост " + preferredSites)); - general.add(serperSearchService - .search("инфляция Казахстан 2024 2025 курс тенге 2024 2025 логистика импорт " + preferredSites)); + String query1 = topic + " рынок " + regionStr + " 2024 2025 статистика объем рост " + preferredSites; + try { + general.add(serperSearchService.search(query1)); + } catch (Exception e) { + // Handle DNS/network errors and other exceptions from Serper API + String errorType = "Unexpected error"; + if (e instanceof java.net.UnknownHostException || e instanceof java.net.ConnectException) { + errorType = "DNS/Network failure"; + } else if (e instanceof java.util.concurrent.TimeoutException) { + errorType = "Timeout"; + } else if (e instanceof java.io.IOException) { + errorType = "IO error"; + } + logger.warn("Serper API {} for query '{}': {}", errorType, query1, e.getMessage()); + general.add(new SerperSearchResult(query1, "ERROR", errorType + ": " + e.getMessage(), List.of())); + } + + String query2 = "инфляция Казахстан 2024 2025 курс тенге 2024 2025 логистика импорт " + preferredSites; + try { + general.add(serperSearchService.search(query2)); + } catch (Exception e) { + // Handle DNS/network errors and other exceptions from Serper API + String errorType = "Unexpected error"; + if (e instanceof java.net.UnknownHostException || e instanceof java.net.ConnectException) { + errorType = "DNS/Network failure"; + } else if (e instanceof java.util.concurrent.TimeoutException) { + errorType = "Timeout"; + } else if (e instanceof java.io.IOException) { + errorType = "IO error"; + } + logger.warn("Serper API {} for query '{}': {}", errorType, query2, e.getMessage()); + general.add(new SerperSearchResult(query2, "ERROR", errorType + ": " + e.getMessage(), List.of())); + } // Lightweight section followups (stored for auditing/citations). Prompts for // III/V/XI may further use these. Map followups = new LinkedHashMap<>(); - followups.put("market", - serperSearchService.search(topic + " объем рынка Казахстан 2024 2025 отчет " + preferredSites)); - followups.put("competitors", - serperSearchService.search(topic + " конкуренты " + regionStr + " топ компании магазины")); - followups.put("finance", - serperSearchService.search( - topic + " себестоимость логистика импорт курс тенге влияние 2024 2025 " + preferredSites)); + String marketQuery = topic + " объем рынка Казахстан 2024 2025 отчет " + preferredSites; + try { + followups.put("market", serperSearchService.search(marketQuery)); + } catch (Exception e) { + // Handle DNS/network errors and other exceptions from Serper API + String errorType = "Unexpected error"; + if (e instanceof java.net.UnknownHostException || e instanceof java.net.ConnectException) { + errorType = "DNS/Network failure"; + } else if (e instanceof java.util.concurrent.TimeoutException) { + errorType = "Timeout"; + } else if (e instanceof java.io.IOException) { + errorType = "IO error"; + } + logger.warn("Serper API {} for market query '{}': {}", errorType, marketQuery, e.getMessage()); + followups.put("market", + new SerperSearchResult(marketQuery, "ERROR", errorType + ": " + e.getMessage(), List.of())); + } + + String competitorsQuery = topic + " конкуренты " + regionStr + " топ компании магазины"; + try { + followups.put("competitors", serperSearchService.search(competitorsQuery)); + } catch (Exception e) { + // Handle DNS/network errors and other exceptions from Serper API + String errorType = "Unexpected error"; + if (e instanceof java.net.UnknownHostException || e instanceof java.net.ConnectException) { + errorType = "DNS/Network failure"; + } else if (e instanceof java.util.concurrent.TimeoutException) { + errorType = "Timeout"; + } else if (e instanceof java.io.IOException) { + errorType = "IO error"; + } + logger.warn("Serper API {} for competitors query '{}': {}", errorType, competitorsQuery, e.getMessage()); + followups.put("competitors", + new SerperSearchResult(competitorsQuery, "ERROR", errorType + ": " + e.getMessage(), List.of())); + } + + String financeQuery = topic + " себестоимость логистика импорт курс тенге влияние 2024 2025 " + preferredSites; + try { + followups.put("finance", serperSearchService.search(financeQuery)); + } catch (Exception e) { + // Handle DNS/network errors and other exceptions from Serper API + String errorType = "Unexpected error"; + if (e instanceof java.net.UnknownHostException || e instanceof java.net.ConnectException) { + errorType = "DNS/Network failure"; + } else if (e instanceof java.util.concurrent.TimeoutException) { + errorType = "Timeout"; + } else if (e instanceof java.io.IOException) { + errorType = "IO error"; + } + logger.warn("Serper API {} for finance query '{}': {}", errorType, financeQuery, e.getMessage()); + followups.put("finance", + new SerperSearchResult(financeQuery, "ERROR", errorType + ": " + e.getMessage(), List.of())); + } pack.put("general", general); pack.put("followups", followups); @@ -907,10 +997,47 @@ public class MarketingAnalysisService { : "Казахстан"; String topic = (niche != null && !niche.isBlank()) ? niche : (product != null ? product : "ниша"); String preferredSites = "(site:kapital.kz OR site:kursiv.media OR site:forbes.kz)"; - SerperSearchResult competitorsEvidence = serperSearchService.search( - topic + " конкуренты " + regionStrForSearch + " Алматы Казахстан топ компании магазины"); - SerperSearchResult competitorsEvidencePreferred = serperSearchService.search( - topic + " рынок " + regionStrForSearch + " лидеры ритейла компании " + preferredSites); + String competitorsQuery1 = topic + " конкуренты " + regionStrForSearch + + " Алматы Казахстан топ компании магазины"; + SerperSearchResult competitorsEvidence; + try { + competitorsEvidence = serperSearchService.search(competitorsQuery1); + } catch (Exception e) { + // Handle DNS/network errors and other exceptions from Serper API + String errorType = "Unexpected error"; + if (e instanceof java.net.UnknownHostException || e instanceof java.net.ConnectException) { + errorType = "DNS/Network failure"; + } else if (e instanceof java.util.concurrent.TimeoutException) { + errorType = "Timeout"; + } else if (e instanceof java.io.IOException) { + errorType = "IO error"; + } + logger.warn("Serper API {} for competitors query '{}': {}", errorType, competitorsQuery1, + e.getMessage()); + competitorsEvidence = new SerperSearchResult(competitorsQuery1, "ERROR", + errorType + ": " + e.getMessage(), List.of()); + } + + String competitorsQuery2 = topic + " рынок " + regionStrForSearch + " лидеры ритейла компании " + + preferredSites; + SerperSearchResult competitorsEvidencePreferred; + try { + competitorsEvidencePreferred = serperSearchService.search(competitorsQuery2); + } catch (Exception e) { + // Handle DNS/network errors and other exceptions from Serper API + String errorType = "Unexpected error"; + if (e instanceof java.net.UnknownHostException || e instanceof java.net.ConnectException) { + errorType = "DNS/Network failure"; + } else if (e instanceof java.util.concurrent.TimeoutException) { + errorType = "Timeout"; + } else if (e instanceof java.io.IOException) { + errorType = "IO error"; + } + logger.warn("Serper API {} for competitors preferred query '{}': {}", errorType, competitorsQuery2, + e.getMessage()); + competitorsEvidencePreferred = new SerperSearchResult(competitorsQuery2, "ERROR", + errorType + ": " + e.getMessage(), List.of()); + } StringBuilder promptBuilder = new StringBuilder(); promptBuilder.append( @@ -1068,10 +1195,43 @@ public class MarketingAnalysisService { : "Казахстан"; String topic = (niche != null && !niche.isBlank()) ? niche : (product != null ? product : "ниша"); String preferredSites = "(site:stat.gov.kz OR site:kapital.kz OR site:kursiv.media OR site:forbes.kz OR site:pwc.kz OR site:kpmg.kz)"; - SerperSearchResult marketEvidence = serperSearchService.search( - topic + " рынок " + regionStrForSearch + " 2024 2025 статистика объем рост " + preferredSites); - SerperSearchResult seasonalityEvidence = serperSearchService.search( - topic + " сезонность спроса " + regionStrForSearch + " 2024 2025 " + preferredSites); + String marketQuery = topic + " рынок " + regionStrForSearch + " 2024 2025 статистика объем рост " + + preferredSites; + SerperSearchResult marketEvidence; + try { + marketEvidence = serperSearchService.search(marketQuery); + } catch (Exception e) { + // Handle DNS/network errors and other exceptions from Serper API + String errorType = "Unexpected error"; + if (e instanceof java.net.UnknownHostException || e instanceof java.net.ConnectException) { + errorType = "DNS/Network failure"; + } else if (e instanceof java.util.concurrent.TimeoutException) { + errorType = "Timeout"; + } else if (e instanceof java.io.IOException) { + errorType = "IO error"; + } + logger.warn("Serper API {} for market query '{}': {}", errorType, marketQuery, e.getMessage()); + marketEvidence = new SerperSearchResult(marketQuery, "ERROR", errorType + ": " + e.getMessage(), List.of()); + } + + String seasonalityQuery = topic + " сезонность спроса " + regionStrForSearch + " 2024 2025 " + preferredSites; + SerperSearchResult seasonalityEvidence; + try { + seasonalityEvidence = serperSearchService.search(seasonalityQuery); + } catch (Exception e) { + // Handle DNS/network errors and other exceptions from Serper API + String errorType = "Unexpected error"; + if (e instanceof java.net.UnknownHostException || e instanceof java.net.ConnectException) { + errorType = "DNS/Network failure"; + } else if (e instanceof java.util.concurrent.TimeoutException) { + errorType = "Timeout"; + } else if (e instanceof java.io.IOException) { + errorType = "IO error"; + } + logger.warn("Serper API {} for seasonality query '{}': {}", errorType, seasonalityQuery, e.getMessage()); + seasonalityEvidence = new SerperSearchResult(seasonalityQuery, "ERROR", errorType + ": " + e.getMessage(), + List.of()); + } StringBuilder promptBuilder = new StringBuilder(); promptBuilder.append( @@ -1466,6 +1626,20 @@ public class MarketingAnalysisService { String fallbackModel, int attempts, String sectionId) { + return generateWithRetryForSection(contextJson, jsonContext, prompt, lang, primaryModel, fallbackModel, + attempts, sectionId, null); + } + + private String generateWithRetryForSection( + Map contextJson, + String jsonContext, + String prompt, + String lang, + String primaryModel, + String fallbackModel, + int attempts, + String sectionId, + Long timeoutOverrideMs) { try { int maxAttempts = Math.max(1, attempts); @@ -1480,11 +1654,55 @@ public class MarketingAnalysisService { if (i == maxAttempts && fallbackModel != null && !fallbackModel.isBlank()) { modelToUse = fallbackModel; } + + boolean is429Error = false; try { - last = openAIAnalyticsService.generateWithInstructionWithModel( - jsonContext, prompt, lang, modelToUse, systemPrompt, maxTokensOverride); + // Acquire semaphore permit before making OpenAI API call + if (requestSemaphore != null) { + try { + requestSemaphore.acquire(); + logger.debug("Acquired semaphore permit for section '{}' (attempt {}/{})", sectionId, i, + maxAttempts); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + logger.warn("Interrupted while waiting for semaphore permit"); + break; + } + } + + try { + last = openAIAnalyticsService.generateWithInstructionWithModel( + jsonContext, prompt, lang, modelToUse, systemPrompt, maxTokensOverride, + timeoutOverrideMs); + } catch (Exception e) { + // Check if it's a 429 rate limit error + if (e instanceof WebClientResponseException) { + WebClientResponseException wcre = (WebClientResponseException) e; + if (wcre.getStatusCode().value() == 429) { + is429Error = true; + logger.warn( + "generateWithRetryForSection: attempt {} failed with 429 rate limit for section '{}'", + i, sectionId); + } + } + logger.warn("generateWithRetry: attempt {} failed with exception: {} for section '{}'", i, + e.getMessage(), sectionId); + last = null; + } finally { + // Always release semaphore permit + if (requestSemaphore != null) { + requestSemaphore.release(); + logger.debug("Released semaphore permit for section '{}' (attempt {}/{})", sectionId, i, + maxAttempts); + } + } } catch (Exception e) { - logger.warn("generateWithRetry: attempt {} failed with exception: {}", i, e.getMessage()); + // Ensure semaphore is released even if there's an unexpected error + if (requestSemaphore != null) { + requestSemaphore.release(); + } + logger.warn("generateWithRetry: attempt {} failed with exception: {} for section '{}'", i, + e.getMessage(), sectionId); last = null; } @@ -1492,10 +1710,20 @@ public class MarketingAnalysisService { return last; } - // Небольшая пауза между попытками, чтобы сгладить кратковременные сбои + // Exponential backoff for 429 errors or empty responses if (i < maxAttempts) { + long delayMs; + if (is429Error || (last == null && i < maxAttempts)) { + // Exponential backoff: 2s, 4s, 8s + delayMs = (long) Math.pow(2, i) * 1000; + logger.info("Waiting {}ms before retry (attempt {}/{}) for section '{}' due to {}", + delayMs, i, maxAttempts, sectionId, is429Error ? "429 rate limit" : "empty response"); + } else { + // Small delay for other errors + delayMs = 250L; + } try { - Thread.sleep(250L); + Thread.sleep(delayMs); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); break; @@ -1505,7 +1733,8 @@ public class MarketingAnalysisService { return last; } catch (Exception e) { - logger.error("generateWithRetryForSection: Error generating section: {}", e.getMessage(), e); + logger.error("generateWithRetryForSection: Error generating section '{}': {}", sectionId, e.getMessage(), + e); logger.error("generateWithRetryForSection: Stack trace: ", e); return null; } @@ -2303,8 +2532,9 @@ public class MarketingAnalysisService { promptBuilder.append(jsonString); // Для сложных разделов используем ТОЛЬКО textModelName (без fallback на mini) + // Увеличенный таймаут для тяжелой секции (800-1100 слов): 180 секунд String result = generateWithRetryForSection(contextJson, jsonString, promptBuilder.toString(), "ru", - textModelName, null, 2, "audience"); + textModelName, null, 2, "audience", 180000L); if (!isValidAiResponse(result)) { logger.warn("genSectionAudience: AI returned invalid or refusal response"); @@ -2417,8 +2647,9 @@ public class MarketingAnalysisService { promptBuilder.append(jsonString); // Для сложных разделов используем ТОЛЬКО textModelName (без fallback на mini) + // Увеличенный таймаут для тяжелой секции (800-1100 слов): 240 секунд String result = generateWithRetryForSection(contextJson, jsonString, promptBuilder.toString(), "ru", - textModelName, null, 2, "competitors"); + textModelName, null, 2, "competitors", 240000L); if (!isValidAiResponse(result)) { logger.warn("genSectionCompetitors: AI returned invalid or refusal response"); @@ -2886,8 +3117,9 @@ public class MarketingAnalysisService { promptBuilder.append("JSON ДАННЫЕ:\n"); promptBuilder.append(jsonString); + // Увеличенный таймаут для тяжелой секции (800-1100 слов): 240 секунд String result = generateWithRetryForSection(contextJson, jsonString, promptBuilder.toString(), "ru", - textModelName, null, 2, "strategy"); + textModelName, null, 2, "strategy", 240000L); if (!isValidAiResponse(result)) { logger.warn("genSectionStrategy: AI returned invalid or refusal response"); diff --git a/src/main/java/kz/konturai/parser/service/OpenAIAnalyticsService.java b/src/main/java/kz/konturai/parser/service/OpenAIAnalyticsService.java index ff03af2..514d0ce 100644 --- a/src/main/java/kz/konturai/parser/service/OpenAIAnalyticsService.java +++ b/src/main/java/kz/konturai/parser/service/OpenAIAnalyticsService.java @@ -322,11 +322,40 @@ public class OpenAIAnalyticsService { String modelName, String systemPrompt, Integer maxTokensOverride) { + return generateWithInstructionWithModel(text, instruction, language, modelName, systemPrompt, maxTokensOverride, null); + } + + /** + * Генерирует текст с поддержкой system prompt, override для max_tokens и кастомного таймаута. + * + * @param text Текст (контекст), который будет добавлен к + * instruction + * @param instruction Основная инструкция для модели + * @param language Язык ответа (исторически передается, но не влияет на + * протокол) + * @param modelName Название модели OpenAI + * @param systemPrompt System prompt (если null/blank — не добавляется) + * @param maxTokensOverride Override для max_tokens (если null — используется + * дефолт) + * @param timeoutOverrideMs Override для таймаута в миллисекундах (если null — используется + * timeoutMsInstruction) + * @return Сгенерированный текст + */ + public String generateWithInstructionWithModel( + String text, + String instruction, + String language, + String modelName, + String systemPrompt, + Integer maxTokensOverride, + Long timeoutOverrideMs) { String prompt = instruction + "\n\n" + text; List> messages = buildChatMessages(prompt, systemPrompt); int maxTokens = maxTokensOverride != null ? maxTokensOverride : DEFAULT_MAX_TOKENS_REPORT; + long timeoutMsToUse = timeoutOverrideMs != null ? timeoutOverrideMs : timeoutMsInstruction; + Map requestBody = buildChatCompletionsRequestBody( modelName, messages, @@ -334,7 +363,7 @@ public class OpenAIAnalyticsService { DEFAULT_TEMPERATURE); try { - logger.debug("Calling OpenAI API with model: {} (timeout: {}ms)", modelName, timeoutMsInstruction); + logger.debug("Calling OpenAI API with model: {} (timeout: {}ms)", modelName, timeoutMsToUse); // Use timeout operator in the reactive chain so timeouts can be retried Mono> responseMono = this.webClient.post() .contentType(MediaType.APPLICATION_JSON) @@ -343,7 +372,7 @@ public class OpenAIAnalyticsService { .retrieve() .bodyToMono(new ParameterizedTypeReference>() { }) - .timeout(Duration.ofMillis(timeoutMsInstruction)) + .timeout(Duration.ofMillis(timeoutMsToUse)) .retryWhen(createRetrySpec("generateWithInstruction")) .onErrorResume(err -> { logger.error("OpenAI request failed after retries: {} - {}", err.getMessage(), @@ -359,17 +388,17 @@ public class OpenAIAnalyticsService { 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.error("OpenAI API request timed out after {}ms", timeoutMsToUse); } 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 + long blockTimeoutMs = timeoutMsToUse + (rateLimitMaxAttempts * rateLimitMaxDelayMs) + 30000; // 30 second buffer logger.debug("Block timeout calculated: {}ms (request: {}ms, retries: {} * {}ms, buffer: 30000ms)", - blockTimeoutMs, timeoutMsInstruction, rateLimitMaxAttempts, rateLimitMaxDelayMs); + blockTimeoutMs, timeoutMsToUse, rateLimitMaxAttempts, rateLimitMaxDelayMs); Map response = responseMono.block(Duration.ofMillis(blockTimeoutMs)); if (response == null) { diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 4c8070b..dc08508 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -72,7 +72,7 @@ openai.retry.multiplier=2.0 openai.retry.rateLimitMaxAttempts=10 openai.retry.rateLimitInitialDelayMs=5000 openai.retry.rateLimitMaxDelayMs=300000 -openai.rateLimit.maxConcurrentRequests=2 +openai.rateLimit.maxConcurrentRequests=3 # Serper.dev Web Search Configuration serper.api.key=837c09c1f2836b888e461a34074b5e6436be06e4