멀티스텝 AI 파이프라인 구현 검토 - 100-hours-a-week/KTB4-13th-wiki GitHub Wiki

def agent_act(req: AgentRequest) -> AgentResponse:
    if cached := get_idempotent_result(req.idempotency_key):   # 같은 키·같은 본문 → 저장된 결과
        return cached

    # 1. 지시어 해소 ("N번" → context_cards[N-1], "이거" → focused_book_id)
    ref = resolve_reference(req.message, req.context_cards, req.focused_book_id)
    if ref.confidence < 0.6:
        return AgentResponse(reply=make_reask(ref), tool_calls=[],           # tool 미실행, 200
                             resolved_reference=ref, selection=None,
                             buttons=[confirm_reference_button(ref)])

    # 2. 의도 분류
    intent = classify_intent(req.message)   # add | qty_edit | cart_view | order_preview | stock | select | compare

    # 3. tool 루프 — 상한 3스텝, 독립 호출은 병렬, 쓰기엔 파생 멱등키
    calls: list[ToolCall] = []
    selection = None
    if intent == "select":                                      # "3만원 안에서 2권"
        cands = call_tool("recommendations.candidates", {"size": 20}, calls)      # 내부 모듈, 감사 기록
        verified = call_tool("inventory.check",                                   # 스냅샷 신뢰 안 함
                             {"book_ids": [c.book_id for c in cands.candidates]}, calls)
        selection = deterministic_pick(cands, verified, req.constraints)          # 조합 전수열거, LLM 아님
        buttons = [confirm_selection_button(selection.chosen)]
    elif intent == "add":
        idem = f"{req.idempotency_key}:{len(calls)}"
        call_tool("cart.add", {"book_id": ref.book_ids[0], "qty": 1, "_idem": idem}, calls)
        buttons = [navigate_button("CART-001")]
    elif intent == "qty_edit":                                   # 수량 변경 / 삭제
        idem = f"{req.idempotency_key}:{len(calls)}"
        call_tool("cart.update",                                  # 수량 0 = 삭제
                  {"book_id": ref.book_ids[0], "qty": req.target_qty, "_idem": idem}, calls)
        buttons = [navigate_button("CART-001")]
    elif intent == "compare":
        for bid in ref.book_ids:
            call_tool("book.detail", {"book_id": bid}, calls)                     # 병렬
        buttons = [confirm_reference_button(ref)]
    else:  # stock / cart_view / order_preview / 범위 밖
        buttons = handle_read_or_out_of_scope(intent, ref, calls)

    # 4. grounding — 발화의 모든 수치·상태는 calls[].result에서만
    draft = llm_compose_reply(intent, ref, calls, selection)
    reply, calls = enforce_grounding(draft, calls)
    if core_claim_ungrounded(reply, intent):                    # 축약해도 핵심 주장이 근거 미확보
        raise GroundingFailed(422)                              # 발화 전체 폐기

    result = AgentResponse(reply=reply, tool_calls=calls, resolved_reference=ref,
                           selection=selection, buttons=buttons)
    save_idempotent_result(req.idempotency_key, result)
    return result

def enforce_grounding(draft: str, calls: list[ToolCall]) -> tuple[str, list[ToolCall]]:
    facts = index_numbers_and_states(calls)     # 가격·재고·수량·합계·별점
    kept = []
    for s in split_sentences(draft):
        needed = extract_claims(s)              # 이 문장이 주장하는 수치·상태
        if all(claim in facts for claim in needed):
            kept.append(s)
        else:
            mark_ungrounded(calls, needed)      # 해당 tool_call.grounded = False, 문장 제거
    return " ".join(kept), calls

def hybrid_candidates(text_query: str, filters: dict, size: int) -> list[Book]:
    kw_hits = bm25_search(text_query, filters)
    query_vec = call_embeddings_api([text_query], purpose="query")[0]  # bge-m3-2026q3
    vec_hits = pgvector_search(query_vec, filters)
    return reciprocal_rank_fusion(kw_hits, vec_hits, k=60)[:size]

def search(query: str, filters: dict, cursor: str | None, size: int = 15) -> SearchResponse:
    fused = hybrid_candidates(query, filters, size)
    ranked = apply_business_rules(fused, popularity=read_replica("v_book_popularity"))  # 복제 테이블. 행 없으면 인기항 0
    if not ranked:
        return SearchResponse(results=[], next_cursor=None,
                               fallback={"message": "원하는 책을 못 찾았어요. AI 추천에게 물어볼까요?"})
    return SearchResponse(results=ranked, next_cursor=make_cursor(ranked, query, filters))

