AI 플랫폼: Feature Store와 RAGOps 블루프린트 2026

┌──────────────────────────────────────────────────────┐
│  Layer 4: Serving (KServe / API Gateway)             │
│  - Feature 조회 + Vector 검색 결과를 LLM에 전달      │
│  - Guardrail, Citation 검증, 응답 필터링             │
├──────────────────────────────────────────────────────┤
│  Layer 3: Feature Store + Vector Store               │
│  - Feast Online Store (Redis) : 정형 피처            │
│  - Vector DB (Qdrant/Weaviate) : 비정형 임베딩       │
│  - 통합 조회 API : 두 store를 하나의 인터페이스로    │
├──────────────────────────────────────────────────────┤
│  Layer 2: Ingestion & Transformation                 │
│  - CDC 파이프라인 (Debezium -> Kafka)                │
│  - 문서 청킹 + 임베딩 파이프라인 (Kubeflow)         │
│  - Feature View 정의 + Vector Index 관리             │
├──────────────────────────────────────────────────────┤
│  Layer 1: Source Data                                │
│  - OLTP DB, Data Warehouse, 문서 저장소, API 로그   │
└──────────────────────────────────────────────────────┘

"""
Feature Store(Feast)와 Vector Store(Qdrant)를 통합 조회하는 서비스.
서빙 레이어에서 모델이나 LLM에 컨텍스트를 제공할 때 사용한다.
"""
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional
import asyncio
from feast import FeatureStore
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue

@dataclass
class UnifiedContext:
    """Feature Store 피처 + Vector 검색 결과를 합친 통합 컨텍스트"""
    entity_id: str
    structured_features: Dict[str, Any]          # Feature Store에서 조회
    retrieved_documents: List[Dict[str, Any]]    # Vector Store에서 검색
    feature_freshness_ms: float = 0.0
    retrieval_latency_ms: float = 0.0

class UnifiedContextService:
    def __init__(
        self,
        feast_repo_path: str,
        qdrant_url: str,
        qdrant_collection: str,
    ):
        self.feast_store = FeatureStore(repo_path=feast_repo_path)
        self.qdrant = QdrantClient(url=qdrant_url, timeout=5.0)
        self.collection = qdrant_collection

    async def get_context(
        self,
        entity_id: str,
        query_embedding: List[float],
        feature_list: List[str],
        top_k: int = 5,
        metadata_filter: Optional[Dict[str, str]] = None,
    ) -> UnifiedContext:
        """
        Feature Store 조회와 Vector 검색을 병렬로 수행하여
        통합 컨텍스트를 반환한다.
        """
        # 병렬 실행: Feature Store 조회 + Vector 검색
        feature_task = asyncio.to_thread(
            self._fetch_features, entity_id, feature_list
        )
        vector_task = asyncio.to_thread(
            self._search_vectors, query_embedding, top_k, metadata_filter, entity_id
        )

        features_result, vectors_result = await asyncio.gather(
            feature_task, vector_task
        )

        return UnifiedContext(
            entity_id=entity_id,
            structured_features=features_result["features"],
            retrieved_documents=vectors_result["documents"],
            feature_freshness_ms=features_result["latency_ms"],
            retrieval_latency_ms=vectors_result["latency_ms"],
        )

    def _fetch_features(self, entity_id: str, feature_list: List[str]) -> dict:
        import time
        start = time.monotonic()
        result = self.feast_store.get_online_features(
            features=feature_list,
            entity_rows=[{"user_id": entity_id}],
        ).to_dict()
        elapsed = (time.monotonic() - start) * 1000

        # dict에서 리스트 값을 스칼라로 변환
        features = {k: v[0] if isinstance(v, list) and v else v for k, v in result.items()}
        return {"features": features, "latency_ms": elapsed}

    def _search_vectors(
        self,
        query_embedding: List[float],
        top_k: int,
        metadata_filter: Optional[Dict[str, str]],
        entity_id: str,
    ) -> dict:
        import time
        start = time.monotonic()

        search_filter = None
        if metadata_filter:
            conditions = [
                FieldCondition(key=k, match=MatchValue(value=v))
                for k, v in metadata_filter.items()
            ]
            search_filter = Filter(must=conditions)

        results = self.qdrant.search(
            collection_name=self.collection,
            query_vector=query_embedding,
            query_filter=search_filter,
            limit=top_k,
            with_payload=True,
        )
        elapsed = (time.monotonic() - start) * 1000

        documents = [
            {
                "text": hit.payload.get("text", ""),
                "source": hit.payload.get("source", ""),
                "score": hit.score,
                "chunk_id": hit.id,
            }
            for hit in results
        ]
        return {"documents": documents, "latency_ms": elapsed}

