3단계는 AI 서버를 api·worker·batch·model-runtime 네 배포 단위로 나눴다. 무엇을 따로 배포하고 따로 스케일할지까지 정했고, 그 단위 안에서 무슨 일이 어떤 순서로 도는지는 정하지 않았다. 4단계는 그 안쪽을 정한다.

상품 카탈로그를 채우는 배치도 여러 단계로 이어지는 오프라인 파이프라인이지만 여기서 다루지 않는다. 크롤링과 속성 추출을 어떤 순서로 돌릴지는 무엇을 수집할 것인가가 정해진 다음의 문제이고, 그 수집 전략은 추후에 확정한다. 3단계가 batch라는 배포 단위로 자리만 잡아뒀다.

4단계 과제가 상정하는 LLM 서비스의 질의 처리 흐름은 사용자 질의를 받아 검색하고 응답을 만드는 모양이다. 니쥬의 대화는 방향이 반대다. 사용자가 묻는 것이 아니라 AI가 물어서 취향을 캐낸다. 그래서 설계의 중심 질문이 들어온 질의를 어떻게 처리할까가 아니라 이번 턴에 무엇을 물을까가 되고, 그 결정을 모델이 아니라 서버 규칙이 한다.

무엇을 물을지는 규칙이 정하고 어떻게 물을지는 모델이 정한다. 규칙은 지금까지 채워진 취향 칸과 턴 수, 사용자 발화 길이를 보고 목표 하나를 고른다. 모델은 그 목표에 해당하는 지시문 한 줄을 받아 문장을 쓴다. 모델은 흐름에 관여하지 않는다.

대가도 있다. 규칙은 사용자가 방금 한 말에 자연스럽게 이어지는 질문을 고르지 못한다. 목표와 문장을 나눈 구조가 그 대가를 흡수한다. 무엇을 물을지의 폭은 규칙이 좁히고, 그 안에서 자연스럽게 잇는 일은 모델이 문장 단계에서 한다.

이 결정은 대화에만 있는 것이 아니라 세 파이프라인에서 같은 모양으로 반복된다.

모델이 하는 일은 문장을 쓰거나 JSON을 내는 것뿐이다.

이것이 뒤의 프레임워크 판단의 전제다. 니쥬에 멀티스텝이 없다는 뜻이 아니다. 파이프라인은 셋이고 단계는 각각 일곱·셋·여섯이다. 없는 것은 흐름 제어를 모델에 넘길 때 생기는 문제이고, 오케스트레이션 프레임워크가 잘 푸는 것이 정확히 그 문제다. 조건부 분기, 재시도 순환, 도구 호출 루프는 다음에 무엇을 할지 모델이 정할 때 필요해진다. 그 전제를 처음부터 받지 않았으므로 그래프가 풀 문제가 남지 않는다.

flowchart LR
    L["세션 상태 로드<br/>Redis"] --> G["턴 목표 판정<br/>규칙 · LLM 미사용"]
    G --> R["응답 생성<br/>응답 모델 · SSE"]
    R --> P["응답 후처리<br/>지시문 절단 · 질문 1개"]
    P --> X["델타 추출<br/>추출 모델 · 구조화 출력"]
    X --> M["병합 · 정정 · 방출<br/>서버 로직"]
    M --> S["세션 상태 저장<br/>Redis"]

    R -.->|사용자 대기 구간| P

    style R fill:#f0f7ff
    style G fill:#fff4e6
    style M fill:#fff4e6

flowchart LR
    C["① 프로필 확정<br/>세션 상태 → TasteProfile"] --> SUM["② 요약문 생성<br/>응답 모델"]
    SUM --> K["③ 표시 키워드 선별<br/>순수 계산"]
    K --> R["profile + summary + keywords"]

    style SUM fill:#f0f7ff
    style C fill:#fff4e6
    style K fill:#fff4e6

flowchart TD
    Q["① 검색 문장 재작성<br/>취향 항목 → 파생 쿼리"] --> E["② 임베딩 배치<br/>Upstage Embed 2 · 1회 호출"]
    E --> V["③ 벡터 검색 · 하드필터<br/>pgvector 단일 쿼리"]
    V --> SC["④ 스코어 · 다양성 선별<br/>순수 계산"]
    SC --> RE["⑤ 추천 이유 생성<br/>응답 모델 · 쿼리 단위"]
    RE --> J["⑥ 응답 조립"]
    J --> D["self + gift 두 리스트"]

    DEL["세션 원문 삭제"]

    style RE fill:#f0f7ff
    style SC fill:#fff4e6
    style DEL stroke-dasharray: 4 4

