digiops/marketing-parser
2
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

.

Browse Source
This commit is contained in:
root
2025-12-27 23:42:13 +05:00
parent e8b99c1fe5
commit dd1bed2536
2 changed files with 54 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/kz/konturai/parser/service/OpenAIAnalyticsService.java
+53 -20
View File
@@ -44,6 +44,9 @@ public class OpenAIAnalyticsService {
@Value("${openai.timeoutMs:90000}") @Value("${openai.timeoutMs:90000}")
private long timeoutMs; private long timeoutMs;
@Value("${openai.timeoutMs.instruction:180000}")
private long timeoutMsInstruction;
@Value("${openai.retry.maxAttempts:5}") @Value("${openai.retry.maxAttempts:5}")
private int maxRetryAttempts; private int maxRetryAttempts;
@@ -76,7 +79,8 @@ public class OpenAIAnalyticsService {
private String openaiUrl; private String openaiUrl;
/** /**
* Helper for building OpenAI chat-completions request body (extracted for unit testing). * Helper for building OpenAI chat-completions request body (extracted for unit
* testing).
*/ */
Map<String, Object> buildChatCompletionsRequestBody( Map<String, Object> buildChatCompletionsRequestBody(
String model, String model,
@@ -92,7 +96,8 @@ public class OpenAIAnalyticsService {
} }
/** /**
* Helper for building chat messages list (system + user) (extracted for unit testing). * Helper for building chat messages list (system + user) (extracted for unit
* testing).
*/ */
List<Map<String, Object>> buildChatMessages(String userPrompt, String systemPrompt) { List<Map<String, Object>> buildChatMessages(String userPrompt, String systemPrompt) {
List<Map<String, Object>> messages = new ArrayList<>(); List<Map<String, Object>> messages = new ArrayList<>();
@@ -237,6 +242,7 @@ public class OpenAIAnalyticsService {
.retrieve() .retrieve()
.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() { .bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
}) })
.timeout(Duration.ofMillis(timeoutMs))
.retryWhen(createRetrySpec("generate")) .retryWhen(createRetrySpec("generate"))
.onErrorResume(err -> { .onErrorResume(err -> {
logger.error("OpenAI request failed after retries: {} - {}", err.getMessage(), logger.error("OpenAI request failed after retries: {} - {}", err.getMessage(),
@@ -251,10 +257,12 @@ public class OpenAIAnalyticsService {
} else if (wcre.getStatusCode().value() >= 500) { } else if (wcre.getStatusCode().value() >= 500) {
logger.error("OpenAI API server error after all retry attempts."); 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);
} }
return Mono.empty(); return Mono.empty();
}) })
.block(Duration.ofMillis(timeoutMs)); .block(Duration.ofMillis(timeoutMs + 10000)); // Add buffer for block timeout
if (response == null) { if (response == null) {
return null; return null;
@@ -282,7 +290,12 @@ public class OpenAIAnalyticsService {
logger.warn("Interrupted while waiting for OpenAI request slot: {}", e.getMessage()); logger.warn("Interrupted while waiting for OpenAI request slot: {}", e.getMessage());
return null; return null;
} catch (Exception e) { } catch (Exception e) {
logger.warn("Error calling OpenAI generate: {}", e.getMessage()); // .block() throws IllegalStateException on timeout, not TimeoutException
if (e.getMessage() != null && e.getMessage().contains("Timeout")) {
logger.warn("Timeout waiting for OpenAI response (block timeout): {}", e.getMessage());
} else {
logger.warn("Error calling OpenAI generate: {} - {}", e.getMessage(), e.getClass().getSimpleName());
}
return null; return null;
} }
} }
@@ -312,12 +325,15 @@ public class OpenAIAnalyticsService {
/** /**
* Генерирует текст с поддержкой system prompt и override для max_tokens. * Генерирует текст с поддержкой system prompt и override для max_tokens.
* *
* @param text Текст (контекст), который будет добавлен к instruction * @param text Текст (контекст), который будет добавлен к
* instruction
* @param instruction Основная инструкция для модели * @param instruction Основная инструкция для модели
* @param language Язык ответа (исторически передается, но не влияет на протокол) * @param language Язык ответа (исторически передается, но не влияет на
* протокол)
* @param modelName Название модели OpenAI * @param modelName Название модели OpenAI
* @param systemPrompt System prompt (если null/blank — не добавляется) * @param systemPrompt System prompt (если null/blank — не добавляется)
* @param maxTokensOverride Override для max_tokens (если null — используется дефолт) * @param maxTokensOverride Override для max_tokens (если null — используется
* дефолт)
* @return Сгенерированный текст * @return Сгенерированный текст
*/ */
public String generateWithInstructionWithModel( public String generateWithInstructionWithModel(
@@ -346,7 +362,8 @@ public class OpenAIAnalyticsService {
} }
try { try {
logger.debug("Calling OpenAI API with model: {}", modelName); logger.debug("Calling OpenAI API with model: {} (timeout: {}ms)", modelName, timeoutMsInstruction);
// Use timeout operator in the reactive chain so timeouts can be retried
Map<String, Object> response = this.webClient.post() Map<String, Object> response = this.webClient.post()
.contentType(MediaType.APPLICATION_JSON) .contentType(MediaType.APPLICATION_JSON)
.accept(MediaType.APPLICATION_JSON) .accept(MediaType.APPLICATION_JSON)
@@ -354,6 +371,7 @@ public class OpenAIAnalyticsService {
.retrieve() .retrieve()
.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() { .bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
}) })
.timeout(Duration.ofMillis(timeoutMsInstruction))
.retryWhen(createRetrySpec("generateWithInstruction")) .retryWhen(createRetrySpec("generateWithInstruction"))
.onErrorResume(err -> { .onErrorResume(err -> {
logger.error("OpenAI request failed after retries: {} - {}", err.getMessage(), logger.error("OpenAI request failed after retries: {} - {}", err.getMessage(),
@@ -368,10 +386,12 @@ public class OpenAIAnalyticsService {
} else if (wcre.getStatusCode().value() >= 500) { } else if (wcre.getStatusCode().value() >= 500) {
logger.error("OpenAI API server error after all retry attempts."); 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);
} }
return Mono.empty(); return Mono.empty();
}) })
.block(Duration.ofMillis(timeoutMs)); .block(Duration.ofMillis(timeoutMsInstruction + 10000)); // Add buffer for block timeout
if (response == null) { if (response == null) {
return null; return null;
@@ -399,7 +419,13 @@ public class OpenAIAnalyticsService {
logger.warn("Interrupted while waiting for OpenAI request slot: {}", e.getMessage()); logger.warn("Interrupted while waiting for OpenAI request slot: {}", e.getMessage());
return null; return null;
} catch (Exception e) { } catch (Exception e) {
logger.warn("Error calling OpenAI generate (generic): {}", e.getMessage()); // .block() throws IllegalStateException on timeout, not TimeoutException
if (e.getMessage() != null && e.getMessage().contains("Timeout")) {
logger.warn("Timeout waiting for OpenAI response (block timeout): {}", e.getMessage());
} else {
logger.warn("Error calling OpenAI generate (generic): {} - {}", e.getMessage(),
e.getClass().getSimpleName());
}
return null; return null;
} }
} }
@@ -407,11 +433,11 @@ public class OpenAIAnalyticsService {
private RetryBackoffSpec createRetrySpec(String operation) { private RetryBackoffSpec createRetrySpec(String operation) {
// Store Retry-After duration for 429 errors // Store Retry-After duration for 429 errors
AtomicReference<Duration> retryAfterDuration = new AtomicReference<>(null); AtomicReference<Duration> retryAfterDuration = new AtomicReference<>(null);
// Use the higher max attempts to handle rate limit errors better // Use the higher max attempts to handle rate limit errors better
// This will apply to all retries, but rate limit errors need more attempts // This will apply to all retries, but rate limit errors need more attempts
int effectiveMaxAttempts = Math.max(maxRetryAttempts, rateLimitMaxAttempts); int effectiveMaxAttempts = Math.max(maxRetryAttempts, rateLimitMaxAttempts);
return Retry.backoff(effectiveMaxAttempts, Duration.ofMillis(initialRetryDelayMs)) return Retry.backoff(effectiveMaxAttempts, Duration.ofMillis(initialRetryDelayMs))
.maxBackoff(Duration.ofMillis(maxRetryDelayMs)) .maxBackoff(Duration.ofMillis(maxRetryDelayMs))
.multiplier(retryMultiplier) .multiplier(retryMultiplier)
@@ -419,7 +445,7 @@ public class OpenAIAnalyticsService {
if (throwable instanceof WebClientResponseException) { if (throwable instanceof WebClientResponseException) {
WebClientResponseException wcre = (WebClientResponseException) throwable; WebClientResponseException wcre = (WebClientResponseException) throwable;
int statusCode = wcre.getStatusCode().value(); int statusCode = wcre.getStatusCode().value();
if (statusCode == 429) { if (statusCode == 429) {
// Extract Retry-After header for 429 errors // Extract Retry-After header for 429 errors
String retryAfterHeader = wcre.getHeaders().getFirst("Retry-After"); String retryAfterHeader = wcre.getHeaders().getFirst("Retry-After");
@@ -442,22 +468,28 @@ public class OpenAIAnalyticsService {
} else { } else {
retryAfterDuration.set(null); retryAfterDuration.set(null);
} }
// Only retry on 429 (rate limit) and 5xx (server errors) // Only retry on 429 (rate limit) and 5xx (server errors)
return statusCode == 429 || statusCode >= 500; return statusCode == 429 || statusCode >= 500;
} }
// Retry on network errors (timeouts, connection issues) // Retry on network errors (timeouts, connection issues)
retryAfterDuration.set(null); retryAfterDuration.set(null);
return throwable instanceof java.util.concurrent.TimeoutException boolean isTimeout = throwable instanceof java.util.concurrent.TimeoutException
|| (throwable.getMessage() != null && throwable.getMessage().contains("Timeout"));
if (isTimeout) {
logger.warn("OpenAI API timeout detected for operation '{}'. Will retry.", operation);
}
return isTimeout
|| throwable instanceof java.net.ConnectException || throwable instanceof java.net.ConnectException
|| throwable instanceof java.io.IOException; || throwable instanceof java.io.IOException;
}) })
.doBeforeRetry(retrySignal -> { .doBeforeRetry(retrySignal -> {
long attempt = retrySignal.totalRetries() + 1; long attempt = retrySignal.totalRetries() + 1;
Throwable failure = retrySignal.failure(); Throwable failure = retrySignal.failure();
// If we have a Retry-After duration for 429 errors, sleep for that duration // If we have a Retry-After duration for 429 errors, sleep for that duration
// This ensures we respect the API's requested delay before Reactor applies its backoff // This ensures we respect the API's requested delay before Reactor applies its
// backoff
Duration customDelay = retryAfterDuration.get(); Duration customDelay = retryAfterDuration.get();
if (customDelay != null && failure instanceof WebClientResponseException) { if (customDelay != null && failure instanceof WebClientResponseException) {
WebClientResponseException wcre = (WebClientResponseException) failure; WebClientResponseException wcre = (WebClientResponseException) failure;
@@ -474,14 +506,15 @@ public class OpenAIAnalyticsService {
} }
} }
} }
// Log retry attempt // Log retry attempt
if (failure instanceof WebClientResponseException) { if (failure instanceof WebClientResponseException) {
WebClientResponseException wcre = (WebClientResponseException) failure; WebClientResponseException wcre = (WebClientResponseException) failure;
int statusCode = wcre.getStatusCode().value(); int statusCode = wcre.getStatusCode().value();
int maxAttempts = statusCode == 429 ? rateLimitMaxAttempts : maxRetryAttempts; int maxAttempts = statusCode == 429 ? rateLimitMaxAttempts : maxRetryAttempts;
if (statusCode == 429 && customDelay == null) { if (statusCode == 429 && customDelay == null) {
logger.warn("OpenAI API returned 429 for operation '{}'. Will retry with exponential backoff (attempt {}/{})", logger.warn(
"OpenAI API returned 429 for operation '{}'. Will retry with exponential backoff (attempt {}/{})",
operation, attempt, maxAttempts); operation, attempt, maxAttempts);
} else if (statusCode != 429) { } else if (statusCode != 429) {
logger.warn("OpenAI API returned {} for operation '{}'. Will retry (attempt {}/{})", logger.warn("OpenAI API returned {} for operation '{}'. Will retry (attempt {}/{})",
@@ -506,5 +539,5 @@ public class OpenAIAnalyticsService {
return retrySignal.failure(); return retrySignal.failure();
}); });
} }
} }
src/main/resources/application.properties
+1
View File
@@ -64,6 +64,7 @@ openai.api.url=https://api.openai.com/v1/chat/completions
openai.model.name=gpt-4o-mini openai.model.name=gpt-4o-mini
openai.model.name.text=gpt-4o openai.model.name.text=gpt-4o
openai.timeoutMs=90000 openai.timeoutMs=90000
openai.timeoutMs.instruction=180000
openai.retry.maxAttempts=3 openai.retry.maxAttempts=3
openai.retry.initialDelayMs=2000 openai.retry.initialDelayMs=2000
openai.retry.maxDelayMs=60000 openai.retry.maxDelayMs=60000