阅读时间：约 14 分钟
前置知识：向量检索（R01）、分块策略（R02）、检索增强（R03）、RAG 评估（R04）

---

前面的文章解决了"搜得准"和"知道准不准"。现在面对最后一个问题：你的 RAG 系统上线了，用户越来越多，文档越来越多，数据越来越实时。怎么办？

四招：缓存（省成本）、增量更新（快索引）、知识图谱（结构化数据）、实时搜索（新鲜数据）。

---

前言：生产环境的真实挑战

一个典型的 RAG 上线后的问题：

• 用户问"这个项目的技术方案是什么"，1000 个人问了 100 遍
• 同一个问题，每次都要跑完整的向量检索，烧掉大量 Token
• 新文档加进来，全量重建索引，耗时 30 分钟，这期间用户搜不到新内容
• 实时数据（订单、库存）需要秒级更新，向量库更新太慢

这些不是算法问题，是工程问题。

📌 本章核心：缓存解决重复查询的 Token 成本，增量更新解决索引效率，知识图谱解决结构化数据检索，实时搜索解决数据新鲜度。

---

第一部分：缓存：省下来的 Token 都是利润

1.1 语义缓存

import hashlib
import time
from typing import Optional

class SemanticCache:
    """
    语义缓存：基于查询语义相似度缓存检索结果
    相同或相似的查询直接返回缓存，不调用向量库
    """
    
    def __init__(self, embedding_model, max_cache_size=1000, similarity_threshold=0.92):
        self.model = embedding_model
        self.max_cache_size = max_cache_size
        self.similarity_threshold = similarity_threshold
        self.cache = {}  # query_hash -> {"query": ..., "result": ..., "embedding": ..., "timestamp": ...}
        self.access_count = {}  # query_hash -> count
    
    def _hash(self, query):
        """生成查询哈希"""
        return hashlib.md5(query.lower().strip().encode()).hexdigest()
    
    def _embed(self, query):
        """生成查询向量"""
        return self.model.encode(query, normalize_embeddings=True)
    
    def get(self, query):
        """
        尝试从缓存获取结果
        返回: (hit: bool, result: list or None)
        """
        query_hash = self._hash(query)
        
        if query_hash in self.cache:
            cached = self.cache[query_hash]
            
            # 检查是否过期（默认 24 小时）
            if time.time() - cached["timestamp"] > 86400:
                del self.cache[query_hash]
                del self.access_count[query_hash]
                return False, None
            
            # 语义相似度检查
            cached_vec = cached["embedding"]
            query_vec = self._embed(query)
            similarity = float(np.dot(query_vec, cached_vec))
            
            if similarity >= self.similarity_threshold:
                self.access_count[query_hash] = self.access_count.get(query_hash, 0) + 1
                return True, cached["result"]
        
        return False, None
    
    def put(self, query, result):
        """将查询结果存入缓存"""
        query_hash = self._hash(query)
        
        # 如果缓存满了，删除最久未使用的
        if len(self.cache) >= self.max_cache_size and query_hash not in self.cache:
            # LRU 淘汰
            oldest_hash = min(self.access_count, key=self.access_count.get)
            del self.cache[oldest_hash]
            del self.access_count[oldest_hash]
        
        self.cache[query_hash] = {
            "query": query,
            "result": result,
            "embedding": self._embed(query),
            "timestamp": time.time()
        }
        self.access_count[query_hash] = self.access_count.get(query_hash, 0) + 1
    
    def stats(self):
        """缓存统计"""
        if not self.cache:
            return {"total_entries": 0, "hit_rate": 0}
        
        total_accesses = sum(self.access_count.values())
        return {
            "total_entries": len(self.cache),
            "total_accesses": total_accesses,
            "avg_accesses_per_entry": round(total_accesses / len(self.cache), 1) if self.cache else 0
        }

1.2 结果缓存

class ResultCache:
    """
    结果缓存：缓存完整 LLM 生成结果
    适合重复问题，比语义缓存更严格（必须完全匹配）
    """
    
    def __init__(self, ttl=3600):
        self.ttl = ttl  # 默认 1 小时
        self.cache = {}
    
    def get(self, key):
        if key in self.cache:
            entry = self.cache[key]
            if time.time() - entry["timestamp"] < self.ttl:
                return entry["result"]
            else:
                del self.cache[key]
        return None
    
    def put(self, key, result):
        self.cache[key] = {"result": result, "timestamp": time.time()}
    
    def clear_expired(self):
        """清理过期缓存"""
        now = time.time()
        expired = [k for k, v in self.cache.items() if now - v["timestamp"] > self.ttl]
        for k in expired:
            del self.cache[k]

