Week 21复习 + RAG v2整合——从0.864到0.948的完整调优路径

日期: 2026-09-19 方向: AI系统工程 / RAG 阶段: Phase 3 - RAG高级模式 (Day 135-148) 标签: #RAG #整合 #调优方法论 #ProductionReady

今日目标

核心结论预告：v1 (baseline) Recall@5 = 0.864 → v2 (full stack) = 0.948。每一层平均贡献 ~2-4% lift，ensemble together is greater than sum of parts。

一、Week 21知识图谱

┌──────────────────────────── RAG高级模式 ─────────────────────────────┐
│                                                                       │
│  Day 135: 基础架构                                                     │
│   └─► 6步流水线 (load→chunk→embed→index→retrieve→generate)            │
│   └─► rag_v1.py: OpenAI embed + Chroma + Claude Sonnet                │
│   └─► Baseline Recall@5 = 0.864                                       │
│                                                                       │
│  Day 136: Embedding选型                                                │
│   └─► OpenAI 3-large最稳, Voyage-3最准, BGE性价比                      │
│   └─► 金融query OpenAI 3-large为baseline                               │
│                                                                       │
│  Day 137: Vector DB选型                                                │
│   └─► Chroma (proto), Qdrant (perf), Pinecone (zero-ops)              │
│   └─► v2选Qdrant (filter快, 自部署或Cloud)                             │
│                                                                       │
│  Day 138: Hybrid Search                                                │
│   └─► BM25 + Dense + RRF融合                                          │
│   └─► +5-6% Recall, 关键term找回率显著提升                             │
│                                                                       │
│  Day 139: Reranking                                                    │
│   └─► bge-reranker-v2-m3最佳ROI                                        │
│   └─► +4-5% Recall, top-50→top-5精度暴涨                              │
│                                                                       │
│  Day 140: Query Rewrite                                                │
│   └─► term_expand (always) + Multi-Query async                        │
│   └─► +5-8% Recall, 短/模糊query救回                                  │
│                                                                       │
└───────────────────────────────────────────────────────────────────────┘

二、调优方法论：先做什么后做什么？

2.1 黄金顺序

[1] 数据质量 (chunking + cleaning)         ← 70%问题在这
       │
       ▼
[2] Embedding 模型选型                       ← 20%
       │
       ▼
[3] Hybrid Search (BM25 + Dense)            ← +5-6%
       │
       ▼
[4] Reranking (bge or Cohere)               ← +4-5%
       │
       ▼
[5] Query Rewrite (term + multi-query)      ← +5-8%
       │
       ▼
[6] Vector DB优化 (HNSW参数调优)             ← <1% (after the above)

关键洞察：Vector DB的微调（HNSW的M、ef）通常是 最后做，因为它的效果在前几步做完后才显现。新手最常犯的错是首先去调DB参数。

2.2 何时止步？

                Recall@5
                  │
        0.95 ━━━┓│                           ← excellent
                ┃▼
        0.90 ━━┃━━┓                          ← good
                  ┃▼
        0.85 ━━━━━━━┓                        ← acceptable
                    ┃▼
        0.80 ━━━━━━━━━━┓                    ← needs work
                       ┃▼
        0.70 ━━━━━━━━━━━━━━┓                 ← broken
                            ▼
   ─ baseline ─ +chunking ─ +hybrid ─ +rerank ─ +query_rw ─ ...

经验阈值：

• Recall@5 ≥ 0.95：production-ready
• 0.90-0.95：acceptable for B2C，B2B继续优化
• <0.90：need major work

三、rag_v2.py完整代码