def recommend_chat(user_id: int, message: str | None, image_ref: str | None,
                    prev_spec: Spec, recent_turns: list, exclude_book_ids: list) -> ChatResponse:
    if image_ref:
        return handle_image_turn(image_ref, prev_spec, user_id)   # 인식, cards는 recognized일 때만

    validate_spec_schema(prev_spec)   # 요청 spec이 6키 스키마 위반 → 422 spec_schema_violation
                                      # (LLM 호출 전. 호출자가 초기 spec으로 1회 재시도)

    # LLM 호출 1 — 발화 해석
    delta = llm_extract_spec_delta(message, prev_spec, recent_turns)   # 6키만
    spec = merge_spec(prev_spec, delta)                                # 서버가 기존 spec에 델타 병합

    # 결정론적 후보 풀 + 채점 — LLM 아님
    profile = load_user_profile(user_id)                              # centroid + computed_at
    hist = read_history_replica(user_id)                              # 복제 테이블 v_user_purchases/library/reviews
    excluded = set(exclude_book_ids) | set(spec["exclude"]) \
             | hist.purchased | hist.rated_low                        # 산 책·별점 1~2 제외
    cands = hybrid_candidates(spec["semantic"] or "", spec["filters"], size=10)
    cands = [c for c in cands if c.book_id not in excluded]

    scored = {c.book_id: (c, combine_score(
                  query_sim=c.vec_sim,
                  taste_sim=cosine(c.vec, profile.centroid),
                  post_profile_hist=hist.since(profile.computed_at),   # computed_at(반영시각) 이후 이력 가산
                  popularity=read_replica("v_book_popularity", c.book_id)))
              for c in cands}
    if not scored:
        return ChatResponse(reply=..., spec=spec, cards=[],
                            followup=make_reclarify_question(spec),
                            buttons=[], degraded=False)

    # LLM 호출 2 (1회) — 10 후보 중 3권 선택 + 그 3권에 이유 2종
    picked = llm_pick_and_write(scored, spec, profile)
    # → [{book_id, reason_short, reason_long, match_basis}] 최대 3개. 배열 순서가 곧 rank
    cards = []
    for rank, r in enumerate(picked, start=1):
        if r.book_id not in scored or r.reason_short is None:   # 후보 밖이거나 한 줄 이유 실패 → 제외
            continue
        book, score = scored[r.book_id]
        cards.append(Card(book_id=book.book_id, rank=rank,
                          match_score=int(score),               # 후보 점수 그대로. rank 순과 무관
                          reason_short=r.reason_short,
                          reason_long=r.reason_long,             # null 가능 → BE가 상세 이유영역 숨김
                          match_basis=r.match_basis))
    return ChatResponse(reply=compose_reply(cards), spec=spec, cards=cards,
                        followup=None, buttons=[], degraded=False)

    # LLM 장애 강등: 위 두 호출을 건너뛰고 요청의 spec으로 채점만 → 점수 상위 3권,
    # degraded=True, reason_short는 규칙, reason_long=null

def rebuild_profile(user_id: int, idempotency_key: str, onboarding: dict, memories: list) -> ProfileResult:
    if cached := get_idempotent_result(idempotency_key):   # 24h 보관, 같은 본문이면 재계산 스킵
        return cached
    if any(m["dim"] != INDEX_DIM for m in memories):
        raise InvalidRequest(400)                          # 차원 불일치

    hist = read_history_replica(user_id)                   # 요청 본문에 없음 — user_id로 복제 테이블 조회
    liked_vecs  = [book_doc_vector(b) for b in onboarding.get("liked_book_ids", [])[:50]]
    author_vecs = [author_vector(m["value"]) for m in memories if m["type"] == "author"]
    tag_vecs    = preset_vectors(onboarding.get("categories", []) + onboarding.get("tags", []))
    mem_vecs    = [m["vector"] for m in memories[-500:]]   # 이미 ⑤에서 임베딩 — 재호출 없음
    hist_vecs   = hist.weighted_doc_vectors()             # 구매3 / 리뷰4~5점2 / 담기1, 리뷰1~2점 제외

    centroid    = weighted_average(liked_vecs + author_vecs + tag_vecs + mem_vecs + hist_vecs)
    tag_weights = aggregate_tags(onboarding, memories, hist)   # 리뷰1~2점은 여기서 −2

    reflected_at = hist.max_row_ts()   # 실제로 읽어 반영한 이력 행들의 최대 시각. now() 아님 —
                                       # 복제 늦게 온 이력이 프로필·가산 양쪽에서 영구 누락되지 않게.
                                       # 반영한 이력이 없으면 None → ③④가 이력 전부 가산
    version = upsert_user_profile(user_id, centroid, tag_weights, computed_at=reflected_at)  # user_id 직렬
    result = ProfileResult(cold_start=not (liked_vecs or mem_vecs or hist.any()),
                           profile_version=version)
    save_idempotent_result(idempotency_key, result)
    return result