flowchart TB
    subgraph Clients["客户端层"]
        RAGEngine["RAG Engine"]
        AgentEngine["Agent Engine"]
        APIClient["外部 API 客户端"]
    end

    subgraph Gateway["网关层"]
        APIGateway["API Gateway<br/>路由/限流/鉴权"]
    end

    subgraph RetrievalSvc["Retrieval Service 服务层"]
        HTTPServer["HTTP Server<br/>FastAPI + Uvicorn"]

        subgraph Middleware["中间件层"]
            RequestID["RequestIDMiddleware<br/>请求追踪"]
            CORS["CORSMiddleware<br/>跨域处理"]
            Auth["AuthMiddleware<br/>认证鉴权"]
            RateLimit["RateLimitMiddleware<br/>速率限制"]
            Idempotency["IdempotencyMiddleware<br/>幂等性保证"]
            LogMetrics["LoggingMetricsMiddleware<br/>日志/指标记录"]
        end

        subgraph Routers["路由层"]
            RetrievalRouter["RetrievalRouter<br/>/api/v1/retrieval/*"]
            HealthRouter["Health Endpoints<br/>/health, /ready, /stats"]
        end

        subgraph Services["服务编排层"]
            RetrievalService["RetrievalService<br/>检索协调器"]
            VectorService["VectorService<br/>向量检索服务"]
            BM25Service["BM25Service<br/>BM25检索服务"]
            HybridService["HybridService<br/>RRF融合服务"]
            RerankService["RerankService<br/>重排序服务"]
            EmbeddingService["EmbeddingService<br/>向量化服务"]
            GraphService["GraphRetrievalService<br/>图谱检索服务"]
            HybridGraphService["HybridGraphService<br/>三路混合检索"]
            QueryExpansion["QueryExpansionService<br/>查询扩展服务"]
        end
    end

    subgraph Infrastructure["基础设施层"]
        Neo4jClient["Neo4jClient<br/>图数据库连接"]
        RedisCache["Redis Cache<br/>语义缓存<br/>TTL=3600s"]
        Metrics["Metrics<br/>Prometheus指标"]
        Logging["Logging<br/>结构化日志"]
        Tracing["Tracing<br/>OpenTelemetry"]
        CostTracker["CostTracker<br/>成本追踪"]
    end

    subgraph Storage["存储层"]
        Milvus["Milvus<br/>向量数据库<br/>HNSW索引"]
        Elasticsearch["Elasticsearch<br/>全文检索<br/>BM25算法"]
        Neo4j["Neo4j<br/>知识图谱<br/>实体关系"]
        RedisDB["Redis<br/>缓存数据库"]
    end

    %% 客户端到网关
    Clients --> APIGateway
    APIGateway --> HTTPServer

    %% HTTP层到中间件
    HTTPServer --> RequestID
    RequestID --> CORS
    CORS --> Auth
    Auth --> RateLimit
    RateLimit --> Idempotency
    Idempotency --> LogMetrics

    %% 中间件到路由层
    LogMetrics --> RetrievalRouter
    LogMetrics --> HealthRouter

    %% 路由层到服务编排层
    RetrievalRouter --> RetrievalService

    %% 服务编排层内部协作
    RetrievalService --> VectorService
    RetrievalService --> BM25Service
    RetrievalService --> HybridService
    RetrievalService --> RerankService
    RetrievalService --> GraphService
    RetrievalService --> HybridGraphService
    RetrievalService --> QueryExpansion

    VectorService --> EmbeddingService
    HybridService --> VectorService
    HybridService --> BM25Service
    HybridGraphService --> VectorService
    HybridGraphService --> BM25Service
    HybridGraphService --> GraphService
    HybridGraphService --> RerankService

    %% 服务层到基础设施
    RetrievalService --> RedisCache
    RetrievalService --> Metrics
    RetrievalService --> Logging
    RetrievalService --> Tracing
    RetrievalService --> CostTracker
    GraphService --> Neo4jClient

    %% 基础设施到存储层
    VectorService --> Milvus
    BM25Service --> Elasticsearch
    Neo4jClient --> Neo4j
    RedisCache --> RedisDB

    style Clients fill:#e3f2fd
    style Gateway fill:#fff9c4
    style RetrievalSvc fill:#fff3e0
    style Middleware fill:#ffe0b2
    style Routers fill:#ffccbc
    style Services fill:#f3e5f5
    style Infrastructure fill:#e0f7fa
    style Storage fill:#e0f2f1

