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..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 @@ -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; @@ -17,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; @@ -24,12 +26,16 @@ @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 Map> documentEmbeddingInFlight = + new ConcurrentHashMap<>(); public DefaultFewShotSearchService( FewShotCaseStore caseStore, @@ -50,22 +56,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 +100,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 +120,113 @@ private List selectWithCohere( } } + private List resolveDocumentEmbeddings( + List candidates, + List documents + ) { + if (!properties.isCacheEnabled()) { + return cohereEmbeddingClient.embedDocuments(documents); + } + 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; + } + + 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; + } + } + if (!owner) { + inFlightReuseCount++; + } + 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() + ); + 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())); + } + } + 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 +289,107 @@ 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); - if (entry == null) { - return null; - } - if (entry.expiresAt().isBefore(Instant.now())) { - cache.remove(key); - return null; + 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) { + 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 entry; + return sha256(source.toString()); + } + + private String documentEmbeddingCacheKey(FewShotCase fewShotCase, String document) { + return sha256( + properties.getDatasetVersion() + + "\n" + defaultString(fewShotCase.id()) + + "\n" + document + ); } - private String cacheKey(FewShotSearchQuery query, int topK) { - return sha256(properties.getDatasetVersion() + "\n" + topK + "\n" + textBuilder.buildQueryText(query)); + 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 +438,29 @@ private static String sha256(String value) { private record LocalScore(FewShotCase fewShotCase, double score) { } - private record CacheEntry(List selectedCases, Instant expiresAt) { - private CacheEntry { + 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); } } + + 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..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,13 +3,23 @@ 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; 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 +63,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 +96,165 @@ 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()); + } + + @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를 개발했습니다."); + } + + private static FewShotSearchQuery query(String caseId, String answer) { return new FewShotSearchQuery( caseId, "백엔드 개발", @@ -77,7 +262,7 @@ private static FewShotSearchQuery query(String caseId) { List.of("Spring Boot API 개발"), List.of("Java"), "지원 직무 경험", - "Spring Boot API를 개발했습니다." + answer ); }