"""
rag_v2.py — Production-grade RAG with all Week 21 optimizations
组件：
  - Loader: PyPDF + Unstructured fallback
  - Chunking: Recursive token-based with overlap
  - Embedding: OpenAI text-embedding-3-large
  - Vector DB: Qdrant (with payload index)
  - BM25: rank-bm25 in-memory
  - Hybrid: RRF fusion
  - Rerank: bge-reranker-v2-m3
  - Query Rewrite: term expansion (always) + multi-query (optional)
  - Generator: Claude Sonnet 4.5 with citation

依赖：
  pip install qdrant-client openai anthropic rank-bm25 sentence-transformers \
              pypdf tiktoken nltk python-dotenv
"""
import os
import time
import asyncio
import hashlib
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from pathlib import Path
import json

import numpy as np
from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue
)
from openai import OpenAI
from anthropic import Anthropic
from rank_bm25 import BM25Okapi
from sentence_transformers import CrossEncoder
from pypdf import PdfReader
import tiktoken
import nltk

nltk.download("punkt_tab", quiet=True)
nltk.download("stopwords", quiet=True)
from nltk.corpus import stopwords
from nltk.tokenize import word_tokenize

STOPWORDS = set(stopwords.words("english"))
encoder = tiktoken.encoding_for_model("gpt-4")
openai_client = OpenAI()
anthropic_client = Anthropic()
qdrant_client = QdrantClient(url=os.environ.get("QDRANT_URL", "http://localhost:6333"))
bge_reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")


# ============================================================
# 配置
# ============================================================
@dataclass
class RAGConfig:
    embed_model: str = "text-embedding-3-large"
    embed_dim: int = 3072
    chunk_size: int = 800
    chunk_overlap: int = 100
    collection: str = "rag_v2"
    initial_top_k: int = 50
    final_top_k: int = 5
    use_rerank: bool = True
    use_multi_query: bool = True
    use_term_expand: bool = True
    use_hybrid: bool = True
    llm_model: str = "claude-sonnet-4-5-20250929"


# ============================================================
# 1. Loader + Chunker
# ============================================================
@dataclass
class Chunk:
    chunk_id: str
    text: str
    metadata: Dict


def load_and_chunk(path: str, cfg: RAGConfig) -> List[Chunk]:
    reader = PdfReader(path)
    full_text = ""
    page_offsets = []
    for i, page in enumerate(reader.pages):
        page_offsets.append(len(full_text))
        full_text += (page.extract_text() or "") + "\n"

    tokens = encoder.encode(full_text)
    doc_id = hashlib.md5(path.encode()).hexdigest()[:12]
    chunks = []
    start = 0
    chunk_idx = 0
    while start < len(tokens):
        end = min(start + cfg.chunk_size, len(tokens))
        chunk_tokens = tokens[start:end]
        chunk_text = encoder.decode(chunk_tokens)

        # 估计chunk所在page
        char_offset = len(encoder.decode(tokens[:start]))
        page_num = sum(1 for o in page_offsets if o <= char_offset)

        chunks.append(Chunk(
            chunk_id=f"{doc_id}_c{chunk_idx}",
            text=chunk_text,
            metadata={
                "source": Path(path).name,
                "doc_id": doc_id,
                "chunk_index": chunk_idx,
                "page": page_num,
            }
        ))
        chunk_idx += 1
        if end == len(tokens):
            break
        start += cfg.chunk_size - cfg.chunk_overlap
    return chunks


# ============================================================
# 2. Indexer (Qdrant + BM25)
# ============================================================
@dataclass
class HybridIndex:
    chunks: List[Chunk]
    bm25: BM25Okapi
    config: RAGConfig


def tokenize(text: str) -> List[str]:
    text_lower = text.lower()
    tokens = word_tokenize(text_lower)
    return [t for t in tokens
            if (t.isalnum() and t not in STOPWORDS and len(t) > 1)
            or t.replace(".", "").replace(",", "").isdigit()]