"""
문서 청킹 + 임베딩 파이프라인.
Kubeflow Pipeline component로 실행하거나 독립 스크립트로 사용한다.
"""
from dataclasses import dataclass
from typing import List, Generator
import hashlib
import json

@dataclass
class ChunkConfig:
    """청킹 설정. 문서 유형별로 다른 전략을 적용한다."""
    chunk_size: int = 512          # 토큰 기준
    chunk_overlap: int = 64        # 청크 간 겹침
    separators: tuple = ("\n\n", "\n", ". ", " ")
    min_chunk_size: int = 50       # 이 미만은 이전 청크에 병합
    metadata_fields: tuple = ("source", "doc_type", "updated_at")

# 문서 유형별 청킹 전략
CHUNK_CONFIGS = {
    "product_review": ChunkConfig(chunk_size=256, chunk_overlap=32),
    "technical_doc": ChunkConfig(chunk_size=512, chunk_overlap=64),
    "faq": ChunkConfig(chunk_size=128, chunk_overlap=0),  # FAQ는 겹침 불필요
    "legal": ChunkConfig(chunk_size=1024, chunk_overlap=128),
}

def recursive_split(text: str, config: ChunkConfig) -> List[str]:
    """재귀적 텍스트 분할. separators를 순서대로 시도한다."""
    if len(text.split()) <= config.chunk_size:
        return [text] if len(text.split()) >= config.min_chunk_size else []

    for sep in config.separators:
        parts = text.split(sep)
        if len(parts) > 1:
            chunks = []
            current = ""
            for part in parts:
                candidate = current + sep + part if current else part
                if len(candidate.split()) > config.chunk_size:
                    if current:
                        chunks.append(current.strip())
                    current = part
                else:
                    current = candidate
            if current:
                chunks.append(current.strip())
            return [c for c in chunks if len(c.split()) >= config.min_chunk_size]

    # 모든 separator로 분할 불가 시 강제 분할
    words = text.split()
    return [
        " ".join(words[i:i + config.chunk_size])
        for i in range(0, len(words), config.chunk_size - config.chunk_overlap)
    ]

def create_chunk_id(source: str, chunk_index: int, text: str) -> str:
    """결정론적 chunk ID 생성. 동일 문서의 동일 청크는 같은 ID를 갖는다."""
    content_hash = hashlib.sha256(text.encode()).hexdigest()[:12]
    return f"{source}::{chunk_index}::{content_hash}"

def process_document(
    text: str,
    source: str,
    doc_type: str,
    metadata: dict,
    embedding_fn,
) -> List[dict]:
    """
    하나의 문서를 청킹하고 임베딩하여 Vector Store에 upsert할
    레코드 리스트를 반환한다.
    """
    config = CHUNK_CONFIGS.get(doc_type, ChunkConfig())
    chunks = recursive_split(text, config)

    records = []
    for i, chunk_text in enumerate(chunks):
        embedding = embedding_fn(chunk_text)
        chunk_id = create_chunk_id(source, i, chunk_text)
        records.append({
            "id": chunk_id,
            "vector": embedding,
            "payload": {
                "text": chunk_text,
                "source": source,
                "doc_type": doc_type,
                "chunk_index": i,
                "total_chunks": len(chunks),
                "chunk_size_tokens": len(chunk_text.split()),
                **{k: metadata.get(k, "") for k in config.metadata_fields},
            },
        })
    return records

"""
Feature Store 피처 + RAG 검색 결과를 결합하여
LLM 프롬프트를 구성하는 패턴.
"""
from typing import Dict, List, Any