flowchart TB
    subgraph Client["客户端"]
        RAG["RAG Engine"]
    end

    subgraph HTTP["HTTP 层"]
        FastAPI["FastAPI Server"]
    end

    subgraph Middleware["中间件层"]
        ReqID["RequestIDMiddleware"]
        CORS["CORSMiddleware"]
        Auth["AuthMiddleware"]
        LogMw["LoggingMetricsMiddleware"]
    end

    subgraph API["API 层"]
        Router["RetrievalRouter<br/>/api/v1/retrieval/hybrid"]
        Validator["请求验证器<br/>Pydantic"]
    end

    subgraph Orchestration["编排层"]
        RetSvc["RetrievalService<br/>主协调器"]
        QueryExp["QueryExpansionService<br/>查询扩展"]
        Cache["RedisCache<br/>语义缓存"]
    end

    subgraph Retrieval["检索层"]
        VecSvc["VectorService"]
        BM25Svc["BM25Service"]
        EmbSvc["EmbeddingService"]
    end

    subgraph Fusion["融合层"]
        Hybrid["HybridService<br/>RRF 融合 k=60"]
        Rerank["RerankService<br/>Cross-Encoder重排序"]
    end

    subgraph Observability["可观测层"]
        Metrics["Metrics<br/>Prometheus"]
        Logging["Logging<br/>结构化日志"]
        Tracing["Tracing<br/>OpenTelemetry"]
    end

    subgraph Storage["存储层"]
        Redis[("Redis")]
        Milvus[("Milvus")]
        ES[("Elasticsearch")]
    end

    RAG -->|"1. POST /hybrid<br/>query, top_k=5"| FastAPI
    FastAPI -->|"2. 请求进入"| ReqID
    ReqID -->|"3. 生成 request_id"| CORS
    CORS -->|"4. CORS 处理"| Auth
    Auth -->|"5. API Key 验证"| LogMw
    LogMw -->|"6. 记录请求开始"| Router

    Router -->|"7. 验证请求<br/>HybridRequest"| Validator
    Validator -->|"8. 调用检索服务"| RetSvc

    RetSvc -->|"9. 可选查询扩展"| QueryExp
    QueryExp -.->|"扩展查询列表"| RetSvc

    RetSvc -->|"10. 查缓存<br/>MD5(query+params)"| Cache
    Cache -.->|"命中: 返回结果<br/>延迟 ~10ms"| RetSvc
    Cache -->|"未命中"| Redis

    RetSvc -->|"11a. 并行检索<br/>top_k=50"| VecSvc
    RetSvc -->|"11b. 并行检索<br/>top_k=50"| BM25Svc

    VecSvc -->|"12. 向量化查询"| EmbSvc
    EmbSvc -.->|"返回向量 dim=1024"| VecSvc
    VecSvc -->|"13. 向量检索"| Milvus
    Milvus -.->|"向量结果 50条<br/>延迟 ~60ms"| VecSvc

    BM25Svc -->|"14. BM25 检索"| ES
    ES -.->|"BM25结果 50条<br/>延迟 ~35ms"| BM25Svc

    VecSvc -.->|"候选集"| RetSvc
    BM25Svc -.->|"候选集"| RetSvc

    RetSvc -->|"15. RRF融合"| Hybrid
    Hybrid -.->|"融合结果 20条"| RetSvc

    RetSvc -->|"16. 重排序<br/>可选"| Rerank
    Rerank -.->|"Top-K 10条<br/>延迟 ~120ms"| RetSvc

    RetSvc -->|"17. 写缓存<br/>TTL=3600s"| Cache
    Cache -->|"异步存储"| Redis

    RetSvc -.->|"18. 返回结果"| Router
    Router -.->|"19. HTTP 响应"| LogMw
    LogMw -->|"20. 记录指标"| Metrics
    LogMw -->|"21. 记录日志"| Logging
    LogMw -->|"22. 记录 Trace"| Tracing
    LogMw -.->|"23. 响应客户端<br/>总延迟 ~215ms"| RAG

    style Client fill:#e3f2fd
    style HTTP fill:#fff9c4
    style Middleware fill:#ffe0b2
    style API fill:#fff3e0
    style Orchestration fill:#f3e5f5
    style Retrieval fill:#e8f5e9
    style Fusion fill:#ffccbc
    style Observability fill:#e0f7fa
    style Storage fill:#e0f2f1

classDiagram
    class RetrievalService {
        -VectorRetriever vector_retriever
        -BM25Retriever bm25_retriever
        -GraphRetriever graph_retriever
        -CrossEncoderReranker reranker
        -RedisCache cache
        -dict stats
        +initialize()
        +retrieve(query, top_k, mode, tenant_id, filters, rerank)
        -_vector_retrieve()
        -_bm25_retrieve()
        -_graph_retrieve()
        -_hybrid_retrieve()
        -_rrf_fusion()
        -_rerank()
        +get_stats()
    }

    class VectorRetriever {
        -VectorStoreClient vector_store_client
        -BGE_M3_Embedder embedder
        +initialize()
        +retrieve(query, top_k, tenant_id, filters)
        +count()
    }

    class BM25Retriever {
        -list corpus
        -list corpus_ids
        -BM25Okapi bm25
        +initialize()
        +retrieve(query, top_k, tenant_id, filters)
        -_tokenize(text)
        +count()
    }

    class GraphRetriever {
        -Neo4jClient neo4j_client
        +initialize()
        +retrieve(query, top_k, tenant_id, filters)
        +cleanup()
    }

    class CrossEncoderReranker {
        -string model_name
        -CrossEncoder model
        +initialize()
        +rerank(query, documents, top_k)
    }

    class RetrievalDocument {
        +string chunk_id
        +string doc_id
        +string content
        +float score
        +string source
        +dict metadata
        +int rank
    }

    class RRFFusion {
        -int rrf_k
        +fuse_results(vector_docs, bm25_docs, top_k)
    }

    RetrievalService "1" *-- "1" VectorRetriever
    RetrievalService "1" *-- "1" BM25Retriever
    RetrievalService "1" *-- "1" GraphRetriever
    RetrievalService "1" *-- "1" CrossEncoderReranker
    RetrievalService ..> RRFFusion : uses
    RetrievalService ..> RetrievalDocument : returns
    VectorRetriever ..> RetrievalDocument : returns
    BM25Retriever ..> RetrievalDocument : returns
    GraphRetriever ..> RetrievalDocument : returns

sequenceDiagram
    autonumber
    participant Client as RAG Engine
    participant Router as RetrievalRouter
    participant Service as VectorService
    participant Embed as EmbeddingService
    participant BGE as BGE-M3 Model
    participant Retriever as VectorRetriever
    participant Milvus as Milvus DB
    participant Redis as Redis Cache

    Client->>Router: POST /api/v1/retrieval/vector
    Note over Client,Router: {query:"什么是RAG", top_k:5}

    Router->>Router: 验证 VectorRequest
    Router->>Service: vector_search(request)

    alt 未提供向量
        Service->>Embed: embed_query(query)
        Embed->>BGE: encode(query + instruction)
        BGE-->>Embed: embedding [1024 dim]
        Note over BGE: 延迟: ~25ms
        Embed-->>Service: query_embedding
    else 已提供向量
        Note over Service: 跳过向量化
    end

    Service->>Retriever: search(query_embedding, top_k=5)
    Retriever->>Retriever: 构建过滤表达式
    Note over Retriever: tenant_id, metadata filters

    Retriever->>Milvus: search(embedding, limit=5, expr)
    Note over Milvus: HNSW 索引搜索<br/>M=16, ef=100
    Milvus-->>Retriever: hits (5条结果)
    Note over Milvus: 延迟: ~60ms

    Retriever->>Retriever: 转换为 RetrievalDocument
    Note over Retriever: chunk_id, content, score<br/>metadata, source="vector"

    Retriever-->>Service: documents [5条]
    Service-->>Router: VectorResponse
    Note over Service: latency_ms: ~85ms
    Router-->>Client: 200 OK
    Note over Router,Client: {documents:[...], latency_ms:85}