def index_chunks(chunks: List[Chunk], cfg: RAGConfig) -> HybridIndex:
    # Qdrant
    try:
        qdrant_client.delete_collection(cfg.collection)
    except: pass
    qdrant_client.create_collection(
        collection_name=cfg.collection,
        vectors_config=VectorParams(size=cfg.embed_dim, distance=Distance.COSINE),
    )

    # Embed batch
    print(f"Embedding {len(chunks)} chunks...")
    BATCH = 100
    points = []
    for i in range(0, len(chunks), BATCH):
        batch = chunks[i:i + BATCH]
        resp = openai_client.embeddings.create(
            model=cfg.embed_model, input=[c.text for c in batch]
        )
        embeds = [d.embedding for d in resp.data]
        for chunk, emb in zip(batch, embeds):
            points.append(PointStruct(
                id=hash(chunk.chunk_id) % (2**63),
                vector=emb,
                payload={**chunk.metadata, "text": chunk.text,
                         "chunk_id": chunk.chunk_id},
            ))
    qdrant_client.upsert(collection_name=cfg.collection, points=points)

    # BM25
    tokenized = [tokenize(c.text) for c in chunks]
    bm25 = BM25Okapi(tokenized, k1=1.5, b=0.75)

    return HybridIndex(chunks=chunks, bm25=bm25, config=cfg)


# ============================================================
# 3. Term Expansion
# ============================================================
FINANCE_DICT = {
    "AAPL": ["Apple Inc.", "Apple"], "MSFT": ["Microsoft"],
    "NVDA": ["NVIDIA"], "TSLA": ["Tesla"], "JPM": ["JPMorgan"],
    "FCF": ["free cash flow"], "EPS": ["earnings per share"],
    "MOIC": ["multiple on invested capital"],
    "10-K": ["annual report"], "10-Q": ["quarterly report"],
    "MD&A": ["management discussion and analysis"],
    "Fed": ["Federal Reserve"], "FOMC": ["Federal Open Market Committee"],
    "EBITDA": ["earnings before interest, taxes, depreciation, and amortization"],
    "YoY": ["year-over-year"], "QoQ": ["quarter-over-quarter"],
}


def term_expand(query: str) -> str:
    import re
    expanded = query
    for term, exps in FINANCE_DICT.items():
        if re.search(r"\b" + re.escape(term) + r"\b", expanded, re.I):
            for e in exps:
                expanded += f" ({e})"
    return expanded


# ============================================================
# 4. Multi-Query (async)
# ============================================================
MQ_PROMPT = """Generate 4 alternative phrasings of this financial question
covering different angles. Output ONLY a JSON array of 4 strings.

Original: {query}

Output:"""


def multi_query_variants(query: str) -> List[str]:
    resp = anthropic_client.messages.create(
        model="claude-haiku-4-5-20250929",
        max_tokens=400,
        messages=[{"role": "user", "content": MQ_PROMPT.format(query=query)}],
    )
    text = resp.content[0].text.strip()
    try:
        variants = json.loads(text[text.index("["):text.rindex("]") + 1])
    except Exception:
        variants = []
    return [query] + variants[:4]


# ============================================================
# 5. Hybrid Retrieval
# ============================================================
def embed_query(text: str, cfg: RAGConfig) -> List[float]:
    return openai_client.embeddings.create(
        model=cfg.embed_model, input=[text]
    ).data[0].embedding