flowchart LR
    subgraph N["니쥬의 세 파이프라인"]
        direction TB
        N1["분기 — 없음<br/>목표 8종은 지시문 문자열만 바뀜"]
        N2["순환 — 없음<br/>0건은 emptyReason으로 반환"]
        N3["병렬 — 없음<br/>세 파이프라인 모두 직선"]
        N4["상태 — Redis 직접 관리<br/>항목 상한 · 방출 규칙이 서버 로직"]
    end

    subgraph F["프레임워크가 필요할때"]
        direction TB
        F1["조건부 분기가 여러 개"]
        F2["재시도·완화 순환"]
        F3["복잡한 fan-out/fan-in"]
        F4["체크포인터 기반 재개"]
    end

    N1 -.->|해당 없음| F1
    N2 -.->|해당 없음| F2
    N3 -.->|gather 한 줄로 충분| F3
    N4 -.->|원문 미저장과 충돌| F4

리스트 재생성은 리스트 생성 잡과 단계가 같다. 진입점이 사용자 확정이 아니라 백엔드 트리거라는 점만 다르고, 확정된 프로필을 받아 도는 것은 동일하다. 같은 코드가 두 진입점을 처리하므로 별도 파이프라인으로 세지 않는다.

다만 이 경로를 잡으로 등록할지 요청 안에서 동기로 끝낼지는 아직 갈려 있다. 1단계 계약은 POST /v1/recommendations를 동기로 분류해 200에 결과를 바로 담고, 3단계는 사용자 확정과 백엔드 재생성 트리거가 같은 잡을 등록한다고 적었다. 단계 구성은 같아도 api에서 도는지 worker에서 도는지가 갈리므로 미확정으로 남긴다.

파이프라인 2와 3 사이에는 사용자 확인이 들어간다. 두 구간이 하나의 잡이 아닌 이유가 여기에 있다. 확인 없이 이어 붙이면 사용자가 정정할 수 없고, 정정 대화로 프로필이 바뀌면 뒤 구간의 산출물이 전부 버려진다.

②의 목표 8종은 조건부 분기처럼 보이지만 파이프라인이 갈라지는 것이 아니다. 어느 목표가 나오든 뒤의 단계는 동일하고, 달라지는 것은 ③에 넘기는 지시문 문자열 하나다. 그래프로 그리면 노드 8개가 같은 다음 노드로 모이는 모양이 되어 표현이 늘 뿐 실행이 달라지지 않는다.

⑤가 실패해도 ③은 이미 사용자에게 전달됐다. 부분 실패를 전체 실패로 만들지 않는다는 1단계 원칙이 파이프라인을 나눈 결과로 성립한다.

20번째 사용자 입력도 ③ 응답부터 ⑦ 저장까지 모두 처리한다. 그 뒤 입력을 잠그고 현재 State를 프로필로 확정한다. 스키마가 덜 채워졌더라도 21번째 질문을 만들지 않으며 completionReason=MAX_CYCLES, profileCompleteness=partial|sufficient로 종료한다. 마지막 추출이 실패하면 직전 State를 사용하고 실패 플래그를 남긴다.

close 요청 안에서 동기로 처리한다.

②만 모델 호출이고 나머지는 순수 계산이다. ②가 실패해도 ①③의 결과는 유효하므로 부분 실패로 처리한다. 세 단계 모두 api 안에서 인프로세스로 이어진다.

세션 원문은 이 구간에서 삭제하지 않는다. 사용자가 확인 화면에서 정정을 요청하면 같은 세션을 재개해야 하기 때문이다.

파이프라인 2와 3 사이에 사람이 들어간다. AI 서버가 도는 구간이 아니므로 단계로 세지 않지만, 앞뒤 파이프라인의 입출력을 결정하므로 여기 적는다.

정정 대화가 세션 TTL(유휴 30분 · 최대 2시간) 안에서만 가능하다는 제약이 여기 걸린다. 확인 대기 중 TTL을 연장할지는 미확정이다.

확정 요청으로 등록되어 worker에서 돈다.

갈래가 없다. 이전 설계에서는 요약문 생성이 검색 갈래와 병렬로 갈라졌는데, 요약문이 사용자 확인 대상이 되어 파이프라인 2로 옮겨가면서 병렬 지점이 사라졌다. estimatedMs를 맞추기 위한 병렬화가 필요했던 이유도 여덟 단계를 한 구간에 넣어야 했기 때문인데, 구간이 나뉘면서 그 제약도 함께 없어졌다.