sequenceDiagram
    autonumber
    participant Client as RAG Engine
    participant Router as RetrievalRouter
    participant Service as BM25Service
    participant Retriever as BM25Retriever
    participant Jieba as Jieba Tokenizer
    participant ES as Elasticsearch

    Client->>Router: POST /api/v1/retrieval/bm25
    Note over Client,Router: {query:"Python asyncio", top_k:5}

    Router->>Router: 验证 BM25Request
    Router->>Service: bm25_search(request)

    Service->>Retriever: search(query, top_k=5)

    Retriever->>Jieba: tokenize(query)
    Note over Jieba: 中文分词<br/>精确模式
    Jieba-->>Retriever: tokens ["Python", "asyncio"]
    Note over Jieba: 延迟: ~2ms

    Retriever->>Retriever: 构建 ES 查询
    Note over Retriever: match query + bool filters<br/>tenant_id, custom filters

    Retriever->>ES: search(index, query, size=5)
    Note over ES: BM25Okapi 算法<br/>倒排索引查询
    ES-->>Retriever: hits (5条结果)
    Note over ES: 延迟: ~35ms

    Retriever->>Retriever: 转换为 RetrievalDocument
    Note over Retriever: chunk_id, content, score<br/>source="bm25"

    Retriever-->>Service: documents [5条]
    Service-->>Router: BM25Response
    Note over Service: latency_ms: ~40ms
    Router-->>Client: 200 OK
    Note over Router,Client: {documents:[...], latency_ms:40}

BM25(D, Q) = Σ(IDF(qi) * (f(qi, D) * (k1 + 1)) / (f(qi, D) + k1 * (1 - b + b * |D| / avgdl)))

其中：
- qi: 查询中的第 i 个词
- f(qi, D): 词 qi 在文档 D 中的词频
- |D|: 文档 D 的长度
- avgdl: 语料库中文档的平均长度
- k1: 词频饱和度参数（默认 1.2）
- b: 长度归一化参数（默认 0.75）
- IDF(qi): 逆文档频率 = log((N - n(qi) + 0.5) / (n(qi) + 0.5))

sequenceDiagram
    autonumber
    participant Client as RAG Engine
    participant Router as RetrievalRouter
    participant RetSvc as RetrievalService
    participant Cache as RedisCache
    participant VecSvc as VectorService
    participant BM25Svc as BM25Service
    participant EmbedSvc as EmbeddingService
    participant Milvus as Milvus
    participant ES as Elasticsearch
    participant Hybrid as HybridService
    participant Rerank as RerankService
    participant CrossEnc as Cross-Encoder Model

    Client->>Router: POST /api/v1/retrieval/hybrid
    Note over Client,Router: {query:"RAG系统架构", top_k:5}

    Router->>Router: 验证 HybridRequest
    Router->>RetSvc: hybrid_search(request)

    %% 缓存检查
    RetSvc->>Cache: check_cache(cache_key)
    Note over Cache: MD5(query+top_k+tenant_id)

    alt 缓存命中
        Cache-->>RetSvc: cached_results
        Note over Cache: 命中率: 30-50%<br/>延迟: ~10ms
        RetSvc-->>Router: HybridResponse (cached=true)
        Router-->>Client: 200 OK (10ms)
    else 缓存未命中

        %% 并行检索
        par 向量检索
            RetSvc->>VecSvc: vector_search(query, top_k=10)
            VecSvc->>EmbedSvc: embed_query(query)
            EmbedSvc-->>VecSvc: embedding [1024 dim]
            Note over EmbedSvc: 延迟: ~25ms
            VecSvc->>Milvus: search(embedding, top_k=10)
            Milvus-->>VecSvc: 向量结果 (10条)
            Note over Milvus: HNSW搜索: ~60ms
            VecSvc-->>RetSvc: vector_docs [10条]
        and BM25检索
            RetSvc->>BM25Svc: bm25_search(query, top_k=10)
            BM25Svc->>ES: search(query, size=10)
            ES-->>BM25Svc: BM25结果 (10条)
            Note over ES: 倒排索引: ~35ms
            BM25Svc-->>RetSvc: bm25_docs [10条]
        end

        Note over RetSvc: 并行检索完成<br/>总延迟: max(60ms, 35ms) ≈ 60ms

        %% RRF融合
        RetSvc->>Hybrid: fuse_results(vector_docs, bm25_docs, top_k=5)
        Hybrid->>Hybrid: 计算 RRF 分数
        Note over Hybrid: RRF(d) = Σ 1/(60 + rank_i)<br/>延迟: ~3ms
        Hybrid-->>RetSvc: fused_docs [5条]

        %% 重排序
        alt 启用重排序
            RetSvc->>Rerank: rerank(query, fused_docs, top_k=5)
            Rerank->>CrossEnc: predict([[query, doc1], [query, doc2], ...])
            Note over CrossEnc: Cross-Encoder 推理<br/>batch_size=5
            CrossEnc-->>Rerank: rerank_scores [5个分数]
            Note over CrossEnc: 延迟: ~120ms
            Rerank->>Rerank: 按分数排序
            Rerank-->>RetSvc: reranked_docs [5条]
        else 跳过重排序
            Note over RetSvc: 直接使用 RRF 结果
        end

        %% 写缓存
        RetSvc->>Cache: set_cache(cache_key, results, ttl=3600)
        Note over Cache: 异步写入<br/>不阻塞响应

        %% 返回结果
        RetSvc-->>Router: HybridResponse
        Note over RetSvc: {documents:[...],<br/>vector_count:10,<br/>bm25_count:10,<br/>reranked:true,<br/>latency_ms:205}
        Router-->>Client: 200 OK (205ms)
    end

class RequestIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # 生成或使用现有 request_id
        request_id = request.headers.get("X-Request-ID") or str(uuid.uuid4())
        request.state.request_id = request_id

        # 传递到下游
        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id

        return response

@app.middleware("http")
async def logging_and_metrics_middleware(request: Request, call_next):
    start_time = time.time()
    request_id = getattr(request.state, "request_id", "unknown")

    # 记录请求开始
    logger.bind(request_id=request_id).info(
        f"Request started: {request.method} {request.url.path}",
        extra={"method": request.method, "path": request.url.path}
    )

    try:
        response = await call_next(request)
        duration = time.time() - start_time

        # 记录指标
        metrics.record_request(
            method=request.method,
            endpoint=request.url.path,
            status=response.status_code,
            duration=duration,
        )

        # 记录请求完成
        logger.bind(request_id=request_id).info(
            f"Request completed: {request.method} {request.url.path} "
            f"[{response.status_code}] in {duration:.3f}s"
        )

        return response
    except Exception as e:
        duration = time.time() - start_time
        logger.bind(request_id=request_id).error(f"Request failed: {e}")
        metrics.record_error(error_type=type(e).__name__)
        raise

class AuthMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, api_keys: Set[str]):
        super().__init__(app)
        self.api_keys = api_keys

    async def dispatch(self, request: Request, call_next):
        # 跳过健康检查和指标端点
        if request.url.path in ["/health", "/ready", "/metrics"]:
            return await call_next(request)

        # 提取 API Key
        auth_header = request.headers.get("Authorization", "")
        if not auth_header.startswith("Bearer "):
            return JSONResponse(
                status_code=401,
                content={"detail": "Missing or invalid Authorization header"}
            )

        api_key = auth_header[7:]  # 移除 "Bearer " 前缀

        if api_key not in self.api_keys:
            return JSONResponse(
                status_code=403,
                content={"detail": "Invalid API key"}
            )

        # 验证通过，继续处理
        return await call_next(request)

class RateLimitMiddleware:
    def __init__(self, redis_client, limit: int = 100, window: int = 60):
        """
        Args:
            limit: 时间窗口内允许的请求数
            window: 时间窗口（秒）
        """
        self.redis = redis_client
        self.limit = limit
        self.window = window

    async def __call__(self, request: Request, call_next):
        # 获取限流键（基于租户或IP）
        tenant_id = request.headers.get("X-Tenant-ID")
        key = f"rate_limit:{tenant_id or request.client.host}:{request.url.path}"

        # 获取当前计数
        count = await self.redis.get(key)

        if count and int(count) >= self.limit:
            return JSONResponse(
                status_code=429,
                content={
                    "detail": f"Rate limit exceeded: {self.limit} requests per {self.window}s"
                },
                headers={"Retry-After": str(self.window)}
            )

        # 增加计数
        pipe = self.redis.pipeline()
        pipe.incr(key)
        pipe.expire(key, self.window)
        await pipe.execute()

        return await call_next(request)

rate_limit:
  default_limit: 100  # 默认限流：100请求/分钟
  by_endpoint:
    "/api/v1/retrieval/hybrid": 50  # 混合检索：50请求/分钟
    "/api/v1/retrieval/graph": 20   # 图谱检索：20请求/分钟（计算密集）
  by_tenant:
    premium: 1000   # 付费用户：1000请求/分钟
    free: 100       # 免费用户：100请求/分钟

class IdempotencyMiddleware:
    def __init__(self, redis_client, ttl: int = 86400):
        self.redis = redis_client
        self.ttl = ttl  # 24小时

    async def __call__(self, request: Request, call_next):
        # 仅对POST/PUT/DELETE请求启用
        if request.method not in ["POST", "PUT", "DELETE"]:
            return await call_next(request)

        # 获取幂等键
        idempotency_key = request.headers.get("Idempotency-Key")
        if not idempotency_key:
            return await call_next(request)

        # 检查是否已处理
        cache_key = f"idempotency:{idempotency_key}"
        cached_response = await self.redis.get(cache_key)

        if cached_response:
            # 返回缓存的响应
            return JSONResponse(
                status_code=200,
                content=json.loads(cached_response),
                headers={"X-Idempotency": "replayed"}
            )

        # 处理请求
        response = await call_next(request)

        # 缓存响应（仅成功响应）
        if 200 <= response.status_code < 300:
            response_body = b""
            async for chunk in response.body_iterator:
                response_body += chunk

            await self.redis.setex(
                cache_key,
                self.ttl,
                response_body.decode()
            )

        return response

# 同义词扩展示例
class QueryExpansionService:
    def __init__(self, methods=["synonym", "spelling"], max_expansions=3):
        self.methods = methods
        self.max_expansions = max_expansions
        # 加载同义词词典
        self.synonym_dict = self._load_synonym_dict("data/synonyms.json")

    async def expand(self, query: str, enable_llm: bool = False):
        """
        查询扩展

        Args:
            query: 原始查询
            enable_llm: 是否启用LLM扩展

        Returns:
            扩展后的查询列表及权重
        """
        expanded_queries = [query]  # 原始查询权重1.0
        weights = [1.0]

        # 1. 同义词扩展
        if "synonym" in self.methods:
            synonyms = self._expand_synonyms(query)
            expanded_queries.extend(synonyms)
            weights.extend([0.8] * len(synonyms))

        # 2. 拼写纠正
        if "spelling" in self.methods:
            corrected = self._spell_correction(query)
            if corrected != query:
                expanded_queries.append(corrected)
                weights.append(0.9)

        # 3. LLM扩展（可选，成本较高）
        if enable_llm:
            llm_queries = await self._llm_expansion(query)
            expanded_queries.extend(llm_queries)
            weights.extend([0.7] * len(llm_queries))

        # 限制扩展数量
        return expanded_queries[:self.max_expansions], weights[:self.max_expansions]