def build_augmented_prompt(
    user_query: str,
    structured_features: Dict[str, Any],
    retrieved_documents: List[Dict[str, Any]],
    system_instruction: str = "",
) -> str:
    """
    정형 피처와 비정형 검색 결과를 결합한 프롬프트 생성.

    원칙:
    1. 정형 피처는 "사용자 컨텍스트" 섹션에 표 형태로 제공
    2. RAG 검색 결과는 "참고 문서" 섹션에 출처와 함께 제공
    3. 피처와 문서 간 충돌 시 피처(실시간 데이터)를 우선하라고 명시
    """
    # 정형 피처 섹션
    feature_lines = []
    feature_display_names = {
        "total_purchases_7d": "최근 7일 구매 횟수",
        "avg_order_value_30d": "최근 30일 평균 주문 금액",
        "preferred_category": "선호 카테고리",
        "membership_tier": "멤버십 등급",
    }
    for key, value in structured_features.items():
        display = feature_display_names.get(key, key)
        feature_lines.append(f"- {display}: {value}")

    features_section = "\n".join(feature_lines) if feature_lines else "없음"

    # RAG 검색 결과 섹션
    doc_sections = []
    for i, doc in enumerate(retrieved_documents, 1):
        score = doc.get("score", 0)
        # relevance score가 낮은 문서는 제외
        if score < 0.7:
            continue
        doc_sections.append(
            f"[문서 {i}] (출처: {doc['source']}, 관련도: {score:.2f})\n{doc['text']}"
        )

    docs_section = "\n\n".join(doc_sections) if doc_sections else "관련 문서 없음"

    prompt = f"""{system_instruction}

## 사용자 컨텍스트 (Feature Store 실시간 데이터)
{features_section}

## 참고 문서 (RAG 검색 결과)
{docs_section}

## 지침
- 사용자 컨텍스트와 참고 문서를 모두 활용하여 답변하세요.
- 정형 데이터(사용자 컨텍스트)와 문서 내용이 충돌하면 정형 데이터를 우선합니다.
- 답변에 사용한 참고 문서 번호를 인용하세요.
- 참고 문서에 없는 내용은 추측하지 마세요.

## 사용자 질문
{user_query}"""

    return prompt

"""
Vector Store 인덱스의 최신성을 추적하는 모니터링 모듈.
문서 원천의 최신 업데이트 시각과 인덱스의 마지막 갱신 시각을
비교하여 stale 상태를 감지한다.
"""
from datetime import datetime, timedelta
from typing import Dict, List
from dataclasses import dataclass
import json

@dataclass
class IndexFreshnessReport:
    collection_name: str
    total_documents: int
    stale_documents: int
    freshness_ratio: float
    oldest_document_age_hours: float
    avg_document_age_hours: float
    recommendations: List[str]

def check_index_freshness(
    qdrant_client,
    collection_name: str,
    freshness_threshold_hours: int = 24,
    sample_size: int = 1000,
) -> IndexFreshnessReport:
    """
    Vector Store 인덱스의 문서 최신성을 점검한다.
    """
    # 샘플 문서의 updated_at 필드를 조회
    points, _ = qdrant_client.scroll(
        collection_name=collection_name,
        limit=sample_size,
        with_payload=["updated_at", "source"],
    )

    now = datetime.utcnow()
    ages_hours = []
    stale_count = 0
    stale_sources = set()

    for point in points:
        updated_at_str = point.payload.get("updated_at", "")
        if not updated_at_str:
            stale_count += 1
            continue
        try:
            updated_at = datetime.fromisoformat(updated_at_str)
            age_hours = (now - updated_at).total_seconds() / 3600
            ages_hours.append(age_hours)
            if age_hours > freshness_threshold_hours:
                stale_count += 1
                stale_sources.add(point.payload.get("source", "unknown"))
        except ValueError:
            stale_count += 1

    total = len(points)
    freshness_ratio = 1.0 - (stale_count / max(total, 1))

    recommendations = []
    if freshness_ratio < 0.95:
        recommendations.append(
            f"Stale 문서 비율이 {(1-freshness_ratio)*100:.1f}%입니다. "
            f"임베딩 파이프라인 실행이 필요합니다."
        )
    if stale_sources:
        recommendations.append(
            f"Stale 문서 출처: {', '.join(list(stale_sources)[:5])}"
        )
    if ages_hours and max(ages_hours) > freshness_threshold_hours * 3:
        recommendations.append(
            "매우 오래된 문서가 존재합니다. 전체 재인덱싱을 고려하세요."
        )

    return IndexFreshnessReport(
        collection_name=collection_name,
        total_documents=total,
        stale_documents=stale_count,
        freshness_ratio=freshness_ratio,
        oldest_document_age_hours=max(ages_hours) if ages_hours else 0,
        avg_document_age_hours=sum(ages_hours) / len(ages_hours) if ages_hours else 0,
        recommendations=recommendations,
    )

