From 804a27405382abfb85a9abd389106048043af940 Mon Sep 17 00:00:00 2001 From: wooh Date: Tue, 8 Sep 2026 14:07:42 +0900 Subject: [PATCH 1/2] =?UTF-8?q?[Fix]=20=EB=8F=99=EC=A0=81=20Few-shot=20?= =?UTF-8?q?=ED=9B=84=EB=B3=B4=20=EC=BA=90=EC=8B=9C=EC=99=80=20=ED=8F=89?= =?UTF-8?q?=EA=B0=80=20=EB=88=84=EC=88=98=20=EB=B0=A9=EC=A7=80=20=EB=B3=B4?= =?UTF-8?q?=EA=B0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 후보별 document embedding 메모리 캐시 적용 - 캐시 miss 후보만 Cohere Embed API로 요청하도록 개선 - datasetVersion, caseId, 후보 내용을 캐시 키에 반영 - 후보 데이터 변경 시 선택 결과와 document embedding 자동 갱신 - TTL 만료 캐시 정리 및 동시 초기화 중 중복 호출 방지 - caseId가 달라도 JD, 문항, 답변이 동일한 후보 제외 - Unicode, 대소문자, 반복 공백을 정규화한 입력 hash 비교 적용 - 자기 참조 조건이 다른 선택 결과의 캐시 재사용 방지 - 캐시 및 평가 누수 방지 회귀 테스트 추가 --- .../fewshot/DefaultFewShotSearchService.java | 189 ++++++++++++++++-- .../DefaultFewShotSearchServiceTest.java | 78 +++++++- 2 files changed, 253 insertions(+), 14 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 fc199397..98774f1a 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 @@ -8,6 +8,7 @@ import java.nio.charset.StandardCharsets; import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; +import java.text.Normalizer; import java.time.Instant; import java.util.ArrayList; import java.util.Comparator; @@ -24,12 +25,15 @@ @Slf4j public class DefaultFewShotSearchService implements FewShotSearchService { 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]+"); private final FewShotCaseStore caseStore; private final FewShotSearchTextBuilder textBuilder; private final CohereEmbeddingClient cohereEmbeddingClient; private final FewShotProperties properties; - private final Map cache = new ConcurrentHashMap<>(); + private final Map selectionCache = new ConcurrentHashMap<>(); + private final Map documentEmbeddingCache = new ConcurrentHashMap<>(); + private final Object documentEmbeddingCacheMonitor = new Object(); public DefaultFewShotSearchService( FewShotCaseStore caseStore, @@ -50,22 +54,23 @@ public List searchRelevantFewShots(FewShotSearchQuery query return List.of(); } int requestedTopK = topK > 0 ? topK : properties.getSearch().getTopK(); - String cacheKey = cacheKey(query, requestedTopK); - CacheEntry cached = readCache(cacheKey); + List activeCases = caseStore.loadActiveCases(); + String datasetFingerprint = datasetFingerprint(activeCases); + String cacheKey = selectionCacheKey(query, requestedTopK, datasetFingerprint); + SelectionCacheEntry cached = readSelectionCache(cacheKey); if (cached != null) { log.debug("few-shot selection cache hit. selectedCount={}, datasetVersion={}", cached.selectedCases().size(), properties.getDatasetVersion()); return cached.selectedCases(); } long startedAt = System.nanoTime(); - List activeCases = caseStore.loadActiveCases(); List candidates = localPrefilter(activeCases, query); List selected = selectWithCohere(query, candidates, requestedTopK); if (selected.isEmpty() && properties.isFallbackEnabled()) { selected = selectLocally(query, candidates, requestedTopK, "local-fallback"); } if (properties.isCacheEnabled()) { - cache.put(cacheKey, new CacheEntry(selected, Instant.now().plus(properties.getCacheTtl()))); + selectionCache.put(cacheKey, new SelectionCacheEntry(selected, expiresAt())); } log.info( "dynamic few-shot selection completed. enabled=true, totalCandidates={}, filteredCandidates={}, selectedIds={}, sources={}, scores={}, latencyMs={}", @@ -93,7 +98,7 @@ private List selectWithCohere( .map(textBuilder::buildCandidateDocument) .toList(); float[] queryEmbedding = cohereEmbeddingClient.embedQuery(queryText); - List documentEmbeddings = cohereEmbeddingClient.embedDocuments(documents); + List documentEmbeddings = resolveDocumentEmbeddings(candidates, documents); List ranked = new ArrayList<>(); for (int i = 0; i < candidates.size(); i++) { double score = cosineSimilarity(queryEmbedding, documentEmbeddings.get(i)); @@ -113,10 +118,73 @@ private List selectWithCohere( } } + private List resolveDocumentEmbeddings( + List candidates, + List documents + ) { + if (!properties.isCacheEnabled()) { + return cohereEmbeddingClient.embedDocuments(documents); + } + synchronized (documentEmbeddingCacheMonitor) { + Instant now = Instant.now(); + documentEmbeddingCache.entrySet().removeIf(entry -> entry.getValue().expiresAt().isBefore(now)); + + List result = new ArrayList<>(java.util.Collections.nCopies(candidates.size(), null)); + List missingKeys = new ArrayList<>(); + List missingDocuments = new ArrayList<>(); + List missingIndexes = new ArrayList<>(); + int cacheHitCount = 0; + + for (int i = 0; i < candidates.size(); i++) { + String key = documentEmbeddingCacheKey(candidates.get(i), documents.get(i)); + DocumentEmbeddingCacheEntry cached = documentEmbeddingCache.get(key); + if (cached != null) { + result.set(i, cached.embedding()); + cacheHitCount++; + continue; + } + missingKeys.add(key); + missingDocuments.add(documents.get(i)); + missingIndexes.add(i); + } + + if (!missingDocuments.isEmpty()) { + List embeddedDocuments = cohereEmbeddingClient.embedDocuments(missingDocuments); + if (embeddedDocuments.size() != missingDocuments.size()) { + throw new IllegalStateException("Cohere document embedding count does not match candidate count."); + } + Instant expiresAt = expiresAt(); + for (int i = 0; i < embeddedDocuments.size(); i++) { + float[] embedding = embeddedDocuments.get(i); + documentEmbeddingCache.put( + missingKeys.get(i), + new DocumentEmbeddingCacheEntry(embedding, expiresAt) + ); + result.set(missingIndexes.get(i), embedding); + } + } + log.debug( + "few-shot document embedding cache resolved. hitCount={}, missCount={}, candidateCount={}, datasetVersion={}", + cacheHitCount, + missingDocuments.size(), + candidates.size(), + properties.getDatasetVersion() + ); + return List.copyOf(result); + } + } + private List localPrefilter(List activeCases, FewShotSearchQuery query) { int limit = Math.max(1, properties.getSearch().getCandidateLimit()); + String queryInputHash = normalizedInputHash( + query.mainTasks(), + query.qualifications(), + query.question(), + query.answer() + ); return activeCases.stream() .filter(fewShotCase -> !sameCase(query.caseId(), fewShotCase.id())) + .filter(fewShotCase -> !sameNormalizedInput(queryInputHash, fewShotCase)) .map(fewShotCase -> new LocalScore(fewShotCase, localScore(query, fewShotCase))) .sorted(Comparator .comparingDouble(LocalScore::score).reversed() @@ -179,29 +247,113 @@ private double localScore(FewShotSearchQuery query, FewShotCase fewShotCase) { return jaccard + fewShotCase.priority() / 1000.0; } - private CacheEntry readCache(String key) { + private SelectionCacheEntry readSelectionCache(String key) { if (!properties.isCacheEnabled()) { return null; } - CacheEntry entry = cache.get(key); + SelectionCacheEntry entry = selectionCache.get(key); if (entry == null) { return null; } if (entry.expiresAt().isBefore(Instant.now())) { - cache.remove(key); + selectionCache.remove(key); return null; } return entry; } - private String cacheKey(FewShotSearchQuery query, int topK) { - return sha256(properties.getDatasetVersion() + "\n" + topK + "\n" + textBuilder.buildQueryText(query)); + private String selectionCacheKey(FewShotSearchQuery query, int topK, String datasetFingerprint) { + return sha256( + datasetFingerprint + + "\n" + topK + + "\n" + defaultString(query.caseId()) + + "\n" + textBuilder.buildQueryText(query) + ); + } + + private String datasetFingerprint(List activeCases) { + StringBuilder source = new StringBuilder(properties.getDatasetVersion()); + for (FewShotCase fewShotCase : activeCases) { + source.append('\n') + .append(defaultString(fewShotCase.id())).append('\u001f') + .append(fewShotCase.source()).append('\u001f') + .append(fewShotCase.priority()).append('\u001f') + .append(textBuilder.buildCandidateDocument(fewShotCase)).append('\u001f') + .append(defaultString(fewShotCase.promptBlock())); + } + return sha256(source.toString()); + } + + private String documentEmbeddingCacheKey(FewShotCase fewShotCase, String document) { + return sha256( + properties.getDatasetVersion() + + "\n" + defaultString(fewShotCase.id()) + + "\n" + document + ); + } + + private Instant expiresAt() { + return Instant.now().plus(properties.getCacheTtl()); } private static boolean sameCase(String queryCaseId, String candidateId) { return StringUtils.hasText(queryCaseId) && queryCaseId.equals(candidateId); } + private static boolean sameNormalizedInput(String queryInputHash, FewShotCase fewShotCase) { + if (!StringUtils.hasText(queryInputHash)) { + return false; + } + String candidateInputHash = normalizedInputHash( + fewShotCase.mainTasks(), + fewShotCase.qualifications(), + fewShotCase.question(), + fewShotCase.sanitizedAnswer() + ); + return queryInputHash.equals(candidateInputHash); + } + + private static String normalizedInputHash( + List mainTasks, + List qualifications, + String question, + String answer + ) { + String normalizedMainTasks = normalizeInputSection(mainTasks == null ? "" : String.join("\n", mainTasks)); + String normalizedQualifications = normalizeInputSection( + qualifications == null ? "" : String.join("\n", qualifications) + ); + String normalizedQuestion = normalizeInputSection(question); + String normalizedAnswer = normalizeInputSection(answer); + if (normalizedMainTasks.isEmpty() + && normalizedQualifications.isEmpty() + && normalizedQuestion.isEmpty() + && normalizedAnswer.isEmpty()) { + return ""; + } + return sha256( + normalizedMainTasks + '\u001f' + + normalizedQualifications + '\u001f' + + normalizedQuestion + '\u001f' + + normalizedAnswer + ); + } + + private static String normalizeInputSection(String value) { + if (!StringUtils.hasText(value)) { + return ""; + } + String unicodeNormalized = Normalizer.normalize(value, Normalizer.Form.NFKC); + return NORMALIZED_INPUT_WHITESPACE_PATTERN.matcher(unicodeNormalized) + .replaceAll(" ") + .trim() + .toLowerCase(Locale.ROOT); + } + + private static String defaultString(String value) { + return value == null ? "" : value; + } + private static Set tokens(String text) { if (!StringUtils.hasText(text)) { return Set.of(); @@ -250,9 +402,20 @@ private static String sha256(String value) { private record LocalScore(FewShotCase fewShotCase, double score) { } - private record CacheEntry(List selectedCases, Instant expiresAt) { - private CacheEntry { + private record SelectionCacheEntry(List selectedCases, Instant expiresAt) { + private SelectionCacheEntry { selectedCases = selectedCases == null ? List.of() : List.copyOf(selectedCases); } } + + private record DocumentEmbeddingCacheEntry(float[] embedding, Instant expiresAt) { + private DocumentEmbeddingCacheEntry { + 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 48d7c9d6..72a4e5e7 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 @@ -10,6 +10,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -53,6 +54,23 @@ void excludesSelfReferenceCandidate() { .containsExactly("EV-02"); } + @Test + @DisplayName("caseId가 달라도 정규화된 JD, 문항, 답변이 같으면 후보에서 제외한다") + void excludesCandidateWithSameNormalizedInput() { + properties.setDynamicSelectionEnabled(true); + when(caseStore.loadActiveCases()).thenReturn(List.of( + caseItem("FS-SAME", " spring BOOT API를 개발했습니다. ", 10), + caseItem("FS-OTHER", "Spring Boot API를 운영하고 장애를 개선했습니다.", 5) + )); + when(cohereEmbeddingClient.embedQuery(any())).thenReturn(new float[]{1, 0}); + when(cohereEmbeddingClient.embedDocuments(any())).thenReturn(List.of(new float[]{1, 0})); + + List result = service.searchRelevantFewShots(query("EV-01"), 3); + + assertThat(result).extracting(item -> item.fewShotCase().id()) + .containsExactly("FS-OTHER"); + } + @Test @DisplayName("Cohere 선택 실패 시 로컬 선택으로 fallback한다") void fallsBackToLocalSelectionWhenCohereFails() { @@ -69,7 +87,65 @@ void fallsBackToLocalSelectionWhenCohereFails() { assertThat(result.getFirst().selectionMethod()).isEqualTo("local-fallback"); } + @Test + @DisplayName("서로 다른 검색 요청에서도 동일한 후보의 document embedding을 재사용한다") + void reusesDocumentEmbeddingAcrossDifferentQueries() { + 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} + )); + + 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(1)).embedDocuments(any()); + } + + @Test + @DisplayName("후보 내용이 변경되면 document embedding을 다시 생성한다") + void refreshesDocumentEmbeddingWhenCandidateContentChanges() { + properties.setDynamicSelectionEnabled(true); + when(caseStore.loadActiveCases()).thenReturn( + List.of(caseItem("FS-1", "Spring Boot API 개발", 0)), + List.of(caseItem("FS-1", "브랜드 운영 경험", 0)) + ); + when(cohereEmbeddingClient.embedQuery(any())).thenReturn(new float[]{1, 0}); + when(cohereEmbeddingClient.embedDocuments(any())).thenReturn(List.of(new float[]{1, 0})); + + 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(2)).embedDocuments(any()); + } + + @Test + @DisplayName("캐시가 비활성화되면 매 요청마다 document embedding을 생성한다") + void embedsDocumentsOnEveryRequestWhenCacheDisabled() { + properties.setDynamicSelectionEnabled(true); + properties.setCacheEnabled(false); + 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})); + + service.searchRelevantFewShots(query("EV-01", "Spring Boot API를 개발했습니다."), 1); + service.searchRelevantFewShots(query("EV-02", "Java 서버를 운영했습니다."), 1); + + verify(cohereEmbeddingClient, times(2)).embedDocuments(any()); + } + private static FewShotSearchQuery query(String caseId) { + return query(caseId, "Spring Boot API를 개발했습니다."); + } + + private static FewShotSearchQuery query(String caseId, String answer) { return new FewShotSearchQuery( caseId, "백엔드 개발", @@ -77,7 +153,7 @@ private static FewShotSearchQuery query(String caseId) { List.of("Spring Boot API 개발"), List.of("Java"), "지원 직무 경험", - "Spring Boot API를 개발했습니다." + answer ); } From 5a73891269a5dcd23ee0a85620907cd468431cb9 Mon Sep 17 00:00:00 2001 From: wooh Date: Tue, 8 Sep 2026 14:27:22 +0900 Subject: [PATCH 2/2] =?UTF-8?q?[Fix]=20Few-shot=20=EC=9E=84=EB=B2=A0?= =?UTF-8?q?=EB=94=A9=20=EC=BA=90=EC=8B=9C=20=EB=8F=99=EC=8B=9C=EC=84=B1?= =?UTF-8?q?=EA=B3=BC=20=EB=A7=8C=EB=A3=8C=20=EC=A0=95=EB=A6=AC=20=EB=B3=B4?= =?UTF-8?q?=EA=B0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 후보별 CompletableFuture 기반 in-flight 초기화 상태 추가 - 동일 후보 임베딩 요청의 Cohere 중복 호출 방지 - Cohere document embedding 호출을 동기화 구간 밖에서 실행 - 서로 다른 후보 키의 임베딩 호출 병렬 실행 허용 - 임베딩 성공 결과를 캐시에 저장한 뒤 대기 요청에 전달 - 임베딩 실패 시 예외 전파 및 in-flight 상태 정리 - selection cache 접근 시 전체 만료 항목 제거 - 동일 키 중복 방지, 서로 다른 키 병렬 처리 및 전역 만료 제거 테스트 추가 --- .../fewshot/DefaultFewShotSearchService.java | 143 ++++++++++++------ .../DefaultFewShotSearchServiceTest.java | 109 +++++++++++++ 2 files changed, 203 insertions(+), 49 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 98774f1a..b5daefb5 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 @@ -18,6 +18,7 @@ import java.util.Locale; import java.util.Map; import java.util.Set; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.regex.Pattern; @@ -33,7 +34,8 @@ public class DefaultFewShotSearchService implements FewShotSearchService { private final FewShotProperties properties; private final Map selectionCache = new ConcurrentHashMap<>(); private final Map documentEmbeddingCache = new ConcurrentHashMap<>(); - private final Object documentEmbeddingCacheMonitor = new Object(); + private final Map> documentEmbeddingInFlight = + new ConcurrentHashMap<>(); public DefaultFewShotSearchService( FewShotCaseStore caseStore, @@ -125,52 +127,92 @@ private List resolveDocumentEmbeddings( if (!properties.isCacheEnabled()) { return cohereEmbeddingClient.embedDocuments(documents); } - synchronized (documentEmbeddingCacheMonitor) { - Instant now = Instant.now(); - documentEmbeddingCache.entrySet().removeIf(entry -> entry.getValue().expiresAt().isBefore(now)); - - List result = new ArrayList<>(java.util.Collections.nCopies(candidates.size(), null)); - List missingKeys = new ArrayList<>(); - List missingDocuments = new ArrayList<>(); - List missingIndexes = new ArrayList<>(); - int cacheHitCount = 0; + Instant now = Instant.now(); + documentEmbeddingCache.entrySet().removeIf(entry -> entry.getValue().expiresAt().isBefore(now)); + + List result = new ArrayList<>(java.util.Collections.nCopies(candidates.size(), null)); + List pending = new ArrayList<>(); + int cacheHitCount = 0; + int inFlightReuseCount = 0; + + for (int i = 0; i < candidates.size(); i++) { + String key = documentEmbeddingCacheKey(candidates.get(i), documents.get(i)); + DocumentEmbeddingCacheEntry cached = documentEmbeddingCache.get(key); + if (cached != null) { + result.set(i, cached.embedding()); + cacheHitCount++; + continue; + } - for (int i = 0; i < candidates.size(); i++) { - String key = documentEmbeddingCacheKey(candidates.get(i), documents.get(i)); - DocumentEmbeddingCacheEntry cached = documentEmbeddingCache.get(key); - if (cached != null) { - result.set(i, cached.embedding()); + CompletableFuture created = new CompletableFuture<>(); + CompletableFuture existing = documentEmbeddingInFlight.putIfAbsent(key, created); + boolean owner = existing == null; + if (owner) { + DocumentEmbeddingCacheEntry cachedAfterClaim = documentEmbeddingCache.get(key); + if (cachedAfterClaim != null) { + created.complete(cachedAfterClaim); + documentEmbeddingInFlight.remove(key, created); + result.set(i, cachedAfterClaim.embedding()); cacheHitCount++; continue; } - missingKeys.add(key); - missingDocuments.add(documents.get(i)); - missingIndexes.add(i); } - - if (!missingDocuments.isEmpty()) { - List embeddedDocuments = cohereEmbeddingClient.embedDocuments(missingDocuments); - if (embeddedDocuments.size() != missingDocuments.size()) { - throw new IllegalStateException("Cohere document embedding count does not match candidate count."); - } - Instant expiresAt = expiresAt(); - for (int i = 0; i < embeddedDocuments.size(); i++) { - float[] embedding = embeddedDocuments.get(i); - documentEmbeddingCache.put( - missingKeys.get(i), - new DocumentEmbeddingCacheEntry(embedding, expiresAt) - ); - result.set(missingIndexes.get(i), embedding); - } + if (!owner) { + inFlightReuseCount++; } - log.debug( - "few-shot document embedding cache resolved. hitCount={}, missCount={}, candidateCount={}, datasetVersion={}", - cacheHitCount, - missingDocuments.size(), - candidates.size(), - properties.getDatasetVersion() + pending.add(new PendingDocumentEmbedding( + i, + key, + documents.get(i), + owner ? created : existing, + owner + )); + } + + initializeMissingDocumentEmbeddings(pending); + for (PendingDocumentEmbedding item : pending) { + result.set(item.index(), item.future().join().embedding()); + } + log.debug( + "few-shot document embedding cache resolved. hitCount={}, initializedCount={}, inFlightReuseCount={}, candidateCount={}, datasetVersion={}", + cacheHitCount, + pending.stream().filter(PendingDocumentEmbedding::owner).count(), + inFlightReuseCount, + candidates.size(), + properties.getDatasetVersion() + ); + return List.copyOf(result); + } + + private void initializeMissingDocumentEmbeddings(List pending) { + List owned = pending.stream() + .filter(PendingDocumentEmbedding::owner) + .toList(); + if (owned.isEmpty()) { + return; + } + try { + List embeddedDocuments = cohereEmbeddingClient.embedDocuments( + owned.stream().map(PendingDocumentEmbedding::document).toList() ); - return List.copyOf(result); + if (embeddedDocuments.size() != owned.size()) { + throw new IllegalStateException("Cohere document embedding count does not match candidate count."); + } + Instant expiresAt = expiresAt(); + for (int i = 0; i < owned.size(); i++) { + PendingDocumentEmbedding item = owned.get(i); + DocumentEmbeddingCacheEntry entry = new DocumentEmbeddingCacheEntry( + embeddedDocuments.get(i), + expiresAt + ); + documentEmbeddingCache.put(item.key(), entry); + item.future().complete(entry); + } + } catch (RuntimeException | Error e) { + owned.forEach(item -> item.future().completeExceptionally(e)); + throw e; + } finally { + owned.forEach(item -> documentEmbeddingInFlight.remove(item.key(), item.future())); } } @@ -251,15 +293,9 @@ private SelectionCacheEntry readSelectionCache(String key) { if (!properties.isCacheEnabled()) { return null; } - SelectionCacheEntry entry = selectionCache.get(key); - if (entry == null) { - return null; - } - if (entry.expiresAt().isBefore(Instant.now())) { - selectionCache.remove(key); - return null; - } - return entry; + Instant now = Instant.now(); + selectionCache.entrySet().removeIf(entry -> entry.getValue().expiresAt().isBefore(now)); + return selectionCache.get(key); } private String selectionCacheKey(FewShotSearchQuery query, int topK, String datasetFingerprint) { @@ -402,6 +438,15 @@ private static String sha256(String value) { private record LocalScore(FewShotCase fewShotCase, double score) { } + private record PendingDocumentEmbedding( + int index, + String key, + String document, + CompletableFuture future, + boolean owner + ) { + } + private record SelectionCacheEntry(List selectedCases, Instant expiresAt) { private SelectionCacheEntry { selectedCases = selectedCases == null ? List.of() : List.copyOf(selectedCases); 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 72a4e5e7..d6972f45 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 @@ -3,8 +3,17 @@ import com.jobdri.jobdri_api.global.cohere.CohereEmbeddingClient; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; +import org.springframework.test.util.ReflectionTestUtils; +import java.time.Duration; import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; @@ -141,6 +150,106 @@ void embedsDocumentsOnEveryRequestWhenCacheDisabled() { verify(cohereEmbeddingClient, times(2)).embedDocuments(any()); } + @Test + @DisplayName("서로 다른 후보의 document embedding 호출은 동시에 실행할 수 있다") + void embedsDifferentDocumentKeysConcurrently() throws Exception { + properties.setDynamicSelectionEnabled(true); + properties.getSearch().setCandidateLimit(1); + when(caseStore.loadActiveCases()).thenReturn(List.of( + caseItem("FS-A", "alphaonly reference", 0), + caseItem("FS-B", "betaonly reference", 0) + )); + when(cohereEmbeddingClient.embedQuery(any())).thenReturn(new float[]{1, 0}); + CountDownLatch callsStarted = new CountDownLatch(2); + CountDownLatch releaseCalls = new CountDownLatch(1); + AtomicInteger activeCalls = new AtomicInteger(); + AtomicInteger maxActiveCalls = new AtomicInteger(); + when(cohereEmbeddingClient.embedDocuments(any())).thenAnswer(invocation -> { + int active = activeCalls.incrementAndGet(); + maxActiveCalls.accumulateAndGet(active, Math::max); + callsStarted.countDown(); + try { + if (!releaseCalls.await(3, TimeUnit.SECONDS)) { + throw new IllegalStateException("concurrent embedding calls did not start in time"); + } + return List.of(new float[]{1, 0}); + } finally { + activeCalls.decrementAndGet(); + } + }); + + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future> alpha = executor.submit( + () -> service.searchRelevantFewShots(query("EV-A", "alphaonly request"), 1) + ); + Future> beta = executor.submit( + () -> service.searchRelevantFewShots(query("EV-B", "betaonly request"), 1) + ); + + assertThat(callsStarted.await(2, TimeUnit.SECONDS)).isTrue(); + assertThat(maxActiveCalls.get()).isEqualTo(2); + releaseCalls.countDown(); + assertThat(alpha.get(2, TimeUnit.SECONDS)).hasSize(1); + assertThat(beta.get(2, TimeUnit.SECONDS)).hasSize(1); + } finally { + releaseCalls.countDown(); + executor.shutdownNow(); + } + } + + @Test + @DisplayName("동일 후보의 동시 요청은 document embedding 호출을 공유한다") + void deduplicatesConcurrentDocumentEmbeddingForSameKey() throws Exception { + properties.setDynamicSelectionEnabled(true); + when(caseStore.loadActiveCases()).thenReturn(List.of(caseItem("FS-1", "shared reference", 0))); + when(cohereEmbeddingClient.embedQuery(any())).thenReturn(new float[]{1, 0}); + CountDownLatch embeddingStarted = new CountDownLatch(1); + CountDownLatch releaseEmbedding = new CountDownLatch(1); + when(cohereEmbeddingClient.embedDocuments(any())).thenAnswer(invocation -> { + embeddingStarted.countDown(); + if (!releaseEmbedding.await(3, TimeUnit.SECONDS)) { + throw new IllegalStateException("document embedding did not finish in time"); + } + return List.of(new float[]{1, 0}); + }); + + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future> first = executor.submit( + () -> service.searchRelevantFewShots(query("EV-A", "first request"), 1) + ); + assertThat(embeddingStarted.await(2, TimeUnit.SECONDS)).isTrue(); + Future> second = executor.submit( + () -> service.searchRelevantFewShots(query("EV-B", "second request"), 1) + ); + + releaseEmbedding.countDown(); + assertThat(first.get(2, TimeUnit.SECONDS)).hasSize(1); + assertThat(second.get(2, TimeUnit.SECONDS)).hasSize(1); + verify(cohereEmbeddingClient, times(1)).embedDocuments(any()); + } finally { + releaseEmbedding.countDown(); + executor.shutdownNow(); + } + } + + @Test + @DisplayName("선택 캐시 접근 시 다른 키의 만료 항목도 함께 제거한다") + void evictsExpiredSelectionEntriesGlobally() { + 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})); + + service.searchRelevantFewShots(query("EV-01", "첫 번째 요청"), 1); + service.searchRelevantFewShots(query("EV-02", "두 번째 요청"), 1); + + Map selectionCache = (Map) ReflectionTestUtils.getField(service, "selectionCache"); + assertThat(selectionCache).hasSize(1); + } + private static FewShotSearchQuery query(String caseId) { return query(caseId, "Spring Boot API를 개발했습니다."); }