This commit is contained in:
root
2025-12-28 00:21:38 +05:00
parent 59779a9206
commit 4dc63d1bf8
3 changed files with 295 additions and 34 deletions
@@ -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<Object> 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<String, Object> 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<String, Object> 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");
@@ -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<Map<String, Object>> messages = buildChatMessages(prompt, systemPrompt);
int maxTokens = maxTokensOverride != null ? maxTokensOverride : DEFAULT_MAX_TOKENS_REPORT;
long timeoutMsToUse = timeoutOverrideMs != null ? timeoutOverrideMs : timeoutMsInstruction;
Map<String, Object> 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<Map<String, Object>> responseMono = this.webClient.post()
.contentType(MediaType.APPLICATION_JSON)
@@ -343,7 +372,7 @@ public class OpenAIAnalyticsService {
.retrieve()
.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
})
.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<String, Object> response = responseMono.block(Duration.ofMillis(blockTimeoutMs));
if (response == null) {
+1 -1
View File
@@ -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