def hybrid_retrieve(idx: HybridIndex, query: str, top_k: int = 50,
                    filter_meta: Optional[Dict] = None) -> List[Tuple[str, str, float]]:
    """Returns [(chunk_id, text, score)]"""
    cfg = idx.config

    # BM25
    bm25_scores = idx.bm25.get_scores(tokenize(query))
    top_bm25_indices = np.argsort(bm25_scores)[::-1][:top_k]
    bm25_results = [
        (idx.chunks[i].chunk_id, idx.chunks[i].text, rank + 1)
        for rank, i in enumerate(top_bm25_indices) if bm25_scores[i] > 0
    ]

    # Dense
    q_emb = embed_query(query, cfg)
    qf = None
    if filter_meta:
        conditions = [FieldCondition(key=k, match=MatchValue(value=v))
                      for k, v in filter_meta.items()]
        qf = Filter(must=conditions)
    dense_results = qdrant_client.search(
        collection_name=cfg.collection,
        query_vector=q_emb,
        limit=top_k, query_filter=qf,
    )
    dense_pairs = [
        (r.payload["chunk_id"], r.payload["text"], rank + 1)
        for rank, r in enumerate(dense_results)
    ]

    # RRF
    rrf = {}
    for cid, text, rank in bm25_results:
        rrf[cid] = rrf.get(cid, {"text": text, "score": 0})
        rrf[cid]["score"] += 1 / (60 + rank)
    for cid, text, rank in dense_pairs:
        rrf[cid] = rrf.get(cid, {"text": text, "score": 0})
        rrf[cid]["score"] += 1 / (60 + rank)

    sorted_items = sorted(rrf.items(), key=lambda x: -x[1]["score"])
    return [(cid, v["text"], v["score"]) for cid, v in sorted_items[:top_k]]


# ============================================================
# 6. Multi-Query Hybrid Retrieval (async)
# ============================================================
async def multi_query_hybrid(idx: HybridIndex, query: str,
                              top_k: int = 50) -> List[Tuple[str, str, float]]:
    variants = multi_query_variants(query)

    loop = asyncio.get_event_loop()
    tasks = [
        loop.run_in_executor(None, hybrid_retrieve, idx, v, top_k)
        for v in variants
    ]
    all_results = await asyncio.gather(*tasks)

    # RRF across variants
    rrf = {}
    for results in all_results:
        for rank, (cid, text, _score) in enumerate(results):
            rrf[cid] = rrf.get(cid, {"text": text, "score": 0})
            rrf[cid]["score"] += 1 / (60 + rank)

    sorted_items = sorted(rrf.items(), key=lambda x: -x[1]["score"])
    return [(cid, v["text"], v["score"]) for cid, v in sorted_items[:top_k]]


# ============================================================
# 7. Rerank
# ============================================================
def bge_rerank(query: str, candidates: List[Tuple[str, str, float]],
               top_k: int = 5) -> List[Tuple[str, str, float]]:
    pairs = [(query, c[1]) for c in candidates]
    scores = bge_reranker.predict(pairs, batch_size=32, show_progress_bar=False)
    indexed = [(candidates[i], float(scores[i])) for i in range(len(candidates))]
    indexed.sort(key=lambda x: -x[1])
    return [(c[0], c[1], float(s)) for (c, s) in indexed[:top_k]]


# ============================================================
# 8. Generation
# ============================================================
SYSTEM = """You are a senior financial analyst. Answer strictly based on the
provided CONTEXT. Cite source filename and chunk_id for every claim.
If insufficient context, say "I cannot find this in the documents."

Format:
ANSWER: <answer>
CITATIONS: [<source>:<chunk_id>], ..."""


def generate(query: str, retrieved: List[Tuple[str, str, float]],
             cfg: RAGConfig) -> str:
    context = "\n\n---\n\n".join(
        f"[chunk_id={cid} | score={score:.3f}]\n{text}"
        for cid, text, score in retrieved
    )
    user_msg = f"CONTEXT:\n{context}\n\nQUESTION: {query}"
    resp = anthropic_client.messages.create(
        model=cfg.llm_model, max_tokens=1024,
        system=SYSTEM, messages=[{"role": "user", "content": user_msg}],
    )
    return resp.content[0].text