③에서 후보가 0건이거나 ④에서 전량이 최소 점수에 미달하면 빈 리스트와 emptyReason을 반환한다. 쿼리를 완화해 ②로 돌아가지 않는다. 카탈로그에 상품이 없어서 0건인 경우 쿼리를 넓혀도 같은 결과가 나오고, 대기 시간만 두 배가 되기 때문이다. 부족한 구간은 카탈로그 커버리지로 집계해 수집으로 해결한다.

①의 입력은 1.3B 추출 모델이 방금 반환한 원시 델타가 아니다. 델타 → 서버 검증·병합 → Redis 누적 State → 종료 시 TasteProfile 확정 → 사용자 승인을 거친 최종 프로필이다. 8B는 그중 query 역할 항목을 상품 문서와 비교하기 좋은 queryText로 바꾼다. 서버는 모델 출력의 tasteNodeIds를 검증하고 원본 프로필에서 confidence, mode, hardFilters를 복원한 뒤에만 ② 임베딩 API로 보낸다. 따라서 추출 모델은 사용자 상태를 기억하거나 임베딩 벡터를 만들지 않고, 8B도 알레르기·기피 조건을 바꿀 권한이 없다.

잡이 실패하면 백엔드가 같은 프로필로 확정 요청을 다시 보낸다. 프로필이 이미 백엔드에 저장되어 있으므로 대화를 다시 하지 않는다. 세션 원문은 이미 지워졌지만 이 경로에는 필요하지 않다.

잡 등록과 같은 시점에 병렬로 실행한다. 잡의 단계에 넣지 않는 이유는 두 가지다.

• 삭제 결과를 기다리면 리스트 생성이 삭제 지연에 묶인다. 두 작업은 서로의 입력을 쓰지 않는다.
• 삭제가 실패해도 잡을 실패시킬 이유가 없다. 실패는 로그로 남기고 세션 TTL이 최종 삭제를 보장한다.

LangChain이 제공하는 것을 니쥬에서 그 자리를 이미 무엇이 채우고 있는지와 나란히 놓으면 이렇다.

여섯 항목 중 다섯은 이미 있는 것을 대체하는 일이고, 그중 셋은 대체하면 손해다. 남는 하나인 체인 조립은 함수를 순서대로 부르는 것 이상이 아니다.

마지막 항목이 판단의 핵심이다. 니쥬의 파이프라인에서 다음에 무엇을 할지는 전부 서버 코드가 정한다. 모델은 문장을 쓰거나 JSON을 내는 일만 하고 흐름에 관여하지 않는다. 흐름 제어를 모델에 맡기지 않기로 한 것은 1단계와 2단계에서 여러 번 반복한 결정이고, 그 결정이 유지되는 한 그래프 오케스트레이션이 풀 문제가 남지 않는다.

의존성이 늘지 않는다는 점이 부수 효과로 따라온다. ranking과 profile 모듈은 순수 파이썬이라 GPU도 DB도 없이 단위 테스트가 된다. 프레임워크를 넣으면 이 모듈들이 프레임워크 타입을 import하게 되어 그 성질이 깨진다.

다만 실제 서비스에서 아래 중 하나라도 성립하면 지금 내용을 재검토한다. 지금은 셋 다 성립하지 않는다.

두 번째 조건은 6단계에서 다시 다룬다. 외부 도구를 표준 프로토콜로 붙이기로 하면 흐름 제어를 모델에 넘기는 첫 지점이 되고, 그때 오케스트레이션 판단이 함께 바뀐다.

한 파이프라인에 모델이 셋 들어가지만 자체 서빙하는 것은 둘이다. 어느 모델을 몇 개 인스턴스에 나눠 올릴지는 7단계에서 정한다.

2단계는 비용과 데이터, 조정 여지 셋을 들어 상용 API에서 자체 서빙으로 옮겼다. 그 셋이 임베딩에는 걸리지 않는다.

여기에 GPU 이득이 붙는다. 임베딩 모델을 올리지 않는 만큼이 KV 캐시 몫으로 가고, 더 낮은 성능의 GPU도 사용 가능하다. 이는 8세션 동시성 목표에 직접 기여한다.

async def handle_turn(session_id: str, utterance: str) -> AsyncIterator[str]:
    state = session_store.load(session_id)          # ①
    goal = policy.decide(state)                     # ② LLM 미사용

    messages = prompts.build_reply(                 # ③ 접두사 고정
        system=SYSTEM_FIXED,                        #    불변
        history=state.history,                      #    누적만
        instruction=goal.instruction,               #    매 턴 바뀜
        utterance=utterance,
    )
    buffer = []
    async for token in chat_model.stream(messages, temperature=0.7):
        buffer.append(token)
        yield token                                 #    SSE

    reply = postprocess.trim(''.join(buffer))       # ④ 지시문 절단·질문 1개

    try:                                            # ⑤ 스트림 종료 후
        delta = await extractor.extract(
            prompts.build_extract(state, reply, utterance),
            schema=DELTA_SCHEMA,                    #    strict
        )
    except SchemaViolation:
        delta = Delta.empty()                       #    그 턴만 버린다

    state = profile.merge(state, delta)             # ⑥ 병합·정정·방출
    session_store.save(session_id, state, ttl_s=1800)   # ⑦

