필사 모드: 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、数据仓库、文档存储、API 日志             │
└──────────────────────────────────────────────────────┘

"""
统一查询 Feature Store(Feast) 与 Vector Store(Qdrant) 的服务。
在 serving 层为模型或 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 中的 list 值转换为标量
        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          # 以 token 为单位
    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]:
    """递归式文本拆分。按顺序尝试各个 separator。"""
    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]:
    """
    对单个文档做分块与嵌入，返回一份可以
    upsert 进 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

"""
把 Feature Store 特征 + RAG 检索结果结合起来
构建 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:
    """
    生成结合了结构化特征与非结构化检索结果的 prompt。

    原则：
    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 失败时: 回退到预先缓存的热门文档
  4. 双方都失败时: 让 LLM 在没有上下文的情况下生成通用回答（附带质量下降警告）