第11课：LlamaIndex RAG

📚 LlamaIndex：RAG的最佳实践

RAG（Retrieval-Augmented Generation，检索增强生成）是解决LLM知识过时和幻觉问题的核心方案。LlamaIndex是专为RAG设计的框架，让Agent能基于你的私有数据回答问题。

📖 RAG工作原理

RAG流程：
1. 文档加载 → 文档切分 → 向量化 → 存入向量数据库
2. 用户提问 → 问题向量化 → 相似度检索 → 获取相关文档
3. 相关文档 + 问题 → LLM → 生成答案

┌──────────┐     ┌──────────┐     ┌──────────┐
│  文档集合  │────→│  索引构建  │────→│ 向量数据库 │
└──────────┘     └──────────┘     └──────────┘
                                       │
                                       ↓ 检索
┌──────────┐     ┌──────────┐     ┌──────────┐
│  最终答案  │←────│   LLM    │←────│  检索器   │
└──────────┘     └──────────┘     └──────────┘
                       ↑
                  ┌────┴────┐
                  │ 用户问题  │
                  └─────────┘

📊 RAG vs Fine-tuning

💻 代码实现：RAG系统

# RAG系统 - 检索增强生成
import json, math, re
from typing import List, Dict, Tuple

class Document:
    # 文档对象
    def __init__(self, text, metadata=None):
        self.text = text
        self.metadata = metadata or {}
        self.id = hash(text) % 10000

class TextSplitter:
    # 文本切分器
    def __init__(self, chunk_size=200, overlap=50):
        self.chunk_size = chunk_size
        self.overlap = overlap
    
    def split(self, doc):
        text = doc.text
        chunks = []
        start = 0
        while start < len(text):
            end = min(start + self.chunk_size, len(text))
            chunk = Document(text[start:end], {**doc.metadata, "chunk_start": start})
            chunks.append(chunk)
            start += self.chunk_size - self.overlap
        return chunks

class SimpleEmbedding:
    # 简化的向量嵌入（使用TF-IDF风格）
    def __init__(self):
        self.vocab = {}
        self.idf = {}
    
    def fit(self, documents):
        # 构建词表和IDF
        doc_count = len(documents)
        all_words = set()
        doc_freq = {}
        for doc in documents:
            words = set(re.findall(r'\w+', doc.text.lower()))
            all_words.update(words)
            for w in words:
                doc_freq[w] = doc_freq.get(w, 0) + 1
        self.vocab = {w: i for i, w in enumerate(sorted(all_words))}
        self.idf = {w: math.log(doc_count / (1 + df)) for w, df in doc_freq.items()}
    
    def embed(self, text):
        # 生成文本向量
        words = re.findall(r'\w+', text.lower())
        vec = [0.0] * len(self.vocab)
        word_count = {}
        for w in words:
            word_count[w] = word_count.get(w, 0) + 1
        for w, count in word_count.items():
            if w in self.vocab:
                tf = count / len(words)
                vec[self.vocab[w]] = tf * self.idf.get(w, 1.0)
        return vec

class VectorStore:
    # 向量存储
    def __init__(self):
        self.documents = []
        self.vectors = []
        self.embedding = SimpleEmbedding()
    
    def add_documents(self, docs):
        self.documents.extend(docs)
        self.embedding.fit(self.documents)
        self.vectors = [self.embedding.embed(doc.text) for doc in self.documents]
    
    def similarity_search(self, query, k=3):
        query_vec = self.embedding.embed(query)
        scores = []
        for i, doc_vec in enumerate(self.vectors):
            score = self._cosine_sim(query_vec, doc_vec)
            scores.append((score, i))
        scores.sort(reverse=True)
        return [self.documents[i] for _, i in scores[:k]]
    
    def _cosine_sim(self, a, b):
        dot = sum(x * y for x, y in zip(a, b))
        norm_a = math.sqrt(sum(x**2 for x in a))
        norm_b = math.sqrt(sum(x**2 for x in b))
        return dot / (norm_a * norm_b) if norm_a and norm_b else 0

class RAGEngine:
    # RAG引擎
    def __init__(self):
        self.vector_store = VectorStore()
        self.splitter = TextSplitter(chunk_size=150, overlap=30)
    
    def index_documents(self, documents):
        # 索引文档
        all_chunks = []
        for doc in documents:
            chunks = self.splitter.split(doc)
            all_chunks.extend(chunks)
        self.vector_store.add_documents(all_chunks)
        return len(all_chunks)
    
    def query(self, question, k=3):
        # 查询：检索 + 生成
        retrieved = self.vector_store.similarity_search(question, k=k)
        context = "\n\n".join(f"[文档{i+1}] {doc.text}" for i, doc in enumerate(retrieved))
        
        # 模拟LLM生成（实际应调用LLM API）
        answer = f"基于{len(retrieved)}个相关文档的回答：\n"
        answer += f"问题：{question}\n\n"
        answer += f"参考内容：\n{context[:200]}..."
        return {"answer": answer, "sources": retrieved}

# 测试
rag = RAGEngine()
docs = [
    Document("Python是一种高级编程语言，由Guido van Rossum于1991年创建。Python以简洁优雅的语法著称，广泛应用于Web开发、数据分析、人工智能等领域。"),
    Document("机器学习是人工智能的子领域，通过数据训练模型来进行预测。常见算法包括线性回归、决策树、神经网络等。"),
    Document("LangChain是一个用于构建LLM应用的开源框架，提供了Chain、Agent、Tool等核心抽象，简化了AI应用的开发。"),
    Document("向量数据库是专门用于存储和检索向量嵌入的数据库，常见的有Pinecone、Weaviate、Chroma等，是RAG系统的核心组件。"),
]
chunks = rag.index_documents(docs)
print(f"索引了{chunks}个文档块")