1.3 缓存效果评估

class CacheEvaluator:
    """缓存效果评估"""
    
    @staticmethod
    def calculate_savings(hit_rate, avg_tokens_per_query, cost_per_1k_tokens):
        """
        计算缓存节省的成本
        """
        tokens_saved = hit_rate * avg_tokens_per_query
        cost_saved = (tokens_saved / 1000) * cost_per_1k_tokens
        return {
            "tokens_saved": round(tokens_saved, 1),
            "cost_saved_per_query": round(cost_saved, 6),
            "hit_rate": hit_rate
        }
    
    @staticmethod
    def evaluate_cache_performance(cache, test_queries):
        """
        评估缓存性能
        """
        hits = 0
        total = len(test_queries)
        
        for query in test_queries:
            hit, result = cache.get(query)
            if hit:
                hits += 1
                # 模拟缓存未命中时的成本
                # 实际项目中这里需要对比有缓存和无缓存的差异
        
        hit_rate = hits / total if total > 0 else 0
        return {
            "hit_rate": round(hit_rate, 3),
            "total_queries": total,
            "cache_hits": hits,
            "cache_misses": total - hits
        }

📌 本章要点：语义缓存（相似查询命中）、结果缓存（完全匹配命中）。语义缓存命中率 30-60%，结果缓存命中率 10-30%。缓存直接节省 Token 成本。

---

缓存解决重复查询。但文档更新了怎么办？

第二部分：增量更新：只更新变化的部分

2.1 增量索引

class IncrementalIndexManager:
    """
    增量索引管理器
    只更新发生变化的文档，不重建整个索引
    """
    
    def __init__(self, vector_store, document_store):
        self.vector_store = vector_store  # 向量数据库
        self.document_store = document_store  # 文档元数据
        self.document_hashes = {}  # doc_id -> content_hash
    
    def update_document(self, doc_id, new_content, chunk_size=500):
        """
        更新单个文档
        """
        # 1. 检查文档是否真的变化了
        content_hash = hashlib.md5(new_content.encode()).hexdigest()
        
        if doc_id in self.document_hashes and self.document_hashes[doc_id] == content_hash:
            return {"status": "unchanged", "message": "文档未变化"}
        
        # 2. 删除旧版本的向量
        self.vector_store.delete(doc_id)
        
        # 3. 重新分块并索引
        chunks = self._chunk_content(new_content, chunk_size)
        for chunk_id, chunk_content in chunks.items():
            self.vector_store.add(
                id=f"{doc_id}_chunk_{chunk_id}",
                content=chunk_content,
                metadata={"doc_id": doc_id, "chunk_id": chunk_id}
            )
        
        # 4. 更新哈希
        self.document_hashes[doc_id] = content_hash
        self.document_store.update(doc_id, new_content)
        
        return {"status": "updated", "chunks": len(chunks)}
    
    def delete_document(self, doc_id):
        """删除文档"""
        self.vector_store.delete_by_prefix(f"{doc_id}_chunk_")
        self.document_store.delete(doc_id)
        self.document_hashes.pop(doc_id, None)
        return {"status": "deleted"}
    
    def bulk_update(self, changes):
        """
        批量更新
        changes: [{"action": "update"|"delete", "doc_id": "...", "content": "..."}]
        """
        results = []
        for change in changes:
            if change["action"] == "update":
                result = self.update_document(change["doc_id"], change["content"])
            elif change["action"] == "delete":
                result = self.delete_document(change["doc_id"])
            else:
                result = {"status": "invalid_action"}
            
            results.append(result)
        
        return results
    
    def _chunk_content(self, content, chunk_size=500):
        """简单的内容分块"""
        chunks = {}
        current_chunk = []
        current_size = 0
        
        for line in content.split('\n'):
            current_chunk.append(line)
            current_size += len(line)
            
            if current_size >= chunk_size:
                chunks[f"{len(chunks)}"] = '\n'.join(current_chunk)
                current_chunk = []
                current_size = 0
        
        if current_chunk:
            chunks[f"{len(chunks)}"] = '\n'.join(current_chunk)
        
        return chunks

2.2 文档版本管理

