diff --git a/docs/fewshot-cache-policy.md b/docs/fewshot-cache-policy.md new file mode 100644 index 0000000..a02716d --- /dev/null +++ b/docs/fewshot-cache-policy.md @@ -0,0 +1,45 @@ +# Few-shot 메모리 캐시 정책 + +## 기본값 + +| 캐시 | 최대 크기 | TTL | +| --- | ---: | ---: | +| selection | 1,000 | 30분 | +| query embedding | 1,000 | 30분 | +| document embedding | 5,000 | 30분 | + +환경변수로 조정할 수 있습니다. + +```text +ANALYSIS_FEW_SHOT_CACHE_TTL=30m +ANALYSIS_FEW_SHOT_SELECTION_CACHE_MAX_SIZE=1000 +ANALYSIS_FEW_SHOT_QUERY_EMBEDDING_CACHE_MAX_SIZE=1000 +ANALYSIS_FEW_SHOT_DOCUMENT_EMBEDDING_CACHE_MAX_SIZE=5000 +ANALYSIS_FEW_SHOT_SELECTION_IN_FLIGHT_WAIT_TIMEOUT=20s +``` + +모든 최대 크기는 1 이상으로 보정합니다. selection과 document embedding 캐시는 접근할 때마다 +만료 항목을 먼저 제거하고, 상한을 넘으면 마지막 접근 시각이 오래된 항목부터 제거합니다. +query embedding 캐시는 기존 주기적 정리와 LRU 제거를 유지합니다. + +selection과 document embedding 생성이 진행 중인 키는 제거 대상에서 제외합니다. 생성이 끝나 in-flight 상태가 +해제된 직후 다시 상한을 적용하므로, 동시에 한 batch가 완료되는 짧은 구간에는 상한을 일시적으로 +넘을 수 있지만 장시간 초과 상태로 남지 않습니다. selection/query/document의 in-flight future는 +캐시 제거와 별도로 유지되므로 동일 키의 동시 요청이 외부 API를 중복 호출하지 않습니다. + +## 관측 + +`fewshot_cache_events_total`을 `cache`, `outcome` 태그로 나눠 확인합니다. + +```promql +sum(rate(fewshot_cache_events_total[10m])) by (cache, outcome) + +sum(rate(fewshot_cache_events_total{outcome="hit"}[10m])) by (cache) +/ +sum(rate(fewshot_cache_events_total{outcome=~"hit|miss"}[10m])) by (cache) + +sum(increase(fewshot_cache_events_total{outcome="evicted"}[1h])) by (cache) +``` + +eviction이 지속 증가하면서 hit 비율이 낮으면 캐시 상한을 늘리기 전에 고유 질의 수, 데이터셋 변경 +빈도와 실제 메모리 사용량을 함께 확인합니다. diff --git a/docs/fewshot-operations.md b/docs/fewshot-operations.md index f3711b0..7f6bf69 100644 --- a/docs/fewshot-operations.md +++ b/docs/fewshot-operations.md @@ -18,6 +18,7 @@ Prometheus endpoint는 관리 포트의 `/actuator/prometheus`입니다. | `fewshot_selection_selected_candidates_count/_sum` | `mode`, `cache_hit` | 선택 후보 수 분포 | | `fewshot_cohere_logical_calls_total` | 없음 | Few-shot이 발생시킨 Cohere 논리 호출량 | | `fewshot_cohere_failure_count_total` | `reason` | Cohere 검색 실패 유형 | +| `fewshot_cache_events_total` | `cache`, `outcome` | 캐시별 hit·hit_after_claim·miss·expired·evicted 횟수 | 원문 JD·답변·검색 텍스트·embedding은 지표 태그나 로그에 넣지 않습니다. `reason`은 정해진 예외 분류만 허용하고 그 외 값은 `Other`로 묶어 tag cardinality 증가를 방지합니다. 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 21dee7a..628dbd6 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 @@ -31,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_SELECTION_IN_FLIGHT_WAIT_TIMEOUT_MILLIS = 20_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}가-힣]+"); @@ -40,6 +41,9 @@ public class DefaultFewShotSearchService implements FewShotSearchService { private final FewShotProperties properties; private final FewShotMetricsRecorder metricsRecorder; private final Map selectionCache = new ConcurrentHashMap<>(); + private final Map> selectionInFlight = new ConcurrentHashMap<>(); + private final AtomicBoolean selectionCacheCleanupInProgress = new AtomicBoolean(); + private final AtomicBoolean selectionCacheCleanupRequested = new AtomicBoolean(); private final Map queryEmbeddingCache = new ConcurrentHashMap<>(); private final Map> queryEmbeddingInFlight = new ConcurrentHashMap<>(); @@ -48,6 +52,8 @@ public class DefaultFewShotSearchService implements FewShotSearchService { private final Map documentEmbeddingCache = new ConcurrentHashMap<>(); private final Map> documentEmbeddingInFlight = new ConcurrentHashMap<>(); + private final AtomicBoolean documentEmbeddingCacheCleanupInProgress = new AtomicBoolean(); + private final AtomicBoolean documentEmbeddingCacheCleanupRequested = new AtomicBoolean(); @Autowired public DefaultFewShotSearchService( @@ -77,46 +83,96 @@ public List searchRelevantFewShots(FewShotSearchQuery query String cacheKey = selectionCacheKey(query, requestedTopK, datasetFingerprint); SelectionCacheEntry cached = readSelectionCache(cacheKey); if (cached != null) { - recordMetrics(cached.selectionMode(), true, cached.selectedCases().size(), startedAt); - log.debug( - "few-shot selection cache hit. selectionMode={}, selectedCount={}, datasetVersion={}", - cached.selectionMode(), - cached.selectedCases().size(), - properties.getDatasetVersion() - ); - return cached.selectedCases(); + return cachedSelection(cached, startedAt, "cache hit"); } - List candidates = localPrefilter(activeCases, query); - List selected = selectWithCohere(query, candidates, requestedTopK); - FewShotSelectionMode selectionMode = selected.isEmpty() - ? FewShotSelectionMode.STATIC_FALLBACK - : FewShotSelectionMode.EMBEDDING; - int minimumSelectedCount = Math.max( - 1, - Math.min(properties.getSearch().getMinimumSelectedCount(), requestedTopK) - ); - if (selected.size() < minimumSelectedCount && properties.isFallbackEnabled()) { - selected = selectLocally(query, candidates, requestedTopK, "local-fallback"); - if (!selected.isEmpty()) { - selectionMode = FewShotSelectionMode.LOCAL_FALLBACK; - } + CompletableFuture created = new CompletableFuture<>(); + CompletableFuture existing = selectionInFlight.putIfAbsent(cacheKey, created); + if (existing != null) { + return cachedSelection(awaitSelection(existing), startedAt, "in-flight reuse"); } - if (properties.isCacheEnabled()) { - selectionCache.put(cacheKey, new SelectionCacheEntry(selected, selectionMode, expiresAt())); + SelectionCacheEntry cachedAfterClaim = readSelectionCache(cacheKey, false); + if (cachedAfterClaim != null) { + created.complete(cachedAfterClaim); + selectionInFlight.remove(cacheKey, created); + return cachedSelection(cachedAfterClaim, startedAt, "cache hit after claim"); } - recordMetrics(selectionMode, false, selected.size(), startedAt); - log.info( - "dynamic few-shot selection completed. enabled=true, selectionMode={}, totalCandidates={}, filteredCandidates={}, selectedIds={}, sources={}, scores={}, latencyMs={}", - selectionMode, - activeCases.size(), - candidates.size(), - selected.stream().map(item -> item.fewShotCase().id()).toList(), - selected.stream().map(item -> item.fewShotCase().source()).toList(), - selected.stream().map(item -> "%.4f".formatted(item.score())).toList(), - (System.nanoTime() - startedAt) / 1_000_000 + + try { + List candidates = localPrefilter(activeCases, query); + List selected = selectWithCohere(query, candidates, requestedTopK); + FewShotSelectionMode selectionMode = selected.isEmpty() + ? FewShotSelectionMode.STATIC_FALLBACK + : FewShotSelectionMode.EMBEDDING; + int minimumSelectedCount = Math.max( + 1, + Math.min(properties.getSearch().getMinimumSelectedCount(), requestedTopK) + ); + if (selected.size() < minimumSelectedCount && properties.isFallbackEnabled()) { + selected = selectLocally(query, candidates, requestedTopK, "local-fallback"); + if (!selected.isEmpty()) { + selectionMode = FewShotSelectionMode.LOCAL_FALLBACK; + } + } + SelectionCacheEntry entry = new SelectionCacheEntry(selected, selectionMode, expiresAt()); + if (properties.isCacheEnabled()) { + selectionCache.put(cacheKey, entry); + maintainSelectionCache(true); + } + created.complete(entry); + recordMetrics(selectionMode, false, selected.size(), startedAt); + log.info( + "dynamic few-shot selection completed. enabled=true, selectionMode={}, totalCandidates={}, filteredCandidates={}, selectedIds={}, sources={}, scores={}, latencyMs={}", + selectionMode, + activeCases.size(), + candidates.size(), + selected.stream().map(item -> item.fewShotCase().id()).toList(), + selected.stream().map(item -> item.fewShotCase().source()).toList(), + selected.stream().map(item -> "%.4f".formatted(item.score())).toList(), + (System.nanoTime() - startedAt) / 1_000_000 + ); + return selected; + } catch (RuntimeException | Error e) { + created.completeExceptionally(e); + throw e; + } finally { + selectionInFlight.remove(cacheKey, created); + if (properties.isCacheEnabled()) { + maintainSelectionCache(true); + } + } + } + + private List cachedSelection( + SelectionCacheEntry cached, + long startedAt, + String source + ) { + recordMetrics(cached.selectionMode(), true, cached.selectedCases().size(), startedAt); + log.debug( + "few-shot selection {}. selectionMode={}, selectedCount={}, datasetVersion={}", + source, + cached.selectionMode(), + cached.selectedCases().size(), + properties.getDatasetVersion() ); - return selected; + return cached.selectedCases(); + } + + private SelectionCacheEntry awaitSelection(CompletableFuture existing) { + long timeoutMillis = properties.getSelectionInFlightWaitTimeout() == null + ? DEFAULT_SELECTION_IN_FLIGHT_WAIT_TIMEOUT_MILLIS + : Math.max(1L, properties.getSelectionInFlightWaitTimeout().toMillis()); + try { + return existing.get(timeoutMillis, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("공유된 Few-shot selection 대기 중 인터럽트되었습니다.", e); + } catch (ExecutionException e) { + throw new IllegalStateException("공유된 Few-shot selection 생성에 실패했습니다.", e.getCause()); + } catch (TimeoutException e) { + throw new IllegalStateException("공유된 Few-shot selection 대기 시간이 초과되었습니다.", e); + } } private void recordMetrics( @@ -213,7 +269,7 @@ private float[] resolveQueryEmbedding(String queryText) { return awaitQueryEmbedding(existing).embedding(); } - QueryEmbeddingCacheEntry cachedAfterClaim = readQueryEmbeddingCache(key, Instant.now()); + QueryEmbeddingCacheEntry cachedAfterClaim = readQueryEmbeddingCache(key, Instant.now(), false); if (cachedAfterClaim != null) { created.complete(cachedAfterClaim); queryEmbeddingInFlight.remove(key, created); @@ -240,16 +296,29 @@ private float[] resolveQueryEmbedding(String queryText) { } private QueryEmbeddingCacheEntry readQueryEmbeddingCache(String key, Instant now) { + return readQueryEmbeddingCache(key, now, true); + } + + private QueryEmbeddingCacheEntry readQueryEmbeddingCache(String key, Instant now, boolean recordEvent) { QueryEmbeddingCacheEntry cached = queryEmbeddingCache.get(key); if (cached == null) { + if (recordEvent) { + metricsRecorder.recordCacheEvent("query_embedding", "miss", 1L); + } return null; } if (cached.expiresAt().isBefore(now)) { queryEmbeddingCache.remove(key, cached); + if (recordEvent) { + metricsRecorder.recordCacheEvent("query_embedding", "expired", 1L); + } return null; } QueryEmbeddingCacheEntry accessed = cached.accessedAt(now); queryEmbeddingCache.replace(key, cached, accessed); + if (recordEvent) { + metricsRecorder.recordCacheEvent("query_embedding", "hit", 1L); + } return accessed; } @@ -285,6 +354,7 @@ private void maintainQueryEmbeddingCache(Instant now) { int sizeBefore = queryEmbeddingCache.size(); queryEmbeddingCache.entrySet().removeIf(entry -> entry.getValue().expiresAt().isBefore(now)); int sizeAfterExpiration = queryEmbeddingCache.size(); + metricsRecorder.recordCacheEvent("query_embedding", "expired", sizeBefore - sizeAfterExpiration); if (sizeAfterExpiration >= maxSize) { int trimTarget = Math.max(1, maxSize - Math.max(1, maxSize / 10)); int removalCount = sizeAfterExpiration - trimTarget; @@ -293,6 +363,8 @@ private void maintainQueryEmbeddingCache(Instant now) { .limit(removalCount) .forEach(entry -> queryEmbeddingCache.remove(entry.getKey(), entry.getValue())); } + int evictedCount = sizeAfterExpiration - queryEmbeddingCache.size(); + metricsRecorder.recordCacheEvent("query_embedding", "evicted", evictedCount); queryEmbeddingCacheNextCleanupAt.set(nowMillis + QUERY_EMBEDDING_CACHE_CLEANUP_INTERVAL_MILLIS); log.debug( "few-shot query embedding cache maintained. sizeBefore={}, sizeAfter={}, maxSize={}", @@ -313,7 +385,7 @@ private List resolveDocumentEmbeddings( return embedDocuments(documents); } Instant now = Instant.now(); - documentEmbeddingCache.entrySet().removeIf(entry -> entry.getValue().expiresAt().isBefore(now)); + maintainDocumentEmbeddingCache(false); List result = new ArrayList<>(java.util.Collections.nCopies(candidates.size(), null)); List pending = new ArrayList<>(); @@ -323,22 +395,39 @@ private List resolveDocumentEmbeddings( 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 && cached.expiresAt().isBefore(now)) { + if (documentEmbeddingCache.remove(key, cached)) { + metricsRecorder.recordCacheEvent("document_embedding", "expired", 1L); + } + cached = null; + } if (cached != null) { - result.set(i, cached.embedding()); + DocumentEmbeddingCacheEntry accessed = cached.accessedAt(now); + documentEmbeddingCache.replace(key, cached, accessed); + result.set(i, accessed.embedding()); cacheHitCount++; + metricsRecorder.recordCacheEvent("document_embedding", "hit", 1L); continue; } + metricsRecorder.recordCacheEvent("document_embedding", "miss", 1L); CompletableFuture created = new CompletableFuture<>(); CompletableFuture existing = documentEmbeddingInFlight.putIfAbsent(key, created); boolean owner = existing == null; if (owner) { DocumentEmbeddingCacheEntry cachedAfterClaim = documentEmbeddingCache.get(key); + if (cachedAfterClaim != null && cachedAfterClaim.expiresAt().isBefore(Instant.now())) { + if (documentEmbeddingCache.remove(key, cachedAfterClaim)) { + metricsRecorder.recordCacheEvent("document_embedding", "expired", 1L); + } + cachedAfterClaim = null; + } if (cachedAfterClaim != null) { created.complete(cachedAfterClaim); documentEmbeddingInFlight.remove(key, created); result.set(i, cachedAfterClaim.embedding()); cacheHitCount++; + metricsRecorder.recordCacheEvent("document_embedding", "hit_after_claim", 1L); continue; } } @@ -369,6 +458,43 @@ private List resolveDocumentEmbeddings( return List.copyOf(result); } + private void maintainDocumentEmbeddingCache(boolean requestFollowUpIfBusy) { + int maxSize = Math.max(1, properties.getDocumentEmbeddingCacheMaxSize()); + if (!documentEmbeddingCacheCleanupInProgress.compareAndSet(false, true)) { + if (requestFollowUpIfBusy) { + documentEmbeddingCacheCleanupRequested.set(true); + } + return; + } + try { + do { + documentEmbeddingCacheCleanupRequested.set(false); + Instant cleanupTime = Instant.now(); + int sizeBefore = documentEmbeddingCache.size(); + documentEmbeddingCache.entrySet().removeIf(entry -> + entry.getValue().expiresAt().isBefore(cleanupTime) + && !documentEmbeddingInFlight.containsKey(entry.getKey())); + int sizeAfterExpiration = documentEmbeddingCache.size(); + metricsRecorder.recordCacheEvent("document_embedding", "expired", sizeBefore - sizeAfterExpiration); + if (sizeAfterExpiration > maxSize) { + int removalCount = sizeAfterExpiration - maxSize; + documentEmbeddingCache.entrySet().stream() + .filter(entry -> !documentEmbeddingInFlight.containsKey(entry.getKey())) + .sorted(Comparator.comparing(entry -> entry.getValue().lastAccessedAt())) + .limit(removalCount) + .forEach(entry -> documentEmbeddingCache.remove(entry.getKey(), entry.getValue())); + } + metricsRecorder.recordCacheEvent( + "document_embedding", "evicted", sizeAfterExpiration - documentEmbeddingCache.size()); + } while (documentEmbeddingCacheCleanupRequested.getAndSet(false)); + } finally { + documentEmbeddingCacheCleanupInProgress.set(false); + if (documentEmbeddingCacheCleanupRequested.getAndSet(false)) { + maintainDocumentEmbeddingCache(true); + } + } + } + private void initializeMissingDocumentEmbeddings(List pending) { List owned = pending.stream() .filter(PendingDocumentEmbedding::owner) @@ -398,6 +524,7 @@ private void initializeMissingDocumentEmbeddings(List throw e; } finally { owned.forEach(item -> documentEmbeddingInFlight.remove(item.key(), item.future())); + maintainDocumentEmbeddingCache(true); } } @@ -475,12 +602,69 @@ private double localScore(FewShotSearchQuery query, FewShotCase fewShotCase) { } private SelectionCacheEntry readSelectionCache(String key) { + return readSelectionCache(key, true); + } + + private SelectionCacheEntry readSelectionCache(String key, boolean recordEvent) { if (!properties.isCacheEnabled()) { return null; } Instant now = Instant.now(); - selectionCache.entrySet().removeIf(entry -> entry.getValue().expiresAt().isBefore(now)); - return selectionCache.get(key); + maintainSelectionCache(false); + SelectionCacheEntry cached = selectionCache.get(key); + if (cached == null) { + if (recordEvent) { + metricsRecorder.recordCacheEvent("selection", "miss", 1L); + } + return null; + } + if (cached.expiresAt().isBefore(now)) { + selectionCache.remove(key, cached); + if (recordEvent) { + metricsRecorder.recordCacheEvent("selection", "expired", 1L); + } + return null; + } + SelectionCacheEntry accessed = cached.accessedAt(now); + selectionCache.replace(key, cached, accessed); + if (recordEvent) { + metricsRecorder.recordCacheEvent("selection", "hit", 1L); + } + return accessed; + } + + private void maintainSelectionCache(boolean requestFollowUpIfBusy) { + int maxSize = Math.max(1, properties.getSelectionCacheMaxSize()); + if (!selectionCacheCleanupInProgress.compareAndSet(false, true)) { + if (requestFollowUpIfBusy) { + selectionCacheCleanupRequested.set(true); + } + return; + } + try { + do { + selectionCacheCleanupRequested.set(false); + Instant cleanupTime = Instant.now(); + int sizeBefore = selectionCache.size(); + selectionCache.entrySet().removeIf(entry -> entry.getValue().expiresAt().isBefore(cleanupTime)); + int sizeAfterExpiration = selectionCache.size(); + metricsRecorder.recordCacheEvent("selection", "expired", sizeBefore - sizeAfterExpiration); + if (sizeAfterExpiration > maxSize) { + int removalCount = sizeAfterExpiration - maxSize; + selectionCache.entrySet().stream() + .filter(entry -> !selectionInFlight.containsKey(entry.getKey())) + .sorted(Comparator.comparing(entry -> entry.getValue().lastAccessedAt())) + .limit(removalCount) + .forEach(entry -> selectionCache.remove(entry.getKey(), entry.getValue())); + } + metricsRecorder.recordCacheEvent("selection", "evicted", sizeAfterExpiration - selectionCache.size()); + } while (selectionCacheCleanupRequested.getAndSet(false)); + } finally { + selectionCacheCleanupInProgress.set(false); + if (selectionCacheCleanupRequested.getAndSet(false)) { + maintainSelectionCache(true); + } + } } private String selectionCacheKey(FewShotSearchQuery query, int topK, String datasetFingerprint) { @@ -603,18 +787,39 @@ private record PendingDocumentEmbedding( private record SelectionCacheEntry( List selectedCases, FewShotSelectionMode selectionMode, - Instant expiresAt + Instant expiresAt, + Instant lastAccessedAt ) { + private SelectionCacheEntry( + List selectedCases, + FewShotSelectionMode selectionMode, + Instant expiresAt + ) { + this(selectedCases, selectionMode, expiresAt, Instant.now()); + } + private SelectionCacheEntry { selectedCases = selectedCases == null ? List.of() : List.copyOf(selectedCases); } + + private SelectionCacheEntry accessedAt(Instant accessedAt) { + return new SelectionCacheEntry(selectedCases, selectionMode, expiresAt, accessedAt); + } } - private record DocumentEmbeddingCacheEntry(float[] embedding, Instant expiresAt) { + private record DocumentEmbeddingCacheEntry(float[] embedding, Instant expiresAt, Instant lastAccessedAt) { + private DocumentEmbeddingCacheEntry(float[] embedding, Instant expiresAt) { + this(embedding, expiresAt, Instant.now()); + } + private DocumentEmbeddingCacheEntry { embedding = embedding == null ? new float[0] : embedding.clone(); } + private DocumentEmbeddingCacheEntry accessedAt(Instant accessedAt) { + return new DocumentEmbeddingCacheEntry(embedding, expiresAt, accessedAt); + } + @Override public float[] embedding() { return embedding.clone(); @@ -654,4 +859,16 @@ private List embedDocuments(List documents) { metricsRecorder.recordCohereLogicalCalls(1L); return cohereEmbeddingClient.embedDocuments(documents); } + + int selectionCacheSize() { + return selectionCache.size(); + } + + int queryEmbeddingCacheSize() { + return queryEmbeddingCache.size(); + } + + int documentEmbeddingCacheSize() { + return documentEmbeddingCache.size(); + } } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/FewShotMetricsRecorder.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/FewShotMetricsRecorder.java index c6d48ab..0d8881d 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/FewShotMetricsRecorder.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/FewShotMetricsRecorder.java @@ -54,6 +54,16 @@ public void recordCohereFailure(String reason) { .increment(); } + public void recordCacheEvent(String cache, String outcome, long count) { + if (count <= 0) { + return; + } + Counter.builder("fewshot.cache.events") + .tags("cache", cache, "outcome", outcome) + .register(meterRegistry) + .increment(count); + } + private String normalizeReason(String reason) { if (reason == null || reason.isBlank()) { return "Unknown"; 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 6a1d61c..710f529 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,7 +16,10 @@ public class FewShotProperties { private boolean fallbackEnabled = true; private boolean cacheEnabled = true; private Duration cacheTtl = Duration.ofMinutes(30); + private int selectionCacheMaxSize = 1_000; private int queryEmbeddingCacheMaxSize = 1_000; + private int documentEmbeddingCacheMaxSize = 5_000; + private Duration selectionInFlightWaitTimeout = Duration.ofSeconds(20); private Duration queryEmbeddingInFlightWaitTimeout = Duration.ofSeconds(20); private Source source = new Source(); private Search search = new Search(); @@ -89,6 +92,22 @@ public int getQueryEmbeddingCacheMaxSize() { return queryEmbeddingCacheMaxSize; } + public int getSelectionCacheMaxSize() { + return selectionCacheMaxSize; + } + + public void setSelectionCacheMaxSize(int selectionCacheMaxSize) { + this.selectionCacheMaxSize = selectionCacheMaxSize; + } + + public int getDocumentEmbeddingCacheMaxSize() { + return documentEmbeddingCacheMaxSize; + } + + public void setDocumentEmbeddingCacheMaxSize(int documentEmbeddingCacheMaxSize) { + this.documentEmbeddingCacheMaxSize = documentEmbeddingCacheMaxSize; + } + public void setQueryEmbeddingCacheMaxSize(int queryEmbeddingCacheMaxSize) { this.queryEmbeddingCacheMaxSize = queryEmbeddingCacheMaxSize; } @@ -97,6 +116,14 @@ public Duration getQueryEmbeddingInFlightWaitTimeout() { return queryEmbeddingInFlightWaitTimeout; } + public Duration getSelectionInFlightWaitTimeout() { + return selectionInFlightWaitTimeout; + } + + public void setSelectionInFlightWaitTimeout(Duration selectionInFlightWaitTimeout) { + this.selectionInFlightWaitTimeout = selectionInFlightWaitTimeout; + } + public void setQueryEmbeddingInFlightWaitTimeout(Duration queryEmbeddingInFlightWaitTimeout) { this.queryEmbeddingInFlightWaitTimeout = queryEmbeddingInFlightWaitTimeout; } diff --git a/src/main/resources/application-analysis-eval.yaml b/src/main/resources/application-analysis-eval.yaml index 4feb046..968e20e 100644 --- a/src/main/resources/application-analysis-eval.yaml +++ b/src/main/resources/application-analysis-eval.yaml @@ -95,7 +95,10 @@ 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} + selection-cache-max-size: ${ANALYSIS_FEW_SHOT_SELECTION_CACHE_MAX_SIZE:1000} query-embedding-cache-max-size: ${ANALYSIS_FEW_SHOT_QUERY_EMBEDDING_CACHE_MAX_SIZE:1000} + document-embedding-cache-max-size: ${ANALYSIS_FEW_SHOT_DOCUMENT_EMBEDDING_CACHE_MAX_SIZE:5000} + selection-in-flight-wait-timeout: ${ANALYSIS_FEW_SHOT_SELECTION_IN_FLIGHT_WAIT_TIMEOUT:20s} 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} diff --git a/src/main/resources/application-dev.yaml b/src/main/resources/application-dev.yaml index c76010a..01eeb8f 100644 --- a/src/main/resources/application-dev.yaml +++ b/src/main/resources/application-dev.yaml @@ -169,7 +169,10 @@ 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} + selection-cache-max-size: ${ANALYSIS_FEW_SHOT_SELECTION_CACHE_MAX_SIZE:1000} query-embedding-cache-max-size: ${ANALYSIS_FEW_SHOT_QUERY_EMBEDDING_CACHE_MAX_SIZE:1000} + document-embedding-cache-max-size: ${ANALYSIS_FEW_SHOT_DOCUMENT_EMBEDDING_CACHE_MAX_SIZE:5000} + selection-in-flight-wait-timeout: ${ANALYSIS_FEW_SHOT_SELECTION_IN_FLIGHT_WAIT_TIMEOUT:20s} 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} diff --git a/src/main/resources/application-prod.yaml b/src/main/resources/application-prod.yaml index ebc6da6..15609e4 100644 --- a/src/main/resources/application-prod.yaml +++ b/src/main/resources/application-prod.yaml @@ -171,7 +171,10 @@ 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} + selection-cache-max-size: ${ANALYSIS_FEW_SHOT_SELECTION_CACHE_MAX_SIZE:1000} query-embedding-cache-max-size: ${ANALYSIS_FEW_SHOT_QUERY_EMBEDDING_CACHE_MAX_SIZE:1000} + document-embedding-cache-max-size: ${ANALYSIS_FEW_SHOT_DOCUMENT_EMBEDDING_CACHE_MAX_SIZE:5000} + selection-in-flight-wait-timeout: ${ANALYSIS_FEW_SHOT_SELECTION_IN_FLIGHT_WAIT_TIMEOUT:20s} 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} 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 2411df2..7829cbc 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 @@ -6,13 +6,16 @@ import org.springframework.test.util.ReflectionTestUtils; import java.time.Duration; +import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; 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.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; @@ -282,6 +285,89 @@ void recoversWaitingRequestsAfterSharedQueryEmbeddingFailure() throws Exception verify(cohereEmbeddingClient, times(2)).embedQuery(any()); } + @Test + @DisplayName("동일한 selection 키의 동시 요청은 하나의 외부 호출 결과를 공유한다") + void reusesInFlightSelectionForSameKey() throws Exception { + properties.setDynamicSelectionEnabled(true); + when(caseStore.loadActiveCases()).thenReturn(List.of(caseItem("FS-1", "Spring Boot API 개발", 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("selection 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", "same selection 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, 1) + ); + assertThatThrownBy(() -> waiter.get(100, TimeUnit.MILLISECONDS)) + .isInstanceOf(java.util.concurrent.TimeoutException.class); + releaseEmbedding.countDown(); + + assertThat(owner.get(2, TimeUnit.SECONDS)).hasSize(1); + assertThat(waiter.get(2, TimeUnit.SECONDS)).hasSize(1); + verify(cohereEmbeddingClient, times(1)).embedQuery(any()); + verify(cohereEmbeddingClient, times(1)).embedDocuments(any()); + Map selectionInFlight = (Map) ReflectionTestUtils.getField(service, "selectionInFlight"); + assertThat(selectionInFlight).isEmpty(); + } finally { + releaseEmbedding.countDown(); + executor.shutdownNow(); + } + } + + @Test + @DisplayName("동일 selection 대기는 설정된 제한 시간을 넘으면 종료된다") + void timesOutWaitingForInFlightSelection() throws Exception { + properties.setDynamicSelectionEnabled(true); + properties.setSelectionInFlightWaitTimeout(Duration.ofMillis(50)); + when(caseStore.loadActiveCases()).thenReturn(List.of(caseItem("FS-1", "Spring Boot API 개발", 0))); + CountDownLatch embeddingStarted = new CountDownLatch(1); + CountDownLatch releaseEmbedding = new CountDownLatch(1); + when(cohereEmbeddingClient.embedQuery(any())).thenAnswer(invocation -> { + embeddingStarted.countDown(); + releaseEmbedding.await(3, TimeUnit.SECONDS); + return new float[]{1, 0}; + }); + when(cohereEmbeddingClient.embedDocuments(any())).thenReturn(List.of(new float[]{1, 0})); + FewShotSearchQuery query = query("EV-01", "selection 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, 1) + ); + + ExecutionException exception = org.assertj.core.api.Assertions.catchThrowableOfType( + () -> waiter.get(1, TimeUnit.SECONDS), + ExecutionException.class + ); + assertThat(exception.getCause()) + .isInstanceOf(IllegalStateException.class) + .hasMessage("공유된 Few-shot selection 대기 시간이 초과되었습니다."); + releaseEmbedding.countDown(); + assertThat(owner.get(2, TimeUnit.SECONDS)).hasSize(1); + } finally { + releaseEmbedding.countDown(); + executor.shutdownNow(); + } + } + @Test @DisplayName("공유 query embedding 대기가 제한 시간을 넘으면 로컬 fallback한다") void fallsBackLocallyWhenSharedQueryEmbeddingWaitTimesOut() throws Exception { @@ -402,6 +488,106 @@ void boundsQueryEmbeddingCacheSize() { assertThat(queryEmbeddingCache).hasSizeLessThanOrEqualTo(3); } + @Test + @DisplayName("selection과 document embedding 캐시는 각각 설정된 최대 크기를 넘지 않는다") + void boundsSelectionAndDocumentEmbeddingCaches() { + properties.setDynamicSelectionEnabled(true); + properties.setSelectionCacheMaxSize(2); + properties.setQueryEmbeddingCacheMaxSize(2); + properties.setDocumentEmbeddingCacheMaxSize(2); + AtomicInteger candidateSequence = new AtomicInteger(); + when(caseStore.loadActiveCases()).thenAnswer(invocation -> { + int sequence = candidateSequence.incrementAndGet(); + return List.of(caseItem("FS-" + sequence, "Spring Boot API 개발 " + sequence, 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, "bounded request " + i), 1); + } + + assertThat(service.selectionCacheSize()).isLessThanOrEqualTo(2); + assertThat(service.queryEmbeddingCacheSize()).isLessThanOrEqualTo(2); + assertThat(service.documentEmbeddingCacheSize()).isLessThanOrEqualTo(2); + } + + @Test + @DisplayName("동시 삽입이 끝난 뒤 selection과 document 캐시는 설정 상한 이내로 정리된다") + void boundsCachesAfterConcurrentInsertions() throws Exception { + int requestCount = 8; + properties.setDynamicSelectionEnabled(true); + properties.setSelectionCacheMaxSize(2); + properties.setDocumentEmbeddingCacheMaxSize(2); + properties.setQueryEmbeddingCacheMaxSize(20); + AtomicInteger candidateSequence = new AtomicInteger(); + when(caseStore.loadActiveCases()).thenAnswer(invocation -> { + int sequence = candidateSequence.incrementAndGet(); + return List.of(caseItem("FS-CONCURRENT-" + sequence, "API 개발 " + sequence, 0)); + }); + when(cohereEmbeddingClient.embedQuery(any())).thenReturn(new float[]{1, 0}); + CountDownLatch documentCallsStarted = new CountDownLatch(requestCount); + CountDownLatch releaseDocuments = new CountDownLatch(1); + when(cohereEmbeddingClient.embedDocuments(any())).thenAnswer(invocation -> { + documentCallsStarted.countDown(); + if (!releaseDocuments.await(3, TimeUnit.SECONDS)) { + throw new IllegalStateException("concurrent document embeddings did not start in time"); + } + List documents = invocation.getArgument(0); + return documents.stream().map(ignored -> new float[]{1, 0}).toList(); + }); + + ExecutorService executor = Executors.newFixedThreadPool(requestCount); + List>> futures = new ArrayList<>(); + try { + for (int i = 0; i < requestCount; i++) { + int request = i; + futures.add(executor.submit(() -> service.searchRelevantFewShots( + query("EV-CONCURRENT-" + request, "concurrent request " + request), 1 + ))); + } + assertThat(documentCallsStarted.await(2, TimeUnit.SECONDS)).isTrue(); + releaseDocuments.countDown(); + for (Future> future : futures) { + assertThat(future.get(3, TimeUnit.SECONDS)).hasSize(1); + } + } finally { + releaseDocuments.countDown(); + executor.shutdownNow(); + } + + assertThat(service.selectionCacheSize()).isLessThanOrEqualTo(2); + assertThat(service.documentEmbeddingCacheSize()).isLessThanOrEqualTo(2); + } + + @Test + @DisplayName("조회가 진행 중인 캐시 정리와 겹쳐도 후속 정리를 예약하지 않는다") + void doesNotRequestFollowUpCleanupForReads() { + AtomicBoolean selectionInProgress = cleanupFlag("selectionCacheCleanupInProgress"); + AtomicBoolean selectionRequested = cleanupFlag("selectionCacheCleanupRequested"); + AtomicBoolean documentInProgress = cleanupFlag("documentEmbeddingCacheCleanupInProgress"); + AtomicBoolean documentRequested = cleanupFlag("documentEmbeddingCacheCleanupRequested"); + selectionInProgress.set(true); + documentInProgress.set(true); + + try { + ReflectionTestUtils.invokeMethod(service, "readSelectionCache", "missing-key"); + List embeddings = ReflectionTestUtils.invokeMethod( + service, + "resolveDocumentEmbeddings", + List.of(), + List.of() + ); + + assertThat(embeddings).isEmpty(); + assertThat(selectionRequested).isFalse(); + assertThat(documentRequested).isFalse(); + } finally { + selectionInProgress.set(false); + documentInProgress.set(false); + } + } + @Test @DisplayName("만료된 query embedding은 동일 질의 재요청에 사용하지 않는다") void doesNotReuseExpiredQueryEmbedding() { @@ -416,6 +602,7 @@ void doesNotReuseExpiredQueryEmbedding() { service.searchRelevantFewShots(query, 2); verify(cohereEmbeddingClient, times(2)).embedQuery(any()); + verify(cohereEmbeddingClient, times(2)).embedDocuments(any()); } @Test @@ -549,7 +736,11 @@ void evictsExpiredSelectionEntriesGlobally() { service.searchRelevantFewShots(query("EV-02", "두 번째 요청"), 1); Map selectionCache = (Map) ReflectionTestUtils.getField(service, "selectionCache"); - assertThat(selectionCache).hasSize(1); + assertThat(selectionCache).isEmpty(); + } + + private AtomicBoolean cleanupFlag(String fieldName) { + return (AtomicBoolean) ReflectionTestUtils.getField(service, fieldName); } private static FewShotSearchQuery query(String caseId) { diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/FewShotMetricsRecorderTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/FewShotMetricsRecorderTest.java index 19ba344..21a9e54 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/FewShotMetricsRecorderTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/ai/fewshot/FewShotMetricsRecorderTest.java @@ -15,6 +15,7 @@ void recordsSelectionCallsAndBoundedFailureReason() { recorder.recordSelection(FewShotSelectionMode.EMBEDDING, false, 3, 42); recorder.recordCohereLogicalCalls(2); recorder.recordCohereFailure("UnexpectedVendorException"); + recorder.recordCacheEvent("selection", "evicted", 2); assertThat(registry.get("fewshot.selection.count") .tags("mode", "EMBEDDING", "cache_hit", "false").counter().count()).isEqualTo(1.0); @@ -25,5 +26,7 @@ void recordsSelectionCallsAndBoundedFailureReason() { assertThat(registry.get("fewshot.cohere.logical.calls").counter().count()).isEqualTo(2.0); assertThat(registry.get("fewshot.cohere.failure.count") .tag("reason", "Other").counter().count()).isEqualTo(1.0); + assertThat(registry.get("fewshot.cache.events") + .tags("cache", "selection", "outcome", "evicted").counter().count()).isEqualTo(2.0); } }