From fc85f5cab882645bf1ed2fa53efb81f0a5db073b Mon Sep 17 00:00:00 2001 From: wooh Date: Tue, 8 Sep 2026 15:18:46 +0900 Subject: [PATCH 1/3] =?UTF-8?q?[Fix]=20Few-shot=20query=20embedding=20?= =?UTF-8?q?=EC=BA=90=EC=8B=9C=20=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 검색 질의 해시 기반 query embedding 캐시 구현 - 동일 질의의 반복 Cohere embedQuery 호출 방지 - topK 또는 후보 데이터셋이 달라도 동일 질의 임베딩 재사용 - CompletableFuture 기반 동일 질의 in-flight 요청 공유 - Cohere API 호출을 캐시 상태 처리 구간 밖에서 실행 - 만료된 query embedding 캐시 항목 전역 정리 - 임베딩 배열 저장 및 조회 시 방어적 복사 적용 - 캐시 비활성화 시 기존 호출 방식 유지 - 반복 요청, 동시 요청 및 캐시 비활성화 회귀 테스트 추가 --- .../fewshot/DefaultFewShotSearchService.java | 61 ++++++++++++++++++- .../DefaultFewShotSearchServiceTest.java | 61 ++++++++++++++++++- 2 files changed, 120 insertions(+), 2 deletions(-) diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchService.java index 032cd3bc..b2305484 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchService.java @@ -33,6 +33,9 @@ public class DefaultFewShotSearchService implements FewShotSearchService { private final CohereEmbeddingClient cohereEmbeddingClient; private final FewShotProperties properties; private final Map selectionCache = new ConcurrentHashMap<>(); + private final Map queryEmbeddingCache = new ConcurrentHashMap<>(); + private final Map> queryEmbeddingInFlight = + new ConcurrentHashMap<>(); private final Map documentEmbeddingCache = new ConcurrentHashMap<>(); private final Map> documentEmbeddingInFlight = new ConcurrentHashMap<>(); @@ -115,7 +118,7 @@ private List selectWithCohere( List documents = candidates.stream() .map(textBuilder::buildCandidateDocument) .toList(); - float[] queryEmbedding = cohereEmbeddingClient.embedQuery(queryText); + float[] queryEmbedding = resolveQueryEmbedding(queryText); List documentEmbeddings = resolveDocumentEmbeddings(candidates, documents); List ranked = new ArrayList<>(); List similarityScores = new ArrayList<>(); @@ -161,6 +164,51 @@ private static String formatScore(double score) { return String.format(Locale.ROOT, "%.4f", score); } + private float[] resolveQueryEmbedding(String queryText) { + if (!properties.isCacheEnabled()) { + return cohereEmbeddingClient.embedQuery(queryText); + } + Instant now = Instant.now(); + queryEmbeddingCache.entrySet().removeIf(entry -> entry.getValue().expiresAt().isBefore(now)); + String key = sha256(queryText); + QueryEmbeddingCacheEntry cached = queryEmbeddingCache.get(key); + if (cached != null) { + log.debug("few-shot query embedding cache hit."); + return cached.embedding(); + } + + CompletableFuture created = new CompletableFuture<>(); + CompletableFuture existing = queryEmbeddingInFlight.putIfAbsent(key, created); + if (existing != null) { + log.debug("few-shot query embedding in-flight request reused."); + return existing.join().embedding(); + } + + QueryEmbeddingCacheEntry cachedAfterClaim = queryEmbeddingCache.get(key); + if (cachedAfterClaim != null) { + created.complete(cachedAfterClaim); + queryEmbeddingInFlight.remove(key, created); + log.debug("few-shot query embedding cache hit after in-flight claim."); + return cachedAfterClaim.embedding(); + } + + try { + QueryEmbeddingCacheEntry initialized = new QueryEmbeddingCacheEntry( + cohereEmbeddingClient.embedQuery(queryText), + expiresAt() + ); + queryEmbeddingCache.put(key, initialized); + created.complete(initialized); + log.debug("few-shot query embedding cache initialized."); + return initialized.embedding(); + } catch (RuntimeException | Error e) { + created.completeExceptionally(e); + throw e; + } finally { + queryEmbeddingInFlight.remove(key, created); + } + } + private List resolveDocumentEmbeddings( List candidates, List documents @@ -512,4 +560,15 @@ public float[] embedding() { return embedding.clone(); } } + + private record QueryEmbeddingCacheEntry(float[] embedding, Instant expiresAt) { + private QueryEmbeddingCacheEntry { + embedding = embedding == null ? new float[0] : embedding.clone(); + } + + @Override + public float[] embedding() { + return embedding.clone(); + } + } } diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchServiceTest.java index 8f9b9968..b1e5cc55 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchServiceTest.java @@ -155,6 +155,64 @@ void reusesDocumentEmbeddingAcrossDifferentQueries() { verify(cohereEmbeddingClient, times(1)).embedDocuments(any()); } + @Test + @DisplayName("동일한 검색 질의를 다른 topK로 요청해도 query embedding을 재사용한다") + void reusesQueryEmbeddingAcrossDifferentTopK() { + properties.setDynamicSelectionEnabled(true); + when(caseStore.loadActiveCases()).thenReturn(List.of( + caseItem("FS-1", "Spring Boot API 개발", 0), + caseItem("FS-2", "브랜드 운영", 0) + )); + when(cohereEmbeddingClient.embedQuery(any())).thenReturn(new float[]{1, 0}); + when(cohereEmbeddingClient.embedDocuments(any())).thenReturn(List.of( + new float[]{1, 0}, + new float[]{0, 1} + )); + FewShotSearchQuery query = query("EV-01", "Spring Boot API를 개발했습니다."); + + service.searchRelevantFewShots(query, 1); + service.searchRelevantFewShots(query, 2); + + verify(cohereEmbeddingClient, times(1)).embedQuery(any()); + } + + @Test + @DisplayName("동일 질의의 동시 요청은 query embedding 호출을 공유한다") + void deduplicatesConcurrentQueryEmbeddingForSameQuery() throws Exception { + properties.setDynamicSelectionEnabled(true); + when(caseStore.loadActiveCases()).thenReturn(List.of(caseItem("FS-1", "shared reference", 0))); + CountDownLatch embeddingStarted = new CountDownLatch(1); + CountDownLatch releaseEmbedding = new CountDownLatch(1); + when(cohereEmbeddingClient.embedQuery(any())).thenAnswer(invocation -> { + embeddingStarted.countDown(); + if (!releaseEmbedding.await(3, TimeUnit.SECONDS)) { + throw new IllegalStateException("query embedding did not finish in time"); + } + return new float[]{1, 0}; + }); + when(cohereEmbeddingClient.embedDocuments(any())).thenReturn(List.of(new float[]{1, 0})); + FewShotSearchQuery query = query("EV-01", "shared request"); + + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future> first = executor.submit( + () -> service.searchRelevantFewShots(query, 1) + ); + assertThat(embeddingStarted.await(2, TimeUnit.SECONDS)).isTrue(); + Future> second = executor.submit( + () -> service.searchRelevantFewShots(query, 2) + ); + + releaseEmbedding.countDown(); + assertThat(first.get(2, TimeUnit.SECONDS)).hasSize(1); + assertThat(second.get(2, TimeUnit.SECONDS)).hasSize(1); + verify(cohereEmbeddingClient, times(1)).embedQuery(any()); + } finally { + releaseEmbedding.countDown(); + executor.shutdownNow(); + } + } + @Test @DisplayName("후보 내용이 변경되면 document embedding을 다시 생성한다") void refreshesDocumentEmbeddingWhenCandidateContentChanges() { @@ -169,7 +227,7 @@ void refreshesDocumentEmbeddingWhenCandidateContentChanges() { service.searchRelevantFewShots(query("EV-01", "Spring Boot API를 개발했습니다."), 1); service.searchRelevantFewShots(query("EV-01", "Spring Boot API를 개발했습니다."), 1); - verify(cohereEmbeddingClient, times(2)).embedQuery(any()); + verify(cohereEmbeddingClient, times(1)).embedQuery(any()); verify(cohereEmbeddingClient, times(2)).embedDocuments(any()); } @@ -185,6 +243,7 @@ void embedsDocumentsOnEveryRequestWhenCacheDisabled() { service.searchRelevantFewShots(query("EV-01", "Spring Boot API를 개발했습니다."), 1); service.searchRelevantFewShots(query("EV-02", "Java 서버를 운영했습니다."), 1); + verify(cohereEmbeddingClient, times(2)).embedQuery(any()); verify(cohereEmbeddingClient, times(2)).embedDocuments(any()); } From 1898c724cbba6f651bd623a97d82b3611288ffe7 Mon Sep 17 00:00:00 2001 From: Woohyeok Choi Date: Wed, 9 Sep 2026 14:43:45 +0900 Subject: [PATCH 2/3] =?UTF-8?q?[Fix]=20Few-shot=20query=20embedding=20?= =?UTF-8?q?=EC=BA=90=EC=8B=9C=20=EC=A0=95=EB=A6=AC=20=EC=A0=95=EC=B1=85=20?= =?UTF-8?q?=EB=B3=B4=EA=B0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - query embedding 캐시의 요청별 전체 만료 항목 순회 제거 - 요청 키의 TTL을 개별적으로 검사하도록 변경 - 1분 간격의 주기적 전역 만료 정리 적용 - query embedding 캐시 최대 크기 설정 추가 - 최대 크기 도달 시 오래된 항목을 일괄 제거하도록 개선 - 동시 캐시 정리 작업 중복 실행 방지 - 실패한 query embedding in-flight 상태 제거 검증 - 실패 후 동일 질의 재시도 및 정상 복구 테스트 추가 - TTL 만료와 최대 캐시 크기 회귀 테스트 추가 --- .../fewshot/DefaultFewShotSearchService.java | 55 +++++++++++++++- .../service/ai/fewshot/FewShotProperties.java | 9 +++ .../resources/application-analysis-eval.yaml | 1 + src/main/resources/application-dev.yaml | 1 + src/main/resources/application-prod.yaml | 1 + .../DefaultFewShotSearchServiceTest.java | 66 +++++++++++++++++++ 6 files changed, 130 insertions(+), 3 deletions(-) diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchService.java index b2305484..7e3ca7d8 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchService.java @@ -20,11 +20,14 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; import java.util.regex.Pattern; @Service @Slf4j public class DefaultFewShotSearchService implements FewShotSearchService { + private static final long QUERY_EMBEDDING_CACHE_CLEANUP_INTERVAL_MILLIS = 60_000L; private static final Pattern TOKEN_SPLIT_PATTERN = Pattern.compile("[^\\p{IsAlphabetic}\\p{IsDigit}가-힣]+"); private static final Pattern NORMALIZED_INPUT_WHITESPACE_PATTERN = Pattern.compile("[\\p{Z}\\s]+"); @@ -36,6 +39,8 @@ public class DefaultFewShotSearchService implements FewShotSearchService { private final Map queryEmbeddingCache = new ConcurrentHashMap<>(); private final Map> queryEmbeddingInFlight = new ConcurrentHashMap<>(); + private final AtomicLong queryEmbeddingCacheNextCleanupAt = new AtomicLong(); + private final AtomicBoolean queryEmbeddingCacheCleanupInProgress = new AtomicBoolean(); private final Map documentEmbeddingCache = new ConcurrentHashMap<>(); private final Map> documentEmbeddingInFlight = new ConcurrentHashMap<>(); @@ -169,9 +174,9 @@ private float[] resolveQueryEmbedding(String queryText) { return cohereEmbeddingClient.embedQuery(queryText); } Instant now = Instant.now(); - queryEmbeddingCache.entrySet().removeIf(entry -> entry.getValue().expiresAt().isBefore(now)); + maintainQueryEmbeddingCache(now); String key = sha256(queryText); - QueryEmbeddingCacheEntry cached = queryEmbeddingCache.get(key); + QueryEmbeddingCacheEntry cached = readQueryEmbeddingCache(key, now); if (cached != null) { log.debug("few-shot query embedding cache hit."); return cached.embedding(); @@ -184,7 +189,7 @@ private float[] resolveQueryEmbedding(String queryText) { return existing.join().embedding(); } - QueryEmbeddingCacheEntry cachedAfterClaim = queryEmbeddingCache.get(key); + QueryEmbeddingCacheEntry cachedAfterClaim = readQueryEmbeddingCache(key, Instant.now()); if (cachedAfterClaim != null) { created.complete(cachedAfterClaim); queryEmbeddingInFlight.remove(key, created); @@ -198,6 +203,7 @@ private float[] resolveQueryEmbedding(String queryText) { expiresAt() ); queryEmbeddingCache.put(key, initialized); + maintainQueryEmbeddingCache(Instant.now()); created.complete(initialized); log.debug("few-shot query embedding cache initialized."); return initialized.embedding(); @@ -209,6 +215,49 @@ private float[] resolveQueryEmbedding(String queryText) { } } + private QueryEmbeddingCacheEntry readQueryEmbeddingCache(String key, Instant now) { + QueryEmbeddingCacheEntry cached = queryEmbeddingCache.get(key); + if (cached == null || !cached.expiresAt().isBefore(now)) { + return cached; + } + queryEmbeddingCache.remove(key, cached); + return null; + } + + private void maintainQueryEmbeddingCache(Instant now) { + int maxSize = Math.max(1, properties.getQueryEmbeddingCacheMaxSize()); + long nowMillis = now.toEpochMilli(); + if (queryEmbeddingCache.size() < maxSize + && nowMillis < queryEmbeddingCacheNextCleanupAt.get()) { + return; + } + if (!queryEmbeddingCacheCleanupInProgress.compareAndSet(false, true)) { + return; + } + try { + int sizeBefore = queryEmbeddingCache.size(); + queryEmbeddingCache.entrySet().removeIf(entry -> entry.getValue().expiresAt().isBefore(now)); + int sizeAfterExpiration = queryEmbeddingCache.size(); + if (sizeAfterExpiration >= maxSize) { + int trimTarget = Math.max(1, maxSize - Math.max(1, maxSize / 10)); + int removalCount = sizeAfterExpiration - trimTarget; + queryEmbeddingCache.entrySet().stream() + .sorted(Comparator.comparing(entry -> entry.getValue().expiresAt())) + .limit(removalCount) + .forEach(entry -> queryEmbeddingCache.remove(entry.getKey(), entry.getValue())); + } + queryEmbeddingCacheNextCleanupAt.set(nowMillis + QUERY_EMBEDDING_CACHE_CLEANUP_INTERVAL_MILLIS); + log.debug( + "few-shot query embedding cache maintained. sizeBefore={}, sizeAfter={}, maxSize={}", + sizeBefore, + queryEmbeddingCache.size(), + maxSize + ); + } finally { + queryEmbeddingCacheCleanupInProgress.set(false); + } + } + private List resolveDocumentEmbeddings( List candidates, List documents diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/FewShotProperties.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/FewShotProperties.java index 7d883fdd..4718a38b 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/FewShotProperties.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/FewShotProperties.java @@ -16,6 +16,7 @@ public class FewShotProperties { private boolean fallbackEnabled = true; private boolean cacheEnabled = true; private Duration cacheTtl = Duration.ofMinutes(30); + private int queryEmbeddingCacheMaxSize = 1_000; private Source source = new Source(); private Search search = new Search(); @@ -83,6 +84,14 @@ public void setCacheTtl(Duration cacheTtl) { this.cacheTtl = cacheTtl; } + public int getQueryEmbeddingCacheMaxSize() { + return queryEmbeddingCacheMaxSize; + } + + public void setQueryEmbeddingCacheMaxSize(int queryEmbeddingCacheMaxSize) { + this.queryEmbeddingCacheMaxSize = queryEmbeddingCacheMaxSize; + } + public Source getSource() { return source; } diff --git a/src/main/resources/application-analysis-eval.yaml b/src/main/resources/application-analysis-eval.yaml index 24f93dc9..61130760 100644 --- a/src/main/resources/application-analysis-eval.yaml +++ b/src/main/resources/application-analysis-eval.yaml @@ -95,6 +95,7 @@ analysis: fallback-enabled: ${ANALYSIS_FEW_SHOT_FALLBACK_ENABLED:true} cache-enabled: ${ANALYSIS_FEW_SHOT_CACHE_ENABLED:true} cache-ttl: ${ANALYSIS_FEW_SHOT_CACHE_TTL:30m} + query-embedding-cache-max-size: ${ANALYSIS_FEW_SHOT_QUERY_EMBEDDING_CACHE_MAX_SIZE:1000} source: fixed-enabled: ${ANALYSIS_FEW_SHOT_FIXED_ENABLED:true} curated-enabled: ${ANALYSIS_FEW_SHOT_CURATED_ENABLED:true} diff --git a/src/main/resources/application-dev.yaml b/src/main/resources/application-dev.yaml index 2f32e674..d7298ea9 100644 --- a/src/main/resources/application-dev.yaml +++ b/src/main/resources/application-dev.yaml @@ -169,6 +169,7 @@ analysis: fallback-enabled: ${ANALYSIS_FEW_SHOT_FALLBACK_ENABLED:true} cache-enabled: ${ANALYSIS_FEW_SHOT_CACHE_ENABLED:true} cache-ttl: ${ANALYSIS_FEW_SHOT_CACHE_TTL:30m} + query-embedding-cache-max-size: ${ANALYSIS_FEW_SHOT_QUERY_EMBEDDING_CACHE_MAX_SIZE:1000} source: fixed-enabled: ${ANALYSIS_FEW_SHOT_FIXED_ENABLED:true} curated-enabled: ${ANALYSIS_FEW_SHOT_CURATED_ENABLED:true} diff --git a/src/main/resources/application-prod.yaml b/src/main/resources/application-prod.yaml index 4426a5f2..0a4af1db 100644 --- a/src/main/resources/application-prod.yaml +++ b/src/main/resources/application-prod.yaml @@ -171,6 +171,7 @@ analysis: fallback-enabled: ${ANALYSIS_FEW_SHOT_FALLBACK_ENABLED:true} cache-enabled: ${ANALYSIS_FEW_SHOT_CACHE_ENABLED:true} cache-ttl: ${ANALYSIS_FEW_SHOT_CACHE_TTL:30m} + query-embedding-cache-max-size: ${ANALYSIS_FEW_SHOT_QUERY_EMBEDDING_CACHE_MAX_SIZE:1000} source: fixed-enabled: ${ANALYSIS_FEW_SHOT_FIXED_ENABLED:true} curated-enabled: ${ANALYSIS_FEW_SHOT_CURATED_ENABLED:true} diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchServiceTest.java index b1e5cc55..acf08aab 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchServiceTest.java @@ -213,6 +213,72 @@ void deduplicatesConcurrentQueryEmbeddingForSameQuery() throws Exception { } } + @Test + @DisplayName("query embedding 실패 후 동일 질의를 재요청하면 Cohere 호출을 다시 시도한다") + void retriesSameQueryAfterQueryEmbeddingFailure() { + properties.setDynamicSelectionEnabled(true); + when(caseStore.loadActiveCases()).thenReturn(List.of( + caseItem("FS-1", "Spring Boot API 개발", 0), + caseItem("FS-2", "브랜드 운영", 0) + )); + when(cohereEmbeddingClient.embedQuery(any())) + .thenThrow(new RuntimeException("cohere down")) + .thenReturn(new float[]{1, 0}); + when(cohereEmbeddingClient.embedDocuments(any())).thenReturn(List.of( + new float[]{1, 0}, + new float[]{0, 1} + )); + FewShotSearchQuery query = query("EV-01", "retry request"); + + List failedAttempt = service.searchRelevantFewShots(query, 1); + Map inFlightAfterFailure = (Map) ReflectionTestUtils.getField( + service, + "queryEmbeddingInFlight" + ); + assertThat(failedAttempt).hasSize(1); + assertThat(failedAttempt.getFirst().selectionMethod()).isEqualTo("local-fallback"); + assertThat(inFlightAfterFailure).isEmpty(); + + List retried = service.searchRelevantFewShots(query, 2); + + assertThat(retried).hasSize(2); + assertThat(retried).allMatch(item -> item.selectionMethod().equals("cohere-embedding")); + verify(cohereEmbeddingClient, times(2)).embedQuery(any()); + } + + @Test + @DisplayName("query embedding 캐시는 설정된 최대 크기를 넘지 않도록 정리한다") + void boundsQueryEmbeddingCacheSize() { + properties.setDynamicSelectionEnabled(true); + properties.setQueryEmbeddingCacheMaxSize(3); + when(caseStore.loadActiveCases()).thenReturn(List.of(caseItem("FS-1", "Spring Boot API 개발", 0))); + when(cohereEmbeddingClient.embedQuery(any())).thenReturn(new float[]{1, 0}); + when(cohereEmbeddingClient.embedDocuments(any())).thenReturn(List.of(new float[]{1, 0})); + + for (int i = 0; i < 5; i++) { + service.searchRelevantFewShots(query("EV-" + i, "unique request " + i), 1); + } + + Map queryEmbeddingCache = (Map) ReflectionTestUtils.getField(service, "queryEmbeddingCache"); + assertThat(queryEmbeddingCache).hasSizeLessThanOrEqualTo(3); + } + + @Test + @DisplayName("만료된 query embedding은 동일 질의 재요청에 사용하지 않는다") + void doesNotReuseExpiredQueryEmbedding() { + properties.setDynamicSelectionEnabled(true); + properties.setCacheTtl(Duration.ZERO); + when(caseStore.loadActiveCases()).thenReturn(List.of(caseItem("FS-1", "Spring Boot API 개발", 0))); + when(cohereEmbeddingClient.embedQuery(any())).thenReturn(new float[]{1, 0}); + when(cohereEmbeddingClient.embedDocuments(any())).thenReturn(List.of(new float[]{1, 0})); + FewShotSearchQuery query = query("EV-01", "expired request"); + + service.searchRelevantFewShots(query, 1); + service.searchRelevantFewShots(query, 2); + + verify(cohereEmbeddingClient, times(2)).embedQuery(any()); + } + @Test @DisplayName("후보 내용이 변경되면 document embedding을 다시 생성한다") void refreshesDocumentEmbeddingWhenCandidateContentChanges() { From 79a5b5ea7e2cd78d0d1a6951628cf162e270dc2b Mon Sep 17 00:00:00 2001 From: Woohyeok Choi Date: Wed, 9 Sep 2026 15:38:51 +0900 Subject: [PATCH 3/3] =?UTF-8?q?[Fix]=20Few-shot=20query=20embedding=20?= =?UTF-8?q?=EC=BA=90=EC=8B=9C=20=EB=8C=80=EA=B8=B0=20=EB=B0=8F=20LRU=20?= =?UTF-8?q?=EC=A0=95=EC=B1=85=20=EB=B3=B4=EA=B0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - query embedding 캐시 hit 시 lastAccessedAt 갱신 - 캐시 최대 크기 정리 기준을 최근 접근 시각으로 변경 - 공유 query embedding future 대기에 timeout 적용 - timeout, 인터럽트 및 공유 future 실패 시 로컬 fallback 유지 - owner 임베딩 실패 시 모든 대기 요청의 fallback 동작 검증 - 실패한 in-flight 상태 제거 후 동일 질의 재시도 검증 - Cohere Retry-After 값을 최대 재시도 backoff 범위로 제한 - 최근 접근 항목 보존 및 미사용 항목 제거 테스트 추가 - in-flight 대기 timeout 설정을 개발·평가·운영 환경에 추가 --- .../fewshot/DefaultFewShotSearchService.java | 49 ++++++- .../service/ai/fewshot/FewShotProperties.java | 9 ++ .../global/cohere/CohereEmbeddingClient.java | 7 +- .../resources/application-analysis-eval.yaml | 1 + src/main/resources/application-dev.yaml | 1 + src/main/resources/application-prod.yaml | 1 + .../DefaultFewShotSearchServiceTest.java | 132 ++++++++++++++++++ .../cohere/CohereEmbeddingClientTest.java | 13 ++ 8 files changed, 205 insertions(+), 8 deletions(-) diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchService.java index 7e3ca7d8..2d548864 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchService.java @@ -20,6 +20,9 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.regex.Pattern; @@ -28,6 +31,7 @@ @Slf4j public class DefaultFewShotSearchService implements FewShotSearchService { private static final long QUERY_EMBEDDING_CACHE_CLEANUP_INTERVAL_MILLIS = 60_000L; + private static final long DEFAULT_QUERY_EMBEDDING_IN_FLIGHT_WAIT_TIMEOUT_MILLIS = 20_000L; private static final Pattern TOKEN_SPLIT_PATTERN = Pattern.compile("[^\\p{IsAlphabetic}\\p{IsDigit}가-힣]+"); private static final Pattern NORMALIZED_INPUT_WHITESPACE_PATTERN = Pattern.compile("[\\p{Z}\\s]+"); @@ -186,7 +190,7 @@ private float[] resolveQueryEmbedding(String queryText) { CompletableFuture existing = queryEmbeddingInFlight.putIfAbsent(key, created); if (existing != null) { log.debug("few-shot query embedding in-flight request reused."); - return existing.join().embedding(); + return awaitQueryEmbedding(existing).embedding(); } QueryEmbeddingCacheEntry cachedAfterClaim = readQueryEmbeddingCache(key, Instant.now()); @@ -217,11 +221,34 @@ private float[] resolveQueryEmbedding(String queryText) { private QueryEmbeddingCacheEntry readQueryEmbeddingCache(String key, Instant now) { QueryEmbeddingCacheEntry cached = queryEmbeddingCache.get(key); - if (cached == null || !cached.expiresAt().isBefore(now)) { - return cached; + if (cached == null) { + return null; + } + if (cached.expiresAt().isBefore(now)) { + queryEmbeddingCache.remove(key, cached); + return null; + } + QueryEmbeddingCacheEntry accessed = cached.accessedAt(now); + queryEmbeddingCache.replace(key, cached, accessed); + return accessed; + } + + private QueryEmbeddingCacheEntry awaitQueryEmbedding( + CompletableFuture existing + ) { + long timeoutMillis = properties.getQueryEmbeddingInFlightWaitTimeout() == null + ? DEFAULT_QUERY_EMBEDDING_IN_FLIGHT_WAIT_TIMEOUT_MILLIS + : Math.max(1L, properties.getQueryEmbeddingInFlightWaitTimeout().toMillis()); + try { + return existing.get(timeoutMillis, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Few-shot query embedding 대기 중 인터럽트되었습니다.", e); + } catch (ExecutionException e) { + throw new IllegalStateException("공유된 Few-shot query embedding 생성에 실패했습니다.", e.getCause()); + } catch (TimeoutException e) { + throw new IllegalStateException("공유된 Few-shot query embedding 대기 시간이 초과되었습니다.", e); } - queryEmbeddingCache.remove(key, cached); - return null; } private void maintainQueryEmbeddingCache(Instant now) { @@ -242,7 +269,7 @@ private void maintainQueryEmbeddingCache(Instant now) { int trimTarget = Math.max(1, maxSize - Math.max(1, maxSize / 10)); int removalCount = sizeAfterExpiration - trimTarget; queryEmbeddingCache.entrySet().stream() - .sorted(Comparator.comparing(entry -> entry.getValue().expiresAt())) + .sorted(Comparator.comparing(entry -> entry.getValue().lastAccessedAt())) .limit(removalCount) .forEach(entry -> queryEmbeddingCache.remove(entry.getKey(), entry.getValue())); } @@ -610,11 +637,19 @@ public float[] embedding() { } } - private record QueryEmbeddingCacheEntry(float[] embedding, Instant expiresAt) { + private record QueryEmbeddingCacheEntry(float[] embedding, Instant expiresAt, Instant lastAccessedAt) { + private QueryEmbeddingCacheEntry(float[] embedding, Instant expiresAt) { + this(embedding, expiresAt, Instant.now()); + } + private QueryEmbeddingCacheEntry { embedding = embedding == null ? new float[0] : embedding.clone(); } + private QueryEmbeddingCacheEntry accessedAt(Instant accessedAt) { + return new QueryEmbeddingCacheEntry(embedding, expiresAt, accessedAt); + } + @Override public float[] embedding() { return embedding.clone(); diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/FewShotProperties.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/FewShotProperties.java index 4718a38b..6a1d61c7 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/FewShotProperties.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/FewShotProperties.java @@ -17,6 +17,7 @@ public class FewShotProperties { private boolean cacheEnabled = true; private Duration cacheTtl = Duration.ofMinutes(30); private int queryEmbeddingCacheMaxSize = 1_000; + private Duration queryEmbeddingInFlightWaitTimeout = Duration.ofSeconds(20); private Source source = new Source(); private Search search = new Search(); @@ -92,6 +93,14 @@ public void setQueryEmbeddingCacheMaxSize(int queryEmbeddingCacheMaxSize) { this.queryEmbeddingCacheMaxSize = queryEmbeddingCacheMaxSize; } + public Duration getQueryEmbeddingInFlightWaitTimeout() { + return queryEmbeddingInFlightWaitTimeout; + } + + public void setQueryEmbeddingInFlightWaitTimeout(Duration queryEmbeddingInFlightWaitTimeout) { + this.queryEmbeddingInFlightWaitTimeout = queryEmbeddingInFlightWaitTimeout; + } + public Source getSource() { return source; } diff --git a/src/main/java/com/jobdri/jobdri_api/global/cohere/CohereEmbeddingClient.java b/src/main/java/com/jobdri/jobdri_api/global/cohere/CohereEmbeddingClient.java index 42d4b6de..86662aa3 100644 --- a/src/main/java/com/jobdri/jobdri_api/global/cohere/CohereEmbeddingClient.java +++ b/src/main/java/com/jobdri/jobdri_api/global/cohere/CohereEmbeddingClient.java @@ -94,7 +94,7 @@ private CohereEmbeddingResponse callCohere(CohereEmbeddingRequest request) { if (attempt == MAX_TRANSIENT_ATTEMPTS) { throw unavailable("Cohere Embed API가 일시적으로 응답할 수 없습니다.", e); } - Duration delay = e.retryAfter() != null ? e.retryAfter() : backoff; + Duration delay = boundedRetryDelay(e.retryAfter(), backoff); log.warn( "Cohere Embed API transient failure. attempt={}, maxAttempts={}, retryAfterMs={}, message={}", attempt, @@ -260,6 +260,11 @@ private static Duration nextBackoff(Duration current) { return next.compareTo(MAX_RETRY_BACKOFF) > 0 ? MAX_RETRY_BACKOFF : next; } + static Duration boundedRetryDelay(Duration retryAfter, Duration backoff) { + Duration requested = retryAfter != null ? retryAfter : backoff; + return requested.compareTo(MAX_RETRY_BACKOFF) > 0 ? MAX_RETRY_BACKOFF : requested; + } + private static void sleepBeforeRetry(Duration delay) { try { Thread.sleep(delay.toMillis()); diff --git a/src/main/resources/application-analysis-eval.yaml b/src/main/resources/application-analysis-eval.yaml index 61130760..e4577618 100644 --- a/src/main/resources/application-analysis-eval.yaml +++ b/src/main/resources/application-analysis-eval.yaml @@ -96,6 +96,7 @@ analysis: cache-enabled: ${ANALYSIS_FEW_SHOT_CACHE_ENABLED:true} cache-ttl: ${ANALYSIS_FEW_SHOT_CACHE_TTL:30m} query-embedding-cache-max-size: ${ANALYSIS_FEW_SHOT_QUERY_EMBEDDING_CACHE_MAX_SIZE:1000} + query-embedding-in-flight-wait-timeout: ${ANALYSIS_FEW_SHOT_QUERY_EMBEDDING_IN_FLIGHT_WAIT_TIMEOUT:20s} source: fixed-enabled: ${ANALYSIS_FEW_SHOT_FIXED_ENABLED:true} curated-enabled: ${ANALYSIS_FEW_SHOT_CURATED_ENABLED:true} diff --git a/src/main/resources/application-dev.yaml b/src/main/resources/application-dev.yaml index d7298ea9..9090d914 100644 --- a/src/main/resources/application-dev.yaml +++ b/src/main/resources/application-dev.yaml @@ -170,6 +170,7 @@ analysis: cache-enabled: ${ANALYSIS_FEW_SHOT_CACHE_ENABLED:true} cache-ttl: ${ANALYSIS_FEW_SHOT_CACHE_TTL:30m} query-embedding-cache-max-size: ${ANALYSIS_FEW_SHOT_QUERY_EMBEDDING_CACHE_MAX_SIZE:1000} + query-embedding-in-flight-wait-timeout: ${ANALYSIS_FEW_SHOT_QUERY_EMBEDDING_IN_FLIGHT_WAIT_TIMEOUT:20s} source: fixed-enabled: ${ANALYSIS_FEW_SHOT_FIXED_ENABLED:true} curated-enabled: ${ANALYSIS_FEW_SHOT_CURATED_ENABLED:true} diff --git a/src/main/resources/application-prod.yaml b/src/main/resources/application-prod.yaml index 0a4af1db..5aef5b92 100644 --- a/src/main/resources/application-prod.yaml +++ b/src/main/resources/application-prod.yaml @@ -172,6 +172,7 @@ analysis: cache-enabled: ${ANALYSIS_FEW_SHOT_CACHE_ENABLED:true} cache-ttl: ${ANALYSIS_FEW_SHOT_CACHE_TTL:30m} query-embedding-cache-max-size: ${ANALYSIS_FEW_SHOT_QUERY_EMBEDDING_CACHE_MAX_SIZE:1000} + query-embedding-in-flight-wait-timeout: ${ANALYSIS_FEW_SHOT_QUERY_EMBEDDING_IN_FLIGHT_WAIT_TIMEOUT:20s} source: fixed-enabled: ${ANALYSIS_FEW_SHOT_FIXED_ENABLED:true} curated-enabled: ${ANALYSIS_FEW_SHOT_CURATED_ENABLED:true} diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchServiceTest.java index acf08aab..0c01cf00 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/DefaultFewShotSearchServiceTest.java @@ -16,6 +16,7 @@ import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -213,6 +214,137 @@ void deduplicatesConcurrentQueryEmbeddingForSameQuery() throws Exception { } } + @Test + @DisplayName("공유 query embedding이 실패하면 모든 대기 요청이 fallback하고 후속 요청은 재시도한다") + void recoversWaitingRequestsAfterSharedQueryEmbeddingFailure() throws Exception { + properties.setDynamicSelectionEnabled(true); + when(caseStore.loadActiveCases()).thenReturn(List.of( + caseItem("FS-1", "Spring Boot API 개발", 0), + caseItem("FS-2", "브랜드 운영", 0), + caseItem("FS-3", "데이터 분석", 0) + )); + CountDownLatch embeddingStarted = new CountDownLatch(1); + CountDownLatch releaseFailure = new CountDownLatch(1); + when(cohereEmbeddingClient.embedQuery(any())) + .thenAnswer(invocation -> { + embeddingStarted.countDown(); + if (!releaseFailure.await(3, TimeUnit.SECONDS)) { + throw new IllegalStateException("query embedding failure was not released in time"); + } + throw new RuntimeException("cohere down"); + }) + .thenReturn(new float[]{1, 0}); + when(cohereEmbeddingClient.embedDocuments(any())).thenReturn(List.of( + new float[]{1, 0}, + new float[]{0, 1}, + new float[]{0.5f, 0.5f} + )); + FewShotSearchQuery query = query("EV-01", "shared failure request"); + + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future> owner = executor.submit( + () -> service.searchRelevantFewShots(query, 1) + ); + assertThat(embeddingStarted.await(2, TimeUnit.SECONDS)).isTrue(); + Future> waiter = executor.submit( + () -> service.searchRelevantFewShots(query, 2) + ); + assertThatThrownBy(() -> waiter.get(100, TimeUnit.MILLISECONDS)) + .isInstanceOf(java.util.concurrent.TimeoutException.class); + + releaseFailure.countDown(); + assertThat(owner.get(2, TimeUnit.SECONDS)) + .allMatch(item -> item.selectionMethod().equals("local-fallback")); + assertThat(waiter.get(2, TimeUnit.SECONDS)) + .allMatch(item -> item.selectionMethod().equals("local-fallback")); + } finally { + releaseFailure.countDown(); + executor.shutdownNow(); + } + Map inFlightAfterFailure = (Map) ReflectionTestUtils.getField( + service, + "queryEmbeddingInFlight" + ); + assertThat(inFlightAfterFailure).isEmpty(); + + List retried = service.searchRelevantFewShots(query, 3); + + assertThat(retried).hasSize(3); + assertThat(retried).allMatch(item -> item.selectionMethod().equals("cohere-embedding")); + verify(cohereEmbeddingClient, times(2)).embedQuery(any()); + } + + @Test + @DisplayName("공유 query embedding 대기가 제한 시간을 넘으면 로컬 fallback한다") + void fallsBackLocallyWhenSharedQueryEmbeddingWaitTimesOut() throws Exception { + properties.setDynamicSelectionEnabled(true); + properties.setQueryEmbeddingInFlightWaitTimeout(Duration.ofMillis(50)); + when(caseStore.loadActiveCases()).thenReturn(List.of( + caseItem("FS-1", "Spring Boot API 개발", 0), + caseItem("FS-2", "브랜드 운영", 0) + )); + CountDownLatch embeddingStarted = new CountDownLatch(1); + CountDownLatch releaseEmbedding = new CountDownLatch(1); + when(cohereEmbeddingClient.embedQuery(any())).thenAnswer(invocation -> { + embeddingStarted.countDown(); + if (!releaseEmbedding.await(3, TimeUnit.SECONDS)) { + throw new IllegalStateException("query embedding did not finish in time"); + } + return new float[]{1, 0}; + }); + when(cohereEmbeddingClient.embedDocuments(any())).thenReturn(List.of( + new float[]{1, 0}, + new float[]{0, 1} + )); + FewShotSearchQuery query = query("EV-01", "timeout request"); + + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future> owner = executor.submit( + () -> service.searchRelevantFewShots(query, 1) + ); + assertThat(embeddingStarted.await(2, TimeUnit.SECONDS)).isTrue(); + Future> waiter = executor.submit( + () -> service.searchRelevantFewShots(query, 2) + ); + + assertThat(waiter.get(1, TimeUnit.SECONDS)) + .allMatch(item -> item.selectionMethod().equals("local-fallback")); + releaseEmbedding.countDown(); + assertThat(owner.get(2, TimeUnit.SECONDS)).hasSize(1); + verify(cohereEmbeddingClient, times(1)).embedQuery(any()); + } finally { + releaseEmbedding.countDown(); + executor.shutdownNow(); + } + } + + @Test + @DisplayName("query embedding 캐시는 최근 접근 항목을 유지하고 오래 사용하지 않은 항목을 제거한다") + void evictsLeastRecentlyAccessedQueryEmbedding() throws Exception { + properties.setDynamicSelectionEnabled(true); + properties.setQueryEmbeddingCacheMaxSize(3); + when(caseStore.loadActiveCases()).thenReturn(List.of(caseItem("FS-1", "Spring Boot API 개발", 0))); + when(cohereEmbeddingClient.embedQuery(any())).thenReturn(new float[]{1, 0}); + when(cohereEmbeddingClient.embedDocuments(any())).thenReturn(List.of(new float[]{1, 0})); + FewShotSearchQuery first = query("EV-A", "first request"); + FewShotSearchQuery second = query("EV-B", "second request"); + FewShotSearchQuery third = query("EV-C", "third request"); + + service.searchRelevantFewShots(first, 1); + TimeUnit.MILLISECONDS.sleep(2); + service.searchRelevantFewShots(second, 1); + TimeUnit.MILLISECONDS.sleep(2); + service.searchRelevantFewShots(first, 2); + TimeUnit.MILLISECONDS.sleep(2); + service.searchRelevantFewShots(third, 1); + service.searchRelevantFewShots(first, 3); + service.searchRelevantFewShots(second, 2); + + verify(cohereEmbeddingClient, times(4)).embedQuery(any()); + } + @Test @DisplayName("query embedding 실패 후 동일 질의를 재요청하면 Cohere 호출을 다시 시도한다") void retriesSameQueryAfterQueryEmbeddingFailure() { diff --git a/src/test/java/com/jobdri/jobdri_api/global/cohere/CohereEmbeddingClientTest.java b/src/test/java/com/jobdri/jobdri_api/global/cohere/CohereEmbeddingClientTest.java index 51970f26..6d2c9e10 100644 --- a/src/test/java/com/jobdri/jobdri_api/global/cohere/CohereEmbeddingClientTest.java +++ b/src/test/java/com/jobdri/jobdri_api/global/cohere/CohereEmbeddingClientTest.java @@ -138,6 +138,19 @@ void retryTransientCohereError() throws Exception { } } + @Test + @DisplayName("서버 Retry-After 값은 최대 재시도 대기 시간을 넘지 않는다") + void capsRetryAfterDelay() { + assertThat(CohereEmbeddingClient.boundedRetryDelay( + Duration.ofHours(1), + Duration.ofMillis(200) + )).isEqualTo(Duration.ofSeconds(2)); + assertThat(CohereEmbeddingClient.boundedRetryDelay( + null, + Duration.ofMillis(200) + )).isEqualTo(Duration.ofMillis(200)); + } + @Test @DisplayName("Cohere 400, 401, 403은 요청 또는 설정 오류로 변환한다") void requestOrConfigurationErrors() throws Exception {