Mapin

분석 파이프라인은 URL 하나당 외부 API를 여러 번 호출합니다. 처음에는 고정 크기 스레드 풀을 사용했는데, 동시 요청이 늘어나면 풀이 고갈되어 대기가 발생했습니다.

// 기존 — 고정 풀, 동시 요청 많으면 대기 발생
@Bean
ExecutorService pipelineExecutor() {
    return Executors.newFixedThreadPool(20);
}

// Java 21 Virtual Threads — URL당 가상 스레드 할당, 한도 없음
@Bean
ExecutorService pipelineExecutor() {
    return Executors.newVirtualThreadPerTaskExecutor();
}

Virtual Threads는 OS 스레드에 1:1 매핑되지 않아 수천 개를 동시에 띄워도 메모리 부담이 작습니다. I/O 대기가 긴 GPT 호출·외부 검색 API에 딱 맞는 구조였고, 스레드 풀 크기를 튜닝하는 고민 자체가 사라졌습니다.

단, synchronized 블록 안에서 I/O를 하면 Virtual Thread가 OS 스레드를 고정(pinning)해 이점이 사라집니다. 이 프로젝트에서는 synchronized를 피하고 ReentrantLock을 쓰도록 주의했습니다.

동일한 URL이 재요청될 때마다 GPT를 호출하면 비용이 선형으로 늘어납니다. 어디서 스킵할 수 있는지 분석해보니 세 단계로 나뉘었습니다.

// L1: opposition_json 존재 → 분석 결과 전부 캐시됨
//     GPT 호출 0회, 외부 검색 0회
if (content.getOppositionJson() != null) {
    return deserialize(content.getOppositionJson());
}

// L2: is_analyzed = true → 소스 분석은 완료, 추천만 재생성
//     GPT 호출 0회, 외부 검색만 수행
if (content.isAnalyzed()) {
    return searchOpposition(content.getKeywords());
}

// L3: content_keywords 풀 → 유사 콘텐츠 키워드 재활용
//     GPT 호출 1회 (스코어링만), 외부 검색 수행
List<Content> pool = keywordRepository.findSimilar(keywords);
if (!pool.isEmpty()) {
    return scoreAndFilter(pool);
}

L1에서 히트하면 파이프라인 전 단계를 스킵합니다. 캐시 키를 URL 해시가 아니라 콘텐츠 ID로 잡은 덕분에 같은 영상을 다른 사람이 저장해도 캐시가 공유됩니다.

처음엔 단순히 "중복 호출을 막자"는 생각이었는데, 실제로 설계하다 보니 "어느 단계부터 캐시 가능한가"를 먼저 따지게 되었고 파이프라인 구조 자체가 더 명확해졌습니다.

반대 관점 후보를 검색하면 10~20개가 나옵니다. 처음 설계는 후보마다 GPT를 1회씩 호출해 관점 점수를 매기는 방식이었는데, 20개면 20번의 API 호출이 발생합니다.

// 개별 호출 — 후보 N개면 GPT N번
for (Content candidate : candidates) {
    double score = gpt.scoreViewpoint(candidate); // N회
}

// 배치 스코어링 — 후보 전체를 JSON 배열로 묶어 1회 호출
String prompt = buildBatchPrompt(candidates); // 전체 후보 직렬화
List<ScoredContent> results = gpt.scoreBatch(prompt); // 1회

GPT에게 JSON 배열로 후보 전체를 넘기고 각 항목의 점수를 배열로 돌려받도록 프롬프트를 설계했습니다. 입력 토큰은 늘어나지만 API 호출 횟수가 1회로 고정되어 레이턴시가 크게 줄었습니다.

응답 파싱 실패 시 개별 호출로 폴백하는 로직도 함께 넣었는데, 실제로 GPT가 간헐적으로 JSON 형식을 지키지 않을 때 이 폴백이 동작하는 걸 확인했습니다.

반대관점 후보를 수집할 때 YouTube API와 Naver 검색 API를 동시에 호출합니다. 처음엔 순차 호출로 구현했는데, 쿼리 2~3개 × 2개 API면 최대 6번의 직렬 호출이 되어 레이턴시가 쌓였습니다.

// 순차 호출 — 쿼리 N개 × API 2개 = 직렬 호출
for (String query : queries) {
    results.addAll(youtubeClient.search(query));
    results.addAll(naverClient.search(query));
}

// 병렬 호출 — 전체를 동시 실행, 각자 실패해도 빈 리스트로 계속
List<CompletableFuture<List<Content>>> futures = queries.stream()
    .flatMap(q -> Stream.of(
        youtubeClient.searchAsync(q)
            .exceptionally(e -> List.of()),  // YouTube 실패 → 빈 리스트
        naverClient.searchAsync(q)
            .exceptionally(e -> List.of())   // Naver 실패 → 빈 리스트
    ))
    .toList();

List<Content> candidates = CompletableFuture
    .allOf(futures.toArray(new CompletableFuture[0]))
    .thenApply(v -> futures.stream()
        .flatMap(f -> f.join().stream())
        .collect(Collectors.toList()))
    .get(30, TimeUnit.SECONDS);

핵심은 exceptionally(() → List.of()) 처리입니다. YouTube가 실패해도 Naver 결과만으로 진행하고, 반대도 마찬가지입니다. 둘 다 실패하면 빈 리스트가 되어 L3 캐시 폴백으로 넘어갑니다.

allOf로 전체를 묶은 뒤 30초 타임아웃을 걸어 파이프라인이 무한정 대기하지 않도록 보호했습니다. 개별 API 타임아웃은 각 클라이언트에서 따로 설정하고, 30초는 파이프라인 전체를 끊는 마지막 안전망입니다.

키워드 텍스트 매칭만으로는 주제는 비슷하지만 관점이 다른 콘텐츠를 찾기 어렵습니다.text-embedding-3-small로 각 키워드를 1536차원 벡터로 변환하고, 최근 분석한 콘텐츠들의 키워드 벡터를 산술 평균해 소스의 주제 벡터를 구성했습니다.

// 1단계: pgvector 코사인 유사도로 주제 유사 콘텐츠 50개 확보
WITH topic_pool AS (
    SELECT c.id, c.viewpoint_score,
           1 - (k.embedding <=> ?::vector) AS topic_sim
    FROM content_keywords ck
    JOIN keywords k ON ck.keyword_id = k.id
    JOIN contents c ON ck.content_id = c.content_id
    WHERE k.embedding IS NOT NULL
    GROUP BY c.id, c.viewpoint_score
    ORDER BY topic_sim DESC
    LIMIT 50  -- poolSize
)
-- 2단계: 풀 안에서 opposition 우선 정렬
SELECT * FROM topic_pool
ORDER BY
    (ABS(viewpoint_score - ?) >= 0.4) DESC,  -- opposition 우선
    (topic_sim * 0.4 + ABS(viewpoint_score - ?) * 0.6) DESC

DB에서 주제 유사도로 50개를 뽑은 뒤, 메모리에서viewpointScore 차이 ≥ 0.4 조건으로 opposition을 먼저 채우고 부족하면 similar 콘텐츠로 보충합니다.

pgvector를 처음 써봤는데, JPA 엔티티에서 지원이 제한적이라 벡터 저장·조회는JdbcTemplate으로 raw SQL을 직접 작성했습니다. 벡터를 float[]로 직렬화해 ?::vector로 캐스팅하는 방식입니다. 임베딩 기능은 설정값으로 활성화·비활성화할 수 있어, 비용이 부담스러울 때는 키워드 매칭만으로 동작하도록 폴백을 열어뒀습니다.