场景：100万次查询/月，启用同义词+拼写纠正

召回率提升：
- 原始召回率：85%
- 扩展后召回率：90-92%
- 提升：5-7%

性能影响：
- 延迟增加：7-15ms
- 对总延迟影响：~5%（从215ms到222ms）

成本影响：
- 无额外成本（本地词典）
- 如启用LLM扩展：+$1000-5000/月（取决于调用频率）

用户体验：
- "零结果"查询减少：30-40%
- 用户满意度提升：10-15%

# 推荐配置（生产环境）
query_expansion:
  enabled: true
  methods:
    - synonym      # 同义词扩展（推荐）
    - spelling     # 拼写纠正（推荐）
    # - llm        # LLM扩展（仅高价值查询）
  max_expansions: 3
  synonym_dict_path: "data/synonyms.json"
  llm_model: "gpt-3.5-turbo"  # LLM扩展使用的模型

# VectorService 降级示例
def _connect(self):
    """连接到 Milvus"""
    try:
        connections.connect(...)
        self.collection = Collection(settings.MILVUS_COLLECTION)
        self.collection.load()
        logger.info("Connected to Milvus")
    except Exception as e:
        logger.error(f"Failed to connect to Milvus: {e}")
        # 不抛出异常，允许服务启动（降级模式）
        self.collection = None

async def search(...) -> List[RetrievalDocument]:
    """向量检索"""
    if not self.collection:
        logger.warning("Milvus not connected, returning empty results")
        return []  # 降级为空结果，由混合检索使用BM25补偿
    # ... 正常检索逻辑

# 降级事件指标
metrics.record_degradation(
    degradation_type="vector_search_failed",
    component="milvus",
    severity="warning",
)

# 告警规则
- alert: VectorSearchDegradation
  expr: rate(degradation_events{type="vector_search_failed"}[5m]) > 0.1
  annotations:
    summary: "向量检索降级率超过 10%"

class SearchRequest(BaseModel):
    query: str                      # 查询文本
    top_k: int = 10                 # 返回结果数
    mode: str = "hybrid"            # 检索模式: vector/bm25/graph/hybrid
    tenant_id: Optional[str] = None # 租户ID
    filters: Optional[Dict] = None  # 过滤条件
    rerank: bool = True             # 是否重排序

class SearchResponse(BaseModel):
    results: List[RetrievalDocument]  # 检索结果列表
    total: int                        # 结果总数
    mode: str                         # 实际使用的检索模式
    latency_ms: float                 # 检索延迟（毫秒）
    cached: bool                      # 是否命中缓存

class RetrievalDocument(BaseModel):
    chunk_id: str                     # 文档块ID
    doc_id: str                       # 原始文档ID
    content: str                      # 文档块内容
    score: float                      # 相关性分数
    source: str                       # 检索来源
    metadata: Dict                    # 元数据
    rank: int                         # 排名

async def retrieve(
    self,
    query: str,
    top_k: int = 10,
    mode: str = "hybrid",
    tenant_id: str = None,
    filters: Dict = None,
    rerank: bool = True,
) -> List[Dict]:
    """
    检索接口（核心逻辑）
    """
    self.stats["total_queries"] += 1

    # 1. 检查缓存
    cache_key = self._generate_cache_key(query, top_k, mode, tenant_id)
    cached_results = await self._get_from_cache(cache_key)
    if cached_results is not None:
        self.stats["cache_hits"] += 1
        return cached_results

    self.stats["cache_misses"] += 1

    # 2. 根据模式执行检索
    if mode == "vector":
        results = await self._vector_retrieve(query, top_k, tenant_id, filters)
    elif mode == "bm25":
        results = await self._bm25_retrieve(query, top_k, tenant_id, filters)
    elif mode == "graph":
        results = await self._graph_retrieve(query, top_k, tenant_id, filters)
    elif mode == "hybrid":
        results = await self._hybrid_retrieve(query, top_k, tenant_id, filters)
    else:
        raise ValueError(f"Unknown retrieval mode: {mode}")

    # 3. 重排序
    if rerank and len(results) > 0:
        results = await self._rerank(query, results, top_k)

    # 4. 写入缓存
    await self._set_to_cache(cache_key, results)

    return results

async def _hybrid_retrieve(
    self, query: str, top_k: int, tenant_id: str, filters: Dict
) -> List[Dict]:
    """
    混合检索：并行执行向量和BM25检索，使用RRF融合
    """
    import asyncio

    # 1. 并行执行多种检索（扩大候选集）
    vector_results, bm25_results = await asyncio.gather(
        self._vector_retrieve(query, top_k * 2, tenant_id, filters),
        self._bm25_retrieve(query, top_k * 2, tenant_id, filters),
    )

    # 2. RRF融合
    fused_results = self._rrf_fusion(
        [vector_results, bm25_results],
        top_k=top_k,
    )

    return fused_results

def _rrf_fusion(self, results_list: List[List[Dict]], top_k: int, k: int = 60) -> List[Dict]:
    """
    Reciprocal Rank Fusion (RRF)

    公式: RRF(d) = Σ 1/(k + rank_i(d))

    Args:
        results_list: 多路检索结果
        top_k: 返回结果数
        k: RRF常数（默认60，平衡不同来源权重）
    """
    # 1. 计算每个文档的RRF分数
    doc_scores = {}

    for results in results_list:
        for rank, result in enumerate(results, 1):
            doc_id = result.get("chunk_id") or result.get("doc_id")

            if doc_id not in doc_scores:
                doc_scores[doc_id] = {
                    "score": 0,
                    "data": result,
                }

            # RRF公式：rank从1开始，k平衡不同来源权重
            doc_scores[doc_id]["score"] += 1 / (k + rank)

    # 2. 按RRF分数排序
    sorted_docs = sorted(
        doc_scores.items(),
        key=lambda x: x[1]["score"],
        reverse=True,
    )

    # 3. 取Top K并标注RRF分数
    results = []
    for doc_id, doc_info in sorted_docs[:top_k]:
        result = doc_info["data"].copy()
        result["rrf_score"] = doc_info["score"]
        results.append(result)

    return results