class DocumentVersionManager:
    """
    文档版本管理器
    保留历史版本，支持回滚
    """
    
    def __init__(self):
        self.versions = {}  # doc_id -> [{"version": n, "content": ..., "timestamp": ...}]
    
    def save_version(self, doc_id, content):
        """保存版本"""
        if doc_id not in self.versions:
            self.versions[doc_id] = []
        
        latest = self.versions[doc_id][-1] if self.versions[doc_id] else {"version": -1}
        version = latest["version"] + 1
        
        self.versions[doc_id].append({
            "version": version,
            "content": content,
            "timestamp": time.time()
        })
        
        return version
    
    def rollback(self, doc_id, version):
        """回滚到指定版本"""
        if doc_id not in self.versions:
            return None
        
        for v in self.versions[doc_id]:
            if v["version"] == version:
                return v["content"]
        
        return None
    
    def get_history(self, doc_id):
        """获取版本历史"""
        return self.versions.get(doc_id, [])

📌 本章要点：增量索引只更新变化的部分，不重建整个索引。文档版本管理支持回滚。批量更新时，先检查是否变化，避免无效操作。

---

向量检索对非结构化数据有效，但结构化数据怎么办？

第三部分：知识图谱：结构化数据的检索

3.1 知识图谱 + 向量检索的混合

import networkx as nx

class KnowledgeGraph:
    """
    知识图谱（简化版）
    存储实体和关系，与向量检索结合使用
    """
    
    def __init__(self):
        self.graph = nx.Graph()  # 无向图
        self.node_vectors = {}   # node_id -> embedding
        self.metadata = {}       # node_id -> metadata
    
    def add_entity(self, entity_id, name, embedding, metadata=None):
        """添加实体"""
        self.graph.add_node(entity_id, name=name)
        self.node_vectors[entity_id] = embedding
        self.metadata[entity_id] = metadata or {}
    
    def add_relationship(self, source, target, relation_type, weight=1.0):
        """添加关系"""
        self.graph.add_edge(source, target, relation=relation_type, weight=weight)
    
    def search_entity(self, query, top_k=5):
        """
        实体搜索
        结合向量相似度和图结构
        """
        from sentence_transformers import SentenceTransformer
        model = SentenceTransformer("bge-large-zh")
        query_vec = model.encode(query)
        
        # 计算实体相似度
        similarities = {}
        for entity_id, vec in self.node_vectors.items():
            sim = float(np.dot(query_vec, vec))
            similarities[entity_id] = sim
        
        # Top-K
        top_entities = sorted(similarities.items(), key=lambda x: x[1], reverse=True)[:top_k]
        
        # 获取关系信息
        results = []
        for entity_id, score in top_entities:
            neighbors = list(self.graph.neighbors(entity_id))
            relations = self.graph[entity_id]
            
            results.append({
                "entity_id": entity_id,
                "name": self.graph.nodes[entity_id]["name"],
                "similarity": round(score, 3),
                "neighbors": neighbors,
                "relations": {
                    neighbor: {"type": self.graph[entity_id][neighbor].get("relation", ""),
                               "weight": self.graph[entity_id][neighbor].get("weight", 1)}
                    for neighbor in neighbors
                }
            })
        
        return results

3.2 混合检索：图谱 + 向量

class HybridGraphVectorSearch:
    """
    混合搜索：知识图谱 + 向量检索
    结构化数据用图谱，非结构化数据用向量
    """
    
    def __init__(self, graph, vector_store):
        self.graph = graph
        self.vector_store = vector_store
    
    def search(self, query, k=5):
        """
        混合搜索
        1. 从图谱中查找相关实体
        2. 从向量库中检索相关文档
        3. 合并排序
        """
        from sentence_transformers import SentenceTransformer
        model = SentenceTransformer("bge-large-zh")
        query_vec = model.encode(query)
        
        # 1. 图谱搜索
        graph_results = self.graph.search_entity(query, top_k=3)
        
        # 2. 向量检索
        vector_results = self.vector_store.search(query, k=5)
        
        # 3. 合并排序
        combined = []
        
        # 图谱结果
        for gr in graph_results:
            combined.append({
                "type": "entity",
                "entity_id": gr["entity_id"],
                "name": gr["name"],
                "score": gr["similarity"],
                "content": f"实体: {gr['name']}, 关系: {gr['relations']}"
            })
        
        # 向量结果
        for vr in vector_results:
            combined.append({
                "type": "document",
                "doc_id": vr["doc_id"],
                "score": vr["score"],
                "content": vr["content"]
            })
        
        # 按分数排序
        combined.sort(key=lambda x: x["score"], reverse=True)
        return combined[:k]
    
    def query_entity_path(self, entity_id, max_depth=2):
        """
        查询实体路径（图谱特有）
        适合查询"X 的上级是谁"、"A 和 B 的关系"
        """
        from collections import deque
        
        visited = set()
        queue = deque([(entity_id, 0)])
        paths = []
        
        while queue:
            current, depth = queue.popleft()
            
            if current in visited:
                continue
            
            visited.add(current)
            paths.append({"entity": current, "depth": depth})
            
            if depth < max_depth:
                for neighbor in self.graph.neighbors(current):
                    if neighbor not in visited:
                        queue.append((neighbor, depth + 1))
        
        return paths