# rag-index-update-pipeline.yaml
# 주기적으로 변경된 문서를 감지하여 벡터 인덱스를 갱신한다.
apiVersion: argoproj.io/v1alpha1
kind: CronWorkflow
metadata:
  name: rag-index-updater
  namespace: ml-pipelines
spec:
  schedule: '0 */4 * * *' # 4시간마다 실행
  concurrencyPolicy: Replace
  workflowSpec:
    entrypoint: update-index
    templates:
      - name: update-index
        steps:
          - - name: detect-changes
              template: detect-document-changes
          - - name: chunk-and-embed
              template: run-embedding-pipeline
              arguments:
                parameters:
                  - name: changed-docs
                    value: '{{steps.detect-changes.outputs.parameters.changed-docs}}'
          - - name: verify-freshness
              template: check-freshness

      - name: detect-document-changes
        container:
          image: ml-platform/doc-change-detector:v1.2.0
          command: [python, detect_changes.py]
          args:
            - --since=4h
            - --output=/tmp/changed_docs.json
          env:
            - name: DOC_STORE_URL
              valueFrom:
                secretKeyRef:
                  name: doc-store-credentials
                  key: url

      - name: run-embedding-pipeline
        container:
          image: ml-platform/embedding-worker:v2.1.0
          command: [python, embed_and_upsert.py]
          args:
            - --docs-file={{inputs.parameters.changed-docs}}
            - --collection=product_knowledge
            - --model=text-embedding-3-large
          resources:
            requests:
              nvidia.com/gpu: 1
              memory: 8Gi

      - name: check-freshness
        container:
          image: ml-platform/rag-monitor:v1.0.0
          command: [python, check_freshness.py]
          args:
            - --collection=product_knowledge
            - --threshold-hours=24
            - --alert-on-failure

증상: RAG 챗봇 답변 품질이 급락. 사용자 피드백 "엉뚱한 답변" 증가.
     Retrieval recall@5가 0.85에서 0.52로 하락.

점검 순서:
  1. 최근 임베딩 파이프라인 실행 로그 확인
     -> 임베딩 모델 버전이 바뀌었는지 체크
  2. Qdrant collection info 확인
     -> vector dimension이 변경되었는지 체크
  3. 원인: 임베딩 모델을 text-embedding-3-small에서 text-embedding-3-large로
     업그레이드했으나 기존 벡터는 재임베딩하지 않아 차원 불일치

해결:
  1. 임베딩 모델 변경 시 반드시 전체 재인덱싱 실행
  2. 새 collection 생성 -> 전체 임베딩 -> alias 전환 (blue-green 패턴)
  3. 재인덱싱 완료 전까지 이전 모델로 롤백

증상: 통합 조회 API에서 간헐적 500 에러.
     에러 로그: "TimeoutError: Feast online store read timed out after 50ms"

원인: Feature Store(Redis)는 타임아웃이지만 Vector Store(Qdrant)는 정상.
     하나의 실패가 전체 요청을 실패시킴.

해결:
  1. 통합 조회 API에 partial failure 허용 로직 추가
  2. Feature Store 실패 시: 기본값(인구통계 평균) 사용, 로그 기록
  3. Vector Store 실패 시: 사전 캐시된 인기 문서로 fallback
  4. 양쪽 모두 실패 시: LLM에 컨텍스트 없이 일반 응답 생성 (품질 저하 경고 포함)