Learning_RoadmapAI_Assistant_BasicsModel_Building_and_APIData_Governance_and_SecurityLLM_Book_IngestionTokenizationAttention_MechanismGPT_ArchitecturePretrainingFineTuningLoRAReasoning_Model_Book_IngestionStarTalent_ArchitectureLLM_Engineer_Handbook_CourseRAG_Building_BlocksLangChain_RunnableAI_Engineering_Chip_HuyenLLM_Ops_FinancePython_BootcampReference_Index
进阶

RAG 系统构建 — 提炼自视频笔记

RAG 系统构建 — 提炼自视频笔记

对应课程: course_C W4 (RAG特征管道) / W9 (RAG推理)

难度: ★★★ (L3 进阶)


代码
# Step 1: MVP 范围定义
mvp = {
    "user": "课程学员 (单一用户群体)",
    "use_case": "技术 AI Q&A (单一用例)",
    "inventory": "视频 transcripts (单一数据源)"
}
print(f"MVP 范围:")
for k, v in mvp.items():
    print(f"  {k}: {v}")
代码
# Step 2: Golden 数据集 (模拟)
golden_dataset = [
    ("什么是决策树?", "transcript_01_supervised_learning"),
    ("如何做错误分析?", "transcript_02_error_analysis"),
    ("过拟合怎么解决?", "transcript_03_regularization"),
    ("什么是迁移学习?", "transcript_04_transfer_learning"),
]

import pandas as pd
df_golden = pd.DataFrame(golden_dataset, columns=["query", "ground_truth_id"])
print(f"Golden 数据集: {len(df_golden)} 对")
print(f"\n开发/测试分割: 3对训练评估, 1对最终测试")
print("\n合成查询技巧: 每个文档生成 type×difficulty 组合")
print("  - 类型: 事实型 / 概念型 / 流程型")
print("  - 难度: 简单 / 中等 / 困难 (含错误假设)")
代码
# Step 3: 检索评估指标
def retrieval_metrics(retrieved_ids, relevant_id, k=3):
    """计算 Precision, Recall@k, MRR"""
    top_k = retrieved_ids[:k]
    
    # Precision@k
    relevant_retrieved = sum(1 for rid in top_k if rid == relevant_id)
    precision_k = relevant_retrieved / k
    
    # Recall@k (假设只有 1 个相关文档)
    recall_k = 1.0 if relevant_id in top_k else 0.0
    
    # MRR (单个查询)
    try:
        rank = top_k.index(relevant_id) + 1
        mrr = 1.0 / rank
    except ValueError:
        mrr = 0.0
    
    return {
        "precision@k": precision_k,
        f"recall@{k}": recall_k,
        "mrr": mrr
    }

# 测试
result = retrieval_metrics(
    retrieved_ids=["t02", "t01", "t03", "t04", "t05"],
    relevant_id="t01",  # 正确答案
    k=3
)
for k, v in result.items():
    print(f"  {k}: {v}")
代码
# Step 4: Classic RAG vs Agentic RAG

class ClassicRAG:
    """经典 RAG: 用户查询 → 检索 → LLM 生成"""
    def __init__(self, llm_endpoint="http://localhost:8080/v1"):
        self.endpoint = llm_endpoint
    
    def query(self, user_input):
        # 1. 检索 (简化)
        context = f"[从知识库检索到的相关文档]"
        # 2. 构建 prompt
        prompt = f"基于以下上下文回答问题:\n{context}\n\n问题: {user_input}"
        # 3. 调用 LLM
        return f"[LLM 根据上下文生成的回答]"

class AgenticRAG:
    """Agent RAG: LLM 自主决定何时检索"""
    def __init__(self, llm_endpoint="http://localhost:8080/v1"):
        self.endpoint = llm_endpoint
    
    def query(self, user_input):
        # LLM 自己决定是否需要调用检索工具
        # 如果用户问"你好", 不需要检索
        # 如果用户问技术问题, 自动触发检索
        return f"[Agent 决定是否检索并生成回答]"

print("经验: 先跑通 Classic RAG → 再升级到 Agentic RAG")
print("因为 Classic RAG 把检索/生成分开, 容易定位问题")
代码
# Step 5: 实验迭代 — 一次只改一个变量
from dataclasses import dataclass

@dataclass
class RAGExperiment:
    name: str
    change: str
    recall_at_1: float
    recall_at_3: float
    bad_framing_rate: float

baseline = RAGExperiment("Baseline", "原始 prompt", 0.2, 0.6, 0.50)
exp_1 = RAGExperiment("Exp 1", "优化 prompt (上下文框架)", 0.2, 0.6, 0.07)

print(f"基线: bad_framing_rate = {baseline.bad_framing_rate:.0%}")
print(f"Exp1: bad_framing_rate = {exp_1.bad_framing_rate:.0%} ← 仅改 prompt")
print(f"改进幅度: {((exp_1.bad_framing_rate - baseline.bad_framing_rate)/baseline.bad_framing_rate)*100:.0f}%")
print()
print("关键: 先做 Error Analysis! 看 50-100 条回复, 找出常见失败模式")
print("常见失败: bad_framing, 过度复杂, 代码过时, 结构混乱")

关键引用

> "Design your project for experimentation, not production."

> — 在设计阶段就内置评估和实验能力, 而不是先搭好再改

> "Almost always the problem is in the source documents."

> — 更好的提取、预处理、分块比改模型更有效


关联素材: references/video-notes/如何在现实世界中构建 RAG 系统.txt

关联课程: 13_LLM_Engineer_Handbook_Course → W4 (RAG特征管道)