⚡ 第15课：知识存储优化

import time from collections import defaultdict from functools import wraps def timed(func): """计时装饰器""" @wraps(func) def wrapper(*args, **kwargs): start = time.perf_counter() result = func(*args, **kwargs) elapsed = (time.perf_counter() - start) * 1000 print(f" ⏱ {func.__name__}: {elapsed:.2f}ms") return result return wrapper class OptimizedGraphStore: """带索引和缓存的知识图谱存储""" def __init__(self): self.triples = [] # 多维索引 self.spo_index = {} # {(s,p,o): idx} self.s_index = defaultdict(list) # {s: [idx]} self.p_index = defaultdict(list) # {p: [idx]} self.o_index = defaultdict(list) # {o: [idx]} self.sp_index = defaultdict(list) # {(s,p): [idx]} self.po_index = defaultdict(list) # {(p,o): [idx]} # 缓存 self.query_cache = {} self.cache_hits = 0 self.cache_misses = 0 def add_triple(self, s, p, o): idx = len(self.triples) triple = (s, p, o) self.triples.append(triple) # 更新所有索引 self.spo_index[triple] = idx self.s_index[s].append(idx) self.p_index[p].append(idx) self.o_index[o].append(idx) self.sp_index[(s, p)].append(idx) self.po_index[(p, o)].append(idx) # 新数据使缓存失效 self.query_cache.clear() def query(self, s=None, p=None, o=None): """带缓存的模式查询""" cache_key = (s, p, o) if cache_key in self.query_cache: self.cache_hits += 1 return self.query_cache[cache_key] self.cache_misses += 1 # 选择最优索引 if s and p: indices = self.sp_index.get((s, p), []) elif p and o: indices = self.po_index.get((p, o), []) elif s: indices = self.s_index.get(s, []) elif p: indices = self.p_index.get(p, []) elif o: indices = self.o_index.get(o, []) else: indices = range(len(self.triples)) results = [self.triples[i] for i in indices] if s and p is None and o is None: results = [(h, r, t) for h, r, t in results if h == s] if o and s is None and p is None: results = [(h, r, t) for h, r, t in results if t == o] self.query_cache[cache_key] = results return results def cache_stats(self): total = self.cache_hits + self.cache_misses return { "缓存命中": self.cache_hits, ">缓存未命中": self.cache_misses, ">命中率": f"{self.cache_hits/total:.2%}" if total > 0 else "N/A", ">缓存大小": len(self.query_cache) } # ========== 性能测试 ========== store = OptimizedGraphStore() # 批量导入三元组 for i in range(1000): s = f"entity_{i % 100}" p = ["创作", "出生地", "属于", "任职", "好友"][i % 5] o = f"entity_{(i + i % 50) % 100}" store.add_triple(s, p, o) print(f"=== 存储规模: {len(store.triples)} 三元组 ===") # 无索引 vs 有索引对比 print(" === 索引查询测试 ===") # 查询1：按主语 r1 = store.query(s="entity_0") print(f" 查询 entity_0 相关: {len(r1)} 结果") # 查询2：按谓词 r2 = store.query(p="创作") print(f" 查询 '创作' 关系: {len(r2)} 结果") # 查询3：按主语+谓词 r3 = store.query(s="entity_0", p="创作") print(f" 查询 entity_0 创作: {len(r3)} 结果") # 缓存测试 print(" === 缓存测试 ===") store.query(s="entity_0") # 缓存未命中 store.query(s="entity_0") # 缓存命中 store.query(s="entity_0") # 缓存命中 for k, v in store.cache_stats().items(): print(f" {k}: {v}")

class TripleStoreBenchmark: """三元组存储引擎基准测试""" def __init__(self): self.engines = {} def register(self, name, store): self.engines[name] = store def benchmark_load(self, triples): """批量导入基准""" results = {} for name, store in self.engines.items(): start = time.perf_counter() for s, p, o in triples: store.add_triple(s, p, o) elapsed = (time.perf_counter() - start) * 1000 results[name] = elapsed return results def benchmark_query(self, queries): """查询基准""" results = {} for name, store in self.engines.items(): times = [] for s, p, o in queries: start = time.perf_counter() store.query(s, p, o) times.append((time.perf_counter() - start) * 1000000) # 微秒 results[name] = { "avg_us": np.mean(times), ">p50_us": np.median(times), ">p99_us": np.percentile(times, 99) } return results # 简单列表存储（无索引基线） class ListTripleStore: def __init__(self): self.triples = [] def add_triple(self, s, p, o): self.triples.append((s, p, o)) def query(self, s=None, p=None, o=None): return [(h,r,t) for h,r,t in self.triples if (s is None or h==s) and (p is None or r==p) and (o is None or t==o)] # 运行基准测试 bench = TripleStoreBenchmark() bench.register("索引存储", OptimizedGraphStore()) bench.register("列表存储", ListTripleStore()) # 生成测试数据 import numpy as np test_triples = [(f"e_{i%100}", f"r_{i%10}", f"e_{(i*7)%100}") for i in range(5000)] load_results = bench.benchmark_load(test_triples) print("=== 导入性能 ===") for name, ms in sorted(load_results.items(), key=lambda x: x[1]): print(f" {name}: {ms:.1f}ms") # 查询基准 query_tests = [("e_0", None, None), (None, "r_0", None), ("e_0", "r_0", None)] query_results = bench.benchmark_query(query_tests * 10) print(" === 查询性能 ===") for name, metrics in query_results.items(): print(f" {name}: avg={metrics['avg_us']:.1f}μs, p50={metrics['p50_us']:.1f}μs")