.
This commit is contained in:
root
@@ -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<MarketingAnalysis> startAnalysis(MarketingAnalysisRequest request, String userId) {
|
||||
List<String> analysisTypes = request.getAnalysisType();
|
||||
if (analysisTypes == null || analysisTypes.isEmpty()) {
|
||||
@@ -685,121 +653,59 @@ public class MarketingAnalysisService {
|
||||
private Map<String, Object> generateJsonAnalysis(MarketingAnalysisRequest request) {
|
||||
Map<String, Object> result = new HashMap<>();
|
||||
|
||||
logger.info("Starting parallel JSON analysis generation");
|
||||
logger.info("Starting sequential JSON analysis generation");
|
||||
|
||||
// Создаем CompletableFuture для каждого типа анализа
|
||||
CompletableFuture<Map<String, Object>> audienceFuture = (CompletableFuture<Map<String, Object>>) CompletableFuture
|
||||
.<Map<String, Object>>supplyAsync(() -> {
|
||||
logger.info("Starting audience analysis generation");
|
||||
Map<String, Object> 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<Map<String, Object>> competitorFuture = (CompletableFuture<Map<String, Object>>) CompletableFuture
|
||||
.<Map<String, Object>>supplyAsync(() -> {
|
||||
logger.info("Starting competitor analysis generation");
|
||||
Map<String, Object> 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<Map<String, Object>> seasonalityFuture = (CompletableFuture<Map<String, Object>>) CompletableFuture
|
||||
.<Map<String, Object>>supplyAsync(() -> {
|
||||
logger.info("Starting seasonality and market analysis generation");
|
||||
Map<String, Object> 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<Map<String, Object>> strategyFuture = (CompletableFuture<Map<String, Object>>) CompletableFuture
|
||||
.<Map<String, Object>>supplyAsync(() -> {
|
||||
logger.info("Starting strategy and funnel analysis generation");
|
||||
Map<String, Object> 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<String, Object> audienceAnalysis = audienceFuture.join();
|
||||
Map<String, Object> competitorAnalysis = competitorFuture.join();
|
||||
Map<String, Object> seasonalityAndMarket = seasonalityFuture.join();
|
||||
Map<String, Object> strategyAndFunnel = strategyFuture.join();
|
||||
|
||||
logger.info("Starting audience analysis generation");
|
||||
Map<String, Object> 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<String, Object> audienceAnalysis = audienceFuture.getNow(new HashMap<>());
|
||||
Map<String, Object> competitorAnalysis = competitorFuture.getNow(new HashMap<>());
|
||||
Map<String, Object> seasonalityAndMarket = seasonalityFuture.getNow(new HashMap<>());
|
||||
Map<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String> section1 = CompletableFuture
|
||||
.supplyAsync(() -> genSectionSummary(prepareSectionContext(analysisJson, "summary")),
|
||||
sectionGenerationExecutor);
|
||||
CompletableFuture<String> section2 = CompletableFuture
|
||||
.supplyAsync(() -> genSectionProduct(prepareSectionContext(analysisJson, "product")),
|
||||
sectionGenerationExecutor);
|
||||
CompletableFuture<String> section3 = CompletableFuture
|
||||
.supplyAsync(() -> genSectionMarket(prepareSectionContext(analysisJson, "market")),
|
||||
sectionGenerationExecutor);
|
||||
CompletableFuture<String> section4 = CompletableFuture
|
||||
.supplyAsync(() -> genSectionAudience(prepareSectionContext(analysisJson, "audience")),
|
||||
sectionGenerationExecutor);
|
||||
CompletableFuture<String> section5 = CompletableFuture
|
||||
.supplyAsync(() -> genSectionCompetitors(prepareSectionContext(analysisJson, "competitors")),
|
||||
sectionGenerationExecutor);
|
||||
CompletableFuture<String> section6 = CompletableFuture
|
||||
.supplyAsync(() -> genSectionSWOT(prepareSectionContext(analysisJson, "swot")),
|
||||
sectionGenerationExecutor);
|
||||
CompletableFuture<String> section7 = CompletableFuture
|
||||
.supplyAsync(() -> genSectionChannels(prepareSectionContext(analysisJson, "channels")),
|
||||
sectionGenerationExecutor);
|
||||
CompletableFuture<String> section8 = CompletableFuture
|
||||
.supplyAsync(() -> genSectionFunnel(prepareSectionContext(analysisJson, "funnel")),
|
||||
sectionGenerationExecutor);
|
||||
CompletableFuture<String> section9 = CompletableFuture
|
||||
.supplyAsync(() -> genSectionPositioning(prepareSectionContext(analysisJson, "positioning")),
|
||||
sectionGenerationExecutor);
|
||||
CompletableFuture<String> section10 = CompletableFuture
|
||||
.supplyAsync(() -> genSectionContent(prepareSectionContext(analysisJson, "content")),
|
||||
sectionGenerationExecutor);
|
||||
CompletableFuture<String> 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()) {
|
||||
|
||||
@@ -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<Map<String, Object>> responseMono = this.webClient.post()
|
||||
.contentType(MediaType.APPLICATION_JSON)
|
||||
.accept(MediaType.APPLICATION_JSON)
|
||||
.bodyValue(requestBody)
|
||||
.retrieve()
|
||||
.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
|
||||
})
|
||||
.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<Map<String, Object>> responseMono = this.webClient.post()
|
||||
.contentType(MediaType.APPLICATION_JSON)
|
||||
.accept(MediaType.APPLICATION_JSON)
|
||||
.bodyValue(requestBody)
|
||||
.retrieve()
|
||||
.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
|
||||
})
|
||||
.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<String, Object> 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<String, Object> response = responseMono.block(Duration.ofMillis(blockTimeoutMs));
|
||||
|
||||
if (response == null) {
|
||||
return null;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
List<Map<String, Object>> choices = (List<Map<String, Object>>) response.get("choices");
|
||||
if (choices != null && !choices.isEmpty()) {
|
||||
Map<String, Object> firstChoice = choices.get(0);
|
||||
Object msg = firstChoice.get("message");
|
||||
if (msg instanceof Map) {
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String, Object> messageObj = (Map<String, Object>) 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<Map<String, Object>> choices = (List<Map<String, Object>>) response.get("choices");
|
||||
if (choices != null && !choices.isEmpty()) {
|
||||
Map<String, Object> firstChoice = choices.get(0);
|
||||
Object msg = firstChoice.get("message");
|
||||
if (msg instanceof Map) {
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String, Object> messageObj = (Map<String, Object>) 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<Map<String, Object>> responseMono = this.webClient.post()
|
||||
.contentType(MediaType.APPLICATION_JSON)
|
||||
.accept(MediaType.APPLICATION_JSON)
|
||||
.bodyValue(requestBody)
|
||||
.retrieve()
|
||||
.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
|
||||
})
|
||||
.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<Map<String, Object>> responseMono = this.webClient.post()
|
||||
.contentType(MediaType.APPLICATION_JSON)
|
||||
.accept(MediaType.APPLICATION_JSON)
|
||||
.bodyValue(requestBody)
|
||||
.retrieve()
|
||||
.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
|
||||
})
|
||||
.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<String, Object> 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<String, Object> response = responseMono.block(Duration.ofMillis(blockTimeoutMs));
|
||||
|
||||
if (response == null) {
|
||||
return null;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
List<Map<String, Object>> choices = (List<Map<String, Object>>) response.get("choices");
|
||||
if (choices != null && !choices.isEmpty()) {
|
||||
Map<String, Object> firstChoice = choices.get(0);
|
||||
Object msg = firstChoice.get("message");
|
||||
if (msg instanceof Map) {
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String, Object> messageObj = (Map<String, Object>) 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<Map<String, Object>> choices = (List<Map<String, Object>>) response.get("choices");
|
||||
if (choices != null && !choices.isEmpty()) {
|
||||
Map<String, Object> firstChoice = choices.get(0);
|
||||
Object msg = firstChoice.get("message");
|
||||
if (msg instanceof Map) {
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String, Object> messageObj = (Map<String, Object>) 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
|
||||
|
||||
Reference in New Issue
Block a user