# ============================================================
# 9. End-to-End RAG v2
# ============================================================
async def rag_v2_query(idx: HybridIndex, query: str, cfg: RAGConfig) -> Dict:
    t0 = time.time()

    # Step 1: term expand
    if cfg.use_term_expand:
        rewritten = term_expand(query)
    else:
        rewritten = query

    # Step 2: retrieve (with multi-query if enabled)
    if cfg.use_multi_query:
        candidates = await multi_query_hybrid(idx, rewritten, top_k=cfg.initial_top_k)
    else:
        candidates = hybrid_retrieve(idx, rewritten, top_k=cfg.initial_top_k)
    t_retr = time.time()

    # Step 3: rerank
    if cfg.use_rerank:
        top = bge_rerank(query, candidates, top_k=cfg.final_top_k)
    else:
        top = candidates[:cfg.final_top_k]
    t_rerank = time.time()

    # Step 4: generate
    answer = generate(query, top, cfg)
    t_gen = time.time()

    return {
        "query": query, "answer": answer,
        "chunks": [{"id": c[0], "score": c[2], "preview": c[1][:200]} for c in top],
        "latency": {
            "retrieve_ms": round((t_retr - t0) * 1000, 1),
            "rerank_ms": round((t_rerank - t_retr) * 1000, 1),
            "generate_ms": round((t_gen - t_rerank) * 1000, 1),
            "total_ms": round((t_gen - t0) * 1000, 1),
        }
    }


# ============================================================
# 10. Demo
# ============================================================
async def main():
    cfg = RAGConfig()

    # Index documents
    all_chunks = []
    for path in ["data/apple_10k_2024.pdf", "data/tesla_10k_2024.pdf",
                 "data/jpmorgan_2024_annual.pdf"]:
        if Path(path).exists():
            all_chunks.extend(load_and_chunk(path, cfg))
    idx = index_chunks(all_chunks, cfg)

    # Test queries
    queries = [
        "What was AAPL Q4 2024 services revenue?",
        "Compare Apple and Tesla R&D as percent of revenue",
        "JPM Tier 1 capital ratio 2024",
        "What AI risks does Apple disclose?",
    ]

    for q in queries:
        result = await rag_v2_query(idx, q, cfg)
        print("\n" + "=" * 80)
        print(f"Q: {q}")
        print(f"A: {result['answer'][:400]}...")
        print(f"   {result['latency']}")


if __name__ == "__main__":
    asyncio.run(main())

四、v1 vs v2 端到端对比

4.1 在50对金融query的全面对比

观察：

• v2 (full stack) 比v1 +9.7% Recall, +11% MRR
• Latency只增加 200ms（rerank和multi-query都优化了）
• Cost 增加 ~40% 但仍 <$0.03/query

4.2 按query类型分层

关键发现：v2在 病态query (短/模糊/缩写) 上提升最大。

五、生产部署架构

                    ┌──────────────────┐
                    │   API Gateway    │
                    │  (rate limit,    │
                    │   auth, cache)   │
                    └────────┬─────────┘
                             ▼
                  ┌─────────────────────┐
                  │   RAG Orchestrator  │
                  │   (FastAPI async)   │
                  └────────┬────────────┘
                           │
        ┌──────────────────┼──────────────────┐
        ▼                  ▼                   ▼
┌──────────────┐   ┌──────────────┐   ┌──────────────┐
│ Query Rewrite│   │ Hybrid       │   │  Reranker    │
│ (Haiku)      │   │ Retrieval    │   │  (bge GPU    │
│              │   │              │   │   service)   │
│ Multi-Query  │   │ ┌──────────┐ │   │              │
│ Term Expand  │   │ │  Qdrant  │ │   └──────┬───────┘
└──────────────┘   │ │  + BM25  │ │          │
                   │ └──────────┘ │          ▼
                   └──────────────┘   ┌──────────────┐
                                       │  Generator   │
                                       │  (Sonnet 4.5 │
                                       │   stream)    │
                                       └──────────────┘

         Background:
         ┌──────────────────────────────────────────┐
         │  Document Ingestor                        │
         │  (PDF→chunk→embed→index, batch + cron)    │
         └──────────────────────────────────────────┘

5.1 K8s部署清单

# 简化版
services:
  - rag-api:        FastAPI, 4 replicas, 2 vCPU
  - rerank-service: bge-reranker GPU, 1 L4 GPU, autoscale
  - qdrant:         3-node cluster, 16GB RAM each
  - redis-cache:    RAG cache (query → result), 4GB