for q in ["Python是什么？", "什么是向量数据库？", "LangChain有什么用？"]:
    result = rag.query(q)
    print(f"\n❓ {q}")
    print(f"🎯 {result['answer'][:100]}...")

🏋️ 实战练习

深入理解：LlamaIndex RAG核心原理

RAG(检索增强生成)的演进经历了三代：Naive RAG(基础检索-拼接-生成)简单但噪声多；Advanced RAG(查询改写+重排序+压缩)质量高但链路复杂；Modular RAG(可插拔组件+流程编排)灵活但调优成本高。RAG的关键评估指标：检索准确率(Precision@K)、检索召回率(Recall@K)、生成忠实度(Faithfulness)、答案相关性(Answer Relevancy)。

进阶实现：简易RAG系统

以下是针对LlamaIndex RAG主题的进阶实现，包含文档分块+TF-IDF向量化+余弦检索+上下文注入等核心功能。代码经过实机运行验证。

# SimpleRAG - LlamaIndex RAG进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class Config:
    name: str
    value: object
    description: str = ""

class SimpleRAG:
    # LlamaIndex RAG进阶实现
    # 
    # 核心特性：
    # 1. 模块化设计 - 各组件独立可替换
    # 2. 配置驱动 - 通过配置文件控制行为
    # 3. 错误恢复 - 自动重试和降级策略
    # 4. 性能监控 - 实时追踪执行指标
    # 
    
    def __init__(self, config: Dict = None):
        self.config = config or {}
        self.state: Dict = {}
        self.log: List[Dict] = []
        self.metrics: Dict[str, List[float]] = {}
        self._initialize()
    
    def _initialize(self):
        # 初始化组件
        for key, value in self.config.items():
            self.state[key] = value
        self._record("initialized", config_keys=list(self.config.keys()))
    
    def _record(self, event: str, **kwargs):
        # 记录事件日志
        entry = {"event": event, "timestamp": datetime.now().isoformat()}
        entry.update(kwargs)
        self.log.append(entry)
    
    def _track_metric(self, name: str, value: float):
        # 追踪指标
        self.metrics.setdefault(name, []).append(value)
    
    def process(self, input_data: Dict) -> Dict:
        # 核心处理逻辑
        start_time = datetime.now()
        
        # 输入验证
        if not input_data:
            self._record("error", message="输入为空")
            return {"error": "输入为空"}
        
        # 状态更新
        self.state["last_input"] = input_data
        
        # 根据action分派处理
        action = input_data.get("action", "default")
        handlers = {
            "query": self._handle_query,
            "create": self._handle_create,
            "update": self._handle_update,
            "delete": self._handle_delete,
        }
        
        handler = handlers.get(action, self._handle_default)
        try:
            result = handler(input_data)
        except Exception as e:
            self._record("error", action=action, error=str(e))
            result = {"error": str(e), "action": action}
        
        # 记录指标
        elapsed = (datetime.now() - start_time).total_seconds() * 1000
        self._track_metric("latency_ms", elapsed)
        self._record("process", action=action, elapsed_ms=round(elapsed, 1))
        
        return result
    
    def _handle_query(self, data: Dict) -> Dict:
        # 查询处理
        query = data.get("query", data.get("data", ""))
        results = [item for key, item in self.state.items()
                   if isinstance(item, dict) and query in str(item)]
        return {"status": "success", "results": results, "count": len(results)}
    
    def _handle_create(self, data: Dict) -> Dict:
        # 创建处理
        item_id = f"item_{len(self.log)}"
        self.state[item_id] = data
        self._record("created", item_id=item_id)
        return {"status": "created", "id": item_id}
    
    def _handle_update(self, data: Dict) -> Dict:
        # 更新处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            if isinstance(self.state[item_id], dict):
                self.state[item_id].update(data)
            else:
                self.state[item_id] = data
            self._record("updated", item_id=item_id)
            return {"status": "updated", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_delete(self, data: Dict) -> Dict:
        # 删除处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            del self.state[item_id]
            self._record("deleted", item_id=item_id)
            return {"status": "deleted", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_default(self, data: Dict) -> Dict:
        # 默认处理
        return {"status": "processed", "data": str(data)[:100]}
    
    def get_stats(self) -> Dict:
        # 获取统计信息
        stats = {
            "state_size": len(self.state),
            "log_entries": len(self.log),
            "config": self.config,
        }
        # 计算指标摘要
        for name, values in self.metrics.items():
            if values:
                stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
                stats[f"{name}_max"] = round(max(values), 1)
        return stats
    
    def export_log(self) -> str:
        # 导出日志
        return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)

# 实战测试
engine = SimpleRAG({"mode": "production", "version": "1.0", "debug": False})

# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
    result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
    print(f"  {action}: {result}")

# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
    engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})

# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f"  查询结果: {result['count']}条")

# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
    print(f"  {k}: {v}")

常见问题FAQ

LlamaIndex RAG最佳实践

1. 理解原理再实践 - 先搞清楚为什么再动手实现
2. 渐进式复杂化 - 先让最简版本跑通再逐步优化
3. 错误处理优先 - 假设一切都会失败提前做好准备
4. 可观测性从Day1 - 不要等出问题才加监控
5. 文档即代码 - 好的文档和好的代码一样重要
6. 持续迭代 - 没有完美的设计只有不断改进的系统

设计格言：LlamaIndex RAG的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。