AI Platform: Feature Store and RAGOps Blueprint 2026

┌──────────────────────────────────────────────────────┐
│  Layer 4: Serving (KServe / API Gateway)             │
│  - Deliver Feature lookup + Vector search to LLM     │
│  - Guardrail, Citation verification, Response filter  │
├──────────────────────────────────────────────────────┤
│  Layer 3: Feature Store + Vector Store               │
│  - Feast Online Store (Redis): Structured features    │
│  - Vector DB (Qdrant/Weaviate): Unstructured embeds   │
│  - Unified Query API: Single interface for both       │
├──────────────────────────────────────────────────────┤
│  Layer 2: Ingestion & Transformation                 │
│  - CDC Pipeline (Debezium -> Kafka)                  │
│  - Document chunking + embedding pipeline (Kubeflow) │
│  - Feature View definition + Vector Index management  │
├──────────────────────────────────────────────────────┤
│  Layer 1: Source Data                                │
│  - OLTP DB, Data Warehouse, Document Store, API Logs │
└──────────────────────────────────────────────────────┘

"""
Service that unifies Feature Store (Feast) and Vector Store (Qdrant) queries.
Used in the serving layer to provide context to models or LLMs.
"""
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:
    """Unified context combining Feature Store features + Vector search results"""
    entity_id: str
    structured_features: Dict[str, Any]          # Retrieved from Feature Store
    retrieved_documents: List[Dict[str, Any]]    # Searched from 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:
        """
        Performs Feature Store lookup and Vector search in parallel
        to return a unified context.
        """
        # Parallel execution: Feature Store lookup + Vector search
        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

        # Convert list values to scalars in the 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}

"""
Document chunking + embedding pipeline.
Can be run as a Kubeflow Pipeline component or as a standalone script.
"""
from dataclasses import dataclass
from typing import List, Generator
import hashlib
import json

@dataclass
class ChunkConfig:
    """Chunking configuration. Apply different strategies per document type."""
    chunk_size: int = 512          # Token-based
    chunk_overlap: int = 64        # Overlap between chunks
    separators: tuple = ("\n\n", "\n", ". ", " ")
    min_chunk_size: int = 50       # Merge with previous chunk if under this
    metadata_fields: tuple = ("source", "doc_type", "updated_at")

# Per document type chunking strategies
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),  # No overlap needed for FAQ
    "legal": ChunkConfig(chunk_size=1024, chunk_overlap=128),
}

def recursive_split(text: str, config: ChunkConfig) -> List[str]:
    """Recursive text splitting. Tries separators in order."""
    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]

    # Force split if no separator works
    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:
    """Deterministic chunk ID generation. Same document + same chunk = same 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]:
    """
    Chunks and embeds a single document, returning a list of records
    ready to upsert into the Vector Store.
    """
    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

"""
Pattern for combining Feature Store features + RAG search results
to compose an LLM prompt.
"""
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:
    """
    Generate a prompt combining structured features and unstructured search results.

    Principles:
    1. Structured features are provided in table format in a "User Context" section
    2. RAG search results are provided in a "Reference Documents" section with sources
    3. Explicitly state that structured data (real-time) takes priority when there is a conflict
    """
    # Structured features section
    feature_lines = []
    feature_display_names = {
        "total_purchases_7d": "Purchases in last 7 days",
        "avg_order_value_30d": "Avg order value (30 days)",
        "preferred_category": "Preferred category",
        "membership_tier": "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 "None"

    # RAG search results section
    doc_sections = []
    for i, doc in enumerate(retrieved_documents, 1):
        score = doc.get("score", 0)
        # Exclude documents with low relevance scores
        if score < 0.7:
            continue
        doc_sections.append(
            f"[Document {i}] (Source: {doc['source']}, Relevance: {score:.2f})\n{doc['text']}"
        )

    docs_section = "\n\n".join(doc_sections) if doc_sections else "No relevant documents found"

    prompt = f"""{system_instruction}

## User Context (Feature Store Real-Time Data)
{features_section}

## Reference Documents (RAG Search Results)
{docs_section}

## Instructions
- Use both user context and reference documents to answer.
- When structured data (user context) conflicts with document content, prioritize the structured data.
- Cite the reference document numbers used in your answer.
- Do not speculate about content not found in the reference documents.

## User Question
{user_query}"""

    return prompt

"""
Monitoring module that tracks Vector Store index freshness.
Compares the latest update time of the document source with the
last update time of the index to detect stale states.
"""
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:
    """
    Checks the document freshness of a Vector Store index.
    """
    # Query updated_at fields from sample documents
    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 document ratio is {(1-freshness_ratio)*100:.1f}%. "
            f"Embedding pipeline execution required."
        )
    if stale_sources:
        recommendations.append(
            f"Stale document sources: {', '.join(list(stale_sources)[:5])}"
        )
    if ages_hours and max(ages_hours) > freshness_threshold_hours * 3:
        recommendations.append(
            "Very old documents exist. Consider a full re-indexing."
        )

    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
# Periodically detects changed documents and updates the vector index.
apiVersion: argoproj.io/v1alpha1
kind: CronWorkflow
metadata:
  name: rag-index-updater
  namespace: ml-pipelines
spec:
  schedule: '0 */4 * * *' # Run every 4 hours
  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

Symptom: RAG chatbot answer quality drops sharply. User feedback of "irrelevant answers" increases.
         Retrieval recall@5 drops from 0.85 to 0.52.

Investigation order:
  1. Check recent embedding pipeline execution logs
     -> Check if embedding model version changed
  2. Check Qdrant collection info
     -> Check if vector dimension changed
  3. Root cause: Upgraded embedding model from text-embedding-3-small to text-embedding-3-large
     but did not re-embed existing vectors, causing dimension mismatch

Resolution:
  1. Always run full re-indexing when changing embedding models
  2. Create new collection -> full embedding -> alias switch (blue-green pattern)
  3. Roll back to previous model until re-indexing is complete

Symptom: Intermittent 500 errors from the unified query API.
         Error log: "TimeoutError: Feast online store read timed out after 50ms"

Root cause: Feature Store (Redis) is timing out but Vector Store (Qdrant) is fine.
            One failure causes the entire request to fail.

Resolution:
  1. Add partial failure tolerance logic to the unified query API
  2. On Feature Store failure: use default values (demographic averages), log the event
  3. On Vector Store failure: fall back to pre-cached popular documents
  4. On both failures: generate general response without context via LLM (include quality degradation warning)