📌 本章要点：知识图谱适合结构化数据（实体关系），向量检索适合非结构化数据（文档内容）。混合搜索把两者结合。图谱支持实体路径查询，向量支持语义搜索。

---

向量库更新有延迟，实时数据怎么搜？

第四部分：实时搜索：数据新鲜度

4.1 实时索引

import asyncio
import threading

class RealtimeIndex:
    """
    实时索引：支持流式数据实时更新
    """
    
    def __init__(self, vector_store, chunk_size=500):
        self.vector_store = vector_store
        self.chunk_size = chunk_size
        self.pending_updates = []
        self.lock = threading.Lock()
    
    def add_realtime_data(self, data_id, content):
        """
        添加实时数据
        """
        chunks = self._chunk(content)
        
        for chunk_id, chunk_content in chunks.items():
            self.pending_updates.append({
                "data_id": data_id,
                "chunk_id": chunk_id,
                "content": chunk_content
            })
        
        # 异步索引（不阻塞主流程）
        threading.Thread(target=self._index_pending, daemon=True).start()
    
    def _index_pending(self):
        """后台索引 pending 数据"""
        with self.lock:
            updates = self.pending_updates.copy()
            self.pending_updates.clear()
        
        for update in updates:
            try:
                self.vector_store.add(
                    id=f"{update['data_id']}_chunk_{update['chunk_id']}",
                    content=update["content"],
                    metadata={"data_id": update["data_id"], "realtime": True}
                )
            except Exception as e:
                # 记录错误，不中断主流程
                print(f"Index error: {e}")
    
    def _chunk(self, content):
        """分块"""
        chunks = {}
        current_chunk = []
        current_size = 0
        
        for line in content.split('\n'):
            current_chunk.append(line)
            current_size += len(line)
            
            if current_size >= self.chunk_size:
                chunks[f"{len(chunks)}"] = '\n'.join(current_chunk)
                current_chunk = []
                current_size = 0
        
        if current_chunk:
            chunks[f"{len(chunks)}"] = '\n'.join(current_chunk)
        
        return chunks

4.2 混合查询：实时 + 历史

class MixedQueryEngine:
    """
    混合查询引擎
    同时查询实时数据和历史数据
    """
    
    def __init__(self, realtime_index, historical_index):
        self.realtime_index = realtime_index
        self.historical_index = historical_index
    
    def search(self, query, k=5, realtime_weight=0.3):
        """
        混合搜索
        realtime_weight: 实时结果的权重
        """
        # 实时数据
        realtime_results = self.realtime_index.search(query, k=3)
        
        # 历史数据
        historical_results = self.historical_index.search(query, k=5)
        
        # 合并
        combined = []
        
        for r in realtime_results:
            r["score"] *= (1 + realtime_weight)  # 提升实时结果权重
            combined.append(r)
        
        combined.extend(historical_results)
        
        # 去重并排序
        seen = set()
        unique_results = []
        for r in combined:
            if r["id"] not in seen:
                seen.add(r["id"])
                unique_results.append(r)
        
        unique_results.sort(key=lambda x: x.get("score", 0), reverse=True)
        return unique_results[:k]

📌 本章要点：实时索引支持流式数据实时更新，异步处理不阻塞主流程。混合查询同时查实时和历史数据，实时结果权重更高。

---

第五部分：优化效果评估

5.1 综合优化指标