async def _rerank(self, query: str, results: List[Dict], top_k: int) -> List[Dict]:
    """
    使用Cross-Encoder重排序

    Cross-Encoder直接计算查询-文档对的相关性分数，
    相比向量相似度（Bi-Encoder），准确性更高但计算成本也更高。

    策略：先用快速的向量/BM25检索获取候选集（top_k*2），
    再用Cross-Encoder精排得到最终top_k结果。
    """
    if not results:
        return results

    # 调用CrossEncoderReranker
    reranked = await self.reranker.rerank(
        query=query,
        documents=results,
        top_k=top_k,
    )

    return reranked

# CrossEncoderReranker.rerank实现
async def rerank(
    self, query: str, documents: List[Dict], top_k: int = 10
) -> List[Dict]:
    """重排序文档"""
    if not documents:
        return []

    # 1. 提取文档内容
    doc_contents = [doc.get("content", "") for doc in documents]

    # 2. 构建查询-文档对
    query_doc_pairs = [[query, doc] for doc in doc_contents]

    # 3. 计算相关性分数（Cross-Encoder模型预测）
    scores = self.model.predict(query_doc_pairs)

    # 4. 将分数添加到文档
    for doc, score in zip(documents, scores):
        doc["rerank_score"] = float(score)

    # 5. 按分数排序并取Top K
    reranked = sorted(documents, key=lambda x: x["rerank_score"], reverse=True)
    top_results = reranked[:top_k]

    return top_results

sequenceDiagram
    autonumber
    participant Client as RAG Engine
    participant HTTP as HTTP Server
    participant Service as RetrievalService
    participant Cache as Redis Cache
    participant Vector as VectorRetriever
    participant BM25 as BM25Retriever
    participant Milvus as Milvus
    participant Reranker as CrossEncoderReranker

    Client->>HTTP: POST /api/v1/retrieval/search
    Note over Client,HTTP: query="什么是RAG", mode="hybrid", top_k=5

    HTTP->>Service: retrieve(query, top_k=5, mode="hybrid")

    Service->>Service: generate_cache_key(query, top_k, mode)
    Service->>Cache: get(cache_key)

    alt 缓存命中
        Cache-->>Service: cached_results
        Service-->>HTTP: results (cached)
        HTTP-->>Client: 200 OK, results
    else 缓存未命中
        Cache-->>Service: None

        Note over Service: 执行混合检索

        par 并行检索（扩大候选集）
            Service->>Vector: retrieve(query, top_k=10)
            Vector->>Vector: embedder.embed_query(query)
            Vector->>Milvus: search(query_vector, top_k=10)
            Milvus-->>Vector: vector_results (10条)
            Vector-->>Service: vector_results
        and
            Service->>BM25: retrieve(query, top_k=10)
            BM25->>BM25: tokenize(query) [jieba分词]
            BM25->>BM25: bm25.get_scores(tokens)
            BM25-->>Service: bm25_results (10条)
        end

        Note over Service: RRF融合
        Service->>Service: rrf_fusion([vector_results, bm25_results], top_k=5)
        Note over Service: 计算RRF分数：<br/>RRF(d) = Σ 1/(60 + rank)

        Service->>Reranker: rerank(query, fused_results, top_k=5)
        Reranker->>Reranker: model.predict([[query, doc1], [query, doc2], ...])
        Note over Reranker: Cross-Encoder精排
        Reranker-->>Service: reranked_results (5条)

        Service->>Cache: set(cache_key, results, ttl=3600)
        Cache-->>Service: ok

        Service-->>HTTP: results
        HTTP-->>Client: 200 OK, results
    end

class RetrievalService:
    """检索编排服务 - 协调所有检索模块"""

    def __init__(self):
        # 初始化核心检索服务
        self.vector_service = VectorService()           # 向量检索
        self.bm25_service = BM25Service()               # BM25检索
        self.hybrid_service = HybridService(rrf_k=60)   # RRF融合
        self.rerank_service = RerankService()           # 重排序
        self.embedding_service = EmbeddingService()     # 向量化

        # 可选的图谱检索（根据配置启用）
        self.neo4j_client = None
        self.graph_service = None
        self.hybrid_graph_service = None

        if settings.ENABLE_GRAPH_RETRIEVAL:
            self.neo4j_client = Neo4jClient(...)
            self.graph_service = GraphRetrievalService(self.neo4j_client)
            self.hybrid_graph_service = HybridGraphService(...)

async def search(
    self,
    query_embedding: List[float],
    top_k: int,
    tenant_id: Optional[str] = None,
    filters: Optional[Dict[str, Any]] = None,
) -> List[RetrievalDocument]:
    """向量检索"""
    if not self.collection:
        logger.warning("Milvus not connected, returning empty results")
        return []  # 降级处理

    # 1. 构建过滤表达式（租户隔离）
    expr = self._build_expression(tenant_id, filters)

    # 2. 配置搜索参数（HNSW索引）
    search_params = {
        "metric_type": "IP",        # Inner Product (Cosine相似度)
        "params": {"ef": 128},      # HNSW搜索候选数
    }

    # 3. 执行检索
    results = self.collection.search(
        data=[query_embedding],
        anns_field="embedding",
        param=search_params,
        limit=top_k,
        expr=expr,
        output_fields=["chunk_id", "document_id", "content", "metadata"],
    )

    # 4. 转换结果格式
    documents = []
    for hits in results:
        for hit in hits:
            doc = RetrievalDocument(
                id=hit.entity.get("document_id", ""),
                chunk_id=hit.entity.get("chunk_id", ""),
                content=hit.entity.get("content", ""),
                score=float(hit.score),  # IP分数，范围约0-1
                metadata=hit.entity.get("metadata", {}),
                source="vector",
            )
            documents.append(doc)

    return documents