async def close_session(session_id: str) -> CloseResult:
    state = session_store.load(session_id)
    prof = profile.finalize(state)                       # ① 순수 계산

    try:                                                 # ② 모델 호출 1회
        summary = await summarizer.write(
            prof.friends_visible(), timeout_s=5
        )
    except (Timeout, LLMUnavailable):
        summary = None                                   #    부분 실패로 처리

    keywords = profile.display_keywords(prof)            # ③ 순수 계산

    session_store.mark_closed(session_id)                #    원문은 남긴다
    return CloseResult(
        profile=prof,
        summary=summary,
        keywords=keywords,
        correction_available=not state.correction_used,
    )

원문을 지우지 않고 mark_closed만 거는 것이 이 함수의 핵심이다. 사용자가 확인 화면에서 정정을 요청하면 같은 세션을 재개해야 한다.

async def confirm_profile(req: ConfirmRequest) -> Job:
    job = job_queue.enqueue(build_lists, req.profile, req.feedback_summary)

    if req.session_id:                                   # 잡과 병렬. 결과를 기다리지 않는다
        asyncio.create_task(purge_session(req.session_id))

    return job


async def build_lists(prof, feedback_summary, job) -> dict[str, RecoList]:
    queries = await rewriter.derive(prof)                             # ①
    job.progress(0.2)

    vectors = await embedder.embed_batch(queries)                     # ② 1회 호출
    candidates = catalog.search(vectors, prof.hard_filter(), k=500)   # ③ 단일 쿼리
    job.progress(0.5)

    out = {}
    for mode in ("self", "gift"):
        picked = ranking.score_and_diversify(candidates, prof, mode)  # ④ 순수
        if not picked:
            out[mode] = RecoList.empty(reason="NO_RELEVANT_CANDIDATE")
            continue
        reasons = await reasoner.per_query(picked, queries, mode)     # ⑤ 쿼리 단위
        out[mode] = RecoList(items=picked, reasons=reasons)
    job.progress(0.9)
    return out                                                        # ⑥ 조립

asyncio.gather가 없어진 것이 이전 설계와 달라진 부분이다. 요약문 생성이 앞 구간으로 옮겨가면서 합류를 기다릴 갈래가 남지 않았다. create_task는 남아 있지만 잡의 결과에 들어가지 않는 원문 삭제에만 쓰인다.

두 모드가 같은 candidates를 공유한다. 모드별로 잡을 나누면 ①~③이 두 번 돈다.

build_lists에 재검색 순환이 없다는 것이 앞의 판단을 코드로 보여주는 지점이다. 후보가 비면 완화 후 재시도하지 않고 빈 리스트에 이유를 붙여 그대로 넘긴다.

1단계는 부분 실패를 전체 실패로 만들지 않는다는 원칙을 정했고, 3단계는 jobId가 멱등 키라는 것까지 정했다. 어느 단계 실패가 어디까지 번지는지는 정해진 바가 없어 여기서 정한다.

프로필 확정 구간은 실패해도 잡이 아니므로 큐 재시도가 없다. ①③이 순수 계산이라 실패 지점이 ② 하나뿐이고, 그 하나가 부분 실패로 흡수된다.

리스트 생성 잡의 재실행 단위는 잡 하나다. 단계별 체크포인트를 두지 않는다. 근거는 셋이다.

• ④는 순수 계산이라 다시 돌아도 같은 결과가 나오고 비용이 거의 없다.
• ②는 임베딩 캐시가 있어 재실행이 대부분 캐시 히트다. 외부 API 호출이지만 재실행 비용이 호출 수에 비례하지 않는다.
• ③은 카탈로그를 읽기만 하고 쓰지 않는다. 파이프라인 전체에 외부 부작용이 없어서 중간부터 이어 붙일 이유가 없다.

잡이 여섯 단계로 짧아져 통째 재실행 비용이 체크포인트 관리 비용보다 싸다. 잡이 길어지면 이 판단을 다시 본다.

잡이 실패하면 백엔드가 같은 프로필로 확정 요청을 다시 보낸다. 세션 원문은 이미 지워졌지만 이 경로에는 필요하지 않다. 사용자에게 대화를 다시 시키지 않는다는 점이 구간을 나눈 부수 효과다.