class OptimizationMetrics:
    """优化效果评估"""
    
    @staticmethod
    def evaluate_optimization(original_metrics, optimized_metrics):
        """
        评估优化效果
        original_metrics: 优化前指标
        optimized_metrics: 优化后指标
        """
        improvements = {}
        
        # 成本节省
        if "total_cost" in original_metrics and "total_cost" in optimized_metrics:
            cost_savings = (original_metrics["total_cost"] - optimized_metrics["total_cost"]) / \
                          original_metrics["total_cost"] * 100
            improvements["cost_savings_pct"] = round(cost_savings, 1)
        
        # 延迟改善
        if "avg_latency_ms" in original_metrics and "avg_latency_ms" in optimized_metrics:
            latency_improvement = (original_metrics["avg_latency_ms"] - optimized_metrics["avg_latency_ms"]) / \
                                 original_metrics["avg_latency_ms"] * 100
            improvements["latency_improvement_pct"] = round(latency_improvement, 1)
        
        # 准确率变化
        if "accuracy" in original_metrics and "accuracy" in optimized_metrics:
            accuracy_delta = optimized_metrics["accuracy"] - original_metrics["accuracy"]
            improvements["accuracy_delta"] = round(accuracy_delta, 3)
        
        return improvements

5.2 优化建议

优化手段	适用场景	预期效果
语义缓存	重复查询多	Token 成本降 30-50%
结果缓存	相同问题多	Token 成本降 10-30%
增量索引	文档频繁更新	索引速度提升 5-10 倍
知识图谱	实体关系查询	结构化数据检索准确率提升 20-40%
实时索引	实时数据场景	数据新鲜度提升到秒级

📌 本章要点：优化效果用成本、延迟、准确率评估。不同优化手段适用不同场景，效果也不同。

---

第六部分：生产级 RAG 系统架构

# ═══════════════════════════════════════════
# ProductionRAGSystem：生产级 RAG 系统
# ═══════════════════════════════════════════

class ProductionRAGSystem:
    """
    生产级 RAG 系统：集成所有优化
    """
    
    def __init__(self):
        # 核心组件
        self.vector_store = VectorStore()
        self.knowledge_graph = KnowledgeGraph()
        
        # 缓存
        self.semantic_cache = SemanticCache(embedding_model=...)
        self.result_cache = ResultCache()
        
        # 增量索引
        self.index_manager = IncrementalIndexManager(self.vector_store, DocumentStore())
        
        # 实时索引
        self.realtime_index = RealtimeIndex(self.vector_store)
        
        # 混合查询
        self.hybrid_search = MixedQueryEngine(self.realtime_index, self.vector_store)
    
    def query(self, query, k=5):
        """
        查询入口
        """
        # 1. 结果缓存（完全匹配）
        cached_result = self.result_cache.get(query)
        if cached_result:
            return {"source": "cache", "result": cached_result}
        
        # 2. 语义缓存（相似查询）
        hit, semantic_result = self.semantic_cache.get(query)
        if hit:
            return {"source": "semantic_cache", "result": semantic_result}
        
        # 3. 混合搜索
        results = self.hybrid_search.search(query, k=k)
        
        # 4. 缓存结果
        self.result_cache.put(query, results)
        self.semantic_cache.put(query, results)
        
        return {"source": "search", "result": results}
    
    def update_document(self, doc_id, content):
        """更新文档"""
        return self.index_manager.update_document(doc_id, content)
    
    def add_realtime_data(self, data_id, content):
        """添加实时数据"""
        self.realtime_index.add_realtime_data(data_id, content)

📌 本章要点：生产级 RAG 系统集成所有优化：缓存（语义 + 结果）、增量索引、知识图谱、实时索引。查询入口先查缓存，再搜索。

---

总结

1. 缓存：语义缓存（相似查询命中 30-60%）+ 结果缓存（完全匹配 10-30%）。直接省 Token 成本。
2. 增量更新：只更新变化的文档，不重建整个索引。文档版本管理支持回滚。
3. 知识图谱：结构化数据用图谱，非结构化数据用向量。混合搜索结合两者优势。
4. 实时搜索：流式数据实时更新，异步处理不阻塞主流程。混合查询同时查实时和历史。
5. 生产级架构：缓存 → 增量索引 → 知识图谱 → 实时搜索。每层独立，可单独优化。

🤔 思考一下：你的 RAG 系统现在有多少重复查询？如果加缓存，能省多少 Token？

---

思维导图

• RAG 优化
  • 缓存
    • 语义缓存（相似查询）
    • 结果缓存（完全匹配）
    • 成本节省 30-50%
  • 增量更新
    • 只更新变化部分
    • 文档版本管理
    • 批量更新优化
  • 知识图谱
    • 结构化数据检索
    • 实体关系查询
    • 与向量混合搜索
  • 实时搜索
    • 流式数据实时更新
    • 异步处理
    • 混合查询实时 + 历史
  • 生产级架构
    • 缓存 → 增量索引 → 知识图谱 → 实时搜索
    • 每层独立