def _build_expression(
    self, tenant_id: Optional[str] = None, filters: Optional[Dict] = None
) -> str:
    """构建Milvus布尔表达式"""
    expressions = []

    # 租户过滤（必须）
    if tenant_id:
        expressions.append(f'tenant_id == "{tenant_id}"')

    # 自定义过滤
    if filters:
        for key, value in filters.items():
            if isinstance(value, str):
                expressions.append(f'{key} == "{value}"')
            elif isinstance(value, (int, float)):
                expressions.append(f"{key} == {value}")
            elif isinstance(value, list):  # IN 操作
                values_str = ", ".join([f'"{v}"' for v in value])
                expressions.append(f"{key} in [{values_str}]")

    return " && ".join(expressions) if expressions else ""

# 单租户
'tenant_id == "tenant_123"'

# 租户+集合过滤
'tenant_id == "tenant_123" && collection_id == "docs_2024"'

# 多值过滤
'tenant_id == "tenant_123" && doc_type in ["pdf", "docx"]'

async def fuse_results(
    self,
    vector_docs: List[RetrievalDocument],
    bm25_docs: List[RetrievalDocument],
    top_k: int,
) -> List[RetrievalDocument]:
    """
    RRF (Reciprocal Rank Fusion) 融合算法

    公式: RRF(d) = Σ 1/(k + rank_i(d))

    参数:
        - k=60: 平衡常数（论文推荐值）
        - rank_i: 文档在第i路检索中的排名（1-based）
    """
    rrf_scores: Dict[str, float] = {}
    doc_map: Dict[str, RetrievalDocument] = {}

    # 向量检索贡献
    for rank, doc in enumerate(vector_docs, start=1):
        key = doc.chunk_id
        rrf_scores[key] = rrf_scores.get(key, 0.0) + (1.0 / (self.rrf_k + rank))
        if key not in doc_map:
            doc_map[key] = doc

    # BM25检索贡献
    for rank, doc in enumerate(bm25_docs, start=1):
        key = doc.chunk_id
        rrf_scores[key] = rrf_scores.get(key, 0.0) + (1.0 / (self.rrf_k + rank))
        if key not in doc_map:
            doc_map[key] = doc

    # 按RRF分数排序
    sorted_keys = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)

    # 构建融合结果
    fused_docs = []
    for key, score in sorted_keys[:top_k]:
        doc = doc_map[key]
        doc.score = score      # 更新为RRF分数
        doc.source = "hybrid"  # 标记来源
        fused_docs.append(doc)

    return fused_docs

# 假设：
vector_docs = [doc_A(rank=1), doc_B(rank=2), doc_C(rank=5)]
bm25_docs = [doc_B(rank=1), doc_C(rank=2), doc_D(rank=3)]

# RRF计算：
doc_A: 1/(60+1) = 0.0164
doc_B: 1/(60+2) + 1/(60+1) = 0.0161 + 0.0164 = 0.0325
doc_C: 1/(60+5) + 1/(60+2) = 0.0154 + 0.0161 = 0.0315
doc_D: 1/(60+3) = 0.0159

# 排序结果：
doc_B (0.0325) > doc_C (0.0315) > doc_A (0.0164) > doc_D (0.0159)

async def _rerank_with_cross_encoder(
    self, query: str, documents: List[RetrievalDocument], top_k: int
) -> List[RetrievalDocument]:
    """使用Cross-Encoder重排序"""
    if not self.cross_encoder:
        return documents[:top_k]  # 降级处理

    # 1. 构建查询-文档对
    pairs = [[query, doc.content] for doc in documents]

    # 2. 在线程池中执行推理（避免阻塞事件循环）
    loop = asyncio.get_event_loop()
    scores = await loop.run_in_executor(None, self.cross_encoder.predict, pairs)

    # 3. 更新分数并排序
    for doc, score in zip(documents, scores):
        doc.score = float(score)

    reranked_docs = sorted(documents, key=lambda x: x.score, reverse=True)[:top_k]

    return reranked_docs

query_vec = encode("什么是RAG?")           # [0.1, 0.3, ..., 0.5]
doc_vec = encode("RAG是一种检索增强生成技术") # [0.2, 0.4, ..., 0.6]
score = cosine(query_vec, doc_vec)        # 独立编码，无交互

input = "[CLS] 什么是RAG? [SEP] RAG是一种检索增强生成技术 [SEP]"
score = model(input)  # 联合编码，捕获查询-文档交互信息

# 1. 批量推理
batch_size = 10
for i in range(0, len(pairs), batch_size):
    batch = pairs[i:i+batch_size]
    scores.extend(self.cross_encoder.predict(batch))

# 2. FP16量化（需GPU）
self.cross_encoder = CrossEncoder(model_name, device="cuda")
# 自动使用混合精度，速度提升50%

# 3. 模型蒸馏（离线）
# 用大模型（bge-reranker-large）标注数据
# 训练小模型（distilbert-reranker），速度提升100%

总延迟 120ms = Tokenization 5ms + Forward Pass 100ms + Post-processing 15ms

