From 8e976626f482fce1870c0cf7a9defb16a59d360f Mon Sep 17 00:00:00 2001 From: Woohyeok Choi Date: Wed, 16 Sep 2026 20:51:58 +0900 Subject: [PATCH 1/3] =?UTF-8?q?[Feat]=20Few-shot=20=EB=A9=94=EB=AA=A8?= =?UTF-8?q?=EB=A6=AC=20=EC=BA=90=EC=8B=9C=20=EC=B5=9C=EB=8C=80=20=ED=81=AC?= =?UTF-8?q?=EA=B8=B0=20=EC=A0=9C=ED=95=9C=20=EC=A0=81=EC=9A=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - selection cache 최대 크기 및 LRU 제거 정책 추가 - document embedding cache 최대 크기 및 LRU 제거 정책 추가 - 기존 query embedding cache와 TTL 정책 정렬 - 동일 selection 키의 동시 요청 결과 공유 - in-flight 요청을 캐시 제거 대상에서 보호 - 캐시별 hit, miss, expired, evicted 메트릭 추가 - dev, prod, evaluation 환경별 캐시 크기 설정 추가 - 캐시 상한과 동시 외부 호출 방지 회귀 테스트 추가 - 캐시 운영 정책 및 Grafana 조회 쿼리 문서화 --- docs/fewshot-cache-policy.md | 44 +++ docs/fewshot-operations.md | 1 + .../fewshot/DefaultFewShotSearchService.java | 250 +++++++++++++++--- .../ai/fewshot/FewShotMetricsRecorder.java | 10 + .../service/ai/fewshot/FewShotProperties.java | 18 ++ .../resources/application-analysis-eval.yaml | 2 + src/main/resources/application-dev.yaml | 2 + src/main/resources/application-prod.yaml | 2 + .../DefaultFewShotSearchServiceTest.java | 68 ++++- .../fewshot/FewShotMetricsRecorderTest.java | 3 + 10 files changed, 357 insertions(+), 43 deletions(-) create mode 100644 docs/fewshot-cache-policy.md diff --git a/docs/fewshot-cache-policy.md b/docs/fewshot-cache-policy.md new file mode 100644 index 0000000..d097fcb --- /dev/null +++ b/docs/fewshot-cache-policy.md @@ -0,0 +1,44 @@ +# 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 +``` + +모든 최대 크기는 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..49d909d 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·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..840c952 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 @@ -40,6 +40,8 @@ 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 Map queryEmbeddingCache = new ConcurrentHashMap<>(); private final Map> queryEmbeddingInFlight = new ConcurrentHashMap<>(); @@ -48,6 +50,7 @@ public class DefaultFewShotSearchService implements FewShotSearchService { private final Map documentEmbeddingCache = new ConcurrentHashMap<>(); private final Map> documentEmbeddingInFlight = new ConcurrentHashMap<>(); + private final AtomicBoolean documentEmbeddingCacheCleanupInProgress = new AtomicBoolean(); @Autowired public DefaultFewShotSearchService( @@ -77,46 +80,80 @@ 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(existing.join(), 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(Instant.now()); + } + 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(Instant.now()); + } + } + } + + 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 void recordMetrics( @@ -213,7 +250,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 +277,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 +335,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 +344,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 +366,7 @@ private List resolveDocumentEmbeddings( return embedDocuments(documents); } Instant now = Instant.now(); - documentEmbeddingCache.entrySet().removeIf(entry -> entry.getValue().expiresAt().isBefore(now)); + maintainDocumentEmbeddingCache(now); List result = new ArrayList<>(java.util.Collections.nCopies(candidates.size(), null)); List pending = new ArrayList<>(); @@ -324,10 +377,14 @@ private List resolveDocumentEmbeddings( String key = documentEmbeddingCacheKey(candidates.get(i), documents.get(i)); DocumentEmbeddingCacheEntry cached = documentEmbeddingCache.get(key); 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); @@ -339,6 +396,7 @@ private List resolveDocumentEmbeddings( documentEmbeddingInFlight.remove(key, created); result.set(i, cachedAfterClaim.embedding()); cacheHitCount++; + metricsRecorder.recordCacheEvent("document_embedding", "hit_after_claim", 1L); continue; } } @@ -369,6 +427,33 @@ private List resolveDocumentEmbeddings( return List.copyOf(result); } + private void maintainDocumentEmbeddingCache(Instant now) { + int maxSize = Math.max(1, properties.getDocumentEmbeddingCacheMaxSize()); + if (!documentEmbeddingCacheCleanupInProgress.compareAndSet(false, true)) { + return; + } + try { + int sizeBefore = documentEmbeddingCache.size(); + documentEmbeddingCache.entrySet().removeIf(entry -> + entry.getValue().expiresAt().isBefore(now) + && !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()); + } finally { + documentEmbeddingCacheCleanupInProgress.set(false); + } + } + private void initializeMissingDocumentEmbeddings(List pending) { List owned = pending.stream() .filter(PendingDocumentEmbedding::owner) @@ -398,6 +483,7 @@ private void initializeMissingDocumentEmbeddings(List throw e; } finally { owned.forEach(item -> documentEmbeddingInFlight.remove(item.key(), item.future())); + maintainDocumentEmbeddingCache(Instant.now()); } } @@ -475,12 +561,59 @@ 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(now); + 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(Instant now) { + int maxSize = Math.max(1, properties.getSelectionCacheMaxSize()); + if (!selectionCacheCleanupInProgress.compareAndSet(false, true)) { + return; + } + try { + int sizeBefore = selectionCache.size(); + selectionCache.entrySet().removeIf(entry -> entry.getValue().expiresAt().isBefore(now)); + 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()); + } finally { + selectionCacheCleanupInProgress.set(false); + } } private String selectionCacheKey(FewShotSearchQuery query, int topK, String datasetFingerprint) { @@ -603,18 +736,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 +808,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..844b0d5 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,9 @@ 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 queryEmbeddingInFlightWaitTimeout = Duration.ofSeconds(20); private Source source = new Source(); private Search search = new Search(); @@ -89,6 +91,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; } diff --git a/src/main/resources/application-analysis-eval.yaml b/src/main/resources/application-analysis-eval.yaml index 4feb046..2e62426 100644 --- a/src/main/resources/application-analysis-eval.yaml +++ b/src/main/resources/application-analysis-eval.yaml @@ -95,7 +95,9 @@ 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} 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..de52c86 100644 --- a/src/main/resources/application-dev.yaml +++ b/src/main/resources/application-dev.yaml @@ -169,7 +169,9 @@ 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} 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..44a20b1 100644 --- a/src/main/resources/application-prod.yaml +++ b/src/main/resources/application-prod.yaml @@ -171,7 +171,9 @@ 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} 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..d42599f 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 @@ -282,6 +282,48 @@ 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("공유 query embedding 대기가 제한 시간을 넘으면 로컬 fallback한다") void fallsBackLocallyWhenSharedQueryEmbeddingWaitTimesOut() throws Exception { @@ -402,6 +444,30 @@ 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("만료된 query embedding은 동일 질의 재요청에 사용하지 않는다") void doesNotReuseExpiredQueryEmbedding() { @@ -549,7 +615,7 @@ void evictsExpiredSelectionEntriesGlobally() { service.searchRelevantFewShots(query("EV-02", "두 번째 요청"), 1); Map selectionCache = (Map) ReflectionTestUtils.getField(service, "selectionCache"); - assertThat(selectionCache).hasSize(1); + assertThat(selectionCache).isEmpty(); } 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); } } From fc7b815846a057adc4143cf95bcd4abc2df4eef8 Mon Sep 17 00:00:00 2001 From: Woohyeok Choi Date: Wed, 16 Sep 2026 23:50:37 +0900 Subject: [PATCH 2/3] =?UTF-8?q?[Fix]=20Few-shot=20=EC=BA=90=EC=8B=9C=20?= =?UTF-8?q?=EB=8F=99=EC=8B=9C=EC=84=B1=20=EB=B0=8F=20=EB=A7=8C=EB=A3=8C=20?= =?UTF-8?q?=EC=B2=98=EB=A6=AC=20=EB=B3=B4=EA=B0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 만료된 document embedding을 cache hit 처리 전에 제거 - selection in-flight 공유 대기에 configurable timeout 적용 - selection 대기의 timeout, interruption, owner failure 예외 처리 - selection 및 document 캐시 정리 중 발생한 추가 정리 요청 보존 - 동시 삽입 완료 후 설정된 캐시 최대 크기를 재적용 - 캐시 동시 삽입 및 selection timeout 회귀 테스트 추가 - hit_after_claim 메트릭 outcome 문서화 - 환경별 selection in-flight wait timeout 설정 추가 --- docs/fewshot-cache-policy.md | 1 + docs/fewshot-operations.md | 2 +- .../fewshot/DefaultFewShotSearchService.java | 107 +++++++++++++----- .../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 | 92 +++++++++++++++ 8 files changed, 183 insertions(+), 31 deletions(-) diff --git a/docs/fewshot-cache-policy.md b/docs/fewshot-cache-policy.md index d097fcb..a02716d 100644 --- a/docs/fewshot-cache-policy.md +++ b/docs/fewshot-cache-policy.md @@ -15,6 +15,7 @@ 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 캐시는 접근할 때마다 diff --git a/docs/fewshot-operations.md b/docs/fewshot-operations.md index 49d909d..7f6bf69 100644 --- a/docs/fewshot-operations.md +++ b/docs/fewshot-operations.md @@ -18,7 +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·miss·expired·evicted 횟수 | +| `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 840c952..7e50415 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}가-힣]+"); @@ -42,6 +43,7 @@ public class DefaultFewShotSearchService implements FewShotSearchService { 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<>(); @@ -51,6 +53,7 @@ public class DefaultFewShotSearchService implements FewShotSearchService { private final Map> documentEmbeddingInFlight = new ConcurrentHashMap<>(); private final AtomicBoolean documentEmbeddingCacheCleanupInProgress = new AtomicBoolean(); + private final AtomicBoolean documentEmbeddingCacheCleanupRequested = new AtomicBoolean(); @Autowired public DefaultFewShotSearchService( @@ -86,7 +89,7 @@ public List searchRelevantFewShots(FewShotSearchQuery query CompletableFuture created = new CompletableFuture<>(); CompletableFuture existing = selectionInFlight.putIfAbsent(cacheKey, created); if (existing != null) { - return cachedSelection(existing.join(), startedAt, "in-flight reuse"); + return cachedSelection(awaitSelection(existing), startedAt, "in-flight reuse"); } SelectionCacheEntry cachedAfterClaim = readSelectionCache(cacheKey, false); if (cachedAfterClaim != null) { @@ -156,6 +159,22 @@ private List cachedSelection( 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( FewShotSelectionMode selectionMode, boolean cacheHit, @@ -376,6 +395,12 @@ 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) { DocumentEmbeddingCacheEntry accessed = cached.accessedAt(now); documentEmbeddingCache.replace(key, cached, accessed); @@ -391,6 +416,12 @@ private List resolveDocumentEmbeddings( 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); @@ -430,27 +461,35 @@ private List resolveDocumentEmbeddings( private void maintainDocumentEmbeddingCache(Instant now) { int maxSize = Math.max(1, properties.getDocumentEmbeddingCacheMaxSize()); if (!documentEmbeddingCacheCleanupInProgress.compareAndSet(false, true)) { + documentEmbeddingCacheCleanupRequested.set(true); return; } try { - int sizeBefore = documentEmbeddingCache.size(); - documentEmbeddingCache.entrySet().removeIf(entry -> - entry.getValue().expiresAt().isBefore(now) - && !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()); + 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(Instant.now()); + } } } @@ -595,24 +634,32 @@ private SelectionCacheEntry readSelectionCache(String key, boolean recordEvent) private void maintainSelectionCache(Instant now) { int maxSize = Math.max(1, properties.getSelectionCacheMaxSize()); if (!selectionCacheCleanupInProgress.compareAndSet(false, true)) { + selectionCacheCleanupRequested.set(true); return; } try { - int sizeBefore = selectionCache.size(); - selectionCache.entrySet().removeIf(entry -> entry.getValue().expiresAt().isBefore(now)); - 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()); + 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(Instant.now()); + } } } 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 844b0d5..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 @@ -19,6 +19,7 @@ public class FewShotProperties { 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(); @@ -115,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 2e62426..968e20e 100644 --- a/src/main/resources/application-analysis-eval.yaml +++ b/src/main/resources/application-analysis-eval.yaml @@ -98,6 +98,7 @@ analysis: 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 de52c86..01eeb8f 100644 --- a/src/main/resources/application-dev.yaml +++ b/src/main/resources/application-dev.yaml @@ -172,6 +172,7 @@ analysis: 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 44a20b1..15609e4 100644 --- a/src/main/resources/application-prod.yaml +++ b/src/main/resources/application-prod.yaml @@ -174,6 +174,7 @@ analysis: 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 d42599f..430cb6b 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,9 +6,11 @@ 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; @@ -324,6 +326,47 @@ void reusesInFlightSelectionForSameKey() throws Exception { } } + @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 { @@ -468,6 +511,54 @@ void boundsSelectionAndDocumentEmbeddingCaches() { 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("만료된 query embedding은 동일 질의 재요청에 사용하지 않는다") void doesNotReuseExpiredQueryEmbedding() { @@ -482,6 +573,7 @@ void doesNotReuseExpiredQueryEmbedding() { service.searchRelevantFewShots(query, 2); verify(cohereEmbeddingClient, times(2)).embedQuery(any()); + verify(cohereEmbeddingClient, times(2)).embedDocuments(any()); } @Test From 121cc7c636d77993bc235a419f9bfbdbbeda9309 Mon Sep 17 00:00:00 2001 From: Woohyeok Choi Date: Thu, 17 Sep 2026 00:02:54 +0900 Subject: [PATCH 3/3] =?UTF-8?q?[Fix]=20Few-shot=20=EC=BA=90=EC=8B=9C=20?= =?UTF-8?q?=EC=A0=95=EB=A6=AC=20=EC=9E=AC=EC=8B=A4=ED=96=89=20=EC=A1=B0?= =?UTF-8?q?=EA=B1=B4=20=EA=B0=9C=EC=84=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - selection 캐시 조회 중 불필요한 후속 정리 예약 방지 - document embedding 캐시 조회 중 불필요한 후속 정리 예약 방지 - 캐시 삽입 및 in-flight 제거 경로에서만 dirty 상태 기록 - 동시 삽입 후 캐시 최대 크기 보장 동작 유지 - 조회와 캐시 정리 경합에 대한 회귀 테스트 추가 --- .../fewshot/DefaultFewShotSearchService.java | 26 ++++++++------- .../DefaultFewShotSearchServiceTest.java | 33 +++++++++++++++++++ 2 files changed, 48 insertions(+), 11 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 7e50415..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 @@ -117,7 +117,7 @@ public List searchRelevantFewShots(FewShotSearchQuery query SelectionCacheEntry entry = new SelectionCacheEntry(selected, selectionMode, expiresAt()); if (properties.isCacheEnabled()) { selectionCache.put(cacheKey, entry); - maintainSelectionCache(Instant.now()); + maintainSelectionCache(true); } created.complete(entry); recordMetrics(selectionMode, false, selected.size(), startedAt); @@ -138,7 +138,7 @@ public List searchRelevantFewShots(FewShotSearchQuery query } finally { selectionInFlight.remove(cacheKey, created); if (properties.isCacheEnabled()) { - maintainSelectionCache(Instant.now()); + maintainSelectionCache(true); } } } @@ -385,7 +385,7 @@ private List resolveDocumentEmbeddings( return embedDocuments(documents); } Instant now = Instant.now(); - maintainDocumentEmbeddingCache(now); + maintainDocumentEmbeddingCache(false); List result = new ArrayList<>(java.util.Collections.nCopies(candidates.size(), null)); List pending = new ArrayList<>(); @@ -458,10 +458,12 @@ private List resolveDocumentEmbeddings( return List.copyOf(result); } - private void maintainDocumentEmbeddingCache(Instant now) { + private void maintainDocumentEmbeddingCache(boolean requestFollowUpIfBusy) { int maxSize = Math.max(1, properties.getDocumentEmbeddingCacheMaxSize()); if (!documentEmbeddingCacheCleanupInProgress.compareAndSet(false, true)) { - documentEmbeddingCacheCleanupRequested.set(true); + if (requestFollowUpIfBusy) { + documentEmbeddingCacheCleanupRequested.set(true); + } return; } try { @@ -488,7 +490,7 @@ private void maintainDocumentEmbeddingCache(Instant now) { } finally { documentEmbeddingCacheCleanupInProgress.set(false); if (documentEmbeddingCacheCleanupRequested.getAndSet(false)) { - maintainDocumentEmbeddingCache(Instant.now()); + maintainDocumentEmbeddingCache(true); } } } @@ -522,7 +524,7 @@ private void initializeMissingDocumentEmbeddings(List throw e; } finally { owned.forEach(item -> documentEmbeddingInFlight.remove(item.key(), item.future())); - maintainDocumentEmbeddingCache(Instant.now()); + maintainDocumentEmbeddingCache(true); } } @@ -608,7 +610,7 @@ private SelectionCacheEntry readSelectionCache(String key, boolean recordEvent) return null; } Instant now = Instant.now(); - maintainSelectionCache(now); + maintainSelectionCache(false); SelectionCacheEntry cached = selectionCache.get(key); if (cached == null) { if (recordEvent) { @@ -631,10 +633,12 @@ private SelectionCacheEntry readSelectionCache(String key, boolean recordEvent) return accessed; } - private void maintainSelectionCache(Instant now) { + private void maintainSelectionCache(boolean requestFollowUpIfBusy) { int maxSize = Math.max(1, properties.getSelectionCacheMaxSize()); if (!selectionCacheCleanupInProgress.compareAndSet(false, true)) { - selectionCacheCleanupRequested.set(true); + if (requestFollowUpIfBusy) { + selectionCacheCleanupRequested.set(true); + } return; } try { @@ -658,7 +662,7 @@ private void maintainSelectionCache(Instant now) { } finally { selectionCacheCleanupInProgress.set(false); if (selectionCacheCleanupRequested.getAndSet(false)) { - maintainSelectionCache(Instant.now()); + maintainSelectionCache(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 430cb6b..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 @@ -15,6 +15,7 @@ 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; @@ -559,6 +560,34 @@ void boundsCachesAfterConcurrentInsertions() throws Exception { 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() { @@ -710,6 +739,10 @@ void evictsExpiredSelectionEntriesGlobally() { assertThat(selectionCache).isEmpty(); } + private AtomicBoolean cleanupFlag(String fieldName) { + return (AtomicBoolean) ReflectionTestUtils.getField(service, fieldName); + } + private static FewShotSearchQuery query(String caseId) { return query(caseId, "Spring Boot API를 개발했습니다."); }