六、生产经验：从0.864到0.948的真实路径

6.1 真实客户案例

某Fintech客户的演进过程：

• Week 1：v1 baseline，Recall 0.83，用户投诉"找不到东西"
• Week 2：换embedding (ada-002 → 3-large)，Recall → 0.86
• Week 3：加chunking优化（recursive splitter），Recall → 0.88
• Week 4：加BM25 hybrid，Recall → 0.90，"找精确名词"投诉消失
• Week 5：加rerank，Recall → 0.94，用户满意度跳升
• Week 6：加multi-query，Recall → 0.95，但latency被吐槽
• Week 7：multi-query → async + Haiku，latency回到Acceptable

总耗时约 6-7周，每一步都是：实测→上线→监控→迭代。

6.2 监控关键指标

- Retrieval Recall@k (人工标注集，每周抽检50 query)
- LLM faithfulness (Ragas, every query采样)
- p50/p95 latency (Prometheus)
- "no answer" rate (system says it cannot find)
- User satisfaction (thumb up/down)
- Cost per query

七、面试题

Q1: 把整个Week 21的RAG优化用一句话讲给VC听。

"我们把RAG的Recall@5从86%提升到95%，关键是用 hybrid (BM25 + dense) 解决精确term检索 + cross-encoder rerank做精排 + LLM-driven query rewrite处理短/模糊query，整体延迟仍在2.5秒内，单次cost仍<$0.03。"

Q2: 上线RAG v2三个月后，你怎么知道哪一层最值得继续投资？

用 ablation testing + offline eval set：(1) 维护一个300+ query的金标集（每月新增）; (2) 每周跑一次"删掉某层"的benchmark：no rerank、no hybrid、no multi-query；(3) 看哪一层 删掉后掉得最多。掉最多的就是最有价值的，可以继续投资优化。同时看 错例分析：把人工标注为"错"的case分类，看哪类问题最多，那是下一步优化方向。

Q3: 如果客户说"我的RAG质量不好"，你不能马上看代码，会问什么问题？

五个问题：(1) 怎么定义"不好"？faithfulness、relevance、还是latency？(2) 错例长什么样？让客户给5-10个bad case；(3) 数据质量？文档是PDF吗、有表格吗、scanned吗；(4) query分布？长query还是短query；(5) 基线分数？有offline eval set吗。一般这五个问题就能定位是数据/模型/检索/生成哪一层的问题。

Q4: 解释为什么同一个改进在不同场景效果差异大？

RAG的lift highly 数据 + query 依赖：(1) 语料专业性：金融术语密集→BM25价值大；通用FAQ→纯dense够用; (2) query长度分布：短query多→multi-query价值大；长query多→不需要; (3) 缩写密度：财报多缩写→term_expand必加；(4) 用户严谨度：分析师query规范→简单baseline够；零售用户随意→需要rewrite。Best practice：每个新domain先跑full stack ablation看lift分布，然后reduce到性价比最高的子集。

Q5: 如果只能做3层优化，哪3层？

1. better chunking (always-on, +2-5%, 几乎0成本)；2. hybrid search (BM25 + Dense + RRF) (+5-6%, latency<10ms增量)；3. rerank (bge-reranker-v2-m3) (+4-5%, 30ms增量, 自部署免费)。这三层组合把baseline 0.86推到 ~0.94，覆盖80%的quality lift，用最少的复杂度。

八、明日预告

Day 142: Hierarchical RAG——固定size chunking有个缺陷：长answer需要的信息可能跨多个chunks，但retrieval只送回小chunks上下文不够。明天我们实现 parent-child chunking 和 sentence window 检索：用小chunk做精确检索，但送给LLM的是 surrounding parent context，结合 auto-merging实现"找到子chunk → 合并相邻子chunks → 自动选择父chunk"的智能上下文扩展。