async def hybrid_search(self, request: HybridRequest) -> HybridResponse:
    """混合检索"""
    # 1. 并行执行向量和BM25检索
    vector_task = self.vector_service.search(
        query_embedding=query_embedding,
        top_k=settings.VECTOR_TOP_K,  # 50
        tenant_id=request.tenant_id,
        filters=request.filters,
    )
    bm25_task = self.bm25_service.search(
        query=request.query,
        top_k=settings.BM25_TOP_K,  # 50
        tenant_id=request.tenant_id,
        filters=request.filters,
    )

    # 并发等待（asyncio.gather）
    vector_docs, bm25_docs = await asyncio.gather(vector_task, bm25_task)

    # 2. RRF融合
    fused_docs = await self.hybrid_service.fuse_results(
        vector_docs=vector_docs,
        bm25_docs=bm25_docs,
        top_k=request.top_k,  # 20
    )

    # 3. 重排序
    if request.enable_rerank:
        fused_docs = await self.rerank_service.rerank(
            query=request.query,
            documents=fused_docs,
            top_k=request.rerank_top_k,  # 10
        )

    return HybridResponse(documents=fused_docs, ...)

串行执行:
|----Vector(80ms)----|----BM25(40ms)----|----RRF(5ms)----|----Rerank(120ms)----|
Total: 245ms

并行执行:
|----Vector(80ms)----|
|----BM25(40ms)------|
                      |----RRF(5ms)----|----Rerank(120ms)----|
Total: max(80, 40) + 5 + 120 = 205ms

性能提升: (245-205)/245 = 16.3%

# 底层机制（简化）
async def gather(*tasks):
    results = []
    pending = set(tasks)

    while pending:
        # 等待任何一个任务完成
        done, pending = await wait(pending, return_when=FIRST_COMPLETED)
        for task in done:
            results.append(task.result())

    return results

# 单租户
expr = 'tenant_id == "tenant_123"'

# 租户+时间范围
expr = 'tenant_id == "tenant_123" && created_at > 1704067200'

# 租户+集合+文档类型
expr = 'tenant_id == "tenant_123" && collection_id == "docs_2024" && doc_type in ["pdf", "docx"]'

{
    "query": {
        "bool": {
            "must": [
                {"match": {"content": "查询文本"}}
            ],
            "filter": [
                {"term": {"tenant_id": "tenant_123"}},
                {"terms": {"doc_type": ["pdf", "docx"]}}
            ]
        }
    }
}

MATCH (e:Entity)-[r:RELATED_TO]-(related:Entity)
WHERE e.tenant_id = $tenant_id AND e.name = $entity_name
RETURN related
LIMIT $top_k

sequenceDiagram
    autonumber
    participant Svc as HybridService
    participant RRF as RRF Algorithm
    participant DocMap as Document Map

    Svc->>RRF: fuse_results(vector_docs, bm25_docs, top_k)
    Note over Svc,RRF: vector_docs: 10条<br/>bm25_docs: 10条<br/>top_k: 5

    RRF->>RRF: 初始化 doc_scores = {}

    loop 遍历向量检索结果
        RRF->>RRF: 获取 doc_id, rank
        RRF->>RRF: 计算 RRF 分数
        Note over RRF: score += 1/(60 + rank)<br/>rank 从 1 开始
        RRF->>DocMap: 更新/创建 doc_scores[doc_id]
    end

    loop 遍历 BM25 检索结果
        RRF->>RRF: 获取 doc_id, rank
        RRF->>RRF: 累加 RRF 分数
        Note over RRF: score += 1/(60 + rank)
        RRF->>DocMap: 更新 doc_scores[doc_id]
    end

    RRF->>RRF: 按 RRF 分数排序
    Note over RRF: sorted(doc_scores,<br/>key=score, reverse=True)

    RRF->>RRF: 取 Top-K (5条)

    loop 构建融合结果
        RRF->>DocMap: 获取文档数据
        RRF->>RRF: 添加 rrf_score 字段
        RRF->>RRF: 设置 source="hybrid"
    end

    RRF-->>Svc: fused_docs [5条]
    Note over RRF,Svc: 延迟: ~3ms

RRF(d) = Σ 1/(k + rank_i(d))

其中：
- d: 文档
- rank_i(d): 文档 d 在第 i 路检索结果中的排名（1-based）
- k: 常数，默认 60（论文推荐值）

假设：
- 向量检索结果: [doc_A(rank=1), doc_B(rank=2), doc_C(rank=3)]
- BM25 检索结果: [doc_B(rank=1), doc_D(rank=2), doc_A(rank=3)]

计算 RRF 分数：
doc_A: 1/(60+1) + 1/(60+3) = 0.0164 + 0.0159 = 0.0323
doc_B: 1/(60+2) + 1/(60+1) = 0.0161 + 0.0164 = 0.0325
doc_C: 1/(60+3) = 0.0159
doc_D: 1/(60+2) = 0.0161

排序: doc_B > doc_A > doc_D > doc_C

sequenceDiagram
    autonumber
    participant Svc as RerankService
    participant CE as Cross-Encoder Model
    participant Batch as Batch Processor

    Svc->>Svc: rerank(query, documents, top_k)
    Note over Svc: documents: 5条<br/>top_k: 5

    Svc->>Svc: 提取文档内容
    loop 遍历文档
        Svc->>Svc: doc_contents.append(doc.content)
    end

    Svc->>Svc: 构建查询-文档对
    Note over Svc: pairs = [[query, doc1],<br/>[query, doc2],<br/>...]

    Svc->>Batch: batch_processor(pairs, batch_size=5)

    Batch->>CE: model.encode(pairs)
    Note over CE: Tokenization<br/>BERT 编码<br/>CLS token
    CE->>CE: forward pass
    Note over CE: 12层 Transformer<br/>参数: 270M
    CE->>CE: 分类头（相关性预测）
    Note over CE: sigmoid(linear(CLS))
    CE-->>Batch: scores [5个分数]
    Note over CE: 延迟: ~120ms<br/>GPU: NVIDIA T4

    Batch-->>Svc: scores

    Svc->>Svc: 添加 rerank_score
    loop 遍历文档和分数
        Svc->>Svc: doc["rerank_score"] = score
    end

    Svc->>Svc: 按 rerank_score 排序
    Note over Svc: sorted(documents,<br/>key=rerank_score,<br/>reverse=True)

    Svc->>Svc: 取 Top-K
    Svc-->>Svc: reranked_docs [5条]